Martino FerrariandClaude Opus 4.6
50d85363a3
Add TCPStreamer/TCPStreamerClient (TCP transport behind the UDPSServerI/UDPSClientI seam), an optional single in-process TCPStreamService serving many streams over worker threads, multiplexed SUBSCRIBE (comma list, stream index in fragmentIdx), and TCPStreamJoin, which merges several streams into one sample-and-hold stream paced by a master. StreamHub takes a comma list of streams over one connection. Includes GTests, E2E scenarios s58-s73 and docs. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
115 lines
12 KiB
Markdown
115 lines
12 KiB
Markdown
# TCPStreamer / TCPStreamerClient Implementation Plan
|
|
|
|
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
|
|
|
**Goal:** Add `TCPStreamer` (output) and `TCPStreamerClient` (input) MARTe2 DataSources that carry the existing UDPS framing over TCP, reusing `UDPStreamer`/`UDPStreamerClient` by making their transport pluggable.
|
|
|
|
**Architecture:** Two abstract interfaces (`UDPSServerI`, `UDPSClientI`) wrap the seven server calls and four client calls the datasources already use. The base datasources own the transport through a heap pointer that a subclass can swap (`SetServer`/`SetClient`) and expose two virtual hooks (`DefaultMaxPayloadSize`, `InitialiseTransport`). `TCPStreamer : UDPStreamer` and `TCPStreamerClient : UDPStreamerClient` plug in the new `TCPSServer` / `TCPSClient`.
|
|
|
|
**Tech Stack:** C++ (MARTe2 conventions, no STL in `Source/`), POSIX sockets, GTest (STL allowed in tests), MARTe2 make system.
|
|
|
|
Spec: `docs/superpowers/specs/2026-11-05-tcp-streamer-design.md`.
|
|
|
|
## Global Constraints
|
|
|
|
- No STL, no OS mutexes in `Source/Components/` (spec: "No STL, no OS mutexes; allocation only in `Start()`" — relaxed to "never on the RT thread"; slots grow lazily on the sender thread).
|
|
- Wire format unchanged: UDPS 17-byte header + CONFIG/DATA payload, `totalFragments` always 1.
|
|
- RT path (`Synchronise()`) must stay "spinlock + memcpy".
|
|
- Config keys: `Port` (44500), `MaxPayloadSize` (60000), `MaxClients` (16), `ClientQueueDepth` (8), `ClientStallTimeout` (5 s, 0 disables); client: `ServerAddress`, `Port`, `SilenceTimeout`, `ReconnectDelay`, `CPUMask`, `StackSize`.
|
|
- `TCPStreamer` rejects `MulticastGroup`, `Interface`, `DataPort`; `TCPStreamerClient` rejects the same.
|
|
- Do not `git add` unrelated working-tree changes (`Client/debugger/go.*`, `run_*.sh`, `Common/Client/c/Makefile`, ...). Add files by name. Commit only when the user asks.
|
|
- Build: `source env.sh`, then `make -f Makefile.gcc core test`. Run from repo root. `make clean` first in Task 1 (class layouts change; `Interfaces/UDPStream` has no depends file).
|
|
|
|
## File Structure
|
|
|
|
| File | Responsibility |
|
|
|---|---|
|
|
| `Source/Components/Interfaces/UDPStream/UDPSTransportI.h` (new) | `UDPSServerI`, `UDPSClientI` |
|
|
| `.../UDPStream/UDPSServer.h`, `UDPSClient.h` (modify) | implement the interfaces |
|
|
| `.../UDPStream/TCPSServer.{h,cpp}` (new) | multi-client non-blocking TCP server, per-client bounded queue |
|
|
| `.../UDPStream/TCPSClient.{h,cpp}` (new) | reconnecting TCP frame reader on its own thread |
|
|
| `.../UDPStream/Makefile.inc` (modify) | add `TCPSServer.x TCPSClient.x` |
|
|
| `Source/Components/DataSources/UDPStreamer/UDPStreamer.{h,cpp}` (modify) | transport via `UDPSServerI*`, hooks |
|
|
| `Source/Components/DataSources/UDPStreamerClient/UDPStreamerClient.{h,cpp}` (modify) | transport via `UDPSClientI*`, hooks |
|
|
| `Source/Components/DataSources/TCPStreamer/` (new) | `TCPStreamer`, Makefiles |
|
|
| `Source/Components/DataSources/TCPStreamerClient/` (new) | `TCPStreamerClient`, Makefiles |
|
|
| `Makefile.gcc` (modify) | build/clean the two new components |
|
|
| `Test/GTest/TCPTestUtil.h` (new) | raw-socket helpers shared by the new tests |
|
|
| `Test/GTest/TCPSServerGTest.cpp`, `TCPSClientGTest.cpp`, `TCPStreamerGTest.cpp` (new) | tests |
|
|
| `Test/GTest/Makefile.inc` (modify) | register tests, include paths |
|
|
| `Docs/TCPStreamer.md`, `CLAUDE.md`, spec (modify/new) | docs |
|
|
|
|
---
|
|
|
|
### Task 1: Pluggable transport seam (pure refactor, UDP behaviour unchanged)
|
|
|
|
**Files:** create `UDPSTransportI.h`; modify `UDPSServer.h`, `UDPSClient.h`, `UDPStreamer.{h,cpp}`, `UDPStreamerClient.{h,cpp}`.
|
|
|
|
**Interfaces:**
|
|
- Produces:
|
|
```cpp
|
|
class UDPSServerI { virtual ~UDPSServerI(){}
|
|
virtual bool Initialise(StructuredDataI&)=0; virtual bool Start()=0; virtual bool Stop()=0;
|
|
virtual void ServiceClients()=0; virtual bool SendConfig(const uint8*,uint32)=0;
|
|
virtual bool SendData(uint32 counter,const uint8*,uint32)=0;
|
|
virtual bool HasClients() const=0; virtual bool IsMulticast() const=0; virtual uint16 GetPort() const=0; };
|
|
class UDPSClientI { virtual ~UDPSClientI(){}
|
|
virtual bool Initialise(StructuredDataI&)=0; virtual void SetListener(UDPSClientListener*)=0;
|
|
virtual bool Start()=0; virtual bool Stop()=0; };
|
|
```
|
|
`UDPStreamer` protected: `virtual uint32 DefaultMaxPayloadSize() const`, `virtual bool InitialiseTransport(StructuredDataI&)`, `void SetServer(UDPSServerI*)`, `UDPSServerI *Transport() const`.
|
|
`UDPStreamerClient` protected: `virtual bool InitialiseTransport(StructuredDataI&)`, `void SetClient(UDPSClientI*)`, members `serverAddress, port, silenceTimeout, cpuMask, stackSize`.
|
|
|
|
- [ ] **Step 1: Baseline.** `make -f Makefile.gcc clean && make -f Makefile.gcc core test`, then run `./Build/x86-linux/GTest/MainGTest.ex` and record pass/fail counts.
|
|
- [ ] **Step 2:** Create `UDPSTransportI.h` (code above; forward-declare `class UDPSClientListener;`).
|
|
- [ ] **Step 3:** `UDPSServer.h`: `#include "UDPSTransportI.h"`, `class UDPSServer : public UDPSServerI`. `UDPSClient.h`: `class UDPSClient : public UDPSClientI, public EmbeddedServiceMethodBinderI`. Signatures already match; no `override` keyword (MARTe2 style).
|
|
- [ ] **Step 4:** `UDPStreamer`: replace `UDPSServer server;` by `UDPSServerI *server;` (private), constructor `server = new UDPSServer();`, destructor stops then deletes it after the executor is stopped, `server.X` -> `server->X` everywhere, `maxPayloadSize = DefaultMaxPayloadSize()` in `Initialise`, move the `serverCfg` block into `UDPStreamer::InitialiseTransport` (default impl, identical code, calling `server->Initialise`). `GetPort()` keeps returning the member.
|
|
- [ ] **Step 5:** `UDPStreamerClient`: same pattern with `UDPSClientI *client`; move the "Forward the transport parameters" block into `InitialiseTransport`; destructor `client->Stop()` when `receiverUp`, then delete.
|
|
- [ ] **Step 6:** `make clean && make core test`; run `MainGTest.ex`: counts must equal Step 1. Run `./Build/x86-linux/Test/Integration/Integration/IntegrationTests.ex` if it passed at baseline.
|
|
|
|
### Task 2: `TCPSServer`
|
|
|
|
**Files:** create `TCPSServer.{h,cpp}`, `Test/GTest/TCPTestUtil.h`, `Test/GTest/TCPSServerGTest.cpp`; modify `UDPStream/Makefile.inc`, `Test/GTest/Makefile.inc`.
|
|
|
|
**Design (see code for the authoritative version):**
|
|
- Raw POSIX sockets. Listener non-blocking, `SO_REUSEADDR`; `Port=0` binds an ephemeral port and `GetPort()` returns the bound port after `Start()`.
|
|
- Per client: `Slot[ClientQueueDepth+2]` queue (`buf, cap, len, isConfig`), `frontSent` (bytes of the front frame already written), `dataQueued`, `lastProgressTicks`.
|
|
- Queue capacity `depth+2` is exact: at most one data-in-flight/old frame, plus two CONFIG frames, plus `depth` DATA frames.
|
|
- `QueueConfig`: drop every queued frame except an in-flight one, append the CONFIG (CONFIG never dropped, never behind stale DATA).
|
|
- `QueueData`: while `dataQueued >= depth`, remove the oldest non-in-flight DATA (count a drop); if none is removable drop the new frame.
|
|
- `Flush`: `send(MSG_NOSIGNAL|MSG_DONTWAIT)` loop; `EAGAIN` stops; other errors evict.
|
|
- `ServiceClients`: accept (non-blocking), refuse beyond `MaxClients`, enqueue cached CONFIG on accept, detect peer close with `recv`, flush, evict clients with no progress for `ClientStallTimeout`, rate-limited (1/s) drop report.
|
|
- Extra key `SendBufferSize` (0 = OS default): `SO_SNDBUF` on accepted sockets, which also bounds latency and makes slow-client tests deterministic.
|
|
- `SendData` returns true unless the frame exceeds 1 MiB; per-client drops are reported through `GetClientDrops()`.
|
|
|
|
- [ ] **Step 1:** write `TCPSServerGTest.cpp` first: `Initialise` rejects (`MaxPayloadSize=0`, `ClientQueueDepth=0`, `MaxClients=0`), `ConfigBeforeData_LateJoin`, `MultipleClients_InOrder`, `ReconfigDeliveredToConnectedClient`, `MaxClientsRefused`, `PeerCloseEvicts`, `SlowClientDropsFastClientComplete` (200 x 60000-byte frames, `SendBufferSize=65536`, slow client `SO_RCVBUF=4096` never reads, fast client thread receives all counters 1..200 in order, `GetClientDrops() > 0`), `StallDisconnect` (`ClientStallTimeout=1`).
|
|
- [ ] **Step 2:** run -> compile failure (no `TCPSServer.h`).
|
|
- [ ] **Step 3:** implement `TCPSServer`; add to `OBJSX`.
|
|
- [ ] **Step 4:** `make -f Makefile.gcc core test && ./Build/x86-linux/GTest/MainGTest.ex --gtest_filter='TCPSServer*'` -> all pass.
|
|
|
|
### Task 3: `TCPSClient`
|
|
|
|
**Files:** create `TCPSClient.{h,cpp}`, `Test/GTest/TCPSClientGTest.cpp`; modify `UDPStream/Makefile.inc`, `Test/GTest/Makefile.inc` (already shared).
|
|
|
|
**Design:** `TCPSClient : UDPSClientI, EmbeddedServiceMethodBinderI`; `SingleThreadService`; `Execute` per iteration: if disconnected honour `ReconnectDelay`, then non-blocking `connect` + `poll` (bounded 500 ms) -> `OnUDPSConnected`; if connected `poll(50 ms)`, `recv` into a `HEADER+1 MiB` buffer, parse complete frames, bad magic / `payloadBytes > 1 MiB` / `totalFragments != 1` -> close + `OnUDPSDisconnected`; peer close/read error/`SilenceTimeout` the same. Keys: `ServerAddr`, `Port`, `SilenceTimeout` (float32 s, default 0 = off), `ReconnectDelay` (float32 s, default 2), `CPUMask`, `StackSize`. `Stop()` closes silently (no callbacks).
|
|
|
|
- [ ] **Step 1:** tests (`RecordingListener` with `std::mutex`): `ReceivesConfigThenData` (real `TCPSServer`), `ReconnectAfterServerRestart` (fixed port, `ReconnectDelay=0.2`), `RejectsBadMagic`, `RejectsOversizeLength`, `PartialFrameReads` (raw server writing a frame in 3 chunks), `SilenceTimeoutReconnects`, `StopIsPrompt`.
|
|
- [ ] **Step 2:** run -> compile failure. **Step 3:** implement. **Step 4:** `--gtest_filter='TCPSClient*'` passes.
|
|
|
|
### Task 4: `TCPStreamer`, `TCPStreamerClient`, build wiring
|
|
|
|
**Files:** create `Source/Components/DataSources/TCPStreamer/{TCPStreamer.h,TCPStreamer.cpp,Makefile.gcc,Makefile.inc}`, same for `TCPStreamerClient`, `Test/GTest/TCPStreamerGTest.cpp`; modify root `Makefile.gcc`, `Test/GTest/Makefile.inc`.
|
|
|
|
- [ ] **Step 1:** `TCPStreamerGTest.cpp`: `Constructor_Defaults` (`GetMaxPayloadSize()==60000`, `GetPort()==44500`), `Initialise_RejectsUdpOnlyKeys` (both classes), `Initialise_TcpKeys`, `ClassRegistered`, `StreamerToRawClient` (RealTimeApplication with `TCPStreamer`, raw client sees CONFIG then DATA while the test calls `Synchronise()`), `RoundTripStreamerToClient` (`TCPSServer`-driven client DS: send CONFIG+DATA, `Synchronise()` on a helper thread, check decoded value).
|
|
- [ ] **Step 2:** run -> compile failure. **Step 3:** implement both classes (each overrides `Initialise` to reject keys, `DefaultMaxPayloadSize`/`InitialiseTransport`, constructor swaps in the TCP transport). Makefile.inc copies UDPStreamer's, with `-lUDPStreamer -lUDPStream` (client: `-lUDPStreamerClient`), packages `Components/DataSources`.
|
|
- [ ] **Step 4:** add to root `Makefile.gcc` `core` and `clean`; if the `MainGTest` link reports undefined `UDPStreamer` symbols wrap `LIBRARIES_STATIC` in `-Wl,--start-group ... -Wl,--end-group`.
|
|
- [ ] **Step 5:** full `make core test`, whole `MainGTest.ex` and UDP datasource tests pass.
|
|
|
|
### Task 5: Docs and spec sync
|
|
|
|
- [ ] `Docs/TCPStreamer.md` (config tables, behaviour, service feasibility pointer), `CLAUDE.md` one-line mention, and update the spec for decisions taken while building: lazy slot growth, `SendBufferSize`, `GetPort()` returns the bound port, `SendData` true unless oversize, client `SilenceTimeout` default 0, no client-side gap handling, Strict frames may exceed `MaxPayloadSize` up to 1 MiB.
|
|
|
|
## Self-review
|
|
|
|
Spec coverage: pluggable transport (T1), `TCPSServer` incl. CONFIG-first, drop, stall (T2), `TCPSClient` (T3), both datasources + rejection of UDP-only keys + tests + regression (T4), docs/service study already in spec section 7 (T5). Naming checked: `UDPSServerI`, `UDPSClientI`, `SetServer`, `SetClient`, `Transport()`, `DefaultMaxPayloadSize`, `InitialiseTransport` are used identically across tasks.
|