swiftswift-concurrency

How to suspend subsequent tasks until first finishes then share its response with tasks that waited?

I have an actor which throttles requests in a way where the first one will suspend subsequent requests until finished, then share its response with them so they don't have to make the same request.

Here's what I'm trying to do:

let cache = Cache()
let operation = OperationStatus()

func execute() async {
    if await operation.isExecuting else {
        await operation.waitUntilFinished()
    } else {
        await operation.set(isExecuting: true)
    }

    if let data = await cache.data {
        return data
    }

    let request = myRequest()
    let response = await myService.send(request)
    await cache.set(data: response)

    await operation.set(isExecuting: false)
}

actor Cache {
    var data: myResponse?

    func set(data: myResponse?) {
        self.data = data
    }
}

actor OperationStatus {
    @Published var isExecuting = false
    private var cancellable = Set<AnyCancellable>()

    func set(isExecuting: Bool) {
        self.isExecuting = isExecuting
    }

    func waitUntilFinished() async {
        guard isExecuting else { return }

        return await withCheckedContinuation { continuation in
            $isExecuting
                .first { !$0 } // Wait until execution toggled off
                .sink { _ in continuation.resume() }
                .store(in: &cancellable)
        }
    }
}

// Do something
DispatchQueue.concurrentPerform(iterations: 1_000_000) { _ in execute() }

This ensures one request at a time, and subsequent calls are waiting until finished. It seems this works but wondering if there's a pure Concurrency way instead of mixing Combine in, and how I can test this? Here's a test I started but I'm confused how to test this:

final class OperationStatusTests: XCTestCase {
    private let iterations = 10_000 // 1_000_000
    private let outerIterations = 10

    actor Storage {
        var counter: Int = 0

        func increment() {
            counter += 1
        }
    }

    func testConcurrency() {
        // Given
        let storage = Storage()
        let operation = OperationStatus()
        let promise = expectation(description: "testConcurrency")
        promise.expectedFulfillmentCount = outerIterations * iterations

        @Sendable func execute() async {
            guard await !operation.isExecuting else {
                await operation.waitUntilFinished()
                promise.fulfill()
                return
            }

            await operation.set(isExecuting: true)
            try? await Task.sleep(seconds: 8)
            await storage.increment()
            await operation.set(isExecuting: false)
            promise.fulfill()
        }

        waitForExpectations(timeout: 10)

        // When
        DispatchQueue.concurrentPerform(iterations: outerIterations) { _ in
            (0..<iterations).forEach { _ in
                Task { await execute() }
            }
        }

        // Then
        // XCTAssertEqual... how to test?
    }
}
Solution

  • Before I tackle a more general example, let us first dispense with some natural examples of sequential execution of asynchronous tasks, passing the result of one as a parameter of the next. Consider:

    func entireProcess() async throws {
    let value = try await first()

    let value2 = try await subsequent(with: value)
    let value3 = try await subsequent(with: value2)
    let value4 = try await subsequent(with: value3)

    // do something with `value4`
}

    Or

    func entireProcess() async throws {
    var value = try await first()

    for _ in 0 ..< 4 {
        value = try await subsequent(with: value)
    }

    // do something with `value`
}

    This is the easiest way to declare a series of async functions, each of which takes the prior result as the input for the next iteration. So, let us expand the above to include some signposts for Instruments’ “Points of Interest” tool:

    import os.log

private let poi = OSSignposter(subsystem: "Test", category: .pointsOfInterest)

func entireProcess() async throws {
    let state = poi.beginInterval(#function, id: poi.makeSignpostID())
    defer { poi.endInterval(#function, state) }

    var value = try await first()

    for i in 0 ..< 4 {
        poi.emitEvent(#function, "Scheduling: \(i) with input of \(value)")
        value = try await subsequent(with: value)
    }
}

func first() async throws -> Int {
    let state = poi.beginInterval(#function, id: poi.makeSignpostID())
    defer { poi.endInterval(#function, state) }

    try await Task.sleep(for: .seconds(1))

    return 42
}

func subsequent(with value: Int) async throws -> Int {
    let state = poi.beginInterval(#function, id: poi.makeSignpostID())
    defer { poi.endInterval(#function, state) }

    try await Task.sleep(for: .seconds(1))

    return value + 1
}

    So, there you see a series of requests that pass their result to the subsequent request. All of that os_signpost signpost stuff is so we can visually see that they are running sequentially in Instrument’s “Points of Interest” tool:

    enter image description here

    You can see ⓢ event signposts as each task is scheduled, and the intervals illustrate the sequential execution of these asynchronous tasks.

    This is the easiest way to have dependencies between tasks, passing values from one task to another.

    Now, that begs the question is how to generalize the above, where we await the prior task before starting the next one.

    One pattern is to write an actor that awaits the result of the prior one. Consider:

    actor SerialTasks<Success> {
    private var previousTask: Task<Success, Error>?

    func add(block: @Sendable @escaping () async throws -> Success) {
        previousTask = Task { [previousTask] in
            let _ = await previousTask?.result
            return try await block()
        }
    }
}

    Unlike the previous example, this does not require that you have a single function from which you initiate the subsequent tasks. E.g., I have used the above when some separate user interaction requires me to add a new task to the end of the list of previously submitted tasks.

    There are two subtle, yet critical, aspects of the above actor:

    1. The add method, itself, must not be an asynchronous function. We need to avoid actor reentrancy. If this were an async function (like in your example), we would lose the sequential execution of the tasks.

    2. The Task has a [previousTask] capture list to capture a copy of the prior task. This way, each task will await the prior one, avoiding any races.

    The above can be used to make a series of tasks run sequentially. But it is not passing values between tasks, itself. I confess that I have used this pattern where I simply need sequential execution of largely independent tasks (e.g., sending separate commands being sent to some Process). But it can probably be adapted for your scenario, in which you want to “share its response with [subsequent requests]”.

    I would suggest that you post a separate question with MCVE with a practical example of precisely what you wanted to pass from one asynchronous function to another. I have, for example, done permutation of the above, passing integer from one task to another. But in practice, that is not of great utility, as it gets more complicated when you start dealing with the reality of heterogenous results parsing. In practice, the simple example with which I started this question is the most practical pattern.

    On the broader question of working with/around actor reentrancy, I would advise keeping an eye out on SE-0306 - Future Directions which explicitly contemplates some potential elegant forthcoming alternatives. I would not be surprised to see some refinements, either in the language itself, or in the Swift Async Algorithms library.

    tl;dr

    I did not want to encumber the above with discussion regarding your code snippets, but there are quite a few issues. So, if you forgive me, here are some observations: