Track base priority separately from max priority whereby base priority

is set at creation time. Create a new API for accessing this state

Radar-Id: rdar://problem/86100376

Author: rokhinip

include/swift/ABI/Task.h

@@ -222,9 +222,9 @@ class AsyncTask : public Job {
     void *Storage[14];
 
     /// Initialize this storage during the creation of a task.
-    void initialize(AsyncTask *task);
-    void initializeWithSlab(AsyncTask *task,
-                            void *slab, size_t slabCapacity);
+    void initialize(JobPriority basePri);
+    void initializeWithSlab(JobPriority basePri, void *slab,
+                            size_t slabCapacity);
 
     /// React to the completion of the enclosing task's execution.
     void complete(AsyncTask *task);

include/swift/Runtime/Concurrency.h

@@ -488,6 +488,12 @@ SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
 JobPriority
 swift_task_currentPriority(AsyncTask *task);
 
+/// Returns the base priority of the task. This function does not exist in the
+/// base ABI of this library and must be deployment limited.
+SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
+JobPriority
+swift_task_basePriority(AsyncTask *task);
+
 /// Create and add an cancellation record to the task.
 SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
 CancellationNotificationStatusRecord*

stdlib/public/Concurrency/Task.cpp

@@ -477,6 +477,13 @@ JobPriority swift::swift_task_currentPriority(AsyncTask *task)
   return oldStatus.getStoredPriority();
 }
 
+JobPriority swift::swift_task_basePriority(AsyncTask *task)
+{
+  JobPriority pri = task->_private().BasePriority;
+  SWIFT_TASK_DEBUG_LOG("Task %p has base priority = %zu", task, pri);
+  return pri;
+}
+
 /// Implementation of task creation.
 SWIFT_CC(swift)
 static AsyncTaskAndContext swift_task_create_commonImpl(
@@ -568,6 +575,8 @@ static AsyncTaskAndContext swift_task_create_commonImpl(
   if (jobFlags.getPriority() == JobPriority::Unspecified)
     jobFlags.setPriority(JobPriority::Default);
 
+  JobPriority basePriority = jobFlags.getPriority();
+
   // Figure out the size of the header.
   size_t headerSize = sizeof(AsyncTask);
   if (parent) {
@@ -711,9 +720,10 @@ static AsyncTaskAndContext swift_task_create_commonImpl(
   if (asyncLet && initialSlabSize > 0) {
     assert(parent);
     void *initialSlab = (char*)allocation + amountToAllocate;
-    task->Private.initializeWithSlab(task, initialSlab, initialSlabSize);
+    task->Private.initializeWithSlab(basePriority, initialSlab,
+                                     initialSlabSize);
   } else {
-    task->Private.initialize(task);
+    task->Private.initialize(basePriority);
   }
 
   // Perform additional linking between parent and child task.

stdlib/public/Concurrency/Task.swift

@@ -297,6 +297,20 @@ extension Task where Success == Never, Failure == Never {
       return TaskPriority(rawValue: UInt8(_getCurrentThreadPriority()))
     }
   }
+
+  /// The current task's base priority.
+  ///
+  /// If you access this property outside of any task, this returns nil
+  @available(SwiftStdlib 9999, *)
+  public static var basePriority: TaskPriority? {
+    withUnsafeCurrentTask { task in
+      // If we are running on behalf of a task, use that task's priority.
+      if let unsafeTask = task {
+         return TaskPriority(rawValue: _taskBasePriority(unsafeTask._task))
+      }
+      return nil
+    }
+  }
 }
 
 @available(SwiftStdlib 5.1, *)
@@ -852,6 +866,9 @@ func _taskIsCancelled(_ task: Builtin.NativeObject) -> Bool
 @_silgen_name("swift_task_currentPriority")
 internal func _taskCurrentPriority(_ task: Builtin.NativeObject) -> UInt8
 
+@_silgen_name("swift_task_basePriority")
+internal func _taskBasePriority(_ task: Builtin.NativeObject) -> UInt8
+
 @available(SwiftStdlib 5.1, *)
 @_silgen_name("swift_task_createNullaryContinuationJob")
 func _taskCreateNullaryContinuationJob(priority: Int, continuation: Builtin.RawUnsafeContinuation) -> Builtin.Job

stdlib/public/Concurrency/TaskGroup.cpp

@@ -488,7 +488,7 @@ static void swift_taskGroup_initializeImpl(TaskGroup *group, const Metadata *T)
 // ==== add / attachChild ------------------------------------------------------
 
 void TaskGroup::addChildTask(AsyncTask *child) {
-  SWIFT_TASK_DEBUG_LOG("attach child task = %p to group = %p", child, group);
+  SWIFT_TASK_DEBUG_LOG("attach child task = %p to group = %p", child, this);
 
   // The counterpart of this (detachChild) is performed by the group itself,
   // when it offers the completed (child) task's value to a waiting task -

stdlib/public/Concurrency/TaskPrivate.h

@@ -169,7 +169,7 @@ class TaskFutureWaitAsyncContext : public AsyncContext {
 /// The current state of a task's status records.
 class alignas(sizeof(void*) * 2) ActiveTaskStatus {
   enum : uintptr_t {
-    /// The current running priority of the task.
+    /// The max priority of the task. This is always >= basePriority in the task
     PriorityMask = 0xFF,
 
     /// Has the task been cancelled?
@@ -206,8 +206,8 @@ class alignas(sizeof(void*) * 2) ActiveTaskStatus {
   ActiveTaskStatus() = default;
 #endif
 
-  constexpr ActiveTaskStatus(JobFlags flags)
-    : Record(nullptr), Flags(uintptr_t(flags.getPriority())) {}
+  constexpr ActiveTaskStatus(JobPriority priority)
+      : Record(nullptr), Flags(uintptr_t(priority)) {}
 
   /// Is the task currently cancelled?
   bool isCancelled() const { return Flags & IsCancelled; }
@@ -317,12 +317,23 @@ struct AsyncTask::PrivateStorage {
   /// The top 32 bits of the task ID. The bottom 32 bits are in Job::Id.
   uint32_t Id;
 
-  PrivateStorage(JobFlags flags)
-      : Status(ActiveTaskStatus(flags)), Local(TaskLocal::Storage()) {}
+  /// Base priority of Task - set only at creation time of task.
+  /// Current max priority of task is ActiveTaskStatus.
+  ///
+  /// TODO (rokhinip): Only 8 bits of the full size_t are used. Change this into
+  /// flagset thing so that remaining bits are available for other non-changing
+  /// task status stuff
+  JobPriority BasePriority;
 
-  PrivateStorage(JobFlags flags, void *slab, size_t slabCapacity)
-      : Status(ActiveTaskStatus(flags)), Allocator(slab, slabCapacity),
-        Local(TaskLocal::Storage()) {}
+  // Always create an async task with max priority in ActiveTaskStatus = base
+  // priority. It will be updated later if needed.
+  PrivateStorage(JobPriority basePri)
+      : Status(ActiveTaskStatus(basePri)), Local(TaskLocal::Storage()),
+        BasePriority(basePri) {}
+
+  PrivateStorage(JobPriority basePri, void *slab, size_t slabCapacity)
+      : Status(ActiveTaskStatus(basePri)), Allocator(slab, slabCapacity),
+        Local(TaskLocal::Storage()), BasePriority(basePri) {}
 
   void complete(AsyncTask *task) {
     // Destroy and deallocate any remaining task local items.
@@ -347,14 +358,12 @@ inline const AsyncTask::PrivateStorage &
 AsyncTask::OpaquePrivateStorage::get() const {
   return reinterpret_cast<const PrivateStorage &>(*this);
 }
-inline void AsyncTask::OpaquePrivateStorage::initialize(AsyncTask *task) {
-  new (this) PrivateStorage(task->Flags);
+inline void AsyncTask::OpaquePrivateStorage::initialize(JobPriority basePri) {
+  new (this) PrivateStorage(basePri);
 }
-inline void
-AsyncTask::OpaquePrivateStorage::initializeWithSlab(AsyncTask *task,
-                                                    void *slab,
-                                                    size_t slabCapacity) {
-  new (this) PrivateStorage(task->Flags, slab, slabCapacity);
+inline void AsyncTask::OpaquePrivateStorage::initializeWithSlab(
+    JobPriority basePri, void *slab, size_t slabCapacity) {
+  new (this) PrivateStorage(basePri, slab, slabCapacity);
 }
 inline void AsyncTask::OpaquePrivateStorage::complete(AsyncTask *task) {
   get().complete(task);

test/Concurrency/async_task_base_priority.swift

@@ -0,0 +1,126 @@
+// RUN: %target-run-simple-swift( -Xfrontend -disable-availability-checking %import-libdispatch -parse-as-library)
+
+// REQUIRES: executable_test
+// REQUIRES: concurrency
+// REQUIRES: libdispatch
+
+// rdar://76038845
+// REQUIRES: concurrency_runtime
+// UNSUPPORTED: back_deployment_runtime
+
+import StdlibUnittest
+import Darwin
+import Dispatch
+
+func loopUntil(priority: TaskPriority) async {
+  while (Task.currentPriority != priority) {
+    await Task.sleep(1_000_000_000)
+  }
+}
+
+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)
+  return basePri
+}
+
+func expectedCurrentPri(priority: TaskPriority) -> TaskPriority {
+  let curPri = Task.currentPriority
+  print("Testing curPri matching expected pri - \(curPri) == \(priority)")
+  expectEqual(curPri, priority)
+  return curPri
+}
+
+func testNestedTaskPriority(basePri: TaskPriority, curPri: TaskPriority) async {
+    let _ = expectedBasePri(priority: basePri)
+    let _ = expectedCurrentPri(priority: curPri)
+}
+
+@main struct Main {
+  static func main() async {
+
+    let top_level = detach { /* To detach from main actor when running work */
+
+      let tests = TestSuite("Task base priority")
+      if #available(SwiftStdlib 5.1, *) {
+
+        tests.test("Structured concurrency base priority propagation") {
+          let task = Task(priority: .background) {
+            await loopUntil(priority: .default)
+
+            let basePri = expectedBasePri(priority: .background)
+            let curPri = expectedCurrentPri(priority: .default)
+
+            // Structured concurrency via async let, escalated priority of
+            // parent should propagate
+            print("Testing propagation for async let structured concurrency child")
+            async let child = testNestedTaskPriority(basePri: basePri, curPri: curPri)
+            await child
+
+            let dispatchGroup = DispatchGroup()
+            // Structured concurrency via task groups, escalated priority should
+            // propagate
+            await withTaskGroup(of: Void.self, returning: Void.self) { group in
+              dispatchGroup.enter()
+              group.addTask {
+                print("Testing propagation for task group regular child")
+                let _ = await testNestedTaskPriority(basePri: basePri, curPri: curPri)
+                dispatchGroup.leave()
+                return
+              }
+
+              dispatchGroup.enter()
+              group.addTask(priority: .utility) {
+                print("Testing propagation for task group child with specified priority")
+                let _ = await testNestedTaskPriority(basePri: .utility, curPri: curPri)
+                dispatchGroup.leave()
+                return
+              }
+
+              // Wait for child tasks to finish running, don't await since that
+              // will escalate them
+              dispatchGroup.wait()
+            }
+          }
+
+          await task.value // Escalate task BG->DEF
+        }
+
+        tests.test("Unstructured base priority propagation") {
+          let task = Task(priority : .background) {
+            await loopUntil(priority: .default)
+
+            let basePri = expectedBasePri(priority: .background)
+            let _ = expectedCurrentPri(priority: .default)
+
+            let group = DispatchGroup()
+
+            // Create an unstructured task
+            group.enter()
+            let _ = Task {
+              let _ = await testNestedTaskPriority(basePri: basePri, curPri: basePri)
+              group.leave()
+              return
+            }
+
+            // Wait for unstructured task to finish running, don't await it
+            // since that will escalate
+            group.wait()
+          }
+
+          await task.value // Escalate task BG->DEF
+        }
+
+      }
+      await runAllTestsAsync()
+    }
+
+    await top_level.value
+  }
+}