Clean up and improve the code and public interface (#231)

This PR addresses many small cleanups and improvements to the code and
the public interface, including some minor breaking changes.

Cleanups:

- Use `deque` for typing
- Replace `__anext__()` with `anext()`
- Rename `recv` argument as `receiver`
- Add missing types to members in `__init__()`
- Anycast: Don't store the `_limit`
- Use the string representation for logging
- Broadcast: Move the `resend_latest` attribute to the end
- Remove `logging` from `Timer` examples
- Make `Broadcast` logger private

Improvements:

- Add `is_closed` property to channels
- Anycast: Clarify what happens when the buffer is full
- Anycast: Add `name` argument
- Add `name` property to bidirectional
- Broadcast: Use `hash()` instead of a UUID
- Event: Make the default name more readable
- Add descriptive `str` and `repr` implementations
- Bump `repo-config` to v0.7.5
- Bump `pylint` to 3.0.2
- Bump `mike` to 2.0.0
- Anycast: Log a warning when a sender is blocked

Breaking changes:

- Make some arguments keyword-only
- Make exception messages `str`
- Rename `maxsize` to `limit`
- Broadcast: Make `name` keyword-only
- Bidirectional: unify `client_id` and `service_id` (fixes #187).

Author: llucax

RELEASE_NOTES.md

@@ -6,26 +6,89 @@ The `Timer` now can be started with a delay.
 
 ## Upgrading
 
-* Internal variable names in the `Anycast` and `Broadcast` implementations are now private.
+* `Anycast`
+
+  - `__init__`: The `maxsize` argument was renamed to `limit` and made keyword-only and a new keyword-only `name` argument was added.
+
+    You should instantiate using `Anycast(name=..., limit=...)` (or `Anycast(name=...)` if the default `limit` is enough) instead of `Anycast(...)` or `Anycast(maxsize=...)`.
+
+* `Bidirectional`
+
+  - The `client_id` and `service_id` arguments were merged into a keyword-only `name`.
+
+    You should instantiate using `Bidirectional(name=...)` instead of `Bidirectional(..., ...)` or `Bidirectional(client_id=..., service_id=...)`.
+
+* `Broadcast`
+
+  - `__init__`: The `name` argument was made optional and keyword-only; `resend_latest` was also made keyword-only. If a `name` is not specified, it will be generated from the `id()` of the instance.
+
+    You should instantiate using `Broadcast(name=name, resend_latest=resend_latest)` (or `Broadcast()` if the defaults are enough) instead of `Broadcast(name)` or `Broadcast(name, resend_latest)`.
+
+  - `new_receiver`: The `maxsize` argument was renamed to `limit` and made keyword-only; the `name` argument was also made keyword-only. If a `name` is not specified, it will be generated from the `id()` of the instance instead of a random UUID.
+
+    You should use `.new_receiver(name=name, limit=limit)` (or `.new_receiver()` if the defaults are enough) instead of `.new_receiver(name)` or `.new_receiver(name, maxsize)`.
+
+* `Event`
+
+  - `__init__`: The `name` argument was made keyword-only. The default was changed to a more readable version of `id(self)`.
+
+    You should instantiate using `Event(name=...)` instead of `Event(...)`.
+
+* All exceptions that took `Any` as the `message` argument now take `str` instead.
+
+  If you were passing a non-`str` value to an exception, you should convert it using `str(value)` before passing it to the exception.
 
 ## New Features
 
-* `Timer()`, `Timer.timeout()`, `Timer.periodic()` and `Timer.reset()` now take an optional `start_delay` option to make the timer start after some delay.
+* `Anycast`
+
+  - The following new read-only properties were added:
+
+    - `name`: The name of the channel.
+    - `limit`: The maximum number of messages that can be sent to the channel.
+    - `is_closed`: Whether the channel is closed.
+
+  - A more useful implementation of `__str__ and `__repr__` were added for the channel and its senders and receivers.
+
+  - A warning will be logged if senders are blocked because the channel buffer is full.
+
+* `Bidirectional`
+
+  - The following new read-only properties were added:
+
+    - `name`: The name of the channel (read-only).
+    - `is_closed`: Whether the channel is closed (read-only).
+
+  - A more useful implementation of `__str__ and `__repr__` were added for the channel and the client and service handles.
+
+* `Broadcast`
+
+  - The following new read-only properties were added:
+
+    - `name`: The name of the channel.
+    - `is_closed`: Whether the channel is closed.
+
+  - A more useful implementation of `__str__ and `__repr__` were added for the channel and the client and service handles.
+
+* `FileWatcher`
+
+  - A more useful implementation of `__str__ and `__repr__` were added.
+
+* `Merge`
+
+  - A more useful implementation of `__str__ and `__repr__` were added.
+
+* `MergeNamed`
 
-  This can be useful, for example, if the timer needs to be *aligned* to a particular time. The alternative to this would be to `sleep()` for the time needed to align the timer, but if the `sleep()` call gets delayed because the event loop is busy, then a re-alignment is needed and this could go on for a while. The only way to guarantee a certain alignment (with a reasonable precision) is to delay the timer start.
+  - A more useful implementation of `__str__ and `__repr__` were added.
 
-* `Broadcast.resend_latest` is now a public attribute, allowing it to be changed after the channel is created.
+* `Peekable`
 
-* The arm64 architecture is now officially supported.
+  - A more useful implementation of `__str__ and `__repr__` were added.
 
-* The documentation was improved to:
+* `Receiver`
 
-  - Show signatures with types.
-  - Show the inherited members.
-  - Documentation for pre-releases are now published.
-  - Show the full tag name as the documentation version.
-  - All development branches now have their documentation published (there is no `next` version anymore).
-  - Fix the order of the documentation versions.
+  - `map()`: The returned map object now has a more useful implementation of `__str__ and `__repr__`.
 
 ## Bug Fixes
 

benchmarks/benchmark_anycast.py

@@ -41,7 +41,9 @@ async def benchmark_anycast(
     Returns:
         Total number of messages received by all channels.
     """
-    channels: list[Anycast[int]] = [Anycast(buffer_size) for _ in range(num_channels)]
+    channels: list[Anycast[int]] = [
+        Anycast(name="test", limit=buffer_size) for _ in range(num_channels)
+    ]
     senders = [
         asyncio.create_task(send_msg(num_messages, bcast.new_sender()))
         for bcast in channels

benchmarks/benchmark_broadcast.py

@@ -61,7 +61,9 @@ async def benchmark_broadcast(
     Returns:
         Total number of messages received by all receivers.
     """
-    channels: list[Broadcast[int]] = [Broadcast("meter") for _ in range(num_channels)]
+    channels: list[Broadcast[int]] = [
+        Broadcast(name="meter") for _ in range(num_channels)
+    ]
     senders: list[asyncio.Task[Any]] = [
         asyncio.create_task(send_msg(num_messages, bcast.new_sender()))
         for bcast in channels
@@ -104,7 +106,9 @@ async def benchmark_single_task_broadcast(
     Returns:
         Total number of messages received by all receivers.
     """
-    channels: list[Broadcast[int]] = [Broadcast("meter") for _ in range(num_channels)]
+    channels: list[Broadcast[int]] = [
+        Broadcast(name="meter") for _ in range(num_channels)
+    ]
     senders = [b.new_sender() for b in channels]
     recv_tracker = 0
 

mkdocs.yml

@@ -93,6 +93,7 @@ plugins:
   - literate-nav:
       nav_file: SUMMARY.md
   - mike:
+      alias_type: redirect
       canonical_version: latest
   - mkdocstrings:
       custom_templates: templates

pyproject.toml

@@ -5,7 +5,7 @@
 requires = [
   "setuptools == 68.1.0",
   "setuptools_scm[toml] == 7.1.0",
-  "frequenz-repo-config[lib] == 0.7.2",
+  "frequenz-repo-config[lib] == 0.7.5",
 ]
 build-backend = "setuptools.build_meta"
 
@@ -47,31 +47,31 @@ dev-formatting = ["black == 23.10.1", "isort == 5.12.0"]
 dev-mkdocs = [
   "black == 23.10.1",
   "Markdown==3.5.1",
-  "mike == 1.1.2",
+  "mike == 2.0.0",
   "mkdocs-gen-files == 0.5.0",
   "mkdocs-literate-nav == 0.6.1",
   "mkdocs-material == 9.4.7",
   "mkdocs-macros-plugin == 1.0.5",
   "mkdocstrings[python] == 0.23.0",
-  "frequenz-repo-config[lib] == 0.7.2",
+  "frequenz-repo-config[lib] == 0.7.5",
 ]
 dev-mypy = [
   "mypy == 1.6.1",
   "types-Markdown == 3.5.0.0",
   # For checking the noxfile, docs/ script, and tests
   "frequenz-channels[dev-mkdocs,dev-noxfile,dev-pytest]",
 ]
-dev-noxfile = ["nox == 2023.4.22", "frequenz-repo-config[lib] == 0.7.2"]
+dev-noxfile = ["nox == 2023.4.22", "frequenz-repo-config[lib] == 0.7.5"]
 dev-pylint = [
-  "pylint == 2.17.7",
+  "pylint == 3.0.2",
   # For checking the noxfile, docs/ script, and tests
   "frequenz-channels[dev-mkdocs,dev-noxfile,dev-pytest]",
 ]
 dev-pytest = [
   "pytest == 7.4.3",
   "async-solipsism == 0.5",
   "hypothesis == 6.88.1",
-  "frequenz-repo-config[extra-lint-examples] == 0.7.2",
+  "frequenz-repo-config[extra-lint-examples] == 0.7.5",
   "pytest-asyncio == 0.21.1",
   "pytest-mock == 3.12.0",
 ]

src/frequenz/channels/_anycast.py

@@ -5,22 +5,31 @@
 
 from __future__ import annotations
 
+import logging
 from asyncio import Condition
 from collections import deque
-from typing import Deque, Generic
+from typing import Generic
 
 from ._base_classes import Receiver as BaseReceiver
 from ._base_classes import Sender as BaseSender
 from ._base_classes import T
 from ._exceptions import ChannelClosedError, ReceiverStoppedError, SenderError
 
+_logger = logging.getLogger(__name__)
+
 
 class Anycast(Generic[T]):
     """A channel for sending data across async tasks.
 
     Anycast channels support multiple senders and multiple receivers.  A message sent
     through a sender will be received by exactly one receiver.
 
+    This channel is buffered, and if the senders are faster than the receivers, then the
+    channel's buffer will fill up. In that case, the senders will block at the
+    [send()][frequenz.channels.Sender.send] method until the receivers consume the
+    messages in the channel's buffer. The channel's buffer size can be configured at
+    creation time via the `limit` argument.
+
     In cases where each message need to be received by every receiver, a
     [Broadcast][frequenz.channels.Broadcast] channel may be used.
 
@@ -61,21 +70,24 @@ async def recv(id: int, receiver: channel.Receiver) -> None:
         Check the `tests` and `benchmarks` directories for more examples.
     """
 
-    def __init__(self, maxsize: int = 10) -> None:
+    def __init__(self, *, name: str, limit: int = 10) -> None:
         """Create an Anycast channel.
 
         Args:
-            maxsize: Size of the channel's buffer.
+            name: The name of the channel. This is for logging purposes, and it will be
+                shown in the string representation of the channel.
+            limit: The size of the internal buffer in number of messages.  If the buffer
+                is full, then the senders will block until the receivers consume the
+                messages in the buffer.
         """
-        self._limit: int = maxsize
-        """The maximum number of values that can be stored in the channel's buffer.
+        self._name: str = name
+        """The name of the channel.
 
-        If the length of channel's buffer reaches the limit, then the sender
-        blocks at the [send()][frequenz.channels.Sender.send] method until
-        a value is consumed.
+        This is for logging purposes, and it will be shown in the string representation
+        of the channel.
         """
 
-        self._deque: Deque[T] = deque(maxlen=maxsize)
+        self._deque: deque[T] = deque(maxlen=limit)
         """The channel's buffer."""
 
         self._send_cv: Condition = Condition()
@@ -97,6 +109,36 @@ def __init__(self, maxsize: int = 10) -> None:
         self._closed: bool = False
         """Whether the channel is closed."""
 
+    @property
+    def name(self) -> str:
+        """The name of this channel.
+
+        This is for debugging purposes, 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
+
+    @property
+    def limit(self) -> int:
+        """The maximum number of values that can be stored in the channel's buffer.
+
+        If the length of channel's buffer reaches the limit, then the sender
+        blocks at the [send()][frequenz.channels.Sender.send] method until
+        a value is consumed.
+        """
+        maxlen = self._deque.maxlen
+        assert maxlen is not None
+        return maxlen
+
     async def close(self) -> None:
         """Close the channel.
 
@@ -131,6 +173,17 @@ def new_receiver(self) -> Receiver[T]:
         """
         return Receiver(self)
 
+    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}, limit={self.limit!r}):<"
+            f"current={len(self._deque)!r}, closed={self._closed!r}>"
+        )
+
 
 class Sender(BaseSender[T]):
     """A sender to send messages to an Anycast channel.
@@ -145,7 +198,7 @@ def __init__(self, chan: Anycast[T]) -> None:
         Args:
             chan: A reference to the channel that this sender belongs to.
         """
-        self._chan = chan
+        self._chan: Anycast[T] = chan
         """The channel that this sender belongs to."""
 
     async def send(self, msg: T) -> None:
@@ -169,14 +222,32 @@ async def send(self, msg: T) -> None:
             raise SenderError("The channel was closed", self) from ChannelClosedError(
                 self._chan
             )
-        while len(self._chan._deque) == self._chan._deque.maxlen:
-            async with self._chan._send_cv:
-                await self._chan._send_cv.wait()
+        if len(self._chan._deque) == self._chan._deque.maxlen:
+            _logger.warning(
+                "Anycast channel [%s] is full, blocking sender until a receiver "
+                "consumes a value",
+                self,
+            )
+            while len(self._chan._deque) == self._chan._deque.maxlen:
+                async with self._chan._send_cv:
+                    await self._chan._send_cv.wait()
+            _logger.info(
+                "Anycast channel [%s] has space again, resuming the blocked sender",
+                self,
+            )
         self._chan._deque.append(msg)
         async with self._chan._recv_cv:
             self._chan._recv_cv.notify(1)
         # pylint: enable=protected-access
 
+    def __str__(self) -> str:
+        """Return a string representation of this sender."""
+        return f"{self._chan}:{type(self).__name__}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this sender."""
+        return f"{type(self).__name__}({self._chan!r})"
+
 
 class _Empty:
     """A sentinel value to indicate that a value has not been set."""
@@ -195,7 +266,7 @@ def __init__(self, chan: Anycast[T]) -> None:
         Args:
             chan: A reference to the channel that this receiver belongs to.
         """
-        self._chan = chan
+        self._chan: Anycast[T] = chan
         """The channel that this receiver belongs to."""
 
         self._next: T | type[_Empty] = _Empty
@@ -251,3 +322,11 @@ def consume(self) -> T:
         self._next = _Empty
 
         return next_val
+
+    def __str__(self) -> str:
+        """Return a string representation of this receiver."""
+        return f"{self._chan}:{type(self).__name__}"
+
+    def __repr__(self) -> str:
+        """Return a string representation of this receiver."""
+        return f"{type(self).__name__}({self._chan!r})"