feat: add operation type to bridge GCD/libdispatch with structured concurrency

Author: soumyamahunt

AsyncObjects.xcodeproj/project.pbxproj

@@ -40,6 +40,7 @@ public actor AsyncEvent: AsyncObject {
     /// By default, event is initially in signaled state.
     ///
     /// - Parameter signaled: The signal state for event.
+    ///
     /// - Returns: The newly created event.
     public init(signaledInitially signaled: Bool = true) {
         self.signaled = signaled

Sources/AsyncObjects/AsyncEvent.swift

@@ -57,9 +57,10 @@ public extension AsyncObject where Self: AnyObject {
     @Sendable
     func wait(forNanoseconds duration: UInt64) async -> TaskTimeoutResult {
         return await waitForTaskCompletion(
-            task: { [weak self] in await self?.wait() },
             withTimeoutInNanoseconds: duration
-        )
+        ) { [weak self] in
+            await self?.wait()
+        }
     }
 }
 
@@ -101,10 +102,9 @@ public func waitForAll(
     _ objects: [any AsyncObject],
     forNanoseconds duration: UInt64
 ) async -> TaskTimeoutResult {
-    return await waitForTaskCompletion(
-        task: { await waitForAll(objects) },
-        withTimeoutInNanoseconds: duration
-    )
+    return await waitForTaskCompletion(withTimeoutInNanoseconds: duration) {
+        await waitForAll(objects)
+    }
 }
 
 /// Waits for multiple objects to green light task execution
@@ -165,10 +165,9 @@ public func waitForAny(
     _ objects: [any AsyncObject],
     forNanoseconds duration: UInt64
 ) async -> TaskTimeoutResult {
-    return await waitForTaskCompletion(
-        task: { await waitForAny(objects) },
-        withTimeoutInNanoseconds: duration
-    )
+    return await waitForTaskCompletion(withTimeoutInNanoseconds: duration) {
+        await waitForAny(objects)
+    }
 }
 
 /// Waits for multiple objects to green light task execution
@@ -201,8 +200,8 @@ public func waitForAny(
 ///            or timed out.
 @inlinable
 public func waitForTaskCompletion(
-    task: @escaping @Sendable () async -> Void,
-    withTimeoutInNanoseconds timeout: UInt64
+    withTimeoutInNanoseconds timeout: UInt64,
+    _ task: @escaping @Sendable () async -> Void
 ) async -> TaskTimeoutResult {
     var timedOut = true
     await withTaskGroup(of: Bool.self) { group in

Sources/AsyncObjects/AsyncObject.swift

@@ -16,3 +16,4 @@ Several synchronization primitives introduced to aid in modern swift concurrency
 ### Tasks Control
 
 - ``CancellationSource``
+- ``TaskOperation``

Sources/AsyncObjects/AsyncObjects.docc/AsyncObjects.md

@@ -58,6 +58,7 @@ public actor AsyncSemaphore: AsyncObject {
     /// Passing a value greater than zero is useful for managing a finite pool of resources, where the pool size is equal to the value.
     ///
     /// - Parameter count: The starting value for the semaphore.
+    ///
     /// - Returns: The newly created semaphore.
     public init(value count: UInt = 0) {
         self.limit = count + 1

Sources/AsyncObjects/AsyncSemaphore.swift

@@ -72,6 +72,8 @@ public actor CancellationSource {
     /// will ensure newly created cancellation source recieve cancellation event.
     ///
     /// - Parameter sources: The cancellation sources the newly created object will be linked to.
+    ///
+    /// - Returns: The newly created cancellation source.
     public convenience init(linkedWith sources: CancellationSource...) async {
         await self.init(linkedWith: sources)
     }
@@ -80,6 +82,8 @@ public actor CancellationSource {
     /// and triggers cancellation event on this object after specified timeout.
     ///
     /// - Parameter nanoseconds: The delay after which cancellation event triggered.
+    ///
+    /// - Returns: The newly created cancellation source.
     public convenience init(cancelAfterNanoseconds nanoseconds: UInt64) {
         self.init()
         Task { [weak self] in

Sources/AsyncObjects/CancellationSource.swift

@@ -0,0 +1,225 @@
+import Foundation
+import Dispatch
+
+/// An object that bridges asynchronous work under structured concurrency
+/// to Grand Central Dispatch (GCD or `libdispatch`) as `Operation`.
+///
+/// Using this object traditional `libdispatch` APIs can be used along with structured concurrency
+/// making concurrent task management flexible in terms of managing dependencies.
+///
+/// You can start the operation by adding it to an `OperationQueue`,
+/// or by manually calling the ``signal()`` or ``start()`` method.
+/// Wait for operation completion asynchronously by calling ``wait()`` method
+/// or its timeout variation ``wait(forNanoseconds:)``.
+public final class TaskOperation<R: Sendable>: Operation, AsyncObject,
+    @unchecked Sendable
+{
+    /// The dispatch queue used to synchronize data access and modifications.
+    private weak var propQueue: DispatchQueue!
+    /// The asynchronous action to perform as part of the operation..
+    private let underlyingAction: @Sendable () async throws -> R
+    /// The top-level task that executes asynchronous action provided
+    /// on behalf of the actor where operation started.
+    private var execTask: Task<R, Error>?
+
+    /// A Boolean value indicating whether the operation executes its task asynchronously.
+    ///
+    /// Always returns true, since the operation always executes its task asynchronously.
+    public override var isConcurrent: Bool { true }
+    /// A Boolean value indicating whether the operation executes its task asynchronously.
+    ///
+    /// Always returns true, since the operation always executes its task asynchronously.
+    public override var isAsynchronous: Bool { true }
+    /// A Boolean value indicating whether the operation has been cancelled.
+    ///
+    /// Returns whether the underlying top-level task is cancelled or not.
+    /// The default value of this property is `false`.
+    /// Calling the ``cancel()`` method of this object sets the value of this property to `true`.
+    public override var isCancelled: Bool { execTask?.isCancelled ?? false }
+
+    /// Private store for boolean value indicating whether the operation is currently executing.
+    private var _isExecuting: Bool = false
+    /// A Boolean value indicating whether the operation is currently executing.
+    ///
+    /// The value of this property is true if the operation is currently executing
+    /// provided asynchronous operation or false if it is not.
+    public override private(set) var isExecuting: Bool {
+        get { propQueue.sync { _isExecuting } }
+        set {
+            willChangeValue(forKey: "isExecuting")
+            propQueue.sync(flags: [.barrier]) { _isExecuting = newValue }
+            didChangeValue(forKey: "isExecuting")
+        }
+    }
+
+    /// Private store for boolean value indicating whether the operation has finished executing its task.
+    private var _isFinished: Bool = false
+    /// A Boolean value indicating whether the operation has finished executing its task.
+    ///
+    /// The value of this property is true if the operation is finished executing or cancelled
+    /// provided asynchronous operation or false if it is not.
+    public override private(set) var isFinished: Bool {
+        get { propQueue.sync { _isFinished } }
+        set {
+            willChangeValue(forKey: "isFinished")
+            propQueue.sync(flags: [.barrier]) {
+                _isFinished = newValue
+                guard newValue, !continuations.isEmpty else { return }
+                continuations.forEach { $0.value.resume() }
+                continuations = [:]
+            }
+            didChangeValue(forKey: "isFinished")
+        }
+    }
+
+    /// The result of provided asynchronous operation execution.
+    ///
+    /// Will be success if provided operation completed successfully,
+    /// or failure returned with error.
+    public var result: Result<R, Error> {
+        get async { (await execTask?.result) ?? .failure(CancellationError()) }
+    }
+
+    /// Creates a new operation that executes the provided throwing asynchronous task.
+    ///
+    /// The provided dispatch queue is used to syncronize operation property access and modifications
+    /// and prevent data races.
+    ///
+    /// - Parameters:
+    ///   - queue: The dispatch queue to be used to synchronize data access and modifications.
+    ///   - operation: The throwing asynchronous operation to execute.
+    ///
+    /// - Returns: The newly created asynchronous operation.
+    public init(
+        queue: DispatchQueue,
+        operation: @escaping @Sendable () async throws -> R
+    ) {
+        self.propQueue = queue
+        self.underlyingAction = operation
+        super.init()
+    }
+
+    /// Creates a new operation that executes the provided nonthrowing asynchronous task.
+    ///
+    /// The provided dispatch queue is used to syncronize operation property access and modifications
+    /// and prevent data races.
+    ///
+    /// - Parameters:
+    ///   - queue: The dispatch queue to be used to synchronize data access and modifications.
+    ///   - operation: The nonthrowing asynchronous operation to execute.
+    ///
+    /// - Returns: The newly created asynchronous operation.
+    public init(
+        queue: DispatchQueue,
+        operation: @escaping @Sendable () async -> R
+    ) {
+        self.propQueue = queue
+        self.underlyingAction = operation
+        super.init()
+    }
+
+    /// Begins the execution of the operation.
+    ///
+    /// Updates the execution state of the operation and
+    /// runs the given operation asynchronously
+    /// as part of a new top-level task on behalf of the current actor.
+    public override func start() {
+        guard !self.isFinished else { return }
+        isFinished = false
+        isExecuting = true
+        main()
+    }
+
+    /// Performs the provided asynchronous task.
+    ///
+    /// Runs the given operation asynchronously
+    /// as part of a new top-level task on behalf of the current actor.
+    public override func main() {
+        guard isExecuting, execTask == nil else { return }
+        execTask = Task { [weak self] in
+            guard let self = self else { throw CancellationError() }
+            let result = try await underlyingAction()
+            self.finish()
+            return result
+        }
+    }
+
+    /// Advises the operation object that it should stop executing its task.
+    ///
+    /// Initiates cooperative cancellation for provided asynchronous operation
+    /// and moves to finshed state.
+    ///
+    /// Calling this method on a task that doesn’t support cancellation has no effect.
+    /// Likewise, if the task has already run past the last point where it would stop early,
+    /// calling this method has no effect.
+    public override func cancel() {
+        execTask?.cancel()
+        finish()
+    }
+
+    /// Moves this operation to finished state.
+    ///
+    /// Must be called either when operation completes or cancelled.
+    @inline(__always)
+    private func finish() {
+        isExecuting = false
+        isFinished = true
+    }
+
+    // MARK: AsyncObject Impl
+    /// The suspended tasks continuation type.
+    private typealias Continuation = UnsafeContinuation<Void, Error>
+    /// The continuations stored with an associated key for all the suspended task that are waitig for opearation completion.
+    private var continuations: [UUID: Continuation] = [:]
+
+    /// Add continuation with the provided key in `continuations` map.
+    ///
+    /// - Parameters:
+    ///   - continuation: The `continuation` to add.
+    ///   - key: The key in the map.
+    @inline(__always)
+    private func addContinuation(
+        _ continuation: Continuation,
+        withKey key: UUID
+    ) {
+        continuations[key] = continuation
+    }
+
+    /// Remove continuation associated with provided key
+    /// from `continuations` map.
+    ///
+    /// - Parameter key: The key in the map.
+    @inline(__always)
+    private func removeContinuation(withKey key: UUID) {
+        continuations.removeValue(forKey: key)
+    }
+
+    /// Starts operation asynchronously
+    /// as part of a new top-level task on behalf of the current actor.
+    @Sendable
+    public func signal() {
+        self.start()
+    }
+
+    /// Waits for opearation to complete successfully or cancelled.
+    ///
+    /// Only waits asynchronously, if opearation is executing,
+    /// until it is completed or cancelled.
+    @Sendable
+    public func wait() async {
+        guard !isFinished else { return }
+        let key = UUID()
+        try? await withUnsafeThrowingContinuationCancellationHandler(
+            handler: { [weak self] (continuation: Continuation) in
+                Task { [weak self] in
+                    self?.removeContinuation(withKey: key)
+                }
+            },
+            { [weak self] (continuation: Continuation) in
+                Task { [weak self] in
+                    self?.addContinuation(continuation, withKey: key)
+                }
+            }
+        )
+    }
+}

Sources/AsyncObjects/TaskOperation.swift

@@ -12,10 +12,7 @@ class AsyncEventTests: XCTestCase {
             try await Task.sleep(nanoseconds: interval)
             await event.signal()
         }
-        await checkExecInterval(
-            for: { await event.wait() },
-            durationInSeconds: seconds
-        )
+        await checkExecInterval(durationInSeconds: seconds, for: event.wait)
     }
 
     func testEventWait() async throws {
@@ -37,12 +34,9 @@ class AsyncEventTests: XCTestCase {
     func testEventWaitWithTimeout() async throws {
         let event = AsyncEvent(signaledInitially: false)
         var result: TaskTimeoutResult = .success
-        await checkExecInterval(
-            for: {
-                result = await event.wait(forNanoseconds: UInt64(4E9))
-            },
-            durationInSeconds: 4
-        )
+        await checkExecInterval(durationInSeconds: 4) {
+            result = await event.wait(forNanoseconds: UInt64(4E9))
+        }
         XCTAssertEqual(result, .timedOut)
     }
 
@@ -53,24 +47,18 @@ class AsyncEventTests: XCTestCase {
             try await Task.sleep(nanoseconds: UInt64(5E9))
             await event.signal()
         }
-        await checkExecInterval(
-            for: {
-                result = await event.wait(forNanoseconds: UInt64(10E9))
-            },
-            durationInSeconds: 5
-        )
+        await checkExecInterval(durationInSeconds: 5) {
+            result = await event.wait(forNanoseconds: UInt64(10E9))
+        }
         XCTAssertEqual(result, .success)
     }
 
     func testReleasedEventWaitSuccessWithoutTimeout() async throws {
         let event = AsyncEvent()
         var result: TaskTimeoutResult = .timedOut
-        await checkExecInterval(
-            for: {
-                result = await event.wait(forNanoseconds: UInt64(10E9))
-            },
-            durationInSeconds: 0
-        )
+        await checkExecInterval(durationInSeconds: 0) {
+            result = await event.wait(forNanoseconds: UInt64(10E9))
+        }
         XCTAssertEqual(result, .success)
     }
 }