[Concurrency] Task.escalatePriority

Author: ktoso

stdlib/public/Concurrency/Task.swift

@@ -900,6 +900,70 @@ extension Task where Success == Never, Failure == Never {
   }
 }
 
+// ==== Manual Priority Escalation ---------------------------------------------
+
+extension Task {
+  /// Escalate the task `priority` of the passed in task to the `newPriority`.
+  ///
+  /// - Warning: This API should rarely be used, and instead you can rely on
+  ///   structured concurrency and implicit priority escalation which happens
+  ///   when a higher priority task awaits on a result of a lower priority task.
+  ///
+  ///   I.e. using `await` on the target task usually is the correct way to
+  ///   escalate the target task to the current priority of the calling task,
+  ///   especially because in such setup if the waiting task gets escalated,
+  ///   the waited on task would be escalated automatically as well.
+  ///
+  /// The concurrency runtime is free to interpret and handle escalation
+  /// depending on platform characteristics.
+  ///
+  /// Priority escalation is propagated to child tasks of the waited-on task,
+  /// and will trigger any priority escalation handlers, if any were registered.
+  ///
+  /// Escalation can only *increase* the priority of a task, and
+  /// de-escalating priority is not supported.
+  ///
+  /// This method can be called from any task or thread.
+  ///
+  /// - Parameters:
+  ///   - task: the task which to escalate the priority of
+  ///   - newPriority: the new priority the task should continue executing on
+  @available(SwiftStdlib 9999, *)
+  public static func escalatePriority(_ task: UnsafeCurrentTask, to newPriority: TaskPriority) {
+    _taskEscalate(task._task, newPriority: newPriority.rawValue)
+  }
+
+  /// Escalate the task `priority` of the passed in task to the `newPriority`.
+  ///
+  /// - Warning: This API should rarely be used, and instead you can rely on
+  ///   structured concurrency and implicit priority escalation which happens
+  ///   when a higher priority task awaits on a result of a lower priority task.
+  ///
+  ///   I.e. using `await` on the target task usually is the correct way to
+  ///   escalate the target task to the current priority of the calling task,
+  ///   especially because in such setup if the waiting task gets escalated,
+  ///   the waited on task would be escalated automatically as well.
+  ///
+  /// The concurrency runtime is free to interpret and handle escalation
+  /// depending on platform characteristics.
+  ///
+  /// Priority escalation is propagated to child tasks of the waited-on task,
+  /// and will trigger any priority escalation handlers, if any were registered.
+  ///
+  /// Escalation can only *increase* the priority of a task, and
+  /// de-escalating priority is not supported.
+  ///
+  /// This method can be called from any task or thread.
+  ///
+  /// - Parameters:
+  ///   - task: the task which to escalate the priority of
+  ///   - newPriority: the new priority the task should continue executing on
+  @available(SwiftStdlib 9999, *)
+  public static func escalatePriority(_ task: UnsafeCurrentTask, to newPriority: TaskPriority) {
+    _taskEscalate(task._task, newPriority: newPriority.rawValue)
+  }
+}
+
 // ==== UnsafeCurrentTask ------------------------------------------------------
 
 /// Calls a closure with an unsafe reference to the current task.
@@ -1152,6 +1216,11 @@ func _taskIsCancelled(_ task: Builtin.NativeObject) -> Bool
 @_silgen_name("swift_task_currentPriority")
 internal func _taskCurrentPriority(_ task: Builtin.NativeObject) -> UInt8
 
+@available(SwiftStdlib 9999, *) // TODO: determine how far back this can back-deploy because it already was in runtime
+@_silgen_name("swift_task_escalate")
+internal func _taskEscalate(_ task: Builtin.NativeObject, newPriority: UInt8)
+
+@available(SwiftStdlib 5.1, *)
 @_silgen_name("swift_task_basePriority")
 internal func _taskBasePriority(_ task: Builtin.NativeObject) -> UInt8
 

test/Concurrency/async_task_escalate_priority.swift

@@ -0,0 +1,109 @@
+// RUN: %empty-directory(%t)
+
+// RUN: %target-build-swift %s -Xfrontend -disable-availability-checking -parse-as-library -o %t/async_task_priority
+// RUN: %target-codesign %t/async_task_priority
+// RUN: %target-run %t/async_task_priority
+
+// REQUIRES: VENDOR=apple
+// REQUIRES: executable_test
+// REQUIRES: concurrency
+// REQUIRES: libdispatch
+
+// rdar://76038845
+// REQUIRES: concurrency_runtime
+// UNSUPPORTED: back_deployment_runtime
+// UNSUPPORTED: back_deploy_concurrency
+
+// rdar://101077408 - Temporarily disable on watchOS & iOS simulator
+// UNSUPPORTED: DARWIN_SIMULATOR=watchos
+// UNSUPPORTED: DARWIN_SIMULATOR=ios
+// UNSUPPORTED: DARWIN_SIMULATOR=tvos
+
+// rdar://107390341 - Temporarily disable for arm64e
+// UNSUPPORTED: CPU=arm64e
+
+import Darwin
+@preconcurrency import Dispatch
+import StdlibUnittest
+
+func loopUntil(priority: TaskPriority) async {
+  var loops = 100
+  var currentPriority = Task.currentPriority
+  while (currentPriority != priority) {
+    print("Current priority = \(currentPriority) != \(priority)")
+    await Task.sleep(1_000_000)
+    currentPriority = Task.currentPriority
+    loops -= 1
+    if loops < 1 {
+      fatalError("Task priority was never: \(priority), over multiple loops")
+    }
+  }
+}
+
+func print(_ s: String = "") {
+  fputs("\(s)\n", stderr)
+}
+
+func expectedBasePri(priority: TaskPriority) -> TaskPriority {
+  let basePri = Task.basePriority!
+  print("Testing basePri matching expected pri - \(basePri) == \(priority)")
+  expectEqual(basePri, priority)
+  withUnsafeCurrentTask { unsafeTask in
+    guard let unsafeTask else {
+      fatalError("Expected to be able to get current task, but could not!")
+    }
+    // The UnsafeCurrentTask must return the same value
+    expectEqual(basePri, unsafeTask.basePriority)
+  }
+
+  return basePri
+}
+
+func expectedEscalatedPri(priority: TaskPriority) -> TaskPriority {
+  let curPri = Task.currentPriority
+  print("Testing escalated matching expected pri - \(curPri) == \(priority)")
+  expectEqual(curPri, priority)
+
+  return curPri
+}
+
+func testNestedTaskPriority(basePri: TaskPriority, curPri: TaskPriority) async {
+  let _ = expectedBasePri(priority: basePri)
+  let _ = expectedEscalatedPri(priority: curPri)
+}
+
+@main struct Main {
+  static func main() async {
+
+    let top_level = Task.detached { /* To detach from main actor when running work */
+
+      let tests = TestSuite("Task Priority escalation")
+      if #available(SwiftStdlib 5.1, *) {
+
+        tests.test("Basic task_escalate when task is running") {
+          let sem1 = DispatchSemaphore(value: 0)
+          let sem2 = DispatchSemaphore(value: 0)
+          let task = Task(priority: .background) {
+            let _ = expectedBasePri(priority: .background)
+
+            // Wait until task is running before asking to be escalated
+            sem1.signal()
+            sleep(1)
+
+            await loopUntil(priority: .default)
+            sem2.signal()
+          }
+
+          // Wait till child runs and asks to be escalated
+          sem1.wait()
+          Task.escalatePriority(task, to: .default)
+          sem2.wait()
+        }
+      }
+
+      await runAllTestsAsync()
+    }
+
+    await top_level.value
+  }
+}