diff --git a/README.md b/README.md index a08ffd64..10ee1cc7 100644 --- a/README.md +++ b/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()) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 96a0240b..f332a36d 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -8,10 +8,38 @@ +- 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 +- `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 + +- 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. diff --git a/docs/user-guide/channels/anycast.md b/docs/user-guide/channels/anycast.md index eca01889..07448cc8 100644 --- a/docs/user-guide/channels/anycast.md +++ b/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: [] diff --git a/docs/user-guide/channels/broadcast.md b/docs/user-guide/channels/broadcast.md index f759bce3..ba8ebe2e 100644 --- a/docs/user-guide/channels/broadcast.md +++ b/docs/user-guide/channels/broadcast.md @@ -1,6 +1,6 @@ # Broadcast -::: frequenz.channels._broadcast.Broadcast +::: frequenz.channels._broadcast.BroadcastChannel options: inherited_members: [] members: [] diff --git a/docs/user-guide/channels/index.md b/docs/user-guide/channels/index.md index c6f2df64..0142ba8d 100644 --- a/docs/user-guide/channels/index.md +++ b/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. diff --git a/docs/user-guide/channels/oneshot.md b/docs/user-guide/channels/oneshot.md new file mode 100644 index 00000000..9405ca72 --- /dev/null +++ b/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 diff --git a/docs/user-guide/receiving/synchronization/index.md b/docs/user-guide/receiving/synchronization/index.md index 60d70d86..2c589daf 100644 --- a/docs/user-guide/receiving/synchronization/index.md +++ b/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}") diff --git a/src/frequenz/channels/__init__.py b/src/frequenz/channels/__init__.py index 627a890f..109be6ef 100644 --- a/src/frequenz/channels/__init__.py +++ b/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", diff --git a/src/frequenz/channels/_anycast.py b/src/frequenz/channels/_anycast.py index 7ccb809c..765b6ac1 100644 --- a/src/frequenz/channels/_anycast.py +++ b/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 diff --git a/src/frequenz/channels/_broadcast.py b/src/frequenz/channels/_broadcast.py index 0f010ea0..9f589c79 100644 --- a/src/frequenz/channels/_broadcast.py +++ b/src/frequenz/channels/_broadcast.py @@ -3,11 +3,15 @@ """A channel to broadcast messages to all receivers.""" +# The docstrings carry long examples. +# pylint: disable=too-many-lines + from __future__ import annotations +import contextlib import logging import weakref -from asyncio import Condition +from asyncio import AbstractEventLoop, Event, get_running_loop from collections import deque from typing import Generic, TypeVar @@ -16,17 +20,246 @@ from ._exceptions import ChannelClosedError from ._generic import ChannelMessageT from ._receiver import Receiver, ReceiverStoppedError -from ._sender import CloneableSubscribableSender, SenderClosedError, SenderError +from ._sender import ( + CloneableSubscribableSender, + NoReceiversError, + SenderClosedError, + SenderError, +) _logger = logging.getLogger(__name__) -@deprecated("Please use BroadcastChannel instead.") -class Broadcast( # pylint: disable=too-many-instance-attributes +class _Broadcast( # pylint: disable=too-many-instance-attributes Generic[ChannelMessageT] ): + """The implementation of broadcast channels. + + It is shared by the deprecated [`Broadcast`][frequenz.channels.Broadcast] class + and by [`BroadcastChannel`][frequenz.channels.BroadcastChannel], and is not + deprecated itself. + """ + + def __init__( + self, + *, + name: str, + resend_latest: bool = False, + auto_close: bool = False, + ) -> None: + """Initialize this channel. + + Args: + name: The name of the channel. This is for logging purposes, and it will be + shown in the string representation of the channel. + resend_latest: When True, every time a new receiver is created with + [`.new_receiver()`][..new_receiver], the last message that was sent + successfully will be sent to the new receiver automatically, unless + it was `None` or the channel was closed. This allows new receivers on + slow streams to get the latest message as soon as they are created, + without having to wait for the next message on the channel to arrive. + It is safe to be set in data/reporting channels, but is not + recommended for use in channels that stream control instructions. + auto_close: If True, the channel is closed when its last sender is + closed or garbage-collected, or when a receiver waits for messages + while the channel has no senders. A `send()` while it has no + receivers raises a + [`NoReceiversError`][frequenz.channels.NoReceiversError]. + """ + self._name: str = name + """The name of the broadcast channel. + + Only used for debugging purposes. + """ + + self._wakeup: Event = Event() + """The event receivers wait on for new messages or state changes. + + Setting it wakes up all receivers waiting on it, and it is then replaced + with a fresh event for the next wait. It can be set from synchronous code + running on the receivers' event loop, like + [`BroadcastReceiver.close()`][frequenz.channels.BroadcastReceiver.close]. + """ + + self._loop: AbstractEventLoop | None = None + """The event loop the receivers last waited on.""" + + self._sender_count: int = 0 + """The number of senders attached to this channel.""" + + self._receivers: dict[ + int, weakref.ReferenceType[BroadcastReceiver[ChannelMessageT]] + ] = {} + """The receivers attached to the channel, indexed by their hash().""" + + self._closed: bool = False + """Whether the channel is closed.""" + + self._latest: ChannelMessageT | None = None + """The latest message sent to the channel.""" + + self._auto_close_enabled: bool = auto_close + """Whether the channel closes automatically, see the `auto_close` argument.""" + + self._subscriber_limit: int = 50 + """The default buffer size of receivers created with `BroadcastSender.subscribe()`.""" + + self._subscriber_warn_on_overflow: bool = True + """The default overflow warning setting of subscribed receivers.""" + + self.resend_latest: bool = resend_latest + """Whether to resend the latest message to new receivers. + + See the `resend_latest` argument of the constructor. + """ + + @property + def name(self) -> str: + """The name of this channel. + + This is for logging purposes, and it will be shown in the string representation + of this channel. + """ + return self._name + + @property + def is_closed(self) -> bool: + """Whether this channel is closed. + + Any further attempts to use this channel after it is closed will result in an + exception. + """ + return self._closed + + async def aclose(self) -> None: + """Close this channel. + + Any further attempts to [`send()`][...Sender.send] data + will raise a [`SenderError`][...SenderError]. + + Receivers will still be able to drain the pending items on their queues, + but after that, subsequent + [`receive()`][...Receiver.receive] calls will raise a + [`ReceiverStoppedError`][...ReceiverStoppedError]. + """ + self._close() + + def _close(self) -> None: + """Close this channel synchronously, see [`.aclose()`][..aclose].""" + self._mark_closed() + self._wake_receivers() + + def _mark_closed(self) -> None: + """Mark this channel as closed, without waking up its receivers.""" + self._latest = None + self._closed = True + + def _wake_receivers(self) -> None: + """Wake up all receivers waiting for new messages.""" + wakeup, self._wakeup = self._wakeup, Event() + try: + wakeup.set() + except RuntimeError: + # `set()` raises if the waiters' loop is closed, and they can't be woken + # anyway. All waiters of an event share one loop, so any other + # `RuntimeError`, like a call from another thread in debug mode, is + # re-raised. + loop = getattr(wakeup, "_loop", None) + if loop is None or not loop.is_closed(): + raise + + def _remove_sender(self, *, from_gc: bool = True) -> None: + """Detach a closed or garbage-collected sender from this channel. + + The garbage collector can run this in any thread, and in the middle of + other code. So when it does, this only updates the channel's state, and + the receivers are woken up later, on the event loop they wait on. + + Args: + from_gc: Whether the garbage collector runs this, through the sender's + finalizer. + """ + self._sender_count -= 1 + if self._auto_close_enabled and self._sender_count == 0: + if from_gc: + self._mark_closed() + self._wake_receivers_soon() + else: + self._close() + + def _wake_receivers_soon(self) -> None: + """Schedule waking up the receivers on the event loop they wait on.""" + if self._loop is not None: + # A closed loop raises `RuntimeError`, and has no receivers left to wake. + with contextlib.suppress(RuntimeError): + self._loop.call_soon_threadsafe(self._wake_receivers) + + def _remove_receiver(self, key: int) -> None: + """Detach a closed or garbage-collected receiver from this channel. + + Args: + key: The key of the receiver in `self._receivers`. + """ + self._receivers.pop(key, None) + + @deprecated("The close() method is deprecated, use aclose() instead") + async def close(self) -> None: # noqa: D402 + """Close the channel, deprecated alias for [`.aclose()`][..aclose].""" # noqa: D402 + return await self.aclose() + + def new_sender(self) -> BroadcastSender[ChannelMessageT]: + """Return a new sender attached to this channel.""" + return BroadcastSender(self) + + def new_receiver( + self, *, name: str | None = None, limit: int = 50, warn_on_overflow: bool = True + ) -> BroadcastReceiver[ChannelMessageT]: + """Return a new receiver attached to this channel. + + Broadcast receivers have their own buffer, and when messages are not + being consumed fast enough and the buffer fills up, old messages will + get dropped just in this receiver. + + Args: + name: A name to identify the receiver in the logs. + limit: Number of messages the receiver can hold in its buffer. + warn_on_overflow: Whether to log a warning when the receiver's + buffer is full and a message is dropped. + + Returns: + A new receiver attached to this channel. + """ + recv: BroadcastReceiver[ChannelMessageT] = BroadcastReceiver( + self, name=name, limit=limit, warn_on_overflow=warn_on_overflow + ) + self._receivers[hash(recv)] = weakref.ref(recv) + if self.resend_latest and self._latest is not None: + recv._enqueue(self._latest) # pylint: disable=protected-access + return recv + + def __str__(self) -> str: + """Return a string representation of this channel.""" + return f"{type(self).__name__.removeprefix('_')}:{self._name}" + + def __repr__(self) -> str: + """Return a string representation of this channel.""" + return ( + f"{type(self).__name__.removeprefix('_')}(name={self._name!r}, " + f"resend_latest={self.resend_latest!r}):<" + f"latest={self._latest!r}, " + f"receivers={len(self._receivers)!r}, " + f"closed={self._closed!r}>" + ) + + +@deprecated("Please use BroadcastChannel instead.") +class Broadcast(_Broadcast[ChannelMessageT]): """A channel that delivers all messages to all receivers. + !!! Warning "Deprecated" + + Please use [`BroadcastChannel`][frequenz.channels.BroadcastChannel] instead. + # Description [`Broadcast`][frequenz.channels.Broadcast] channels can have multiple @@ -127,14 +360,14 @@ async def main() -> None: received 2 ``` - Example: Send a few number from multiple senders to multiple receivers + Example: Send a few numbers from multiple senders to multiple receivers This is a more complex example that sends a few numbers from multiple senders to - multiple receivers, using a small buffer to force the senders to block. + multiple receivers. The receivers stop when the channel is closed. ```python import asyncio - from frequenz.channels import Broadcast, Receiver, ReceiverStoppedError, Sender + from frequenz.channels import Broadcast, Receiver, Sender async def send(name: str, sender: Sender[int], start: int, stop: int) -> None: @@ -144,28 +377,28 @@ async def send(name: str, sender: Sender[int], start: int, stop: int) -> None: async def recv(name: str, receiver: Receiver[int]) -> None: - try: - async for message in receiver: - print(f"{name} received {message}") + async for message in receiver: + print(f"{name} received {message}") await asyncio.sleep(0.1) # sleep (or work) with the data - except ReceiverStoppedError: - pass async def main() -> None: acast = Broadcast[int](name="numbers") + receiver_1 = acast.new_receiver() + receiver_2 = acast.new_receiver() async with asyncio.TaskGroup() as task_group: - task_group.create_task(send("sender_1", acast.new_sender(), 10, 13)) - task_group.create_task(send("sender_2", acast.new_sender(), 20, 22)) - task_group.create_task(recv("receiver_1", acast.new_receiver())) - task_group.create_task(recv("receiver_2", acast.new_receiver())) + task_group.create_task(recv("receiver_1", receiver_1)) + task_group.create_task(recv("receiver_2", receiver_2)) + await send("sender_1", acast.new_sender(), 10, 13) + await send("sender_2", acast.new_sender(), 20, 22) + await acast.aclose() asyncio.run(main()) ``` - The output should look something like this(although the sending and received + The output should look something like this (although the sending and received might appear interleaved in a different way): ``` @@ -175,163 +408,18 @@ async def main() -> None: sender_2 sending 20 sender_2 sending 21 receiver_1 received 10 - receiver_1 received 11 - receiver_1 received 12 - receiver_1 received 20 - receiver_1 received 21 receiver_2 received 10 + receiver_1 received 11 receiver_2 received 11 + receiver_1 received 12 receiver_2 received 12 + receiver_1 received 20 receiver_2 received 20 + receiver_1 received 21 receiver_2 received 21 ``` """ - def __init__( - self, - *, - name: str, - resend_latest: bool = False, - auto_close: bool = False, - ) -> None: - """Initialize this channel. - - Args: - name: The name of the channel. This is for logging purposes, and it will be - shown in the string representation of the channel. - resend_latest: When True, every time a new receiver is created with - [`.new_receiver()`][..new_receiver], the last message seen by the channel - will be sent to the - new receiver automatically. This allows new receivers on slow streams to - get the latest message as soon as they are created, without having to - wait for the next message on the channel to arrive. It is safe to be - set in data/reporting channels, but is not recommended for use in - channels that stream control instructions. - auto_close: If True, the channel will be closed when all senders or all - receivers are closed. - """ - self._name: str = name - """The name of the broadcast channel. - - Only used for debugging purposes. - """ - - self._recv_cv: Condition = Condition() - """The condition to wait for data in the channel's buffer.""" - - self._sender_count: int = 0 - """The number of senders attached to this channel.""" - - self._receivers: dict[ - int, weakref.ReferenceType[BroadcastReceiver[ChannelMessageT]] - ] = {} - """The receivers attached to the channel, indexed by their hash().""" - - self._closed: bool = False - """Whether the channel is closed.""" - - self._latest: ChannelMessageT | None = None - """The latest message sent to the channel.""" - - self._auto_close_enabled: bool = auto_close - """Whether to close the channel when all senders or all receivers are closed.""" - - self.resend_latest: bool = resend_latest - """Whether to resend the latest message to new receivers. - - When `True`, every time a new receiver is created with [`.new_receiver()`][..new_receiver], - it will - automatically get sent the latest message on the channel. This allows new - receivers on slow streams to get the latest message as soon as they are created, - without having to wait for the next message on the channel to arrive. - - It is safe to be set in data/reporting channels, but is not recommended for use - in channels that stream control instructions. - """ - - @property - def name(self) -> str: - """The name of this channel. - - This is for logging purposes, and it will be shown in the string representation - of this channel. - """ - return self._name - - @property - def is_closed(self) -> bool: - """Whether this channel is closed. - - Any further attempts to use this channel after it is closed will result in an - exception. - """ - return self._closed - - async def aclose(self) -> None: - """Close this channel. - - Any further attempts to [`send()`][...Sender.send] data - will raise a [`SenderError`][...SenderError]. - - Receivers will still be able to drain the pending items on their queues, - but after that, subsequent - [`receive()`][...Receiver.receive] calls will return `None` - immediately. - """ - self._latest = None - self._closed = True - async with self._recv_cv: - self._recv_cv.notify_all() - - @deprecated("The close() method is deprecated, use aclose() instead") - async def close(self) -> None: # noqa: D402 - """Close the channel, deprecated alias for [`.aclose()`][..aclose].""" # noqa: D402 - return await self.aclose() - - def new_sender(self) -> BroadcastSender[ChannelMessageT]: - """Return a new sender attached to this channel.""" - return BroadcastSender(self) - - def new_receiver( - self, *, name: str | None = None, limit: int = 50, warn_on_overflow: bool = True - ) -> BroadcastReceiver[ChannelMessageT]: - """Return a new receiver attached to this channel. - - Broadcast receivers have their own buffer, and when messages are not - being consumed fast enough and the buffer fills up, old messages will - get dropped just in this receiver. - - Args: - name: A name to identify the receiver in the logs. - limit: Number of messages the receiver can hold in its buffer. - warn_on_overflow: Whether to log a warning when the receiver's - buffer is full and a message is dropped. - - Returns: - A new receiver attached to this channel. - """ - recv: BroadcastReceiver[ChannelMessageT] = BroadcastReceiver( - self, name=name, limit=limit, warn_on_overflow=warn_on_overflow - ) - self._receivers[hash(recv)] = weakref.ref(recv) - if self.resend_latest and self._latest is not None: - recv._enqueue(self._latest) # pylint: disable=protected-access - return recv - - def __str__(self) -> str: - """Return a string representation of this channel.""" - return f"{type(self).__name__}:{self._name}" - - def __repr__(self) -> str: - """Return a string representation of this channel.""" - return ( - f"{type(self).__name__}(name={self._name!r}, " - f"resend_latest={self.resend_latest!r}):<" - f"latest={self._latest!r}, " - f"receivers={len(self._receivers)!r}, " - f"closed={self._closed!r}>" - ) - _T = TypeVar("_T") @@ -339,30 +427,51 @@ def __repr__(self) -> str: class BroadcastSender(CloneableSubscribableSender[_T]): """A sender to send messages to the broadcast channel. - Should not be created directly, but through the - [`.clone()`][.clone] - method of an existing sender. + Should not be created directly, but through + [`BroadcastChannel`][frequenz.channels.BroadcastChannel] or the + [`.clone()`][.clone] method of an existing sender. """ - def __init__(self, channel: Broadcast[_T], /) -> None: + def __init__(self, channel: _Broadcast[_T], /) -> None: """Initialize this sender. Args: channel: A reference to the broadcast channel this sender belongs to. """ - self._channel: Broadcast[_T] = channel + self._channel: _Broadcast[_T] = channel """The broadcast channel this sender belongs to.""" self._closed: bool = False """Whether this sender is closed.""" - self._channel._sender_count += 1 + # pylint: disable=protected-access + self._detach = weakref.finalize(self, channel._remove_sender) + """Detaches this sender from the channel when it is garbage-collected.""" + + channel._sender_count += 1 + # pylint: enable=protected-access @property def sender_count(self) -> int: """The number of open senders attached to this sender's channel.""" return self._channel._sender_count # pylint: disable=protected-access + @property + def has_receivers(self) -> bool: + """Whether this sender's channel is open and has any receivers. + + Producers can check this to stop producing messages that nobody would + receive. When this is `False` on an open + [`BroadcastChannel`][frequenz.channels.BroadcastChannel], + [`send()`][..send] on an open sender raises a + [`NoReceiversError`][frequenz.channels.NoReceiversError] until a new + receiver is created with [`subscribe()`][..subscribe]. + + This doesn't check whether this sender itself is closed. + """ + # pylint: disable=protected-access + return not self._channel._closed and bool(self._channel._receivers) + @override async def send(self, message: _T, /) -> None: """Send a message to all broadcast receivers. @@ -375,6 +484,10 @@ async def send(self, message: _T, /) -> None: A [`ChannelClosedError`][...ChannelClosedError] is set as the cause. SenderClosedError: If this sender was closed. + NoReceiversError: If this sender and its channel are open, but the + channel has no receivers, and it is a + [`BroadcastChannel`][frequenz.channels.BroadcastChannel] or a + `Broadcast` created with `auto_close=True`. """ if self._closed: raise SenderClosedError(self) @@ -383,23 +496,15 @@ async def send(self, message: _T, /) -> None: raise SenderError("The channel was closed", self) from ChannelClosedError( self._channel ) - stale_refs = [] - for _hash, recv_ref in self._channel._receivers.items(): - recv = recv_ref() - if recv is None: - stale_refs.append(_hash) - continue - recv._enqueue(message) - for _hash in stale_refs: - del self._channel._receivers[_hash] - if self._channel._auto_close_enabled and len(self._channel._receivers) == 0: - await self._channel.aclose() - raise SenderError("The channel was closed", self) from ChannelClosedError( - self._channel - ) + if self._channel._auto_close_enabled and not self._channel._receivers: + raise NoReceiversError(self) + # Iterate over a copy, as receivers garbage-collected during the loop remove + # themselves from the dict. + for recv_ref in list(self._channel._receivers.values()): + if (recv := recv_ref()) is not None: + recv._enqueue(message) self._channel._latest = message - async with self._channel._recv_cv: - self._channel._recv_cv.notify_all() + self._channel._wake_receivers() # pylint: enable=protected-access @override @@ -413,33 +518,47 @@ async def aclose(self) -> None: if self._closed: return self._closed = True - self._channel._sender_count -= 1 - - if ( - self._channel._sender_count == 0 # pylint: disable=protected-access - and self._channel._auto_close_enabled # pylint: disable=protected-access - ): - await self._channel.aclose() + if self._detach.detach() is not None: + # pylint: disable-next=protected-access + self._channel._remove_sender(from_gc=False) @override def clone(self) -> BroadcastSender[_T]: - """Return a clone of this sender.""" + """Return a clone of this sender. + + Returns: + A new sender attached to this sender's channel. + + Raises: + SenderError: If the underlying channel is closed. + SenderClosedError: If this sender is closed. + """ + if self._closed: + raise SenderClosedError(self) + if self._channel._closed: # pylint: disable=protected-access + raise SenderError("The channel was closed", self) from ChannelClosedError( + self._channel + ) return BroadcastSender(self._channel) @override def subscribe( self, name: str | None = None, - limit: int = 50, - warn_on_overflow: bool = True, + limit: int | None = None, + warn_on_overflow: bool | None = None, ) -> BroadcastReceiver[_T]: """Return a new receiver attached to this sender's channel. Args: name: A name to identify the receiver in the logs. - limit: Number of messages the receiver can hold in its buffer. + limit: Number of messages the receiver can hold in its buffer. If + `None`, the `limit` the `BroadcastChannel` was created with is used, + or `50` for a `Broadcast`. warn_on_overflow: Whether to log a warning when the receiver's buffer is - full and a message is dropped. + full and a message is dropped. If `None`, the `warn_on_overflow` + setting the `BroadcastChannel` was created with is used, or `True` + for a `Broadcast`. Returns: A new receiver attached to this sender's channel. @@ -450,13 +569,21 @@ def subscribe( """ if self._closed: raise SenderClosedError(self) - if self._channel._closed: # pylint: disable=protected-access + # pylint: disable=protected-access + if self._channel._closed: raise SenderError("The channel was closed", self) from ChannelClosedError( self._channel ) return self._channel.new_receiver( - name=name, limit=limit, warn_on_overflow=warn_on_overflow + name=name, + limit=self._channel._subscriber_limit if limit is None else limit, + warn_on_overflow=( + self._channel._subscriber_warn_on_overflow + if warn_on_overflow is None + else warn_on_overflow + ), ) + # pylint: enable=protected-access def __str__(self) -> str: """Return a string representation of this sender.""" @@ -470,14 +597,15 @@ def __repr__(self) -> str: class BroadcastReceiver(Receiver[_T]): """A receiver to receive messages from the broadcast channel. - Should not be created directly, but through the + Should not be created directly, but through + [`BroadcastChannel`][frequenz.channels.BroadcastChannel] or the [`BroadcastSender.subscribe()`][frequenz.channels.BroadcastSender.subscribe] method of an existing sender. """ def __init__( self, - channel: Broadcast[_T], + channel: _Broadcast[_T], /, *, name: str | None, @@ -491,8 +619,7 @@ def __init__( get dropped just in this receiver. Args: - channel: A reference to the [`Broadcast`][...Broadcast] - channel that this receiver belongs to. + channel: A reference to the broadcast channel this receiver belongs to. name: A name to identify the receiver in the logs. If `None` an `id(self)`-based name will be used. This is only for debugging purposes, it will be shown in the string representation of the @@ -509,7 +636,7 @@ def __init__( Only used for debugging purposes. """ - self._channel: Broadcast[_T] = channel + self._channel: _Broadcast[_T] = channel """The broadcast channel that this receiver belongs to.""" self._q: deque[_T] = deque(maxlen=limit) @@ -518,6 +645,10 @@ def __init__( self._closed: bool = False """Whether the receiver is closed.""" + # pylint: disable-next=protected-access + self._detach = weakref.finalize(self, channel._remove_receiver, hash(self)) + """Detaches this receiver from the channel when closed or garbage-collected.""" + def _enqueue(self, message: _T, /) -> None: """Put a message into this receiver's queue. @@ -537,6 +668,20 @@ def _enqueue(self, message: _T, /) -> None: ) self._q.append(message) + @property + def has_senders(self) -> bool: + """Whether this receiver's channel is open and has any senders. + + A [`BroadcastChannel`][frequenz.channels.BroadcastChannel] is closed as soon + as its last sender is closed or garbage-collected. So once this is `False` + on a `BroadcastChannel`, it stays `False`, and no new messages will arrive. + The messages already in this receiver's buffer can still be received. + + This doesn't check whether this receiver itself is closed. + """ + # pylint: disable=protected-access + return not self._channel._closed and self._channel._sender_count > 0 + def __len__(self) -> int: """Return the number of unconsumed messages in the broadcast receiver. @@ -558,24 +703,25 @@ async def ready(self) -> bool: Returns: Whether the receiver is still active. """ + if self._closed: + return False # if there are still messages to consume from the queue, return immediately if self._q: return True - # Use a while loop here, to handle spurious wakeups of condition variables. - # - # The condition also makes sure that if there are already messages ready to be - # consumed, then we return immediately. + self._channel._loop = get_running_loop() # pylint: disable=protected-access + + # Use a while loop here, because receivers are woken up on every channel + # state change, not only when a message for this receiver arrives. # pylint: disable=protected-access while len(self._q) == 0: if self._channel._closed or self._closed: return False if self._channel._auto_close_enabled and self._channel._sender_count == 0: - await self._channel.aclose() + self._channel._close() return False - async with self._channel._recv_cv: - await self._channel._recv_cv.wait() - return True + await self._channel._wakeup.wait() + return not self._closed # pylint: enable=protected-access @override @@ -601,14 +747,13 @@ def consume(self) -> _T: def close(self) -> None: """Close the receiver. - After calling this method, new messages will not be received. Once the - receiver's buffer is drained, trying to receive a message will raise a - [`ReceiverStoppedError`][...ReceiverStoppedError]. + After calling this method, trying to receive a message raises a + [`ReceiverStoppedError`][...ReceiverStoppedError], even if there are still + messages in the receiver's buffer. """ self._closed = True - self._channel._receivers.pop( # pylint: disable=protected-access - hash(self), None - ) + self._detach() + self._channel._wake_receivers() # pylint: disable=protected-access def __str__(self) -> str: """Return a string representation of this receiver.""" @@ -663,24 +808,46 @@ class BroadcastChannel( receivers that can't keep up with the senders, and not for the whole channel. - Instantiating this class will create a new broadcast channel, and return an - initial sender and a receiver. Further senders and receivers can be created - with the [`BroadcastSender.clone()`][frequenz.channels.BroadcastSender.clone], + Instantiating this class creates a new broadcast channel, and returns an + initial sender and receiver. Further senders and receivers can be created + with the [`BroadcastSender.clone()`][frequenz.channels.BroadcastSender.clone] and [`BroadcastSender.subscribe()`][frequenz.channels.BroadcastSender.subscribe] methods respectively. When a sender or a receiver is not needed anymore, it should be closed with [`BroadcastSender.aclose()`][frequenz.channels.BroadcastSender.aclose] or - [`BroadcastReceiver.close()`][frequenz.channels.BroadcastReceiver.close]. This will prevent - further attempts to [`send()`][frequenz.channels.BroadcastSender.send] data, - and will allow receivers to drain the pending items on their queues, but - after that, subsequent [`receive()`][frequenz.channels.Receiver.receive] calls - will raise a - [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError]. - - When all senders of a channel are closed, all its receivers will be - automatically closed, and vice versa. + [`BroadcastReceiver.close()`][frequenz.channels.BroadcastReceiver.close]. + + The channel closes itself when all its senders are gone. Senders and receivers + are gone when they are closed or garbage-collected. This is what happens when + either side is gone: + + * As soon as all senders are gone, the channel is closed. Receivers can still + drain the messages pending in their buffers, but after that, + [`receive()`][frequenz.channels.Receiver.receive] raises a + [`ReceiverStoppedError`][frequenz.channels.ReceiverStoppedError], and `async + for` loops over them end. + + * While all receivers are gone, + [`send()`][frequenz.channels.BroadcastSender.send] drops the message and + raises a [`NoReceiversError`][frequenz.channels.NoReceiversError]. The + channel stays open, so once + [`subscribe()`][frequenz.channels.BroadcastSender.subscribe] creates a new + receiver, sending works again. + + Senders can check + [`has_receivers`][frequenz.channels.BroadcastSender.has_receivers] to stop + producing messages that nobody would receive, and receivers can check + [`has_senders`][frequenz.channels.BroadcastReceiver.has_senders] to find out + whether new messages can still arrive. + + Tip: + Because garbage-collected senders count as closed, keep a reference to every + sender that is still needed. For example, after + `_, receiver = BroadcastChannel[int](name="numbers")`, the channel closes as + soon as `_` is reassigned or goes out of scope. The receiver then stops once + it has received the messages already in its buffer. This channel is useful, for example, to implement a pub/sub pattern, where multiple consumers can subscribe to a channel to receive all messages. @@ -689,12 +856,12 @@ class BroadcastChannel( Example: Send a few numbers to a receiver This is a very simple example that sends a few numbers from a single sender to - a single receiver. + a single receiver. The receiver stops when the sender is closed. ```python import asyncio - from frequenz.channels import BroadcastChannel, Sender + from frequenz.channels import BroadcastChannel, Receiver, Sender async def send(sender: Sender[int]) -> None: @@ -704,15 +871,18 @@ async def send(sender: Sender[int]) -> None: await sender.aclose() + async def receive(receiver: Receiver[int]) -> None: + async for message in receiver: + print(f"received {message}") + print("all senders are closed") + + async def main() -> None: sender, receiver = BroadcastChannel[int](name="numbers") async with asyncio.TaskGroup() as task_group: task_group.create_task(send(sender)) - for _ in range(3): - message = await receiver.receive() - print(f"received {message}") - await asyncio.sleep(0.1) # sleep (or work) with the data + task_group.create_task(receive(receiver)) asyncio.run(main()) @@ -728,16 +898,17 @@ async def main() -> None: received 0 received 1 received 2 + all senders are closed ``` - Example: Send a few number from multiple senders to multiple receivers + Example: Send a few numbers from multiple senders to multiple receivers This is a more complex example that sends a few numbers from multiple senders to - multiple receivers, using a small buffer to force the senders to block. + multiple receivers. The receivers stop when all senders are closed. ```python import asyncio - from frequenz.channels import BroadcastChannel, Receiver, ReceiverStoppedError, Sender + from frequenz.channels import BroadcastChannel, Receiver, Sender async def send(name: str, sender: Sender[int], start: int, stop: int) -> None: @@ -747,30 +918,28 @@ async def send(name: str, sender: Sender[int], start: int, stop: int) -> None: await sender.aclose() - async def recv(name: str, receiver: Receiver[int]) -> None: - try: - async for message in receiver: - print(f"{name} received {message}") + async def receive(name: str, receiver: Receiver[int]) -> None: + async for message in receiver: + print(f"{name} received {message}") await asyncio.sleep(0.1) # sleep (or work) with the data - except ReceiverStoppedError: - pass async def main() -> None: sender_1, receiver_1 = BroadcastChannel[int](name="numbers") sender_2 = sender_1.clone() receiver_2 = sender_1.subscribe() + async with asyncio.TaskGroup() as task_group: task_group.create_task(send("sender_1", sender_1, 10, 13)) task_group.create_task(send("sender_2", sender_2, 20, 22)) - task_group.create_task(recv("receiver_1", receiver_1)) - task_group.create_task(recv("receiver_2", receiver_2)) + task_group.create_task(receive("receiver_1", receiver_1)) + task_group.create_task(receive("receiver_2", receiver_2)) asyncio.run(main()) ``` - The output should look something like this(although the sending and received + The output should look something like this (although the sending and received might appear interleaved in a different way): ``` @@ -780,16 +949,61 @@ async def main() -> None: sender_2 sending 20 sender_2 sending 21 receiver_1 received 10 - receiver_1 received 11 - receiver_1 received 12 - receiver_1 received 20 - receiver_1 received 21 receiver_2 received 10 + receiver_1 received 11 receiver_2 received 11 + receiver_1 received 12 receiver_2 received 12 + receiver_1 received 20 receiver_2 received 20 + receiver_1 received 21 receiver_2 received 21 ``` + + Example: Stop producing messages when nobody is listening + This example produces messages only for as long as there are receivers to + receive them. + + ```python + import asyncio + + from frequenz.channels import BroadcastChannel, BroadcastSender, Receiver + + + async def produce(sender: BroadcastSender[int]) -> None: + counter = 0 + while sender.has_receivers: + await sender.send(counter) + counter += 1 + await asyncio.sleep(0.1) # wait for (or compute) the next message + print(f"no receivers left, stopping after {counter} messages") + + + async def consume(receiver: Receiver[int]) -> None: + for _ in range(3): + print(f"received {await receiver.receive()}") + receiver.close() + + + async def main() -> None: + sender, receiver = BroadcastChannel[int](name="numbers") + + async with asyncio.TaskGroup() as task_group: + task_group.create_task(produce(sender)) + task_group.create_task(consume(receiver)) + + + asyncio.run(main()) + ``` + + The output should be: + + ``` + received 0 + received 1 + received 2 + no receivers left, stopping after 3 messages + ``` """ def __new__( @@ -805,32 +1019,32 @@ def __new__( name: The name of the channel. This is for logging purposes, and it will be shown in the string representation of the channel. resend_latest: When True, every time a new receiver is created with - [`Broadcast.new_receiver()`][...Broadcast.new_receiver], - the last message seen by the channel will be sent to the - new receiver automatically. This allows new receivers on slow streams to - get the latest message as soon as they are created, without having to - wait for the next message on the channel to arrive. It is safe to be - set in data/reporting channels, but is not recommended for use in - channels that stream control instructions. - limit: Number of messages the receivers can hold in their buffers. + [`BroadcastSender.subscribe()`][...BroadcastSender.subscribe], + the last message that was sent successfully will be sent to the new + receiver automatically, unless it was `None`. This allows new + receivers on slow streams to get the latest message as soon as they + are created, without having to wait for the next message on the + channel to arrive. It is safe to be set in data/reporting channels, + but is not recommended for use in channels that stream control + instructions. + limit: Number of messages each receiver can hold in its buffer, for the + initial receiver and, by default, for receivers created with + [`BroadcastSender.subscribe()`][frequenz.channels.BroadcastSender.subscribe]. warn_on_overflow: Whether to log a warning when a receiver's buffer is full - and a message is dropped. + and a message is dropped. Applies to the initial receiver and, by + default, to receivers created with + [`BroadcastSender.subscribe()`][frequenz.channels.BroadcastSender.subscribe]. Returns: A new BroadcastChannel instance that can be destructured into an initial sender and receiver. """ - channel = Broadcast[ChannelMessageT]( + channel = _Broadcast[ChannelMessageT]( name=name, resend_latest=resend_latest, auto_close=True ) - return tuple.__new__( - cls, - ( - channel.new_sender(), - channel.new_receiver( - name=f"{name}_receiver", - limit=limit, - warn_on_overflow=warn_on_overflow, - ), - ), - ) + # pylint: disable=protected-access + channel._subscriber_limit = limit + channel._subscriber_warn_on_overflow = warn_on_overflow + # pylint: enable=protected-access + sender = channel.new_sender() + return tuple.__new__(cls, (sender, sender.subscribe(name=f"{name}_receiver"))) diff --git a/src/frequenz/channels/_exceptions.py b/src/frequenz/channels/_exceptions.py index b9e5f95d..ff88b05e 100644 --- a/src/frequenz/channels/_exceptions.py +++ b/src/frequenz/channels/_exceptions.py @@ -19,11 +19,10 @@ was closed while sending or receiving a message, the original exception will be available as the cause of the exception: -```python show_lines="6:" -from frequenz.channels import Anycast, ChannelClosedError, SenderError +```python show_lines="5:" +from frequenz.channels import BroadcastChannel, ChannelClosedError, SenderError -channel = Anycast[int](name="test-channel") -sender = channel.new_sender() +sender, receiver = BroadcastChannel[int](name="test-channel") try: await sender.send(42) @@ -44,11 +43,10 @@ explicitly calling [`receive()`][..Receiver.receive] on the receiver after the iteration is done: - ```python show_lines="6:" - from frequenz.channels import Anycast, ChannelClosedError, ReceiverStoppedError + ```python show_lines="5:" + from frequenz.channels import BroadcastChannel, ChannelClosedError, ReceiverStoppedError - channel = Anycast[int](name="test-channel") - receiver = channel.new_receiver() + sender, receiver = BroadcastChannel[int](name="test-channel") async for message in receiver: print(message) diff --git a/src/frequenz/channels/_latest_value_cache.py b/src/frequenz/channels/_latest_value_cache.py index 2530dbe4..32c4fe16 100644 --- a/src/frequenz/channels/_latest_value_cache.py +++ b/src/frequenz/channels/_latest_value_cache.py @@ -17,16 +17,18 @@ Example: ```python - from frequenz.channels import Broadcast, LatestValueCache + import asyncio - channel = Broadcast[int](name="lvc_test") + from frequenz.channels import BroadcastChannel, LatestValueCache - cache = LatestValueCache(channel.new_receiver()) - sender = channel.new_sender() + sender, receiver = BroadcastChannel[int](name="lvc_test") + + cache = LatestValueCache(receiver) assert not cache.has_value() await sender.send(5) + await asyncio.sleep(0) # Let the cache process the message assert cache.has_value() assert cache.get() == 5 diff --git a/src/frequenz/channels/_merge.py b/src/frequenz/channels/_merge.py index b412d4ca..ce066171 100644 --- a/src/frequenz/channels/_merge.py +++ b/src/frequenz/channels/_merge.py @@ -9,21 +9,19 @@ stream, you can use [`merge()`][.merge] to create a new receiver that will receive messages from all the given receivers: -```python show_lines="8:" -from frequenz.channels import Anycast, Receiver, merge +```python show_lines="6:" +from frequenz.channels import BroadcastChannel, Receiver, merge -channel1: Anycast[int] = Anycast(name="channel1") -channel2: Anycast[int] = Anycast(name="channel2") -receiver1 = channel1.new_receiver() -receiver2 = channel2.new_receiver() +sender1, receiver1 = BroadcastChannel[int](name="channel1") +sender2, receiver2 = BroadcastChannel[int](name="channel2") async for message in merge(receiver1, receiver2): print(message) ``` -If the first message comes from `channel2` and the second message from `channel1`, the -first message will be received immediately, and the second message will be received as -soon as it is available. +If the first message comes from `receiver2` and the second message from `receiver1`, +the first message will be received immediately, and the second message will be +received as soon as it is available. This can be helpful when you just need to receive messages and don't care about where are they coming from specifically. If you need to know where the message came @@ -69,12 +67,10 @@ def merge(*receivers: Receiver[ReceiverMessageT_co]) -> Merger[ReceiverMessageT_ stream like this: ```python - from frequenz.channels import Broadcast + from frequenz.channels import BroadcastChannel, merge - channel1 = Broadcast[int](name="input-channel-1") - channel2 = Broadcast[int](name="input-channel-2") - receiver1 = channel1.new_receiver() - receiver2 = channel2.new_receiver() + sender1, receiver1 = BroadcastChannel[int](name="input-channel-1") + sender2, receiver2 = BroadcastChannel[int](name="input-channel-2") async for message in merge(receiver1, receiver2): print(f"received {message}") diff --git a/src/frequenz/channels/_oneshot.py b/src/frequenz/channels/_oneshot.py index 3147d4ac..4d2bb9c4 100644 --- a/src/frequenz/channels/_oneshot.py +++ b/src/frequenz/channels/_oneshot.py @@ -25,7 +25,7 @@ class _Oneshot(typing.Generic[ChannelMessageT]): A one-shot channel is a channel that can only send one message. After the first message is sent, the sender is closed and any further attempts to send a message - will raise a [`SenderClosedError`][..SenderClosedError]. + will raise a [`SenderClosedError`][frequenz.channels.SenderClosedError]. """ def __init__(self) -> None: @@ -116,7 +116,7 @@ class OneshotChannel( A one-shot channel is a channel that can only send one message. After the first message is sent, the sender is closed and any further attempts to send a message - will raise a [`SenderClosedError`][..SenderClosedError]. + will raise a [`SenderClosedError`][frequenz.channels.SenderClosedError]. Example: Sending a message from one task to another ```python diff --git a/src/frequenz/channels/_receiver.py b/src/frequenz/channels/_receiver.py index ef1ea594..6cdc4a9f 100644 --- a/src/frequenz/channels/_receiver.py +++ b/src/frequenz/channels/_receiver.py @@ -7,15 +7,15 @@ Messages are received from [channels](/user-guide/channels/index.md) through [`Receiver`][..Receiver] objects. [`Receiver`][..Receiver]s -are usually created by calling `channel.new_receiver()` and are [async -iterators][typing.AsyncIterator], so the easiest way to receive messages from them as -a stream is to use `async for`: +usually come from creating a channel, like a +[`BroadcastChannel`][..BroadcastChannel], or from calling +`sender.subscribe()`. They are [async iterators][typing.AsyncIterator], so the +easiest way to receive messages from them as a stream is to use `async for`: -```python show_lines="6:" -from frequenz.channels import Anycast +```python show_lines="5:" +from frequenz.channels import BroadcastChannel -channel = Anycast[int](name="test-channel") -receiver = channel.new_receiver() +sender, receiver = BroadcastChannel[int](name="test-channel") async for message in receiver: print(message) @@ -24,11 +24,10 @@ If you need to receive messages in different places or expecting a particular sequence, you can use the [`receive()`][..Receiver.receive] method: -```python show_lines="6:" -from frequenz.channels import Anycast +```python show_lines="5:" +from frequenz.channels import BroadcastChannel -channel = Anycast[int](name="test-channel") -receiver = channel.new_receiver() +sender, receiver = BroadcastChannel[int](name="test-channel") first_message = await receiver.receive() print(f"First message: {first_message}") @@ -42,11 +41,10 @@ If you need to transform the received messages, receivers provide a [`map()`][..Receiver.map] method to easily do so: -```python show_lines="6:" -from frequenz.channels import Anycast +```python show_lines="5:" +from frequenz.channels import BroadcastChannel -channel = Anycast[int](name="test-channel") -receiver = channel.new_receiver() +sender, receiver = BroadcastChannel[int](name="test-channel") async for message in receiver.map(lambda x: x + 1): print(message) @@ -60,11 +58,10 @@ If you need to filter the received messages, receivers provide a [`filter()`][..Receiver.filter] method to easily do so: -```python show_lines="6:" -from frequenz.channels import Anycast +```python show_lines="5:" +from frequenz.channels import BroadcastChannel -channel = Anycast[int](name="test-channel") -receiver = channel.new_receiver() +sender, receiver = BroadcastChannel[int](name="test-channel") async for message in receiver.filter(lambda x: x % 2 == 0): print(message) # Only even numbers will be printed @@ -90,11 +87,10 @@ closed), a [`ReceiverStoppedError`][..ReceiverStoppedError] exception is raised by [`receive()`][..Receiver.receive] method. -```python show_lines="6:" -from frequenz.channels import Anycast +```python show_lines="5:" +from frequenz.channels import BroadcastChannel, ReceiverError, ReceiverStoppedError -channel = Anycast[int](name="test-channel") -receiver = channel.new_receiver() +sender, receiver = BroadcastChannel[int](name="test-channel") try: await receiver.receive() @@ -107,11 +103,10 @@ When used as an async iterator, the iteration will just stop without raising an exception: -```python show_lines="6:" -from frequenz.channels import Anycast +```python show_lines="5:" +from frequenz.channels import BroadcastChannel, ReceiverError, ReceiverStoppedError -channel = Anycast[int](name="test-channel") -receiver = channel.new_receiver() +sender, receiver = BroadcastChannel[int](name="test-channel") try: async for message in receiver: @@ -131,11 +126,10 @@ [`Receiver.close()`][..Receiver.close] method can be used for this purpose. -After [`Receiver.close()`][..Receiver.close] is called, we can still receive -messages that are in the receiver's -buffer, but as soon as the receiver's buffer has been drained, trying to receive further -messages from a *closed* receiver will raise a -[`ReceiverStoppedError`][..ReceiverStoppedError]. +After [`Receiver.close()`][..Receiver.close] is called, some receivers still +deliver the messages in their buffer, and raise a +[`ReceiverStoppedError`][..ReceiverStoppedError] once it is drained. Others, +like [`BroadcastReceiver`][..BroadcastReceiver], raise it at once. # Advanced Usage @@ -235,8 +229,10 @@ def close(self) -> None: """Close the receiver. After calling this method, new messages will not be available from the receiver. - Once the receiver's buffer is drained, trying to receive a message will raise a - [`ReceiverStoppedError`][...ReceiverStoppedError]. + Trying to receive a message raises a + [`ReceiverStoppedError`][...ReceiverStoppedError], either at once or once the + messages already in the receiver's buffer are received, depending on the + receiver. """ raise NotImplementedError("close() must be implemented by subclasses") diff --git a/src/frequenz/channels/_select.py b/src/frequenz/channels/_select.py index d6130b5e..655c0145 100644 --- a/src/frequenz/channels/_select.py +++ b/src/frequenz/channels/_select.py @@ -18,13 +18,11 @@ function to check the source of the message, and the [`message`][..Selected.message] attribute to access the message: -```python show_lines="8:" -from frequenz.channels import Anycast, ReceiverStoppedError, select, selected_from +```python show_lines="6:" +from frequenz.channels import BroadcastChannel, ReceiverStoppedError, select, selected_from -channel1: Anycast[int] = Anycast(name="channel1") -channel2: Anycast[str] = Anycast(name="channel2") -receiver1 = channel1.new_receiver() -receiver2 = channel2.new_receiver() +sender1, receiver1 = BroadcastChannel[int](name="channel1") +sender2, receiver2 = BroadcastChannel[str](name="channel2") async for selected in select(receiver1, receiver2): if selected_from(selected, receiver1): @@ -46,13 +44,11 @@ If for some reason you want to ignore a received message, just add the receiver to the if-chain and do nothing with the message: - ```python show_lines="8:" - from frequenz.channels import Anycast, select, selected_from + ```python show_lines="6:" + from frequenz.channels import BroadcastChannel, select, selected_from - channel1: Anycast[int] = Anycast(name="channel1") - channel2: Anycast[str] = Anycast(name="channel2") - receiver1 = channel1.new_receiver() - receiver2 = channel2.new_receiver() + sender1, receiver1 = BroadcastChannel[int](name="channel1") + sender2, receiver2 = BroadcastChannel[str](name="channel2") async for selected in select(receiver1, receiver2): if selected_from(selected, receiver1): @@ -71,13 +67,11 @@ [`was_stopped`][..Selected.was_stopped] method to check if the selected [`Receiver`][..Receiver] was stopped: -```python show_lines="8:" -from frequenz.channels import Anycast, select, selected_from +```python show_lines="6:" +from frequenz.channels import BroadcastChannel, select, selected_from -channel1: Anycast[int] = Anycast(name="channel1") -channel2: Anycast[str] = Anycast(name="channel2") -receiver1 = channel1.new_receiver() -receiver2 = channel2.new_receiver() +sender1, receiver1 = BroadcastChannel[int](name="channel1") +sender2, receiver2 = BroadcastChannel[str](name="channel2") async for selected in select(receiver1, receiver2): if selected_from(selected, receiver1): @@ -106,13 +100,11 @@ You can use a try-except block to handle exceptions as usual: -```python show_lines="8:" -from frequenz.channels import Anycast, ReceiverStoppedError, select, selected_from +```python show_lines="6:" +from frequenz.channels import BroadcastChannel, ReceiverStoppedError, select, selected_from -channel1: Anycast[int] = Anycast(name="channel1") -channel2: Anycast[str] = Anycast(name="channel2") -receiver1 = channel1.new_receiver() -receiver2 = channel2.new_receiver() +sender1, receiver1 = BroadcastChannel[int](name="channel1") +sender2, receiver2 = BroadcastChannel[str](name="channel2") async for selected in select(receiver1, receiver2): if selected_from(selected, receiver1): diff --git a/src/frequenz/channels/_sender.py b/src/frequenz/channels/_sender.py index 7149880c..86ca72ad 100644 --- a/src/frequenz/channels/_sender.py +++ b/src/frequenz/channels/_sender.py @@ -6,15 +6,16 @@ # Senders Messages are sent to a [channel](/user-guide/channels) through -[`Sender`][..Sender] objects. [`Sender`][..Sender]s are -usually created by calling `channel.new_sender()`, and are a very simple abstraction -that only provides a single [`send()`][..Sender.send] method: +[`Sender`][..Sender] objects. [`Sender`][..Sender]s +usually come from creating a channel, like a +[`BroadcastChannel`][..BroadcastChannel], or from calling +`sender.clone()`. They are a very simple abstraction that mainly provides a +[`send()`][..Sender.send] method: -```python show_lines="6:" -from frequenz.channels import Anycast +```python show_lines="5:" +from frequenz.channels import BroadcastChannel -channel = Anycast[int](name="test-channel") -sender = channel.new_sender() +sender, receiver = BroadcastChannel[str](name="test-channel") await sender.send("Hello, world!") ``` @@ -36,11 +37,10 @@ If there is any failure sending a message, a [`SenderError`][..SenderError] exception is raised. -```python show_lines="6:" -from frequenz.channels import Anycast +```python show_lines="5:" +from frequenz.channels import BroadcastChannel, SenderError -channel = Anycast[int](name="test-channel") -sender = channel.new_sender() +sender, receiver = BroadcastChannel[str](name="test-channel") try: await sender.send("Hello, world!") @@ -114,6 +114,24 @@ def __init__(self, sender: Sender[SenderMessageT_co]): super().__init__("Sender is closed", sender) +class NoReceiversError(SenderError[SenderMessageT_co]): + """An error indicating that a message was sent to a channel without receivers. + + It is raised by the senders of a + [`BroadcastChannel`][frequenz.channels.BroadcastChannel], and of a `Broadcast` + created with `auto_close=True`. The message is dropped, but the channel stays + open, so sending works again once a new receiver is attached. + """ + + def __init__(self, sender: Sender[SenderMessageT_co]): + """Initialize this error. + + Args: + sender: The [`Sender`][...Sender] that tried to send the message. + """ + super().__init__(f"Sender {sender} has no receivers", sender) + + class SubscribableSender(Sender[SenderMessageT_contra], ABC): """A [`Sender`][..Sender] that can be subscribed to.""" diff --git a/src/frequenz/channels/event.py b/src/frequenz/channels/event.py index c40b4f6d..79767c21 100644 --- a/src/frequenz/channels/event.py +++ b/src/frequenz/channels/event.py @@ -46,12 +46,10 @@ class Event(Receiver[None]): ```python import asyncio - from frequenz.channels import Anycast, select, selected_from + from frequenz.channels import BroadcastChannel, select, selected_from from frequenz.channels.event import Event - channel: Anycast[int] = Anycast(name="channel") - receiver = channel.new_receiver() - sender = channel.new_sender() + sender, receiver = BroadcastChannel[int](name="channel") stop_event = Event(name="stop") diff --git a/src/frequenz/channels/experimental/_grouping_latest_value_cache.py b/src/frequenz/channels/experimental/_grouping_latest_value_cache.py index 96f59029..e6c2d12d 100644 --- a/src/frequenz/channels/experimental/_grouping_latest_value_cache.py +++ b/src/frequenz/channels/experimental/_grouping_latest_value_cache.py @@ -67,18 +67,20 @@ class GroupingLatestValueCache(Mapping[HashableT, ValueT_co]): Example: ```python - from frequenz.channels import Broadcast + import asyncio + + from frequenz.channels import BroadcastChannel from frequenz.channels.experimental import GroupingLatestValueCache - channel = Broadcast[tuple[int, str]](name="lvc_test") + sender, receiver = BroadcastChannel[tuple[int, str]](name="lvc_test") - cache = GroupingLatestValueCache(channel.new_receiver(), key=lambda x: x[0]) - sender = channel.new_sender() + cache = GroupingLatestValueCache(receiver, key=lambda x: x[0]) assert cache.get(6) is None assert 6 not in cache await sender.send((6, "twenty-six")) + await asyncio.sleep(0) # Let the cache process the message assert 6 in cache assert cache.get(6) == (6, "twenty-six") diff --git a/src/frequenz/channels/experimental/_pipe.py b/src/frequenz/channels/experimental/_pipe.py index b54f1917..7bdc4efa 100644 --- a/src/frequenz/channels/experimental/_pipe.py +++ b/src/frequenz/channels/experimental/_pipe.py @@ -15,7 +15,7 @@ from .._generic import ChannelMessageT from .._receiver import Receiver -from .._sender import Sender +from .._sender import Sender, SenderError class Pipe(typing.Generic[ChannelMessageT]): @@ -25,33 +25,39 @@ class Pipe(typing.Generic[ChannelMessageT]): creates a pipe between them by forwarding all the messages received by the receiver to the sender. + When its receiver stops, the pipe closes its sender and stops. If sending fails + with a [`SenderError`][frequenz.channels.SenderError], for example because the + sender's channel has no receivers, the pipe closes both its sender and its + receiver, and stops. [`stop()`][frequenz.channels.experimental.Pipe.stop] + leaves both open. Give the pipe a clone of the sender if you still need to send + after the pipe stops on its own. + Example: ```python import asyncio - from contextlib import closing, aclosing, AsyncExitStack + from contextlib import closing, AsyncExitStack - from frequenz.channels import Broadcast + from frequenz.channels import BroadcastChannel from frequenz.channels.experimental import Pipe async def main() -> None: - # Channels, receivers and Pipe are in AsyncExitStack + # Senders, receivers and Pipe are in AsyncExitStack # to close and stop them at the end. async with AsyncExitStack() as stack: - source_channel = await stack.enter_async_context( - aclosing(Broadcast[int](name="source channel")) + source_sender, source_receiver = BroadcastChannel[int]( + name="source channel" ) - source_receiver = stack.enter_context(closing(source_channel.new_receiver())) + stack.push_async_callback(source_sender.aclose) + stack.enter_context(closing(source_receiver)) - forwarding_channel = await stack.enter_async_context( - aclosing(Broadcast[int](name="forwarding channel")) - ) - await stack.enter_async_context( - Pipe(source_receiver, forwarding_channel.new_sender()) + forwarding_sender, receiver = BroadcastChannel[int]( + name="forwarding channel" ) + stack.push_async_callback(forwarding_sender.aclose) + stack.enter_context(closing(receiver)) - receiver = stack.enter_context(closing(forwarding_channel.new_receiver())) + await stack.enter_async_context(Pipe(source_receiver, forwarding_sender)) - source_sender = source_channel.new_sender() await source_sender.send(10) assert await receiver.receive() == 10 @@ -102,4 +108,12 @@ async def stop(self) -> None: async def _run(self) -> None: async for value in self._receiver: - await self._sender.send(value) + try: + await self._sender.send(value) + except SenderError: + try: + await self._sender.aclose() + finally: + self._receiver.close() + return + await self._sender.aclose() diff --git a/src/frequenz/channels/experimental/_relay_sender.py b/src/frequenz/channels/experimental/_relay_sender.py index 5960229d..c412b33c 100644 --- a/src/frequenz/channels/experimental/_relay_sender.py +++ b/src/frequenz/channels/experimental/_relay_sender.py @@ -24,16 +24,13 @@ class RelaySender(Sender[SenderMessageT_contra]): Example: ```python - from frequenz.channels import Broadcast + from frequenz.channels import BroadcastChannel from frequenz.channels.experimental import RelaySender - channel1: Broadcast[int] = Broadcast(name="channel1") - channel2: Broadcast[int] = Broadcast(name="channel2") + sender1, receiver1 = BroadcastChannel[int](name="channel1") + sender2, receiver2 = BroadcastChannel[int](name="channel2") - receiver1 = channel1.new_receiver() - receiver2 = channel2.new_receiver() - - tee_sender = RelaySender(channel1.new_sender(), channel2.new_sender()) + tee_sender = RelaySender(sender1, sender2) await tee_sender.send(5) assert await receiver1.receive() == 5 diff --git a/src/frequenz/channels/experimental/_with_previous.py b/src/frequenz/channels/experimental/_with_previous.py index 9d7420f0..822df390 100644 --- a/src/frequenz/channels/experimental/_with_previous.py +++ b/src/frequenz/channels/experimental/_with_previous.py @@ -30,12 +30,11 @@ class WithPrevious(Generic[ChannelMessageT]): Example: Receiving only messages that are different from the previous one. ```python - from frequenz.channels import Broadcast + from frequenz.channels import BroadcastChannel from frequenz.channels.experimental import WithPrevious - channel = Broadcast[int](name="example") - receiver = channel.new_receiver().filter(WithPrevious(lambda old, new: old != new)) - sender = channel.new_sender() + sender, receiver = BroadcastChannel[int](name="example") + receiver = receiver.filter(WithPrevious(lambda old, new: old != new)) # This message will be received as it is the first message. await sender.send(1) @@ -51,14 +50,13 @@ class WithPrevious(Generic[ChannelMessageT]): Example: Receiving only messages if they are bigger than the previous one. ```python - from frequenz.channels import Broadcast + from frequenz.channels import BroadcastChannel from frequenz.channels.experimental import WithPrevious - channel = Broadcast[int](name="example") - receiver = channel.new_receiver().filter( + sender, receiver = BroadcastChannel[int](name="example") + receiver = receiver.filter( WithPrevious(lambda old, new: new > old, first_is_true=False) ) - sender = channel.new_sender() # This message will skipped as first_is_true is False. await sender.send(1) diff --git a/src/frequenz/channels/timer.py b/src/frequenz/channels/timer.py index 26c41958..37c958ec 100644 --- a/src/frequenz/channels/timer.py +++ b/src/frequenz/channels/timer.py @@ -21,7 +21,7 @@ import asyncio from datetime import datetime, timedelta - from frequenz.channels.timer import Timer + from frequenz.channels.timer import Timer, TriggerAllMissed async def main() -> None: @@ -46,13 +46,12 @@ async def main() -> None: import asyncio from datetime import timedelta - from frequenz.channels import Anycast, select, selected_from - from frequenz.channels.timer import Timer + from frequenz.channels import BroadcastChannel, select, selected_from + from frequenz.channels.timer import SkipMissedAndDrift, Timer async def main() -> None: - channel = Anycast[int](name="data-channel") - data_receiver = channel.new_receiver() + sender, data_receiver = BroadcastChannel[int](name="data-channel") timer = Timer(timedelta(seconds=1.0), SkipMissedAndDrift()) diff --git a/tests/experimental/test_pipe.py b/tests/experimental/test_pipe.py index f0158e14..dc73e24c 100644 --- a/tests/experimental/test_pipe.py +++ b/tests/experimental/test_pipe.py @@ -7,7 +7,17 @@ import typing from contextlib import AsyncExitStack, aclosing, closing -from frequenz.channels import Broadcast, Receiver +import pytest + +from frequenz.channels import ( + Broadcast, + BroadcastChannel, + NoReceiversError, + Receiver, + Sender, + SenderClosedError, + SenderError, +) from frequenz.channels.experimental import Pipe T = typing.TypeVar("T") @@ -25,6 +35,13 @@ async def receive_timeout(recv: Receiver[T], timeout: float = 0.1) -> T | type[T return Timeout +async def wait_stopped(pipe: Pipe[int]) -> None: + """Wait until the task of a started pipe ends.""" + task = pipe._task # pylint: disable=protected-access + assert task is not None + await asyncio.wait_for(task, timeout=1.0) + + async def test_pipe() -> None: """Test pipe.""" channel1: Broadcast[int] = Broadcast(name="channel1") @@ -79,3 +96,117 @@ async def test_pipe_with_async_exit_stack() -> None: source_sender = source_channel.new_sender() await source_sender.send(10) assert await receiver.receive() == 10 + + +async def test_pipe_closes_both_ends_when_sending_fails() -> None: + """Ensure a pipe closes its sender and receiver when it can't send.""" + source_sender, source_receiver = BroadcastChannel[int](name="source") + sender, receiver = BroadcastChannel[int](name="forwarding") + receiver.close() + + pipe = Pipe(source_receiver, sender) + await pipe.start() + with pytest.raises(NoReceiversError): + await sender.send(0) + + await source_sender.send(1) + await wait_stopped(pipe) + + assert not source_sender.has_receivers + with pytest.raises(SenderClosedError): + await sender.send(2) + + +async def test_pipe_closes_sender_when_receiver_stops() -> None: + """Ensure a pipe closes its sender when its receiver stops.""" + source_sender, source_receiver = BroadcastChannel[int](name="source") + sender, receiver = BroadcastChannel[int](name="forwarding") + + pipe = Pipe(source_receiver, sender) + await pipe.start() + await source_sender.send(1) + assert await receive_timeout(receiver) == 1 + + await source_sender.aclose() + await wait_stopped(pipe) + + with pytest.raises(SenderClosedError): + await sender.send(2) + assert not receiver.has_senders + + +async def test_pipe_stop_leaves_both_ends_open() -> None: + """Ensure stopping a pipe leaves its sender and receiver open.""" + source_sender, source_receiver = BroadcastChannel[int](name="source") + sender, receiver = BroadcastChannel[int](name="forwarding") + + pipe = Pipe(source_receiver, sender) + await pipe.start() + await asyncio.sleep(0) + await pipe.stop() + + await sender.send(1) + assert await receive_timeout(receiver) == 1 + await source_sender.send(2) + assert await receive_timeout(source_receiver) == 2 + + +class UncloseableReceiver(Receiver[int]): + """A receiver whose `close()` fails.""" + + def __init__(self, receiver: Receiver[int]) -> None: + """Wrap the given receiver.""" + self._receiver = receiver + + async def ready(self) -> bool: + """Wait until the wrapped receiver is ready.""" + return await self._receiver.ready() + + def consume(self) -> int: + """Return the wrapped receiver's message.""" + return self._receiver.consume() + + def close(self) -> None: + """Fail to close.""" + raise NotImplementedError("close() is not supported") + + +async def test_pipe_closes_sender_when_receiver_cannot_close() -> None: + """Ensure a failed send closes the sender even if the receiver can't close.""" + source_sender, source_receiver = BroadcastChannel[int](name="source") + sender, receiver = BroadcastChannel[int](name="forwarding") + receiver.close() + + pipe = Pipe(UncloseableReceiver(source_receiver), sender) + await pipe.start() + await source_sender.send(1) + with pytest.raises(NotImplementedError): + await wait_stopped(pipe) + + with pytest.raises(SenderClosedError): + await sender.send(2) + + +class UncloseableSender(Sender[int]): + """A sender whose sends and `aclose()` fail.""" + + async def send(self, message: int, /) -> None: + """Fail to send.""" + raise SenderError("Sending failed", self) + + async def aclose(self) -> None: + """Fail to close.""" + raise RuntimeError("Closing failed") + + +async def test_pipe_closes_receiver_when_sender_cannot_close() -> None: + """Ensure a failed send closes the receiver even if the sender can't close.""" + source_sender, source_receiver = BroadcastChannel[int](name="source") + + pipe = Pipe(source_receiver, UncloseableSender()) + await pipe.start() + await source_sender.send(1) + with pytest.raises(RuntimeError, match="Closing failed"): + await wait_stopped(pipe) + + assert not source_sender.has_receivers diff --git a/tests/test_broadcast.py b/tests/test_broadcast.py index 7035250b..d537e638 100644 --- a/tests/test_broadcast.py +++ b/tests/test_broadcast.py @@ -4,6 +4,9 @@ """Tests for the Broadcast implementation.""" import asyncio +import gc +import re +import warnings from dataclasses import dataclass from typing import TypeGuard, assert_never @@ -13,10 +16,14 @@ Broadcast, BroadcastChannel, ChannelClosedError, + NoReceiversError, Receiver, ReceiverStoppedError, Sender, + SenderClosedError, SenderError, + select, + selected_from, ) @@ -427,8 +434,117 @@ async def test_broadcast_close_receiver() -> None: _ = await receiver_2.receive() -async def test_broadcast_auto_close_1() -> None: - """Ensure broadcast auto close works when all receivers are closed.""" +async def test_broadcast_close_receiver_while_waiting() -> None: + """Ensure closing a receiver wakes up a task waiting on it.""" + chan = Broadcast[int](name="input-chan") + receiver = chan.new_receiver() + + waiter = asyncio.create_task(receiver.ready()) + await asyncio.sleep(0) + assert not waiter.done() + + receiver.close() + + assert await asyncio.wait_for(waiter, timeout=1.0) is False + + +async def test_broadcast_closed_receiver_with_buffered_messages() -> None: + """Ensure a closed receiver with buffered messages stops `select()`.""" + chan = Broadcast[int](name="input-chan") + sender = chan.new_sender() + receiver = chan.new_receiver() + filtered = chan.new_receiver().filter(lambda _: True) + await sender.send(1) + receiver.close() + filtered.close() + + assert not await receiver.ready() + + async def stopped(recv: Receiver[int]) -> list[bool]: + return [ + selected.was_stopped + async for selected in select(recv) + if selected_from(selected, recv) + ] + + assert await asyncio.wait_for(stopped(receiver), timeout=1.0) == [True] + assert await asyncio.wait_for(stopped(filtered), timeout=1.0) == [True] + + +async def test_broadcast_receiver_closed_while_waking() -> None: + """Ensure a receiver closed while it wakes up isn't reported as ready.""" + chan = Broadcast[int](name="input-chan") + sender = chan.new_sender() + receiver = chan.new_receiver() + + waiter = asyncio.create_task(receiver.ready()) + await asyncio.sleep(0) + await sender.send(1) + receiver.close() + + assert await asyncio.wait_for(waiter, timeout=1.0) is False + + +def _wait_on_closed_loop(receiver: Receiver[int]) -> None: + """Leave a task waiting on the receiver in an event loop that is then closed.""" + + async def wait() -> None: + asyncio.create_task(receiver.ready()) + await asyncio.sleep(0) + + # The loop is closed without cancelling the waiting task. + loop = asyncio.new_event_loop() + try: + loop.run_until_complete(wait()) + finally: + loop.close() + + +def test_broadcast_wake_after_loop_closed() -> None: + """Ensure waking receivers works after a waiting receiver's loop closed.""" + chan = Broadcast[int](name="input-chan") + sender = chan.new_sender() + receiver = chan.new_receiver() + receiver_2 = chan.new_receiver() + + _wait_on_closed_loop(receiver) + receiver_2.close() + with pytest.raises(ReceiverStoppedError): + receiver_2.consume() + + _wait_on_closed_loop(receiver) + asyncio.run(sender.send(1)) + assert receiver.consume() == 1 + + _wait_on_closed_loop(receiver) + asyncio.run(chan.aclose()) + assert chan.is_closed + + +def test_broadcast_wake_from_other_thread_in_debug_mode() -> None: + """Ensure a wake-up from another thread still fails in debug mode.""" + + async def close_in_thread() -> None: + chan = Broadcast[int](name="input-chan") + receiver = chan.new_receiver() + receiver_2 = chan.new_receiver() + asyncio.create_task(receiver.ready()) + await asyncio.sleep(0) + with pytest.raises(RuntimeError, match="Non-thread-safe operation"): + await asyncio.to_thread(receiver_2.close) + + # The failed wake-up leaves the waiting task stuck, so this loop is closed + # without cancelling it. + loop = asyncio.new_event_loop() + loop.set_debug(True) + try: + loop.run_until_complete(close_in_thread()) + finally: + loop.close() + + +async def test_broadcast_no_receivers_after_all_receivers_closed() -> None: + """Ensure sending fails once all receivers are closed.""" sender, receiver = BroadcastChannel[int](name="auto-close-test") receiver_2 = sender.subscribe() @@ -446,12 +562,16 @@ async def test_broadcast_auto_close_1() -> None: receiver_2.close() - with pytest.raises(SenderError) as excinfo: + with pytest.raises( + SenderError, + match=f"^Sender {re.escape(str(sender))} has no receivers$", + ) as excinfo: await sender.send(3) - assert isinstance(excinfo.value.__cause__, ChannelClosedError) + assert isinstance(excinfo.value, NoReceiversError) + assert excinfo.value.__cause__ is None -async def test_broadcast_auto_close_2() -> None: +async def test_broadcast_auto_close_on_all_senders_closed() -> None: """Ensure broadcast auto close works when all senders are closed.""" sender, receiver = BroadcastChannel[int](name="auto-close-test") @@ -474,11 +594,200 @@ async def test_broadcast_auto_close_2() -> None: assert isinstance(excinfo.value.__cause__, ChannelClosedError) -async def test_broadcast_closed_channels_remain_closed() -> None: - """Ensure that a closed channel can't be resurrected.""" +async def test_broadcast_subscribe_after_last_receiver_closed() -> None: + """Ensure a channel without receivers can get new ones after a failed send.""" sender, receiver = BroadcastChannel[int](name="auto-close-test") receiver.close() + with pytest.raises(NoReceiversError): + await sender.send(0) + + receiver_2 = sender.subscribe() + await sender.send(1) + assert await receiver_2.receive() == 1 + + +async def test_broadcast_no_receivers_after_last_receiver_dropped() -> None: + """Ensure sending fails once the last receiver is garbage-collected.""" + sender, receiver = BroadcastChannel[int](name="auto-close-test") + + del receiver + gc.collect() + + with pytest.raises(NoReceiversError): + await sender.send(1) + + +async def test_broadcast_no_auto_close_send_without_receivers() -> None: + """Ensure a non-auto-closing `Broadcast` keeps a message sent without receivers. + + With `resend_latest=True`, a receiver created later gets it. + """ + chan = Broadcast[int](name="no-auto-close-test", resend_latest=True) + sender = chan.new_sender() + + await sender.send(1) + + receiver = chan.new_receiver() + assert await asyncio.wait_for(receiver.receive(), timeout=1.0) == 1 + + +async def test_broadcast_failed_send_not_resent() -> None: + """Ensure a message sent without receivers isn't resent to new receivers.""" + sender, receiver = BroadcastChannel[int](name="resend-test", resend_latest=True) + await sender.send(1) + receiver.close() + + with pytest.raises(NoReceiversError): + await sender.send(2) + + receiver_2 = sender.subscribe() + assert await asyncio.wait_for(receiver_2.receive(), timeout=1.0) == 1 + + +async def test_broadcast_no_receivers_deprecated_auto_close() -> None: + """Ensure an auto-closing `Broadcast` stays open when sending without receivers.""" + chan = Broadcast[int](name="auto-close-test", auto_close=True) + sender = chan.new_sender() + + with pytest.raises(NoReceiversError): + await sender.send(1) + assert not chan.is_closed + + +async def test_broadcast_auto_close_on_last_sender_dropped() -> None: + """Ensure dropping the last sender stops a receiver waiting for messages.""" + sender, receiver = BroadcastChannel[int](name="auto-close-test") + sender_2 = sender.clone() + await sender.aclose() + + waiter = asyncio.create_task(receiver.receive()) + await asyncio.sleep(0) + assert not waiter.done() + + del sender_2 + gc.collect() + + with pytest.raises(ReceiverStoppedError) as excinfo: + await asyncio.wait_for(waiter, timeout=1.0) + assert isinstance(excinfo.value.__cause__, ChannelClosedError) + + +def test_broadcast_auto_close_on_last_sender_dropped_in_thread() -> None: + """Ensure a sender garbage-collected in another thread wakes up the receivers.""" + + async def drop_sender_in_thread() -> None: + sender, receiver = BroadcastChannel[int](name="auto-close-test") + senders = [sender] + del sender + + waiter = asyncio.create_task(receiver.receive()) + await asyncio.sleep(0) + assert not waiter.done() + + await asyncio.to_thread(senders.clear) + + done, _ = await asyncio.wait({waiter}, timeout=1.0) + assert waiter in done + with pytest.raises(ReceiverStoppedError) as excinfo: + waiter.result() + assert isinstance(excinfo.value.__cause__, ChannelClosedError) + + # Debug mode makes the loop raise if the worker thread wakes the receivers + # directly, without `call_soon_threadsafe()`, which leaves the waiter stuck. + # This loop is closed without cancelling its tasks, so a stuck waiter fails + # the test instead of hanging it. + loop = asyncio.new_event_loop() + loop.set_debug(True) + try: + loop.run_until_complete(drop_sender_in_thread()) + finally: + loop.close() + + +async def test_broadcast_last_sender_aclose_wakes_receivers() -> None: + """Ensure closing the last sender wakes up a waiting receiver at once.""" + sender, receiver = BroadcastChannel[int](name="auto-close-test") + + waiter = asyncio.create_task(receiver.ready()) + await asyncio.sleep(0) + assert not waiter.done() + + await sender.aclose() + await asyncio.sleep(0) + + assert waiter.done() + assert waiter.result() is False + + +async def test_broadcast_closed_sender_dropped() -> None: + """Ensure a closed sender isn't detached again when it is garbage-collected.""" + sender, receiver = BroadcastChannel[int](name="auto-close-test") + sender_2 = sender.clone() + + await sender.aclose() + del sender + gc.collect() + + assert sender_2.sender_count == 1 + await sender_2.send(1) + assert await receiver.receive() == 1 + + +async def test_broadcast_finalized_sender_closed() -> None: + """Ensure closing a sender whose finalizer already ran doesn't detach it again.""" + sender, receiver = BroadcastChannel[int](name="auto-close-test") + sender_2 = sender.clone() + + # Interpreter exit runs the finalizers of senders that are still alive. + sender._detach() # pylint: disable=protected-access + await sender.aclose() + + assert sender_2.sender_count == 1 + await sender_2.send(1) + assert await receiver.receive() == 1 + + +def test_broadcast_last_sender_dropped_after_loop_closed() -> None: + """Ensure dropping the last sender after the receivers' loop closed is fine.""" + sender, receiver = BroadcastChannel[int](name="auto-close-test") + + async def wait() -> None: + # Let the receiver wait, so the channel knows the receivers' event loop. + _ = asyncio.create_task(receiver.ready()) + await asyncio.sleep(0) + + asyncio.run(wait()) + + del sender + gc.collect() + + with pytest.raises(ReceiverStoppedError): + asyncio.run(receiver.receive()) + + +async def test_broadcast_no_auto_close_on_sender_dropped() -> None: + """Ensure dropping all senders doesn't close a non-auto-closing channel.""" + chan = Broadcast[int](name="input-chan") + receiver = chan.new_receiver() + + sender = chan.new_sender() + del sender + gc.collect() + + assert not chan.is_closed + sender = chan.new_sender() + assert sender.sender_count == 1 + await sender.send(1) + assert await receiver.receive() == 1 + + +async def test_broadcast_closed_channels_remain_closed() -> None: + """Ensure that a closed channel can't be resurrected.""" + chan = Broadcast[int](name="closed-test") + sender = chan.new_sender() + + await chan.aclose() with pytest.raises(SenderError) as excinfo: await sender.send(1) @@ -487,3 +796,111 @@ async def test_broadcast_closed_channels_remain_closed() -> None: with pytest.raises(SenderError) as excinfo: _ = sender.subscribe() assert isinstance(excinfo.value.__cause__, ChannelClosedError) + + with pytest.raises(SenderError) as excinfo: + _ = sender.clone() + assert isinstance(excinfo.value.__cause__, ChannelClosedError) + + +async def test_broadcast_closed_sender() -> None: + """Ensure a closed sender can't be used to send, clone or subscribe.""" + sender, _receiver = BroadcastChannel[int](name="closed-sender-test") + sender_2 = sender.clone() + await sender.aclose() + assert sender_2.sender_count == 1 + + with pytest.raises(SenderClosedError): + await sender.send(1) + with pytest.raises(SenderClosedError): + _ = sender.clone() + with pytest.raises(SenderClosedError): + _ = sender.subscribe() + + +async def test_broadcast_subscribe_inherits_receiver_options() -> None: + """Ensure subscribed receivers default to the channel's receiver options.""" + sender, receiver = BroadcastChannel[int]( + name="options-test", limit=2, warn_on_overflow=False + ) + subscribed = sender.subscribe() + overridden = sender.subscribe(limit=3, warn_on_overflow=True) + + for i in range(4): + await sender.send(i) + + assert [len(receiver), len(subscribed), len(overridden)] == [2, 2, 3] + # pylint: disable=protected-access + assert not subscribed._warn_on_overflow + assert overridden._warn_on_overflow + # pylint: enable=protected-access + + +async def test_broadcast_has_receivers() -> None: + """Ensure senders report whether their channel has receivers.""" + sender, receiver = BroadcastChannel[int](name="has-receivers-test") + receiver_2 = sender.subscribe() + assert sender.has_receivers + + receiver.close() + assert sender.has_receivers + + receiver_2.close() + assert not sender.has_receivers + + +async def test_broadcast_has_senders() -> None: + """Ensure receivers report whether their channel has senders.""" + sender, receiver = BroadcastChannel[int](name="has-senders-test") + sender_2 = sender.clone() + assert receiver.has_senders + + await sender.aclose() + assert receiver.has_senders + + await sender_2.send(1) + await sender_2.aclose() + assert not receiver.has_senders + assert await receiver.receive() == 1 + + +async def test_broadcast_has_receivers_and_senders_no_auto_close() -> None: + """Ensure non-auto-closing channels report their senders and receivers.""" + chan = Broadcast[int](name="input-chan") + sender = chan.new_sender() + assert not sender.has_receivers + + receiver = chan.new_receiver() + assert sender.has_receivers + assert receiver.has_senders + + await sender.aclose() + assert not receiver.has_senders + + sender_2 = chan.new_sender() + assert receiver.has_senders + + await chan.aclose() + assert not receiver.has_senders + assert not sender_2.has_receivers + + +async def test_broadcast_channel_str() -> None: + """Ensure BroadcastChannel endpoints are named like Broadcast ones.""" + sender, receiver = BroadcastChannel[int](name="str-test") + + assert str(sender) == "Broadcast:str-test:BroadcastSender" + assert str(receiver) == "Broadcast:str-test:BroadcastReceiver:str-test_receiver" + assert repr(sender).startswith("BroadcastSender(Broadcast(name='str-test'") + + +def test_broadcast_deprecated() -> None: + """Ensure creating a Broadcast emits a deprecation warning.""" + with pytest.deprecated_call(match="Please use BroadcastChannel instead."): + _ = Broadcast[int](name="deprecated-test") + + +async def test_broadcast_channel_not_deprecated() -> None: + """Ensure creating a BroadcastChannel doesn't emit deprecation warnings.""" + with warnings.catch_warnings(): + warnings.simplefilter("error") + _ = BroadcastChannel[int](name="no-warnings-test")