+task Implement Task.currentPriority

cleanups

Author: ktoso

stdlib/public/Concurrency/TaskGroup.swift

@@ -73,23 +73,23 @@ extension Task {
     // Set up the job flags for a new task.
     var groupFlags = JobFlags()
     groupFlags.kind = .task // TODO: .taskGroup?
-    groupFlags.priority = .default // TODO: parent's priority // await Task.currentPriority()
+    groupFlags.priority = await Task.currentPriority()
     groupFlags.isFuture = true
+    groupFlags.isChildTask = true
 
     // 1. Prepare the Group task
     // FIXME: do we have to rather prepare it inside the task we spawn; and yield it back along with the result instead?
     var group = Task.Group<TaskResult>(parentTask: parent)
 
     let (groupTask, context) =
-      Builtin.createAsyncTaskFuture(groupFlags.bits, nil) { () async throws -> BodyResult in
-        let got = await try body(&group)
-        return got
+      Builtin.createAsyncTaskFuture(groupFlags.bits, parent) { () async throws -> BodyResult in
+        await try body(&group)
       }
     let groupHandle = Handle<BodyResult>(task: groupTask)
 
     // 2.0) Run the task!
-    DispatchQueue.global(priority: .default).async {
-      groupHandle.run() // TODO: this synchronously runs
+    DispatchQueue.global(priority: .default).async { // TODO: use executors when they land
+      groupHandle.run()
     }
 
     // 2.1) ensure that if we fail and exit by throwing we will cancel all tasks,

test/Concurrency/Runtime/async_task_group_basic.swift

@@ -20,6 +20,7 @@ extension DispatchQueue {
 }
 
 @available(*, deprecated, message: "This is a temporary hack")
+@discardableResult
 func launch<R>(operation: @escaping () async -> R) -> Task.Handle<R> {
   let handle = Task.runDetached(operation: operation)
 

test/Concurrency/Runtime/async_task_group_cancellation.swift

@@ -0,0 +1,103 @@
+// RUN: %target-run-simple-swift(-Xfrontend -enable-experimental-concurrency) | %FileCheck %s --dump-input always
+// REQUIRES: executable_test
+// REQUIRES: concurrency
+// REQUIRES: OS=macosx
+
+import Dispatch
+
+// ==== ------------------------------------------------------------------------
+// MARK: "Infrastructure" for the tests
+
+extension DispatchQueue {
+  func async<R>(operation: @escaping () async -> R) -> Task.Handle<R> {
+    let handle = Task.runDetached(operation: operation)
+
+    // Run the task
+    _ = { self.async { handle.run() } }() // force invoking the non-async version
+
+    return handle
+  }
+}
+
+@available(*, deprecated, message: "This is a temporary hack")
+@discardableResult
+func launch<R>(operation: @escaping () async -> R) -> Task.Handle<R> {
+  let handle = Task.runDetached(operation: operation)
+
+  // Run the task
+  _ = DispatchQueue.global(priority: .default).async { handle.run() }
+
+  return handle
+}
+
+// ==== ------------------------------------------------------------------------
+// MARK: Tests
+
+func test_taskGroup_cancel_internally_afterOne() {
+  let taskHandle = launch { () async -> Int in
+    return await try! Task.withGroup(resultType: Int.self) { (group) async -> Int in
+
+      func submit(_ n: Int) async {
+        await group.add {
+          do {
+            await try Task.checkCancellation()
+            print("submitted: \(n)")
+            return n
+          } catch {
+            print("submit failed: \(error)")
+            throw error
+          }
+        }
+      }
+
+      func pull() async -> Result<Int, Error>? {
+        do {
+          if let r = await try group.next() {
+            return .success(r)
+          } else {
+            return nil
+          }
+        } catch {
+          return .failure(error)
+        }
+      }
+
+      await submit(1)
+
+      var sum = 0
+      while let r = await pull() {
+        print("next: \(r)")
+        await submit(1)
+
+        if sum >= 2 {
+          print("task group returning: \(sum)")
+          return sum
+        }
+
+        sum += 1
+        group.cancelAll() // cancel all after we receive one element
+      }
+
+      print("bad, task group returning: \(sum)")
+      return 0
+    }
+  }
+
+  // CHECK: main task
+  // CHECK: next: 1
+  // CHECK: next: 2
+  // CHECK: task group returning: 3
+
+  launch { () async in
+    let sum = await try! taskHandle.get()
+    // CHECK: result: 2
+    print("result: \(sum)")
+    exit(0)
+  }
+
+  print("main task")
+}
+
+test_taskGroup_cancel_internally_afterOne()
+
+dispatchMain()

test/Concurrency/async_task_groups.swift

@@ -33,11 +33,11 @@ func test_taskGroup_add() async throws -> Int {
 
 func test_taskGroup_addHandles() async throws -> Int {
   await try Task.withGroup(resultType: Int.self) { group in
-    let one = await group.addWithHandle {
+    let one = await group.add {
       await asyncFunc()
     }
 
-    let two = await group.addWithHandle {
+    let two = await group.add {
       await asyncFunc()
     }
 
@@ -48,11 +48,11 @@ func test_taskGroup_addHandles() async throws -> Int {
 
 func test_taskGroup_cancel_handles() async throws {
   await try Task.withGroup(resultType: Int.self) { group in
-    let one = await group.addWithHandle {
+    let one = await group.add {
       await try asyncThrowsOnCancel()
     }
 
-    let two = await group.addWithHandle {
+    let two = await group.add {
       await asyncFunc()
     }