Remove transaction inheritance
Transaction inheritance causes issues such as: 1. higher fragmentation in the db file, causing storage inefficiencies. 2. stale transactions holding an old view of the database, causing notes to be unavailable in certain scenarios This should be safe because direct access to transactions is hidden within the NostrDB Swift modules behind safe interfaces that prevent direct access, thus reducing or eliminating the chance of accidentally starting two transactions within the same thread (which would cause a deadlock). Closes: https://github.com/damus-io/damus/issues/3681 Changelog-Fixed: Disabled transaction inhe Signed-off-by: Daniel D’Aquino <daniel@daquino.me> Co-Authored-By: alltheseas <alltheseas@users.noreply.github.com>
This commit is contained in:
Daniel D’Aquino
co-authored by
alltheseas
alltheseas
parent
0a801b01cb
commit
c1fb8b592f
+80
-113
@@ -11,18 +11,26 @@ import Foundation
|
||||
fileprivate var txn_count: Int = 0
|
||||
#endif
|
||||
|
||||
/// Standard timeout for nostrdb transaction operations.
|
||||
///
|
||||
/// This timeout prevents indefinite hangs while allowing normal database operations to complete.
|
||||
/// The value is chosen empirically: long enough for typical database operations on the main thread,
|
||||
/// short enough to prevent user-visible freezes.
|
||||
private extension DispatchTimeInterval {
|
||||
static let ndbTransactionTimeout = DispatchTimeInterval.milliseconds(200)
|
||||
}
|
||||
|
||||
// Would use struct and ~Copyable but generics aren't supported well
|
||||
class NdbTxn<T>: RawNdbTxnAccessible {
|
||||
var txn: ndb_txn
|
||||
private var val: T!
|
||||
var moved: Bool
|
||||
var inherited: Bool
|
||||
var ndb: Ndb
|
||||
var generation: Int
|
||||
var name: String
|
||||
|
||||
static func pure(ndb: Ndb, val: T) -> NdbTxn<T> {
|
||||
.init(ndb: ndb, txn: ndb_txn(), val: val, generation: ndb.generation, inherited: true, name: "pure_txn")
|
||||
.init(ndb: ndb, txn: ndb_txn(), val: val, generation: ndb.generation, name: "pure_txn")
|
||||
}
|
||||
|
||||
/// Simple helper struct for the init function to avoid compiler errors encountered by using other techniques
|
||||
@@ -36,36 +44,21 @@ class NdbTxn<T>: RawNdbTxnAccessible {
|
||||
self.name = name ?? "txn"
|
||||
self.ndb = ndb
|
||||
self.generation = ndb.generation
|
||||
if let active_txn = Thread.current.threadDictionary["ndb_txn"] as? ndb_txn,
|
||||
let txn_generation = Thread.current.threadDictionary["txn_generation"] as? Int,
|
||||
txn_generation == ndb.generation
|
||||
{
|
||||
// some parent thread is active, use that instead
|
||||
print("txn: inherited txn")
|
||||
self.txn = active_txn
|
||||
self.inherited = true
|
||||
self.generation = Thread.current.threadDictionary["txn_generation"] as! Int
|
||||
let ref_count = Thread.current.threadDictionary["ndb_txn_ref_count"] as! Int
|
||||
let new_ref_count = ref_count + 1
|
||||
Thread.current.threadDictionary["ndb_txn_ref_count"] = new_ref_count
|
||||
} else {
|
||||
let result: R? = try? ndb.withNdb({
|
||||
var txn = ndb_txn()
|
||||
#if TXNDEBUG
|
||||
txn_count += 1
|
||||
#endif
|
||||
let ok = ndb_begin_query(ndb.ndb.ndb, &txn) != 0
|
||||
guard ok else { return .none }
|
||||
return .some(R(txn: txn, generation: ndb.generation))
|
||||
}, maxWaitTimeout: .milliseconds(200))
|
||||
guard let result else { return nil }
|
||||
self.txn = result.txn
|
||||
self.generation = result.generation
|
||||
Thread.current.threadDictionary["ndb_txn"] = self.txn
|
||||
Thread.current.threadDictionary["ndb_txn_ref_count"] = 1
|
||||
Thread.current.threadDictionary["txn_generation"] = ndb.generation
|
||||
self.inherited = false
|
||||
}
|
||||
|
||||
// Always create fresh transaction
|
||||
let result: R? = try? ndb.withNdb({
|
||||
var txn = ndb_txn()
|
||||
let ok = ndb_begin_query(ndb.ndb.ndb, &txn) != 0
|
||||
guard ok else { return .none }
|
||||
#if TXNDEBUG
|
||||
txn_count += 1
|
||||
#endif
|
||||
return .some(R(txn: txn, generation: ndb.generation))
|
||||
}, maxWaitTimeout: .ndbTransactionTimeout)
|
||||
guard let result else { return nil }
|
||||
self.txn = result.txn
|
||||
self.generation = result.generation
|
||||
|
||||
#if TXNDEBUG
|
||||
print("txn: open gen\(self.generation) '\(self.name)' \(txn_count)")
|
||||
#endif
|
||||
@@ -73,11 +66,10 @@ class NdbTxn<T>: RawNdbTxnAccessible {
|
||||
self.val = with(self)
|
||||
}
|
||||
|
||||
private init(ndb: Ndb, txn: ndb_txn, val: T, generation: Int, inherited: Bool, name: String) {
|
||||
private init(ndb: Ndb, txn: ndb_txn, val: T, generation: Int, name: String) {
|
||||
self.txn = txn
|
||||
self.val = val
|
||||
self.moved = false
|
||||
self.inherited = inherited
|
||||
self.ndb = ndb
|
||||
self.generation = generation
|
||||
self.name = name
|
||||
@@ -99,27 +91,15 @@ class NdbTxn<T>: RawNdbTxnAccessible {
|
||||
print("txn: not closing. db closed")
|
||||
return
|
||||
}
|
||||
if let ref_count = Thread.current.threadDictionary["ndb_txn_ref_count"] as? Int {
|
||||
let new_ref_count = ref_count - 1
|
||||
Thread.current.threadDictionary["ndb_txn_ref_count"] = new_ref_count
|
||||
assert(new_ref_count >= 0, "NdbTxn reference count should never be below zero")
|
||||
if new_ref_count <= 0 {
|
||||
_ = try? ndb.withNdb({
|
||||
ndb_end_query(&self.txn)
|
||||
}, maxWaitTimeout: .milliseconds(200))
|
||||
Thread.current.threadDictionary.removeObject(forKey: "ndb_txn")
|
||||
Thread.current.threadDictionary.removeObject(forKey: "ndb_txn_ref_count")
|
||||
}
|
||||
}
|
||||
if inherited {
|
||||
print("txn: not closing. inherited ")
|
||||
return
|
||||
}
|
||||
if moved {
|
||||
//print("txn: not closing. moved")
|
||||
return
|
||||
}
|
||||
|
||||
_ = try? ndb.withNdb({
|
||||
ndb_end_query(&self.txn)
|
||||
}, maxWaitTimeout: .ndbTransactionTimeout)
|
||||
|
||||
#if TXNDEBUG
|
||||
txn_count -= 1;
|
||||
print("txn: close gen\(generation) '\(name)' \(txn_count)")
|
||||
@@ -129,14 +109,14 @@ class NdbTxn<T>: RawNdbTxnAccessible {
|
||||
// functor
|
||||
func map<Y>(_ transform: (T) -> Y) -> NdbTxn<Y> {
|
||||
self.moved = true
|
||||
return .init(ndb: self.ndb, txn: self.txn, val: transform(val), generation: generation, inherited: inherited, name: self.name)
|
||||
return .init(ndb: self.ndb, txn: self.txn, val: transform(val), generation: generation, name: self.name)
|
||||
}
|
||||
|
||||
// comonad!?
|
||||
// useful for moving ownership of a transaction to another value
|
||||
func extend<Y>(_ with: (NdbTxn<T>) -> Y) -> NdbTxn<Y> {
|
||||
self.moved = true
|
||||
return .init(ndb: self.ndb, txn: self.txn, val: with(self), generation: generation, inherited: inherited, name: self.name)
|
||||
return .init(ndb: self.ndb, txn: self.txn, val: with(self), generation: generation, name: self.name)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -156,13 +136,12 @@ class SafeNdbTxn<T: ~Copyable> {
|
||||
var txn: ndb_txn
|
||||
var val: T!
|
||||
var moved: Bool
|
||||
var inherited: Bool
|
||||
var ndb: Ndb
|
||||
var generation: Int
|
||||
var name: String
|
||||
|
||||
static func pure(ndb: Ndb, val: consuming T) -> SafeNdbTxn<T> {
|
||||
.init(ndb: ndb, txn: ndb_txn(), val: val, generation: ndb.generation, inherited: true, name: "pure_txn")
|
||||
.init(ndb: ndb, txn: ndb_txn(), val: val, generation: ndb.generation, name: "pure_txn")
|
||||
}
|
||||
|
||||
/// Simple helper struct for the init function to avoid compiler errors encountered by using other techniques
|
||||
@@ -173,52 +152,42 @@ class SafeNdbTxn<T: ~Copyable> {
|
||||
|
||||
static func new(on ndb: Ndb, with valueGetter: (PlaceholderNdbTxn) -> T? = { _ in () }, name: String = "txn") -> SafeNdbTxn<T>? {
|
||||
guard !ndb.is_closed else { return nil }
|
||||
let generation: Int
|
||||
let txn: ndb_txn
|
||||
let inherited: Bool
|
||||
if let active_txn = Thread.current.threadDictionary["ndb_txn"] as? ndb_txn,
|
||||
let txn_generation = Thread.current.threadDictionary["txn_generation"] as? Int,
|
||||
txn_generation == ndb.generation
|
||||
{
|
||||
// some parent thread is active, use that instead
|
||||
print("txn: inherited txn")
|
||||
txn = active_txn
|
||||
inherited = true
|
||||
generation = Thread.current.threadDictionary["txn_generation"] as! Int
|
||||
let ref_count = Thread.current.threadDictionary["ndb_txn_ref_count"] as! Int
|
||||
let new_ref_count = ref_count + 1
|
||||
Thread.current.threadDictionary["ndb_txn_ref_count"] = new_ref_count
|
||||
} else {
|
||||
let result: R? = try? ndb.withNdb({
|
||||
var txn = ndb_txn()
|
||||
#if TXNDEBUG
|
||||
txn_count += 1
|
||||
#endif
|
||||
let ok = ndb_begin_query(ndb.ndb.ndb, &txn) != 0
|
||||
guard ok else { return .none }
|
||||
return .some(R(txn: txn, generation: ndb.generation))
|
||||
}, maxWaitTimeout: .milliseconds(200))
|
||||
guard let result else { return nil }
|
||||
txn = result.txn
|
||||
generation = result.generation
|
||||
Thread.current.threadDictionary["ndb_txn"] = txn
|
||||
Thread.current.threadDictionary["ndb_txn_ref_count"] = 1
|
||||
Thread.current.threadDictionary["txn_generation"] = ndb.generation
|
||||
inherited = false
|
||||
}
|
||||
|
||||
// Always create fresh transaction
|
||||
let result: R? = try? ndb.withNdb({
|
||||
var txn = ndb_txn()
|
||||
let ok = ndb_begin_query(ndb.ndb.ndb, &txn) != 0
|
||||
guard ok else { return .none }
|
||||
#if TXNDEBUG
|
||||
txn_count += 1
|
||||
#endif
|
||||
return .some(R(txn: txn, generation: ndb.generation))
|
||||
}, maxWaitTimeout: .ndbTransactionTimeout)
|
||||
guard let result else { return nil }
|
||||
let txn = result.txn
|
||||
let generation = result.generation
|
||||
|
||||
#if TXNDEBUG
|
||||
print("txn: open gen\(generation) '\(name)' \(txn_count)")
|
||||
#endif
|
||||
let placeholderTxn = PlaceholderNdbTxn(txn: txn)
|
||||
guard let val = valueGetter(placeholderTxn) else { return nil }
|
||||
return SafeNdbTxn<T>(ndb: ndb, txn: txn, val: val, generation: generation, inherited: inherited, name: name)
|
||||
guard let val = valueGetter(placeholderTxn) else {
|
||||
// Fix leak: Close transaction before returning nil
|
||||
var mutableTxn = txn
|
||||
_ = try? ndb.withNdb({ ndb_end_query(&mutableTxn) }, maxWaitTimeout: .ndbTransactionTimeout)
|
||||
#if TXNDEBUG
|
||||
txn_count -= 1
|
||||
print("txn: close (valueGetter nil) gen\(generation) '\(name)' \(txn_count)")
|
||||
#endif
|
||||
return nil
|
||||
}
|
||||
return SafeNdbTxn<T>(ndb: ndb, txn: txn, val: val, generation: generation, name: name)
|
||||
}
|
||||
|
||||
private init(ndb: Ndb, txn: ndb_txn, val: consuming T, generation: Int, inherited: Bool, name: String) {
|
||||
private init(ndb: Ndb, txn: ndb_txn, val: consuming T, generation: Int, name: String) {
|
||||
self.txn = txn
|
||||
self.val = consume val
|
||||
self.moved = false
|
||||
self.inherited = inherited
|
||||
self.ndb = ndb
|
||||
self.generation = generation
|
||||
self.name = name
|
||||
@@ -233,27 +202,15 @@ class SafeNdbTxn<T: ~Copyable> {
|
||||
print("txn: not closing. db closed")
|
||||
return
|
||||
}
|
||||
if let ref_count = Thread.current.threadDictionary["ndb_txn_ref_count"] as? Int {
|
||||
let new_ref_count = ref_count - 1
|
||||
Thread.current.threadDictionary["ndb_txn_ref_count"] = new_ref_count
|
||||
assert(new_ref_count >= 0, "NdbTxn reference count should never be below zero")
|
||||
if new_ref_count <= 0 {
|
||||
_ = try? ndb.withNdb({
|
||||
ndb_end_query(&self.txn)
|
||||
}, maxWaitTimeout: .milliseconds(200))
|
||||
Thread.current.threadDictionary.removeObject(forKey: "ndb_txn")
|
||||
Thread.current.threadDictionary.removeObject(forKey: "ndb_txn_ref_count")
|
||||
}
|
||||
}
|
||||
if inherited {
|
||||
print("txn: not closing. inherited ")
|
||||
return
|
||||
}
|
||||
if moved {
|
||||
//print("txn: not closing. moved")
|
||||
return
|
||||
}
|
||||
|
||||
_ = try? ndb.withNdb({
|
||||
ndb_end_query(&self.txn)
|
||||
}, maxWaitTimeout: .ndbTransactionTimeout)
|
||||
|
||||
#if TXNDEBUG
|
||||
txn_count -= 1;
|
||||
print("txn: close gen\(generation) '\(name)' \(txn_count)")
|
||||
@@ -263,14 +220,14 @@ class SafeNdbTxn<T: ~Copyable> {
|
||||
// functor
|
||||
func map<Y>(_ transform: (borrowing T) -> Y) -> SafeNdbTxn<Y> {
|
||||
self.moved = true
|
||||
return .init(ndb: self.ndb, txn: self.txn, val: transform(val), generation: generation, inherited: inherited, name: self.name)
|
||||
return .init(ndb: self.ndb, txn: self.txn, val: transform(val), generation: generation, name: self.name)
|
||||
}
|
||||
|
||||
// comonad!?
|
||||
// useful for moving ownership of a transaction to another value
|
||||
func extend<Y>(_ with: (SafeNdbTxn<T>) -> Y) -> SafeNdbTxn<Y> {
|
||||
self.moved = true
|
||||
return .init(ndb: self.ndb, txn: self.txn, val: with(self), generation: generation, inherited: inherited, name: self.name)
|
||||
return .init(ndb: self.ndb, txn: self.txn, val: with(self), generation: generation, name: self.name)
|
||||
}
|
||||
|
||||
consuming func maybeExtend<Y>(_ with: (consuming SafeNdbTxn<T>) -> Y?) -> SafeNdbTxn<Y>? where Y: ~Copyable {
|
||||
@@ -278,10 +235,20 @@ class SafeNdbTxn<T: ~Copyable> {
|
||||
let ndb = self.ndb
|
||||
let txn = self.txn
|
||||
let generation = self.generation
|
||||
let inherited = self.inherited
|
||||
let name = self.name
|
||||
guard let newVal = with(consume self) else { return nil }
|
||||
return .init(ndb: ndb, txn: txn, val: newVal, generation: generation, inherited: inherited, name: name)
|
||||
|
||||
guard let newVal = with(consume self) else {
|
||||
// Fix leak: Close transaction on nil path if we own it
|
||||
var mutableTxn = txn
|
||||
_ = try? ndb.withNdb({ ndb_end_query(&mutableTxn) }, maxWaitTimeout: .ndbTransactionTimeout)
|
||||
#if TXNDEBUG
|
||||
txn_count -= 1
|
||||
print("txn: close (maybeExtend nil) gen\(generation) '\(name)' \(txn_count)")
|
||||
#endif
|
||||
return nil
|
||||
}
|
||||
|
||||
return .init(ndb: ndb, txn: txn, val: newVal, generation: generation, name: name)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -304,7 +271,7 @@ extension NdbTxn where T: OptionalType {
|
||||
return nil
|
||||
}
|
||||
self.moved = true
|
||||
return NdbTxn<T.Wrapped>(ndb: self.ndb, txn: self.txn, val: unwrappedVal, generation: generation, inherited: inherited, name: name)
|
||||
return NdbTxn<T.Wrapped>(ndb: self.ndb, txn: self.txn, val: unwrappedVal, generation: generation, name: name)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user