Merge pull request #40222 from rjmccall/actor-priority-order

Teach default actors to execute jobs in priority order

Author: rjmccall

stdlib/public/Concurrency/Actor.cpp

@@ -34,6 +34,7 @@
 #include "swift/Runtime/ThreadLocalStorage.h"
 #include "swift/ABI/Task.h"
 #include "swift/ABI/Actor.h"
+#include "swift/Basic/ListMerger.h"
 #ifndef SWIFT_CONCURRENCY_BACK_DEPLOYMENT
 #include "llvm/Config/config.h"
 #else
@@ -1110,6 +1111,22 @@ void DefaultActorImpl::OverrideJobCache::commit() {
   }
 }
 
+namespace {
+
+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());
+  }
+};
+
+} // end anonymous namespace
+
 /// Preprocess the prefix of the actor's queue that hasn't already
 /// been preprocessed:
 ///
@@ -1128,7 +1145,8 @@ static Job *preprocessQueue(JobRef first,
   if (!first.needsPreprocessing())
     return first.getAsPreprocessedJob();
 
-  Job *firstNewJob = nullptr;
+  using ListMerger = swift::ListMerger<Job*, JobQueueTraits>;
+  ListMerger newJobs;
 
   while (first != previousFirst) {
     // If we find something that doesn't need preprocessing, it must've
@@ -1160,27 +1178,20 @@ static Job *preprocessQueue(JobRef first,
     // jobs; since enqueue() always adds jobs to the front, reversing
     // the order effectively makes the actor queue FIFO, which is what
     // we want.
-    // FIXME: but we should also sort by priority
     auto job = first.getAsJob();
     first = getNextJobInQueue(job);
-    setNextJobInQueue(job, JobRef::getPreprocessed(firstNewJob));
-    firstNewJob = job;
+    newJobs.insertAtFront(job);
   }
 
   // If there are jobs already in the queue, put the new jobs at the end.
+  auto firstNewJob = newJobs.release();
   if (!firstNewJob) {
     firstNewJob = previousFirstNewJob;
   } else if (previousFirstNewJob) {
-    auto cur = previousFirstNewJob;
-    while (true) {
-      auto next = getNextJobInQueue(cur).getAsPreprocessedJob();
-      if (!next) {
-        setNextJobInQueue(cur, JobRef::getPreprocessed(firstNewJob));
-        break;
-      }
-      cur = next;
-    }
-    firstNewJob = previousFirstNewJob;
+    // Merge the jobs we just processed into the existing job list.
+    ListMerger merge(previousFirstNewJob);
+    merge.merge(firstNewJob);
+    firstNewJob = merge.release();
   }
 
   return firstNewJob;

unittests/runtime/Actor.cpp

@@ -48,7 +48,7 @@ static void enqueueGlobal(Job *job,
   // Add the job after (i.e. before in execution order) all jobs
   // with lower priority.
   for (auto i = globalQueue.begin(), e = globalQueue.end(); i != e; ++i) {
-    if (job->getPriority() <= (*i)->getPriority()) {
+    if (descendingPriorityOrder((*i)->getPriority(), job->getPriority()) <= 0) {
       globalQueue.insert(i, job);
       return;
     }
@@ -158,6 +158,19 @@ static AsyncTask *createTask(JobPriority priority, Fn &&fn) {
            .first;
 }
 
+template <class Fn>
+static AsyncTask *createAndEnqueueTask(JobPriority priority,
+                                       TestActor *actor,
+                                       Fn &&fn) {
+  auto task = createTaskWithContext<AsyncContext, Fn>(priority, std::move(fn))
+           .first;
+  if (actor)
+    swift_task_enqueue(task, ExecutorRef::forDefaultActor(actor));
+  else
+    swift_task_enqueueGlobal(task);
+  return task;
+}
+
 template <class Context, class Fn>
 static void parkTask(AsyncTask *task, Context *context, Fn &&fn) {
   auto invoke =
@@ -312,7 +325,7 @@ TEST(ActorTest, actorContention) {
 
         parkTask(task, context,
           [](Context *context) SWIFT_CC(swiftasync) {
-            EXPECT_PROGRESS(3);
+            EXPECT_PROGRESS(2);
             auto executor = swift_task_getCurrentExecutor();
             EXPECT_FALSE(executor.isGeneric());
             EXPECT_EQ(ExecutorRef::forDefaultActor(context->get<1>()),
@@ -336,7 +349,7 @@ TEST(ActorTest, actorContention) {
     auto task1 = createTaskStoring(JobPriority::Background,
                                    (AsyncTask*) nullptr, actor,
       [](Context *context) SWIFT_CC(swiftasync) {
-        EXPECT_PROGRESS(2);
+        EXPECT_PROGRESS(3);
         auto executor = swift_task_getCurrentExecutor();
         EXPECT_FALSE(executor.isGeneric());
         EXPECT_EQ(ExecutorRef::forDefaultActor(context->get<1>()),
@@ -361,3 +374,115 @@ TEST(ActorTest, actorContention) {
     EXPECT_PROGRESS(0);
   });
 }
+
+TEST(ActorTest, actorPriority) {
+  run([] {
+    auto actor = createActor();
+
+    createAndEnqueueTask(JobPriority::Background, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(4);
+      return context->ResumeParent(context);
+    });
+
+    createAndEnqueueTask(JobPriority::Utility, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(1);
+      return context->ResumeParent(context);
+    });
+
+    createAndEnqueueTask(JobPriority::Background, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(5);
+      finishTest();
+      return context->ResumeParent(context);
+    });
+
+    createAndEnqueueTask(JobPriority::Utility, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(2);
+      return context->ResumeParent(context);
+    });
+
+    createAndEnqueueTask(JobPriority::Default, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(0);
+      return context->ResumeParent(context);
+    });
+
+    createAndEnqueueTask(JobPriority::Utility, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(3);
+      return context->ResumeParent(context);
+    });
+  });
+}
+
+TEST(ActorTest, actorPriority2) {
+  run([] {
+    auto actor = createActor();
+
+    createAndEnqueueTask(JobPriority::Background, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(7);
+      return context->ResumeParent(context);
+    });
+
+    createAndEnqueueTask(JobPriority::Utility, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(1);
+
+      createAndEnqueueTask(JobPriority::Utility, actor,
+                           [=](AsyncContext *context) {
+        EXPECT_PROGRESS(5);
+        return context->ResumeParent(context);
+      });
+
+      createAndEnqueueTask(JobPriority::Default, actor,
+                           [](AsyncContext *context) {
+        EXPECT_PROGRESS(2);
+        return context->ResumeParent(context);
+      });
+
+      return context->ResumeParent(context);
+    });
+
+    createAndEnqueueTask(JobPriority::Background, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(8);
+      return context->ResumeParent(context);
+    });
+
+    createAndEnqueueTask(JobPriority::Utility, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(3);
+
+      createAndEnqueueTask(JobPriority::Background, actor,
+                           [=](AsyncContext *context) {
+        EXPECT_PROGRESS(9);
+        finishTest();
+        return context->ResumeParent(context);
+      });
+
+      createAndEnqueueTask(JobPriority::Utility, actor,
+                           [=](AsyncContext *context) {
+        EXPECT_PROGRESS(6);
+        return context->ResumeParent(context);
+      });
+
+      return context->ResumeParent(context);
+    });
+
+    createAndEnqueueTask(JobPriority::Default, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(0);
+      return context->ResumeParent(context);
+    });
+
+    createAndEnqueueTask(JobPriority::Utility, actor,
+                         [=](AsyncContext *context) {
+      EXPECT_PROGRESS(4);
+      return context->ResumeParent(context);
+    });
+  });
+}