feat: async broadcast, stream map and async changes

[algebraic-dev]

This PR makes a lot of changes to the Async library:

• Adds a Std.Sync.Broadcast structure that is a multi-producer and multi-consumer broadcast queue. Right now, only the Bounded version is defined because of the ring buffer used to store messages. Broadcast channels can grow very quickly in size since all subscribers need to read each message before it can be dequeued.
• Adds more functions to the Async module.
• Adds a Std.Sync.Notify structure that can send to one or multiple consumers a signal.
• Adds a Std.Sync.CancellationToken structure that can send signals to tokens in a tree, so they can be used to cancel. It's a nice way to create structured concurrency.
• Fix a simple TCP problem in reference counting
• Adds support for multiple bytearrays in send.

[leanprover-community-bot]

Mathlib CI status (docs):

• ❗ Batteries/Mathlib CI will not be attempted unless your PR branches off the nightly-with-mathlib branch. Try git rebase db3fb47109de07fcfbb3e493d387086b0b8c01ae --onto 797985e3197ad2cf805310911854565996e72c48. You can force Mathlib CI using the force-mathlib-ci label. (2025-08-26 22:23:17)
• ❗ Batteries/Mathlib CI will not be attempted unless your PR branches off the nightly-with-mathlib branch. Try git rebase 0b84c3912ece79c0c438eab1dcdef636412a33b6 --onto fc6a6cc4e2ed529abacae437447ef5206527a370. You can force Mathlib CI using the force-mathlib-ci label. (2025-09-11 04:09:52)

[leanprover-bot]

Reference manual CI status:

• ❗ Reference manual CI will not be attempted unless your PR branches off the nightly-with-manual branch. Try git rebase db3fb47109de07fcfbb3e493d387086b0b8c01ae --onto d3dda9f6d4428a906c096067ecb75e432afc4615. You can force reference manual CI using the force-manual-ci label. (2025-08-26 22:23:19)
• ❗ Reference manual CI will not be attempted unless your PR branches off the nightly-with-manual branch. Try git rebase 0b84c3912ece79c0c438eab1dcdef636412a33b6 --onto d3dda9f6d4428a906c096067ecb75e432afc4615. You can force reference manual CI using the force-manual-ci label. (2025-09-11 04:09:53)

[TwoFX]

Haven't looked at Broadcast, CancellationToken and StreamMap yet. All of them are missing tests, please add them.

[TwoFX]

Please add a docstring.

[TwoFX]

Also (perhaps a docstring would clear this up), It's not really clear to me when you would use this over Selectable.one, and also it seems like this implementation doesn't have the fairness properties of Selectable.one because it does not shuffle the selectables. Is this by design?

[algebraic-dev]

I'm going to add the fairness property (do we need the overhead of the property, I think that tokio::select! does not do that)

Regarding using it instead of one, when I was coding the StreamMap, I needed some way to recv from multiple channels at the same time so I used to use Selectable.one for the StreamMap.recv function. However, I also need to expose a function that returns a Selector itself in a function like StreamMap.recvSelector itself, so that while I can receive from multiple maps, I can also use this operation in another Selectable for example with a promise to cancel the entire operation.

[hargoniX]

tokio::select! is fair: https://docs.rs/tokio/latest/tokio/macro.select.html#fairness, this is one of the key requirements when building such a primitive right.

[algebraic-dev]

Oh wow, I misremembered. I thought it wasn’t like that.

[TwoFX]

How does this differ from Std.Condvar?

[algebraic-dev]

The Std.Condvar depends on a Std.Mutex and blocks the entire thread that the Task is using while waiting. If I try to use it with async and a lot of Tasks like this:

def condvar : Async Unit := do
  let condvar ← Std.Condvar.new
  let mutex ← Std.Mutex.new false

  for i in [0:threads] do
    background do
      IO.println s!"start {i + 1}"
      await =<< (show IO (ETask _ _) from IO.asTask (mutex.atomically (condvar.wait mutex)))
      IO.println s!"end {i + 1}"

  IO.sleep 2000
  condvar.notifyAll

It causes some weird behavior because some tasks start running and get notified, while others don’t, because condvar.wait blocks the Task entire task and right now afaik it blocks an entire thread and cannot be paused while doing blocking operations like that.

Notify uses Promises so it’s better suited for concurrency. The Task is not blocked while waiting for a notification which makes it simpler for use cases that just involve notifying:

def notify : Async Unit := do
  let notify ← Std.Notify.new

  for i in [0:threads] do
    background do
      IO.println s!"start {i}"
      notify.wait
      IO.println s!"end {i}"

  IO.sleep 2000
  notify.notify

[algebraic-dev]

Closing this PR since I split the changes into multiple smaller PRs:
#10366, #10367, #10370, #10368, #10369 and #10400