[Concurrency] Improve in order synchronous enqueue of startSynchronously

Previously there was still a sneaky hop which caused ordering issues.
This introduced a specific test startSynchronously_order which checks
that the task enqueues indeed are "immediate" and cleans up how we
handle this.

This also prepares for the being discussed in SE review direction of
this API that it SHOULD be ALLOWED to actually hop and NOT be
synchronous at all IF the isolation is specified on the closure and is
DIFFERENT than the callers dynamic isolation.

This effectively implements "synchronously run right now if dynamically
on the exact isolation as requested by the closure; otherwise enqueue
the task as usual".

resolves rdar://149284186
cc @drexin

Author: ktoso

include/swift/Runtime/Concurrency.h

@@ -1069,7 +1069,7 @@ SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
 void swift_task_startOnMainActor(AsyncTask* job);
 
 SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
-void swift_task_startSynchronously(AsyncTask* job);
+void swift_task_startSynchronously(AsyncTask* job, SerialExecutorRef targetExecutor);
 
 /// Donate this thread to the global executor until either the
 /// given condition returns true or we've run out of cooperative

stdlib/public/CompatibilityOverride/CompatibilityOverrideConcurrency.def

@@ -441,7 +441,8 @@ OVERRIDE_TASK(task_startOnMainActor, void,
 // In ACTOR since we need ExecutorTracking info
 OVERRIDE_ACTOR(task_startSynchronously, void,
                SWIFT_EXPORT_FROM(swift_Concurrency), SWIFT_CC(swift),
-               swift::, (AsyncTask *task), (task))
+               swift::, (AsyncTask *task, SerialExecutorRef targetExecutor),
+               (task, targetExecutor))
 
 #undef OVERRIDE
 #undef OVERRIDE_ACTOR

stdlib/public/Concurrency/Actor.cpp

@@ -2531,22 +2531,31 @@ static void swift_task_switchImpl(SWIFT_ASYNC_CONTEXT AsyncContext *resumeContex
 }
 
 SWIFT_CC(swift)
-static void swift_task_startSynchronouslyImpl(AsyncTask* task) {
+static void
+swift_task_startSynchronouslyImpl(AsyncTask *task,
+                                  SerialExecutorRef targetExecutor) {
   swift_retain(task);
-
-  auto currentTracking = ExecutorTrackingInfo::current();
-  if (currentTracking) {
-    auto currentExecutor = currentTracking->getActiveExecutor();
-    AsyncTask * originalTask = _swift_task_clearCurrent();
-
-    swift_job_run(task, currentExecutor);
-    _swift_task_setCurrent(originalTask);
+  if (targetExecutor.isGeneric()) {
+  // If the target is generic, it means that the closure did not specify
+  // an isolation explicitly. According to the "start synchronously" rules,
+  // we should therefore ignore the global and just start running on the
+  // caller immediately.
+  SerialExecutorRef executor = SerialExecutorRef::forSynchronousStart();
+
+  auto originalTask = ActiveTask::swap(task);
+  swift_job_run(task, executor);
+  _swift_task_setCurrent(originalTask);
   } else {
-    auto originalTask = ActiveTask::swap(task);
-    assert(!originalTask);
+    assert(swift_task_isCurrentExecutor(targetExecutor) &&
+           "startSynchronously must only be invoked when it is correctly in "
+           "the same isolation already, but wasn't!");
+
+    // We can run synchronously, we're on the expected executor so running in
+    // the caller context is going to be in the same context as the requested
+    // "caller" context.
+    AsyncTask *originalTask = _swift_task_clearCurrent();
 
-    SerialExecutorRef executor = SerialExecutorRef::forSynchronousStart();
-     swift_job_run(task, executor);
+    swift_job_run(task, targetExecutor);
     _swift_task_setCurrent(originalTask);
   }
 }

stdlib/public/Concurrency/Task+startSynchronously.swift.gyb

@@ -59,14 +59,30 @@ extension Task where Failure == ${FAILURE_TYPE} {
   public static func startSynchronously(
     name: String? = nil,
     priority: TaskPriority? = nil,
-    @_inheritActorContext @_implicitSelfCapture _ operation: __owned sending @escaping () async throws -> Success
+    % # NOTE: This closure cannot be 'sending' because we'll trigger ' pattern that the region based isolation checker does not understand how to check'
+    % #       In this case: `func syncOnMyGlobalActor() { Task.startSynchronously { @MyGlobalActor in } }`
+    @_implicitSelfCapture _ operation: __owned @isolated(any) @escaping () async throws -> Success
   ) -> Task<Success, ${FAILURE_TYPE}> {
+
+    let builtinSerialExecutor =
+      unsafe Builtin.extractFunctionIsolation(operation)?.unownedExecutor.executor
+
+    // Determine if we're switching isolation dynamically.
+    // If not, we can run the task synchronously and therefore MUST NOT "enqueue" it.
+    let flagsMustNotCrash: UInt64 = 0
+    let canRunSynchronously: Bool =
+    if let builtinSerialExecutor {
+      _taskIsCurrentExecutor(executor: builtinSerialExecutor, flags: flagsMustNotCrash)
+    } else {
+      true // if there is not target executor, we can run synchronously
+    }
+
     let flags = taskCreateFlags(
       priority: priority,
       isChildTask: false,
       copyTaskLocals: true,
       inheritContext: true,
-      enqueueJob: false, // don't enqueue, we'll run it manually
+      enqueueJob: !canRunSynchronously,
       addPendingGroupTaskUnconditionally: false,
       isDiscardingTask: false,
       isSynchronousStart: true
@@ -79,6 +95,7 @@ extension Task where Failure == ${FAILURE_TYPE} {
         unsafe name.utf8CString.withUnsafeBufferPointer { nameBytes in
           Builtin.createTask(
             flags: flags,
+            initialSerialExecutor: builtinSerialExecutor,
             taskName: nameBytes.baseAddress!._rawValue,
             operation: operation).0
         }
@@ -91,7 +108,9 @@ extension Task where Failure == ${FAILURE_TYPE} {
         operation: operation).0
     }
 
-    _startTaskSynchronously(task!)
+    if canRunSynchronously {
+      _startTaskSynchronously(task!, targetExecutor: builtinSerialExecutor)
+    }
     return Task<Success, ${FAILURE_TYPE}>(task!)
   }
 }
@@ -161,7 +180,7 @@ extension ${GROUP_TYPE} {
   public func ${METHOD_NAME}( // in ${GROUP_TYPE}
     name: String? = nil,
     priority: TaskPriority? = nil,
-    operation: sending @escaping () async ${THROWS}-> ${RESULT_TYPE}
+    @_inheritActorContext @_implicitSelfCapture operation: sending @isolated(any) @escaping () async ${THROWS}-> ${RESULT_TYPE}
   ) {
     let flags = taskCreateFlags(
       priority: priority,
@@ -174,13 +193,18 @@ extension ${GROUP_TYPE} {
       isSynchronousStart: true
     )
 
+    // Create the asynchronous task.
+    let builtinSerialExecutor =
+      unsafe Builtin.extractFunctionIsolation(operation)?.unownedExecutor.executor
+
     // Create the task in this group.
     let (task, _) = Builtin.createTask(
       flags: flags,
+      initialSerialExecutor: builtinSerialExecutor,
       taskGroup: self._group,
       operation: operation
     )
-    _startTaskSynchronously(task)
+    _startTaskSynchronously(task, targetExecutor: builtinSerialExecutor)
   }
 }
 %   end # METHOD_NAMES
@@ -241,4 +265,4 @@ extension Task where Failure == ${FAILURE_TYPE} {
 internal func _startTaskOnMainActor(_ task: Builtin.NativeObject)
 
 @_silgen_name("swift_task_startSynchronously")
-internal func _startTaskSynchronously(_ task: Builtin.NativeObject)
+internal func _startTaskSynchronously(_ task: Builtin.NativeObject, targetExecutor: Builtin.Executor?)

stdlib/public/Concurrency/Task.swift

@@ -807,8 +807,7 @@ extension Task where Failure == Error {
       unsafe Builtin.extractFunctionIsolation(operation)?.unownedExecutor.executor
 
     let (task, _) = Builtin.createTask(flags: flags,
-                                       initialSerialExecutor:
-                                         builtinSerialExecutor,
+                                       initialSerialExecutor: builtinSerialExecutor,
                                        operation: operation)
 
     self._task = task

test/Concurrency/Runtime/startSynchronously.swift

@@ -1,4 +1,3 @@
-// REQUIRES: rdar145735542
 // RUN: %empty-directory(%t)
 // RUN: %target-build-swift -Xfrontend -disable-availability-checking %s %import-libdispatch -swift-version 6 -o %t/a.out
 // RUN: %target-codesign %t/a.out
@@ -62,6 +61,42 @@ extension ThreadID: @unchecked Sendable {}
 @globalActor
 actor MyGlobalActor {
   static let shared: MyGlobalActor = MyGlobalActor()
+
+  @MyGlobalActor
+  static func test() {}
+}
+
+final class NaiveQueueExecutor: SerialExecutor {
+  let queue: DispatchQueue
+
+  init(queue: DispatchQueue) {
+    self.queue = queue
+  }
+
+  public func enqueue(_ job: consuming ExecutorJob) {
+    let unowned = UnownedJob(job)
+    print("NaiveQueueExecutor(\(self.queue.label)) enqueue [thread:\(getCurrentThreadID())]")
+    queue.async {
+      unowned.runSynchronously(on: self.asUnownedSerialExecutor())
+    }
+  }
+}
+
+@globalActor
+actor DifferentGlobalActor {
+  static let queue = DispatchQueue(label: "DifferentGlobalActor-queue")
+  let executor: NaiveQueueExecutor
+  nonisolated let unownedExecutor: UnownedSerialExecutor
+
+  init() {
+    self.executor = NaiveQueueExecutor(queue: DifferentGlobalActor.queue)
+    self.unownedExecutor = executor.asUnownedSerialExecutor()
+  }
+
+  static let shared: DifferentGlobalActor = DifferentGlobalActor()
+
+  @DifferentGlobalActor
+  static func test() {}
 }
 
 // Test on all platforms
@@ -94,6 +129,49 @@ func syncOnMyGlobalActor() -> [Task<Void, Never>] {
   return [t1, tt]
 }
 
+func syncOnMyGlobalActorHopToDifferentActor() -> [Task<Void, Never>] {
+  MyGlobalActor.shared.preconditionIsolated("Should be executing on the global actor here")
+  print("Confirmed to be on @MyGlobalActor")
+
+  // This task must be guaranteed to happen AFTER 'tt' because we are already on this actor
+  // so this enqueue must happen after we give up the actor.
+  print("schedule Task { @DifferentGlobalActor }, before startSynchronously [thread:\(getCurrentThreadID())] @ :\(#line)")
+  let t1 = Task { @DifferentGlobalActor in
+    print("inside Task { @DifferentGlobalActor } [thread:\(getCurrentThreadID())] @ :\(#line)")
+    DifferentGlobalActor.shared.preconditionIsolated("Expected Task{} to be on DifferentGlobalActor")
+  }
+
+  print("before startSynchronously [thread:\(getCurrentThreadID())] @ :\(#line)")
+  let outerTID = getCurrentThreadID()
+  let tt = Task.startSynchronously { @DifferentGlobalActor in
+    let innerTID = getCurrentThreadID()
+    print("inside startSynchronously, outer thread = \(outerTID)")
+    print("inside startSynchronously, inner thread = \(innerTID)")
+    if (compareThreadIDs(outerTID, .equal, innerTID)) {
+      // This case specifically is NOT synchronously run because we specified a different isolation for the closure
+      // and FORCED a hop to the DifferentGlobalActor executor.
+      print("ERROR! Outer Thread ID must NOT equal Thread ID inside runSynchronously synchronous part!")
+    }
+    // We crucially need to see this task be enqueued on the different global actor,
+    // so it did not execute "synchronously" after all - it had to hop to the other actor.
+    dispatchPrecondition(condition: .onQueue(DifferentGlobalActor.queue))
+    DifferentGlobalActor.shared.preconditionIsolated("Expected Task.startSynchronously { @DifferentGlobalActor in } to be on DifferentGlobalActor")
+
+    print("inside startSynchronously, sleep now [thread:\(getCurrentThreadID())] @ :\(#line)")
+    _ = try? await Task.sleep(for: .milliseconds(100))
+
+    print("inside startSynchronously, after sleep [thread:\(getCurrentThreadID())] @ :\(#line)")
+    dispatchPrecondition(condition: .onQueue(DifferentGlobalActor.queue))
+    DifferentGlobalActor.shared.preconditionIsolated("Expected Task.startSynchronously { @DifferentGlobalActor in } to be on DifferentGlobalActor")
+
+    // do something here
+    await MyGlobalActor.test()
+    DifferentGlobalActor.test()
+  }
+
+  return [t1, tt]
+}
+
 func syncOnNonTaskThread(synchronousTask behavior: SynchronousTaskBehavior) {
   let sem1 = DispatchSemaphore(value: 0)
   let sem2 = DispatchSemaphore(value: 0)
@@ -162,6 +240,33 @@ await Task { @MyGlobalActor in
 // resume on some other thread
 // CHECK: after sleep, inside startSynchronously
 
+print("\n\n==== ------------------------------------------------------------------")
+print("syncOnMyGlobalActorHopToDifferentActor()")
+
+await Task { @MyGlobalActor in
+  MyGlobalActor.shared.preconditionIsolated("Should be executing on the global actor here")
+  for t in syncOnMyGlobalActorHopToDifferentActor() {
+    await t.value
+  }
+}.value
+
+// Assertion Notes: We expect the task to be on the specified queue as we force the Task.startSynchronously
+// task to enqueue on the DifferentGlobalActor, however we CANNOT use threads to verify this behavior,
+// because dispatch may still pull tricks and reuse threads. We can only verify that we're on the right
+// queue, and that the `enqueue` calls on the target executor happen when we expect them to.
+//
+// CHECK: syncOnMyGlobalActorHopToDifferentActor()
+// CHECK: Confirmed to be on @MyGlobalActor
+// CHECK: before startSynchronously
+
+// This IS actually enqueueing on the target actor (not synchronous), as expected:
+// CHECK: NaiveQueueExecutor(DifferentGlobalActor-queue) enqueue
+// CHECK: inside startSynchronously, sleep now
+
+// After the sleep we get back onto the specified executor as expected
+// CHECK: NaiveQueueExecutor(DifferentGlobalActor-queue) enqueue
+// CHECK: inside startSynchronously, after sleep
+
 print("\n\n==== ------------------------------------------------------------------")
 var behavior: SynchronousTaskBehavior = .suspend
 print("syncOnNonTaskThread(synchronousTask: \(behavior))")
@@ -347,23 +452,6 @@ callActorFromStartSynchronousTask(recipient: .recipientOnQueue(RecipientOnQueue(
 // CHECK-NOT: ERROR!
 // CHECK: inside startSynchronously, done
 
-final class NaiveQueueExecutor: SerialExecutor {
-  let queue: DispatchQueue
-
-  init(queue: DispatchQueue) {
-    self.queue = queue
-  }
-
-  public func enqueue(_ job: consuming ExecutorJob) {
-    let unowned = UnownedJob(job)
-    print("NaiveQueueExecutor(\(self.queue.label)) enqueue... [thread:\(getCurrentThreadID())]")
-    queue.async {
-    print("NaiveQueueExecutor(\(self.queue.label)) enqueue: run [thread:\(getCurrentThreadID())]")
-      unowned.runSynchronously(on: self.asUnownedSerialExecutor())
-    }
-  }
-}
-
 actor RecipientOnQueue {
   let executor: NaiveQueueExecutor
   nonisolated let unownedExecutor: UnownedSerialExecutor

test/Concurrency/Runtime/startSynchronously_order.swift

@@ -0,0 +1,64 @@
+// RUN: %empty-directory(%t)
+// RUN: %target-build-swift -Xfrontend -disable-availability-checking %s %import-libdispatch -swift-version 6 -o %t/a.out
+// RUN: %target-codesign %t/a.out
+// RUN: %target-run %t/a.out | %FileCheck %s
+
+// REQUIRES: executable_test
+// REQUIRES: concurrency
+// REQUIRES: concurrency_runtime
+
+// UNSUPPORTED: back_deployment_runtime
+// UNSUPPORTED: back_deploy_concurrency
+// UNSUPPORTED: use_os_stdlib
+// UNSUPPORTED: freestanding
+
+import _Concurrency
+
+let max = 1000
+
+func bar(x: Int, cc: CheckedContinuation<Void, Never>) {
+  Task.startSynchronously {
+    print("Task \(x) started")
+    try! await Task.sleep(nanoseconds: 10000)
+    if (x == max) {
+      cc.resume()
+    }
+  }
+}
+
+await withCheckedContinuation { (cc: CheckedContinuation<Void, Never>) in
+  for i in 1...max {
+    bar(x: i, cc: cc)
+  }
+}
+
+// CHECK: Task 1 started
+// CHECK: Task 2 started
+// CHECK: Task 3 started
+// CHECK: Task 4 started
+// CHECK: Task 5 started
+// CHECK: Task 6 started
+// CHECK: Task 7 started
+// CHECK: Task 8 started
+// CHECK: Task 9 started
+// CHECK: Task 10 started
+// CHECK: Task 11 started
+// CHECK: Task 12 started
+// CHECK: Task 13 started
+// CHECK: Task 14 started
+// CHECK: Task 15 started
+// CHECK: Task 16 started
+// CHECK: Task 17 started
+// CHECK: Task 18 started
+// CHECK: Task 19 started
+// CHECK: Task 20 started
+// CHECK: Task 21 started
+// CHECK: Task 22 started
+// CHECK: Task 23 started
+// CHECK: Task 24 started
+// CHECK: Task 25 started
+// CHECK: Task 26 started
+// CHECK: Task 27 started
+// CHECK: Task 28 started
+// CHECK: Task 29 started
+// CHECK: Task 30 started