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,12 +8,48 @@
 
 <!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
 
+- 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.  Before, it left the sender open.
+
+- When sending fails with a `SenderError`, for example because the sender's `BroadcastChannel` has no receivers, a `Pipe` now closes both its sender and its receiver, and logs the error: a `NoReceiversError` as a warning, and any other error with its traceback.  Before, the pipe stopped but left both open.
+
+- `Pipe.stop()` still leaves a pipe's sender and receiver open.  So does any exception while receiving, other than the receiver stopping, and any exception from sending that isn't a `SenderError`.
+
+- To keep a `Pipe` running after a `SenderError`, pass `close_pipe_on_send_error=False`.  It then drops the message, keeps sending the messages that follow, and logs a `NoReceiversError` at debug level.
+
+- Give a `Pipe` a clone of its sender if you still need to send after the pipe stops on its own.
+
+- `BroadcastSender.clone()` now raises a `SenderClosedError` if the sender is closed, and a `SenderError` if the channel is closed, like `send()` and `subscribe()` do.  Clone a sender before closing it, or catch `SenderError`.
+
+- `BroadcastSender.subscribe()` now uses the `limit` and `warn_on_overflow` values the `BroadcastChannel` was created with as defaults, instead of `50` and `True`.  Pass `limit=50, warn_on_overflow=True` to keep the old values.
+
+- Call `BroadcastReceiver.close()` from the thread that runs the event loop of the channel's receivers.  From another thread, tasks waiting on any receiver of the channel might not wake up.  If asyncio debug mode is on and a task is waiting, `close()` also raises a `RuntimeError`, and the waiting tasks stay stuck.  Use `loop.call_soon_threadsafe(receiver.close)` instead.
+
+- `Broadcast` is now a deprecated alias of the class behind `BroadcastChannel`.  The `DeprecationWarning` is now emitted when `Broadcast` is imported or accessed from `frequenz.channels`, including by `from frequenz.channels import *`, instead of each time a channel is created, and type checkers no longer report it as deprecated.  Please use `BroadcastChannel` instead.
+
+- This package now depends on `frequenz-core >= 1.6.0, < 2`, and needs `typing-extensions >= 4.13.0`.
+
 ## 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, but with `resend_latest=True`, new receivers still get it if it is the last message sent and it isn't `None`.
+
+- `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.  A sender that was garbage-collected without being closed still counts.
+
 ## Bug Fixes
 
 <!-- Here goes notable bug fixes that are worth a special mention or explanation -->
 
 - Docstrings now use absolute cross-references to public paths instead of relative ones. Relative cross-references are expanded from the module where an object is defined since mkdocstrings-python 2.0.9, which broke some links, and projects rendering docstrings inherited from this library needed `relative_crossrefs: true` for them to work at all. See [frequenz-repo-config-python#655](https://github.com/frequenz-floss/frequenz-repo-config-python/issues/655) for details.
+
+- 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.
+
+- Closing a broadcast receiver now drops the messages left in its buffer, so it is no longer reported as ready.  So `select()` no longer reports it as stopped 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/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,7 @@ 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)
 
 More might be added in the future, and you can also create your own.

docs/user-guide/channels/anycast.md → docs/user-guide/channels/oneshot.md

@@ -1,6 +1,6 @@
-# Anycast
+# Oneshot
 
-::: frequenz.channels._anycast.Anycast
+::: frequenz.channels._oneshot.OneshotChannel
     options:
         inherited_members: []
         members: []

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}")
@@ -16,9 +16,9 @@ message = await receiver2.receive()
 print(f"Received from channel2: {message}")
 ```
 
-The problem is that if the first message is not available in `channel1` but in
-`channel2`, the program will be blocked until a message is available in
-`channel1`, but you probably want to receive the first message from `channel2`
+The problem is that if the first message is not available in `receiver1` but in
+`receiver2`, the program will be blocked until a message is available in
+`receiver1`, but you probably want to receive the first message from `receiver2`
 as soon as it is available.
 
 Frequenz Channels provides two tools to solve this issue:

pyproject.toml

@@ -26,7 +26,8 @@ classifiers = [
 ]
 requires-python = ">= 3.11, < 4"
 dependencies = [
-  "typing-extensions >= 4.12.0, < 5",
+  "frequenz-core >= 1.6.0, < 2",
+  "typing-extensions >= 4.13.0, < 5",
   "watchfiles >= 0.15.0, < 1.4.0",
 ]
 dynamic = ["version"]

src/frequenz/channels/__init__.py

@@ -17,15 +17,29 @@
 * [`Sender`][frequenz.channels.Sender]: An object that can send messages to
   a channel.
 
+* [`CloneableSender`][frequenz.channels.CloneableSender],
+  [`SubscribableSender`][frequenz.channels.SubscribableSender] and
+  [`CloneableSubscribableSender`][frequenz.channels.CloneableSubscribableSender]:
+  Senders that can create more senders, or receivers, attached to the same channel.
+
 Channels:
 
-* [`Anycast`][frequenz.channels.Anycast]: A channel that supports multiple
+* [`BroadcastChannel`][frequenz.channels.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.  A `send()` while it has no receivers raises a
+  [`NoReceiversError`][frequenz.channels.NoReceiversError].
+
+* [`OneshotChannel`][frequenz.channels.OneshotChannel]: A channel that can send a
+  single message from one sender to one receiver.
+
+* [`Anycast`][frequenz.channels.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`][frequenz.channels.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`][frequenz.channels.Broadcast] (deprecated): The predecessor of
+  [`BroadcastChannel`][frequenz.channels.BroadcastChannel].  Unless it is created with
+  `auto_close=True`, it must be closed with `aclose()`.
 
 Utilities to work with channels:
 
@@ -54,6 +68,10 @@
 * [`SenderError`][frequenz.channels.SenderError]: Base class for all errors
   related to senders.
 
+* [`NoReceiversError`][frequenz.channels.NoReceiversError]: Error raised when sending
+  a message to a [`BroadcastChannel`][frequenz.channels.BroadcastChannel] that has no
+  receivers.
+
 * [`ReceiverError`][frequenz.channels.ReceiverError]: Base class for all errors
   related to receivers.
 
@@ -79,8 +97,12 @@
   given amount of time.
 """
 
+from typing import TYPE_CHECKING
+
+from frequenz.core.warnings import DeprecatedAlias, deprecated_aliases
+
 from ._anycast import Anycast
-from ._broadcast import Broadcast, BroadcastChannel, BroadcastReceiver, BroadcastSender
+from ._broadcast import BroadcastChannel, BroadcastReceiver, BroadcastSender
 from ._exceptions import ChannelClosedError, ChannelError, Error
 from ._generic import (
     ChannelMessageT,
@@ -104,12 +126,27 @@
 from ._sender import (
     CloneableSender,
     CloneableSubscribableSender,
+    NoReceiversError,
     Sender,
     SenderClosedError,
     SenderError,
     SubscribableSender,
 )
 
+if TYPE_CHECKING:
+    from ._broadcast import _Broadcast as Broadcast
+else:
+    __getattr__ = deprecated_aliases(
+        __name__,
+        DeprecatedAlias(
+            "Broadcast",
+            new_module="frequenz.channels._broadcast",
+            new_name="_Broadcast",
+            since="v1.12.0",
+            message="Please use BroadcastChannel instead.",
+        ),
+    )
+
 __all__ = [
     "Anycast",
     "Broadcast",
@@ -126,6 +163,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
 
@@ -280,8 +280,8 @@ async def aclose(self) -> None:
 
         Receivers will still be able to drain the pending messages on the channel,
         but after that, subsequent
-        [`receive()`][frequenz.channels.Receiver.receive] calls will return `None`
-        immediately.
+        [`receive()`][frequenz.channels.Receiver.receive] calls will raise a
+        [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError].
         """
         self._closed = True
         async with self._send_cv: