Can't set AsyncStream `onTermination` handler

user_341994 0
Viewed 289

Has anyone succeeded in creating an AsyncStream and setting its onTermination handler? I can't do it. The following is copied and pasted directly from the proposal (https://github.com/apple/swift-evolution/blob/main/proposals/0314-async-stream.md), except I got rid of the warnings by modernizing detach into Task.detached:

    let t = Task.detached {
      func make123Stream() -> AsyncStream<Int> {
        AsyncStream { continuation in
          continuation.onTermination = { termination in
            switch termination {
            case .finished:
                print("Regular finish")
            case .cancelled:
                print("Cancellation")
            }
          }
            Task.detached {
            for n in 1...3 {
              continuation.yield(n)
              sleep(2)
            }
            continuation.finish()
          }
        }
      }

      for await n in make123Stream() {
        print("for-in: \(n)")
      }
      print("After")
    }
    sleep(3)
    t.cancel()

Looks great, but it doesn't compile, and I can't find a way to make it compile. The error message on the onTermination setter reads:

 Converting non-concurrent function value to 
 '@Sendable (AsyncStream<Int>.Continuation.Termination) -> Void' 
 may introduce data races

I don't know what the compiler is asking me to do. Has anyone worked this out, and what's the solution?

(I've filed a bug on this.)

swift-concurrency async-await swift5 swift
1 Answers

Update:

You can work around this bug by adding @Sendable as the first thing inside the closure (before the capture list, parameters, and the in keyword), like:

continuation.onTermination = { @Sendable termination in
    switch termination {
    case .finished:
        print("Regular finish")
    case .cancelled:
        print("Cancellation")
    @unknown default:
        break
    }
}

Original answer:

Yeah I'm guessing it's a bug, because I was able to get it to compile by adding:

as (@Sendable (AsyncStream<Int>.Continuation.Termination) -> Void)

after the closure, like:

continuation.onTermination = { termination in
    switch termination {
    case .finished:
        print("Regular finish")
    case .cancelled:
        print("Cancellation")
    @unknown default:
        break
    }
} as (@Sendable (AsyncStream<Int>.Continuation.Termination) -> Void)

(I also added the @unknown default case to silence a new warning that appeared.)

user_9882600
Related