[Concurrency] detach, spawnUnlessCancelled, priority param cleanup

Author: ktoso

include/swift/Runtime/Concurrency.h

@@ -117,6 +117,9 @@ SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
 JobPriority
 swift_task_escalate(AsyncTask *task, JobPriority newPriority);
 
+// TODO: "async let wait" and "async let destroy" would be expressed
+//       similar to like TaskFutureWait;
+
 /// This matches the ABI of a closure `<T>(Builtin.NativeObject) async -> T`
 using TaskFutureWaitSignature =
     SWIFT_CC(swiftasync)
@@ -237,11 +240,12 @@ void swift_taskGroup_destroy(TaskGroup *group);
 ///
 /// \code
 /// func swift_taskGroup_addPending(
-///     group: Builtin.RawPointer
+///     group: Builtin.RawPointer,
+///     unconditionally: Bool
 /// ) -> Bool
 /// \endcode
 SWIFT_EXPORT_FROM(swift_Concurrency) SWIFT_CC(swift)
-bool swift_taskGroup_addPending(TaskGroup *group);
+bool swift_taskGroup_addPending(TaskGroup *group, bool unconditionally);
 
 /// Cancel all tasks in the group.
 /// This also prevents new tasks from being added.

lib/SILGen/SILGenDecl.cpp

@@ -1142,6 +1142,8 @@ void SILGenFunction::emitPatternBinding(PatternBindingDecl *PBD,
   auto initialization = emitPatternBindingInitialization(PBD->getPattern(idx),
                                                          JumpDest::invalid());
 
+  // TODO: need to allocate the variable, stackalloc it? pass the address to the start()
+
   // If this is an async let, create a child task to compute the initializer
   // value.
   if (PBD->isAsyncLet()) {
@@ -1157,11 +1159,35 @@ void SILGenFunction::emitPatternBinding(PatternBindingDecl *PBD,
            "Could not find async let autoclosure");
     bool isThrowing = init->getType()->castTo<AnyFunctionType>()->isThrowing();
 
+    // TODO: there's a builtin to make an address into a raw pointer
+    //       --- note dont need that; just have the builtin take it inout?
+    //       --- the builtin can take the address (for start())
+
+    // TODO: make a builtin start async let
+    // Builtin.startAsyncLet -- and in the builtin create the async let record
+
+    // TODO: make a builtin for end async let
+
+    // TODO: IRGen would make a local allocation for the builtins
+
+    // TODO: remember if we did an await already?
+
+    // TODO: force in typesystem that we always await; then end aysnc let does not have to be async
+    // the local let variable is actually owning the result
+    // - but since throwing we can't know; maybe we didnt await on a thing yet
+    // so we do need the tracking if we waited on a thing
+
+    // TODO: awaiting an async let should be able to take ownership
+    // that means we will not await on this async let again, maybe?
+    // it means that the async let destroy should not destroy the result anymore
+
     // Emit the closure for the child task.
     SILValue childTask;
     {
       FullExpr Scope(Cleanups, CleanupLocation(init));
       SILLocation loc(PBD);
+      // TODO: opaque object in the async context that represents the async let
+      //
       childTask = emitRunChildTask(
           loc,
           init->getType(),
@@ -1173,7 +1199,7 @@ void SILGenFunction::emitPatternBinding(PatternBindingDecl *PBD,
     enterDestroyCleanup(childTask);
 
     // Push a cleanup that will cancel the child task at the end of the scope.
-    enterCancelAsyncTaskCleanup(childTask);
+    enterCancelAsyncTaskCleanup(childTask); // TODO: this is "went out scope" rather than just a cancel
 
     // Save the child task so we can await it as needed.
     AsyncLetChildTasks[{PBD, idx}] = { childTask, isThrowing };

stdlib/public/CompatibilityOverride/CompatibilityOverrideConcurrency.def

@@ -180,7 +180,8 @@ OVERRIDE_TASK_GROUP(taskGroup_cancelAll, void,
 
 OVERRIDE_TASK_GROUP(taskGroup_addPending, bool,
                     SWIFT_EXPORT_FROM(swift_Concurrency), SWIFT_CC(swift),
-                    swift::, (TaskGroup *group), (group))
+                    swift::, (TaskGroup *group, bool unconditionally),
+                    (group, unconditionally))
 
 OVERRIDE_TASK_LOCAL(task_localValuePush, void,
                     SWIFT_EXPORT_FROM(swift_Concurrency), SWIFT_CC(swift),

stdlib/public/Concurrency/Task.swift

@@ -396,7 +396,7 @@ extension Task {
 @available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
 @discardableResult
 public func detach<T>(
-  priority: Task.Priority = .default,
+  priority: Task.Priority = .unspecified,
   operation: __owned @Sendable @escaping () async -> T
 ) -> Task.Handle<T, Never> {
   // Set up the job flags for a new task.
@@ -448,7 +448,7 @@ public func detach<T>(
 ///     throw the error the operation has thrown when awaited on.
 @discardableResult
 public func detach<T, Failure>(
-  priority: Task.Priority = .default,
+  priority: Task.Priority = .unspecified,
   operation: __owned @Sendable @escaping () async throws -> T
 ) -> Task.Handle<T, Failure> {
   // Set up the job flags for a new task.
@@ -470,7 +470,7 @@ public func detach<T, Failure>(
 
 @available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
 // TODO: remove this?
-public func _runAsyncHandler(operation: @escaping () async -> ()) {
+func _runAsyncHandler(operation: @escaping () async -> ()) {
   typealias ConcurrentFunctionType = @Sendable () async -> ()
   detach(
     operation: unsafeBitCast(operation, to: ConcurrentFunctionType.self)

stdlib/public/Concurrency/TaskGroup.cpp

@@ -361,13 +361,19 @@ class TaskGroupImpl: public TaskGroupTaskStatusRecord {
   /// is currently executing the group. Here we only need the counts of
   /// pending/ready tasks.
   ///
+  /// If the `unconditionally` parameter is `true` the operation always successfully
+  /// adds a pending task, even if the group is cancelled. If the unconditionally
+  /// flag is `false`, the added pending count will be *reverted* before returning.
+  /// This is because we will NOT add a task to a cancelled group, unless doing
+  /// so unconditionally.
+  ///
   /// Returns *assumed* new status, including the just performed +1.
-  GroupStatus statusAddPendingTaskRelaxed() {
+  GroupStatus statusAddPendingTaskRelaxed(bool unconditionally) {
     auto old = status.fetch_add(GroupStatus::onePendingTask,
                                 std::memory_order_relaxed);
     auto s = GroupStatus{old + GroupStatus::onePendingTask};
 
-    if (s.isCancelled()) {
+    if (!unconditionally && s.isCancelled()) {
       // revert that add, it was meaningless
       auto o = status.fetch_sub(GroupStatus::onePendingTask,
                                 std::memory_order_relaxed);
@@ -821,8 +827,8 @@ bool TaskGroupImpl::cancelAll() {
 // =============================================================================
 // ==== addPending -------------------------------------------------------------
 SWIFT_CC(swift)
-static bool swift_taskGroup_addPendingImpl(TaskGroup *group) {
-  auto assumedStatus = asImpl(group)->statusAddPendingTaskRelaxed();
+static bool swift_taskGroup_addPendingImpl(TaskGroup *group, bool unconditionally) {
+  auto assumedStatus = asImpl(group)->statusAddPendingTaskRelaxed(unconditionally);
   return !assumedStatus.isCancelled();
 }
 

stdlib/public/Concurrency/TaskGroup.swift

@@ -220,22 +220,17 @@ public struct TaskGroup<ChildTaskResult: Sendable> {
   /// - Returns:
   ///   - `true` if the operation was added to the group successfully,
   ///     `false` otherwise (e.g. because the group `isCancelled`)
-  @discardableResult
   public mutating func spawn(
-    overridingPriority priorityOverride: Task.Priority? = nil,
+    priority: Task.Priority = .unspecified,
     operation: __owned @Sendable @escaping () async -> ChildTaskResult
-  ) -> Self.Spawned {
-    let canAdd = _taskGroupAddPendingTask(group: _group)
-    
-    guard canAdd else {
-      // the group is cancelled and is not accepting any new work
-      return Spawned(handle: nil)
-    }
+  ) {
+    _ = _taskGroupAddPendingTask(group: _group, unconditionally: true)
 
     // Set up the job flags for a new task.
     var flags = Task.JobFlags()
     flags.kind = .task
-    flags.priority = priorityOverride ?? getJobFlags(_task).priority
+    flags.priority = priority != Task.Priority.unspecified ?
+        priority : getJobFlags(_task).priority
     flags.isFuture = true
     flags.isChildTask = true
     flags.isGroupChildTask = true
@@ -249,25 +244,56 @@ public struct TaskGroup<ChildTaskResult: Sendable> {
 
     // Enqueue the resulting job.
     _enqueueJobGlobal(Builtin.convertTaskToJob(childTask))
-    
-    return Spawned(handle: Task.Handle(childTask))
   }
 
-  public struct Spawned: Sendable {
-    /// Returns `true` if the task was successfully spawned in the task group,
-    /// `false` otherwise which means that the group was already cancelled and
-    /// refused to accept spawn a new child task.
-    public var successfully: Bool { handle != nil }
-
-    /// Task handle for the spawned task group child task,
-    /// or `nil` if it was not spawned successfully.
-    public let handle: Task.Handle<ChildTaskResult, Never>?
+  /// Add a child task to the group.
+  ///
+  /// ### Error handling
+  /// Operations are allowed to `throw`, in which case the `try await next()`
+  /// invocation corresponding to the failed task will re-throw the given task.
+  ///
+  /// The `add` function will never (re-)throw errors from the `operation`.
+  /// Instead, the corresponding `next()` call will throw the error when necessary.
+  ///
+  /// - Parameters:
+  ///   - overridingPriority: override priority of the operation task
+  ///   - operation: operation to execute and add to the group
+  /// - Returns:
+  ///   - `true` if the operation was added to the group successfully,
+  ///     `false` otherwise (e.g. because the group `isCancelled`)
+  public mutating func spawnUnlessCancelled(
+    priority: Task.Priority = .unspecified,
+    operation: __owned @Sendable @escaping () async -> ChildTaskResult
+  ) -> Bool {
+    let canAdd = _taskGroupAddPendingTask(group: _group, unconditionally: false)
 
-    init(handle: Task.Handle<ChildTaskResult, Never>?) {
-      self.handle = handle
+    guard canAdd else {
+      // the group is cancelled and is not accepting any new work
+      return false
     }
+
+    // Set up the job flags for a new task.
+    var flags = Task.JobFlags()
+    flags.kind = .task
+    flags.priority = priority != Task.Priority.unspecified ?
+        priority : getJobFlags(_task).priority
+    flags.isFuture = true
+    flags.isChildTask = true
+    flags.isGroupChildTask = true
+
+    // Create the asynchronous task future.
+    let (childTask, _) = Builtin.createAsyncTaskGroupFuture(
+      flags.bits, _group, operation)
+
+    // Attach it to the group's task record in the current task.
+    _ = _taskGroupAttachChild(group: _group, child: childTask)
+
+    // Enqueue the resulting job.
+    _enqueueJobGlobal(Builtin.convertTaskToJob(childTask))
+
+    return true
   }
-  
+
   /// Wait for the a child task that was added to the group to complete,
   /// and return (or rethrow) the value it completed with. If no tasks are
   /// pending in the task group this function returns `nil`, allowing the
@@ -417,7 +443,7 @@ public struct ThrowingTaskGroup<ChildTaskResult: Sendable, Failure: Error> {
     self._group = group
   }
 
-  /// Add a child task to the group.
+  /// Spawn, unconditionally, a child task in the group.
   ///
   /// ### Error handling
   /// Operations are allowed to `throw`, in which case the `try await next()`
@@ -432,22 +458,64 @@ public struct ThrowingTaskGroup<ChildTaskResult: Sendable, Failure: Error> {
   /// - Returns:
   ///   - `true` if the operation was added to the group successfully,
   ///     `false` otherwise (e.g. because the group `isCancelled`)
-  @discardableResult
   public mutating func spawn(
-    overridingPriority priorityOverride: Task.Priority? = nil,
+    priority: Task.Priority = .unspecified,
     operation: __owned @Sendable @escaping () async throws -> ChildTaskResult
-  ) -> Self.Spawned {
-    let canAdd = _taskGroupAddPendingTask(group: _group)
+  ) {
+    // we always add, so no need to check if group was cancelled
+    _ = _taskGroupAddPendingTask(group: _group, unconditionally: true)
+
+    // Set up the job flags for a new task.
+    var flags = Task.JobFlags()
+    flags.kind = .task
+    flags.priority = priority != Task.Priority.unspecified ?
+        priority : getJobFlags(_task).priority
+    flags.isFuture = true
+    flags.isChildTask = true
+    flags.isGroupChildTask = true
+
+    // Create the asynchronous task future.
+    let (childTask, _) = Builtin.createAsyncTaskGroupFuture(
+      flags.bits, _group, operation)
+
+    // Attach it to the group's task record in the current task.
+    _ = _taskGroupAttachChild(group: _group, child: childTask)
+
+    // Enqueue the resulting job.
+    _enqueueJobGlobal(Builtin.convertTaskToJob(childTask))
+  }
+
+  /// Add a child task to the group.
+  ///
+  /// ### Error handling
+  /// Operations are allowed to `throw`, in which case the `try await next()`
+  /// invocation corresponding to the failed task will re-throw the given task.
+  ///
+  /// The `add` function will never (re-)throw errors from the `operation`.
+  /// Instead, the corresponding `next()` call will throw the error when necessary.
+  ///
+  /// - Parameters:
+  ///   - overridingPriority: override priority of the operation task
+  ///   - operation: operation to execute and add to the group
+  /// - Returns:
+  ///   - `true` if the operation was added to the group successfully,
+  ///     `false` otherwise (e.g. because the group `isCancelled`)
+  public mutating func spawnUnlessCancelled(
+    priority: Task.Priority = .unspecified,
+    operation: __owned @Sendable @escaping () async throws -> ChildTaskResult
+  ) -> Bool {
+    let canAdd = _taskGroupAddPendingTask(group: _group, unconditionally: false)
 
     guard canAdd else {
       // the group is cancelled and is not accepting any new work
-      return Spawned(handle: nil)
+      return false
     }
 
     // Set up the job flags for a new task.
     var flags = Task.JobFlags()
     flags.kind = .task
-    flags.priority = priorityOverride ?? getJobFlags(_task).priority
+    flags.priority = priority != Task.Priority.unspecified ?
+        priority : getJobFlags(_task).priority
     flags.isFuture = true
     flags.isChildTask = true
     flags.isGroupChildTask = true
@@ -462,22 +530,7 @@ public struct ThrowingTaskGroup<ChildTaskResult: Sendable, Failure: Error> {
     // Enqueue the resulting job.
     _enqueueJobGlobal(Builtin.convertTaskToJob(childTask))
 
-    return Spawned(handle: Task.Handle(childTask))
-  }
-
-  public struct Spawned: Sendable  {
-    /// Returns `true` if the task was successfully spawned in the task group,
-    /// `false` otherwise which means that the group was already cancelled and
-    /// refused to accept spawn a new child task.
-    public var successfully: Bool { handle != nil }
-
-    /// Task handle for the spawned task group child task,
-    /// or `nil` if it was not spawned successfully.
-    public let handle: Task.Handle<ChildTaskResult, Error>?
-
-    init(handle: Task.Handle<ChildTaskResult, Error>?) {
-      self.handle = handle
-    }
+    return true
   }
 
   /// Wait for the a child task that was added to the group to complete,
@@ -754,7 +807,8 @@ func _taskGroupDestroy(group: __owned Builtin.RawPointer)
 @available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
 @_silgen_name("swift_taskGroup_addPending")
 func _taskGroupAddPendingTask(
-  group: Builtin.RawPointer
+  group: Builtin.RawPointer,
+  unconditionally: Bool
 ) -> Bool
 
 @available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)

test/Concurrency/Runtime/async_task_detached.swift renamed to test/Concurrency/Runtime/async_task_detach.swift

@@ -22,7 +22,7 @@ class X {
 func test_detach() async {
   for _ in 1...3 {
     let x = X()
-    let h = Task.runDetached {
+    let h = detach {
       print("inside: \(x)")
     }
     await h.get()

test/Concurrency/Runtime/async_task_priority_current.swift

@@ -12,7 +12,7 @@ import Dispatch
 @available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
 func test_detach() async {
   let a1 = Task.currentPriority
-  print("a1: \(a1)") // CHECK: a1: default
+  print("a1: \(a1)") // CHECK: a1: unspecified
 
   // 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
@@ -23,7 +23,7 @@ func test_detach() async {
   }.get()
 
   let a3 = Task.currentPriority
-  print("a3: \(a3)") // CHECK: a3: default
+  print("a3: \(a3)") // CHECK: a3: unspecified
 }
 
 @available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)