Using multiple insertion points to ensure all jobs are always inserted in O(1)

Author: nickolas-pohilets

stdlib/public/Concurrency/Actor.cpp

@@ -983,28 +983,38 @@ class alignas(sizeof(void *) * 2) ActiveActorStatus {
 
 /// Given that a job is enqueued normally on a default actor, get/set
 /// the next job in the actor's queue.
-static JobRef getNextJobInQueue(Job *job) {
+static JobRef getNextJobRef(Job *job) {
   return *reinterpret_cast<JobRef *>(job->SchedulerPrivate);
 }
-static void setNextJobInQueue(Job *job, JobRef next) {
+static void setNextJobRef(Job *job, JobRef next) {
   *reinterpret_cast<JobRef *>(job->SchedulerPrivate) = next;
 }
 
-namespace {
+static Job* getNextJob(Job *job) {
+  return getNextJobRef(job).getAsPreprocessedJob();
+}
+static void setNextJob(Job *job, Job* next) {
+  setNextJobRef(job, JobRef::getPreprocessed(next));
+}
 
-struct JobQueueTraits {
-  static Job *getNext(Job *job) {
-    return getNextJobInQueue(job).getAsPreprocessedJob();
-  }
-  static void setNext(Job *job, Job *next) {
-    setNextJobInQueue(job, JobRef::getPreprocessed(next));
-  }
-  static int compare(Job *lhs, Job *rhs) {
-    return descendingPriorityOrder(lhs->getPriority(), rhs->getPriority());
+enum { NumPriorityBuckets = 5 };
+
+static int getPriorityIndex(JobPriority priority) {
+  switch (priority) {
+    case JobPriority::UserInteractive:
+      return 0;
+    case JobPriority::UserInitiated:
+      return 1;
+    case JobPriority::Unspecified:
+      assert(false);
+    case JobPriority::Default:
+      return 2;
+    case JobPriority::Utility:
+      return 3;
+    case JobPriority::Background:
+      return 4;
   }
-};
-
-} // end anonymous namespace
+}
 
 #endif
 
@@ -1080,9 +1090,14 @@ class DefaultActorImpl : public HeapObject {
   // the future
   alignas(sizeof(ActiveActorStatus)) char StatusStorage[sizeof(ActiveActorStatus)];
 
-  using ListMerger = swift::ListMerger<Job *, JobQueueTraits>;
-  ListMerger::LastInsertionPoint lastInsertionPoint =
-      ListMerger::LastInsertionPoint();
+  // All jobs are stored in a single linked list
+  // Head of the list is stored inside ActiveActorStatus
+  // List starts with a zero or more unprocessed jobs of mixed priorities in LIFO order.
+  // After that list contains processed jobs sorted by priority.
+  // Within the same priority jobs are sorted in the FIFO order.
+  // Elements of `insertionPoints` point to the last job of the corresponding priority,
+  // or are null if no jobs of that priority exist.
+  Job *insertionPoints[NumPriorityBuckets];
 #endif
   // TODO (rokhinip): Make this a flagset
   bool isDistributedRemoteActor;
@@ -1161,7 +1176,8 @@ class DefaultActorImpl : public HeapObject {
 
   Job *preprocessQueue(JobRef start);
   Job *preprocessQueue(JobRef unprocessedStart, JobRef unprocessedEnd,
-                       Job *existingProcessedJobsToMergeInto);
+                       Job *processedJobsStart);
+  Job *preprocessJobs(Job* jobsToProcess, Job *processedJobsStart);
 #endif /* !SWIFT_CONCURRENCY_ACTORS_AS_LOCKS */
 
   void deallocateUnconditional();
@@ -1251,34 +1267,28 @@ static NonDefaultDistributedActorImpl *asImpl(NonDefaultDistributedActor *actor)
 // preprocessQueue
 Job *DefaultActorImpl::preprocessQueue(JobRef unprocessedStart,
                                        JobRef unprocessedEnd,
-                                       Job *existingProcessedJobsToMergeInto) {
-  assert(existingProcessedJobsToMergeInto != NULL);
+                                       Job *processedJobsStart) {
+  assert(processedJobsStart != NULL);
   assert(unprocessedStart.needsPreprocessing());
   assert(unprocessedStart.getAsJob() != unprocessedEnd.getAsJob());
 
   // Build up a list of jobs we need to preprocess
-  ListMerger jobsToProcess;
+  Job* jobsToProcess = nullptr;
 
   // Get just the prefix list of unprocessed jobs
   auto current = unprocessedStart;
   while (current != unprocessedEnd) {
     assert(current.needsPreprocessing());
     // Advance current to next pointer and process current unprocessed job
     auto job = current.getAsJob();
-    current = getNextJobInQueue(job);
+    current = getNextJobRef(job);
 
-    jobsToProcess.insertAtFront(job);
+    setNextJob(job, jobsToProcess);
+    jobsToProcess = job;
   }
 
   // Finish processing the unprocessed jobs
-  Job *newProcessedJobs = std::get<0>(jobsToProcess.release());
-  assert(newProcessedJobs);
-
-  ListMerger mergedList(existingProcessedJobsToMergeInto, lastInsertionPoint);
-  mergedList.merge(newProcessedJobs);
-  Job *result;
-  std::tie(result, lastInsertionPoint) = mergedList.release();
-  return result;
+  return preprocessJobs(jobsToProcess, processedJobsStart);
 }
 
 // Called with the actor drain lock held.
@@ -1297,49 +1307,56 @@ Job *DefaultActorImpl::preprocessQueue(JobRef start) {
   // There exist some jobs which haven't been preprocessed
 
   // Build up a list of jobs we need to preprocess
-  ListMerger jobsToProcess;
+  Job* jobsToProcess = nullptr;
 
-  Job *wellFormedListStart = NULL;
+  Job *processedJobsStart = NULL;
 
-  auto current = start;
-  while (current) {
+  for (JobRef current = start; current; ) {
     if (!current.needsPreprocessing()) {
       // We can assume that everything from here onwards as being well formed
       // and sorted
-      wellFormedListStart = current.getAsPreprocessedJob();
+      processedJobsStart = current.getAsPreprocessedJob();
       break;
     }
 
     // Advance current to next pointer and insert current fella to jobsToProcess
     // list
     auto job = current.getAsJob();
-    current = getNextJobInQueue(job);
+    current = getNextJobRef(job);
 
-    jobsToProcess.insertAtFront(job);
+    setNextJob(job, jobsToProcess);
+    jobsToProcess = job;
   }
 
   // Finish processing the unprocessed jobs
-  auto processedJobHead = std::get<0>(jobsToProcess.release());
-  assert(processedJobHead);
-
-  Job *firstJob = NULL;
-  if (wellFormedListStart) {
-    // Merge it with already known well formed list if we have one.
-    ListMerger mergedList(wellFormedListStart, lastInsertionPoint);
-    mergedList.merge(processedJobHead);
-    std::tie(firstJob, lastInsertionPoint) = mergedList.release();
-  } else {
-    // Nothing to merge with, just return the head we already have
-    firstJob = processedJobHead;
-    lastInsertionPoint = ListMerger::LastInsertionPoint();
-  }
+  return preprocessJobs(jobsToProcess, processedJobsStart);
+}
 
-  return firstJob;
+Job *DefaultActorImpl::preprocessJobs(Job* jobsToProcess, Job *processedJobsStart) {
+  for (auto current = jobsToProcess; current; ) {
+    auto next = getNextJob(current);
+    int priorityIndex = getPriorityIndex(current->getPriority());
+    for (int i = priorityIndex; ; i--) {
+      if (i < 0) {
+        setNextJob(current, processedJobsStart);
+        processedJobsStart = current;
+        break;
+      }
+      if (insertionPoints[i]) {
+        setNextJob(current, getNextJob(insertionPoints[i]));
+        setNextJob(insertionPoints[i], current);
+        break;
+      }
+    }
+    insertionPoints[priorityIndex] = current;
+    current = next;
+  }
+  return processedJobsStart;
 }
 
 static void traceJobQueue(DefaultActorImpl *actor, Job *first) {
   concurrency::trace::actor_note_job_queue(actor, first, [](Job *job) {
-    return getNextJobInQueue(job).getAsPreprocessedJob();
+    return getNextJob(job);
   });
 }
 
@@ -1381,7 +1398,7 @@ void DefaultActorImpl::enqueue(Job *job, JobPriority priority) {
 
     // Link this into the queue in the atomic state
     JobRef currentHead = oldState.getFirstJob();
-    setNextJobInQueue(job, currentHead);
+    setNextJobRef(job, currentHead);
     JobRef newHead = JobRef::getUnpreprocessed(job);
 
     newState = newState.withFirstJob(newHead);
@@ -1533,11 +1550,14 @@ Job * DefaultActorImpl::drainOne() {
 
     auto newState = oldState;
     // Dequeue the first job and set up a new head
-    newState = newState.withFirstJob(getNextJobInQueue(firstJob));
+    newState = newState.withFirstJob(getNextJobRef(firstJob));
     if (_status().compare_exchange_weak(oldState, newState,
                             /* success */ std::memory_order_relaxed,
                             /* failure */ std::memory_order_relaxed)) {
-      lastInsertionPoint.nodeWasRemoved(firstJob);
+      int priorityIndex = getPriorityIndex(firstJob->getPriority());
+      if (insertionPoints[priorityIndex] == firstJob) {
+        insertionPoints[priorityIndex] = nullptr;
+      }
       SWIFT_TASK_DEBUG_LOG("Drained first job %p from actor %p", firstJob, this);
       traceActorStateTransition(this, oldState, newState, distributedActorIsRemote);
       concurrency::trace::actor_dequeue(this, firstJob);