diff --git a/Package.swift b/Package.swift index fbbefc2..d3e596a 100644 --- a/Package.swift +++ b/Package.swift @@ -18,6 +18,7 @@ let swiftSettings: [PackageDescription.SwiftSetting] = [ .enableUpcomingFeature("InternalImportsByDefault"), .enableUpcomingFeature("MemberImportVisibility"), .enableUpcomingFeature("InferIsolatedConformances"), + .enableUpcomingFeature("NonisolatedNonsendingByDefault"), .enableUpcomingFeature("ImmutableWeakCaptures"), ] @@ -38,19 +39,26 @@ let package = Package( dependencies: [ .package(url: "https://github.com/mochidev/AsyncSequenceReader.git", .upToNextMinor(from: "0.5.0")), .package(url: "https://github.com/mochidev/Bytes.git", .upToNextMinor(from: "0.6.2")), + .package(url: "https://github.com/mochidev/swift-questionable-concurrency", .upToNextMinor(from: "0.2.2")), ], targets: [ .target( name: "CodableDatastore", dependencies: [ "AsyncSequenceReader", - "Bytes" + "Bytes", + .product(name: "QuestionableConcurrency", package: "swift-questionable-concurrency") ], swiftSettings: swiftSettings ), .testTarget( name: "CodableDatastoreTests", - dependencies: ["CodableDatastore"], + dependencies: [ + "AsyncSequenceReader", + "CodableDatastore", + "Bytes", + .product(name: "QuestionableConcurrency", package: "swift-questionable-concurrency") + ], swiftSettings: swiftSettings ), ] diff --git a/Sources/CodableDatastore/Datastore/Datastore.swift b/Sources/CodableDatastore/Datastore/Datastore.swift index a16410b..2c87f25 100644 --- a/Sources/CodableDatastore/Datastore/Datastore.swift +++ b/Sources/CodableDatastore/Datastore/Datastore.swift @@ -125,7 +125,7 @@ extension Datastore { return descriptor } - func decoder(for version: Version) throws -> @Sendable (_ data: Data) async throws -> (id: IdentifierType, instance: InstanceType) { + nonisolated func decoder(for version: Version) throws -> @Sendable (_ data: Data) async throws -> (id: IdentifierType, instance: InstanceType) { guard let decoder = decoders[version] else { throw DatastoreError.missingDecoder(version: String(describing: version)) } @@ -533,7 +533,7 @@ extension Datastore { datastoreKey: self.key ) { versionData, instanceData in let entryVersion = try Version(versionData) - let decoder = try await self.decoder(for: entryVersion) + let decoder = try self.decoder(for: entryVersion) let decodedValue = try await decoder(instanceData) try await provider.yield(decodedValue) @@ -676,7 +676,7 @@ extension Datastore { datastoreKey: self.key ) { versionData, instanceData in let entryVersion = try Version(versionData) - let decoder = try await self.decoder(for: entryVersion) + let decoder = try self.decoder(for: entryVersion) let instance = try await decoder(instanceData).instance try await provider.yield(instance) @@ -690,7 +690,7 @@ extension Datastore { let persistedEntry = try await transaction.primaryIndexCursor(for: identifier, datastoreKey: self.key) let entryVersion = try Version(persistedEntry.versionData) - let decoder = try await self.decoder(for: entryVersion) + let decoder = try self.decoder(for: entryVersion) let instance = try await decoder(persistedEntry.instanceData).instance try await provider.yield(instance) @@ -953,7 +953,7 @@ extension Datastore { }.compactMap { event in try? await event.mapEntries { entry in let version = try Version(entry.versionData) - let decoder = try await self.decoder(for: version) + let decoder = try self.decoder(for: version) let instance = try await decoder(entry.instanceData).instance return instance } diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift index 45f22b9..d3030d3 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Datastore/PersistenceDatastore.swift @@ -27,8 +27,6 @@ extension DiskPersistence { var cachedRootObject: DatastoreRootManifest? - var lastUpdateDescriptorTask: Task? - /// The root objects that are being tracked in memory. var trackedRootObjects: [RootObject.ID : WeakValue] = [:] var trackedIndexes: [Index.ID : WeakValue] = [:] diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift index 58ae78f..3da6c28 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/DiskPersistence.swift @@ -8,6 +8,7 @@ // public import Foundation +import QuestionableConcurrency public actor DiskPersistence: Persistence { /// The location of this persistence. @@ -16,15 +17,16 @@ public actor DiskPersistence: Persistence { /// A cached instance of the store info as last loaded from disk. var cachedStoreInfo: StoreInfo? - /// A pointer to the last store info updater, so updates can be serialized after the last request - var lastUpdateStoreInfoTask: Task? + /// A transaction stream for storeInfo updates, so reads and writes can be serialized in request order. + var storeInfoTransactionStream = TransactionStream() /// The loaded Snapshots var snapshots: [SnapshotIdentifier: Snapshot] = [:] var registeredDatastores: [DatastoreKey : [WeakDatastore]] = [:] - var lastTransaction: Transaction? + var lastMutatingTransaction: Transaction? + var rootTransactionStream = TransactionStream() /// Shared caches across all snapshots and datastores. var rollingRootObjectCacheIndex = 0 @@ -36,7 +38,8 @@ public actor DiskPersistence: Persistence { var rollingPageCacheIndex = 0 var rollingPageCache: [Datastore.Page] = [] - var transactionCounter = 0 + nonisolated(unsafe) var transactionCounter = 0 + let transactionCounterLock = UnfairLock() /// Initialize a ``DiskPersistence`` with a read-write URL. /// @@ -166,35 +169,28 @@ extension DiskPersistence { cachedStoreInfo = storeInfo } - /// Load and update the store info in an updater, returning the task for the updater. + /// Load and update the store info in an updater. /// /// This method loads the ``StoreInfo`` from cache, offers it to be mutated, then writes it back to disk, if it changed. It is up to the caller to update the modification date of the store. /// /// - Note: Calling this method when no store info exists on disk will create it, even if no changes occur in the block. /// - Parameter updater: An updater that takes a mutable reference to a store info, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func updateStoreInfo( - @_inheritActorContext updater: @Sendable @escaping (_ storeInfo: inout StoreInfo) async throws -> T - ) -> Task where AccessMode == ReadWrite { + /// - Returns: The value returned from the `updater`. + func withStoreInfo( + updater: (_ storeInfo: inout StoreInfo) async throws -> T + ) async throws -> T where AccessMode == ReadWrite { if let storeInfo = DiskPersistenceTaskLocals.storeInfo(for: self) { - return Task { - var updatedStoreInfo = storeInfo - let returnValue = try await updater(&updatedStoreInfo) - - guard updatedStoreInfo == storeInfo else { - throw DiskPersistenceInternalError.nestedStoreWrite - } - - return returnValue + var updatedStoreInfo = storeInfo + let returnValue = try await updater(&updatedStoreInfo) + + guard updatedStoreInfo == storeInfo else { + throw DiskPersistenceInternalError.nestedStoreWrite } + + return returnValue } - /// Grab the last task so we can chain off of it in a serial manner. - let lastUpdaterTask = lastUpdateStoreInfoTask - let updaterTask = Task { - /// We don't care if the last request throws an error or not, but we do want it to complete first. - _ = try? await lastUpdaterTask?.value - + return try await storeInfoTransactionStream.withTransaction { /// Load the store info so we have a fresh copy, unless we have a cached copy already. var storeInfo = try cachedStoreInfo ?? self.loadStoreInfo() @@ -209,31 +205,23 @@ extension DiskPersistence { } return returnValue } - /// Assign the task to our pointer so we can depend on it the next time. Also, re-wrap it so we can keep proper type information when returning from this method. - lastUpdateStoreInfoTask = Task { try await updaterTask.value } - - return updaterTask } - /// Load the store info in an accessor, returning the task for the updater. + /// Load the store info in an updater. /// /// This method loads the ``StoreInfo`` from cache. /// /// - Parameter accessor: An accessor that takes an immutable reference to a store info, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func updateStoreInfo( - @_inheritActorContext accessor: @Sendable @escaping (_ storeInfo: StoreInfo) async throws -> T - ) -> Task { + /// - Returns: The value returned from the `accessor`. + @_disfavoredOverload + func withStoreInfo( + accessor: (_ storeInfo: StoreInfo) async throws -> T + ) async throws -> T { if let storeInfo = DiskPersistenceTaskLocals.storeInfo(for: self) { - return Task { try await accessor(storeInfo) } + return try await accessor(storeInfo) } - /// Grab the last task so we can chain off of it in a serial manner. - let lastUpdaterTask = lastUpdateStoreInfoTask - let updaterTask = Task { - /// We don't care if the last request throws an error or not, but we do want it to complete first. - _ = try? await lastUpdaterTask?.value - + return try await storeInfoTransactionStream.withTransaction { /// Load the store info so we have a fresh copy, unless we have a cached copy already. let storeInfo = try cachedStoreInfo ?? self.loadStoreInfo() @@ -242,40 +230,6 @@ extension DiskPersistence { try await accessor(storeInfo) } } - /// Assign the task to our pointer so we can depend on it the next time. Also, re-wrap it so we can keep proper type information when returning from this method. - lastUpdateStoreInfoTask = Task { try await updaterTask.value } - - return updaterTask - } - - /// Load and update the store info in an updater. - /// - /// This method loads the ``StoreInfo`` from cache, offers it to be mutated, then writes it back to disk, if it changed. It is up to the caller to update the modification date of the store. - /// - /// - Note: Calling this method when no store info exists on disk will create it, even if no changes occur in the block. - /// - Parameter updater: An updater that takes a mutable reference to a store info, and will forward the returned value to the caller. - /// - Returns: The value returned from the `updater`. - func withStoreInfo( - updater: @Sendable (_ storeInfo: inout StoreInfo) async throws -> T - ) async throws -> T where AccessMode == ReadWrite { - try await withoutActuallyEscaping(updater) { escapingClosure in - try await updateStoreInfo(updater: escapingClosure).value - } - } - - /// Load the store info in an updater. - /// - /// This method loads the ``StoreInfo`` from cache. - /// - /// - Parameter accessor: An accessor that takes an immutable reference to a store info, and will forward the returned value to the caller. - /// - Returns: The value returned from the `accessor`. - @_disfavoredOverload - func withStoreInfo( - accessor: @Sendable (_ storeInfo: StoreInfo) async throws -> T - ) async throws -> T { - try await withoutActuallyEscaping(accessor) { escapingClosure in - try await updateStoreInfo(accessor: escapingClosure).value - } } } @@ -304,22 +258,22 @@ extension DiskPersistence { return snapshot } - /// Load and update the current snapshot in an updater, returning the task for the updater. + /// Load the current snapshot in an updater. /// /// This method loads the current ``Snapshot`` so it can be updated. /// /// - Note: Calling this method when no store info exists on disk will create it. /// - Parameter dateUpdate: The method to which to update the date of the main store with. /// - Parameter updater: An updater that takes a reference to the current ``Snapshot``, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func updateCurrentSnapshot( + /// - Returns: The value returned from the `accessor`. + func updatingCurrentSnapshot( dateUpdate: ModificationUpdate = .updateOnWrite, - updater: @escaping (_ snapshot: Snapshot) async throws -> T - ) -> Task where AccessMode == ReadWrite { + updater: sending (_ snapshot: Snapshot) async throws -> T + ) async throws -> T where AccessMode == ReadWrite { /// Grab access to the store info to load and update it. - return updateStoreInfo { storeInfo in + try await withStoreInfo { storeInfo in /// Grab the current snapshot from the store info - let snapshot = self.loadSnapshot(from: storeInfo) + let snapshot = loadSnapshot(from: storeInfo) /// Load a modification date to use let modificationDate = dateUpdate.modificationDate(for: storeInfo.modificationDate) @@ -335,40 +289,6 @@ extension DiskPersistence { } } - /// Load the current snapshot in an accessor, returning the task for the accessor. - /// - /// - Parameter accessor: An accessor that takes a reference to the current ``Snapshot``, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func updateCurrentSnapshot( - accessor: @escaping (_ snapshot: Snapshot) async throws -> T - ) -> Task { - /// Grab access to the store info to load and update it. - return updateStoreInfo { storeInfo in - /// Grab the current snapshot from the store info - let snapshot = self.loadSnapshot(from: storeInfo) - - /// Let the accessor do what it needs to do with the snapshot - return try await accessor(snapshot) - } - } - - /// Load the current snapshot in an updater. - /// - /// This method loads the current ``Snapshot`` so it can be updated. - /// - /// - Note: Calling this method when no store info exists on disk will create it. - /// - Parameter dateUpdate: The method to which to update the date of the main store with. - /// - Parameter updater: An updater that takes a reference to the current ``Snapshot``, and will forward the returned value to the caller. - /// - Returns: The value returned from the `accessor`. - func updatingCurrentSnapshot( - dateUpdate: ModificationUpdate = .updateOnWrite, - updater: @Sendable (_ snapshot: Snapshot) async throws -> T - ) async throws -> T where AccessMode == ReadWrite { - try await withoutActuallyEscaping(updater) { escapingClosure in - try await updateCurrentSnapshot(dateUpdate: dateUpdate, updater: escapingClosure).value - } - } - /// Load the current snapshot in an accessor. /// /// This method loads the current ``Snapshot`` so it can be accessed. @@ -379,10 +299,15 @@ extension DiskPersistence { /// - Returns: The value returned from the `accessor`. @_disfavoredOverload func readingCurrentSnapshot( - accessor: @Sendable (_ snapshot: Snapshot) async throws -> T + accessor: sending @Sendable (_ snapshot: Snapshot) async throws -> T ) async throws -> T { - try await withoutActuallyEscaping(accessor) { escapingClosure in - try await updateCurrentSnapshot(accessor: escapingClosure).value + /// Grab access to the store info to load and update it. + try await withStoreInfo { storeInfo in + /// Grab the current snapshot from the store info + let snapshot = self.loadSnapshot(from: storeInfo) + + /// Let the accessor do what it needs to do with the snapshot + return try await accessor(snapshot) } } @@ -405,7 +330,7 @@ extension DiskPersistence { var currentSnapshot: Snapshot { // TODO: This should return a readonly snapshot, but we need to be able to make a read-only copy from the persistence first. get async throws { - try await withStoreInfo { await loadSnapshot(from: $0) } + try await withStoreInfo { loadSnapshot(from: $0) } } } @@ -540,48 +465,82 @@ extension DiskPersistence { // MARK: - Transactions extension DiskPersistence { - func nextTransactionCounter() -> Int { - let transactionIndex = transactionCounter - transactionCounter += 1 - return transactionIndex + nonisolated func nextTransactionCounter() -> Int { + transactionCounterLock.withLock { + let transactionIndex = transactionCounter + transactionCounter += 1 + return transactionIndex + } + } + + /// Replace the last mutating root transaction with the specified transaction, and return the previous one. + func enqueue(rootTransaction: Transaction) -> Transaction? { + let lastTransaction = lastMutatingTransaction + if !rootTransaction.options.contains(.readOnly) { + self.lastMutatingTransaction = rootTransaction + } + return lastTransaction } public func _withTransaction( + isolation actor: isolated (any Actor)? = #isolation, actionName: String?, options: UnsafeTransactionOptions, - transaction: @Sendable (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + transaction operation: (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T ) async throws -> T { - try await withoutActuallyEscaping(transaction) { escapingTransaction in - /// If the transaction is starting in the context of another persistence's transaction, make sure it is a read-only one. Otherwise assert and throw an error as it likely indicates a mistake and could lead to unexpected consistency violations if one persistence succeeds while the other fails. - if - Transaction.isTransactingExternally(to: self), - !options.contains(.readOnly) - { + /// If the transaction is starting in the context of another persistence's transaction, make sure it is a read-only one. Otherwise assert and throw an error as it likely indicates a mistake and could lead to unexpected consistency violations if one persistence succeeds while the other fails. + if isTransactingExternally() { + guard options.contains(.readOnly) + else { assertionFailure(DatastoreInterfaceError.transactingWithinExternalPersistence.localizedDescription) throw DatastoreInterfaceError.transactingWithinExternalPersistence } - - let currentCounter = nextTransactionCounter() -// print("[CDS] [\(storeURL.lastPathComponent)] Starting transaction \(currentCounter) “\(actionName ?? "")” - \(options)") - let (transaction, task) = await Transaction.makeTransaction( + } + + let currentCounter = nextTransactionCounter() +// print("[CDS] [\(storeURL.lastPathComponent)] Starting transaction \(currentCounter) “\(actionName ?? "")” - \(options)") + + if let parent = Transaction.unsafeCurrentTransaction(for: self) { +// print("[CDS] [\(storeURL.lastPathComponent)] Found parent \(parent.transactionIndex), making child \(currentCounter)") + assert(!parent.options.contains(.readOnly) || options.contains(.readOnly), "A child transaction was declared read-write, even though its parent was read-only!") + let transaction = Transaction( persistence: self, transactionIndex: currentCounter, - lastTransaction: lastTransaction, + parent: parent, actionName: actionName, options: options - ) { interface, isDurable in - try await escapingTransaction(interface, isDurable) - } + ) - /// Save the last non-concurrent top-level transaction from the list. Note that disk persistence currently does not support concurrent idempotent transactions. - if !options.contains(.readOnly), transaction.parent == nil { - lastTransaction = transaction - } + /// Enqueue and get the last non-concurrent transaction from the parent's list. Note that disk persistence currently does not support concurrent idempotent transactions. + let lastChildTransaction = try await parent.enqueue(childTransaction: transaction) - let result = try await task.value + let result = try await transaction.run( + lastTransaction: !options.contains(.readOnly) ? lastChildTransaction : nil, + isDurable: false, + operation: operation + ) // print("[CDS] [\(storeURL.lastPathComponent)] Finished transaction \(currentCounter) “\(actionName ?? "")” - \(options)") return result } + + let transaction = Transaction( + persistence: self, + transactionIndex: currentCounter, + parent: nil, + actionName: actionName, + options: options + ) + + /// Save the last non-concurrent root transaction from the list. Note that disk persistence currently does not support concurrent idempotent transactions. + let lastTransaction = await enqueue(rootTransaction: transaction) + + let result = try await transaction.run( + lastTransaction: !options.contains(.readOnly) ? lastTransaction : nil, + isDurable: options.isDisjoint(with: [.collateWrites, .readOnly]), + operation: operation + ) +// print("[CDS] [\(storeURL.lastPathComponent)] Finished transaction \(currentCounter) “\(actionName ?? "")” - \(options)") + return result } func persist( @@ -726,6 +685,7 @@ private enum DiskPersistenceTaskLocals { } static func with( + isolation actor: isolated (any Actor)? = #isolation, storeInfo: StoreInfo, for persistence: DiskPersistence, operation: () async throws -> R diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift index 55f93c2..bdd0839 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Snapshot/Snapshot.swift @@ -8,6 +8,7 @@ // import Foundation +import QuestionableConcurrency typealias SnapshotIdentifier = DatedIdentifier> @@ -34,8 +35,8 @@ actor Snapshot { /// A cached instance of the current iteration as last loaded from disk. var cachedIteration: SnapshotIteration? - /// A pointer to the last manifest updater, so updates can be serialized after the last request. - var lastUpdateManifestTask: Task? + /// A transaction stream for manifest updates, so reads and writes can be serialized in request order. + var manifestTransactionStream = TransactionStream() /// The loaded datastores. var datastores: [DatastoreIdentifier: DiskPersistence.Datastore] = [:] @@ -170,35 +171,28 @@ extension Snapshot { cachedIteration = iteration } - /// Load and update the manifest in an updater, returning the task for the updater. + /// Load and update the manifest in an updater. /// /// This method loads the ``SnapshotManifest`` from cache, offers it to be mutated, then writes it back to disk, if it changed. It is up to the caller to update the modification date of the store. /// /// - Note: Calling this method when no manifest exists on disk will create it, even if no changes occur in the block. /// - Parameter updater: An updater that takes a mutable reference to a manifest, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func updateManifest( - updater: @Sendable @escaping (_ manifest: inout SnapshotManifest, _ iteration: inout SnapshotIteration) async throws -> T - ) -> Task where AccessMode == ReadWrite { + /// - Returns: The value returned from the `updater`. + func updatingManifest( + updater: (_ manifest: inout SnapshotManifest, _ iteration: inout SnapshotIteration) async throws -> T + ) async throws -> T where AccessMode == ReadWrite { if let (manifest, iteration) = SnapshotTaskLocals.manifest(for: persistence) { - return Task { - var updatedManifest = manifest - var updatedIteration = iteration - let returnValue = try await updater(&updatedManifest, &updatedIteration) - - guard updatedManifest == manifest, updatedIteration == iteration - else { throw DiskPersistenceInternalError.nestedSnapshotWrite } - - return returnValue - } + var updatedManifest = manifest + var updatedIteration = iteration + let returnValue = try await updater(&updatedManifest, &updatedIteration) + + guard updatedManifest == manifest, updatedIteration == iteration + else { throw DiskPersistenceInternalError.nestedSnapshotWrite } + + return returnValue } - /// Grab the last task so we can chain off of it in a serial manner. - let lastUpdaterTask = lastUpdateManifestTask - let updaterTask = Task { - /// We don't care if the last request throws an error or not, but we do want it to complete first. - _ = try? await lastUpdaterTask?.value - + return try await manifestTransactionStream.withTransaction { /// Load the manifest so we have a fresh copy, unless we have a cached copy already. var manifest = try cachedManifest ?? self.loadManifest() var iteration: SnapshotIteration @@ -233,32 +227,23 @@ extension Snapshot { } return returnValue } - /// Assign the task to our pointer so we can depend on it the next time. Also, re-wrap it so we can keep proper type information when returning from this method. - lastUpdateManifestTask = Task { try await updaterTask.value } - - return updaterTask } - /// Load the manifest in an accessor, returning the task for the updater. + /// Load the manifest in an updater. /// /// This method loads the ``SnapshotManifest`` from cache. /// /// - Parameter accessor: An accessor that takes an immutable reference to a manifest, and will forward the returned value to the caller. - /// - Returns: A ``/Swift/Task`` which contains the value of the updater upon completion. - func readManifest( - accessor: @Sendable @escaping (_ manifest: SnapshotManifest, _ iteration: SnapshotIteration) async throws -> T - ) -> Task { - + /// - Returns: The value returned from the `accessor`. + @_disfavoredOverload + func readingManifest( + accessor: (_ manifest: SnapshotManifest, _ iteration: SnapshotIteration) async throws -> T + ) async throws -> T { if let (manifest, iteration) = SnapshotTaskLocals.manifest(for: persistence) { - return Task { try await accessor(manifest, iteration) } + return try await accessor(manifest, iteration) } - /// Grab the last task so we can chain off of it in a serial manner. - let lastUpdaterTask = lastUpdateManifestTask - let readerTask = Task { - /// We don't care if the last request throws an error or not, but we do want it to complete first. - _ = try? await lastUpdaterTask?.value - + return try await manifestTransactionStream.withTransaction { /// Load the manifest so we have a fresh copy, unless we have a cached copy already. let manifest = try cachedManifest ?? self.loadManifest() var iteration: SnapshotIteration @@ -276,40 +261,6 @@ extension Snapshot { try await accessor(manifest, iteration) } } - /// Assign the task to our pointer so we can depend on it the next time. Also, re-wrap it so we can keep proper type information when returning from this method. - lastUpdateManifestTask = Task { try await readerTask.value } - - return readerTask - } - - /// Load and update the manifest in an updater. - /// - /// This method loads the ``SnapshotManifest`` from cache, offers it to be mutated, then writes it back to disk, if it changed. It is up to the caller to update the modification date of the store. - /// - /// - Note: Calling this method when no manifest exists on disk will create it, even if no changes occur in the block. - /// - Parameter updater: An updater that takes a mutable reference to a manifest, and will forward the returned value to the caller. - /// - Returns: The value returned from the `updater`. - func updatingManifest( - updater: @Sendable (_ manifest: inout SnapshotManifest, _ iteration: inout SnapshotIteration) async throws -> T - ) async throws -> T where AccessMode == ReadWrite { - try await withoutActuallyEscaping(updater) { escapingClosure in - try await updateManifest(updater: escapingClosure).value - } - } - - /// Load the manifest in an updater. - /// - /// This method loads the ``SnapshotManifest`` from cache. - /// - /// - Parameter accessor: An accessor that takes an immutable reference to a manifest, and will forward the returned value to the caller. - /// - Returns: The value returned from the `accessor`. - @_disfavoredOverload - func readingManifest( - accessor: @Sendable (_ manifest: SnapshotManifest, _ iteration: SnapshotIteration) async throws -> T - ) async throws -> T { - try await withoutActuallyEscaping(accessor) { escapingClosure in - try await readManifest(accessor: escapingClosure).value - } } } @@ -322,6 +273,7 @@ private enum SnapshotTaskLocals { } static func with( + isolation actor: isolated (any Actor)? = #isolation, manifest: SnapshotManifest, iteration: SnapshotIteration, for persistence: DiskPersistence, @@ -389,7 +341,7 @@ extension Snapshot { /// Iterate through each datastore and copy the data over for (_, datastoreInfo) in iteration.dataStores { - let (datastore, _) = await loadDatastore(for: datastoreInfo.key, from: iteration) + let (datastore, _) = loadDatastore(for: datastoreInfo.key, from: iteration) try await datastore.copy( rootIdentifier: datastoreInfo.root, datastoreKey: datastoreInfo.key, diff --git a/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift b/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift index 50320a6..7ea84bc 100644 --- a/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift +++ b/Sources/CodableDatastore/Persistence/Disk Persistence/Transaction/Transaction.swift @@ -7,15 +7,21 @@ // mochidev-codable-datastore: 8A3D87799CB24B2BA7A7661369B88325 // -import Foundation import Bytes +import Foundation +import QuestionableConcurrency extension DiskPersistence { actor Transaction: AnyDiskTransaction { unowned let persistence: DiskPersistence unowned let parent: Transaction? - weak var lastReadWriteChildTransaction: Transaction? + weak var lastMutatingChildTransaction: Transaction? + + var workDidFinishPromise: Promise? + var transactionDidPersistPromise: Promise? + let workDidFinishResult: AsyncResult + let transactionDidPersistResult: AsyncResult private(set) var task: Task! let transactionIndex: Int @@ -37,7 +43,7 @@ extension DiskPersistence { var isActive = false - private init( + init( persistence: DiskPersistence, transactionIndex: Int, parent: Transaction?, @@ -49,39 +55,96 @@ extension DiskPersistence { self.actionName = actionName self.options = options self.transactionIndex = transactionIndex + + let workDidFinishPromise = Promise(name: "DiskPersistence.Transaction.workDidFinish - \"\(actionName ?? "")\"") + let transactionDidPersistPromise = Promise(name: "DiskPersistence.Transaction.transactionDidPersist - \"\(actionName ?? "")\"") + + self.workDidFinishResult = workDidFinishPromise.future + self.transactionDidPersistResult = transactionDidPersistPromise.future + self.workDidFinishPromise = consume workDidFinishPromise + self.transactionDidPersistPromise = consume transactionDidPersistPromise } - private func attachTask( - options: UnsafeTransactionOptions, - @_inheritActorContext handler: @Sendable @escaping () async throws -> T - ) async -> Task { - let task = Task { - isActive = true - let returnValue = try await TransactionTaskLocals.with(transaction: self, for: persistence) { - try await handler() + /// An internal method to attach an operation to the transaction. + /// - Warning: Do not call this method more than once for a single transaction. + /// - Parameters: + /// - lastTransaction: The last transaction this one should be chained to. + /// - isDurable: A flag that reports whether or not the operation was durable. + /// - operation: The operation itself, which takes a reference to the owning transaction and the `isDurable` flag + /// - Returns: The return value of the operation, if any. + func run( + isolation actor: isolated (any Actor)? = #isolation, + lastTransaction: Transaction?, + isDurable: Bool, + operation: (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + ) async throws -> T { + let returnValue: T + do { + await setIsActive(true) + + /// If provided, wait for the last transaction to properly finish before startign the next one. + if let lastTransaction { + await lastTransaction.yieldPersistenceCompletion() } - isActive = false - /// If we don't care to collate our writes, go ahead and wait for the persistence to stick - if !options.contains(.collateWrites) { - try await self.persist() + /// Perform the operation with the current transaction set as the task-local. + returnValue = try await TransactionTaskLocals.with(transaction: self, for: persistence) { + try await operation(self, isDurable) } - return returnValue + await setIsActive(false) + + /// Mark the work as done so opportunistic transaction may begin. + await reportWorkDidFinish() + } catch { + /// If the operation failed, make sure to resume both continuations so the next transaction can start. + await reportWorkDidFinish() + await reportTransactionDidPersist() + + throw error } - self.task = Task { - _ = try await task.value - - /// If we previously skipped persisting, go ahead and do so now. - if options.contains(.collateWrites) { + /// Persist the work that was just completed to signal to the next transaction that it can start, but check for the requested timing option first. ``persist()`` takes care of signaling when the persistence is finished. + if options.contains(.collateWrites) { + /// If we are skipping immediate writes, kick off persistence in a separate task. + Task { try await self.persist() } + } else { + /// If we don't care to collate our writes, go ahead and wait for the persistence to stick + try await self.persist() } - return task + return returnValue + } + + private func reportWorkDidFinish() { + guard let workDidFinishPromise = workDidFinishPromise.take() + else { fatalError("workDidFinishPromise was already consumed. Please report this as a critical bug!") } + workDidFinishPromise.resume() + } + + private func reportTransactionDidPersist() { + guard let transactionDidPersistPromise = transactionDidPersistPromise.take() + else { fatalError("transactionDidPersistPromise was already consumed. Please report this as a critical bug!") } + transactionDidPersistPromise.resume() + } + + /// Yield and wait for the main work of the transaction to be complete. + func yieldWorkCompletion() async { + await workDidFinishResult.yield() } + /// Yield and wait for the transaction to finish persisting, which is the ideal point to start another mutating transaction. + func yieldPersistenceCompletion() async { + await transactionDidPersistResult.yield() + } + + private func setIsActive(_ isActive: Bool) { + self.isActive = isActive + } + + /// Ensure the receiving transaction is currenrtly active. func checkIsActive() throws { guard isActive else { assertionFailure(DatastoreInterfaceError.transactionInactive.localizedDescription) @@ -137,6 +200,10 @@ extension DiskPersistence { deletedRootObjects.removeAll() deletedIndexes.removeAll() deletedPages.removeAll() + + guard let transactionDidPersistPromise = transactionDidPersistPromise.take() + else { fatalError("transactionDidPersistPromise was already consumed. Please report this as a critical bug!") } + transactionDidPersistPromise.resume() } /// If the transaction is read-only, stop here without applying anything to the parent. @@ -203,76 +270,17 @@ extension DiskPersistence { } } - static func makeTransaction( - persistence: DiskPersistence, - transactionIndex: Int, - lastTransaction: Transaction?, - actionName: String?, - options: UnsafeTransactionOptions, - @_inheritActorContext handler: @Sendable @escaping (_ transaction: Transaction, _ isDurable: Bool) async throws -> T - ) async -> (Transaction, Task) { - if let parent = Self.unsafeCurrentTransaction(for: persistence) { -// print("[CDS] [\(persistence.storeURL.lastPathComponent)] Found parent \(parent.transactionIndex), making child \(transactionIndex)") - let (child, task) = await parent.childTransaction( - transactionIndex: transactionIndex, - actionName: actionName, - options: options, - handler: handler - ) - return (child, task) - } - - let transaction = Transaction( - persistence: persistence, - transactionIndex: transactionIndex, - parent: nil, - actionName: actionName, - options: options - ) - - let task = await transaction.attachTask(options: options) { - /// If the transaction is not read only, wait for the last transaction to properly finish before starting the next one. - if !options.contains(.readOnly) { - try? await lastTransaction?.task.value - } - return try await handler(transaction, options.isDisjoint(with: [.collateWrites, .readOnly])) - } - - return (transaction, task) - } - - func childTransaction( - transactionIndex: Int, - actionName: String?, - options: UnsafeTransactionOptions, - @_inheritActorContext handler: @Sendable @escaping (_ transaction: Transaction, _ isDurable: Bool) async throws -> T - ) async -> (Transaction, Task) { - assert(!self.options.contains(.readOnly) || options.contains(.readOnly), "A child transaction was declared read-write, even though its parent was read-only!") - let childTransaction = Transaction( - persistence: persistence, - transactionIndex: transactionIndex, - parent: self, - actionName: actionName, - options: options - ) + /// Replace the last mutating child transaction with the specified transaction, and return the previous one. + func enqueue(childTransaction: Transaction) throws -> Transaction? { + /// Make sure the parent transaction actually started, as we'll be dumping state into it as new child transactions complete. + try checkIsActive() - /// Get the last non-concurrent transaction from the list. Note that disk persistence currently does not support concurrent idempotent transactions. - let lastChild = lastReadWriteChildTransaction + /// Enqueue and get the last non-concurrent transaction from the parent's list. Note that disk persistence currently does not support concurrent idempotent transactions. + let lastChildTransaction = lastMutatingChildTransaction if !childTransaction.options.contains(.readOnly) { - lastReadWriteChildTransaction = childTransaction - } - - let task = await childTransaction.attachTask(options: options) { - try self.checkIsActive() - - /// If the transaction is not read only, wait for the last transaction to properly finish before starting the next one. - if !options.contains(.readOnly) { - _ = try? await lastChild?.task.value - } - return try await handler(childTransaction, false) + lastMutatingChildTransaction = childTransaction } - - return (childTransaction, task) + return lastChildTransaction } func rootObject(for datastoreKey: DatastoreKey) async throws -> Datastore.RootObject? { @@ -1370,7 +1378,16 @@ extension DiskPersistence.Transaction { } } -// MARK: - Helper Types +// MARK: - Task Locals + +extension DiskPersistence { + /// Check to see if we are in a transaction that pertains to a different persistence than the one provided. + /// + /// This is determined by checking if the only transactions present belong to other persistences. If there are no other transactions, or a transaction for the persistence is already registered, it has already been vetted and is no longer considered external. + nonisolated func isTransactingExternally() -> Bool { + !TransactionTaskLocals.transactionStorage.isEmpty && TransactionTaskLocals.transactionStorage[ObjectIdentifier(self)] == nil + } +} fileprivate protocol AnyDiskTransaction: Sendable {} @@ -1383,6 +1400,7 @@ fileprivate enum TransactionTaskLocals { } static func with( + isolation actor: isolated (any Actor)? = #isolation, transaction: any AnyDiskTransaction, for persistence: DiskPersistence, operation: () async throws -> R diff --git a/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift b/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift index e1a94cc..8f79eb9 100644 --- a/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift +++ b/Sources/CodableDatastore/Persistence/Memory Persistence/MemoryPersistence.swift @@ -15,9 +15,10 @@ public actor MemoryPersistence: Persistence { extension MemoryPersistence { public func _withTransaction( + isolation actor: isolated (any Actor)? = #isolation, actionName: String?, options: UnsafeTransactionOptions, - transaction: @Sendable (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + transaction: (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T ) async throws -> T { preconditionFailure("Unimplemented") } diff --git a/Sources/CodableDatastore/Persistence/Persistence.swift b/Sources/CodableDatastore/Persistence/Persistence.swift index c80d41d..2420051 100644 --- a/Sources/CodableDatastore/Persistence/Persistence.swift +++ b/Sources/CodableDatastore/Persistence/Persistence.swift @@ -19,13 +19,33 @@ public protocol Persistence: Sendable { /// - options: The options to use while building the transaction. /// - transaction: A closure representing the transaction with which to perform operations on. You should not escape the provided transaction. func _withTransaction( + isolation actor: isolated (any Actor)?, actionName: String?, options: UnsafeTransactionOptions, - @_inheritActorContext transaction: @Sendable (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + transaction: (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T ) async throws -> T } extension Persistence { + /// Perform a transaction on the persistence with the specified options. + /// - Parameters: + /// - actionName: The name of the action to use in an undo operation. Only the names of top-level transactions are recorded. We recommend you use the Base localization name here, as you may no longer have the localized version in future versions of your app. + /// - options: The options to use while building the transaction. + /// - transaction: A closure representing the transaction with which to perform operations on. You should not escape the provided transaction. + func _withTransaction( + isolation actor: isolated (any Actor)? = #isolation, + actionName: String?, + options: UnsafeTransactionOptions, + transaction: (_ transaction: any DatastoreInterfaceProtocol, _ isDurable: Bool) async throws -> T + ) async throws -> T { + try await _withTransaction( + isolation: actor, + actionName: actionName, + options: options, + transaction: transaction + ) + } + /// Perform a set of operations as a single transaction. /// /// Within the transaction block, perform operations on multiple ``Datastore``s such that if any one of them were to fail, none will be persisted, and if all of them succeed, they will be persisted atomically. @@ -39,9 +59,10 @@ extension Persistence { /// - transaction: A closure with the set of operations to perform. Parameters include a reference to the persistence, and a flag indicating if the transaction is durable. /// - SeeAlso: ``Persistence/perform(actionName:options:transaction:)-1tpvd`` public func perform( + isolation actor: isolated (any Actor)? = #isolation, actionName: String? = nil, options: TransactionOptions = [], - _inheritActorContext transaction: @Sendable (_ persistence: Self, _ isDurable: Bool) async throws -> T + transaction: (_ persistence: Self, _ isDurable: Bool) async throws -> T ) async throws -> T { try await _withTransaction( actionName: actionName, @@ -64,9 +85,10 @@ extension Persistence { /// - transaction: A closure with the set of operations to perform. /// - SeeAlso: ``Persistence/perform(actionName:options:transaction:)-5476l`` public func perform( + isolation actor: isolated (any Actor)? = #isolation, actionName: String? = nil, options: TransactionOptions = [], - @_inheritActorContext transaction: @Sendable () async throws -> T + transaction: () async throws -> T ) async throws -> T { try await _withTransaction( actionName: actionName, diff --git a/Tests/CodableDatastoreTests/DiskPersistenceTests.swift b/Tests/CodableDatastoreTests/DiskPersistenceTests.swift index 60a8a64..590eff6 100644 --- a/Tests/CodableDatastoreTests/DiskPersistenceTests.swift +++ b/Tests/CodableDatastoreTests/DiskPersistenceTests.swift @@ -131,20 +131,17 @@ final class DiskPersistenceTests: XCTestCase, @unchecked Sendable { let date1 = Date(timeIntervalSince1970: 0) let date2 = Date(timeIntervalSince1970: 10) - let task1 = await persistence.updateStoreInfo { storeInfo in + try await persistence.withStoreInfo { storeInfo in sleep(1) XCTAssertNotEqual(storeInfo.modificationDate, date2) storeInfo.modificationDate = date1 } - let task2 = await persistence.updateStoreInfo { storeInfo in + try await persistence.withStoreInfo { storeInfo in XCTAssertEqual(storeInfo.modificationDate, date1) storeInfo.modificationDate = date2 } - try await task1.value - try await task2.value - let currentStoreInfo = await persistence.cachedStoreInfo XCTAssertEqual(currentStoreInfo?.modificationDate, date2) } diff --git a/Tests/CodableDatastoreTests/DiskTransactionTests.swift b/Tests/CodableDatastoreTests/DiskTransactionTests.swift index aff22d5..a56ead7 100644 --- a/Tests/CodableDatastoreTests/DiskTransactionTests.swift +++ b/Tests/CodableDatastoreTests/DiskTransactionTests.swift @@ -70,4 +70,162 @@ final class DiskTransactionTests: XCTestCase, @unchecked Sendable { XCTAssertEqual(existingDescriptor, descriptor) } } + + func testSerialMutableRootTransactions() async throws { + let persistence = try DiskPersistence(readWriteURL: temporaryStoreURL) + + struct TestFormat: DatastoreFormat { + enum Version: Int, CaseIterable { + case zero + } + + struct Instance: Codable {} + typealias Identifier = UUID + + static let defaultKey: DatastoreKey = "test" + static let currentVersion = Version.zero + } + + var tasks: [Task] = [] + + let iterations = 5000 + nonisolated(unsafe) var count = 0 + nonisolated(unsafe) var lastStartedIndex = 0 + for index in 0..] = [] + + let iterations = 5000 + nonisolated(unsafe) var count = 0 + for index in 0..] = [] + + let iterations = 5000 + nonisolated(unsafe) var count = 0 + nonisolated(unsafe) var lastStartedIndex = 0 + try await persistence._withTransaction(actionName: nil, options: []) { _, _ in + for index in 0..] = [] + + let iterations = 5000 + nonisolated(unsafe) var count = 0 + try await persistence._withTransaction(actionName: nil, options: []) { _, _ in + for index in 0..