AsyncStream & AsyncThrowingStream behavioral divergences & misleading documentation

[jamieQ]

Description

issues:

1. AsyncStream supports multiple concurrent consumption, but its documentation states that it does not
2. AsyncThrowingStream does not support multiple concurrent consumption (its documentation is accurate)

Reproduction

// non-throwing stream
func test_stream() async {
  let (s, c) = AsyncStream<Int>.makeStream()

  let t1 = Task { @MainActor in
    for await _ in s {}
  }
  await MainActor.run {}
  print("installed consumer 1")

  let t2 = Task { @MainActor in
    for await _ in s {}
  }
  await MainActor.run {}
  print("installed consumer 2")
  // works
}

// throwing stream
func test_throwing_stream() async {
  let (s, c) = AsyncThrowingStream<Int, Error>.makeStream()

  let t1 = Task { @MainActor in
    for try await _ in s {}
  }
  await MainActor.run {}
  print("installed consumer 1")

  let t2 = Task { @MainActor in
    for try await _ in s {}
  }
  await MainActor.run {}
  print("installed consumer 2")
  // fatal error shortly after this point
}

Task { await test_stream() }
Task { await test_throwing_stream() }

Expected behavior

the async stream implementations should function analogously with respect to multi-consumption, and the corresponding documentation should accurately reflect their behaviors. in particular, the throwing variant should behave in the same manner that the non-throwing stream does.

Environment

Apple Swift version 5.10 (swiftlang-5.10.0.13 clang-1500.3.9.4)
Target: arm64-apple-macosx14.0

Additional information

relevant forum posts:

1. https://forums.swift.org/t/consuming-an-asyncstream-from-multiple-tasks-redux/69818
2. https://forums.swift.org/t/question-how-to-re-introduce-concurrent-iteration-runtime-error-for-asyncstream/71831

[malhal]

I believe the hypothesis that "AsyncStream supports multiple concurrent consumption" is incorrect, e.g. in the example below only a single consumer receives:

func test_stream() async {
    let (s, c) = AsyncStream<Int>.makeStream()
    
    let t1 = Task { @MainActor in
        for await x in s {
            
            print("consumer 1 received \(x)")
        }
    }
    await MainActor.run {}
    print("installed consumer 1")
    
    let t2 = Task { @MainActor in
        for await x in s {
            print("consumer 2 received \(x)")
        }
    }
    await MainActor.run {}
    print("installed consumer 2")
    
    c.yield(1) // only consumer 1 receives
}

Output

installed consumer 1
installed consumer 2
consumer 1 received 1

In general when I see a top level Task not at the top level it makes me think that async/await is being applied inversely from how it is designed.

[jamieQ]

I believe the hypothesis that "AsyncStream supports multiple concurrent consumption" is incorrect

@malhal you are correct that only a single 'consumer' will receive a particular value produced by the stream. perhaps i should have phrased this as 'concurrent iteration' rather than 'concurrent consumption' (or if you have a better suggestion, please share!). to clarify what i mean by this – AsyncStream supports behavior where a single backing 'source' of the stream can be iterated concurrently from two different concurrency domains. e.g. you can write this:

let stream = AsyncStream(...)
let t1 = Task.detached {
  for await _ in stream { print("task 1") }
}
let t2 = Task.detached {
  for await _ in stream { print("task 2") }
}
await (t1.value, t2.value)

and if you push a bunch of elements through the stream, they will be 'split' in some manner across the consuming Tasks. however, AsyncThrowingStream does not work this way – if at any point a second continuation is registered via an async iterator, the implementation raises a fatalError(). AsyncStream used to have the same behavior, but was changed at some point (see the linked forum posts for more info).

so the issue here is that the two types have some (IMO) confusing behavioral divergences. this is separate from the question of supporting 'multiple consumption' in the sense i believe you are thinking of it (every pending continuation receives every stream element).

[malhal]

I would suggest if a second consuming client is attempted then that should fatal error, since after all it is misuse.

[jamieQ]

I would suggest if a second consuming client is attempted then that should fatal error, since after all it is misuse.

i also suggested that here, but there was concern about breaking client code.

[malhal]

I would suggest if a second consuming client is attempted then that should fatal error, since after all it is misuse.

i also suggested that here, but there was concern about breaking client code.

I replied that it wouldn't break valid code which is likely the majority.

However, perhaps another way of flagging the misuse would be a Swift 6 style compilation error.

[malhal]

Any idea why they wanted AsyncStream to be Sendable? #41713 Doesn't make sense to me since it should only be used once, in one place, thus not passed around.