Merge pull request #38709 from DougGregor/actor-scheduling-without-priorities

Actor scheduling without priorities

Author: DougGregor

stdlib/public/Concurrency/Actor.cpp

@@ -1050,8 +1050,7 @@ static void wakeOverrides(ProcessOverrideJob *nextOverride,
     nextOverride = cur->NextJob.getAsPreprocessedOverride();
 
     if (hasAlreadyActivated ||
-        !targetPriority ||
-        cur->getPriority() != *targetPriority)
+        !targetPriority)
       cur->wakeAndAbandon();
     else
       hasAlreadyActivated = cur->wakeAndActivate();
@@ -1214,8 +1213,7 @@ void DefaultActorImpl::giveUpThread(RunningJobInfo runner) {
     }
 
     bool hasMoreJobs = (bool) newState.FirstJob;
-    bool hasOverrideAtNewPriority =
-      (runner.Priority < oldState.Flags.getMaxPriority());
+    bool hasOverrideAtNewPriority = false;
     bool hasActiveInlineJob = newState.Flags.hasActiveInlineJob();
     bool needsNewProcessJob = hasMoreJobs && !hasOverrideAtNewPriority;
 
@@ -1316,8 +1314,7 @@ Job *DefaultActorImpl::claimNextJobOrGiveUp(bool actorIsOwned,
 
       // If the actor is out of work, or its priority doesn't match our
       // priority, don't try to take over the actor.
-      if (!oldState.FirstJob ||
-          oldState.Flags.getMaxPriority() != runner.Priority) {
+      if (!oldState.FirstJob) {
 
         // The only change we need here is inline-runner bookkeeping.
         if (!tryUpdateForInlineRunner())
@@ -1399,8 +1396,7 @@ Job *DefaultActorImpl::claimNextJobOrGiveUp(bool actorIsOwned,
     // FIXME: should this be an exact match in priority instead of
     // potentially running jobs with too high a priority?
     Job *jobToRun;
-    if (oldState.Flags.getMaxPriority() <= runner.Priority &&
-        newFirstJob) {
+    if (newFirstJob) {
       jobToRun = newFirstJob;
       newState.FirstJob = getNextJobInQueue(newFirstJob);
       newState.Flags.setStatus(Status::Running);
@@ -1646,7 +1642,7 @@ void DefaultActorImpl::enqueue(Job *job) {
 
     // If we need an override job, create it (if necessary) and
     // register it with the queue.
-    bool needsOverride = !wasIdle && newPriority != oldPriority;
+    bool needsOverride = false;
     if (needsOverride) {
       overrideJob.addToState(this, newState);
     } else {

stdlib/public/Concurrency/Task.cpp

@@ -496,15 +496,18 @@ static AsyncTaskAndContext swift_task_create_commonImpl(
 
     if (currentTask)
       jobFlags.setPriority(currentTask->getPriority());
-    else
-      // FIXME: Ideally, this should be setting priority based on
-      // swift_task_getCurrentThreadPriority(). However, that creates
-      // priority differences which lead to different kinds of hangs
-      // Temporarily use Unspecified to work around that.
-      // See also: PR #37939.
-      jobFlags.setPriority(JobPriority::Unspecified);
+    else if (taskCreateFlags.inheritContext())
+      jobFlags.setPriority(swift_task_getCurrentThreadPriority());
   }
 
+  // Adjust user-interactive priorities down to user-initiated.
+  if (jobFlags.getPriority() == JobPriority::UserInteractive)
+    jobFlags.setPriority(JobPriority::UserInitiated);
+
+  // If there is still no job priority, use the default priority.
+  if (jobFlags.getPriority() == JobPriority::Unspecified)
+    jobFlags.setPriority(JobPriority::Default);
+
   // Figure out the size of the header.
   size_t headerSize = sizeof(AsyncTask);
   if (parent) {

stdlib/public/Concurrency/Task.swift

@@ -269,7 +269,7 @@ extension Task where Success == Never, Failure == Never {
       }
 
       // Otherwise, query the system.
-      return TaskPriority(rawValue: UInt8(0))
+      return TaskPriority(rawValue: UInt8(_getCurrentThreadPriority()))
     }
   }
 }
@@ -632,11 +632,10 @@ extension Task where Success == Never, Failure == Never {
   /// As such,
   /// this method isn't necessarily a way to avoid resource starvation.
   public static func yield() async {
-    let currentTask = Builtin.getCurrentAsyncTask()
-    let priority = getJobFlags(currentTask).priority ?? Task.currentPriority._downgradeUserInteractive
-
     return await Builtin.withUnsafeContinuation { (continuation: Builtin.RawUnsafeContinuation) -> Void in
-      let job = _taskCreateNullaryContinuationJob(priority: Int(priority.rawValue), continuation: continuation)
+      let job = _taskCreateNullaryContinuationJob(
+          priority: Int(Task.currentPriority.rawValue),
+          continuation: continuation)
       _enqueueJobGlobal(job)
     }
   }
@@ -708,7 +707,8 @@ public struct UnsafeCurrentTask {
   /// - SeeAlso: `TaskPriority`
   /// - SeeAlso: `Task.currentPriority`
   public var priority: TaskPriority {
-    getJobFlags(_task).priority ?? .unspecified
+    getJobFlags(_task).priority ?? TaskPriority(
+        rawValue: UInt8(_getCurrentThreadPriority()))
   }
 
   /// Cancel the current task.

stdlib/public/Concurrency/TaskPrivate.h

@@ -236,7 +236,7 @@ class alignas(sizeof(void*) * 2) ActiveTaskStatus {
   ActiveTaskStatus withEscalatedPriority(JobPriority priority) const {
     assert(priority > getStoredPriority());
     return ActiveTaskStatus(Record,
-                            (Flags & PriorityMask)
+                            (Flags & ~PriorityMask)
                                | IsEscalated | uintptr_t(priority));
   }
   ActiveTaskStatus withoutStoredPriorityEscalation() const {

stdlib/public/Concurrency/TaskSleep.swift

@@ -19,11 +19,10 @@ extension Task where Success == Never, Failure == Never {
   ///
   /// This function does _not_ block the underlying thread.
   public static func sleep(_ duration: UInt64) async {
-    let currentTask = Builtin.getCurrentAsyncTask()
-    let priority = getJobFlags(currentTask).priority ?? Task.currentPriority._downgradeUserInteractive
-
     return await Builtin.withUnsafeContinuation { (continuation: Builtin.RawUnsafeContinuation) -> Void in
-      let job = _taskCreateNullaryContinuationJob(priority: Int(priority.rawValue), continuation: continuation)
+      let job = _taskCreateNullaryContinuationJob(
+          priority: Int(Task.currentPriority.rawValue),
+          continuation: continuation)
       _enqueueJobGlobalWithDelay(duration, job)
     }
   }

test/Concurrency/Runtime/actor_counters.swift

@@ -36,6 +36,11 @@ actor Counter {
   }
 }
 
+// Produce a random priority.
+nonisolated var randomPriority: TaskPriority? {
+  let priorities: [TaskPriority?] = [ .background, .low, .medium, .high, nil ]
+  return priorities.randomElement()!
+}
 
 @available(SwiftStdlib 5.5, *)
 func worker(identity: Int, counters: [Counter], numIterations: Int) async {
@@ -59,8 +64,8 @@ func runTest(numCounters: Int, numWorkers: Int, numIterations: Int) async {
   var workers: [Task.Handle<Void, Error>] = []
   for i in 0..<numWorkers {
     workers.append(
-      detach { [counters] in
-        await Task.sleep(UInt64.random(in: 0..<100) * 1_000_000)
+      Task.detached(priority: randomPriority) { [counters] in
+        await try! Task.sleep(nanoseconds: UInt64.random(in: 0..<100) * 1_000_000)
         await worker(identity: i, counters: counters, numIterations: numIterations)
       }
     )

test/Concurrency/Runtime/async.swift

@@ -1,11 +1,9 @@
-// REQUIRES: rdar79670222
 // RUN: %target-run-simple-swift( -Xfrontend -disable-availability-checking %import-libdispatch)
 
 // REQUIRES: executable_test
 // REQUIRES: concurrency
 // REQUIRES: libdispatch
 
-// rdar://76038845
 // UNSUPPORTED: use_os_stdlib
 
 import Dispatch

test/Concurrency/Runtime/async_task_priority_current.swift

@@ -13,7 +13,7 @@ import Dispatch
 @available(SwiftStdlib 5.5, *)
 func test_detach() async {
   let a1 = Task.currentPriority
-  print("a1: \(a1)") // CHECK: TaskPriority(rawValue: 0)
+  print("a1: \(a1)") // CHECK: TaskPriority(rawValue: 21)
 
   // Note: remember to detach using a higher priority, otherwise a lower one
   // might be escalated by the get() and we could see `default` in the detached
@@ -24,7 +24,7 @@ func test_detach() async {
   }.get()
 
   let a3 = Task.currentPriority
-  print("a3: \(a3)") // CHECK: a3: TaskPriority(rawValue: 0)
+  print("a3: \(a3)") // CHECK: a3: TaskPriority(rawValue: 21)
 }
 
 @available(SwiftStdlib 5.5, *)