Give up default actors in swift_job_run if we switch to them.

Previously, if this happened, we simply left the actor in a running
state, causing any further jobs submitted to it to never be executed.
I can only speculate why this wasn't showing up in testing.

Also, change swift_job_run so that it prevents switching if the executor
passed in is not generic.  This is an entrypoint for arbitrary executors
and generally should not allow unexpected switching (if someday custom
executors participate in that scheme).  This infrastructure will also
be useful for implementing the `async let` semantics of running
synchronously until the task reaches a suspension point.

Finally, improve the #if'ed logging code throughout the task/actor runtime.

Author: rjmccall

stdlib/public/Concurrency/Actor.cpp

@@ -27,9 +27,6 @@
 #include "llvm/ADT/PointerIntPair.h"
 #include "TaskPrivate.h"
 
-// Uncomment to enable helpful debug spew to stderr
-//#define SWIFT_TASK_PRINTF_DEBUG 1
-
 using namespace swift;
 
 /// Should we yield the thread?
@@ -74,6 +71,11 @@ class ExecutorTrackingInfo {
   /// The active executor.
   ExecutorRef ActiveExecutor = ExecutorRef::generic();
 
+  /// 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.
+  bool AllowsSwitching = true;
+
   /// The tracking info that was active when this one was entered.
   ExecutorTrackingInfo *SavedInfo;
 
@@ -114,21 +116,27 @@ class ExecutorTrackingInfo {
     return ActiveExecutor;
   }
 
-  static ExecutorRef getActiveExecutorInThread() {
-    if (auto activeInfo = ActiveInfoInThread.get())
-      return activeInfo->getActiveExecutor();
-    return ExecutorRef::generic();
+  void setActiveExecutor(ExecutorRef newExecutor) {
+    ActiveExecutor = newExecutor;
   }
 
-  void leave() {
-    ActiveInfoInThread.set(SavedInfo);
+
+  bool allowsSwitching() const {
+    return AllowsSwitching;
+  }
+
+  /// Disallow switching in this tracking context.  This should only
+  /// be set on a new tracking info, before any jobs are run in it.
+  void disallowSwitching() {
+    AllowsSwitching = false;
   }
-  
-  static ExecutorRef getActiveExecutorOnThread() {
-    if (auto active = ActiveInfoInThread.get())
-      return active->getActiveExecutor();
 
-    swift_unreachable("no active executor?!");
+  static ExecutorTrackingInfo *current() {
+    return ActiveInfoInThread.get();
+  }
+
+  void leave() {
+    ActiveInfoInThread.set(SavedInfo);
   }
 };
 
@@ -152,16 +160,6 @@ SWIFT_RUNTIME_DECLARE_THREAD_LOCAL(
 
 } // end anonymous namespace
 
-SWIFT_CC(swift)
-static void swift_job_runImpl(Job *job, ExecutorRef executor) {
-  ExecutorTrackingInfo trackingInfo;
-  trackingInfo.enterAndShadow(executor);
-
-  runJobInEstablishedExecutorContext(job);
-
-  trackingInfo.leave();
-}
-
 void swift::runJobInEstablishedExecutorContext(Job *job) {
   _swift_tsan_acquire(job);
 
@@ -199,9 +197,11 @@ AsyncTask *swift::_swift_task_clearCurrent() {
 
 SWIFT_CC(swift)
 static ExecutorRef swift_task_getCurrentExecutorImpl() {
-  auto result = ExecutorTrackingInfo::getActiveExecutorOnThread();
+  auto currentTracking = ExecutorTrackingInfo::current();
+  auto result = (currentTracking ? currentTracking->getActiveExecutor()
+                                 : ExecutorRef::generic());
 #if SWIFT_TASK_PRINTF_DEBUG
-  fprintf(stderr, "%p getting current executor %p\n", pthread_self(), (void*)result.getRawValue());
+  fprintf(stderr, "[%p] getting current executor %p\n", pthread_self(), result.getIdentity());
 #endif
   return result;
 }
@@ -885,7 +885,7 @@ static Job *preprocessQueue(JobRef first,
 
 void DefaultActorImpl::giveUpThread(RunningJobInfo runner) {
 #if SWIFT_TASK_PRINTF_DEBUG
-  fprintf(stderr, "%p %p.giveUpThread\n", pthread_self(), this);
+  fprintf(stderr, "[%p] giving up thread for actor %p\n", pthread_self(), this);
 #endif
   auto oldState = CurrentState.load(std::memory_order_acquire);
   assert(oldState.Flags.getStatus() == Status::Running);
@@ -936,8 +936,8 @@ void DefaultActorImpl::giveUpThread(RunningJobInfo runner) {
     }
 
 #if SWIFT_TASK_PRINTF_DEBUG
-#  define LOG_STATE_TRANSITION fprintf(stderr, "%p transition from %zx to %zx in %p.%s\n", \
-  pthread_self(), oldState.Flags.getOpaqueValue(), newState.Flags.getOpaqueValue(), this, __FUNCTION__)
+#  define LOG_STATE_TRANSITION fprintf(stderr, "[%p] actor %p transitioned from %zx to %zx (%s)\n", \
+  pthread_self(), this, oldState.Flags.getOpaqueValue(), newState.Flags.getOpaqueValue(), __FUNCTION__)
 #else
 #  define LOG_STATE_TRANSITION ((void)0)
 #endif
@@ -1141,6 +1141,34 @@ Job *DefaultActorImpl::claimNextJobOrGiveUp(bool actorIsOwned,
   }
 }
 
+SWIFT_CC(swift)
+static void swift_job_runImpl(Job *job, ExecutorRef executor) {
+  ExecutorTrackingInfo trackingInfo;
+
+  // swift_job_run is a primary entrypoint for executors telling us to
+  // run jobs.  Actor executors won't expect us to switch off them
+  // during this operation.  But do allow switching if the executor
+  // is generic.
+  if (!executor.isGeneric()) trackingInfo.disallowSwitching();
+
+  trackingInfo.enterAndShadow(executor);
+
+  runJobInEstablishedExecutorContext(job);
+
+  trackingInfo.leave();
+
+  // Give up the current executor if this is a switching context
+  // (which, remember, only happens if we started out on a generic
+  // executor) and we've switched to a default actor.
+  auto currentExecutor = trackingInfo.getActiveExecutor();
+  if (trackingInfo.allowsSwitching() && currentExecutor.isDefaultActor()) {
+    // Use an underestimated priority; if this means we create an
+    // extra processing job in some cases, that's probably okay.
+    auto runner = RunningJobInfo::forOther(JobPriority(0));
+    asImpl(currentExecutor.getDefaultActor())->giveUpThread(runner);
+  }
+}
+
 /// The primary function for processing an actor on a thread.  Start
 /// processing the given default actor as the active default actor on
 /// the current thread, and keep processing whatever actor we're
@@ -1154,15 +1182,18 @@ Job *DefaultActorImpl::claimNextJobOrGiveUp(bool actorIsOwned,
 static void processDefaultActor(DefaultActorImpl *currentActor,
                                 RunningJobInfo runner) {
 #if SWIFT_TASK_PRINTF_DEBUG
-  fprintf(stderr, "%p processDefaultActor %p\n", pthread_self(), currentActor);
+  fprintf(stderr, "[%p] processDefaultActor %p\n", pthread_self(), currentActor);
 #endif
   DefaultActorImpl *actor = currentActor;
 
-  // Register that we're processing a default actor in this frame.
+  // If we actually have work to do, we'll need to set up tracking info.
+  // Optimistically assume that we will; the alternative (an override job
+  // took over the actor first) is rare.
   ExecutorTrackingInfo trackingInfo;
-  auto activeTrackingInfo = trackingInfo.enterOrUpdate(
+  trackingInfo.enterAndShadow(
     ExecutorRef::forDefaultActor(asAbstract(currentActor)));
 
+  // Remember whether we've already taken over the actor.
   bool threadIsRunningActor = false;
   while (true) {
     assert(currentActor);
@@ -1171,12 +1202,13 @@ static void processDefaultActor(DefaultActorImpl *currentActor,
     if (shouldYieldThread())
       break;
 
-    // Claim another job from the current actor.
+    // Try to claim another job from the current actor, taking it over
+    // if we haven't already.
     auto job = currentActor->claimNextJobOrGiveUp(threadIsRunningActor,
                                                   runner);
 
 #if SWIFT_TASK_PRINTF_DEBUG
-    fprintf(stderr, "%p processDefaultActor %p claimed job %p\n", pthread_self(), currentActor, job);
+    fprintf(stderr, "[%p] processDefaultActor %p claimed job %p\n", pthread_self(), currentActor, job);
 #endif
 
     // If we failed to claim a job, we have nothing to do.
@@ -1187,15 +1219,18 @@ static void processDefaultActor(DefaultActorImpl *currentActor,
       break;
     }
 
+    // This thread now owns the current actor.
+    threadIsRunningActor = true;
+
     // Run the job.
     runJobInEstablishedExecutorContext(job);
 
     // The current actor may have changed after the job.
     // If it's become nil, or not a default actor, we have nothing to do.
-    auto currentExecutor = activeTrackingInfo->getActiveExecutor();
+    auto currentExecutor = trackingInfo.getActiveExecutor();
 
 #if SWIFT_TASK_PRINTF_DEBUG
-    fprintf(stderr, "%p processDefaultActor %p current executor now %p\n", pthread_self(), currentActor, (void*)currentExecutor.getRawValue());
+    fprintf(stderr, "[%p] processDefaultActor %p current executor now %p\n", pthread_self(), currentActor, currentExecutor.getIdentity());
 #endif
 
     if (!currentExecutor.isDefaultActor()) {
@@ -1205,17 +1240,15 @@ static void processDefaultActor(DefaultActorImpl *currentActor,
       break;
     }
     currentActor = asImpl(currentExecutor.getDefaultActor());
-
-    // Otherwise, we know that we're running the actor on this thread.
-    threadIsRunningActor = true;
   }
 
-  if (activeTrackingInfo == &trackingInfo)
-    trackingInfo.leave();
+  // Leave the tracking info.
+  trackingInfo.leave();
 
   // If we still have an active actor, we should give it up.
-  if (currentActor)
+  if (threadIsRunningActor && currentActor) {
     currentActor->giveUpThread(runner);
+  }
 
   swift_release(actor);
 }
@@ -1386,7 +1419,14 @@ void swift::swift_MainActor_register(HeapObject *actor) {
 /*****************************************************************************/
 
 /// Can the current executor give up its thread?
-static bool canGiveUpThreadForSwitch(ExecutorRef currentExecutor) {
+static bool canGiveUpThreadForSwitch(ExecutorTrackingInfo *trackingInfo,
+                                     ExecutorRef currentExecutor) {
+  assert(trackingInfo || currentExecutor.isGeneric());
+
+  // Some contexts don't allow switching at all.
+  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())
@@ -1439,31 +1479,38 @@ static bool tryAssumeThreadForSwitch(ExecutorRef newExecutor,
 /// continue to run the given task on it.
 SWIFT_CC(swiftasync)
 static void runOnAssumedThread(AsyncTask *task, ExecutorRef executor,
+                               ExecutorTrackingInfo *oldTracking,
                                RunningJobInfo runner) {
-  // Set that this actor is now the active default actor on this thread,
-  // and set up tracking info if there isn't any already.
-  ExecutorTrackingInfo trackingInfo;
-  auto activeTrackingInfo = trackingInfo.enterOrUpdate(executor);
+  // If there's alreaady tracking info set up, just change the executor
+  // there and tail-call the task.  We don't want these frames to
+  // potentially accumulate linearly.
+  if (oldTracking) {
+    oldTracking->setActiveExecutor(executor);
 
-  // If one already existed, we should just tail-call the task; we don't
-  // want these frames to potentially accumulate linearly.
-  if (activeTrackingInfo != &trackingInfo) {
     // FIXME: force tail call
     return task->runInFullyEstablishedContext();
   }
 
-  // Otherwise, run the new task.
+  // Otherwise, set up tracking info.
+  ExecutorTrackingInfo trackingInfo;
+  trackingInfo.enterAndShadow(executor);
+
+  // Run the new task.
   task->runInFullyEstablishedContext();
 
   // Leave the tracking frame, and give up the current actor if
   // we have one.
   //
-  // In principle, we could execute more tasks here, but that's probably
-  // not a reasonable thing to do in an assumed context rather than a
-  // dedicated actor-processing job.
+  // In principle, we could execute more tasks from the actor here, but
+  // that's probably not a reasonable thing to do in an assumed context
+  // rather than a dedicated actor-processing job.
   executor = trackingInfo.getActiveExecutor();
   trackingInfo.leave();
 
+#if SWIFT_TASK_PRINTF_DEBUG
+  fprintf(stderr, "[%p] leaving assumed thread, current executor is %p\n", pthread_self(), executor.getIdentity());
+#endif
+
   if (executor.isDefaultActor())
     asImpl(executor.getDefaultActor())->giveUpThread(runner);
 }
@@ -1472,9 +1519,12 @@ SWIFT_CC(swiftasync)
 static void swift_task_switchImpl(SWIFT_ASYNC_CONTEXT AsyncContext *resumeContext,
                                   TaskContinuationFunction *resumeFunction,
                                   ExecutorRef newExecutor) {
-  auto currentExecutor = ExecutorTrackingInfo::getActiveExecutorInThread();
+  auto trackingInfo = ExecutorTrackingInfo::current();
+  auto currentExecutor =
+    (trackingInfo ? trackingInfo->getActiveExecutor()
+                  : ExecutorRef::generic());
 #if SWIFT_TASK_PRINTF_DEBUG
-  fprintf(stderr, "%p switch %p -> %p\n", pthread_self(), (void*)currentExecutor.getRawValue(), (void*)newExecutor.getRawValue());
+  fprintf(stderr, "[%p] trying to switch from executor %p to %p\n", pthread_self(), currentExecutor.getIdentity(), newExecutor.getIdentity());
 #endif
 
   // If the current executor is compatible with running the new executor,
@@ -1499,16 +1549,22 @@ 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.
-  if (canGiveUpThreadForSwitch(currentExecutor) &&
+  if (canGiveUpThreadForSwitch(trackingInfo, currentExecutor) &&
       !shouldYieldThread() &&
       tryAssumeThreadForSwitch(newExecutor, runner)) {
+#if SWIFT_TASK_PRINTF_DEBUG
+    fprintf(stderr, "[%p] switch succeeded, task %p assumed thread for executor %p\n", pthread_self(), task, newExecutor.getIdentity());
+#endif
     giveUpThreadForSwitch(currentExecutor, runner);
     // FIXME: force tail call
-    return runOnAssumedThread(task, newExecutor, runner);
+    return runOnAssumedThread(task, newExecutor, trackingInfo, runner);
   }
 
   // Otherwise, just asynchronously enqueue the task on the given
   // executor.
+#if SWIFT_TASK_PRINTF_DEBUG
+  fprintf(stderr, "[%p] switch failed, task %p enqueued on executor %p\n", pthread_self(), task, newExecutor.getIdentity());
+#endif
   swift_task_enqueue(task, newExecutor);
 }
 
@@ -1519,7 +1575,7 @@ static void swift_task_switchImpl(SWIFT_ASYNC_CONTEXT AsyncContext *resumeContex
 SWIFT_CC(swift)
 static void swift_task_enqueueImpl(Job *job, ExecutorRef executor) {
 #if SWIFT_TASK_PRINTF_DEBUG
-  fprintf(stderr, "%p enqueue %p\n", pthread_self(), (void*)executor.getRawValue());
+  fprintf(stderr, "[%p] enqueue job %p on executor %p\n", pthread_self(), job, executor.getIdentity());
 #endif
 
   assert(job && "no job provided");