Cancellable interface for FIFOQueue

[vladyslavsosiuk]

Goals

• Introduce a cancellable interface to the existing FIFOQueue implementation.

Technical Details

• Tasks can be cancelled before they start, and the cancellation will be propagated correctly.
• Task cancellation is handled using withTaskCancellationHandler to ensure proper cleanup.

API Overview

    public func enqueue<T: Sendable>(
        _ task: @escaping @Sendable () async throws -> T
    ) async throws -> T

Example Usage

let queue = FIFOQueue()

let task = Task {
    do {
        let result = try await queue.enqueueWithCancellation {
            try Task.sleep(seconds: 1)
        }
        print("Task completed with result: \(result)")
    } catch {
        print("Task was cancelled or failed with error: \(error)")
    }
}
task.cancel()

I have developed a similar queue functionality that has a cancellable enqueue method implementation. This allowed tasks to be enqueued with the ability to cancel them if needed either when they already started or when they are waiting in the queue. I believe this feature could be a valuable addition to your library. Would you be interested in me opening a pull request with this functionality?

Update:
I discovered that while implementing cancellation is possible, it loses the ability to propagate Task locals to the enqueued tasks. Extra work is required to hold a handle on a task to cancel it without starting it immediately and this involves starting a task from the escaped boundary, which means it doesn't retain locals of the tasks that enqueued the task. I believe this behavior would be unexpected and confusing with the proposed interface moving forward.

[dfed]

I like the idea of adding cancellation! I've played with it locally over the years but haven't prioritized because I haven't hit this need myself.

This repo's queues are very much in the world of Unstructured Concurrency. I don't think we necessarily want to introduce structured concurrency where tasks created within a particular context can all be cancelled in one fell swoop.

Thoughts on an API that looks like:

public func enqueue<Success: Sendable, Failure: Error>(_ task: @escaping @Sendable () async -> Void) -> FIFOTask<Success, Failure>

Where the API for FIFOTask looks vaguely like a Task:

public struct FIFOTask<Success, Failure> {
    public var value: Success { get async throws }

    public var isCancelled: Bool { get }

    public func cancel()
}

extension FIFOTask where Failure == Never {
    public var value: Success { get async }
}

extension FIFOTask where Success == Never, Failure == Never {
    public static var isCancelled: Bool { get }
    public static func checkCancellation() throws
}

Since FIFO queues are definitionally FIFO, implementing the static methods might be pretty straightforward. My only qualm with this approach is that I don't think it scales nicely to the ActorQueue type. Thoughts?

[vladyslavsosiuk]

Thanks for the response! Your approach with FIFOTask introduces a separate task-like abstraction, which is quite different from what I originally intended. My idea was to allow the cancellation of an individual enqueuement in the FIFOQueue. The main challenge I encountered was that starting tasks from an escaped closure breaks Task local propagation, which may be unexpected behavior for the consumers of the library.

Your proposed FIFOTask abstraction shifts the model away from simply making the enqueue function of FIFOQueue cancellable and this changes how tasks interact with the queue rather than just enabling cancellation.

[dfed]

My idea was to allow the cancellation of an individual enqueuement in the FIFOQueue

My proposal does allow for this!

The main challenge I encountered was that starting tasks from an escaped closure breaks Task local propagation

Can you give me some examples of how you would task cancellation propagation to work within this library? I think I'm missing something here. Given that these tasks aren't structured – we're not using child tasks – we're firmly in the world of Unstructured Concurrency, which doesn't propagate outer cancellation to inner tasks. For example, this unit test fails:

@Test func test_taskCancellation() async {
    let semaphore = Semaphore()
    let outerTask = Task {
        let innerTask = Task {
            await semaphore.wait() // wait until after the outer task has been cancelled to continue.
            #expect(Task.isCancelled) // Fails! If outer task cancellation propagated to tasks declared within the outer task, then this would succeed.
        }
        await innerTask.value
        #expect(Task.isCancelled) // Passes!
    }
    outerTask.cancel()
    await semaphore.signal()
    await outerTask.value
}

[vladyslavsosiuk]

You would need to wrap the task closure in a private LazyTask class—similar to what you described, but private to ensure its internal logic for lazy execution and cancellation remains correctly isolated within the queue. Then, you would wrap withUnsafeContinuation inside withTaskCancellationHandler. In the onCancel closure, you would call the cancel function of LazyTask. Finally, the LazyTask would only be started within the closure yielded to the stream.

However, while writing this, I realized that FIFOQueue is not an actor, and the async sequence iteration runs in a detached task. Due to these implementation details, achieving the above might not be feasible, as there’s likely no way to guarantee that start and cancel won’t interleave, potentially leading to undefined behavior.

extension FIFOQueue {
    private class LazyTask {
        let operation: () async throws -> Void
        var task: Task<Void, Error>?
        var isCancelled: Bool = false

        func start() async throws {
            precondition(task == nil)
            if isCancelled { throw CancellationError() }
            task = Task { try await operation() }
            return try await task!.value
        }

        func cancel() {
            isCancelled = true
            task?.cancel()
        }
    }
}

[dfed]

Got it! That's an elucidating bit of code. And yes I agree with your statement that the ordering here is difficult to get right. Additionally, the cancellation semantic you describe departs somewhat significantly from how Task cancellation works, since cancellation only occurs if the task is cancelled before the task begins executing.

I think the general goal here should be to emulate the cancellation that Swift Concurrency provides, given that this library's goal is just trying to improve the ordering of tasks enqueued from nonisolated regions to isolated regions. Does that make sense?

Full disclosure: I do not have an immediate need for this API, and my bandwidth is currently pretty limited, so I am not likely to implement any of the APIs proposed in this library in the near term. I am happy to continue discussing what we think would be a good API though, in the hopes that someone else might have the time to offer an implementation.

[vladyslavsosiuk]

I think the general goal here should be to emulate the cancellation that Swift Concurrency provides, given that this library's goal is just trying to improve the ordering of tasks enqueued from nonisolated regions to isolated regions. Does that make sense?

Yes.

To address the data race I think LazyTask can leverage NSRecursiveLock or even NSLock:

extension FIFOQueue {
    private class LazyTask {
        let operation: () async throws -> Void
        var task: Task<Void, Error>?
        var isCancelled: Bool = false
        let lock = NSRecursiveLock()

        func start() async throws {
            try await withLock {
                precondition(task == nil)
                if isCancelled { 
                    throw CancellationError() 
                }
                task = Task { try await operation() }
                return task
            }!.value
        }

        func cancel() {
            withLock {
                isCancelled = true
                task?.cancel()
            }
        }

        private func withLock<T>(_ operation: () throws -> T) rethrows -> T {
            defer { lock.unlock() }
            lock.lock()
            return operation()
        }
    }
}

This should help avoiding data races, the cancellation would be implemented using the withTaskCancellationHandler.

If that looks good I can work on a PR implementing this.

[dfed]

Apologies for the slow response. Work's kept me busy!

I think we may be misunderstanding one another a little here. I do not think AsyncQueue should have an API that inherits cancellation from a parent context, since AsyncQueue enqueued tasks are more akin to a Task than a structured concurrency TaskGroup.

I do think that our tasks should be cancellable. But via a direct cancel() API, similar to Task. The best possible API is likely to have both FIFO queue and Actor queue methods return an actual Task that waits its turn in the underlying queue structure, allows for cancellation. That would be a breaking change, but given that we're still beta and the above would unlock a much nicer API, it might be worthy of such a thing.

[dfed]

Basically thinking about having the following API for both FIFOQueue and ActorQueue:

Task(enqueuedOn: queue) {
    // Your task here.
}

This way you get all of the semantics of Task, but with the benefits of enqueueing the tasks in order. I'm hopeful I can get this to work, but I'm not 100% sure yet.

[dfed]

PR is out! Curious what you think of this approach.