Add has_receivers/has_senders and fix BroadcastChannel auto-close

README.md

@@ -67,13 +67,11 @@ python3 -m pip install frequenz-channels
 ```python
 import asyncio
 
-from frequenz.channels import Anycast
+from frequenz.channels import BroadcastChannel
 
 
 async def main() -> None:
-    hello_channel = Anycast[str](name="hello-world-channel")
-    sender = hello_channel.new_sender()
-    receiver = hello_channel.new_receiver()
+    sender, receiver = BroadcastChannel[str](name="hello-world-channel")
 
     await sender.send("Hello World!")
     message = await receiver.receive()
@@ -100,8 +98,7 @@ from enum import Enum, auto
 from typing import assert_never
 
 from frequenz.channels import (
-    Anycast,
-    Broadcast,
+    BroadcastChannel,
     Receiver,
     Sender,
     merge,
@@ -151,6 +148,7 @@ async def send(
                     await control_reply.send(Reply(ReplyCommand.PONG, str(sender)))
                 case _ as unknown:
                     assert_never(unknown)
+    await sender.aclose()
     print(f"{sender}: Finished")
 
 
@@ -159,13 +157,16 @@ async def receive(
     control_command: Receiver[Command],
     control_reply: Sender[Reply],
 ) -> None:
-    """Receive data from multiple channels, until no more data is received for 2 seconds."""
+    """Receive data from multiple channels, until they're closed or idle for 2 seconds."""
     print("receive: Starting")
     timer = Timer(timedelta(seconds=2.0), SkipMissedAndDrift())
     print(f"{timer=}")
     merged = merge(*receivers)
     async for selected in select(merged, timer, control_command):
         if selected_from(selected, merged):
+            if selected.was_stopped:
+                print("receive: All data channels were closed")
+                break
             message = selected.message
             print(f"receive: Received {message=}")
             timer.reset()
@@ -191,51 +192,41 @@ async def receive(
 
 
 async def main() -> None:
-    data_channel_1 = Anycast[str](name="data-channel-1")
-    data_channel_2 = Anycast[str](name="data-channel-2")
-    command_channel = Broadcast[Command](name="control-channel")  # (1)!
-    reply_channel = Anycast[Reply](name="reply-channel")
+    data_sender_1, data_receiver_1 = BroadcastChannel[str](name="data-channel-1")
+    data_sender_2, data_receiver_2 = BroadcastChannel[str](name="data-channel-2")
+    command_sender, command_receiver = BroadcastChannel[Command](
+        name="control-channel"
+    )
+    reply_sender, reply_receiver = BroadcastChannel[Reply](name="reply-channel")
 
     async with asyncio.TaskGroup() as tasks:
         tasks.create_task(
-            send(
-                data_channel_1.new_sender(),
-                command_channel.new_receiver(),
-                reply_channel.new_sender(),
-            ),
+            send(data_sender_1, command_receiver, reply_sender.clone()),
             name="send-channel-1",
         )
         tasks.create_task(
-            send(
-                data_channel_2.new_sender(),
-                command_channel.new_receiver(),
-                reply_channel.new_sender(),
-            ),
+            send(data_sender_2, command_sender.subscribe(), reply_sender.clone()),
             name="send-channel-2",
         )
         tasks.create_task(
             receive(
-                [data_channel_1.new_receiver(), data_channel_2.new_receiver()],
-                command_channel.new_receiver(),
-                reply_channel.new_sender(),
+                [data_receiver_1, data_receiver_2],
+                command_sender.subscribe(),
+                reply_sender,
             ),
             name="receive",
         )
 
-        control_sender = command_channel.new_sender()
-        reply_receiver = reply_channel.new_receiver()
-
         # Send a ping command to all tasks and wait for the replies
-        await control_sender.send(Command.PING)
+        await command_sender.send(Command.PING)
         print(f"main: {await reply_receiver.receive()}")
         print(f"main: {await reply_receiver.receive()}")
         print(f"main: {await reply_receiver.receive()}")
 
         await asyncio.sleep(5.0)
 
-        # Stop senders, after 2 seconds not receiving any data,
-        # the receiver will stop too
-        await control_sender.send(Command.STOP_SENDER)
+        # Stop the senders.  This closes the data channels, so the receiver stops too.
+        await command_sender.send(Command.STOP_SENDER)
 
 
 asyncio.run(main())

RELEASE_NOTES.md

@@ -8,10 +8,38 @@
 
 <!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
 
+- Garbage-collected broadcast senders now count as closed.  A `BroadcastChannel`, or a `Broadcast` created with `auto_close=True`, now closes as soon as its last sender is closed or garbage-collected, so keep a reference to every sender that is still needed.  On all broadcast channels, `sender_count` no longer counts garbage-collected senders.
+
+- A `BroadcastChannel`, or a `Broadcast` created with `auto_close=True`, no longer closes when a message is sent while it has no receivers.  `send()` now drops the message and raises a `NoReceiversError`, a `SenderError` without a cause.  The channel stays open: `subscribe()` keeps working, and sending works again once a new receiver is attached.
+
+- Code that waited for such a channel to close after a send without receivers, for example by checking for a `ChannelClosedError` cause or for `is_closed`, should catch `NoReceiversError` instead.  It can then stop sending, or close all of the channel's senders to close the channel.
+
+- An experimental `Pipe` now closes its sender when its receiver stops.  When sending fails with a `SenderError`, for example because the sender's `BroadcastChannel` has no receivers, it closes its receiver too.  Before, the pipe stopped but left both open.  `Pipe.stop()` still leaves both open.
+
+- Give a `Pipe` a clone of its sender if you still need to send after the pipe stops on its own.
+
+- For errors from a `BroadcastChannel` that are caused by a `ChannelClosedError`, `isinstance(error.__cause__.channel, Broadcast)` is now `False`.
+
+- `BroadcastSender.clone()` now raises a `SenderClosedError` if the sender is closed, and a `SenderError` if the channel is closed, like `send()` and `subscribe()` do.
+
+- `BroadcastSender.subscribe()` now uses the `limit` and `warn_on_overflow` values the `BroadcastChannel` was created with as defaults, instead of `50` and `True`.
+
 ## New Features
 
 <!-- Here goes the main new features and examples or instructions on how to use them -->
 
+- `NoReceiversError` is a new `SenderError`.  An open sender of a `BroadcastChannel`, or of a `Broadcast` created with `auto_close=True`, raises it from `send()` while the channel is open but has no receivers.  The message is dropped.  With `resend_latest=True`, new receivers get the last message that was sent successfully instead.
+
+- `BroadcastSender.has_receivers` and `BroadcastReceiver.has_senders` tell whether the other side of a broadcast channel is still there, so producers can stop producing messages that nobody would receive.
+
 ## Bug Fixes
 
 <!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+
+- Closing a broadcast receiver now wakes up tasks waiting for messages on it, instead of leaving them blocked until the next message arrived or the channel was closed.
+
+- A closed broadcast receiver is no longer reported as ready while messages are left in its buffer.  So `select()` reports it as stopped once instead of over and over, and no longer fails with a `SelectError` for a `filter()` on it.
+
+- Sending to or closing a broadcast channel no longer raises `RuntimeError: Event loop is closed` when a receiver was left waiting on an event loop that has since been closed.
+
+- Creating a `BroadcastChannel` no longer emits a deprecation warning telling users to use `BroadcastChannel` instead.

docs/user-guide/channels/anycast.md

@@ -1,5 +1,11 @@
 # Anycast
 
+!!! Warning "Deprecated"
+
+    Anycast channels are deprecated, because of the high cost of maintaining
+    something for which we don't have a use case.  The implementation will remain
+    in the codebase in its current form at least until the next major version.
+
 ::: frequenz.channels._anycast.Anycast
     options:
         inherited_members: []

docs/user-guide/channels/broadcast.md

@@ -1,6 +1,6 @@
 # Broadcast
 
-::: frequenz.channels._broadcast.Broadcast
+::: frequenz.channels._broadcast.BroadcastChannel
     options:
         inherited_members: []
         members: []

docs/user-guide/channels/index.md

@@ -35,7 +35,8 @@ various factors:
 Because these questions can have many answers, there are different types of
 channels. Frequenz Channels offers a few of them:
 
-* [Anycast](anycast.md)
 * [Broadcast](broadcast.md)
+* [Oneshot](oneshot.md)
+* [Anycast](anycast.md) (deprecated)
 
 More might be added in the future, and you can also create your own.

docs/user-guide/channels/oneshot.md

@@ -0,0 +1,10 @@
+# Oneshot
+
+::: frequenz.channels._oneshot.OneshotChannel
+    options:
+        inherited_members: []
+        members: []
+        show_bases: false
+        show_root_heading: false
+        show_root_toc_entry: false
+        show_source: false

docs/user-guide/receiving/synchronization/index.md

@@ -6,8 +6,8 @@ as soon as it is available in one single task. A naive approach like this will
 not work:
 
 ```python show_lines="4:"
-receiver1: Receiver[int] = channel1.new_receiver()
-receiver2: Receiver[int] = channel2.new_receiver()
+sender1, receiver1 = BroadcastChannel[int](name="channel1")
+sender2, receiver2 = BroadcastChannel[int](name="channel2")
 
 message = await receiver1.receive()
 print(f"Received from channel1: {message}")

src/frequenz/channels/__init__.py

@@ -17,15 +17,29 @@
 * [`Sender`][.Sender]: An object that can send messages to
   a channel.
 
+* [`CloneableSender`][.CloneableSender],
+  [`SubscribableSender`][.SubscribableSender] and
+  [`CloneableSubscribableSender`][.CloneableSubscribableSender]: Senders that can
+  create more senders, or receivers, attached to the same channel.
+
 Channels:
 
-* [`Anycast`][.Anycast]: A channel that supports multiple
+* [`BroadcastChannel`][.BroadcastChannel]: A channel to broadcast messages
+  from multiple senders to multiple receivers. Each message sent through any of
+  the senders is received by all of the receivers.  The channel closes itself
+  when all its senders are closed or garbage-collected.  A `send()` while it has
+  no receivers raises a [`NoReceiversError`][.NoReceiversError].
+
+* [`OneshotChannel`][.OneshotChannel]: A channel that can send a single
+  message from one sender to one receiver.
+
+* [`Anycast`][.Anycast] (deprecated): A channel that supports multiple
   senders and multiple receivers.  A message sent through a sender will be
   received by exactly one receiver.
 
-* [`Broadcast`][.Broadcast]: A channel to broadcast messages
-  from multiple senders to multiple receivers. Each message sent through any of
-  the senders is received by all of the receivers.
+* [`Broadcast`][.Broadcast] (deprecated): The predecessor of
+  [`BroadcastChannel`][.BroadcastChannel].  Unless it is created with
+  `auto_close=True`, it must be closed with `aclose()`.
 
 Utilities to work with channels:
 
@@ -54,6 +68,9 @@
 * [`SenderError`][.SenderError]: Base class for all errors
   related to senders.
 
+* [`NoReceiversError`][.NoReceiversError]: Error raised when sending
+  a message to a [`BroadcastChannel`][.BroadcastChannel] that has no receivers.
+
 * [`ReceiverError`][.ReceiverError]: Base class for all errors
   related to receivers.
 
@@ -104,6 +121,7 @@
 from ._sender import (
     CloneableSender,
     CloneableSubscribableSender,
+    NoReceiversError,
     Sender,
     SenderClosedError,
     SenderError,
@@ -126,6 +144,7 @@
     "LatestValueCache",
     "MappedMessageT_co",
     "Merger",
+    "NoReceiversError",
     "OneshotChannel",
     "OneshotReceiver",
     "OneshotSender",

src/frequenz/channels/_anycast.py

@@ -93,7 +93,7 @@ class Anycast(Generic[ChannelMessageT]):
     This channel is useful, for example, to distribute work across multiple workers.
 
     In cases where each message needs to be received by every receiver, a
-    [broadcast][frequenz.channels.Broadcast] channel may be used.
+    [`BroadcastChannel`][frequenz.channels.BroadcastChannel] may be used.
 
     # Examples