Merge pull request #237 from vkottler/dev/tftp

Initial tftp scaffolding

.github/workflows/python-package.yml

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

.gitignore

@@ -19,3 +19,4 @@ compile_commands.json
 src
 *.webm
 *.log
+tmp

.pylintrc

@@ -1,6 +1,6 @@
 [DESIGN]
 max-args=9
-max-attributes=14
+max-attributes=15
 max-parents=13
 max-branches=13
 

README.md

@@ -2,11 +2,11 @@
     =====================================
     generator=datazen
     version=3.1.4
-    hash=a98449cc670c0d11f756d6044f1bd45f
+    hash=bc8310897ee0818dfb82ec8021aefc60
     =====================================
 -->
 
-# runtimepy ([5.0.1](https://pypi.org/project/runtimepy/))
+# runtimepy ([5.1.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/arbiter/tftp.sh

@@ -0,0 +1,63 @@
+#!/bin/bash
+
+set -e
+
+REPO=$(git rev-parse --show-toplevel)
+REL=local/arbiter
+CWD=$REPO/$REL
+TMP=$CWD/tmp
+
+PORT=8001
+
+tftp_cmd() {
+	rlwrap tftp -m octet localhost $PORT "$@"
+}
+
+test_get_file() {
+	for FILE in LICENSE README.md tags; do
+		if [ -f "$REPO/$FILE" ]; then
+			tftp_cmd -c get "$FILE"
+			diff "$REPO/$FILE" "$TMP/$FILE"
+		fi
+	done
+}
+
+test_large_file() {
+	fallocate -l 30M "$REPO/dummy.bin"
+	tftp_cmd -c get dummy.bin
+	diff "$REPO/dummy.bin" "$TMP/dummy.bin"
+	rm "$REPO/dummy.bin"
+}
+
+clear_tmp() {
+	rm -f "$TMP/*"
+}
+
+test_write_files() {
+	for FILE in LICENSE README.md tags; do
+		tftp_cmd -c put "$REPO/$FILE" $REL/tmp/$FILE
+		sleep 0.25
+		diff "$REPO/$FILE" "$TMP/$FILE"
+	done
+}
+
+mkdir -p "$TMP"
+pushd "$TMP" >/dev/null || exit
+set -x
+
+# Test that we can retrieve files.
+test_get_file
+test_large_file
+
+# Clear directory.
+clear_tmp
+
+# Test that we can write files.
+test_write_files
+
+set +x
+popd >/dev/null || exit
+
+# rm -rf "$TMP"
+
+echo "Success."

local/configs/package.yaml

@@ -5,7 +5,7 @@ description: A framework for implementing Python services.
 entry: {{entry}}
 
 requirements:
-  - vcorelib>=3.2.8
+  - vcorelib>=3.3.1
   - svgen>=0.6.7
   - websockets
   - psutil

local/variables/package.yaml

@@ -1,5 +1,5 @@
 ---
 major: 5
-minor: 0
-patch: 1
+minor: 1
+patch: 0
 entry: runtimepy

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta:__legacy__"
 
 [project]
 name = "runtimepy"
-version = "5.0.1"
+version = "5.1.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=2c34399f27189207cd0cbd855a086ead
+# hash=c57910000e21ff16644bf037b474eeb4
 # =====================================
 
 """
@@ -10,7 +10,7 @@
 
 DESCRIPTION = "A framework for implementing Python services."
 PKG_NAME = "runtimepy"
-VERSION = "5.0.1"
+VERSION = "5.1.0"
 
 # runtimepy-specific content.
 METRICS_NAME = "metrics"

runtimepy/commands/common.py

@@ -5,7 +5,7 @@
 # built-in
 from argparse import ArgumentParser as _ArgumentParser
 from argparse import Namespace as _Namespace
-from contextlib import contextmanager
+from contextlib import contextmanager, suppress
 from typing import Any, Iterator
 
 # third-party
@@ -16,10 +16,9 @@
 # internal
 from runtimepy import DEFAULT_EXT, PKG_NAME
 
-try:
-    import curses as _curses
-except ModuleNotFoundError:  # pragma: nocover
-    _curses = {}  # type: ignore
+_curses = {}  # type: ignore
+with suppress(ModuleNotFoundError):
+    import curses as _curses  # type: ignore
 
 FACTORIES = f"package://{PKG_NAME}/factories.{DEFAULT_EXT}"
 

runtimepy/data/factories.yaml

@@ -20,6 +20,7 @@ factories:
     namespaces: [websocket, "null"]
 
   # Useful protocols.
+  - {name: runtimepy.net.factories.Tftp}
   - {name: runtimepy.net.factories.Http}
   - {name: runtimepy.net.factories.RuntimepyHttp}
   - {name: runtimepy.net.factories.RuntimepyWebsocketJson}

runtimepy/data/js/classes/OverlayManager.js

@@ -62,7 +62,7 @@ class OverlayManager {
       /* Show amount of time captured. */
       if (this.minTimestamp != null && this.maxTimestamp) {
         let nanos = nanosString(this.maxTimestamp - this.minTimestamp);
-        this.writeLn(nanos[0] + nanos[1] + "s (y-axis     )");
+        this.writeLn(nanos[0] + nanos[1] + "s (x-axis     )");
       }
 
       this.writeLn(String(this.bufferDepth) + "       (max samples)");

runtimepy/data/tftp_server.yaml

@@ -0,0 +1,12 @@
+---
+includes:
+  - package://runtimepy/factories.yaml
+
+ports:
+  - {name: tftp_server, type: udp}
+
+clients:
+  - factory: tftp
+    name: tftp_server
+    kwargs:
+      local_addr: [localhost, "$tftp_server"]

runtimepy/enum/__init__.py

@@ -11,6 +11,7 @@
 # third-party
 from vcorelib.io.types import JsonObject as _JsonObject
 from vcorelib.io.types import JsonValue as _JsonValue
+from vcorelib.python import StrToBool
 
 # internal
 from runtimepy.enum.types import EnumType as _EnumType
@@ -19,7 +20,6 @@
 from runtimepy.registry.bool import BooleanRegistry as _BooleanRegistry
 from runtimepy.registry.item import RegistryItem as _RegistryItem
 from runtimepy.registry.name import NameRegistry as _NameRegistry
-from runtimepy.util import StrToBool
 
 
 class RuntimeEnum(_RegistryItem):

runtimepy/metrics/task.py

@@ -3,12 +3,17 @@
 """
 
 # built-in
-from asyncio import AbstractEventLoop
 from contextlib import contextmanager
 from typing import Iterator, NamedTuple
 
 # third-party
-from vcorelib.math import MovingAverage, RateTracker, to_nanos
+from vcorelib.math import (
+    MovingAverage,
+    RateTracker,
+    from_nanos,
+    metrics_time_ns,
+)
+from vcorelib.math.keeper import TimeSource
 
 # internal
 from runtimepy.primitives import Double as _Double
@@ -28,30 +33,36 @@ class PeriodicTaskMetrics(NamedTuple):
     overruns: _Uint16
 
     @staticmethod
-    def create() -> "PeriodicTaskMetrics":
+    def create(
+        time_source: TimeSource = metrics_time_ns,
+    ) -> "PeriodicTaskMetrics":
         """Create a new metrics instance."""
 
         return PeriodicTaskMetrics(
-            _Uint32(), _Float(), _Float(), _Float(), _Float(), _Uint16()
+            _Uint32(time_source=time_source),
+            _Float(time_source=time_source),
+            _Float(time_source=time_source),
+            _Float(time_source=time_source),
+            _Float(time_source=time_source),
+            _Uint16(time_source=time_source),
         )
 
     @contextmanager
     def measure(
         self,
-        eloop: AbstractEventLoop,
         rate: RateTracker,
         dispatch: MovingAverage,
         iter_time: _Double,
         period_s: float,
     ) -> Iterator[None]:
         """Measure the time spent yielding and update data."""
 
-        start = eloop.time()
-        self.rate_hz.value = rate(to_nanos(start))
+        start = metrics_time_ns()
+        self.rate_hz.value = rate(start)
 
         yield
 
-        iter_time.value = eloop.time() - start
+        iter_time.value = from_nanos(metrics_time_ns() - start)
 
         # Update runtime metrics.
         self.dispatches.value += 1

runtimepy/net/arbiter/housekeeping/__init__.py

@@ -5,6 +5,7 @@
 
 # built-in
 import asyncio
+from typing import Awaitable
 
 # internal
 from runtimepy.mixins.async_command import AsyncCommandProcessingMixin
@@ -50,12 +51,16 @@ async def dispatch(self) -> bool:
             self.manager.poll_metrics()
 
         # Handle any incoming commands.
-        processors = []
+        processors: list[Awaitable[None]] = []
         for mapping in self.app.connections.values(), self.app.tasks.values():
             for item in mapping:
                 if isinstance(item, AsyncCommandProcessingMixin):
                     processors.append(item.process_command_queue())
 
+        # Service connection tasks. The connection manager should probably do
+        # this on its own at some point.
+        processors += list(self.app.conn_manager.connection_tasks)
+
         if processors:
             await asyncio.gather(*processors)
 

runtimepy/net/connection.py

@@ -5,7 +5,9 @@
 # built-in
 from abc import ABC as _ABC
 import asyncio as _asyncio
-from contextlib import suppress as _suppress
+from contextlib import asynccontextmanager, suppress
+from typing import AsyncIterator
+from typing import Iterator as _Iterator
 from typing import Optional as _Optional
 from typing import Union as _Union
 
@@ -61,7 +63,12 @@ def __init__(
         self._binary_messages: _asyncio.Queue[BinaryMessage] = _asyncio.Queue()
         self.tx_binary_hwm: int = 0
 
+        # Tasks common to connection processing.
         self._tasks: list[_asyncio.Task[None]] = []
+
+        # Connection-specific tasks.
+        self._conn_tasks: list[_asyncio.Task[None]] = []
+
         self.initialized = _asyncio.Event()
         self.exited = _asyncio.Event()
 
@@ -190,7 +197,7 @@ def disable(self, reason: str) -> None:
             self.disable_extra()
 
             # Cancel tasks.
-            for task in self._tasks:
+            for task in self._tasks + list(self.tasks):
                 if not task.done():
                     task.cancel()
 
@@ -249,6 +256,17 @@ async def _handle_restart(
 
             self._restart_attempts.raw.value += 1
 
+    @asynccontextmanager
+    async def process_then_disable(self, **kwargs) -> AsyncIterator[None]:
+        """Process this connection, then disable and wait for completion."""
+
+        task = _asyncio.create_task(self.process(**kwargs))
+        try:
+            yield
+        finally:
+            self.disable("nominal")
+            await task
+
     async def process(
         self,
         stop_sig: _asyncio.Event = None,
@@ -304,7 +322,7 @@ async def process(
     async def _process_read(self) -> None:
         """Process incoming messages while this connection is active."""
 
-        with _suppress(KeyboardInterrupt):
+        with suppress(KeyboardInterrupt):
             while self._enabled:
                 # Attempt to get the next message.
                 message = await self._await_message()
@@ -349,6 +367,22 @@ async def _process_write_binary(self) -> None:
                 await self._send_binay_message(data)
                 queue.task_done()
 
+    @property
+    def tasks(self) -> _Iterator[_asyncio.Task[None]]:
+        """
+        Get active connection tasks. Instance uses this opportunity to release
+        references to any completed tasks.
+        """
+
+        active = []
+
+        for task in self._conn_tasks:
+            if not task.done():
+                active.append(task)
+                yield task
+
+        self._conn_tasks = active
+
 
 class EchoConnection(Connection):
     """A connection that just echoes what it was sent."""