ibex10
October 1, 2024, 12:09am
41
Summary
Does taskGroup.cancelAll() require active co-operation to finish properly?
Because cancellation is cooperative, i.e. a Task always requires whatever it is executing to explicitly check for cancellation, unwind its stack and return before CancellationError() can be thrown by the continuation.withTaskCancellationHandler() so that you can signal your subsystem to stop what it's doing and return early.
Summary
Post
@Test func raceCondition() async {
let object = CustomClass()
await withTaskGroup(of: Void.self) { group in
for _ in 1...10_000 {
group.addTask { object.property += 1 }
}
}
#expect(object.property == 10_000) // ❌
}
This code compiles just fine in Swift 6 language mode, yet the test will not pass. This is because when doing object.property += 1 there are two steps (reading then setting), and each are individually locked. This allows multiple threads to interleave, causing a thread to set a new value in between another thread's read+set.
Beyond the basics of structured concurrency
Summary
Benefits of cooperative multitasking
This is a significant distinction, but it's not the distinction that defines "cooperative" multitasking.
At least in the Apple world, cooperative multitasking means non-preemptive multitasking. Preemptive multitasking is about ejecting the current execution from a thread on a schedule , while cooperative multitasking is about ejecting the current execution from a thread at a point chosen by the current execution . Both deal with the behavior of single threads.
Summary
Post
On all of our supported platforms, Swift co-exists with C and is built on top of an underlying C runtime. In particular, all work done by Swift is actually running on some C thread. C threads are typically scheduled preemptively, usually directly by the kernel, and there is always a potential for there to be threads in a process that are not running Swift code. Thus, at a low level, all Swift work is scheduled preemptively.
Summary
Yes, but with some caveats.
There are potential suspension points on entry/exit edges between async functions because of the possibility for isolation change, and those are not explicit in the source of the function. However, since every call to an async function has to be awaited, you can think of that await as also covering those points.
A somewhat more egregious case is that there's an implicit suspension point when an async let goes out of scope. Normally , that suspension point does not actually suspend because you've already awaited the async let. But there can be paths out of the scope that haven't passed through that await (e.g. throwing an error), and the containing function still has to wait for the subtask to finish before those paths can continue. In that proposal, we just decided that this was an acceptable deviation from the general rule that there's a visible await.
Like resuming a CheckedContinuation, yielding into an AsyncStream's Continuation is not a potential suspension point: it unblocks any tasks waiting on the stream and allows them to be scheduled again, but the current task continues running without interruption.
Summary
I am trying to execute the submission of tickets in serial order however it prints it in random sequence.
actor TicketSubmission {
let executor: TicketExecutor
let queueManager: TicketExectionManager
let queueIdentifier = "com.happy.ticket.queue"
init() {
self.executor = TicketExecutor(queue: DispatchQueue(label: queueIdentifier))
self.queueManager = TicketExectionManager(ticketExecutor: executor)
}
func submitTicket(ticketId: String) async {
await queueManager.execute {
print("Ticket id \(ticketId)")
}
}
}
final class TicketExecutor: SerialExecutor {
private let queue: DispatchQueue
init(queue: DispatchQueue) {
self.queue = queue
}
func asUnownedSerialExecutor() -> UnownedSerialExecutor {
UnownedSerialExecutor(ordinary: self)
}
func enqueue(_ job: consuming ExecutorJob) {
let unownedJob = UnownedJob(job)
let unownedExecutor = asUnownedSerialExecutor()
queue.async {
unownedJob.runSynchronously(on: unownedExecutor)
}
}
}
actor TicketExectionManager {
private let ticketExecutor: TicketExecutor
init(ticketExecutor: TicketExecutor) {
self.ticketExecutor = ticketExecutor
}
nonisolated var unownedExecutor: UnownedSerialExecutor {
ticketExecutor.asUnownedSerialExecutor()
}
func execute(action: () async -> Void ) async {
await action()
}
}
let ticketSubmission = TicketSubmission()
print("Start")
Task {
await ticketSubmission.submitTicket(ticketId: "1")
}
Task {
await ticketSubmission.submitTicket(ticketId: "2")
}
Task {
await ticketSubmission.submitTicket(ticketId: "3")
}
print("End")
It should print in order 1 , 2 and 3.
Summary
Here is the revised version using the AsyncStream.
func suspendResumeTasks (M: Int = 5) async throws {
let chan = Channel <CheckedContinuation <Void, Never>> ()
for i in 0..<M {
Task {
await withCheckedContinuation {
chan.send ($0)
print ("\(i) suspending")
}
print ("\(i) resumed")
}
}
var n = 0
for await cont in chan.stream {
cont.resume()
n += 1
if n == M {
break
}
}
}
struct Channel <T>: Sendable where T:Sendable {
let stream : AsyncStream <T>
let cont : AsyncStream <T>.Continuation
init () {
let u = Self.makeStreamAndCont ()
self.stream = u.0
self.cont = u.1
}
func send (_ u: T?) {
if let u {
cont.yield(u)
}
else {
cont.finish()
}
}
static func makeStreamAndCont () -> (AsyncStream <T>, AsyncStream <T>.Continuation) {
let t = AsyncStream <T>.makeStream()
return t
}
}
Note that my surprise was not about the requirement of active participation of the "child" (which was not, in fact a child). The community has done a good job driving home the fact that you have to participate in cancelation.
What got me was that the task group itself, couldn't return from the await where it was created until all the subtasks had actively co-operated. I just imagined it could return and its child tasks would finish (or not) at some point.
Indeed, the documentation on one version of task group was careful to spell this out, unfortunately, it was the version I wasn't using...