Initial idea for global channel environment

.github/workflows/python-package.yml

@@ -68,7 +68,7 @@ jobs:
 
       - run: |
           mk python-release owner=vkottler \
-                            repo=runtimepy version=3.7.6
+                            repo=runtimepy version=4.0.0
         if: |
           matrix.python-version == '3.11'
           && matrix.system == 'ubuntu-latest'

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.4
-    hash=5e658f86474b5fde932a9ad02e57bb7b
+    hash=0eae1fb0651f9608411daf093deabfbc
     =====================================
 -->
 
-# runtimepy ([3.7.6](https://pypi.org/project/runtimepy/))
+# runtimepy ([4.0.0](https://pypi.org/project/runtimepy/))
 
 [![python](https://img.shields.io/pypi/pyversions/runtimepy.svg)](https://pypi.org/project/runtimepy/)
 ![Build Status](https://github.com/vkottler/runtimepy/workflows/Python%20Package/badge.svg)

local/variables/package.yaml

@@ -1,5 +1,5 @@
 ---
-major: 3
-minor: 7
-patch: 6
+major: 4
+minor: 0
+patch: 0
 entry: runtimepy

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta:__legacy__"
 
 [project]
 name = "runtimepy"
-version = "3.7.6"
+version = "4.0.0"
 description = "A framework for implementing Python services."
 readme = "README.md"
 requires-python = ">=3.11"

runtimepy/__init__.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.4
-# hash=f807afa762f8662fce8e0c03e04b39d8
+# hash=372c7409fed9799f8307142ddbc1c739
 # =====================================
 
 """
@@ -10,7 +10,7 @@
 
 DESCRIPTION = "A framework for implementing Python services."
 PKG_NAME = "runtimepy"
-VERSION = "3.7.6"
+VERSION = "4.0.0"
 
 # runtimepy-specific content.
 METRICS_NAME = "metrics"

runtimepy/channel/environment/__init__.py

@@ -2,6 +2,9 @@
 A module implementing a channel environment.
 """
 
+# built-in
+from typing import Iterator as _Iterator
+
 # internal
 from runtimepy.channel.environment.array import (
     ArrayChannelEnvironment as _ArrayChannelEnvironment,
@@ -12,15 +15,16 @@
 from runtimepy.channel.environment.file import (
     FileChannelEnvironment as _FileChannelEnvironment,
 )
-from runtimepy.channel.environment.names import (
-    ChannelNameEnvironment as _ChannelNameEnvironment,
-)
 
 
 class ChannelEnvironment(
     _ArrayChannelEnvironment,
     _FileChannelEnvironment,
     _CreateChannelEnvironment,
-    _ChannelNameEnvironment,
 ):
     """A class integrating channel and enumeration registries."""
+
+    @property
+    def names(self) -> _Iterator[str]:
+        """Iterate over registered names in the environment."""
+        yield from self.channels.names.names

runtimepy/channel/registry.py

@@ -4,6 +4,7 @@
 
 # built-in
 from typing import Any as _Any
+from typing import NamedTuple
 from typing import Optional as _Optional
 from typing import Type as _Type
 from typing import Union
@@ -29,6 +30,46 @@ class ChannelNameRegistry(_NameRegistry):
     name_regex = _CHANNEL_PATTERN
 
 
+class ChannelCreation(NamedTuple):
+    """A container for channel-creation parameters."""
+
+    name: str
+    kind: Union[Primitive[_Any], _Primitivelike]
+    commandable: bool = False
+    enum: _Optional[_RegistryKey] = None
+    scaling: _Optional[ChannelScaling] = None
+
+
+class GlobalEnvironment:
+    """A global environment management interface."""
+
+    registry: "ChannelRegistry"
+
+    def __init__(self) -> None:
+        """Initialize this instance."""
+
+        self.duplicates: list[tuple[_AnyChannel, ChannelCreation]] = []
+
+        # We should add a 'num_duplicates' channel, maybe 'num_channels' as
+        # well?
+
+    def handle(self, channel: _AnyChannel, meta: ChannelCreation) -> None:
+        """Handle global channel registration (used for instrumentation)."""
+
+        if not self.registry.channel(
+            meta.name,
+            meta.kind,
+            commandable=meta.commandable,
+            enum=meta.enum,
+            scaling=meta.scaling,
+        ):
+            self.duplicates.append((channel, meta))
+
+
+# Keep track of global channels.
+GLOBAL: GlobalEnvironment = GlobalEnvironment()
+
+
 class ChannelRegistry(_Registry[_Channel[_Any]]):
     """A runtime enumeration registry."""
 
@@ -77,4 +118,18 @@ def channel(
         if result is not None:
             result.raw = primitive
 
+            GLOBAL.handle(
+                result,
+                ChannelCreation(
+                    name,
+                    kind,
+                    commandable=commandable,
+                    enum=enum,
+                    scaling=scaling,
+                ),
+            )
+
         return result
+
+
+GLOBAL.registry = ChannelRegistry.create()

runtimepy/net/arbiter/base.py

@@ -182,6 +182,7 @@ async def _entry(
         """
 
         result = -1
+        info: Optional[AppInfo] = None
 
         try:
             # Wait for servers to start.
@@ -211,8 +212,6 @@ async def _entry(
             for name, conn in self._connections.items():
                 register_env(name, conn.command)
 
-            info: Optional[AppInfo] = None
-
             # Run application, but only if all the registered connections are
             # still alive after initialization.
             if not check_connections or not any(

runtimepy/net/manager.py

@@ -14,6 +14,7 @@
 
 # third-party
 from vcorelib.asyncio import log_exceptions as _log_exceptions
+from vcorelib.logging import LoggerMixin
 from vcorelib.math import default_time_ns as _default_time_ns
 
 # internal
@@ -22,11 +23,13 @@
 T = _TypeVar("T", bound=_Connection)
 
 
-class ConnectionManager:
+class ConnectionManager(LoggerMixin):
     """A class for managing connection processing at runtime."""
 
     def __init__(self) -> None:
         """Initialize this connection manager."""
+
+        super().__init__()
         self.queue: _asyncio.Queue[_Connection] = _asyncio.Queue()
         self._running = False
         self._conns: _List[_Connection] = []
@@ -113,10 +116,11 @@ async def manage(self, stop_sig: _asyncio.Event) -> None:
 
             tasks = next_tasks
 
-        # Allow existing tasks to clean up.
-        if new_conn_task is not None:
-            new_conn_task.cancel()
-        for task in tasks:
-            await task
+        with self.log_time("Shutting down", reminder=True):
+            # Allow existing tasks to clean up.
+            if new_conn_task is not None:
+                new_conn_task.cancel()
+            for task in tasks:
+                await task
 
         self._running = False