[Concurrency] StartSynchronously special executor to avoid switching

Author: ktoso

include/swift/ABI/Executor.h

@@ -77,6 +77,8 @@ class SerialExecutorRef {
     /// Executor that may need to participate in complex "same context" checks,
     /// by invoking `isSameExclusiveExecutionContext` when comparing execution contexts.
     ComplexEquality = 0b01,
+    ///
+    StartSynchronously = 0b10,
   };
 
   static_assert(static_cast<uintptr_t>(ExecutorKind::Ordinary) == 0);
@@ -101,6 +103,16 @@ class SerialExecutorRef {
     return SerialExecutorRef(actor, 0);
   }
 
+  static SerialExecutorRef forSynchronousStart() {
+    auto wtable = reinterpret_cast<uintptr_t>(nullptr) |
+                  static_cast<uintptr_t>(ExecutorKind::StartSynchronously);
+    return SerialExecutorRef(nullptr, wtable);
+  }
+  bool isForSynchronousStart() const {
+    return getIdentity() == nullptr &&
+           getExecutorKind() == ExecutorKind::StartSynchronously;
+  }
+
   /// Given a pointer to a serial executor and its SerialExecutor
   /// conformance, return an executor reference for it.
   static SerialExecutorRef forOrdinary(HeapObject *identity,
@@ -127,7 +139,7 @@ class SerialExecutorRef {
 
   const char* getIdentityDebugName() const {
     return isMainExecutor() ? " (MainActorExecutor)"
-           : isGeneric()    ? " (GenericExecutor)"
+           : isGeneric()    ? (isForSynchronousStart() ? " (GenericExecutor/SynchronousStart)" : " (GenericExecutor)")
                             : "";
   }
 

include/swift/ABI/MetadataValues.h

@@ -2668,13 +2668,13 @@ class TaskCreateFlags : public FlagSet<size_t> {
     Task_EnqueueJob                               = 12,
     Task_AddPendingGroupTaskUnconditionally       = 13,
     Task_IsDiscardingTask                         = 14,
-
     /// The task function is consumed by calling it (@callee_owned).
     /// The context pointer should be treated as opaque and non-copyable;
     /// in particular, it should not be retained or released.
     ///
     /// Supported starting in Swift 6.1.
-    Task_IsTaskFunctionConsumed                       = 15,
+    Task_IsTaskFunctionConsumed                   = 15,
+    Task_IsStartSynchronouslyTask                 = 16,
   };
 
   explicit constexpr TaskCreateFlags(size_t bits) : FlagSet(bits) {}
@@ -2707,6 +2707,9 @@ class TaskCreateFlags : public FlagSet<size_t> {
   FLAGSET_DEFINE_FLAG_ACCESSORS(Task_IsTaskFunctionConsumed,
                                 isTaskFunctionConsumed,
                                 setIsTaskFunctionConsumed)
+  FLAGSET_DEFINE_FLAG_ACCESSORS(Task_IsStartSynchronouslyTask,
+                                isSynchronousStartTask,
+                                setIsSYnchronousStartTask)
 };
 
 /// Flags for schedulable jobs.

stdlib/public/Concurrency/Actor.cpp

@@ -134,6 +134,8 @@ class ExecutorTrackingInfo {
   /// is `generic`.
   TaskExecutorRef TaskExecutor = TaskExecutorRef::undefined();
 
+  bool StartedSynchronouslySkipSwitchOnce = false;
+
   /// Whether this context allows switching.  Some contexts do not;
   /// for example, we do not allow switching from swift_job_run
   /// unless the passed-in executor is generic.
@@ -177,7 +179,7 @@ class ExecutorTrackingInfo {
   }
 
   bool allowsSwitching() const {
-    return AllowsSwitching;
+    return AllowsSwitching && !StartedSynchronouslySkipSwitchOnce;
   }
 
   /// Disallow switching in this tracking context.  This should only
@@ -186,6 +188,16 @@ class ExecutorTrackingInfo {
     AllowsSwitching = false;
   }
 
+  void markSynchronousStart() {
+    StartedSynchronouslySkipSwitchOnce = true;
+  }
+  bool isSynchronousStart() const {
+    return StartedSynchronouslySkipSwitchOnce;
+  }
+  void withoutStartSynchronously() {
+    StartedSynchronouslySkipSwitchOnce = false;
+  }
+
   static ExecutorTrackingInfo *current() {
     return ActiveInfoInThread.get();
   }
@@ -2151,13 +2163,23 @@ static bool canGiveUpThreadForSwitch(ExecutorTrackingInfo *trackingInfo,
   assert(trackingInfo || currentExecutor.isGeneric());
 
   // Some contexts don't allow switching at all.
-  if (trackingInfo && !trackingInfo->allowsSwitching())
+  if (trackingInfo && !trackingInfo->allowsSwitching()) {
     return false;
+  }
 
   // We can certainly "give up" a generic executor to try to run
   // a task for an actor.
-  if (currentExecutor.isGeneric())
+  if (currentExecutor.isGeneric()) {
+    if (trackingInfo->isSynchronousStart()) {
+      return false;
+    }
+
+    if (currentExecutor.isForSynchronousStart()) {
+      return false;
+    }
+
     return true;
+  }
 
   // If the current executor is a default actor, we know how to make
   // it give up its thread.
@@ -2278,7 +2300,8 @@ static void swift_task_switchImpl(SWIFT_ASYNC_CONTEXT AsyncContext *resumeContex
                                            : TaskExecutorRef::undefined());
   auto newTaskExecutor = task->getPreferredTaskExecutor();
   SWIFT_TASK_DEBUG_LOG("Task %p trying to switch executors: executor %p%s to "
-                       "new serial executor: %p%s; task executor: from %p%s to %p%s",
+                       "new serial executor: %p%s; task executor: from %p%s to %p%s"
+                       "%s",
                        task,
                        currentExecutor.getIdentity(),
                        currentExecutor.getIdentityDebugName(),
@@ -2287,13 +2310,15 @@ static void swift_task_switchImpl(SWIFT_ASYNC_CONTEXT AsyncContext *resumeContex
                        currentTaskExecutor.getIdentity(),
                        currentTaskExecutor.isDefined() ? "" : " (undefined)",
                        newTaskExecutor.getIdentity(),
-                       newTaskExecutor.isDefined() ? "" : " (undefined)");
+                       newTaskExecutor.isDefined() ? "" : " (undefined)",
+                       trackingInfo->isSynchronousStart() ? "[synchronous start]" : "[NOT SYNC START]");
 
   // If the current executor is compatible with running the new executor,
   // we can just immediately continue running with the resume function
   // we were passed in.
   if (!mustSwitchToRun(currentExecutor, newExecutor, currentTaskExecutor,
                        newTaskExecutor)) {
+    SWIFT_TASK_DEBUG_LOG("Task %p run inline", task);
     return resumeFunction(resumeContext); // 'return' forces tail call
   }
 
@@ -2305,6 +2330,7 @@ static void swift_task_switchImpl(SWIFT_ASYNC_CONTEXT AsyncContext *resumeContex
   // If the current executor can give up its thread, and the new executor
   // can take over a thread, try to do so; but don't do this if we've
   // been asked to yield the thread.
+  SWIFT_TASK_DEBUG_LOG("Task %p can give up thread?", task);
   if (currentTaskExecutor.isUndefined() &&
       canGiveUpThreadForSwitch(trackingInfo, currentExecutor) &&
       !shouldYieldThread() &&
@@ -2339,8 +2365,11 @@ static void swift_task_startSynchronouslyImpl(AsyncTask* task) {
     swift_job_run(task, currentExecutor);
     _swift_task_setCurrent(originalTask);
   } else {
-    // FIXME: this is not correct; we'll keep reusing this thread when we should not have been. See tests with actors.
-    SerialExecutorRef executor = SerialExecutorRef::generic(); // _task_task_getRunSynchronouslyExecutor();
+    SerialExecutorRef executor = SerialExecutorRef::forSynchronousStart();
+
+    ExecutorTrackingInfo trackingInfo;
+//    trackingInfo.markSynchronousStart();
+
     auto originalTask = ActiveTask::swap(task);
     assert(!originalTask);
 

stdlib/public/Concurrency/DiscardingTaskGroup.swift

@@ -205,13 +205,15 @@ public struct DiscardingTaskGroup {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: false,
-      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true
+      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true,
+      isSynchronousStart: false
     )
 #else
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true
+      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true,
+      isSynchronousStart: false
     )
 #endif
 
@@ -252,13 +254,15 @@ public struct DiscardingTaskGroup {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: false,
-      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true
+      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true,
+      isSynchronousStart: false
     )
 #else
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true
+      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true,
+      isSynchronousStart: false
     )
 #endif
 
@@ -281,7 +285,8 @@ public struct DiscardingTaskGroup {
     let flags = taskCreateFlags(
       priority: nil, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true
+      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true,
+      isSynchronousStart: false
     )
 
     // Create the task in this group.
@@ -317,7 +322,8 @@ public struct DiscardingTaskGroup {
     let flags = taskCreateFlags(
       priority: nil, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true
+      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true,
+      isSynchronousStart: false
     )
 
     // Create the task in this group.
@@ -635,7 +641,8 @@ public struct ThrowingDiscardingTaskGroup<Failure: Error> {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true
+      addPendingGroupTaskUnconditionally: true, isDiscardingTask: true,
+      isSynchronousStart: false
     )
 
     // Create the task in this group.
@@ -666,7 +673,8 @@ public struct ThrowingDiscardingTaskGroup<Failure: Error> {
     let flags = taskCreateFlags(
       priority: priority, isChildTask: true, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
-      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true
+      addPendingGroupTaskUnconditionally: false, isDiscardingTask: true,
+      isSynchronousStart: false
     )
 
     // Create the task in this group.

stdlib/public/Concurrency/Task+TaskExecutor.swift

@@ -240,7 +240,7 @@ extension Task where Failure == Never {
       priority: priority, isChildTask: false, copyTaskLocals: true,
       inheritContext: true, enqueueJob: true,
       addPendingGroupTaskUnconditionally: false,
-      isDiscardingTask: false)
+      isDiscardingTask: false, isSynchronousStart: false)
 
 #if $BuiltinCreateAsyncTaskOwnedTaskExecutor
     let (task, _) = Builtin.createTask(
@@ -303,7 +303,7 @@ extension Task where Failure == Error {
       priority: priority, isChildTask: false, copyTaskLocals: true,
       inheritContext: true, enqueueJob: true,
       addPendingGroupTaskUnconditionally: false,
-      isDiscardingTask: false)
+      isDiscardingTask: false, isSynchronousStart: false)
 
 #if $BuiltinCreateAsyncTaskOwnedTaskExecutor
     let (task, _) = Builtin.createTask(
@@ -364,7 +364,7 @@ extension Task where Failure == Never {
       priority: priority, isChildTask: false, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
       addPendingGroupTaskUnconditionally: false,
-      isDiscardingTask: false)
+      isDiscardingTask: false, isSynchronousStart: false)
 
 #if $BuiltinCreateAsyncTaskOwnedTaskExecutor
     let (task, _) = Builtin.createTask(
@@ -425,7 +425,7 @@ extension Task where Failure == Error {
       priority: priority, isChildTask: false, copyTaskLocals: false,
       inheritContext: false, enqueueJob: true,
       addPendingGroupTaskUnconditionally: false,
-      isDiscardingTask: false)
+      isDiscardingTask: false, isSynchronousStart: false)
 
 #if $BuiltinCreateAsyncTaskOwnedTaskExecutor
     let (task, _) = Builtin.createTask(

stdlib/public/Concurrency/Task.cpp

@@ -847,7 +847,8 @@ swift_task_create_commonImpl(size_t rawTaskCreateFlags,
      // basePriority should already be the right value
 
   } else if (taskIsUnstructured(taskCreateFlags, jobFlags)) {
-     SWIFT_TASK_DEBUG_LOG("Creating an unstructured task from %p", currentTask);
+     SWIFT_TASK_DEBUG_LOG("Creating an unstructured task from %p%s", currentTask,
+                          taskCreateFlags.isSynchronousStartTask() ? " [start synchronously]" : "");
 
     if (isUnspecified(basePriority)) {
       // Case 1: No priority specified