fix: Fix a variety of issues found in DispatchAsync while implementing new tests.

Author: scottmarchant

Sources/DispatchAsync/AsyncSemaphore.swift

@@ -12,7 +12,7 @@
 //
 //===----------------------------------------------------------------------===//
 
-/// Provides a semaphore implantation in `async` context, with a safe wait method. Provides easy safer replacement
+/// Provides a semaphore implantation in `async` context, with a safe wait method. Provides easy safe replacement
 /// for DispatchSemaphore usage.
 @available(macOS 10.15, *)
 actor AsyncSemaphore {
@@ -25,6 +25,7 @@ actor AsyncSemaphore {
 
     func wait() async {
         value -= 1
+
         if value >= 0 { return }
         await withCheckedContinuation {
             waiters.append($0)

Sources/DispatchAsync/DispatchGroup.swift

@@ -23,34 +23,29 @@
 /// for more details,
 @available(macOS 10.15, *)
 public class DispatchGroup: @unchecked Sendable {
-    /// Used to ensure FIFO access to the enter and leave calls
-    @globalActor
-    private actor DispatchGroupEntryActor: GlobalActor {
-        static let shared = DispatchGroupEntryActor()
-    }
-
-    private let group = AsyncGroup()
+    private let group = _AsyncGroup()
+    private let queue = FIFOQueue()
 
     public func enter() {
-        Task { @DispatchGroupEntryActor [] in
-            // ^--- Ensures serial FIFO entrance/exit into the group
+        queue.enqueue { [weak self] in
+            guard let self else { return }
             await group.enter()
         }
     }
 
     public func leave() {
-        Task { @DispatchGroupEntryActor [] in
-            // ^--- Ensures serial FIFO entrance/exit into the group
+        queue.enqueue { [weak self] in
+            guard let self else { return }
             await group.leave()
         }
     }
 
-    public func notify(queue: DispatchQueue, execute work: @escaping @Sendable @convention(block) () -> Void) {
-        Task { @DispatchGroupEntryActor [] in
-            // ^--- Ensures serial FIFO entrance/exit into the group
+    public func notify(queue notificationQueue: DispatchQueue, execute work: @escaping @Sendable @convention(block) () -> Void) {
+        queue.enqueue { [weak self] in
+            guard let self else { return }
             await group.notify {
                 await withCheckedContinuation { continuation in
-                    queue.async {
+                    notificationQueue.async {
                         work()
                         continuation.resume()
                     }
@@ -60,7 +55,22 @@ public class DispatchGroup: @unchecked Sendable {
     }
 
     func wait() async {
-        await group.wait()
+        await withCheckedContinuation { continuation in
+            queue.enqueue { [weak self] in
+                guard let self else { return }
+                // NOTE: We use a task for the wait, because
+                // otherwise the queue won't execute any more
+                // tasks until the wait finishes, which is not the
+                // behavior we want here. We want to enqueue the wait
+                // in FIFO call order, but then we want to allow the wait
+                // to be non-blocking for the queue until the last leave
+                // is called on the group.
+                Task {
+                    await group.wait()
+                    continuation.resume()
+                }
+            }
+        }
     }
 
     public init() {}
@@ -69,10 +79,8 @@ public class DispatchGroup: @unchecked Sendable {
 // MARK: - Private Interface for Async Usage -
 
 @available(macOS 10.15, *)
-fileprivate actor AsyncGroup {
+fileprivate actor _AsyncGroup {
     private var taskCount = 0
-    private var continuation: CheckedContinuation<Void, Never>?
-    private var isWaiting = false
     private var notifyHandlers: [@Sendable () async -> Void] = []
 
     func enter() {
@@ -100,30 +108,22 @@ fileprivate actor AsyncGroup {
             return
         }
 
-        isWaiting = true
-
         await withCheckedContinuation { (continuation: CheckedContinuation<Void, Never>) in
-            self.continuation = continuation
+            notify {
+                continuation.resume()
+            }
             checkCompletion()
         }
     }
 
     private func checkCompletion() {
-        if taskCount <= 0 {
-            if isWaiting {
-                continuation?.resume()
-                continuation = nil
-                isWaiting = false
-            }
+        if taskCount <= 0, !notifyHandlers.isEmpty {
+            let handlers = notifyHandlers
+            notifyHandlers.removeAll()
 
-            if !notifyHandlers.isEmpty {
-                let handlers = notifyHandlers
-                notifyHandlers.removeAll()
-
-                for handler in handlers {
-                    Task {
-                        await handler()
-                    }
+            for handler in handlers {
+                Task {
+                    await handler()
                 }
             }
         }

Sources/DispatchAsync/DispatchQueue.swift

@@ -23,17 +23,26 @@
 public class DispatchQueue: @unchecked Sendable {
     public static let main = DispatchQueue(isMain: true)
 
-    private static let _global = DispatchQueue()
+    private static let _global = DispatchQueue(attributes: .concurrent)
     public static func global() -> DispatchQueue {
         Self._global
     }
 
     public enum Attributes {
         case concurrent
+
+        fileprivate var isConcurrent: Bool {
+            switch self {
+            case .concurrent:
+                return true
+            }
+        }
     }
 
     private let targetQueue: DispatchQueue?
 
+    private let serialQueue = FIFOQueue()
+
     /// Indicates whether calling context is running from the main DispatchQueue instance, or some other DispatchQueue instance.
     @TaskLocal public static var isMain = false
 
@@ -56,6 +65,10 @@ public class DispatchQueue: @unchecked Sendable {
         attributes: DispatchQueue.Attributes? = nil,
         target: DispatchQueue? = nil
     ) {
+        if isMain, attributes == .concurrent {
+            assertionFailure("Should never create a concurrent main queue. Main queue should always be serial.")
+        }
+
         self.isMain = isMain
         self.label = label
         self.attributes = attributes
@@ -77,10 +90,60 @@ public class DispatchQueue: @unchecked Sendable {
                     }
                 }
             } else {
-                Task {
-                    work()
+                if attributes?.isConcurrent == true {
+                    Task { // FIFO is not important for concurrent queues, using global task executor here
+                        work()
+                    }
+                } else {
+                    // We don't need to use a task for enqueing work to a non-main serial queue
+                    // because the enqueue process is very light-weight, and it is important to
+                    // preserve FIFO entry into the queue as much as possible.
+                    serialQueue.enqueue(work)
                 }
             }
         }
     }
 }
+
+/// A tiny FIFO job runner that executes each submitted async closure
+/// strictly in the order it was enqueued.
+///
+/// This is NOT part of the original GCD API. So it is intentionally kept
+/// internal for now.
+@available(macOS 10.15, *)
+actor FIFOQueue {
+    /// A single item in the stream, which is a block of work that can be completed.
+    typealias WorkItem = @Sendable () async -> Void
+
+    /// The stream’s continuation; lives inside the actor so nobody
+    /// else can yield into it.
+    private let continuation: AsyncStream<WorkItem>.Continuation
+
+    /// Spin up the stream and the single draining task.
+    init(bufferingPolicy: AsyncStream<WorkItem>.Continuation.BufferingPolicy = .unbounded) {
+        let stream: AsyncStream<WorkItem>
+        (stream, self.continuation) = AsyncStream.makeStream(of: WorkItem.self, bufferingPolicy: bufferingPolicy)
+
+        // Dedicated worker that processes work items one-by-one.
+        Task {
+            for await work in stream {
+                // Run each job in order, allowing suspension, and awaiting full
+                // completion, before running the next work item
+                await work()
+            }
+        }
+    }
+
+    /// Enqueue a new unit of work.
+    @discardableResult
+    nonisolated
+    func enqueue(_ workItem: @escaping WorkItem) -> AsyncStream<WorkItem>.Continuation.YieldResult {
+        // Never suspends, preserves order
+        continuation.yield(workItem)
+    }
+
+    deinit {
+        // Clean shutdown on deinit
+        continuation.finish()
+    }
+}

Sources/DispatchAsync/DispatchTime.swift

@@ -13,30 +13,39 @@
 //===----------------------------------------------------------------------===//
 
 @available(macOS 13, *)
-public typealias DispatchTime = ContinuousClock.Instant
+public struct DispatchTime {
+    private let instant: ContinuousClock.Instant
 
-/// The very first time someone tries to reference a `uptimeNanoseconds` or a similar
-/// function that references a beginning point, this variable will be initialized as a beginning
-/// reference point. This guarantees that all calls to `uptimeNanoseconds` or similar
-/// will be 0 or greater.
-///
-/// By design, it is not possible to related `ContinuousClock.Instant` to
-/// `ProcessInfo.processInfo.systemUptime`, and even if one devised such
-/// a mechanism, it would open the door for fingerprinting. It's best to let the concept
-/// of uptime be relative to previous uptime calls.
-@available(macOS 13, *)
-private let uptimeBeginning: DispatchTime = DispatchTime.now()
+    /// The very first time someone intializes a DispatchTime instance, we
+    /// reference this static let, causing it to be initialized.
+    ///
+    /// This is the closest we can get to snapshotting the start time of the running
+    /// executable, without using OS-specific calls. We want
+    /// to avoid OS-specific calls to maximize portability.
+    ///
+    /// To keep this robust, we initialize `self.durationSinceBeginning`
+    /// to this value using a default value, which is guaranteed to run before any
+    /// initializers run. This guarantees that uptimeBeginning will be the very
+    /// first
+    @available(macOS 13, *)
+    private static let uptimeBeginning: ContinuousClock.Instant = ContinuousClock.Instant.now
+
+    /// See documentation for ``uptimeBeginning``. We intentionally
+    /// use this to guarantee a capture of `now` to uptimeBeginnin BEFORE
+    /// any DispatchTime instances are initialized.
+    private let durationSinceUptime = uptimeBeginning.duration(to: ContinuousClock.Instant.now)
+
+    public init() {
+        self.instant = ContinuousClock.Instant.now
+    }
 
-@available(macOS 13, *)
-extension DispatchTime {
     public static func now() -> DispatchTime {
-        now
+        DispatchTime()
     }
 
     public var uptimeNanoseconds: UInt64 {
-        let beginning = uptimeBeginning
-        let rightNow = DispatchTime.now()
-        let uptimeDuration: Int64 = beginning.duration(to: rightNow).nanosecondsClamped
+        let beginning = DispatchTime.uptimeBeginning
+        let uptimeDuration: Int64 = beginning.duration(to: self.instant).nanosecondsClamped
         guard uptimeDuration >= 0 else {
             assertionFailure("It shouldn't be possible to get a negative duration since uptimeBeginning.")
             return 0