Remove channel registry

RELEASE_NOTES.md

@@ -17,3 +17,5 @@
 - Improved formula validation: Consistent error messages for invalid formulas and conventional span semantics.
 
 - This fixes a rare power distributor bug where some battery inverters becoming unreachable because of network outages would lead to excess power values getting set.  This is fixed by measuring the power of the unreachable inverters through their fallback meters and excluding that power from what is distributed to the other inverters.
+
+- Removed ChannelRegistry. Streams are not being set up using one-shot channels and owned by the data source.

benchmarks/timeseries/benchmark_datasourcing.py

@@ -16,15 +16,17 @@
 from time import perf_counter
 from typing import Any
 
-from frequenz.channels import Broadcast, Receiver, ReceiverStoppedError
+from frequenz.channels import Broadcast, OneshotChannel, Receiver, ReceiverStoppedError
+from frequenz.channels._broadcast import BroadcastReceiver
 from frequenz.client.microgrid.metrics import Metric
+from frequenz.quantities import Quantity
 
 from frequenz.sdk import microgrid
-from frequenz.sdk._internal._channels import ChannelRegistry
 from frequenz.sdk.microgrid._data_sourcing import (
     ComponentMetricRequest,
     DataSourcingActor,
 )
+from frequenz.sdk.timeseries import Sample
 
 try:
     from tests.timeseries.mock_microgrid import MockMicrogrid
@@ -80,7 +82,6 @@ async def benchmark_data_sourcing(  # pylint: disable=too-many-locals
         name="DataSourcingActor Request Channel"
     )
 
-    channel_registry = ChannelRegistry(name="Microgrid Channel Registry")
     request_receiver = request_channel.new_receiver(
         name="datasourcing-benchmark",
         limit=(num_ev_chargers * len(COMPONENT_METRIC_IDS)),
@@ -105,18 +106,21 @@ async def consume(channel: Receiver[Any]) -> None:
 
     for evc_id in mock_grid.evc_ids:
         for component_metric_id in COMPONENT_METRIC_IDS:
+            telem_stream_sender, telem_stream_receiver = OneshotChannel[
+                BroadcastReceiver[Sample[Quantity]]
+            ]()
             request = ComponentMetricRequest(
-                "current_phase_requests", evc_id, component_metric_id, None
+                namespace="current_phase_requests",
+                component_id=evc_id,
+                metric=component_metric_id,
+                start_time=None,
+                telem_stream_sender=telem_stream_sender,
             )
-
-            recv_channel = channel_registry.get_or_create(
-                ComponentMetricRequest, request.get_channel_name()
-            ).new_receiver()
-
             await request_sender.send(request)
-            consume_tasks.append(asyncio.create_task(consume(recv_channel)))
+            stream_receiver = await telem_stream_receiver.receive()
+            consume_tasks.append(asyncio.create_task(consume(stream_receiver)))
 
-    async with DataSourcingActor(request_receiver, channel_registry):
+    async with DataSourcingActor(request_receiver):
         await asyncio.gather(*consume_tasks)
 
         time_taken = perf_counter() - start_time

pyproject.toml

@@ -32,7 +32,7 @@ dependencies = [
   "frequenz-client-microgrid >= 0.18.1, < 0.19.0",
   "frequenz-microgrid-component-graph >= 0.3.4, < 0.4",
   "frequenz-client-common >= 0.3.6, < 0.4.0",
-  "frequenz-channels >= 1.6.1, < 2.0.0",
+  "frequenz-channels @ git+https://github.com/shsms/frequenz-channels-python.git@auto-closing-broadcast",
   "frequenz-quantities[marshmallow] >= 1.0.0, < 2.0.0",
   "numpy >= 2.1.0, < 3",
   "typing_extensions >= 4.14.1, < 5",

src/frequenz/sdk/_internal/_channels.py

@@ -4,12 +4,10 @@
 """General purpose classes for use with channels."""
 
 import abc
-import dataclasses
 import logging
-import traceback
 import typing
 
-from frequenz.channels import Broadcast, Receiver
+from frequenz.channels import Receiver
 
 _logger = logging.getLogger(__name__)
 
@@ -61,141 +59,3 @@ def new_receiver(self, *, limit: int = 50) -> Receiver[U_co]:
             A receiver instance.
         """
         return self._mapping_function(self._fetcher.new_receiver(limit=limit))
-
-
-class ChannelRegistry:
-    """Dynamically creates, own and provide access to broadcast channels.
-
-    It can be used by actors to dynamically establish a communication channel
-    between each other.
-
-    The registry is responsible for creating channels when they are first requested via
-    the [`get_or_create()`][frequenz.sdk.actor.ChannelRegistry.get_or_create] method.
-
-    The registry also stores type information to make sure that the same channel is not
-    used for different message types.
-
-    Since the registry owns the channels, it is also responsible for closing them when
-    they are no longer needed. There is no way to remove a channel without closing it.
-
-    Note:
-        This registry stores [`Broadcast`][frequenz.channels.Broadcast] channels.
-    """
-
-    def __init__(self, *, name: str) -> None:
-        """Initialize this registry.
-
-        Args:
-            name: A name to identify the registry in the logs. This name is also used as
-                a prefix for the channel names.
-        """
-        self._name = name
-        self._channels: dict[str, _Entry] = {}
-
-    @property
-    def name(self) -> str:
-        """The name of this registry."""
-        return self._name
-
-    def message_type(self, key: str) -> type:
-        """Get the message type of the channel for the given key.
-
-        Args:
-            key: The key to identify the channel.
-
-        Returns:
-            The message type of the channel.
-
-        Raises:
-            KeyError: If the channel does not exist.
-        """
-        entry = self._channels.get(key)
-        if entry is None:
-            raise KeyError(f"No channel for key {key!r} exists.")
-        return entry.message_type
-
-    def __contains__(self, key: str) -> bool:
-        """Check whether the channel for the given `key` exists."""
-        return key in self._channels
-
-    def get_or_create(self, message_type: type[T], key: str) -> Broadcast[T]:
-        """Get or create a channel for the given key.
-
-        If a channel for the given key already exists, the message type of the existing
-        channel is checked against the requested message type. If they do not match,
-        a `ValueError` is raised.
-
-        Note:
-            The types have to match exactly, it doesn't do a subtype check due to
-            technical limitations. In the future subtype checks might be supported.
-
-        Args:
-            message_type: The type of the message that is sent through the channel.
-            key: The key to identify the channel.
-
-        Returns:
-            The channel for the given key.
-
-        Raises:
-            ValueError: If the channel exists and the message type does not match.
-        """
-        if key not in self._channels:
-            if _logger.isEnabledFor(logging.DEBUG):
-                _logger.debug(
-                    "Creating a new channel for key %r with type %s at:\n%s",
-                    key,
-                    message_type,
-                    "".join(traceback.format_stack(limit=10)[:9]),
-                )
-            self._channels[key] = _Entry(
-                message_type, Broadcast(name=f"{self._name}-{key}")
-            )
-
-        entry = self._channels[key]
-        if entry.message_type is not message_type:
-            error_message = (
-                f"Type mismatch, a channel for key {key!r} exists and the requested "
-                f"message type {message_type} is not the same as the existing "
-                f"message type {entry.message_type}."
-            )
-            if _logger.isEnabledFor(logging.DEBUG):
-                _logger.debug(
-                    "%s at:\n%s",
-                    error_message,
-                    # We skip the last frame because it's this method, and limit the
-                    # stack to 9 frames to avoid adding too much noise.
-                    "".join(traceback.format_stack(limit=10)[:9]),
-                )
-            raise ValueError(error_message)
-
-        return typing.cast(Broadcast[T], entry.channel)
-
-    async def close_and_remove(self, key: str) -> None:
-        """Remove the channel for the given key.
-
-        Args:
-            key: The key to identify the channel.
-
-        Raises:
-            KeyError: If the channel does not exist.
-        """
-        entry = self._channels.pop(key, None)
-        if entry is None:
-            raise KeyError(f"No channel for key {key!r} exists.")
-        await entry.channel.close()
-
-
-@dataclasses.dataclass(frozen=True)
-class _Entry:
-    """An entry in a channel registry."""
-
-    message_type: type
-    """The type of the message that is sent through the channel in this entry."""
-
-    # We use object instead of Any to minimize the chances of hindering type checking.
-    # If for some reason the channel is not casted to the proper underlaying type, when
-    # using object at least accessing any member that's not part of the object base
-    # class will yield a type error, while if we used Any, it would not and the issue
-    # would be much harder to find.
-    channel: Broadcast[object]
-    """The channel in this entry."""

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -20,7 +20,6 @@
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.client.microgrid.component import Battery, EvCharger, SolarInverter
 
-from .._internal._channels import ChannelRegistry
 from ..actor._actor import Actor
 from ..timeseries import ResamplerConfig
 from ..timeseries._voltage_streamer import VoltageStreamer
@@ -96,29 +95,22 @@ def __init__(
         """
         self._resampler_config: ResamplerConfig = resampler_config
 
-        self._channel_registry: ChannelRegistry = ChannelRegistry(
-            name="Data Pipeline Registry"
-        )
-
         self._data_sourcing_actor: _ActorInfo | None = None
         self._resampling_actor: _ActorInfo | None = None
 
         self._battery_power_wrapper = PowerWrapper(
-            self._channel_registry,
             api_power_request_timeout=api_power_request_timeout,
             power_manager_algorithm=battery_power_manager_algorithm,
             default_power=DefaultPower.ZERO,
             component_class=Battery,
         )
         self._ev_power_wrapper = PowerWrapper(
-            self._channel_registry,
             api_power_request_timeout=api_power_request_timeout,
             power_manager_algorithm=PowerManagerAlgorithm.MATRYOSHKA,
             default_power=DefaultPower.MAX,
             component_class=EvCharger,
         )
         self._pv_power_wrapper = PowerWrapper(
-            self._channel_registry,
             api_power_request_timeout=api_power_request_timeout,
             power_manager_algorithm=PowerManagerAlgorithm.MATRYOSHKA,
             default_power=DefaultPower.MIN,
@@ -158,18 +150,14 @@ def frequency(self) -> GridFrequency:
         if self._frequency_instance is None:
             self._frequency_instance = GridFrequency(
                 self._data_sourcing_request_sender(),
-                self._channel_registry,
             )
 
         return self._frequency_instance
 
     def voltage_per_phase(self) -> VoltageStreamer:
         """Return the per-phase voltage measuring point."""
         if not self._voltage_instance:
-            self._voltage_instance = VoltageStreamer(
-                self._resampling_request_sender(),
-                self._channel_registry,
-            )
+            self._voltage_instance = VoltageStreamer(self._resampling_request_sender())
 
         return self._voltage_instance
 
@@ -179,7 +167,6 @@ def logical_meter(self) -> LogicalMeter:
 
         if self._logical_meter is None:
             self._logical_meter = LogicalMeter(
-                channel_registry=self._channel_registry,
                 resampler_subscription_sender=self._resampling_request_sender(),
             )
         return self._logical_meter
@@ -190,7 +177,6 @@ def consumer(self) -> Consumer:
 
         if self._consumer is None:
             self._consumer = Consumer(
-                channel_registry=self._channel_registry,
                 resampler_subscription_sender=self._resampling_request_sender(),
             )
         return self._consumer
@@ -201,7 +187,6 @@ def producer(self) -> Producer:
 
         if self._producer is None:
             self._producer = Producer(
-                channel_registry=self._channel_registry,
                 resampler_subscription_sender=self._resampling_request_sender(),
             )
         return self._producer
@@ -213,7 +198,6 @@ def grid(self) -> Grid:
 
         if self._grid is None:
             initialize_grid(
-                channel_registry=self._channel_registry,
                 resampler_subscription_sender=self._resampling_request_sender(),
             )
             self._grid = get_grid()
@@ -273,7 +257,6 @@ def new_ev_charger_pool(
         if ref_store_key not in self._ev_charger_pool_reference_stores:
             self._ev_charger_pool_reference_stores[ref_store_key] = (
                 EVChargerPoolReferenceStore(
-                    channel_registry=self._channel_registry,
                     resampler_subscription_sender=self._resampling_request_sender(),
                     status_receiver=self._ev_power_wrapper.status_channel.new_receiver(
                         limit=1
@@ -346,7 +329,6 @@ def new_pv_pool(
 
         if ref_store_key not in self._pv_pool_reference_stores:
             self._pv_pool_reference_stores[ref_store_key] = PVPoolReferenceStore(
-                channel_registry=self._channel_registry,
                 resampler_subscription_sender=self._resampling_request_sender(),
                 status_receiver=(
                     self._pv_power_wrapper.status_channel.new_receiver(limit=1)
@@ -422,7 +404,6 @@ def new_battery_pool(
         if ref_store_key not in self._battery_pool_reference_stores:
             self._battery_pool_reference_stores[ref_store_key] = (
                 BatteryPoolReferenceStore(
-                    channel_registry=self._channel_registry,
                     resampler_subscription_sender=self._resampling_request_sender(),
                     batteries_status_receiver=(
                         self._battery_power_wrapper.status_channel.new_receiver(limit=1)
@@ -461,7 +442,6 @@ def _data_sourcing_request_sender(self) -> Sender[ComponentMetricRequest]:
             )
             actor = DataSourcingActor(
                 request_receiver=channel.new_receiver(limit=_REQUEST_RECV_BUFFER_SIZE),
-                registry=self._channel_registry,
             )
             self._data_sourcing_actor = _ActorInfo(actor, channel)
             self._data_sourcing_actor.actor.start()
@@ -482,7 +462,6 @@ def _resampling_request_sender(self) -> Sender[ComponentMetricRequest]:
                 name="Data Pipeline: Component Metric Resampling Actor Request Channel"
             )
             actor = ComponentMetricsResamplingActor(
-                channel_registry=self._channel_registry,
                 data_sourcing_request_sender=self._data_sourcing_request_sender(),
                 resampling_request_receiver=channel.new_receiver(
                     limit=_REQUEST_RECV_BUFFER_SIZE,

src/frequenz/sdk/microgrid/_data_sourcing/_component_metric_request.py

@@ -6,13 +6,18 @@
 from dataclasses import dataclass
 from datetime import datetime
 
+from frequenz.channels._broadcast import BroadcastReceiver
+from frequenz.channels._oneshot import OneshotSender
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.client.microgrid.metrics import Metric
+from frequenz.quantities import Quantity
 
 from frequenz.sdk.microgrid._old_component_data import TransitionalMetric
 
 __all__ = ["ComponentMetricRequest", "Metric"]
 
+from frequenz.sdk.timeseries import Sample
+
 
 @dataclass
 class ComponentMetricRequest:
@@ -51,6 +56,9 @@ class ComponentMetricRequest:
     If None, only live data is streamed.
     """
 
+    telem_stream_sender: OneshotSender[BroadcastReceiver[Sample[Quantity]]]
+    """Sender of a oneshot channel used to send the data stream back to the requester."""
+
     def get_channel_name(self) -> str:
         """Construct the channel name based on the request parameters.