Revert "Revert some changes for cleaner diff"

This reverts commit d3a2530.

Author: nickolas-pohilets

stdlib/public/Concurrency/Actor.cpp

@@ -1085,9 +1085,13 @@ class DefaultActorImpl
   /// new priority
   void enqueueStealer(Job *job, JobPriority priority);
 
-  /// Processes incoming jobs if needed and dequeues one job of the highest priority.
-  /// The calling thread must be holding the actor lock while calling this.
+  /// Dequeues one job from `prioritisedJobs`.
+  /// The calling thread must be holding the actor lock while calling this
   Job *drainOne();
+
+  /// Atomically claims incoming jobs from ActiveActorStatus, and calls `handleUnprioritizedJobs()`.
+  /// Called with actor lock held on current thread.
+  void processIncomingQueue();
 #endif
 
   /// Check if the actor is actually a distributed *remote* actor.
@@ -1124,10 +1128,6 @@ class DefaultActorImpl
   /// when actor gets a priority override and we schedule a stealer.
   void scheduleActorProcessJob(JobPriority priority);
 
-  /// Atomically claims incoming jobs from ActiveActorStatus, and calls `handleUnprioritizedJobs()`.
-  /// Called with actor lock held on current thread.
-  void processIncomingQueue();
-
   /// Processes claimed incoming jobs into `prioritizedJobs`.
   /// Incoming jobs are of mixed priorities and in LIFO order.
   /// Called with actor lock held on current thread.
@@ -1434,7 +1434,6 @@ void DefaultActorImpl::handleUnprioritizedJobs(Job *head) {
 Job *DefaultActorImpl::drainOne() {
   SWIFT_TASK_DEBUG_LOG("Draining one job from default actor %p", this);
 
-  processIncomingQueue();
   traceJobQueue(this, prioritizedJobs.peek());
   auto firstJob = prioritizedJobs.dequeue();
   if (!firstJob) {
@@ -1499,39 +1498,40 @@ static void defaultActorDrain(DefaultActorImpl *actor) {
       TaskExecutorRef::undefined());
 
   while (true) {
-    if (shouldYieldThread()) {
-      currentActor->unlock(true);
-      break;
-    }
-
     Job *job = currentActor->drainOne();
     if (job == NULL) {
       // No work left to do, try unlocking the actor. This may fail if there is
       // work concurrently enqueued in which case, we'd try again in the loop
-      if (!currentActor->unlock(false)) {
-        continue;
+      if (currentActor->unlock(false)) {
+        break;
+      }
+    } else {
+      if (AsyncTask *task = dyn_cast<AsyncTask>(job)) {
+        auto taskExecutor = task->getPreferredTaskExecutor();
+        trackingInfo.setTaskExecutor(taskExecutor);
       }
-      break;
-    }
 
-    if (AsyncTask *task = dyn_cast<AsyncTask>(job)) {
-      auto taskExecutor = task->getPreferredTaskExecutor();
-      trackingInfo.setTaskExecutor(taskExecutor);
+      // This thread is now going to follow the task on this actor. It may hop off
+      // the actor
+      runJobInEstablishedExecutorContext(job);
+
+      // We could have come back from the job on a generic executor and not as
+      // part of a default actor. If so, there is no more work left for us to do
+      // here.
+      auto currentExecutor = trackingInfo.getActiveExecutor();
+      if (!currentExecutor.isDefaultActor()) {
+        currentActor = nullptr;
+        break;
+      }
+      currentActor = asImpl(currentExecutor.getDefaultActor());
     }
 
-    // This thread is now going to follow the task on this actor. It may hop off
-    // the actor
-    runJobInEstablishedExecutorContext(job);
-
-    // We could have come back from the job on a generic executor and not as
-    // part of a default actor. If so, there is no more work left for us to do
-    // here.
-    auto currentExecutor = trackingInfo.getActiveExecutor();
-    if (!currentExecutor.isDefaultActor()) {
-      currentActor = nullptr;
+    if (shouldYieldThread()) {
+      currentActor->unlock(true);
       break;
     }
-    currentActor = asImpl(currentExecutor.getDefaultActor());
+
+    currentActor->processIncomingQueue();
   }
 
   // Leave the tracking info.