Merge pull request #105 from vkottler/dev/2.6.3

2.6.3 - Refactor periodic task metrics

.github/workflows/python-package.yml

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

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.3
-    hash=a3111c8378903228c0a4d8a5626629ec
+    hash=50d5b938880799c5ddb535fc10d86e73
     =====================================
 -->
 
-# runtimepy ([2.6.2](https://pypi.org/project/runtimepy/))
+# runtimepy ([2.6.3](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: 2
 minor: 6
-patch: 2
+patch: 3
 entry: runtimepy

pyproject.toml

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

runtimepy/__init__.py

@@ -1,7 +1,7 @@
 # =====================================
 # generator=datazen
 # version=3.1.3
-# hash=f1035fea70e8a5cff741d00f44d92cb5
+# hash=61810068c8810947c395b39e8217ee1f
 # =====================================
 
 """
@@ -10,4 +10,4 @@
 
 DESCRIPTION = "A framework for implementing Python services."
 PKG_NAME = "runtimepy"
-VERSION = "2.6.2"
+VERSION = "2.6.3"

runtimepy/metrics/task.py

@@ -3,9 +3,15 @@
 """
 
 # built-in
-from typing import NamedTuple
+from asyncio import AbstractEventLoop
+from contextlib import contextmanager
+from typing import Iterator, NamedTuple
+
+# third-party
+from vcorelib.math import MovingAverage, RateTracker
 
 # internal
+from runtimepy.primitives import Double as _Double
 from runtimepy.primitives import Float as _Float
 from runtimepy.primitives import Uint32 as _Uint32
 
@@ -26,3 +32,26 @@ def create() -> "PeriodicTaskMetrics":
         return PeriodicTaskMetrics(
             _Uint32(), _Float(), _Float(), _Float(), _Float()
         )
+
+    @contextmanager
+    def measure(
+        self,
+        eloop: AbstractEventLoop,
+        rate: RateTracker,
+        dispatch: MovingAverage,
+        iter_time: _Double,
+    ) -> Iterator[None]:
+        """Measure the time spent yielding and update data."""
+
+        start = eloop.time()
+        self.rate_hz.raw.value = rate(int(start * 1e9))
+
+        yield
+
+        iter_time.value = eloop.time() - start
+
+        # Update runtime metrics.
+        self.dispatches.raw.value += 1
+        self.average_s.raw.value = dispatch(iter_time.value)
+        self.max_s.raw.value = dispatch.max
+        self.min_s.raw.value = dispatch.min

runtimepy/task/basic/periodic.py

@@ -24,6 +24,7 @@
 from runtimepy.metrics import PeriodicTaskMetrics
 from runtimepy.mixins.environment import ChannelEnvironmentMixin
 from runtimepy.primitives import Bool as _Bool
+from runtimepy.primitives import Double as _Double
 from runtimepy.primitives import Float as _Float
 
 
@@ -112,34 +113,26 @@ async def run(
         )
 
         eloop = _asyncio.get_running_loop()
+        iter_time = _Double()
 
         while self._enabled:
-            start = eloop.time()
-
-            # Keep track of the rate that this task is running at.
-            self.metrics.rate_hz.raw.value = self._dispatch_rate(
-                int(start * 1e9)
-            )
-
-            self._enabled.raw.value = await _asyncio.shield(self.dispatch())
-            iter_time = eloop.time() - start
-
-            # Update runtime metrics.
-            self.metrics.dispatches.raw.value += 1
-            self.metrics.average_s.raw.value = self._dispatch_time(iter_time)
-            self.metrics.max_s.raw.value = self._dispatch_time.max
-            self.metrics.min_s.raw.value = self._dispatch_time.min
+            with self.metrics.measure(
+                eloop, self._dispatch_rate, self._dispatch_time, iter_time
+            ):
+                self._enabled.raw.value = await _asyncio.shield(
+                    self.dispatch()
+                )
 
             # Check this synchronously. This may not be suitable for tasks
             # with long periods.
             if stop_sig is not None:
                 self._enabled.raw.value = not stop_sig.is_set()
 
-            sleep_s = self.period_s.value - iter_time
-
             if self._enabled:
                 try:
-                    await _asyncio.sleep(max(sleep_s, 0))
+                    await _asyncio.sleep(
+                        max(self.period_s.value - iter_time.value, 0)
+                    )
                 except _asyncio.CancelledError:
                     self.logger.info("Task was cancelled.")
                     self.disable()