spawnDetached to detach, non-suspending group.spawn, spawnUnlessCancelled

This commit is contained in:
Konrad `ktoso` Malawski
2021-03-26 23:37:43 +09:00
parent 31144b9a52
commit 34f8e768b5
39 changed files with 262 additions and 136 deletions

View File

@@ -105,7 +105,7 @@ extension Task {
/// ### Priority inheritance
/// Child tasks automatically inherit their parent task's priority.
///
/// Detached tasks (created by `spawnDetached`) DO NOT inherit task priority,
/// Detached tasks (created by `detach`) DO NOT inherit task priority,
/// as they are "detached" from their parent tasks after all.
///
/// ### Priority elevation
@@ -150,7 +150,7 @@ extension Task {
/// i.e. the task will run regardless of the handle still being present or not.
/// Dropping a handle however means losing the ability to await on the task's result
/// and losing the ability to cancel it.
public struct Handle<Success, Failure: Error> {
public struct Handle<Success, Failure: Error>: Sendable {
internal let _task: Builtin.NativeObject
internal init(_ task: Builtin.NativeObject) {
@@ -395,7 +395,7 @@ extension Task {
/// throw the error the operation has thrown when awaited on.
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
@discardableResult
public func spawnDetached<T>(
public func detach<T>(
priority: Task.Priority = .default,
operation: __owned @Sendable @escaping () async -> T
) -> Task.Handle<T, Never> {
@@ -447,7 +447,7 @@ public func spawnDetached<T>(
/// tasks result or `cancel` it. If the operation fails the handle will
/// throw the error the operation has thrown when awaited on.
@discardableResult
public func spawnDetached<T, Failure>(
public func detach<T, Failure>(
priority: Task.Priority = .default,
operation: __owned @Sendable @escaping () async throws -> T
) -> Task.Handle<T, Failure> {
@@ -472,7 +472,7 @@ public func spawnDetached<T, Failure>(
// TODO: remove this?
public func _runAsyncHandler(operation: @escaping () async -> ()) {
typealias ConcurrentFunctionType = @Sendable () async -> ()
spawnDetached(
detach(
operation: unsafeBitCast(operation, to: ConcurrentFunctionType.self)
)
}
@@ -629,7 +629,7 @@ public func _asyncMainDrainQueue() -> Never
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
public func _runAsyncMain(_ asyncFun: @escaping () async throws -> ()) {
#if os(Windows)
spawnDetached {
detach {
do {
try await asyncFun()
exit(0)
@@ -647,7 +647,7 @@ public func _runAsyncMain(_ asyncFun: @escaping () async throws -> ()) {
}
}
spawnDetached {
detach {
await _doMain(asyncFun)
exit(0)
}
@@ -705,12 +705,12 @@ func _taskIsCancelled(_ task: Builtin.NativeObject) -> Bool
@_alwaysEmitIntoClient
@usableFromInline
internal func _runTaskForBridgedAsyncMethod(_ body: @escaping () async -> Void) {
// TODO: We can probably do better than spawnDetached
// TODO: We can probably do better than detach
// if we're already running on behalf of a task,
// if the receiver of the method invocation is itself an Actor, or in other
// situations.
#if compiler(>=5.5) && $Sendable
spawnDetached { await body() }
detach { await body() }
#endif
}

View File

@@ -61,10 +61,8 @@ import Swift
/// - if the body returns normally:
/// - the group will await any not yet complete tasks,
/// - once the `withTaskGroup` returns the group is guaranteed to be empty.
/// - if the body throws:
/// - all tasks remaining in the group will be automatically cancelled.
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
public func withTaskGroup<ChildTaskResult, GroupResult>(
public func withTaskGroup<ChildTaskResult: Sendable, GroupResult>(
of childTaskResultType: ChildTaskResult.Type,
returning returnType: GroupResult.Type = GroupResult.self,
body: (inout TaskGroup<ChildTaskResult>) async -> GroupResult
@@ -144,7 +142,7 @@ public func withTaskGroup<ChildTaskResult, GroupResult>(
/// - once the `withTaskGroup` returns the group is guaranteed to be empty.
/// - if the body throws:
/// - all tasks remaining in the group will be automatically cancelled.
public func withThrowingTaskGroup<ChildTaskResult, GroupResult>(
public func withThrowingTaskGroup<ChildTaskResult: Sendable, GroupResult>(
of childTaskResultType: ChildTaskResult.Type,
returning returnType: GroupResult.Type = GroupResult.self,
body: (inout ThrowingTaskGroup<ChildTaskResult, Error>) async throws -> GroupResult
@@ -191,7 +189,7 @@ public func withThrowingTaskGroup<ChildTaskResult, GroupResult>(
/// A task group serves as storage for dynamically spawned tasks.
///
/// It is created by the `withTaskGroup` function.
public struct TaskGroup<ChildTaskResult> {
public struct TaskGroup<ChildTaskResult: Sendable> {
private let _task: Builtin.NativeObject
/// Group task into which child tasks offer their results,
@@ -225,13 +223,13 @@ public struct TaskGroup<ChildTaskResult> {
@discardableResult
public mutating func spawn(
overridingPriority priorityOverride: Task.Priority? = nil,
operation: @Sendable @escaping () async -> ChildTaskResult
) async -> Bool {
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 false
return Spawned(handle: nil)
}
// Set up the job flags for a new task.
@@ -252,7 +250,22 @@ public struct TaskGroup<ChildTaskResult> {
// Enqueue the resulting job.
_enqueueJobGlobal(Builtin.convertTaskToJob(childTask))
return true
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>?
init(handle: Task.Handle<ChildTaskResult, Never>?) {
self.handle = handle
}
}
/// Wait for the a child task that was added to the group to complete,
@@ -296,8 +309,8 @@ public struct TaskGroup<ChildTaskResult> {
/// Order of values returned by next() is *completion order*, and not
/// submission order. I.e. if tasks are added to the group one after another:
///
/// await group.spawn { 1 }
/// await group.spawn { 2 }
/// group.spawn { 1 }
/// group.spawn { 2 }
///
/// print(await group.next())
/// /// Prints "1" OR "2"
@@ -388,7 +401,7 @@ public struct TaskGroup<ChildTaskResult> {
/// child tasks.
///
/// It is created by the `withTaskGroup` function.
public struct ThrowingTaskGroup<ChildTaskResult, Failure: Error> {
public struct ThrowingTaskGroup<ChildTaskResult: Sendable, Failure: Error> {
private let _task: Builtin.NativeObject
/// Group task into which child tasks offer their results,
@@ -423,12 +436,12 @@ public struct ThrowingTaskGroup<ChildTaskResult, Failure: Error> {
public mutating func spawn(
overridingPriority priorityOverride: Task.Priority? = nil,
operation: __owned @Sendable @escaping () async throws -> ChildTaskResult
) async -> Bool {
) -> Self.Spawned {
let canAdd = _taskGroupAddPendingTask(group: _group)
guard canAdd else {
// the group is cancelled and is not accepting any new work
return false
return Spawned(handle: nil)
}
// Set up the job flags for a new task.
@@ -449,7 +462,22 @@ public struct ThrowingTaskGroup<ChildTaskResult, Failure: Error> {
// Enqueue the resulting job.
_enqueueJobGlobal(Builtin.convertTaskToJob(childTask))
return true
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
}
}
/// Wait for the a child task that was added to the group to complete,
@@ -493,8 +521,8 @@ public struct ThrowingTaskGroup<ChildTaskResult, Failure: Error> {
/// Order of values returned by next() is *completion order*, and not
/// submission order. I.e. if tasks are added to the group one after another:
///
/// await group.spawn { 1 }
/// await group.spawn { 2 }
/// group.spawn { 1 }
/// group.spawn { 2 }
///
/// print(await group.next())
/// /// Prints "1" OR "2"
@@ -519,6 +547,29 @@ public struct ThrowingTaskGroup<ChildTaskResult, Failure: Error> {
return try await _taskGroupWaitNext(group: _group)
}
/// - SeeAlso: `next()`
public mutating func nextResult() async throws -> Result<ChildTaskResult, Failure>? {
#if NDEBUG
let callingTask = Builtin.getCurrentAsyncTask() // can't inline into the assert sadly
assert(unsafeBitCast(callingTask, to: size_t.self) ==
unsafeBitCast(_task, to: size_t.self),
"""
group.next() invoked from task other than the task which created the group! \
This means the group must have illegally escaped the withTaskGroup{} scope.
""")
#endif
do {
guard let success: ChildTaskResult = try await _taskGroupWaitNext(group: _group) else {
return nil
}
return .success(success)
} catch {
return .failure(error as! Failure) // as!-safe, because we are only allowed to throw Failure (Error)
}
}
/// Query whether the group has any remaining tasks.
///
/// Task groups are always empty upon entry to the `withTaskGroup` body, and
@@ -650,6 +701,7 @@ extension ThrowingTaskGroup: AsyncSequence {
/// - SeeAlso: `ThrowingTaskGroup.next()` for a detailed discussion its semantics.
public mutating func next() async throws -> Element? {
guard !finished else { return nil }
do {
guard let element = try await group.next() else {
finished = true

View File

@@ -52,7 +52,7 @@ func runTest(numCounters: Int, numWorkers: Int, numIterations: Int) async {
var workers: [Task.Handle<Void, Error>] = []
for i in 0..<numWorkers {
workers.append(
spawnDetached { [counters] in
detach { [counters] in
await Task.sleep(UInt64.random(in: 0..<100) * 1_000_000)
await worker(identity: i, counters: counters, numIterations: numIterations)
}

View File

@@ -10,9 +10,9 @@
import Dispatch
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func test_spawnDetached_cancel_child_early() async {
print(#function) // CHECK: test_spawnDetached_cancel_child_early
let h: Task.Handle<Bool, Error> = spawnDetached {
func test_detach_cancel_child_early() async {
print(#function) // CHECK: test_detach_cancel_child_early
let h: Task.Handle<Bool, Error> = detach {
async let childCancelled: Bool = { () -> Bool in
await Task.sleep(2_000_000_000)
return Task.isCancelled
@@ -37,6 +37,6 @@ func test_spawnDetached_cancel_child_early() async {
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
@main struct Main {
static func main() async {
await test_spawnDetached_cancel_child_early()
await test_detach_cancel_child_early()
}
}

View File

@@ -10,8 +10,8 @@
import Dispatch
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func test_spawnDetached_cancel_while_child_running() async {
let h: Task.Handle<Bool, Error> = spawnDetached {
func test_detach_cancel_while_child_running() async {
let h: Task.Handle<Bool, Error> = detach {
async let childCancelled: Bool = { () -> Bool in
await Task.sleep(3_000_000_000)
return Task.isCancelled
@@ -36,6 +36,6 @@ func test_spawnDetached_cancel_while_child_running() async {
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
@main struct Main {
static func main() async {
await test_spawnDetached_cancel_while_child_running()
await test_detach_cancel_while_child_running()
}
}

View File

@@ -11,12 +11,12 @@
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func simple() async {
print("\(#function) -----------------------")
let one = await Task.current()
let two = await Task.current()
let one = Task.current!
let two = Task.current!
print("same equal: \(one == two)") // CHECK: same equal: true
print("hashes equal: \(one.hashValue == two.hashValue)") // CHECK: hashes equal: true
async let x = Task.current()
async let x = Task.current
let three = await x
print("parent/child equal: \(three == two)") // CHECK: parent/child equal: false
@@ -26,12 +26,12 @@ func simple() async {
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func unsafe() async {
print("\(#function) -----------------------")
let one = Task.unsafeCurrent!
let two = Task.unsafeCurrent!
let one = withUnsafeCurrentTask { $0! }
let two = withUnsafeCurrentTask { $0! }
print("unsafe same equal: \(one == two)") // CHECK: same equal: true
print("unsafe hashes equal: \(one.hashValue == two.hashValue)") // CHECK: hashes equal: true
async let x = Task.unsafeCurrent!
async let x = withUnsafeCurrentTask { $0! }
let three = await x
print("unsafe parent/child equal: \(three == two)") // CHECK: parent/child equal: false
@@ -44,8 +44,8 @@ func unsafe() async {
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func unsafeSync() {
print("\(#function) -----------------------")
let one = Task.unsafeCurrent!
let two = Task.unsafeCurrent!
let one = withUnsafeCurrentTask { $0! }
let two = withUnsafeCurrentTask { $0! }
print("unsafe same equal: \(one == two)") // CHECK: same equal: true
print("unsafe hashes equal: \(one.hashValue == two.hashValue)") // CHECK: hashes equal: true
}

View File

@@ -12,7 +12,7 @@
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
@main struct Main {
static func main() async {
let handle = spawnDetached {
let handle = detach {
while (!Task.isCancelled) { // no need for await here, yay
print("waiting")
}

View File

@@ -51,7 +51,7 @@ func groups() async {
// no value in parent, value in child
let x1: Int = try! await withTaskGroup(of: Int.self) { group in
await group.spawn {
group.spawn {
printTaskLocal(\.number) // CHECK: NumberKey: 0 {{.*}}
// inside the child task, set a value
await Task.withLocal(\.number, boundTo: 1) {
@@ -71,7 +71,7 @@ func groups() async {
let x2: Int = try! await withTaskGroup(of: Int.self) { group in
printTaskLocal(\.number) // CHECK: NumberKey: 2 {{.*}}
await group.spawn {
group.spawn {
printTaskLocal(\.number) // CHECK: NumberKey: 2 {{.*}}
async let childInsideGroupChild: () = printTaskLocal(\.number)

View File

@@ -68,14 +68,15 @@ func test_async_group() async {
await Task.withLocal(\.string, boundTo: "top") {
printTaskLocal(\.string) // CHECK: StringKey: top {{.*}}
try! await withTaskGroup(of: Void.self) { group -> Void? in
await withTaskGroup(of: Int.self, returning: Void.self) { group in
printTaskLocal(\.string) // CHECK: StringKey: top {{.*}}
await group.spawn {
group.spawn {
printTaskLocal(\.string) // CHECK: StringKey: <undefined> {{.*}}
return 0
}
return try! await group.next()
_ = await group.next()
}
}
}

View File

@@ -17,7 +17,7 @@ func test_detach() async {
// 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
// task.
await spawnDetached(priority: .userInitiated) {
await detach(priority: .userInitiated) {
let a2 = Task.currentPriority
print("a2: \(a2)") // CHECK: a2: userInitiated
}.get()
@@ -35,16 +35,16 @@ func test_multiple_lo_indirectly_escalated() async {
}
}
let z = spawnDetached(priority: .background) {
let z = detach(priority: .background) {
await loopUntil(priority: .userInitiated)
}
let x = spawnDetached(priority: .background) {
let x = detach(priority: .background) {
_ = await z // waiting on `z`, but it won't complete since we're also background
await loopUntil(priority: .userInitiated)
}
// detach, don't wait
spawnDetached(priority: .userInitiated) {
detach(priority: .userInitiated) {
await x // escalates x, which waits on z, so z also escalates
}

View File

@@ -32,7 +32,7 @@ import Dispatch
static func testSleepDoesNotBlock() async {
// FIXME: Should run on main executor
let task = spawnDetached {
let task = detach {
print("Run first")
}

View File

@@ -3,6 +3,7 @@
// REQUIRES: concurrency
// XFAIL: linux
// XFAIL: windows
struct Boom: Error {}
func boom() async throws -> Int {
@@ -10,12 +11,10 @@ func boom() async throws -> Int {
}
func test_taskGroup_next() async {
let sum: Int = await Task.withGroup(resultType: Int.self) { group in
let sum = await withThrowingTaskGroup(of: Int.self, returning: Int.self) { group in
for n in 1...10 {
await group.add {
return n.isMultiple(of: 3)
? try await boom()
: n
group.spawn {
return n.isMultiple(of: 3) ? try await boom() : n
}
}
@@ -42,12 +41,10 @@ func test_taskGroup_next() async {
}
func test_taskGroup_for_in() async {
let sum: Int = await Task.withGroup(resultType: Int.self) { group in
let sum = await withThrowingTaskGroup(of: Int.self, returning: Int.self) { group in
for n in 1...10 {
await group.add {
return n.isMultiple(of: 3)
? try await boom()
: n
group.spawn {
return n.isMultiple(of: 3) ? try await boom() : n
}
}
@@ -74,12 +71,10 @@ func test_taskGroup_for_in() async {
}
func test_taskGroup_asyncIterator() async {
let sum: Int = await Task.withGroup(resultType: Int.self) { group in
let sum = await withThrowingTaskGroup(of: Int.self, returning: Int.self) { group in
for n in 1...10 {
await group.add {
return n.isMultiple(of: 3)
? try await boom()
: n
group.spawn {
return n.isMultiple(of: 3) ? try await boom() : n
}
}

View File

@@ -21,7 +21,7 @@ func test_taskGroup_cancelAll_onlySpecificGroup() async {
async let g1: Int = withTaskGroup(of: Int.self) { group in
for i in 1...5 {
await group.spawn {
group.spawn {
await Task.sleep(1_000_000_000)
let c = Task.isCancelled
print("add: \(i) (cancelled: \(c))")
@@ -46,7 +46,7 @@ func test_taskGroup_cancelAll_onlySpecificGroup() async {
// The cancellation os g2 should have no impact on g1
let g2: Int = try! await withTaskGroup(of: Int.self) { group in
for i in 1...3 {
await group.spawn {
group.spawn {
await Task.sleep(1_000_000_000)
let c = Task.isCancelled
print("g1 task \(i) (cancelled: \(c))")

View File

@@ -21,20 +21,20 @@ func test_taskGroup_cancel_from_inside_child() async {
}
let result = await withTaskGroup(of: Int.self, returning: Int.self) { group in
let firstAdded = await group.spawn { [group] in // must explicitly capture, as the task executes concurrently
let firstAdded = group.spawn { [group] in // must explicitly capture, as the task executes concurrently
group.cancelAll() // allowed
print("first")
return 1
}
print("firstAdded: \(firstAdded)") // CHECK: firstAdded: true
print("firstAdded: \(firstAdded.successfully)") // CHECK: firstAdded: true
let one = await group.next()
let secondAdded = await group.spawn {
let secondAdded = group.spawn {
print("second")
return 2
}
print("secondAdded: \(secondAdded)") // CHECK: secondAdded: false
print("secondAdded: \(secondAdded.successfully)") // CHECK: secondAdded: false
return 1
}

View File

@@ -0,0 +1,51 @@
// RUN: %target-run-simple-swift(-Xfrontend -enable-experimental-concurrency %import-libdispatch -parse-as-library) | %FileCheck %s --dump-input always
// REQUIRES: executable_test
// REQUIRES: concurrency
// REQUIRES: libdispatch
import Dispatch
func asyncEcho(_ value: Int) async -> Int {
value
}
func test_taskGroup_cancel_then_add() async {
// CHECK: test_taskGroup_cancel_then_add
print("\(#function)")
let result: Int = await withTaskGroup(of: Int.self) { group in
let addedFirst = group.spawn {
while !Task.isCancelled {
await Task.sleep(50)
}
print("first done")
return 1
}
print("added first: \(addedFirst.successfully)") // CHECK: added first: true
print("cancel first") // CHECK: cancel first
addedFirst.handle?.cancel() // CHECK: first done
let one = await group.next()!
print("next first: \(one)") // CHECK: next first: 1
let addedSecond = group.spawn { 2 }
print("added second: \(addedSecond.successfully)") // CHECK: added second: true
let none = await group.next()!
print("next second: \(none)") // CHECK: next second: 2
return 0
}
print("result: \(result)") // CHECK: result: 0
}
@main struct Main {
static func main() async {
await test_taskGroup_cancel_then_add()
}
}

View File

@@ -17,9 +17,9 @@ func asyncEcho(_ value: Int) async -> Int {
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func test_taskGroup_cancel_parent_affects_group() async {
let x = spawnDetached {
let x = detach {
await withTaskGroup(of: Int.self, returning: Void.self) { group in
await group.spawn {
group.spawn {
await Task.sleep(3_000_000_000)
let c = Task.isCancelled
print("group task isCancelled: \(c)")

View File

@@ -20,8 +20,8 @@ func test_taskGroup_cancel_then_add() async {
print("\(#function)")
let result: Int = await withTaskGroup(of: Int.self) { group in
let addedFirst = await group.spawn { 1 }
print("added first: \(addedFirst)") // CHECK: added first: true
let addedFirst = group.spawn { 1 }
print("added first: \(addedFirst.successfully)") // CHECK: added first: true
let one = await group.next()!
print("next first: \(one)") // CHECK: next first: 1
@@ -29,8 +29,8 @@ func test_taskGroup_cancel_then_add() async {
group.cancelAll()
print("cancelAll")
let addedSecond = await group.spawn { 1 }
print("added second: \(addedSecond)") // CHECK: added second: false
let addedSecond = group.spawn { 1 }
print("added second: \(addedSecond.successfully)") // CHECK: added second: false
let none = await group.next()
print("next second: \(none)") // CHECK: next second: nil

View File

@@ -14,32 +14,45 @@ func asyncEcho(_ value: Int) async -> Int {
value
}
// FIXME: this is a workaround since (A, B) today isn't inferred to be Sendable
// and causes an error, but should be a warning (this year at least)
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
struct SendableTuple2<A: Sendable, B: Sendable>: Sendable {
let first: A
let second: B
init(_ first: A, _ second: B) {
self.first = first
self.second = second
}
}
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func test_taskGroup_cancel_then_completions() async {
// CHECK: test_taskGroup_cancel_then_completions
print("before \(#function)")
let result: Int = await withTaskGroup(of: (Int, Bool).self) { group in
let result: Int = await withTaskGroup(of: SendableTuple2<Int, Bool>.self) { group in
print("group cancelled: \(group.isCancelled)") // CHECK: group cancelled: false
let addedFirst = await group.spawn {
let spawnedFirst = group.spawn {
print("start first")
await Task.sleep(1_000_000_000)
print("done first")
return (1, Task.isCancelled)
return SendableTuple2(1, Task.isCancelled)
}
print("added first: \(addedFirst)") // CHECK: added first: true
assert(addedFirst)
print("spawned first: \(spawnedFirst.successfully)") // CHECK: spawned first: true
assert(spawnedFirst.successfully)
let addedSecond = await group.spawn {
let spawnedSecond = group.spawn {
print("start second")
await Task.sleep(3_000_000_000)
print("done second")
return (2, Task.isCancelled)
return SendableTuple2(2, Task.isCancelled)
}
print("added second: \(addedSecond)") // CHECK: added second: true
assert(addedSecond)
print("spawned second: \(spawnedSecond.successfully)") // CHECK: spawned second: true
assert(spawnedSecond.successfully)
group.cancelAll() // FIXME: dont make it async
group.cancelAll()
print("cancelAll") // CHECK: cancelAll
// let outerCancelled = await outer // should not be cancelled
@@ -47,13 +60,13 @@ func test_taskGroup_cancel_then_completions() async {
// print("group cancelled: \(group.isCancelled)") // COM: CHECK: outer cancelled: false
let one = await group.next()
print("first: \(one)") // CHECK: first: Optional((1,
print("first: \(one)") // CHECK: first: Optional(main.SendableTuple2<Swift.Int, Swift.Bool>(first: 1,
let two = await group.next()
print("second: \(two)") // CHECK: second: Optional((2,
print("second: \(two)") // CHECK: second: Optional(main.SendableTuple2<Swift.Int, Swift.Bool>(first: 2,
let none = await group.next()
print("none: \(none)") // CHECK: none: nil
return (one?.0 ?? 0) + (two?.0 ?? 0) + (none?.0 ?? 0)
return (one?.first ?? 0) + (two?.first ?? 0) + (none?.first ?? 0)
}
print("result: \(result)") // CHECK: result: 3

View File

@@ -15,7 +15,7 @@ func test_taskGroup_is_asyncSequence() async {
let sum = await withTaskGroup(of: Int.self, returning: Int.self) { group in
for n in 1...10 {
await group.spawn {
group.spawn {
print("add \(n)")
return n
}
@@ -38,7 +38,7 @@ func test_throwingTaskGroup_is_asyncSequence() async throws {
let sum = try await withThrowingTaskGroup(of: Int.self, returning: Int.self) { group in
for n in 1...10 {
await group.spawn {
group.spawn {
print("add \(n)")
return n
}

View File

@@ -21,7 +21,7 @@ func test_taskGroup_isEmpty() async {
// CHECK: before add: isEmpty=true
print("before add: isEmpty=\(group.isEmpty)")
await group.spawn {
group.spawn {
await Task.sleep(2_000_000_000)
return await asyncEcho(1)
}

View File

@@ -16,7 +16,7 @@ func test_skipCallingNext_butInvokeCancelAll() async {
let result = try! await withTaskGroup(of: Int.self) { (group) async -> Int in
for n in numbers {
print("group.spawn { \(n) }")
await group.spawn { [group] () async -> Int in
group.spawn { [group] () async -> Int in
await Task.sleep(1_000_000_000)
print(" inside group.spawn { \(n) }")
print(" inside group.spawn { \(n) } (group cancelled: \(group.isCancelled))")

View File

@@ -16,7 +16,7 @@ func test_skipCallingNext() async {
let result = try! await withTaskGroup(of: Int.self) { (group) async -> Int in
for n in numbers {
print("group.spawn { \(n) }")
await group.spawn { () async -> Int in
group.spawn { () async -> Int in
await Task.sleep(1_000_000_000)
let c = Task.isCancelled
print(" inside group.spawn { \(n) } (canceled: \(c))")

View File

@@ -17,7 +17,7 @@ func test_sum_nextOnCompleted() async {
let sum = try! await withTaskGroup(of: Int.self) {
(group) async -> Int in
for n in numbers {
await group.spawn {
group.spawn {
() async -> Int in
print(" complete group.spawn { \(n) }")
return n

View File

@@ -24,7 +24,7 @@ func test_sum_nextOnPending() async {
let sum = try! await withTaskGroup(of: Int.self) { (group) async -> Int in
for n in numbers {
await group.spawn {
group.spawn {
let res = await completeSlowly(n: n)
return res
}

View File

@@ -22,7 +22,7 @@ func boom() async throws -> Int {
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func test_taskGroup_throws() async {
let got: Int = try await withThrowingTaskGroup(of: Int.self) { group in
await group.spawn { try await boom() }
group.spawn { try await boom() }
do {
while let r = try await group.next() {
@@ -34,7 +34,7 @@ func test_taskGroup_throws() async {
let gc = group.isCancelled
print("group cancelled: \(gc)")
await group.spawn { () async -> Int in
group.spawn { () async -> Int in
let c = Task.isCancelled
print("task 3 (cancelled: \(c))")
return 3

View File

@@ -20,9 +20,9 @@ func boom() async throws -> Int { throw Boom() }
func test_taskGroup_throws_rethrows() async {
do {
let got = try await withThrowingTaskGroup(of: Int.self, returning: Int.self) { group in
await group.spawn { await echo(1) }
await group.spawn { await echo(2) }
await group.spawn { try await boom() }
group.spawn { await echo(1) }
group.spawn { await echo(2) }
group.spawn { try await boom() }
do {
while let r = try await group.next() {

View File

@@ -26,7 +26,7 @@ func testSimple(
var completed = false
let taskHandle: Task.Handle<String, Error> = spawnDetached {
let taskHandle: Task.Handle<String, Error> = detach {
let greeting = await formGreeting(name: name)
// If the intent is to test suspending, wait a bit so the second task

View File

@@ -26,7 +26,7 @@ struct App {
var n = 0
for _ in 1...NUM_TASKS {
let c = Capture()
let r = spawnDetached {
let r = detach {
c.doSomething()
}
await r.get()

View File

@@ -32,7 +32,7 @@ actor Container {
for _ in 0..<n {
let id = generation
generation += 1
let t = spawnDetached { [weak self] in
let t = detach { [weak self] in
let r = Runner()
await r.run()
await self?.remove(id)

View File

@@ -26,11 +26,11 @@ func asyncFib(_ n: Int) async -> Int {
return n
}
let first = spawnDetached {
let first = detach {
await asyncFib(n - 2)
}
let second = spawnDetached {
let second = detach {
await asyncFib(n - 1)
}

View File

@@ -211,7 +211,7 @@ struct MyGlobalActor {
// expected-error@+3{{actor-isolated var 'number' cannot be passed 'inout' to 'async' function call}}
// expected-error@+2{{var 'number' isolated to global actor 'MyGlobalActor' can not be used 'inout' from a non-isolated context}}
if #available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *) {
let _ = spawnDetached { await { (_ foo: inout Int) async in foo += 1 }(&number) }
let _ = detach { await { (_ foo: inout Int) async in foo += 1 }(&number) }
}
// attempt to pass global state owned by the global actor to another async function

View File

@@ -706,7 +706,7 @@ class SomeClassWithInits {
}
func hasDetached() {
spawnDetached {
detach {
// okay
await self.isolated() // expected-warning{{cannot use parameter 'self' with a non-sendable type 'SomeClassWithInits' from concurrently-executed code}}
self.isolated() // expected-warning{{cannot use parameter 'self' with a non-sendable type 'SomeClassWithInits' from concurrently-executed code}}

View File

@@ -27,7 +27,7 @@ struct SomeFile: Sendable {
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func test_cancellation_withCancellationHandler(_ anything: Any) async -> PictureData {
let handle: Task.Handle<PictureData, Error> = spawnDetached {
let handle: Task.Handle<PictureData, Error> = detach {
let file = SomeFile()
return await Task.withCancellationHandler(

View File

@@ -21,12 +21,12 @@ func asyncThrowsOnCancel() async throws -> Int {
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func test_taskGroup_add() async throws -> Int {
try await withTaskGroup(of: Int.self) { group in
await group.spawn {
try await withThrowingTaskGroup(of: Int.self) { group in
group.spawn {
await asyncFunc()
}
await group.spawn {
group.spawn {
await asyncFunc()
}
@@ -50,10 +50,10 @@ func boom() async throws -> Int { throw Boom() }
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func first_allMustSucceed() async throws {
let first: Int = try await withTaskGroup(of: Int.self) { group in
await group.spawn { await work() }
await group.spawn { await work() }
await group.spawn { try await boom() }
let first: Int = try await withThrowingTaskGroup(of: Int.self) { group in
group.spawn { await work() }
group.spawn { await work() }
group.spawn { try await boom() }
if let first = try await group.next() {
return first
@@ -71,10 +71,10 @@ func first_ignoreFailures() async throws {
@Sendable func work() async -> Int { 42 }
@Sendable func boom() async throws -> Int { throw Boom() }
let first: Int = try await withTaskGroup(of: Int.self) { group in
await group.spawn { await work() }
await group.spawn { await work() }
await group.spawn {
let first: Int = try await withThrowingTaskGroup(of: Int.self) { group in
group.spawn { await work() }
group.spawn { await work() }
group.spawn {
do {
return try await boom()
} catch {
@@ -119,9 +119,9 @@ func test_taskGroup_quorum_thenCancel() async {
///
/// - Returns: `true` iff `N/2 + 1` followers return `.yay`, `false` otherwise.
func gatherQuorum(followers: [Follower]) async -> Bool {
try! await withTaskGroup(of: Vote.self) { group in
try! await withThrowingTaskGroup(of: Vote.self) { group in
for follower in followers {
await group.spawn { try await follower.vote() }
group.spawn { try await follower.vote() }
}
defer {
@@ -154,7 +154,19 @@ func test_taskGroup_quorum_thenCancel() async {
_ = await gatherQuorum(followers: [Follower("A"), Follower("B"), Follower("C")])
}
// FIXME: this is a workaround since (A, B) today isn't inferred to be Sendable
// and causes an error, but should be a warning (this year at least)
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
struct SendableTuple2<A: Sendable, B: Sendable>: Sendable {
let first: A
let second: B
init(_ first: A, _ second: B) {
self.first = first
self.second = second
}
}
extension Collection where Self: Sendable, Element: Sendable, Self.Index: Sendable {
/// Just another example of how one might use task groups.
@@ -171,7 +183,7 @@ extension Collection where Self: Sendable, Element: Sendable, Self.Index: Sendab
return []
}
return try await withTaskGroup(of: (Int, T).self) { group in
return try await withThrowingTaskGroup(of: SendableTuple2<Int, T>.self) { group in
var result = ContiguousArray<T>()
result.reserveCapacity(n)
@@ -179,9 +191,9 @@ extension Collection where Self: Sendable, Element: Sendable, Self.Index: Sendab
var submitted = 0
func submitNext() async throws {
await group.spawn { [submitted,i] in
group.spawn { [submitted,i] in
let value = try await transform(self[i])
return (submitted, value)
return SendableTuple2(submitted, value)
}
submitted += 1
formIndex(after: &i)
@@ -192,7 +204,9 @@ extension Collection where Self: Sendable, Element: Sendable, Self.Index: Sendab
try await submitNext()
}
while let (index, taskResult) = try await group.next() {
while let tuple = try await group.next() {
let index = tuple.first
let taskResult = tuple.second
result[index] = taskResult
try Task.checkCancellation()

View File

@@ -84,7 +84,7 @@ func test_unsafeThrowingContinuations() async throws {
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func test_detached() async throws {
let handle = spawnDetached() {
let handle = detach() {
await someAsyncFunc() // able to call async functions
}
@@ -94,7 +94,7 @@ func test_detached() async throws {
@available(macOS 9999, iOS 9999, watchOS 9999, tvOS 9999, *)
func test_detached_throwing() async -> String {
let handle: Task.Handle<String, Error> = spawnDetached() {
let handle: Task.Handle<String, Error> = detach() {
try await someThrowingAsyncFunc() // able to call async functions
}

View File

@@ -8,7 +8,7 @@ extension Collection {
return await try withTaskGroup(of: Element.self) { group in
var i = self.startIndex
func doit() async throws {
await group.spawn { [i] in
group.spawn { [i] in
return self[i]
}
}

View File

@@ -64,7 +64,7 @@ func runTest(numCounters: Int, numWorkers: Int, numIterations: Int) async {
var workers: [Task.Handle<Void, Error>] = []
for i in 0..<numWorkers {
workers.append(
spawnDetached { [counters] in
detach { [counters] in
usleep(UInt32.random(in: 0..<100) * 1000)
await worker(identity: i, counters: counters, numIterations: numIterations)
}

View File

@@ -30,7 +30,7 @@ func testSimple(
var completed = false
let taskHandle: Task.Handle<String, Error> = spawnDetached {
let taskHandle: Task.Handle<String, Error> = detach {
let greeting = await formGreeting(name: name)
// If the intent is to test suspending, wait a bit so the second task

View File

@@ -46,7 +46,7 @@ func runTest(numCounters: Int, numWorkers: Int, numIterations: Int) async {
var workers: [Task.Handle<Void, Error>] = []
for i in 0..<numWorkers {
workers.append(
spawnDetached { [counters] in
detach { [counters] in
await worker(identity: i, counters: counters, numIterations: numIterations)
}
)