Reuse DOSCTRL RPC connections to prevent socket exhaustion

README.md

@@ -234,3 +234,12 @@ The implementation exposes basic interrupt status and loopback registers;
 real interrupt-driven serial applications need their own compatibility tests.
 Register reference: [National Semiconductor UART databook](https://bitsavers.org/components/national/_dataBooks/1987_National_Microcommunications_Elements_Databook_UARTs_NSC800_Family.pdf).
 SYMDEB reference: [Microsoft MS-DOS Encyclopedia](https://msarchive.pcjs.org/mspl13/msdos/encyclopedia/section4/).
+
+DOSCTRL mailbox RPC reuses one JSON-lines TCP connection per controller instead
+of opening one per poll. This avoids Windows ephemeral-port exhaustion during
+long compiler/file-transfer sessions. Call `RPC.close()` when finished; the CLI
+and real-DOS session do this automatically. A timeout, malformed reply or wrong
+request ID closes the channel and raises without replaying the request. Keep the
+existing pending-command recovery rules: do not resubmit an uncertain EXEC/write.
+Connection-count evidence: `docs/dosctrl_persistent_rpc_20261002.json` (500 polls:
+500 connections before, one after); deadline/uncertain-write controls remain.

docs/HANDOFF.md

@@ -1,3 +1,14 @@
+# Handoff - DOSCTRL persistent RPC, 2026-10-02
+
+FINISHED local controller fix: reuse one JSON-lines TCP channel for mailbox
+polling; close it on uncertainty without replaying a request. CLI and dos_session
+close explicitly. A real shared-main TP6 batch failed with Windows socket10048
+after12 compiles. Independent 500-poll probe measures500 connections before
+and1 after; docs/dosctrl_persistent_rpc_20261002.json.52 focused controller
+tests pass, including deadline and no-replay controls.
+WIP public CI/PR integration and a fresh complete parent TP6 batch. No game
+semantics/CPU/device change. Earlier compiler products remain a failed checkpoint.
+
 # Handoff - public input export refreshed, 2026-09-30
 
 FINISHED: PR63 exports the frozen PyPC execution revision ec610f3 and final

docs/dosctrl_persistent_rpc_20261002.json

@@ -0,0 +1,10 @@
+{
+  "before": {
+    "polls": 500,
+    "connections": 500
+  },
+  "after": {
+    "polls": 500,
+    "connections": 1
+  }
+}

guest/dos_control.py

@@ -127,6 +127,18 @@ def __init__(self, port: int):
         self.port = port
         self.sequence = 0
         self.timeout = 20.0
+        self._socket = None
+        self._stream = None
+
+    def close(self) -> None:
+        stream, sock = self._stream, self._socket
+        self._stream = self._socket = None
+        try:
+            if stream is not None:
+                stream.close()
+        finally:
+            if sock is not None:
+                sock.close()
 
     def call(self, method: str, params: dict | None = None, *,
              deadline: float | None = None) -> dict:
@@ -144,21 +156,29 @@ def remaining_timeout() -> float:
         request_id = self.sequence
         request = {'jsonrpc': '2.0', 'id': request_id,
                    'method': method, 'params': params or {}}
-        with socket.create_connection(('127.0.0.1', self.port),
-                                      remaining_timeout()) as sock:
-            sock.settimeout(remaining_timeout())
-            sock.sendall(json.dumps(request).encode('utf-8') + b'\n')
-            sock.settimeout(remaining_timeout())
-            with sock.makefile('rb') as stream:
-                line = stream.readline(16 * 1024 * 1024 + 1)
-        if not line.endswith(b'\n') or len(line) > 16 * 1024 * 1024:
-            raise RuntimeError('incomplete or oversized PyPC reply')
-        reply = json.loads(line)
-        if reply.get('id') != request_id:
-            raise RuntimeError('PyPC reply id mismatch')
-        if 'error' in reply:
-            raise RuntimeError(f'{method}: {reply["error"]}')
-        return reply['result']
+        # The JSON-lines server accepts multiple requests per connection.
+        # Mailbox polling must not consume a Windows ephemeral port per RPC.
+        # On any uncertain response, discard the channel; never replay a write.
+        try:
+            if self._socket is None:
+                self._socket = socket.create_connection(
+                    ('127.0.0.1', self.port), remaining_timeout())
+                self._stream = self._socket.makefile('rb')
+            self._socket.settimeout(remaining_timeout())
+            self._socket.sendall(json.dumps(request).encode('utf-8') + b'\n')
+            self._socket.settimeout(remaining_timeout())
+            line = self._stream.readline(16 * 1024 * 1024 + 1)
+            if not line.endswith(b'\n') or len(line) > 16 * 1024 * 1024:
+                raise RuntimeError('incomplete or oversized PyPC reply')
+            reply = json.loads(line)
+            if reply.get('id') != request_id:
+                raise RuntimeError('PyPC reply id mismatch')
+            if 'error' in reply:
+                raise RuntimeError(f'{method}: {reply["error"]}')
+            return reply['result']
+        except Exception:
+            self.close()
+            raise
 
     def read(self, address: int, length: int, *,
              deadline: float | None = None) -> bytes:
@@ -538,8 +558,9 @@ def main() -> int:
     running.add_argument('tail', nargs='?', default='')
     running.add_argument('--output', help='DOS path for captured standard handles')
     args = parser.parse_args()
+    rpc = RPC(args.rpc_port)
     try:
-        worker = DOSControl(RPC(args.rpc_port), args.timeout)
+        worker = DOSControl(rpc, args.timeout)
         if args.command == 'ready':
             result = {'ready': worker.ready()}
         elif args.command == 'list':
@@ -572,6 +593,8 @@ def main() -> int:
     except (DOSCommandError, DOSClientInputError, TimeoutError,
             RuntimeError, OSError, ValueError) as exc:
         return _print_cli_failure(exc)
+    finally:
+        rpc.close()
     print(json.dumps(result, indent=2))
     if args.command in ('exec', 'collect-exec'):
         # DOS termination type is independent of AL: never report abnormal

guest/dos_session.py

@@ -253,6 +253,7 @@ def at_prompt():
                 except subprocess.TimeoutExpired:
                     process.kill()
                     process.wait(timeout=5)
+            live.rpc.close()
             live.save()
             assert digest(ROOT / 'harddisk.img') == boot_hash, 'tracked boot disk was modified'
             print('LIVE_DOS_REPORT=' + json.dumps(report, sort_keys=True))

tests/test_dos_control_cli.py

@@ -59,14 +59,18 @@ def dispatch(self, method, params):
 
 class WorkerHandler(socketserver.StreamRequestHandler):
     def handle(self):
-        request = json.loads(self.rfile.readline())
-        try:
-            result = self.server.dispatch(request['method'], request['params'])
-            reply = {'jsonrpc': '2.0', 'id': request['id'], 'result': result}
-        except Exception as exc:
-            reply = {'jsonrpc': '2.0', 'id': request['id'],
-                     'error': {'code': -32603, 'message': str(exc)}}
-        self.wfile.write(json.dumps(reply).encode('utf-8') + b'\n')
+        while True:
+            line = self.rfile.readline()
+            if not line:
+                return
+            request = json.loads(line)
+            try:
+                result = self.server.dispatch(request['method'], request['params'])
+                reply = {'jsonrpc': '2.0', 'id': request['id'], 'result': result}
+            except Exception as exc:
+                reply = {'jsonrpc': '2.0', 'id': request['id'],
+                         'error': {'code': -32603, 'message': str(exc)}}
+            self.wfile.write(json.dumps(reply).encode('utf-8') + b'\n')
 
 
 @pytest.mark.parametrize(('exit_code', 'termination_type', 'host_code'), [

tests/test_dos_control_cli_errors.py

@@ -57,14 +57,18 @@ def dispatch(self, method, params):
 
 class ErrorHandler(socketserver.StreamRequestHandler):
     def handle(self):
-        request = json.loads(self.rfile.readline())
-        try:
-            result = self.server.dispatch(request['method'], request['params'])
-            reply = {'jsonrpc': '2.0', 'id': request['id'], 'result': result}
-        except Exception as exc:
-            reply = {'jsonrpc': '2.0', 'id': request['id'],
-                     'error': {'code': -32603, 'message': str(exc)}}
-        self.wfile.write(json.dumps(reply).encode('utf-8') + b'\n')
+        while True:
+            line = self.rfile.readline()
+            if not line:
+                return
+            request = json.loads(line)
+            try:
+                result = self.server.dispatch(request['method'], request['params'])
+                reply = {'jsonrpc': '2.0', 'id': request['id'], 'result': result}
+            except Exception as exc:
+                reply = {'jsonrpc': '2.0', 'id': request['id'],
+                         'error': {'code': -32603, 'message': str(exc)}}
+            self.wfile.write(json.dumps(reply).encode('utf-8') + b'\n')
 
 
 def run_with_peer(peer, *arguments):

tests/test_dos_control_rpc_connection.py

@@ -0,0 +1,92 @@
+"""Mailbox polling reuses a channel; uncertain writes are never replayed."""
+import json
+import socketserver
+import threading
+import time
+
+import pytest
+from guest.dos_control import RPC
+
+
+class Peer(socketserver.ThreadingTCPServer):
+    allow_reuse_address = True
+    daemon_threads = True
+
+    def __init__(self, bad_reply=None):
+        super().__init__(('127.0.0.1', 0), Handler)
+        self.connections = 0
+        self.requests = []
+        self.bad_reply = bad_reply
+        self.lock = threading.Lock()
+
+
+class Handler(socketserver.StreamRequestHandler):
+    def handle(self):
+        with self.server.lock:
+            self.server.connections += 1
+        while line := self.rfile.readline():
+            request = json.loads(line)
+            with self.server.lock:
+                self.server.requests.append(request)
+                number = len(self.server.requests)
+            if number == 1 and self.server.bad_reply == 'timeout':
+                time.sleep(0.15)  # Write was applied, but the reply is late.
+            reply = {'jsonrpc': '2.0', 'id': request['id'], 'result': {'calls': number}}
+            if number == 1 and self.server.bad_reply == 'id':
+                reply['id'] += 100
+            try:
+                self.wfile.write(json.dumps(reply).encode() + b'\n')
+            except (BrokenPipeError, ConnectionResetError):
+                return
+
+
+@pytest.fixture
+def peer_factory():
+    peers = []
+
+    def create(bad_reply=None):
+        peer = Peer(bad_reply)
+        thread = threading.Thread(target=peer.serve_forever, daemon=True)
+        thread.start()
+        peers.append((peer, thread))
+        return peer
+
+    yield create
+    for peer, thread in peers:
+        peer.shutdown()
+        peer.server_close()
+        thread.join(timeout=1)
+
+
+def test_many_polls_use_one_connection_and_close_is_idempotent(peer_factory):
+    peer = peer_factory()
+    rpc = RPC(peer.server_address[1])
+    try:
+        for number in range(1, 501):
+            assert rpc.call('memory.read', {'address': 0xD8000}) == {'calls': number}
+        assert peer.connections == 1
+        assert [r['id'] for r in peer.requests] == list(range(1, 501))
+        rpc.close()
+        rpc.close()
+        assert rpc.call('memory.read') == {'calls': 501}
+        assert peer.connections == 2
+    finally:
+        rpc.close()
+
+
+@pytest.mark.parametrize('failure', ['id', 'timeout'])
+def test_uncertain_write_closes_channel_without_replay(peer_factory, failure):
+    peer = peer_factory(failure)
+    rpc = RPC(peer.server_address[1])
+    try:
+        expected = TimeoutError if failure == 'timeout' else RuntimeError
+        with pytest.raises(expected):
+            rpc.call('memory.write', {'address': 0xD8000},
+                     deadline=time.monotonic() + (0.03 if failure == 'timeout' else 2))
+        assert rpc._socket is None and rpc._stream is None
+        assert [r['method'] for r in peer.requests] == ['memory.write']
+        assert rpc.call('memory.read') == {'calls': 2}
+        assert peer.connections == 2
+        assert [r['method'] for r in peer.requests] == ['memory.write', 'memory.read']
+    finally:
+        rpc.close()

tests/test_dos_control_rpc_deadline.py

@@ -68,6 +68,9 @@ def __enter__(self):
         def __exit__(self, *exc_info):
             return False
 
+        def close(self):
+            pass
+
         def readline(self, limit):
             assert socket_timeouts[-1] == pytest.approx(0.01)
             clock[0] += 0.03
@@ -80,6 +83,9 @@ def __enter__(self):
         def __exit__(self, *exc_info):
             return False
 
+        def close(self):
+            pass
+
         def settimeout(self, timeout):
             socket_timeouts.append(timeout)