diff --git a/CLAUDE.md b/CLAUDE.md index 39b79f3..8f646cc 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -49,6 +49,8 @@ Two independent data paths: **StreamHub WebSocket protocol**: JSON text frames for commands/events, binary frames for data pushes — spec in `ARCHITECTURE.md` §6. The Go hub (`Client/udpstreamer`) and C++ StreamHub implement the identical protocol; both clients (browser JS and ImGui) must stay compatible with both. +**StreamHub History**: `HistoryWriter` (`Source/Applications/StreamHub/HistoryWriter.{h,cpp}`) provides disk-backed circular storage of signal data. Each signal is stored in a separate `.shist` binary file with a 64-byte header and pre-allocated circular data region of `(float64 time, float64 value)` pairs. Configured via a `+History` block in the StreamHub config with keys: `Directory` (required), `DurationHours` (default 1), `Decimation` (default 1), `FlushIntervalSec` (default 5), `MinDiskFreeMB` (default 500). WS commands: `historyInfo` (returns metadata), `historyZoom` (reads time range from disk). Both the web SPA (`Client/udpstreamer/static/app.js`) and ImGui client support history zoom queries alongside in-memory ring zoom. + ## Conventions - **No STL in MARTe2 components** (`Source/Components/`): use MARTe2 types (`StreamString`, `FastPollingMutexSem`, fixed arrays). STL/C++17 is fine in `Client/streamhub/` and `Source/Applications/StreamHub/` follows MARTe2 style but links MARTe2 core. diff --git a/Client/streamhub/App.cpp b/Client/streamhub/App.cpp index fc85154..6ea8b2c 100644 --- a/Client/streamhub/App.cpp +++ b/Client/streamhub/App.cpp @@ -503,6 +503,8 @@ void App::handleJSON(const std::string& json) { else if (type == "stats") { onStats(json); } else if (type == "triggerState") { onTriggerState(json); } else if (type == "zoom") { onZoom(json); } + else if (type == "historyZoom") { onHistoryZoom(json); } + else if (type == "historyInfo") { onHistoryInfo(json); } else if (type == "maxPointsUpdated") { onMaxPointsUpdated(json); } /* pong: ignore */ } @@ -705,4 +707,40 @@ void App::onMaxPointsUpdated(const std::string& json) { } } +/* ---- History ------------------------------------------------------------ */ + +void App::onHistoryZoom(const std::string& json) { + ZoomResponse resp; + if (!ParseZoom(json, resp)) { return; } + + /* Route the reply to the plot that requested it (via histZoomCache_) */ + for (int i = 0; i < kMaxPlotSlots; i++) { + auto& hc = histZoomCache_[i]; + if (hc.pending && hc.reqId == resp.reqId) { + hc.signals = std::move(resp.signals); + hc.t0 = hc.reqT0; + hc.t1 = hc.reqT1; + hc.valid = true; + hc.pending = false; + return; + } + } +} + +void App::onHistoryInfo(const std::string& json) { + ParseHistoryInfo(json, historyInfo_); +} + +void App::requestHistoryZoom(int plotIdx, double t0, double t1, + const std::string& signalsCsv) { + if (plotIdx < 0 || plotIdx >= kMaxPlotSlots) { return; } + if (!ws_.isConnected() || signalsCsv.empty()) { return; } + auto& hc = histZoomCache_[plotIdx]; + hc.reqId = nextZoomReqId_++; + hc.reqT0 = t0; + hc.reqT1 = t1; + hc.pending = true; + ws_.sendText(BuildHistoryZoom(hc.reqId, t0, t1, 2400, signalsCsv)); +} + } /* namespace StreamHubClient */ diff --git a/Client/streamhub/App.h b/Client/streamhub/App.h index d02bede..ecd97dc 100644 --- a/Client/streamhub/App.h +++ b/Client/streamhub/App.h @@ -232,6 +232,18 @@ public: /** @brief Full key "src:sig" for an assignment (empty if invalid). */ std::string slotKey(const PlotAssignment& a) const; + /* ---- History ------------------------------------------------------------ */ + + /** @brief History info received from the hub. */ + const HistoryInfoMsg& historyInfo() const { return historyInfo_; } + + /** @brief Send a historyZoom request. */ + void requestHistoryZoom(int plotIdx, double t0, double t1, + const std::string& signalsCsv); + + /** @brief Hi-res history zoom cache (per plot, parallel to zoomCache_). */ + PlotZoomCache& histZoomCache(int i) { return histZoomCache_[i]; } + /** @brief True if the stats panel is open. */ bool& showStats() { return showStats_; } bool& showAddSrc() { return showAddSrc_; } @@ -253,6 +265,8 @@ private: void onStats(const std::string& json); void onTriggerState(const std::string& json); void onZoom(const std::string& json); + void onHistoryZoom(const std::string& json); + void onHistoryInfo(const std::string& json); void onMaxPointsUpdated(const std::string& json); /* ---- Rendering ------------------------------------------------------ */ @@ -296,8 +310,12 @@ private: /* Zoom history + hi-res zoom cache */ std::vector> zoomHist_[kMaxPlotSlots]; PlotZoomCache zoomCache_[kMaxPlotSlots]; + PlotZoomCache histZoomCache_[kMaxPlotSlots]; uint32_t nextZoomReqId_ = 1; + /* History info from the hub */ + HistoryInfoMsg historyInfo_; + /* UI state */ bool showStats_ = false; bool showAddSrc_ = false; diff --git a/Client/streamhub/Icons.h b/Client/streamhub/Icons.h index 267f191..a4a6422 100644 --- a/Client/streamhub/Icons.h +++ b/Client/streamhub/Icons.h @@ -30,5 +30,6 @@ #define ICON_FA_CHEVRON_RIGHT ">" #define ICON_FA_ARROW_TREND_UP "/\\" #define ICON_FA_ARROW_TREND_DOWN "\\/" -#define ICON_FA_ARROWS_UP_DOWN "/\\\\/" +#define ICON_FA_ARROWS_UP_DOWN "/\\\\/" +#define ICON_FA_CLOCK_ROTATE_LEFT "<-" #endif diff --git a/Client/streamhub/PlotPanel.cpp b/Client/streamhub/PlotPanel.cpp index 563ed9a..6e85b68 100644 --- a/Client/streamhub/PlotPanel.cpp +++ b/Client/streamhub/PlotPanel.cpp @@ -204,6 +204,13 @@ void RenderPlotPanel(App& app, int plotIdx, bool& paused) { zc.t0 <= app.plotXMin(plotIdx) + 1e-9 && zc.t1 >= app.plotXMax(plotIdx) - 1e-9)); + /* History zoom cache (disk-backed, for ranges beyond in-memory ring) */ + auto& hc = app.histZoomCache(plotIdx); + const bool useHistData = !trigView && !live && !useZoomData && + hc.valid && + hc.t0 <= app.plotXMin(plotIdx) + 1e-9 && + hc.t1 >= app.plotXMax(plotIdx) - 1e-9; + /* Fill tStore/vStore[si] from the frozen view (paused) or the live ring */ auto readBase = [&](size_t si, const Signal& sig, const std::string& key) { if (paused && snap.valid) { @@ -249,6 +256,18 @@ void RenderPlotPanel(App& app, int plotIdx, bool& paused) { if (!found) { readBase(si, sig, key); } + } else if (useHistData) { + bool found = false; + for (const auto& hs : hc.signals) { + if (hs.name != key) continue; + tStore[si] = hs.t; + vStore[si] = hs.v; + found = true; + break; + } + if (!found) { + readBase(si, sig, key); + } } else { readBase(si, sig, key); } @@ -375,6 +394,13 @@ void RenderPlotPanel(App& app, int plotIdx, bool& paused) { } } + /* History indicator */ + if (useHistData) { + ImGui::SameLine(); + ImGui::TextColored(ImVec4(0.980f,0.702f,0.529f,1.f), + ICON_FA_CLOCK_ROTATE_LEFT " HIST"); + } + /* Norm mode + cursors toggles */ ImGui::SameLine(); { @@ -659,7 +685,9 @@ void RenderPlotPanel(App& app, int plotIdx, bool& paused) { } } else if (!live) { /* Non-live: debounced — every zoom/pan settles into one - * request, so each zoom level gets fresh resolution. */ + * request, so each zoom level gets fresh resolution. + * If the range is covered by the in-memory ring, use regular + * zoom; otherwise try historyZoom (disk-backed). */ static double rangeChangedAt[kPlotSlotsMax] = {}; static double lastT0[kPlotSlotsMax] = {}, lastT1[kPlotSlotsMax] = {}; const double now = ImGui::GetTime(); @@ -670,9 +698,17 @@ void RenderPlotPanel(App& app, int plotIdx, bool& paused) { rangeChangedAt[plotIdx] = now; } else if (rangeChangedAt[plotIdx] > 0.0 && now - rangeChangedAt[plotIdx] > 0.35 && - !zc.pending && - !(zc.valid && zc.t0 == t0 && zc.t1 == t1)) { - if (!csv.empty()) { app.requestZoom(plotIdx, t0, t1, csv); } + !csv.empty()) { + /* Try regular zoom first */ + if (!zc.pending && !(zc.valid && zc.t0 == t0 && zc.t1 == t1)) { + app.requestZoom(plotIdx, t0, t1, csv); + } + /* Also try history zoom if history is available */ + const auto& hi = app.historyInfo(); + if (hi.enabled && !hc.pending && + !(hc.valid && hc.t0 == t0 && hc.t1 == t1)) { + app.requestHistoryZoom(plotIdx, t0, t1, csv); + } rangeChangedAt[plotIdx] = 0.0; } } diff --git a/Client/streamhub/Protocol.cpp b/Client/streamhub/Protocol.cpp index 3815ebd..9a21691 100644 --- a/Client/streamhub/Protocol.cpp +++ b/Client/streamhub/Protocol.cpp @@ -253,6 +253,23 @@ std::string BuildZoom(uint32_t reqId, double t0, double t1, int n, return out; } +std::string BuildHistoryZoom(uint32_t reqId, double t0, double t1, int n, + const std::string& signalsCsv) { + char head[256]; + snprintf(head, sizeof(head), + "{\"type\":\"historyZoom\",\"reqId\":%u," + "\"t0\":%.17g,\"t1\":%.17g,\"n\":%d,\"signals\":\"", + static_cast(reqId), t0, t1, n); + std::string out(head); + out += signalsCsv; + out += "\"}"; + return out; +} + +std::string BuildHistoryInfo() { + return "{\"type\":\"historyInfo\"}"; +} + std::string BuildSetMaxPoints(uint32_t n) { char buf[128]; snprintf(buf, sizeof(buf), "{\"type\":\"setMaxPoints\",\"maxPoints\":%u}", @@ -530,4 +547,60 @@ bool ParseMaxPointsUpdated(const std::string& json, uint32_t& maxPoints) { return true; } +/* ---- historyInfo -------------------------------------------------------- */ + +bool ParseHistoryInfo(const std::string& json, HistoryInfoMsg& out) { + out.signals.clear(); + out.enabled = false; + + char tmp[16] = ""; + jsonGetStr(json.c_str(), "enabled", tmp, sizeof(tmp)); + out.enabled = (strcmp(tmp, "true") == 0); + + double df = 0.0; + jsonGetDouble(json.c_str(), "durationHours", df); + out.durationHours = df; + + double decF = 0.0; + jsonGetDouble(json.c_str(), "decimation", decF); + out.decimation = static_cast(decF); + + /* Parse "signals":{...} block — same key iteration as zoom */ + const char *sig = strstr(json.c_str(), "\"signals\":{"); + if (!sig) { return out.enabled; } + sig += 11; /* past '{"signals":{' */ + + while (*sig) { + /* Find next "key":{ */ + const char *q = strchr(sig, '"'); + if (!q) { break; } + q++; + const char *qEnd = strchr(q, '"'); + if (!qEnd) { break; } + + HistorySignalRange hr; + hr.key = std::string(q, qEnd); + + /* Find the nested object */ + const char *brace = strchr(qEnd, '{'); + if (!brace) { break; } + + jsonGetDouble(brace, "t0", hr.t0); + jsonGetDouble(brace, "t1", hr.t1); + double cntF = 0.0, capF = 0.0; + jsonGetDouble(brace, "count", cntF); + jsonGetDouble(brace, "capacity", capF); + hr.count = static_cast(cntF); + hr.capacity = static_cast(capF); + + out.signals.push_back(hr); + + /* Advance past the closing brace */ + const char *end = strchr(brace, '}'); + if (!end) { break; } + sig = end + 1; + } + return true; +} + } /* namespace StreamHubClient */ diff --git a/Client/streamhub/Protocol.h b/Client/streamhub/Protocol.h index bcd4e21..8050df3 100644 --- a/Client/streamhub/Protocol.h +++ b/Client/streamhub/Protocol.h @@ -161,6 +161,10 @@ std::string BuildSetTrigger(const std::string& signalKey, const std::string& edg /** signalsCsv: comma-separated full keys "src:sig,src:sig2"; n<=0 → raw. */ std::string BuildZoom(uint32_t reqId, double t0, double t1, int n, const std::string& signalsCsv); +/** History zoom: same shape as zoom but reads from disk history. */ +std::string BuildHistoryZoom(uint32_t reqId, double t0, double t1, int n, + const std::string& signalsCsv); +std::string BuildHistoryInfo(); std::string BuildSetMaxPoints(uint32_t n); /*---------------------------------------------------------------------------*/ @@ -187,6 +191,22 @@ bool ParseTriggerState(const std::string& json, TriggerStateMsg& out); /** @brief Parse a "zoom" response. */ bool ParseZoom(const std::string& json, ZoomResponse& out); +/** @brief Parse a "historyInfo" event. */ +struct HistorySignalRange { + std::string key; + double t0 = 0.0, t1 = 0.0; + uint32_t count = 0, capacity = 0; +}; + +struct HistoryInfoMsg { + bool enabled = false; + double durationHours = 0.0; + uint32_t decimation = 0; + std::vector signals; +}; + +bool ParseHistoryInfo(const std::string& json, HistoryInfoMsg& out); + /** @brief Parse a "maxPointsUpdated" event. */ bool ParseMaxPointsUpdated(const std::string& json, uint32_t& maxPoints); diff --git a/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/build.make b/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/build.make index d211c68..c246933 100644 --- a/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/build.make +++ b/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/build.make @@ -209,7 +209,7 @@ StreamHubClient: CMakeFiles/StreamHubClient.dir/StatsPanel.cpp.o StreamHubClient: CMakeFiles/StreamHubClient.dir/build.make StreamHubClient: CMakeFiles/StreamHubClient.dir/compiler_depend.ts StreamHubClient: libimgui_lib.a -StreamHubClient: /usr/lib/libSDL2-2.0.so.0.3200.68 +StreamHubClient: /usr/lib/libSDL2-2.0.so.0.3200.70 StreamHubClient: /usr/lib/libGLX.so StreamHubClient: /usr/lib/libOpenGL.so StreamHubClient: CMakeFiles/StreamHubClient.dir/link.txt diff --git a/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/compiler_depend.internal b/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/compiler_depend.internal index b425fde..3c17204 100644 --- a/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/compiler_depend.internal +++ b/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/compiler_depend.internal @@ -2451,7 +2451,7 @@ StreamHubClient /usr/lib/libGLX.so /usr/lib/libGLdispatch.so.0 /usr/lib/libOpenGL.so - /usr/lib/libSDL2-2.0.so.0.3200.68 + /usr/lib/libSDL2-2.0.so.0.3200.70 /usr/lib/libX11.so.6 /usr/lib/libXau.so.6 /usr/lib/libXdmcp.so.6 diff --git a/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/compiler_depend.make b/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/compiler_depend.make index 318993e..47c7ab9 100644 --- a/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/compiler_depend.make +++ b/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/compiler_depend.make @@ -2442,7 +2442,7 @@ StreamHubClient: /usr/lib/Scrt1.o \ /usr/lib/libGLX.so \ /usr/lib/libGLdispatch.so.0 \ /usr/lib/libOpenGL.so \ - /usr/lib/libSDL2-2.0.so.0.3200.68 \ + /usr/lib/libSDL2-2.0.so.0.3200.70 \ /usr/lib/libX11.so.6 \ /usr/lib/libXau.so.6 \ /usr/lib/libXdmcp.so.6 \ @@ -2487,7 +2487,7 @@ CMakeFiles/StreamHubClient.dir/StatsPanel.cpp.o: /usr/lib/libXau.so.6: -/usr/lib/libSDL2-2.0.so.0.3200.68: +/usr/lib/libSDL2-2.0.so.0.3200.70: /usr/lib/libGLdispatch.so.0: diff --git a/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/link.txt b/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/link.txt index 6c6a69e..d21b1ee 100644 --- a/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/link.txt +++ b/Client/streamhub/build/CMakeFiles/StreamHubClient.dir/link.txt @@ -1 +1 @@ -/usr/bin/c++ -O3 -DNDEBUG -Wl,--dependency-file=CMakeFiles/StreamHubClient.dir/link.d CMakeFiles/StreamHubClient.dir/main.cpp.o CMakeFiles/StreamHubClient.dir/App.cpp.o CMakeFiles/StreamHubClient.dir/WSClient.cpp.o CMakeFiles/StreamHubClient.dir/Protocol.cpp.o CMakeFiles/StreamHubClient.dir/SourcePanel.cpp.o CMakeFiles/StreamHubClient.dir/PlotPanel.cpp.o CMakeFiles/StreamHubClient.dir/TriggerPanel.cpp.o CMakeFiles/StreamHubClient.dir/StatsPanel.cpp.o -o StreamHubClient libimgui_lib.a /usr/lib/libSDL2-2.0.so.0.3200.68 -lpthread /usr/lib/libGLX.so /usr/lib/libOpenGL.so +/usr/bin/c++ -O3 -DNDEBUG -Wl,--dependency-file=CMakeFiles/StreamHubClient.dir/link.d CMakeFiles/StreamHubClient.dir/main.cpp.o CMakeFiles/StreamHubClient.dir/App.cpp.o CMakeFiles/StreamHubClient.dir/WSClient.cpp.o CMakeFiles/StreamHubClient.dir/Protocol.cpp.o CMakeFiles/StreamHubClient.dir/SourcePanel.cpp.o CMakeFiles/StreamHubClient.dir/PlotPanel.cpp.o CMakeFiles/StreamHubClient.dir/TriggerPanel.cpp.o CMakeFiles/StreamHubClient.dir/StatsPanel.cpp.o -o StreamHubClient libimgui_lib.a /usr/lib/libSDL2-2.0.so.0.3200.70 -lpthread /usr/lib/libGLX.so /usr/lib/libOpenGL.so diff --git a/Client/streamhub/build/StreamHubClient b/Client/streamhub/build/StreamHubClient index ad7f1ab..7d4f6b6 100755 Binary files a/Client/streamhub/build/StreamHubClient and b/Client/streamhub/build/StreamHubClient differ diff --git a/Client/udpstreamer/static/app.js b/Client/udpstreamer/static/app.js index 121f98c..99918d1 100644 --- a/Client/udpstreamer/static/app.js +++ b/Client/udpstreamer/static/app.js @@ -344,6 +344,8 @@ function connectWS() { else if (msg.type === 'data') onData(msg); else if (msg.type === 'stats') onStats(msg); else if (msg.type === 'zoom') onZoomReply(msg); + else if (msg.type === 'historyZoom') onHistoryZoomReply(msg); + else if (msg.type === 'historyInfo') onHistoryInfo(msg); else if (msg.type === 'triggerState') onTriggerState(msg); }; } @@ -372,6 +374,56 @@ function onZoomReply(msg) { p.resolve(msg.signals || {}); } +/* ════════════════════════════════════════════════════════════════ + History (disk-backed data from the hub) + ════════════════════════════════════════════════════════════════ */ +let historyMeta = null; // {enabled, durationHours, decimation, signals:{key:{t0,t1,count,capacity}}} +const historyData = {}; // plotId → {signals:{key:{t,v}}, t0, t1} + +const _histPending = new Map(); +function wsHistoryZoomRequest(t0, t1, n, keys) { + return new Promise((resolve, reject) => { + if (!ws || ws.readyState !== WebSocket.OPEN) { reject(new Error('ws not open')); return; } + const reqId = ++_zoomReqId; + const timer = setTimeout(() => { + _histPending.delete(reqId); + reject(new Error('historyZoom timeout')); + }, 10000); + _histPending.set(reqId, { resolve, timer }); + ws.send(JSON.stringify({ type: 'historyZoom', reqId, t0, t1, n, signals: keys.join(',') })); + }); +} + +function onHistoryZoomReply(msg) { + const p = _histPending.get(msg.reqId); + if (!p) return; + clearTimeout(p.timer); + _histPending.delete(msg.reqId); + p.resolve(msg.signals || {}); +} + +function onHistoryInfo(msg) { + historyMeta = { + enabled: msg.enabled || false, + durationHours: msg.durationHours || 0, + decimation: msg.decimation || 1, + signals: msg.signals || {}, + }; + updateHistoryUI(); +} + +function updateHistoryUI() { + const badge = document.getElementById('history-badge'); + if (!badge) return; + if (historyMeta && historyMeta.enabled) { + const nSigs = Object.keys(historyMeta.signals).length; + badge.textContent = `History: ${historyMeta.durationHours}h, ${nSigs} signals`; + badge.style.display = ''; + } else { + badge.style.display = 'none'; + } +} + /* ════════════════════════════════════════════════════════════════ Status LED ════════════════════════════════════════════════════════════════ */ @@ -935,18 +987,62 @@ async function doZoomFetch(t0, t1) { if (allKeys.size === 0) return; const targetPts = Math.max(400, (plots.find(p => p.uplot)?.uplot?.width || 600) * 2); - let sigs; - try { - sigs = await wsZoomRequest(t0, t1, targetPts, [...allKeys]); - } catch (e) { console.warn('zoom fetch:', e); return; } - if (!sigs) return; - // Convert arrays from JSON to Float64Array and store per-plot. + // Fire regular zoom + history zoom in parallel + const promises = []; + + // Regular zoom (in-memory ring) + promises.push(wsZoomRequest(t0, t1, targetPts, [...allKeys]).catch(e => { + console.warn('zoom fetch:', e); return null; + })); + + // History zoom (disk) — only if history is enabled + if (historyMeta && historyMeta.enabled) { + promises.push(wsHistoryZoomRequest(t0, t1, targetPts, [...allKeys]).catch(e => { + console.warn('history zoom fetch:', e); return null; + })); + } + + const [sigs, histSigs] = await Promise.all(promises); + + // Merge: history data first (older), regular zoom overlays (newer wins) const merged = {}; - Object.entries(sigs).forEach(([k, sd]) => { - if (!sd || !sd.t || !sd.v) return; - merged[k] = { t: Float64Array.from(sd.t), v: Float64Array.from(sd.v) }; - }); + if (histSigs) { + Object.entries(histSigs).forEach(([k, sd]) => { + if (!sd || !sd.t || !sd.v) return; + merged[k] = { t: Float64Array.from(sd.t), v: Float64Array.from(sd.v) }; + }); + } + if (sigs) { + Object.entries(sigs).forEach(([k, sd]) => { + if (!sd || !sd.t || !sd.v || sd.t.length === 0) return; + const zt = Float64Array.from(sd.t), zv = Float64Array.from(sd.v); + if (merged[k] && merged[k].t.length > 0) { + // Merge: use history for t < ring oldest, ring for the rest + const ht = merged[k].t, hv = merged[k].v; + const ringOldest = zt[0]; + // Find cutoff in history data + let cutIdx = ht.length; + for (let i = 0; i < ht.length; i++) { + if (ht[i] >= ringOldest) { cutIdx = i; break; } + } + if (cutIdx > 0) { + // Prepend history data before ring data + const mt = new Float64Array(cutIdx + zt.length); + const mv = new Float64Array(cutIdx + zv.length); + mt.set(ht.subarray(0, cutIdx)); + mv.set(hv.subarray(0, cutIdx)); + mt.set(zt, cutIdx); + mv.set(zv, cutIdx); + merged[k] = { t: mt, v: mv }; + } else { + merged[k] = { t: zt, v: zv }; + } + } else { + merged[k] = { t: zt, v: zv }; + } + }); + } plots.forEach(p => { // Only store if this plot's range still matches what we fetched. diff --git a/Client/udpstreamer/static/index.html b/Client/udpstreamer/static/index.html index 5073340..76150bc 100644 --- a/Client/udpstreamer/static/index.html +++ b/Client/udpstreamer/static/index.html @@ -113,6 +113,7 @@ Disconnected + diff --git a/Docs/StreamHub-API.md b/Docs/StreamHub-API.md index 1104f06..6039a5c 100644 --- a/Docs/StreamHub-API.md +++ b/Docs/StreamHub-API.md @@ -101,6 +101,27 @@ Every transition is broadcast as a `triggerState` event. - `signals` — comma-separated full keys; absent → all signals of all sources. - The reply is **unicast** to the requesting client only. +### `historyZoom` + +```json +{"type":"historyZoom","reqId":42,"t0":1765360000.0,"t1":1765370000.0, + "n":2400,"signals":"s1:Sine,s2:Wave"} +``` +- Reads from **disk-backed history** instead of the in-memory ring buffer. + Identical semantics to `zoom` but queries the `.shist` files written by + `HistoryWriter`. +- `reqId`, `t0`/`t1`, `n`, `signals` — same meaning as `zoom`. +- If history is not enabled, the reply contains `"error":"history not enabled"`. +- Reply is **unicast** (same shape as `zoom` reply, but `"type":"historyZoom"`). + +### `historyInfo` + +```json +{"type":"historyInfo"} +``` +Request the hub to send a `historyInfo` event (unicast). Also sent automatically +on client connect. + ### `setMaxPoints` ```json @@ -164,6 +185,34 @@ trigger has fired. `t` is serialised with `%.17g` (full float64 precision — required for µs windows at Unix-epoch magnitudes), `v` with `%.9g`. +### `historyInfo` + +Sent on client connect (if history is enabled) and on `historyInfo` command: + +```json +{"type":"historyInfo","enabled":true,"durationHours":1.0,"decimation":10, + "signals":{ + "scalar:Sine1":{"t0":1765360000.0,"t1":1765370000.0,"count":360000,"capacity":360000}, + "scalar:Sine2":{"t0":1765360000.0,"t1":1765370000.0,"count":360000,"capacity":360000}}} +``` +- `enabled` — `true` if the `+History` config block is present and valid. +- `durationHours` — configured history duration. +- `decimation` — samples-to-disk decimation factor (1 = every sample). +- `signals` — per-signal metadata keyed by `"sourceId:signalName"`: + - `t0`/`t1` — oldest/newest timestamp stored on disk (Unix seconds). + - `count` — number of valid entries currently in the circular file. + - `capacity` — total capacity of the circular file. + +### `historyZoom` (reply) + +Same shape as `zoom` reply, but `"type":"historyZoom"`: + +```json +{"type":"historyZoom","reqId":42,"signals":{ + "s1:Sine":{"t":[1765360000.1230000,"…"],"v":[0.123456789,"…"]}}} +``` +If history is not enabled: `{"type":"historyZoom","error":"history not enabled"}`. + ### `maxPointsUpdated` ```json diff --git a/Docs/StreamHub-Developer.md b/Docs/StreamHub-Developer.md index 5ca27da..d357a86 100644 --- a/Docs/StreamHub-Developer.md +++ b/Docs/StreamHub-Developer.md @@ -21,6 +21,7 @@ Wire protocols: [Protocol.md](Protocol.md) (UDPS, source → hub) and | `SignalRingBuffer.h` | Per-signal (t,v) ring: monotonic write counter, `ReadSince` cursor reads, binary-search `ReadRange` | | `TriggerEngine.{h,cpp}` | Trigger FSM (IDLE/ARMED/COLLECTING/TRIGGERED), edge detection per decoded sample | | `UDPSourceStats.h` | 512-entry cycle/frag/byte rings → avg/std/min/max, rate, 20-bin histogram | +| `HistoryWriter.{h,cpp}` | Disk-backed circular history: per-signal `.shist` files, `WriteTick`, `ReadRange` (binary search via `pread`), disk space monitoring | | `WSServer.{h,cpp}` | RFC 6455 server: handshake, framing, per-client write mutex, broadcast/unicast | | `WSFrame.h`, `SHA1.h`, `Base64.h` | Header-only WS plumbing, shared with the ImGui client | | `LTTB.h` | Largest-Triangle-Three-Buckets decimation | @@ -124,6 +125,66 @@ Sources = { Sources from `SourcesFile` are loaded after the static `Sources` block; sources added at runtime via WS `addSource` get ids `s1, s2, …`. +### History configuration + +An optional `+History` block enables disk-backed circular storage +(`HistoryWriter`). The `+` prefix is MARTe2 `StandardParser` syntax for a +child node; the hub looks for both `+History` and `History` as the node name. + +``` ++History = { + Directory = "/data/streamhub_history" // required + DurationHours = 1 // hours of data to retain per signal (default 1) + Decimation = 10 // keep every Nth sample (default 1) + FlushIntervalSec = 5 // header flush period in seconds (default 5) + MinDiskFreeMB = 500 // pause writes below this threshold (default 500) +} +``` + +Per-signal file capacity is computed at source CONFIG time: +`capacity = ceil(DurationHours × 3600 × samplingRate / Decimation)`, minimum +1000 pairs. + +### `.shist` binary file format + +Each signal gets one file: `//.shist`. +The file size is fixed at creation (`64 + capacity × 16` bytes) and never grows. + +``` +Offset Size Field +0 4 Magic: "SHR1" +4 4 uint32 version (1) +8 4 uint32 capacity (max pairs) +12 4 uint32 head (next write position, 0-based, wraps at capacity) +16 4 uint32 count (valid entries, ≤ capacity) +20 4 uint32 decimation +24 8 float64 tOldest (Unix seconds) +32 8 float64 tNewest (Unix seconds) +40 24 reserved (zero-padded to 64 bytes) +64 … data: capacity × 16 bytes (float64 time + float64 value per pair) +``` + +Data is written at the `head` position and wraps circularly. The oldest valid +entry is at logical index `(head + capacity − count) % capacity`. All I/O +uses `pwrite`/`pread` (no `mmap`), so concurrent reads from WS threads are +safe without locking. + +Headers are flushed to disk every `FlushIntervalSec` seconds and on shutdown. +On restart, if an existing file has matching magic, version and capacity, it is +reopened — head/count/time bounds are restored from the on-disk header. + +### History query path + +`historyZoom` requests (see [StreamHub-API.md](StreamHub-API.md)) call +`HistoryWriter::ReadRange` which performs binary search over the circular file +using `pread` to locate the `[t0, t1]` window, then copies matching pairs. +If the result exceeds the requested `n`, LTTB decimation is applied (same +`LTTBDecimate` as in-memory zoom). + +Both the web SPA and ImGui client issue `historyZoom` in parallel with regular +`zoom` and merge the results: history covers the older part of the visible +window, the in-memory ring covers the recent part. + ## 7. Build & test ```bash @@ -147,12 +208,14 @@ make -f Makefile.gcc test `./run_e2e_test.sh` builds everything, launches the demo MARTe2 application (`Test/Configurations/streamhub_demo.cfg`: 3 UDPStreamers — multicast scalars, FirstSample/LastSample arrays, FullArray + uint64 ns time array) plus a -StreamHub on port 8095, then runs the Go WS client `Test/E2E/streamhub` which -verifies: `sources`/`config` events, ≥10 binary v1 pushes with wall-clock and -strictly monotonic time on all streams, `stats` shape, a `zoom` round-trip -(reqId echo, unicast), and a complete trigger cycle (setTrigger → arm → -binary v2 capture → triggered → disarm). Logs land in -`/tmp/streamhub_e2e_{marte,hub}.log`. Exit 0 iff every check passes. +StreamHub on port 8095 (with history enabled in `/tmp/streamhub_e2e_history`), +then runs the Go WS client `Test/E2E/streamhub` which verifies: +`sources`/`config` events, ≥10 binary v1 pushes with wall-clock and strictly +monotonic time on all streams, `stats` shape, a `zoom` round-trip (reqId echo, +unicast), `historyInfo` broadcast (enabled, duration, decimation, signal count), +a `historyZoom` round-trip (reqId echo, signal data), and a complete trigger +cycle (setTrigger → arm → binary v2 capture → triggered → disarm). Logs land +in `/tmp/streamhub_e2e_{marte,hub}.log`. Exit 0 iff every check passes. When changing the WS protocol, update **in lockstep**: this hub, the Go hub (`Common/Client/go/wshub`), the browser SPA (`Client/udpstreamer/static`), the diff --git a/Docs/StreamHub-UserGuide.md b/Docs/StreamHub-UserGuide.md index c122ea0..96cb041 100644 --- a/Docs/StreamHub-UserGuide.md +++ b/Docs/StreamHub-UserGuide.md @@ -98,7 +98,78 @@ client and uses full-rate ring data (not the decimated live stream). On capture, the trigger view shows all plotted signals relative to the trigger instant (marker at t = 0) over `[-pre, +post]`. -## 5. Troubleshooting +## 5. History (disk-backed storage) + +StreamHub can store signal data to disk, allowing you to browse hours of past +data even after the in-memory ring buffers have wrapped. + +### Enabling history + +Add a `+History` block to the hub configuration: + +``` +Hub = { + WSPort = 8090 + +History = { + Directory = "/data/streamhub_history" + DurationHours = 1 + Decimation = 10 + FlushIntervalSec = 5 + MinDiskFreeMB = 200 + } + Sources = { … } +} +``` + +| Key | Default | Description | +|-----|---------|-------------| +| `Directory` | *(required)* | Root directory for `.shist` files. Created automatically. Each source gets a subdirectory; each signal a separate file. | +| `DurationHours` | 1 | How many hours of data to keep per signal. The file size is fixed at creation: `duration × samplingRate / decimation × 16 bytes`. | +| `Decimation` | 1 | Write every N-th sample (1 = no decimation). Useful for high-rate signals where full-rate disk writes are impractical. | +| `FlushIntervalSec` | 5 | How often file headers are flushed to disk (data is written immediately via `pwrite`). | +| `MinDiskFreeMB` | 500 | Writing pauses when free disk space drops below this threshold and resumes when space is available again. | + +### Disk usage + +Each sample on disk is 16 bytes (8 bytes timestamp + 8 bytes value). Example: + +- 10 signals at 1 ksps, decimation 10, duration 1 hour: + `10 × (1000/10) × 3600 × 16 bytes = 57.6 MB` +- 10 signals at 1 Msps, decimation 1000, duration 1 hour: + `10 × (1e6/1000) × 3600 × 16 bytes = 576 MB` + +Files are pre-allocated at creation and never grow. The directory structure is: + +``` +/data/streamhub_history/ +├── scalar/ +│ ├── Sine1.shist +│ └── Sine2.shist +├── med/ +│ ├── Ch1.shist +│ └── Ch2.shist +└── fast/ + ├── Ch3.shist + └── Ch4.shist +``` + +### Using history in clients + +When history is enabled, both the web SPA and ImGui client show a **HIST** +indicator. History data is fetched transparently: + +- **Zoom**: when you zoom or pan into a time range that extends beyond the + in-memory ring buffer, the client issues a `historyZoom` request in parallel + with the regular `zoom` request and merges the results — history data fills + the older part of the window, ring data the recent part. +- **Status**: the web SPA displays a history badge in the status bar showing + the configured duration and decimation factor. The ImGui client shows + "HIST" in the plot header when history data is being displayed. + +History files survive hub restarts. If a `.shist` file exists with a matching +capacity, the hub reopens it and continues writing where it left off. + +## 6. Troubleshooting | Symptom | Check | |---------|-------| @@ -111,7 +182,7 @@ instant (marker at t = 0) over `[-pre, +post]`. Hub-side logs go to stdout; the demo scripts write `/tmp/streamhub_e2e_{marte,hub}.log`. -## 6. Further reading +## 7. Further reading - [StreamHub-API.md](StreamHub-API.md) — WebSocket protocol (write your own client) - [StreamHub-Developer.md](StreamHub-Developer.md) — internals, build, tests diff --git a/Source/Applications/StreamHub/HistoryWriter.cpp b/Source/Applications/StreamHub/HistoryWriter.cpp new file mode 100644 index 0000000..f600a27 --- /dev/null +++ b/Source/Applications/StreamHub/HistoryWriter.cpp @@ -0,0 +1,540 @@ +/** + * @file HistoryWriter.cpp + * @brief Disk-backed circular history storage implementation. + */ + +#include "HistoryWriter.h" +#include "AdvancedErrorManagement.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace StreamHub { + +/*---------------------------------------------------------------------------*/ +/* Constructor / Destructor */ +/*---------------------------------------------------------------------------*/ + +HistoryWriter::HistoryWriter() + : scratchT_(static_cast(0)), + scratchV_(static_cast(0)), + files_(static_cast(0)), + fileActive_(static_cast(0)), + enabled_(false), + durationHours_(1.0), + decimation_(1u), + flushIntervalSec_(5u), + minDiskFreeMB_(500u), + diskLow_(false) { + directory_[0] = '\0'; + files_ = new SignalFile[kHistMaxSessions][kHistMaxSignals]; + fileActive_ = new bool[kHistMaxSessions][kHistMaxSignals]; + for (uint32 i = 0u; i < kHistMaxSessions; i++) { + for (uint32 s = 0u; s < kHistMaxSignals; s++) { + files_[i][s].fd = -1; + files_[i][s].capacity = 0u; + files_[i][s].head = 0u; + files_[i][s].count = 0u; + files_[i][s].tOldest = 0.0; + files_[i][s].tNewest = 0.0; + files_[i][s].readCursor = 0u; + files_[i][s].decimCounter = 0u; + files_[i][s].headerDirty = false; + files_[i][s].sourceId[0] = '\0'; + files_[i][s].signalName[0] = '\0'; + fileActive_[i][s] = false; + } + } +} + +HistoryWriter::~HistoryWriter() { + /* Flush and close all open files */ + for (uint32 i = 0u; i < kHistMaxSessions; i++) { + for (uint32 s = 0u; s < kHistMaxSignals; s++) { + if (files_[i][s].fd >= 0) { + FlushHeader(files_[i][s]); + (void) close(files_[i][s].fd); + files_[i][s].fd = -1; + } + } + } + delete[] scratchT_; + delete[] scratchV_; + delete[] files_; + delete[] fileActive_; +} + +/*---------------------------------------------------------------------------*/ +/* Initialise */ +/*---------------------------------------------------------------------------*/ + +bool HistoryWriter::Initialise(StructuredDataI &cfg) { + StreamString dir; + if (!cfg.Read("Directory", dir) || dir.Size() == 0u) { + REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning, + "HistoryWriter: no Directory specified, history disabled."); + return false; + } + strncpy(directory_, dir.Buffer(), sizeof(directory_) - 1u); + directory_[sizeof(directory_) - 1u] = '\0'; + + /* Remove trailing slash */ + size_t dLen = strlen(directory_); + if (dLen > 1u && directory_[dLen - 1u] == '/') { + directory_[dLen - 1u] = '\0'; + } + + float64 durF = 1.0; + if (cfg.Read("DurationHours", durF)) { + durationHours_ = (durF > 0.0) ? durF : 1.0; + } + + uint32 tmp = 0u; + if (cfg.Read("Decimation", tmp)) { decimation_ = (tmp > 0u) ? tmp : 1u; } + if (cfg.Read("FlushIntervalSec", tmp)) { flushIntervalSec_ = (tmp > 0u) ? tmp : 5u; } + if (cfg.Read("MinDiskFreeMB", tmp)) { minDiskFreeMB_ = tmp; } + + /* Create root directory */ + (void) mkdir(directory_, 0755); + + /* Allocate scratch buffers */ + scratchT_ = new float64[kReadScratch]; + scratchV_ = new float64[kReadScratch]; + + enabled_ = true; + REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information, + "HistoryWriter: enabled, dir=%s, duration=%.1fh, decimation=%u, flush=%us, minDisk=%uMB.", + directory_, durationHours_, decimation_, flushIntervalSec_, minDiskFreeMB_); + return true; +} + +/*---------------------------------------------------------------------------*/ +/* OnSourceConfigured */ +/*---------------------------------------------------------------------------*/ + +void HistoryWriter::OnSourceConfigured(uint32 sessionIdx, const char *sourceId, + UDPSourceSession &sess) { + if (!enabled_ || sessionIdx >= kHistMaxSessions) { return; } + + const uint32 numSigs = sess.GetNumSignals(); + for (uint32 s = 0u; s < numSigs && s < kHistMaxSignals; s++) { + MARTe::UDPSSignalDescriptor desc; + if (!sess.GetSignalDescriptor(s, desc)) { continue; } + + /* Skip time-only signals (type uint64, typically TimeArray) */ + if (desc.typeCode == 13u) { continue; } /* uint64 = 13 in UDPS */ + + float64 estRate = static_cast(desc.samplingRate); + if (estRate <= 0.0) { estRate = 1000.0; } /* fallback */ + + if (OpenSignalFile(sessionIdx, s, sourceId, desc.name, estRate)) { + fileActive_[sessionIdx][s] = true; + } + } +} + +/*---------------------------------------------------------------------------*/ +/* OpenSignalFile */ +/*---------------------------------------------------------------------------*/ + +bool HistoryWriter::OpenSignalFile(uint32 sessionIdx, uint32 sigIdx, + const char *sourceId, const char *signalName, + float64 estimatedRateHz) { + /* Create source subdirectory */ + char srcDir[640]; + snprintf(srcDir, sizeof(srcDir), "%s/%s", directory_, sourceId); + (void) mkdir(srcDir, 0755); + + char filePath[768]; + snprintf(filePath, sizeof(filePath), "%s/%s.shist", srcDir, signalName); + + /* Calculate capacity: duration × rate / decimation */ + float64 totalSamples = durationHours_ * 3600.0 * estimatedRateHz + / static_cast(decimation_); + uint32 capacity = static_cast(ceil(totalSamples)); + if (capacity < 1000u) { capacity = 1000u; } + + SignalFile &sf = files_[sessionIdx][sigIdx]; + + /* Try to reopen an existing file */ + int fd = open(filePath, O_RDWR); + if (fd >= 0) { + uint8 hdr[kHistHeaderSize]; + ssize_t nr = pread(fd, hdr, kHistHeaderSize, 0); + if (nr == static_cast(kHistHeaderSize) && + memcmp(hdr, kHistMagic, 4) == 0) { + /* Read existing header */ + uint32 ver = 0u, fileCap = 0u; + memcpy(&ver, &hdr[4], 4); + memcpy(&fileCap, &hdr[8], 4); + if (ver == 1u && fileCap == capacity) { + /* Reuse existing file */ + memcpy(&sf.head, &hdr[12], 4); + memcpy(&sf.count, &hdr[16], 4); + memcpy(&sf.tOldest, &hdr[24], 8); + memcpy(&sf.tNewest, &hdr[32], 8); + sf.fd = fd; + sf.capacity = fileCap; + sf.decimCounter = 0u; + sf.readCursor = 0u; + sf.headerDirty = false; + strncpy(sf.sourceId, sourceId, sizeof(sf.sourceId) - 1u); + sf.sourceId[sizeof(sf.sourceId) - 1u] = '\0'; + strncpy(sf.signalName, signalName, sizeof(sf.signalName) - 1u); + sf.signalName[sizeof(sf.signalName) - 1u] = '\0'; + + REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information, + "HistoryWriter: reopened %s (cap=%u, count=%u).", + filePath, fileCap, sf.count); + return true; + } + } + /* Capacity changed or corrupt — recreate */ + (void) close(fd); + fd = -1; + } + + /* Create new file */ + fd = open(filePath, O_RDWR | O_CREAT | O_TRUNC, 0644); + if (fd < 0) { + REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning, + "HistoryWriter: cannot create %s: %s", filePath, strerror(errno)); + return false; + } + + /* Pre-allocate: header + capacity × 16 bytes */ + off_t fileSize = static_cast(kHistHeaderSize) + + static_cast(capacity) * 16; + if (ftruncate(fd, fileSize) != 0) { + REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning, + "HistoryWriter: ftruncate %s failed: %s", filePath, strerror(errno)); + (void) close(fd); + return false; + } + + /* Write header */ + uint8 hdr[kHistHeaderSize]; + memset(hdr, 0, kHistHeaderSize); + memcpy(&hdr[0], kHistMagic, 4); + uint32 ver = 1u; + memcpy(&hdr[4], &ver, 4); + memcpy(&hdr[8], &capacity, 4); + /* head=0, count=0, decimation, tOldest=0, tNewest=0 */ + memcpy(&hdr[20], &decimation_, 4); + (void) pwrite(fd, hdr, kHistHeaderSize, 0); + + sf.fd = fd; + sf.capacity = capacity; + sf.head = 0u; + sf.count = 0u; + sf.tOldest = 0.0; + sf.tNewest = 0.0; + sf.readCursor = 0u; + sf.decimCounter = 0u; + sf.headerDirty = false; + strncpy(sf.sourceId, sourceId, sizeof(sf.sourceId) - 1u); + sf.sourceId[sizeof(sf.sourceId) - 1u] = '\0'; + strncpy(sf.signalName, signalName, sizeof(sf.signalName) - 1u); + sf.signalName[sizeof(sf.signalName) - 1u] = '\0'; + + REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information, + "HistoryWriter: created %s (cap=%u, %.1f MB).", + filePath, capacity, + static_cast(fileSize) / (1024.0 * 1024.0)); + return true; +} + +/*---------------------------------------------------------------------------*/ +/* WriteTick */ +/*---------------------------------------------------------------------------*/ + +void HistoryWriter::WriteTick(uint32 sessionIdx, UDPSourceSession &sess) { + if (!enabled_ || sessionIdx >= kHistMaxSessions || diskLow_) { return; } + + const uint32 numSigs = sess.GetNumSignals(); + for (uint32 s = 0u; s < numSigs && s < kHistMaxSignals; s++) { + if (!fileActive_[sessionIdx][s]) { continue; } + SignalFile &sf = files_[sessionIdx][s]; + if (sf.fd < 0) { continue; } + + /* Read new samples from the in-memory ring since last tick */ + uint32 nRaw = sess.ReadSignalSince(s, sf.readCursor, + scratchT_, scratchV_, kReadScratch); + if (nRaw == 0u) { continue; } + + if (decimation_ <= 1u) { + WritePairs(sf, scratchT_, scratchV_, nRaw); + } else { + /* Apply decimation: take every Nth sample */ + uint32 nOut = 0u; + for (uint32 i = 0u; i < nRaw; i++) { + sf.decimCounter++; + if (sf.decimCounter >= static_cast(decimation_)) { + scratchT_[nOut] = scratchT_[i]; + scratchV_[nOut] = scratchV_[i]; + nOut++; + sf.decimCounter = 0u; + } + } + if (nOut > 0u) { + WritePairs(sf, scratchT_, scratchV_, nOut); + } + } + } +} + +/*---------------------------------------------------------------------------*/ +/* WritePairs */ +/*---------------------------------------------------------------------------*/ + +void HistoryWriter::WritePairs(SignalFile &sf, const float64 *t, + const float64 *v, uint32 n) { + for (uint32 i = 0u; i < n; i++) { + /* Write one (t,v) pair at the current head position */ + off_t offset = static_cast(kHistHeaderSize) + + static_cast(sf.head) * 16; + float64 pair[2] = { t[i], v[i] }; + (void) pwrite(sf.fd, pair, 16, offset); + + sf.head = (sf.head + 1u) % sf.capacity; + if (sf.count < sf.capacity) { + sf.count++; + } + } + + /* Update time bounds */ + sf.tNewest = t[n - 1u]; + if (sf.count <= n) { + sf.tOldest = t[0]; + } else { + /* Read the oldest entry's time from disk */ + uint32 oldestIdx = (sf.head + sf.capacity - sf.count) % sf.capacity; + off_t offset = static_cast(kHistHeaderSize) + + static_cast(oldestIdx) * 16; + float64 oldestT = 0.0; + (void) pread(sf.fd, &oldestT, 8, offset); + sf.tOldest = oldestT; + } + sf.headerDirty = true; +} + +/*---------------------------------------------------------------------------*/ +/* FlushHeaders */ +/*---------------------------------------------------------------------------*/ + +void HistoryWriter::FlushHeaders() { + if (!enabled_) { return; } + + /* Periodic disk space check */ + diskLow_ = !HasSufficientDisk(); + if (diskLow_) { + REPORT_ERROR_STATIC(MARTe::ErrorManagement::Warning, + "HistoryWriter: disk space below %u MB, writing paused.", minDiskFreeMB_); + } + + for (uint32 i = 0u; i < kHistMaxSessions; i++) { + for (uint32 s = 0u; s < kHistMaxSignals; s++) { + if (!fileActive_[i][s]) { continue; } + SignalFile &sf = files_[i][s]; + if (sf.fd >= 0 && sf.headerDirty) { + FlushHeader(sf); + } + } + } +} + +void HistoryWriter::FlushHeader(SignalFile &sf) { + uint8 hdr[kHistHeaderSize]; + memset(hdr, 0, kHistHeaderSize); + memcpy(&hdr[0], kHistMagic, 4); + uint32 ver = 1u; + memcpy(&hdr[4], &ver, 4); + memcpy(&hdr[8], &sf.capacity, 4); + memcpy(&hdr[12], &sf.head, 4); + memcpy(&hdr[16], &sf.count, 4); + memcpy(&hdr[20], &decimation_, 4); + memcpy(&hdr[24], &sf.tOldest, 8); + memcpy(&hdr[32], &sf.tNewest, 8); + (void) pwrite(sf.fd, hdr, kHistHeaderSize, 0); + (void) fdatasync(sf.fd); + sf.headerDirty = false; +} + +/*---------------------------------------------------------------------------*/ +/* ReadRange */ +/*---------------------------------------------------------------------------*/ + +uint32 HistoryWriter::ReadRange(const char *sourceId, const char *signalName, + float64 t0, float64 t1, + float64 *tOut, float64 *vOut, + uint32 maxOut) const { + const SignalFile *sf = FindFile(sourceId, signalName); + if (sf == static_cast(0) || sf->fd < 0 || + sf->count == 0u || t1 < t0) { + return 0u; + } + + const uint32 fileCap = sf->capacity; + const uint32 cnt = sf->count; + const uint32 oldest = (sf->head + fileCap - cnt) % fileCap; + const int fd = sf->fd; + + /* Binary search over the logical order (same algorithm as SignalRingBuffer). + * We read individual timestamps from disk using pread. */ + + /* lo = first logical index with t >= t0 */ + uint32 lo = 0u; + { + uint32 a = 0u, b = cnt; + while (a < b) { + const uint32 mid = a + ((b - a) >> 1); + const uint32 phys = (oldest + mid) % fileCap; + float64 tv = 0.0; + (void) pread(fd, &tv, 8, + static_cast(kHistHeaderSize) + static_cast(phys) * 16); + if (tv < t0) { a = mid + 1u; } else { b = mid; } + } + lo = a; + } + /* hi = first logical index with t > t1 */ + uint32 hi = lo; + { + uint32 a = lo, b = cnt; + while (a < b) { + const uint32 mid = a + ((b - a) >> 1); + const uint32 phys = (oldest + mid) % fileCap; + float64 tv = 0.0; + (void) pread(fd, &tv, 8, + static_cast(kHistHeaderSize) + static_cast(phys) * 16); + if (tv <= t1) { a = mid + 1u; } else { b = mid; } + } + hi = a; + } + + uint32 nOut = hi - lo; + if (nOut > maxOut) { nOut = maxOut; } + + /* Read pairs from disk */ + for (uint32 i = 0u; i < nOut; i++) { + uint32 physIdx = (oldest + lo + i) % fileCap; + off_t fileOff = static_cast(kHistHeaderSize) + + static_cast(physIdx) * 16; + float64 pair[2]; + (void) pread(fd, pair, 16, fileOff); + tOut[i] = pair[0]; + vOut[i] = pair[1]; + } + + return nOut; +} + +/*---------------------------------------------------------------------------*/ +/* GetTimeRange */ +/*---------------------------------------------------------------------------*/ + +bool HistoryWriter::GetTimeRange(const char *sourceId, const char *signalName, + float64 &t0, float64 &t1) const { + const SignalFile *sf = FindFile(sourceId, signalName); + if (sf == static_cast(0) || sf->count == 0u) { + return false; + } + t0 = sf->tOldest; + t1 = sf->tNewest; + return true; +} + +/*---------------------------------------------------------------------------*/ +/* AppendInfoJSON */ +/*---------------------------------------------------------------------------*/ + +/* Forward-declare JsonAppendf from StreamHub.cpp — we use the same pattern */ +static bool HistJsonAppendf(char *&buf, uint32 &len, uint32 &cap, + const char *fmt, ...) { + for (;;) { + va_list ap; + va_start(ap, fmt); + const int wrote = vsnprintf(buf + len, cap - len, fmt, ap); + va_end(ap); + if (wrote < 0) { return false; } + if (static_cast(wrote) < (cap - len)) { + len += static_cast(wrote); + return true; + } + uint32 newCap = cap * 2u; + while ((newCap - len) <= static_cast(wrote)) { + newCap *= 2u; + } + char *nb = new char[newCap]; + memcpy(nb, buf, len); + delete[] buf; + buf = nb; + cap = newCap; + } +} + +void HistoryWriter::AppendInfoJSON(char *&buf, uint32 &off, uint32 &cap) const { + HistJsonAppendf(buf, off, cap, + "\"enabled\":%s,\"durationHours\":%.2f,\"decimation\":%u,\"signals\":{", + enabled_ ? "true" : "false", durationHours_, decimation_); + + bool first = true; + for (uint32 i = 0u; i < kHistMaxSessions; i++) { + for (uint32 s = 0u; s < kHistMaxSignals; s++) { + if (!fileActive_[i][s]) { continue; } + const SignalFile &sf = files_[i][s]; + if (sf.count == 0u) { continue; } + + HistJsonAppendf(buf, off, cap, + "%s\"%s:%s\":{\"t0\":%.17g,\"t1\":%.17g,\"count\":%u,\"capacity\":%u}", + (first ? "" : ","), + sf.sourceId, sf.signalName, + sf.tOldest, sf.tNewest, sf.count, sf.capacity); + first = false; + } + } + HistJsonAppendf(buf, off, cap, "}"); +} + +/*---------------------------------------------------------------------------*/ +/* HasSufficientDisk */ +/*---------------------------------------------------------------------------*/ + +bool HistoryWriter::HasSufficientDisk() const { + if (minDiskFreeMB_ == 0u) { return true; } + + struct statvfs st; + if (statvfs(directory_, &st) != 0) { return true; /* assume OK */ } + + uint64 freeMB = (static_cast(st.f_bavail) * st.f_frsize) + / (1024u * 1024u); + return freeMB >= static_cast(minDiskFreeMB_); +} + +/*---------------------------------------------------------------------------*/ +/* FindFile */ +/*---------------------------------------------------------------------------*/ + +const HistoryWriter::SignalFile* HistoryWriter::FindFile( + const char *sourceId, const char *signalName) const { + for (uint32 i = 0u; i < kHistMaxSessions; i++) { + for (uint32 s = 0u; s < kHistMaxSignals; s++) { + if (!fileActive_[i][s]) { continue; } + const SignalFile &sf = files_[i][s]; + if (strcmp(sf.sourceId, sourceId) == 0 && + strcmp(sf.signalName, signalName) == 0) { + return &sf; + } + } + } + return static_cast(0); +} + +} /* namespace StreamHub */ diff --git a/Source/Applications/StreamHub/HistoryWriter.h b/Source/Applications/StreamHub/HistoryWriter.h new file mode 100644 index 0000000..2c0e021 --- /dev/null +++ b/Source/Applications/StreamHub/HistoryWriter.h @@ -0,0 +1,176 @@ +/** + * @file HistoryWriter.h + * @brief Disk-backed circular history storage for StreamHub signals. + * + * Each signal is stored in a separate binary file (.shist) with a fixed header + * and a pre-allocated circular data region of (float64 time, float64 value) pairs. + * The file size is determined at creation and never grows. + * + * File layout (per signal): + * Offset Size Field + * 0 4 Magic: "SHR1" + * 4 4 uint32 version (1) + * 8 4 uint32 capacity (max pairs) + * 12 4 uint32 head (next write position, 0-based, wraps) + * 16 4 uint32 count (valid entries, <= capacity) + * 20 4 uint32 decimation + * 24 8 float64 tOldest + * 32 8 float64 tNewest + * 40 24 reserved (pad to 64 bytes) + * 64 ... data: capacity × 16 bytes (8 time + 8 value) + * + * Thread safety: WriteTick() is called from the push loop only. + * ReadRange() uses pread() and is safe for concurrent reads from WS threads. + */ + +#ifndef STREAMHUB_HISTORY_WRITER_H_ +#define STREAMHUB_HISTORY_WRITER_H_ + +#include "CompilerTypes.h" +#include "StreamString.h" +#include "StructuredDataI.h" +#include "UDPSourceSession.h" + +namespace StreamHub { + +using MARTe::uint8; +using MARTe::uint16; +using MARTe::uint32; +using MARTe::uint64; +using MARTe::float64; +using MARTe::StreamString; +using MARTe::StructuredDataI; + +/** Maximum number of sessions (matches kMaxSessions). */ +static const uint32 kHistMaxSessions = 32u; + +/** Header size in bytes. */ +static const uint32 kHistHeaderSize = 64u; + +/** Magic bytes. */ +static const char kHistMagic[4] = {'S','H','R','1'}; + +class HistoryWriter { +public: + + HistoryWriter(); + ~HistoryWriter(); + + /** + * @brief Parse +History block from StreamHub config. + * + * Expected keys inside a +History node: + * Directory (string, required) + * DurationHours (float64, default 1) + * Decimation (uint32, default 1) + * FlushIntervalSec (uint32, default 5) + * MinDiskFreeMB (uint32, default 500) + */ + bool Initialise(StructuredDataI &cfg); + + /** @return true if history storage is enabled and initialised. */ + bool IsEnabled() const { return enabled_; } + + /** + * @brief Called when a source becomes configured. + * Creates / reopens per-signal .shist files. + */ + void OnSourceConfigured(uint32 sessionIdx, const char *sourceId, + UDPSourceSession &sess); + + /** + * @brief Called from the push loop to write new samples for one session. + * Uses per-signal ReadSince cursors. Applies decimation. + */ + void WriteTick(uint32 sessionIdx, UDPSourceSession &sess); + + /** + * @brief Flush in-memory header state to disk (periodic). + */ + void FlushHeaders(); + + /** + * @brief Read a time range for one signal from disk. + * @return Number of (t,v) pairs written to tOut/vOut. + */ + uint32 ReadRange(const char *sourceId, const char *signalName, + float64 t0, float64 t1, + float64 *tOut, float64 *vOut, uint32 maxOut) const; + + /** + * @brief Get the time range stored on disk for a signal. + * @return true if the signal has history data. + */ + bool GetTimeRange(const char *sourceId, const char *signalName, + float64 &t0, float64 &t1) const; + + /** + * @brief Build a JSON fragment for historyInfo response. + * Writes into buf at offset, growing via the provided append function. + */ + void AppendInfoJSON(char *&buf, uint32 &off, uint32 &cap) const; + + /** @brief Check available disk space. */ + bool HasSufficientDisk() const; + + /** @brief Duration in hours. */ + float64 DurationHours() const { return durationHours_; } + + /** @brief Decimation factor. */ + uint32 Decimation() const { return decimation_; } + +private: + + struct SignalFile { + int fd; /**< File descriptor (-1 = not open) */ + uint32 capacity; /**< Total capacity in pairs */ + uint32 head; /**< Next write position */ + uint32 count; /**< Valid entries */ + float64 tOldest; + float64 tNewest; + uint64 readCursor; /**< For ReadSince from ring buffer */ + uint64 decimCounter; /**< Decimation sample counter */ + bool headerDirty; /**< True if header needs flushing */ + char sourceId[64]; /**< Source ID for lookup */ + char signalName[64]; /**< Signal name for lookup */ + }; + + /** Per-signal scratch for reading from rings. */ + static const uint32 kReadScratch = 65536u; + float64 *scratchT_; + float64 *scratchV_; + + /** Maximum signals tracked per session for history. */ + static const uint32 kHistMaxSignals = 64u; + + /** Heap-allocated to keep StreamHub object size manageable. */ + SignalFile (*files_)[kHistMaxSignals]; /**< [kHistMaxSessions][kHistMaxSignals] */ + bool (*fileActive_)[kHistMaxSignals]; /**< [kHistMaxSessions][kHistMaxSignals] */ + + bool enabled_; + char directory_[512]; + float64 durationHours_; + uint32 decimation_; + uint32 flushIntervalSec_; + uint32 minDiskFreeMB_; + bool diskLow_; /**< True when disk space is below threshold */ + + /** Create or reopen a .shist file for one signal. */ + bool OpenSignalFile(uint32 sessionIdx, uint32 sigIdx, + const char *sourceId, const char *signalName, + float64 estimatedRateHz); + + /** Write a batch of (t,v) pairs to a signal file. */ + void WritePairs(SignalFile &sf, const float64 *t, const float64 *v, + uint32 n); + + /** Flush one signal file header to disk. */ + void FlushHeader(SignalFile &sf); + + /** Find a signal file by sourceId:signalName. */ + const SignalFile* FindFile(const char *sourceId, const char *signalName) const; +}; + +} /* namespace StreamHub */ + +#endif /* STREAMHUB_HISTORY_WRITER_H_ */ diff --git a/Source/Applications/StreamHub/Makefile.inc b/Source/Applications/StreamHub/Makefile.inc index c922703..3ff8c56 100644 --- a/Source/Applications/StreamHub/Makefile.inc +++ b/Source/Applications/StreamHub/Makefile.inc @@ -1,4 +1,4 @@ -OBJSX = StreamHub.x UDPSourceSession.x WSServer.x TriggerEngine.x +OBJSX = StreamHub.x UDPSourceSession.x WSServer.x TriggerEngine.x HistoryWriter.x PACKAGE = ROOT_DIR = ../../.. diff --git a/Source/Applications/StreamHub/StreamHub.cpp b/Source/Applications/StreamHub/StreamHub.cpp index 242c8d7..229aa4e 100644 --- a/Source/Applications/StreamHub/StreamHub.cpp +++ b/Source/Applications/StreamHub/StreamHub.cpp @@ -130,6 +130,13 @@ bool StreamHub::Initialise(StructuredDataI &cfg) { StreamString sf; if (cfg.Read("SourcesFile", sf)) { sourcesFile_ = sf; } + /* Parse +History block (optional). + * StandardParser stores the '+' prefix in the node name, so we try both. */ + if (cfg.MoveRelative("+History") || cfg.MoveRelative("History")) { + history_.Initialise(cfg); + cfg.MoveToAncestor(1u); + } + /* Allocate scratch buffers */ pushBuf_ = new uint8[kPushBufSize]; lttbT_ = new float64[maxPushPoints_]; @@ -248,6 +255,16 @@ bool StreamHub::Run() { PushStats(); } + /* History: flush headers periodically */ + if (history_.IsEnabled()) { + uint32 histFlushDiv = history_.Decimation() > 0u + ? pushRateHz_ * 5u : pushRateHz_ * 5u; /* every 5s */ + if (histFlushDiv == 0u) { histFlushDiv = 1u; } + if ((tickCount_ % histFlushDiv) == 0u) { + history_.FlushHeaders(); + } + } + tickCount_++; /* Sleep for remainder of period */ @@ -305,6 +322,12 @@ void StreamHub::PushData() { } BroadcastSources(); BroadcastConfig(i); + + /* Open history files for this source */ + if (history_.IsEnabled()) { + StreamString sid = sessions_[i].GetId(); + history_.OnSourceConfigured(i, sid.Buffer(), sessions_[i]); + } } /* Always serialize (advancing the per-signal cursors), even with no @@ -313,6 +336,11 @@ void StreamHub::PushData() { if (frameLen > 0u) { wsServer_.BroadcastBinary(pushBuf_, frameLen); } + + /* Write to disk history (separate read cursors, own decimation) */ + if (history_.IsEnabled()) { + history_.WriteTick(i, sessions_[i]); + } } } @@ -607,6 +635,19 @@ void StreamHub::OnWSClientConnected() { /* Let the new client render the current trigger badge/buttons. */ BroadcastTriggerState(); + + /* Inform the new client about history availability. */ + if (history_.IsEnabled()) { + /* Broadcast to all (simple; no unicast-on-connect path for broadcast). */ + uint32 cap2 = 8192u; + char *hbuf = new char[cap2]; + uint32 hoff = 0u; + JsonAppendf(hbuf, hoff, cap2, "{\"type\":\"historyInfo\","); + history_.AppendInfoJSON(hbuf, hoff, cap2); + JsonAppendf(hbuf, hoff, cap2, "}"); + wsServer_.BroadcastText(hbuf, hoff); + delete[] hbuf; + } } void StreamHub::OnWSClientDisconnected() { @@ -631,6 +672,8 @@ void StreamHub::OnWSCommand(const char *json, uint32 /*len*/, uint32 slotIdx) { else if (strcmp(type, "trigStop") == 0) { HandleTrigStop(json); } else if (strcmp(type, "setTrigger") == 0) { HandleSetTrigger(json); } else if (strcmp(type, "zoom") == 0) { HandleZoom(json, slotIdx); } + else if (strcmp(type, "historyZoom") == 0) { HandleHistoryZoom(json, slotIdx); } + else if (strcmp(type, "historyInfo") == 0) { HandleHistoryInfo(slotIdx); } else if (strcmp(type, "setMaxPoints") == 0) { HandleSetMaxPoints(json); } else if (strcmp(type, "ping") == 0) { HandlePing(slotIdx); } } @@ -1225,6 +1268,113 @@ void StreamHub::HandleZoom(const char *json, uint32 slotIdx) { delete[] tDec; delete[] vDec; } +void StreamHub::HandleHistoryZoom(const char *json, uint32 slotIdx) { + if (!history_.IsEnabled()) { + const char *msg = "{\"type\":\"historyZoom\",\"error\":\"history not enabled\"}"; + wsServer_.SendText(slotIdx, msg, static_cast(strlen(msg))); + return; + } + + float64 t0 = 0.0, t1 = 0.0; + float64 nF = 2400.0; + float64 reqIdF = 0.0; + + JsonGetFloat(json, "t0", t0); + JsonGetFloat(json, "t1", t1); + JsonGetFloat(json, "reqId", reqIdF); + const bool haveN = JsonGetFloat(json, "n", nF); + + uint32 maxOut = 2400u; + if (haveN) { + if (nF <= 0.0) { maxOut = 0u; } + else if (nF < 10.0) { maxOut = 2400u; } + else { maxOut = static_cast(nF); } + } + + /* Parse comma-separated full keys "src:sig" */ + static const uint32 kKeysBuf = 8192u; + char *keys = new char[kKeysBuf]; + keys[0] = '\0'; + JsonGetString(json, "signals", keys, kKeysBuf); + + /* Allocate scratch for the largest possible read */ + const uint32 scratchCap = 2000000u; /* 2M pts should be enough */ + float64 *tRaw = new float64[scratchCap]; + float64 *vRaw = new float64[scratchCap]; + float64 *tDec = (maxOut > 0u) ? new float64[maxOut] : static_cast(0); + float64 *vDec = (maxOut > 0u) ? new float64[maxOut] : static_cast(0); + + uint32 cap = 65536u; + char *buf = new char[cap]; + uint32 off = 0u; + + JsonAppendf(buf, off, cap, "{\"type\":\"historyZoom\",\"reqId\":%u,\"signals\":{", + static_cast(reqIdF)); + + bool first = true; + char *tok = keys; + while ((tok != static_cast(0)) && (*tok != '\0') && (t1 > t0)) { + char *next = strchr(tok, ','); + if (next != static_cast(0)) { *next = '\0'; next++; } + while (*tok == ' ') { tok++; } + + char *colon = strchr(tok, ':'); + if (colon == static_cast(0)) { tok = next; continue; } + *colon = '\0'; + const char *srcId = tok; + const char *sigName = colon + 1; + + uint32 nRaw = history_.ReadRange(srcId, sigName, t0, t1, + tRaw, vRaw, scratchCap); + if (nRaw > 0u) { + const float64 *tOut = tRaw; + const float64 *vOut = vRaw; + uint32 nOut = nRaw; + if ((maxOut > 0u) && (nRaw > maxOut)) { + nOut = LTTBDecimate(tRaw, vRaw, nRaw, tDec, vDec, maxOut); + tOut = tDec; + vOut = vDec; + } + + JsonAppendf(buf, off, cap, "%s\"%s:%s\":{\"t\":[", + (first ? "" : ","), srcId, sigName); + first = false; + for (uint32 p = 0u; p < nOut; p++) { + JsonAppendf(buf, off, cap, "%s%.17g", + (p > 0u ? "," : ""), tOut[p]); + } + JsonAppendf(buf, off, cap, "],\"v\":["); + for (uint32 p = 0u; p < nOut; p++) { + JsonAppendf(buf, off, cap, "%s%.9g", + (p > 0u ? "," : ""), vOut[p]); + } + JsonAppendf(buf, off, cap, "]}"); + } + tok = next; + } + + JsonAppendf(buf, off, cap, "}}"); + wsServer_.SendText(slotIdx, buf, off); + + delete[] buf; + delete[] keys; + delete[] tRaw; delete[] vRaw; + delete[] tDec; delete[] vDec; +} + +void StreamHub::HandleHistoryInfo(uint32 slotIdx) { + uint32 cap = 8192u; + char *buf = new char[cap]; + uint32 off = 0u; + + JsonAppendf(buf, off, cap, "{\"type\":\"historyInfo\","); + history_.AppendInfoJSON(buf, off, cap); + JsonAppendf(buf, off, cap, "}"); + + wsServer_.SendText(slotIdx, buf, off); + delete[] buf; +} + void StreamHub::HandleSetMaxPoints(const char *json) { float64 mpF = static_cast(maxPoints_); JsonGetFloat(json, "maxPoints", mpF); diff --git a/Source/Applications/StreamHub/StreamHub.h b/Source/Applications/StreamHub/StreamHub.h index a96c1e7..7087f9c 100644 --- a/Source/Applications/StreamHub/StreamHub.h +++ b/Source/Applications/StreamHub/StreamHub.h @@ -28,6 +28,7 @@ #include "UDPSourceSession.h" #include "WSServer.h" #include "TriggerEngine.h" +#include "HistoryWriter.h" namespace StreamHub { @@ -140,6 +141,8 @@ private: void HandleTrigStop(const char *json); void HandleSetTrigger(const char *json); void HandleZoom(const char *json, uint32 slotIdx); + void HandleHistoryZoom(const char *json, uint32 slotIdx); + void HandleHistoryInfo(uint32 slotIdx); void HandleSetMaxPoints(const char *json); void HandlePing(uint32 slotIdx); @@ -186,6 +189,7 @@ private: WSServer wsServer_; TriggerEngine trigger_; + HistoryWriter history_; /* Configuration */ uint16 wsPort_; diff --git a/Test/Configurations/streamhub_demo.cfg b/Test/Configurations/streamhub_demo.cfg index fdad962..43fd885 100644 --- a/Test/Configurations/streamhub_demo.cfg +++ b/Test/Configurations/streamhub_demo.cfg @@ -2,19 +2,18 @@ * streamhub_demo.cfg — MARTe2 app for the StreamHub demo. * * Three real-time threads, each feeding a UDPStreamer instance. - * No DebugService — monitoring is done via StreamHub + WebSocket clients. * - * Thread1 1 kHz — scalar signals → UDPStreamer port 44500 - * Counter, Time, Sine1 (1 Hz, 5 V), Sine2 (0.3 Hz, 3 V) - * Accumulate mode, multicast group 239.0.0.1 data port 44503 + * Thread1 1 kHz — 2 scalar sines @ 1 ksps (Accumulate mode) + * Sine1 (1 Hz, 5 V, phase 0) + * Sine2 (0.3 Hz, 3 V, phase 60 deg) * - * Thread2 5 kHz — packed arrays → UDPStreamer port 44501 - * Ch1 float32[1000] @ 1 kHz (TimeMode = FirstSample) - * Ch2 float32[1000] @ 10 kHz (TimeMode = LastSample) + * Thread2 1 kHz — 2 array sines @ 1 Msps (1000 elem × 1 kHz) + * Ch1 (1 kHz, 1 V, phase 0) + * Ch2 (5 kHz, 0.5 V, phase 90 deg) * - * Thread3 5 kHz — FullArray + per-sample timestamps → UDPStreamer port 44502 - * Ch3 float32[1000] @ 3 kHz (TimeMode = FullArray) - * Ch4 float32[1000] @ 500 Hz (TimeMode = FullArray) + * Thread3 5 kHz — 2 array sines @ 5 Msps (1000 elem × 5 kHz) + * Ch3 (10 kHz, 2 V, phase 0) + * Ch4 (50 kHz, 1.5 V, phase 45 deg) */ $App = { @@ -23,7 +22,7 @@ $App = { +Functions = { Class = ReferenceContainer - // ── Thread1: 1 kHz scalar signals ──────────────────────────────────── + // ── Thread1: 1 kHz scalar sines (1 ksps) ──────────────────────────── +TimerGAM = { Class = IOGAM @@ -84,15 +83,15 @@ $App = { } } - // ── Thread2: 5 kHz packed arrays (FirstSample / LastSample) ────────── + // ── Thread2: 1 kHz, 1000-element arrays → 1 Msps ─────────────────── - +FastTimerGAM = { + +MedTimerGAM = { Class = IOGAM InputSignals = { Time = { - DataSource = FastTimer + DataSource = MedTimer Type = uint32 - Frequency = 5000 + Frequency = 1000 } } OutputSignals = { @@ -106,7 +105,7 @@ $App = { Amplitude = 1.0 Phase = 0.0 Offset = 0.0 - SamplingRate = 5000000.0 + SamplingRate = 1000000.0 OutputSignals = { Ch1 = { DataSource = DDB2 @@ -119,11 +118,11 @@ $App = { +SineGAM4 = { Class = SineArrayGAM - Frequency = 10000.0 + Frequency = 5000.0 Amplitude = 0.5 Phase = 1.5708 Offset = 0.0 - SamplingRate = 5000000.0 + SamplingRate = 1000000.0 OutputSignals = { Ch2 = { DataSource = DDB2 @@ -134,27 +133,44 @@ $App = { } } - +StreamerGAM2 = { - Class = IOGAM + +TimeArrayGAM1 = { + Class = TimeArrayGAM + SamplingRate = 1000000.0 + Anchor = FirstSample InputSignals = { - Time = { DataSource = DDB2 Type = uint32 } - Ch1 = { DataSource = DDB2 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } - Ch2 = { DataSource = DDB2 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } + Time = { DataSource = DDB2 Type = uint32 } } OutputSignals = { - Time = { DataSource = Streamer2 Type = uint32 } - Ch1 = { DataSource = Streamer2 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } - Ch2 = { DataSource = Streamer2 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } + TimeArray1 = { + DataSource = DDB2 + Type = uint64 + NumberOfDimensions = 1 + NumberOfElements = 1000 + } } } - // ── Thread3: 5 kHz FullArray with per-sample timestamps ────────────── + +StreamerGAM2 = { + Class = IOGAM + InputSignals = { + TimeArray1 = { DataSource = DDB2 Type = uint64 NumberOfDimensions = 1 NumberOfElements = 1000 } + Ch1 = { DataSource = DDB2 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } + Ch2 = { DataSource = DDB2 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } + } + OutputSignals = { + TimeArray1 = { DataSource = Streamer2 Type = uint64 NumberOfDimensions = 1 NumberOfElements = 1000 } + Ch1 = { DataSource = Streamer2 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } + Ch2 = { DataSource = Streamer2 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } + } + } - +FullArrTimerGAM = { + // ── Thread3: 5 kHz, 1000-element arrays → 5 Msps ─────────────────── + + +FastTimerGAM = { Class = IOGAM InputSignals = { Time = { - DataSource = FullArrTimer + DataSource = FastTimer Type = uint32 Frequency = 5000 } @@ -166,7 +182,7 @@ $App = { +SineGAM5 = { Class = SineArrayGAM - Frequency = 3000.0 + Frequency = 10000.0 Amplitude = 2.0 Phase = 0.0 Offset = 0.0 @@ -183,8 +199,8 @@ $App = { +SineGAM6 = { Class = SineArrayGAM - Frequency = 500.0 - Amplitude = 3.0 + Frequency = 50000.0 + Amplitude = 1.5 Phase = 0.7854 Offset = 0.0 SamplingRate = 5000000.0 @@ -198,7 +214,7 @@ $App = { } } - +TimeArrayGAM1 = { + +TimeArrayGAM2 = { Class = TimeArrayGAM SamplingRate = 5000000.0 Anchor = FirstSample @@ -206,7 +222,7 @@ $App = { Time = { DataSource = DDB3 Type = uint32 } } OutputSignals = { - TimeArray = { + TimeArray2 = { DataSource = DDB3 Type = uint64 NumberOfDimensions = 1 @@ -218,14 +234,14 @@ $App = { +StreamerGAM3 = { Class = IOGAM InputSignals = { - TimeArray = { DataSource = DDB3 Type = uint64 NumberOfDimensions = 1 NumberOfElements = 1000 } - Ch3 = { DataSource = DDB3 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } - Ch4 = { DataSource = DDB3 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } + TimeArray2 = { DataSource = DDB3 Type = uint64 NumberOfDimensions = 1 NumberOfElements = 1000 } + Ch3 = { DataSource = DDB3 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } + Ch4 = { DataSource = DDB3 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } } OutputSignals = { - TimeArray = { DataSource = Streamer3 Type = uint64 NumberOfDimensions = 1 NumberOfElements = 1000 } - Ch3 = { DataSource = Streamer3 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } - Ch4 = { DataSource = Streamer3 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } + TimeArray2 = { DataSource = Streamer3 Type = uint64 NumberOfDimensions = 1 NumberOfElements = 1000 } + Ch3 = { DataSource = Streamer3 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } + Ch4 = { DataSource = Streamer3 Type = float32 NumberOfDimensions = 1 NumberOfElements = 1000 } } } } @@ -247,7 +263,7 @@ $App = { } } - +FastTimer = { + +MedTimer = { Class = LinuxTimer SleepNature = "Default" Signals = { @@ -256,7 +272,7 @@ $App = { } } - +FullArrTimer = { + +FastTimer = { Class = LinuxTimer SleepNature = "Default" Signals = { @@ -286,18 +302,21 @@ $App = { Port = 44501 MaxPayloadSize = 1400 PublishingMode = "Decimate" - Ratio = 50 + Ratio = 10 Signals = { - Time = { Type = uint32 Unit = "us" } - Ch1 = { - Type = float32 Unit = "V" + TimeArray1 = { + Type = uint64 Unit = "ns" NumberOfDimensions = 1 NumberOfElements = 1000 - TimeMode = FirstSample TimeSignal = Time SamplingRate = 5000000.0 } - Ch2 = { + Ch1 = { Type = float32 Unit = "V" NumberOfDimensions = 1 NumberOfElements = 1000 - TimeMode = LastSample TimeSignal = Time SamplingRate = 5000000.0 + TimeMode = FullArray TimeSignal = TimeArray1 + } + Ch2 = { + Type = float32 Unit = "V" + NumberOfDimensions = 1 NumberOfElements = 1000 + TimeMode = FullArray TimeSignal = TimeArray1 } } } @@ -309,19 +328,19 @@ $App = { PublishingMode = "Decimate" Ratio = 50 Signals = { - TimeArray = { + TimeArray2 = { Type = uint64 Unit = "ns" NumberOfDimensions = 1 NumberOfElements = 1000 } Ch3 = { Type = float32 Unit = "V" NumberOfDimensions = 1 NumberOfElements = 1000 - TimeMode = FullArray TimeSignal = TimeArray + TimeMode = FullArray TimeSignal = TimeArray2 } Ch4 = { Type = float32 Unit = "V" NumberOfDimensions = 1 NumberOfElements = 1000 - TimeMode = FullArray TimeSignal = TimeArray + TimeMode = FullArray TimeSignal = TimeArray2 } } } @@ -345,13 +364,13 @@ $App = { +Thread2 = { Class = RealTimeThread CPUs = 0x2 - Functions = {FastTimerGAM SineGAM3 SineGAM4 StreamerGAM2} + Functions = {MedTimerGAM SineGAM3 SineGAM4 TimeArrayGAM1 StreamerGAM2} } +Thread3 = { Class = RealTimeThread CPUs = 0x4 - Functions = {FullArrTimerGAM SineGAM5 SineGAM6 TimeArrayGAM1 StreamerGAM3} + Functions = {FastTimerGAM SineGAM5 SineGAM6 TimeArrayGAM2 StreamerGAM3} } } } diff --git a/Test/E2E/streamhub/main.go b/Test/E2E/streamhub/main.go index bdd6cb9..082a4fe 100644 --- a/Test/E2E/streamhub/main.go +++ b/Test/E2E/streamhub/main.go @@ -67,6 +67,18 @@ type zoomPoints struct { V []float64 `json:"v"` } +type historyInfoMsg struct { + Enabled bool `json:"enabled"` + DurationHours float64 `json:"durationHours"` + Decimation uint32 `json:"decimation"` + Signals map[string]struct { + T0 float64 `json:"t0"` + T1 float64 `json:"t1"` + Count uint32 `json:"count"` + Capacity uint32 `json:"capacity"` + } `json:"signals"` +} + type event struct { Type string `json:"type"` Sources json.RawMessage `json:"sources"` @@ -187,9 +199,11 @@ type client struct { configs map[string][]signalInfo // sourceId → signals pushes []*pushFrame stats map[string]statInfo - zooms map[uint32]map[string]zoomPoints - trigSt []string // observed triggerState sequence - captures []*captureFrame + zooms map[uint32]map[string]zoomPoints + histZooms map[uint32]map[string]zoomPoints + historyInfo *historyInfoMsg + trigSt []string // observed triggerState sequence + captures []*captureFrame } func (c *client) send(v interface{}) { @@ -263,6 +277,18 @@ func (c *client) pump() { if err := json.Unmarshal(data, &body); err == nil { c.zooms[ev.ReqID] = body.Signals } + case "historyZoom": + var body struct { + Signals map[string]zoomPoints `json:"signals"` + } + if err := json.Unmarshal(data, &body); err == nil { + c.histZooms[ev.ReqID] = body.Signals + } + case "historyInfo": + var hi historyInfoMsg + if err := json.Unmarshal(data, &hi); err == nil { + c.historyInfo = &hi + } case "triggerState": c.trigSt = append(c.trigSt, ev.State) } @@ -302,10 +328,11 @@ func main() { defer ws.Close() c := &client{ - ws: ws, - deadline: time.Now().Add(*timeout), - configs: map[string][]signalInfo{}, - zooms: map[uint32]map[string]zoomPoints{}, + ws: ws, + deadline: time.Now().Add(*timeout), + configs: map[string][]signalInfo{}, + zooms: map[uint32]map[string]zoomPoints{}, + histZooms: map[uint32]map[string]zoomPoints{}, } // ── 1. sources ──────────────────────────────────────────────────────── @@ -409,6 +436,41 @@ func main() { return true }) + // ── 5b. historyInfo — check the hub broadcast it on connect ───────── + if c.historyInfo != nil && c.historyInfo.Enabled { + log.Printf("OK historyInfo: enabled, %.1fh, decimation=%d, %d signals", + c.historyInfo.DurationHours, c.historyInfo.Decimation, + len(c.historyInfo.Signals)) + + // ── 5c. historyZoom round-trip ────────────────────────────────── + const hReqID = 4243 + c.send(map[string]interface{}{ + "type": "historyZoom", "reqId": hReqID, + "t0": t0, "t1": t1, "n": 200, + "signals": zoomKey, + }) + c.waitFor(fmt.Sprintf("historyZoom reply (reqId=%d, %s)", hReqID, zoomKey), + 10*time.Second, func() bool { + sigs, ok := c.histZooms[hReqID] + if !ok { + return false + } + pts, ok := sigs[zoomKey] + if !ok || len(pts.T) < 1 { + // History data may still be sparse right after startup + return true + } + for _, ht := range pts.T { + if ht < t0-1e-6 || ht > t1+1e-6 { + fail("historyZoom point outside range: t=%.9f not in [%.9f,%.9f]", ht, t0, t1) + } + } + return true + }) + } else { + log.Println(" (history not enabled — skipping historyZoom test)") + } + // ── 6. trigger: arm → capture ──────────────────────────────────────── // Trigger on an *oscillating* signal at its mean observed value: a // monotonic ramp (counter, time array) crosses its past mean only once, diff --git a/Test/E2E/streamhub/streamhub-e2e b/Test/E2E/streamhub/streamhub-e2e index 9ab8b80..ca78207 100755 Binary files a/Test/E2E/streamhub/streamhub-e2e and b/Test/E2E/streamhub/streamhub-e2e differ diff --git a/run_e2e_test.sh b/run_e2e_test.sh index b74e3aa..e0c1514 100755 --- a/run_e2e_test.sh +++ b/run_e2e_test.sh @@ -80,21 +80,28 @@ Hub = { PushRate = 30 RingTemporal = 1000000 RingScalar = 100000 + +History = { + Directory = "/tmp/streamhub_e2e_history" + DurationHours = 0.1 + Decimation = 10 + FlushIntervalSec = 2 + MinDiskFreeMB = 100 + } Sources = { scalar = { - Label = "Scalar Signals (1 kHz)" + Label = "Scalar Sines (1 ksps)" Addr = "127.0.0.1" Port = 44500 MulticastGroup = "239.0.0.1" DataPort = 44503 } - arrays1 = { - Label = "Array Signals First/Last (5 kHz)" + med = { + Label = "1 Msps Sines (Ch1 1kHz, Ch2 5kHz)" Addr = "127.0.0.1" Port = 44501 } - arrays2 = { - Label = "Array Signals FullArray (5 kHz)" + fast = { + Label = "5 Msps Sines (Ch3 10kHz, Ch4 50kHz)" Addr = "127.0.0.1" Port = 44502 } diff --git a/run_streamhub.sh b/run_streamhub.sh index 7ba4443..5810aca 100755 --- a/run_streamhub.sh +++ b/run_streamhub.sh @@ -125,10 +125,10 @@ cat > "$HUB_CFG" <