Concurrency 101

[ibex10]

More Interesting Topics

Task Groups

Summary

Does taskGroup.cancelAll() require active co-operation to finish properly?

@layoutSubviews

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.
Depending on your use case, you may need to wrap your continuation with withTaskCancellationHandler() so that you can signal your subsystem to stop what it's doing and return early.

@ktoso

We did a talk specifically around the details of structured concurrency over here: Beyond the basics of structured concurrency - WWDC23 - Videos - Apple Developer

Cancellation does propagate to child tasks, it’s just that they also must react cooperatively. There’s no “interrupt” per se, other than the cancellation handlers which also are just another way to cooperatively react.

Race Conditions May Occur When Setting Properties

Summary

Post

@mbrandonw

@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.

Structured concurrency

Beyond the basics of structured concurrency

Cooperative vs preemptive multitasking

Summary

Benefits of cooperative multitasking

@QuinceyMorris

@dima_kozhinov

You suggest doing much of the work in a thread pool, that is in parallel. But cooperative multitasking is more about one-thread concurrency without parallelism.

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.

Swift Running on C Threads

Summary

Post

@John_McCall

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.

Invisible Suspension Points

Summary

@John_McCall

@Gero

Is it fair to say that all these points are marked with the await keyword?

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.

@Gero

I am especially wondering about a Continuation's yield, as that is not an async method.

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.

Serial Executor

Summary

@swifto

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.

Suspending/Resuming Tasks by Using Continuations

Summary

@ibex10

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
    }
}