avoid actor hopping on SubmissionGroup

Author: lhoward

Sources/IORing/IORing.swift

@@ -336,9 +336,9 @@ public actor IORing: CustomStringConvertible {
   func withSubmissionGroup<T: Sendable>(_ body: (
     SubmissionGroup<T>
   ) async throws -> ()) async throws -> [T] {
-    let submissionGroup = try await SubmissionGroup<T>(ring: self)
+    let submissionGroup = try SubmissionGroup<T>(ring: self)
     try await body(submissionGroup)
-    return try await submissionGroup.finish()
+    return try await submissionGroup.finish(ring: self)
   }
 
   fileprivate func prepareAndSubmit<T: Sendable>(

Sources/IORing/Submission.swift

@@ -201,7 +201,7 @@ final class SingleshotSubmission<T: Sendable>: Submission<T>, @unchecked Sendabl
       handler: handler
     )
     if let group {
-      await group.enqueue(submission: self)
+      group.enqueue(submission: self, ring: ring)
     }
   }
 
@@ -211,7 +211,7 @@ final class SingleshotSubmission<T: Sendable>: Submission<T>, @unchecked Sendabl
         // guaranteed to run immediately
         self.continuation = continuation
         if group != nil {
-          Task { await ready() }
+          ready()
         } else {
           _ = try? ring.submit()
         }

Sources/IORing/SubmissionGroup.swift

@@ -19,7 +19,7 @@ import AsyncQueue
 import Glibc
 
 extension SingleshotSubmission {
-  func enqueue() {
+  func enqueue(ring: isolated IORing) {
     guard let group else { return }
     Task {
       do {
@@ -31,34 +31,30 @@ extension SingleshotSubmission {
     }
   }
 
-  func ready() async {
-    await group?.readinessChannel.send(())
+  func ready() {
+    Task { await group?.readinessChannel.send(()) }
   }
 }
 
-actor SubmissionGroup<T: Sendable> {
+final class SubmissionGroup<T: Sendable>: Sendable {
   private let ring: IORing
-  private let queue = ActorQueue<SubmissionGroup>()
-  private var submissions = [SingleshotSubmission<T>]()
+  private nonisolated(unsafe) var submissions = [SingleshotSubmission<T>]()
 
   fileprivate let readinessChannel = AsyncChannel<()>()
   fileprivate let resultChannel = AsyncThrowingChannel<T, Error>()
 
-  init(ring: IORing) async throws {
+  init(ring: isolated IORing) throws {
     self.ring = ring
-    queue.adoptExecutionContext(of: self)
   }
 
   ///
   /// Asynchronously enqueues an submission. Submission must call `ready()` when
   /// its continuation is registered in the SQE `user_data` otherwise the group
   /// will never be submitted.
   ///
-  func enqueue(submission: SingleshotSubmission<T>) {
+  func enqueue(submission: SingleshotSubmission<T>, ring: isolated IORing) {
     submissions.append(submission)
-    Task(on: queue) { _ in
-      submission.enqueue()
-    }
+    submission.enqueue(ring: ring)
   }
 
   private func allReady() async {
@@ -75,15 +71,13 @@ actor SubmissionGroup<T: Sendable> {
   ///
   /// Completing the submission group involves the following:
   ///
-  /// - Await all submissions to be scheduled on queue
   /// - Wait for all submissions to have continuations registered
   /// - Submit SQEs to I/O ring
   /// - Collect results from results channel
   ///
-  func finish() async throws -> [T] {
-    await Task(on: queue) { _ in }.value
+  func finish(ring: isolated IORing) async throws -> [T] {
     await allReady()
-    try await ring.submit()
+    try ring.submit()
     readinessChannel.finish()
     return try await allComplete()
   }