Thanks for pushing this forward, and for the updated pitch. Addressing the cancellation handling is really important since the current implementation leads to enqueued jobs outliving the cancellation which can be an attack vector in services.
I want to step back and look at the overall shape, because I think several of the moving parts here are connected and point toward a cleaner division of responsibility between the runtime and executors.
The problems I see with this updated pitch:
- Cancellation lives outside the executor. In the current design
Task.sleep needs an AtomicSleepCancellationState, a var token, and a withTaskCancellationHandler wrapping the whole suspension. All of that exists only because the executor can't install a cancellation handler from inside its enqueue method, so the call site has to bridge cancellation to the executor through a token and a hand-rolled state machine. I think this is entirely unnecessary since executors already track state of jobs, their deadlines and any additional resources. More importantly, the executors already need to synchronize this state so the atomic in the runtime can be avoided.
- As a result of (1), the implementation of throwing a
CancellationError in Task.sleep is actually racy. Since the executor is not responsible for handling the cancellation the code inside Task.sleep calls try Task.checkCancellation() after the executor resumed the job. This check might throw a CancellationError even though the timer successfully ran and was cancelled in the window between the executor resuming the job and the cancellation check. Only the executor actually knows if the job ran successfully or was cancelled.
cleanUp on JobCancellationToken externalizes executor state. The DispatchSource lifetime is the executor's concern. The executor knows when the job runs and when cancel(jobWithToken:) is called so it can release the source itself. A cleanup closure in the public API feels to me like a workaround for Dispatch not being a first-class SchedulingExecutor, and I don't think that implementation detail should surface in the public API.CancellationBehavior.drop isn't valid for task-based jobs. Every job we enqueue today in this new proposed API is a task, and dropping one hangs the task forever. I personally think we should remove it now and re-propose it once a use-case arises.
In my opinion, all four are symptoms of the same thing: we're doing work in the runtime, and pushing state into the job and the token, that the executor should simply own. Executors already manage exactly this kind of state directly: Dispatch already tracks its sources, a cooperative executor already has its timer queues, a native pthread based executor also tracks its jobs and deadlines, and any executor that can be cancelled from another thread already has synchronization for its bookkeeping. In most cases, cancellation state is just a jobID -> source (or a small pending-jobs map) next to state the executor already keeps, protected by the synchronization primitive it already uses.
So I don't think we need to smuggle executor state through the job's private data or a token's opaqueData/cleanUp at all. Keying off job.id in the executor's own structure is enough, and it's strictly less machinery than reconstructing a parallel token + cleanup + drop/execute policy in the runtime.
As a concrete alternative approach I was thinking we could do this instead:
job.addTaskCancellationHandler { } / job.removeTaskCancellationHandler(_:): A synchronous pair on ExecutorJob that installs (and later removes) a task-cancellation handler using the task context already embedded in the job. This mirrors how withTaskCancellationHandler is already built (add on entry, remove on exit); we're just exposing it so an executor can straddle its own enqueue/run boundary instead of a lexical scope.
The important constraint is that neither enqueue nor the cancellation handler ever runs the job inline. The handler can fire on another thread, and, in the already-cancelled case, it can fire before the task has reached its suspension point, so resuming there would resume a continuation that hasn't suspended yet. Instead, both paths just move the job into the executor's ready state and let the next tick of the executor's own loop actually run it. That keeps a single, clean invariant: jobs are only ever run from the executor's run loop.
// Pseudocode
enum PendingJob {
case reserved
case stored(ExecutorJob, TaskCancellationHandler)
case cancelled(ExecutorJob?) // nil only transiently, until enqueue completes
}
func enqueue(_ job: consuming ExecutorJob, run at: FireTime<C>, ...) {
let jobID = job.id
pendingJobs[jobID] = .reserved // under the executor's lock
let handler = job.addTaskCancellationHandler { // may fire now, or later, on any thread
switch pendingJobs.removeValue(forKey: jobID) {
case .stored(let job, _): pendingJobs[jobID] = .cancelled(job); wakeUp()
case .reserved: pendingJobs[jobID] = .cancelled(nil) // cancelled mid-enqueue
case .cancelled: preconditionFailure("handler fired twice") // genuine invariant violation
case nil: break // loop already claimed it (timer fired) — no-op
}
}
switch pendingJobs.removeValue(forKey: jobID) {
case .reserved: pendingJobs[jobID] = .stored(consume job, handler)
case .cancelled: pendingJobs[jobID] = .cancelled(consume job); wakeUp()
case .stored, nil: preconditionFailure("unreachable")
}
}
On a loop tick the executor removes any job that is either cancelled or the deadline has passed:
// on a loop tick, for a job whose deadline passed or that was cancelled:
switch pendingJobs.removeValue(forKey: jobID) {
case .stored(let job, let handler): // fired normally
job.removeTaskCancellationHandler(handler)
job.runSynchronously(on: self, with: .success(()))
case .cancelled(let job?): // cancelled; deliver the error
job.runSynchronously(on: self, with: .failure(CancellationError()))
case .cancelled(nil), .reserved, nil:
preconditionFailure("unreachable at run time")
}
Which leads me to the second part that executors should be able to pass/return values to jobs they are running. This will allow the executor to correctly inform the job if it got cancelled or not.
TypedExecutorJob<T, E: Error> with runSynchronously(on:, with: Result<T, E>): An executor-side counterpart of UnsafeContinuation<T, E>. A job is fundamentally a function thunk, so we should be able to pass a value in when we run it, rather than storing a buffer in the job or threading the result through a side channel. The executor delivers the outcome directly through the job, which removes the racy trailing Task.checkCancellation().
I believe with both of these changes we can:
- Remove a lot of the proposed API surface
- Make the implementation more performant in the executor implementations
- Pave the way for a more general I/O executor which also needs to return hypothetical
IOResult and IOError to the job.
On Dispatch specifically
The current implementation also uses multiple dispatch sources in the global Dispatch-backed executor. That is far from ideal since every source is at least one allocation and one kernel object. We should be able to use just one source per priority that is armed to the closest deadline.
I'd rather we make Dispatch implement SchedulingExecutor natively than keep runtime machinery that compensates for it not doing so. Dispatch owns its timer sources; if it implements the interface directly it manages source lifetime and cancellation itself, on its own queues, which is also the highest-performance option. This is similar to how we would implement it in the PThreadExecutor in GitHub - swiftlang/swift-platform-executors: This package provides platform-native executors for Swift Concurrency. · GitHub.