Revert "Process incoming queue when obtaining non-drainer lock as well"

This reverts commit 74d25eb.

Author: nickolas-pohilets

stdlib/public/Concurrency/Actor.cpp

@@ -984,13 +984,6 @@ class DefaultActorImplFooter {
   // prioritizedJobs in the appropriate priority bucket.
   //
   PriorityQueue prioritizedJobs;
-
-  // As an optimisation, we handle unprioritized jobs when obtaining actor lock.
-  // This flag is used to skip `processIncomingQueue()` on the iteration of the
-  // drainer loop right after taking the lock.
-  // This applies both to obtaining lock when entering the drainer loop, and
-  // when assuming thread for switching to a new actor.
-  bool shouldProcessIncomingQueue;
 #endif
 };
 
@@ -1057,7 +1050,6 @@ class DefaultActorImpl
 #else
    _status().store(ActiveActorStatus(), std::memory_order_relaxed);
    new (&this->prioritizedJobs) PriorityQueue();
-   this->shouldProcessIncomingQueue = true;
 #endif
     SWIFT_TASK_DEBUG_LOG("Creating default actor %p", this);
     concurrency::trace::actor_create(this);
@@ -1374,14 +1366,10 @@ void DefaultActorImpl::enqueueStealer(Job *job, JobPriority priority) {
 #endif
     }
   }
+
 }
 
 void DefaultActorImpl::processIncomingQueue() {
-  if (!shouldProcessIncomingQueue) {
-    SWIFT_TASK_DEBUG_LOG("Skip processing incoming queue of default actor %p", this);
-    return;
-  }
-
   // Pairs with the store release in DefaultActorImpl::enqueue
   bool distributedActorIsRemote = swift_distributed_actor_is_remote(this);
   auto oldState = _status().load(SWIFT_MEMORY_ORDER_CONSUME);
@@ -1427,7 +1415,6 @@ void DefaultActorImpl::handleUnprioritizedJobs(Job *head) {
     head = next;
   }
   prioritizedJobs.enqueueContentsOf(reversed);
-  shouldProcessIncomingQueue = false;
 }
 
 // Called with actor lock held on current thread
@@ -1442,7 +1429,6 @@ Job *DefaultActorImpl::drainOne() {
     SWIFT_TASK_DEBUG_LOG("Drained first job %p from actor %p", firstJob, this);
     concurrency::trace::actor_dequeue(this, firstJob);
   }
-  shouldProcessIncomingQueue = true;
   return firstJob;
 }
 
@@ -1677,9 +1663,11 @@ retry:;
     auto newState = oldState.withRunning();
     newState = newState.withoutEscalatedPriority();
 
-    // Claim incoming jobs when obtaining lock, to save one
+    // Claim incoming jobs when obtaining lock as a drainer, to save one
     // round of atomic load and compare-exchange.
-    newState = newState.withFirstUnprioritisedJob(nullptr);
+    if (asDrainer) {
+      newState = newState.withFirstUnprioritisedJob(nullptr);
+    }
 
     // This needs an acquire since we are taking a lock
     if (_status().compare_exchange_weak(oldState, newState,
@@ -1690,7 +1678,9 @@ retry:;
         assert(prioritizedJobs.empty());
       }
       traceActorStateTransition(this, oldState, newState, distributedActorIsRemote);
-      handleUnprioritizedJobs(oldState.getFirstUnprioritisedJob());
+      if (asDrainer) {
+        handleUnprioritizedJobs(oldState.getFirstUnprioritisedJob());
+      }
       return true;
     }
   }