diff --git a/Client/WebUI/hub.go b/Client/WebUI/hub.go index bddab8c..68c3ea8 100644 --- a/Client/WebUI/hub.go +++ b/Client/WebUI/hub.go @@ -10,6 +10,7 @@ import ( "strings" "sync" "time" + "unsafe" "github.com/gorilla/websocket" ) @@ -163,6 +164,12 @@ type Hub struct { ringsMu sync.RWMutex rings map[string]*sigRing // "sourceId:signalKey" → ring + // lastZoomAt tracks the last time a zoom request was served. + // Ring buffer writes are skipped when no zoom has been requested + // in the last 10 s, saving substantial CPU on LTTB + ring writes. + lastZoomAt time.Time + zoomAtMu sync.Mutex + statsMu sync.RWMutex statsMap map[string]*SourceStat } @@ -189,9 +196,17 @@ func (h *Hub) getRing(key string) *sigRing { return rb } -// HandleZoom serves GET /api/zoom?t0=...&t1=...&n=...&signals=key1,key2 -// It reads from the ring buffers (safe for concurrent access) and returns -// LTTB-decimated signal data for the requested time range. +// shouldWriteRing returns true if zoom was requested within the last 10 seconds. +// When false, the hot path can skip LTTB decimation for the ring buffer entirely. +func (h *Hub) shouldWriteRing() bool { + h.zoomAtMu.Lock() + ok := time.Since(h.lastZoomAt) < 10*time.Second + h.zoomAtMu.Unlock() + return ok +} + +// HandleZoom serves GET /api/zoom?... It also records the access time +// so the ring buffer knows zoom is active and worth populating. func (h *Hub) HandleZoom(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() t0, err0 := strconv.ParseFloat(q.Get("t0"), 64) @@ -214,6 +229,15 @@ func (h *Hub) HandleZoom(w http.ResponseWriter, r *http.Request) { } } + // Record zoom access time so ring writes stay active. + // Skip n=0 requests (full-data exports / trigger snapshots) — only + // interactive zooms should keep the ring buffer populated. + if n > 0 { + h.zoomAtMu.Lock() + h.lastZoomAt = time.Now() + h.zoomAtMu.Unlock() + } + keys := strings.Split(q.Get("signals"), ",") // Collect ring references under a brief RLock. @@ -506,13 +530,31 @@ func (h *Hub) Run() { } } +// float64ToBytes reinterprets a []float64 as []byte without copying. +// The caller must ensure the slice is not modified while the byte view is in use. +func float64ToBytes(f []float64) []byte { + if len(f) == 0 { + return nil + } + return unsafe.Slice((*byte)(unsafe.Pointer(&f[0])), len(f)*8) +} + +// writeFloat64s encodes a []float64 as little-endian bytes into buf at offset +// and returns the new offset. Uses unsafe to avoid per-element PutUint64 calls. +func writeFloat64s(buf []byte, off int, f []float64) int { + copy(buf[off:], float64ToBytes(f)) + return off + len(f)*8 +} + // ─── Data serialisation ─────────────────────────────────────────────────────── // maxPushPoints is the LTTB target for data pushed over WebSocket per 30 Hz tick. -// At 2 k pts/tick × 30 Hz = 60 k pts/s; the 600 k frontend buffer covers ~10 s. -// Kept deliberately low so the rolling window shows plenty of history even for -// multi-MHz signals — zoom resolution comes from the ring buffer instead. -const maxPushPoints = 2_000 +// With cascaded LTTB the server selects the most visually significant pts from +// each batch; the browser accumulates ticks × pts/tick for the rolling window. +// For a 1 200 px plot showing a 5 s window at 30 Hz: 2×1200/(5×30) ≈ 16 pts/tick +// needed. 50 gives 4× headroom: 50×30×5 = 7 500 pts → trivial 120 KB buffer. +// Zoom resolution is unaffected — the ring buffer (maxRingPoints) serves /api/zoom. +const maxPushPoints = 50 // maxRingPoints is the LTTB target written into the ring buffer per tick. // At 5 Msps / 5 kHz packet rate ≈ 167 k raw samples/tick → LTTB to 20 k → @@ -800,6 +842,7 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []DataS sigs := src.signals pfx := src.id + ":" + writeRing := h.shouldWriteRing() // ---- Phase 1: collect (t,v) for each signal (same logic as JSON path) ---- @@ -864,10 +907,12 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []DataS allV = append(allV, vals[k]) } } - // Write hi-res LTTB data to ring. - ringT, ringV := lttbDecimate(allT, allV, maxRingPoints) - if rb := h.getRing(pfx + sig.Name); rb != nil { - rb.write(ringT, ringV) + // Write hi-res LTTB data to ring (only if zoom is active). + if writeRing { + ringT, ringV := lttbDecimate(allT, allV, maxRingPoints) + if rb := h.getRing(pfx + sig.Name); rb != nil { + rb.write(ringT, ringV) + } } // Decimate for push. decimT, decimV := lttbDecimate(allT, allV, maxPushPoints) @@ -930,8 +975,10 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []DataS ts = append(ts, float64(s.WallTime.UnixNano())/1e9) vs = append(vs, vals[0]) } - if rb := h.getRing(pfx + sig.Name); rb != nil { - rb.write(ts, vs) + if writeRing { + if rb := h.getRing(pfx + sig.Name); rb != nil { + rb.write(ts, vs) + } } pairs[sig.Name] = pairBuf{t: ts, v: vs} @@ -948,19 +995,21 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []DataS ts = append(ts, float64(s.WallTime.UnixNano())/1e9) vs = append(vs, vals[i]) } - if rb := h.getRing(pfx + key); rb != nil { - rb.write(ts, vs) + if writeRing { + if rb := h.getRing(pfx + key); rb != nil { + rb.write(ts, vs) + } } pairs[key] = pairBuf{t: ts, v: vs} } } } - // ---- Phase 2: compute total size for pre-allocation ---- + // ---- Phase 2: compute total size and serialize ---- totalSize := 1 + 1 + len(src.id) + 4 // version + srcIdLen + srcId + numSigs for key, p := range pairs { - totalSize += 2 + len(key) + 4 // keyLen + key + pairCount - totalSize += len(p.t) * 16 // t + v, each float64 = 8 bytes + totalSize += 2 + len(key) + 4 // keyLen + key + pairCount + totalSize += len(p.t)*8 + len(p.v)*8 // t + v float64 data } buf := make([]byte, totalSize) @@ -978,14 +1027,8 @@ func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []DataS off += len(key) binary.LittleEndian.PutUint32(buf[off:], uint32(len(p.t))) off += 4 - for i := 0; i < len(p.t); i++ { - binary.LittleEndian.PutUint64(buf[off:], math.Float64bits(p.t[i])) - off += 8 - } - for i := 0; i < len(p.v); i++ { - binary.LittleEndian.PutUint64(buf[off:], math.Float64bits(p.v[i])) - off += 8 - } + off = writeFloat64s(buf, off, p.t) + off = writeFloat64s(buf, off, p.v) } return buf diff --git a/Client/WebUI/static/app.js b/Client/WebUI/static/app.js index df0e311..3cde343 100644 --- a/Client/WebUI/static/app.js +++ b/Client/WebUI/static/app.js @@ -3,7 +3,10 @@ Constants ════════════════════════════════════════════════════════════════ */ const DEFAULT_CAP = 10_000; -const TEMPORAL_CAP = 600_000; +// Temporal signals receive 50 pts/tick × 30 Hz = 1 500 pts/s from the server. +// 50 000 cap → ~33 s of rolling history; well beyond the max supported window. +// Zoom uses /api/zoom (server ring buffer) so the cap only affects rolling display. +const TEMPORAL_CAP = 50_000; const LTTB_MIN = 200; // never decimate below this many points const TRACE_COLORS = [ @@ -276,8 +279,8 @@ function onBinaryData(buf) { } } - // Trigger check - if (trigVals) checkTrigger(trigVals); + // Trigger check — pass as keyed object matching checkTrigger's expected format + if (trigVals) checkTrigger({ [trig.signal]: trigVals }); if (trig.enabled && trig.collecting && (Date.now() / 1000) >= trig.trigTime + trigPostSec()) finaliseTriggerCapture(); @@ -528,6 +531,69 @@ function makeSeriesPath(key) { }; } +/* ════════════════════════════════════════════════════════════════ + LTTB Web Worker — offloads decimation off the main thread. + Cache key: "::::" + On cache-hit → render uses cached {t, v} immediately (stale-while-revalidate). + On cache-miss → render falls back to sync lttb once (first zoom render only), + then worker takes over for subsequent updates. + ════════════════════════════════════════════════════════════════ */ +const lttbCache = new Map(); // key → {t, v} +const lttbPending = new Set(); // keys currently in-flight + +let _lttbWorker = null; +try { + _lttbWorker = new Worker('lttb-worker.js'); + _lttbWorker.onmessage = function({ data: { id, t, v } }) { + lttbPending.delete(id); + lttbCache.set(id, { t, v }); + // Invalidate and redraw the owning plot. + const plotId = parseInt(id.split(':')[0], 10); + const p = plots.find(q => q.id === plotId); + if (p) { p.needsRedraw = true; } + }; + _lttbWorker.onerror = e => console.warn('[lttb-worker] error:', e); +} catch(e) { + console.warn('[lttb-worker] unavailable, using sync fallback:', e); +} + +// Submit a LTTB job to the worker (or run sync if worker unavailable). +// Returns cached {t, v} if fresh, null if a worker job was just submitted, +// or a sync result if the worker is unavailable. +function lttbAsync(cacheKey, t, v, threshold) { + const cached = lttbCache.get(cacheKey); + if (cached) return cached; // cache hit — use immediately + + if (!lttbPending.has(cacheKey)) { + lttbPending.add(cacheKey); + if (_lttbWorker) { + // Send copies so the main thread retains the originals. + const tCopy = new Float64Array(t); + const vCopy = new Float64Array(v); + _lttbWorker.postMessage({ id: cacheKey, t: tCopy, v: vCopy, threshold }, + [tCopy.buffer, vCopy.buffer]); + } else { + // Synchronous fallback (worker unavailable). + const result = lttb(t, v, threshold); + lttbCache.set(cacheKey, result); + lttbPending.delete(cacheKey); + return result; + } + } + return null; // worker job in-flight — caller should use fallback +} + +// Evict stale LTTB cache entries for a plot (call when zoom range changes). +function lttbCacheEvict(plotId) { + const prefix = plotId + ':'; + for (const k of [...lttbCache.keys()]) { + if (k.startsWith(prefix)) lttbCache.delete(k); + } + for (const k of [...lttbPending]) { + if (k.startsWith(prefix)) lttbPending.delete(k); + } +} + // LTTB decimation — O(n). Returns {t, v} unchanged when len ≤ threshold. function lttb(t, v, threshold) { const len = t.length; @@ -1147,17 +1213,30 @@ function buildLiveData(p) { if (!masterRaw || masterRaw.t.length === 0) return [new Float64Array(0), ...p.traces.map(() => new Float64Array(0))]; - // In rolling mode, Go backend already LTTB-decimated temporal signals to - // maxPushPoints (2000) and scalar points per tick are naturally limited. - // Skip JS-side LTTB entirely — just use the raw buffer data as-is. - // In zoomed mode, run pixel-adaptive LTTB for display quality. + // Decimate to pixel-adaptive point count in both rolling and zoomed modes. + // The server sends ≤2000 pts/tick but the buffer accumulates many ticks, so + // the full window slice can easily reach 100k–300k pts — far more than uPlot + // needs for a 1200px-wide canvas. Always run LTTB via the background worker + // (stale-while-revalidate: use cached result; fall back to sync on first render). + // Rolling-mode cache key uses _dataGen so the result refreshes on new data. + const targetPts = Math.max(LTTB_MIN, ((p.uplot ? p.uplot.width : p.div.clientWidth) || 600) * 2); + const cacheKey = isRolling + ? `${p.id}:${masterKey}:rolling:${_dataGen}` + : `${p.id}:${masterKey}:${t0.toFixed(6)}:${t1.toFixed(6)}:${masterRaw.t.length}`; let sharedT, masterV; - if (isRolling) { + if (masterRaw.t.length <= targetPts) { + // Data already sparse enough — use directly (no decimation needed). sharedT = masterRaw.t; masterV = masterRaw.v; } else { - const targetPts = Math.max(LTTB_MIN, ((p.uplot ? p.uplot.width : p.div.clientWidth) || 600) * 2); - const dec = lttb(masterRaw.t, masterRaw.v, targetPts); + const cached = lttbAsync(cacheKey, masterRaw.t, masterRaw.v, targetPts); + let dec; + if (cached) { + dec = cached; + } else { + // Worker job submitted — sync fallback this frame so the plot isn't blank. + dec = lttb(masterRaw.t, masterRaw.v, targetPts); + } sharedT = dec.t; masterV = dec.v; } @@ -1203,7 +1282,11 @@ function buildTrigData(p) { if (!masterRaw || masterRaw.t.length === 0) return [new Float64Array(0), ...p.traces.map(() => new Float64Array(0))]; - const dec = lttb(masterRaw.t, masterRaw.v, targetPts); + // Trig snapshot is fixed (no new data arrives after capture), so the cache + // key only needs range + data length. Worker result persists until re-arm. + const cacheKey = `${p.id}:${masterKey}:${t0.toFixed(6)}:${t1.toFixed(6)}:${masterRaw.t.length}`; + const cachedDec = lttbAsync(cacheKey, masterRaw.t, masterRaw.v, targetPts); + const dec = cachedDec || lttb(masterRaw.t, masterRaw.v, targetPts); // Convert absolute → relative seconds const sharedT = new Float64Array(dec.t.length); for (let i = 0; i < dec.t.length; i++) sharedT[i] = dec.t[i] - trigT; @@ -1252,6 +1335,9 @@ function onZoom(sourcePlotId, min, max) { }); zoomGuard = false; + // Evict stale LTTB cache entries — new zoom range needs fresh decimation. + plots.forEach(p => lttbCacheEvict(p.id)); + // Mark all plots dirty (re-slice data to the new range for full resolution) plots.forEach(p => { p.needsRedraw = true; }); @@ -1903,6 +1989,7 @@ function deletePlot(plotId) { const p = plots[idx]; if (p.ro) p.ro.disconnect(); if (p.uplot) p.uplot.destroy(); + lttbCacheEvict(plotId); plots.splice(idx, 1); document.querySelector('[data-plot-id="' + plotId + '"]').remove(); } @@ -1913,22 +2000,14 @@ function deletePlot(plotId) { let _dbgTick = 0; let _dataGen = 0; // incremented each time new data arrives function renderDirtyPlots() { + // Compute global "now" once — shared by all rolling-window plots this frame. + const globalPlotNow = getGlobalNow(); - // Diagnostic: every ~5 s print buffer state to the browser console. - // Open DevTools → Console to see timestamps and sizes. + // Lightweight diagnostic: every ~10 s log buffer count if (++_dbgTick % 300 === 0) { - const wallNow = Date.now() / 1000; - let anyData = false; - Object.entries(buffers).forEach(([k, b]) => { - if (b.size === 0) return; - anyData = true; - const newest = b.t[(b.head - 1 + b.cap) % b.cap]; - const oldest = b.t[(b.size === b.cap ? b.head : 0) % b.cap]; - const sliceLen = getBufferSlice(b).t.length; - console.log(`[buf] ${k}: size=${b.size} oldest=${oldest.toFixed(3)} newest=${newest.toFixed(3)} wallNow=${wallNow.toFixed(3)} ts-age=${(wallNow - newest).toFixed(3)}s slice(${windowSec}s)=${sliceLen}pts`); - }); - if (!anyData) console.log('[buf] all buffers empty — no data received yet'); - plots.forEach(p => console.log(`[plot${p.id}] traces=${JSON.stringify(p.traces)} xRange=${JSON.stringify(p.xRange)} needsRedraw=${p.needsRedraw}`)); + let bufCount = 0, totalPts = 0; + Object.values(buffers).forEach(b => { if (b.size > 0) { bufCount++; totalPts += b.size; } }); + if (bufCount > 0) console.log(`[buf] ${bufCount} buffers, ${totalPts} total pts`); } // Rolling-window mode: detect stale xRange (buffer has scrolled past a frozen @@ -1982,8 +2061,7 @@ function renderDirtyPlots() { if (isRolling && _dataGen === p.lastDataGen && p.uplot.data && p.uplot.data[0] && p.uplot.data[0].length > 0) { p.needsRedraw = false; zoomGuard = true; - const plotNow = computePlotNow(p); - p.uplot.setScale('x', { min: plotNow - windowSec, max: plotNow }); + p.uplot.setScale('x', { min: globalPlotNow - windowSec, max: globalPlotNow }); zoomGuard = false; return; } @@ -2040,6 +2118,8 @@ document.getElementById('btn-pause-global').addEventListener('click', () => { document.getElementById('window-select').addEventListener('change', e => { windowSec = parseFloat(e.target.value); + // Evict rolling-mode LTTB cache — window size change invalidates all cached results. + plots.forEach(p => lttbCacheEvict(p.id)); // Don't trigger redraws while in trigger mode without a snapshot if (trig.enabled && !trig.snapshot) return; plots.forEach(p => { if (!p.xRange) p.needsRedraw = true; }); diff --git a/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp b/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp index e724e18..2999760 100644 --- a/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp +++ b/Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp @@ -101,20 +101,23 @@ UDPStreamer::UDPStreamer() : MemoryDataSourceI(), EmbeddedServiceMethodBinderI(), executor(*this) { - port = UDPS_DEFAULT_PORT; - maxPayloadSize = UDPS_DEFAULT_MAX_PAYLOAD; - cpuMask = 0xFFFFFFFFu; - stackSize = THREADS_DEFAULT_STACKSIZE; - numSigs = 0u; - signalInfos = NULL_PTR(UDPStreamerSignalInfo *); - readyBuffer = NULL_PTR(uint8 *); - scratchBuffer = NULL_PTR(uint8 *); - wireBuffer = NULL_PTR(uint8 *); - totalSrcBytes = 0u; - totalWireBytes = 0u; - syncTimestamp = 0u; - clientConnected = false; - packetCounter = 0u; + port = UDPS_DEFAULT_PORT; + maxPayloadSize = UDPS_DEFAULT_MAX_PAYLOAD; + cpuMask = 0xFFFFFFFFu; + stackSize = THREADS_DEFAULT_STACKSIZE; + publishMode = UDPStreamerPublishStrict; + minRefreshRate = 0.0; + flushPeriodTicks = 0u; + numSigs = 0u; + signalInfos = NULL_PTR(UDPStreamerSignalInfo *); + readyBuffer = NULL_PTR(uint8 *); + scratchBuffer = NULL_PTR(uint8 *); + wireBuffer = NULL_PTR(uint8 *); + totalSrcBytes = 0u; + totalWireBytes = 0u; + syncTimestamp = 0u; + clientConnected = false; + packetCounter = 0u; if (!dataSem.Create()) { REPORT_ERROR(ErrorManagement::FatalError, "Could not create EventSem."); @@ -218,6 +221,40 @@ bool UDPStreamer::Initialise(StructuredDataI &data) { } } + if (ok) { + StreamString publishStr = ""; + (void) data.Read("PublishingMode", publishStr); + if ((publishStr.Size() == 0u) || (publishStr == "Strict")) { + publishMode = UDPStreamerPublishStrict; + } + else if (publishStr == "Auto") { + publishMode = UDPStreamerPublishAuto; + } + else { + REPORT_ERROR(ErrorManagement::ParametersError, + "Unknown PublishingMode '%s'. Allowed: Strict|Auto.", + publishStr.Buffer()); + ok = false; + } + } + + if (ok && (publishMode == UDPStreamerPublishAuto)) { + if (!data.Read("MinRefreshRate", minRefreshRate) || (minRefreshRate <= 0.0)) { + REPORT_ERROR(ErrorManagement::ParametersError, + "MinRefreshRate > 0 is required when PublishingMode = Auto."); + ok = false; + } + else { + /* Pre-compute the HRT tick count for one flush interval */ + float64 hrtFreq = static_cast(HighResolutionTimer::Frequency()); + flushPeriodTicks = static_cast(hrtFreq / minRefreshRate); + REPORT_ERROR(ErrorManagement::Information, + "Auto mode: MinRefreshRate=%.1f Hz, flushPeriodTicks=%llu.", + minRefreshRate, + static_cast(flushPeriodTicks)); + } + } + return ok; } @@ -611,10 +648,17 @@ bool UDPStreamer::Synchronise() { ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) { ErrorManagement::ErrorType ret = ErrorManagement::NoError; + /* nextFlushTick: HRT counter target for the next Auto-mode flush. + * Declared static so it persists across Execute() calls (the framework + * calls Execute() in a tight loop for the MainStage). */ + static uint64 nextFlushTick = 0u; + if (info.GetStage() == ExecutionInfo::StartupStage) { + nextFlushTick = HighResolutionTimer::Counter(); REPORT_ERROR(ErrorManagement::Information, - "UDPStreamer background thread started (port %u).", - static_cast(port)); + "UDPStreamer background thread started (port %u, mode %s).", + static_cast(port), + (publishMode == UDPStreamerPublishAuto) ? "Auto" : "Strict"); } if (info.GetStage() == ExecutionInfo::MainStage) { @@ -646,21 +690,44 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) { } if (dataReady && clientConnected) { - /* Copy readyBuffer → scratchBuffer under brief spinlock */ - uint64 ts = 0u; - bufMutex.FastLock(TTInfiniteWait); - (void) MemoryOperationsHelper::Copy(scratchBuffer, readyBuffer, totalSrcBytes); - ts = syncTimestamp; - bufMutex.FastUnLock(); + /* In Auto mode, only flush when the HRT flush interval has elapsed. + * This reduces send syscalls and network load by (rtHz / minRefreshRate)x + * without any changes to the RT thread or the wire protocol. + * The most-recent signal values (captured in readyBuffer) are used. */ + bool shouldSend = true; + if (publishMode == UDPStreamerPublishAuto) { + uint64 now = HighResolutionTimer::Counter(); + if (now < nextFlushTick) { + shouldSend = false; + } + else { + /* Advance deadline by one full period (keeps phase-locked). */ + nextFlushTick += flushPeriodTicks; + /* Guard against clock drift: if we're already more than one + * period behind, reset to avoid a burst of back-to-back sends. */ + if (nextFlushTick < now) { + nextFlushTick = now + flushPeriodTicks; + } + } + } - /* Serialize signal data into wireBuffer */ - QuantizeAndSerialize(scratchBuffer, ts); + if (shouldSend) { + /* Copy readyBuffer → scratchBuffer under brief spinlock */ + uint64 ts = 0u; + bufMutex.FastLock(TTInfiniteWait); + (void) MemoryOperationsHelper::Copy(scratchBuffer, readyBuffer, totalSrcBytes); + ts = syncTimestamp; + bufMutex.FastUnLock(); - /* Send (fragmented if needed) */ - packetCounter++; - if (!SendFragmented(UDPS_TYPE_DATA, packetCounter, wireBuffer, totalWireBytes)) { - REPORT_ERROR(ErrorManagement::Warning, - "Failed to send DATA packet (counter=%u).", packetCounter); + /* Serialize signal data into wireBuffer */ + QuantizeAndSerialize(scratchBuffer, ts); + + /* Send (fragmented if needed) */ + packetCounter++; + if (!SendFragmented(UDPS_TYPE_DATA, packetCounter, wireBuffer, totalWireBytes)) { + REPORT_ERROR(ErrorManagement::Warning, + "Failed to send DATA packet (counter=%u).", packetCounter); + } } } } diff --git a/Source/Components/DataSources/UDPStreamer/UDPStreamer.h b/Source/Components/DataSources/UDPStreamer/UDPStreamer.h index 50d572d..79b4976 100644 --- a/Source/Components/DataSources/UDPStreamer/UDPStreamer.h +++ b/Source/Components/DataSources/UDPStreamer/UDPStreamer.h @@ -46,6 +46,18 @@ namespace MARTe { +/** + * @brief Publishing mode for the background sender thread. + * + * - Strict: send one UDP packet on every Synchronise() call (legacy behaviour). + * - Auto: buffer successive ticks and only flush when the HRT-based flush + * interval has elapsed. Controlled by MinRefreshRate (Hz). + */ +typedef enum { + UDPStreamerPublishStrict = 0u, /**< Send on every RT cycle (default) */ + UDPStreamerPublishAuto = 1u /**< Rate-limited: flush at MinRefreshRate Hz */ +} UDPStreamerPublishMode; + /** * @brief Quantization types for float signals. */ @@ -153,6 +165,8 @@ static const uint8 UDPS_TYPECODE_UNKNOWN = 255u; * MaxPayloadSize = 1400 // Optional (default 1400); max payload bytes per UDP datagram * CPUMask = 0x2 // Optional, affinity for background thread * StackSize = 1048576 // Optional, stack size for background thread + * PublishingMode = "Auto" // Optional: "Strict" (default) | "Auto" + * MinRefreshRate = 120 // Optional (Hz); only used when PublishingMode = "Auto" * Signals = { * Time = { * Type = uint64 @@ -292,10 +306,13 @@ private: static uint8 TypeDescriptorToCode(TypeDescriptor td); /* Configuration parameters */ - uint16 port; /**< UDP server port */ - uint32 maxPayloadSize; /**< Max payload bytes per UDP packet (excluding header) */ - uint32 cpuMask; /**< Background thread CPU affinity */ - uint32 stackSize; /**< Background thread stack size */ + uint16 port; /**< UDP server port */ + uint32 maxPayloadSize; /**< Max payload bytes per UDP packet (excluding header) */ + uint32 cpuMask; /**< Background thread CPU affinity */ + uint32 stackSize; /**< Background thread stack size */ + UDPStreamerPublishMode publishMode; /**< Strict or Auto publishing mode */ + float64 minRefreshRate; /**< Minimum flush rate (Hz) for Auto mode */ + uint64 flushPeriodTicks; /**< HRT ticks per flush interval (computed from minRefreshRate) */ /* Signal metadata */ uint32 numSigs; /**< Number of signals */ diff --git a/Test/Components/DataSources/UDPStreamer/UDPStreamerGTest.cpp b/Test/Components/DataSources/UDPStreamer/UDPStreamerGTest.cpp index f6cb6ff..514e90e 100644 --- a/Test/Components/DataSources/UDPStreamer/UDPStreamerGTest.cpp +++ b/Test/Components/DataSources/UDPStreamer/UDPStreamerGTest.cpp @@ -61,6 +61,21 @@ TEST(UDPStreamerGTest, TestInitialise_ZeroStackSize) { ASSERT_TRUE(test.TestInitialise_ZeroStackSize()); } +TEST(UDPStreamerGTest, TestInitialise_AutoMode_Valid) { + UDPStreamerTest test; + ASSERT_TRUE(test.TestInitialise_AutoMode_Valid()); +} + +TEST(UDPStreamerGTest, TestInitialise_AutoMode_MissingRefreshRate) { + UDPStreamerTest test; + ASSERT_TRUE(test.TestInitialise_AutoMode_MissingRefreshRate()); +} + +TEST(UDPStreamerGTest, TestInitialise_UnknownPublishingMode) { + UDPStreamerTest test; + ASSERT_TRUE(test.TestInitialise_UnknownPublishingMode()); +} + TEST(UDPStreamerGTest, TestGetBrokerName_Output) { UDPStreamerTest test; ASSERT_TRUE(test.TestGetBrokerName_Output()); diff --git a/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp b/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp index 268fda8..7ce3836 100644 --- a/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp +++ b/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.cpp @@ -230,6 +230,41 @@ bool UDPStreamerTest::TestInitialise_ZeroStackSize() { return ok; } +bool UDPStreamerTest::TestInitialise_AutoMode_Valid() { + using namespace MARTe; + UDPStreamer ds; + ConfigurationDatabase cdb; + cdb.Write("Port", 44502u); + cdb.Write("PublishingMode", "Auto"); + cdb.Write("MinRefreshRate", 120.0); + cdb.CreateRelative("Signals"); + cdb.MoveToRoot(); + return ds.Initialise(cdb); +} + +bool UDPStreamerTest::TestInitialise_AutoMode_MissingRefreshRate() { + using namespace MARTe; + UDPStreamer ds; + ConfigurationDatabase cdb; + cdb.Write("Port", 44503u); + cdb.Write("PublishingMode", "Auto"); + /* MinRefreshRate intentionally omitted */ + cdb.CreateRelative("Signals"); + cdb.MoveToRoot(); + return !ds.Initialise(cdb); /* Must fail */ +} + +bool UDPStreamerTest::TestInitialise_UnknownPublishingMode() { + using namespace MARTe; + UDPStreamer ds; + ConfigurationDatabase cdb; + cdb.Write("Port", 44504u); + cdb.Write("PublishingMode", "Invalid"); + cdb.CreateRelative("Signals"); + cdb.MoveToRoot(); + return !ds.Initialise(cdb); /* Must fail */ +} + bool UDPStreamerTest::TestGetBrokerName_Output() { using namespace MARTe; UDPStreamer ds; @@ -1157,36 +1192,88 @@ bool UDPStreamerTest::TestIsClientConnected_InitiallyFalse() { * Fragment count at MaxPayloadSize=1400: * chunk = 1400 - 17 = 1383 B → ceil(8016 / 1383) = 6 fragments */ +/* MARTe2 StandardParser uses whitespace/newlines as delimiters. + * Semicolons are NOT statement separators — they are consumed as part of + * token values. All inline { key = val; key = val } blocks below have been + * rewritten to use one key-value pair per line. */ + #define HF_FUNCTIONS_BLOCK \ " +Functions = {\n" \ " Class = ReferenceContainer\n" \ " +Writer = {\n" \ " Class = UDPStreamerTestOutputGAM\n" \ " OutputSignals = {\n" \ - " T0 = { DataSource = Streamer; Type = uint64 }\n" \ - " Ch1 = { DataSource = Streamer; Type = int16; NumberOfElements = 1000 }\n" \ - " Ch2 = { DataSource = Streamer; Type = int16; NumberOfElements = 1000 }\n" \ - " Ch3 = { DataSource = Streamer; Type = int16; NumberOfElements = 1000 }\n" \ - " Ch4 = { DataSource = Streamer; Type = int16; NumberOfElements = 1000 }\n" \ + " T0 = {\n" \ + " DataSource = Streamer\n" \ + " Type = uint64\n" \ + " }\n" \ + " Ch1 = {\n" \ + " DataSource = Streamer\n" \ + " Type = int16\n" \ + " NumberOfElements = 1000\n" \ + " }\n" \ + " Ch2 = {\n" \ + " DataSource = Streamer\n" \ + " Type = int16\n" \ + " NumberOfElements = 1000\n" \ + " }\n" \ + " Ch3 = {\n" \ + " DataSource = Streamer\n" \ + " Type = int16\n" \ + " NumberOfElements = 1000\n" \ + " }\n" \ + " Ch4 = {\n" \ + " DataSource = Streamer\n" \ + " Type = int16\n" \ + " NumberOfElements = 1000\n" \ + " }\n" \ " }\n" \ " }\n" \ " }\n" #define HF_SIGNALS_BLOCK \ " Signals = {\n" \ - " T0 = { Type = uint64 }\n" \ - " Ch1 = { Type = int16; NumberOfDimensions = 1; NumberOfElements = 1000;\n" \ - " TimeMode = FirstSample; TimeSignal = T0; SamplingRate = 1000000.0 }\n" \ - " Ch2 = { Type = int16; NumberOfDimensions = 1; NumberOfElements = 1000;\n" \ - " TimeMode = FirstSample; TimeSignal = T0; SamplingRate = 1000000.0 }\n" \ - " Ch3 = { Type = int16; NumberOfDimensions = 1; NumberOfElements = 1000;\n" \ - " TimeMode = FirstSample; TimeSignal = T0; SamplingRate = 1000000.0 }\n" \ - " Ch4 = { Type = int16; NumberOfDimensions = 1; NumberOfElements = 1000;\n" \ - " TimeMode = FirstSample; TimeSignal = T0; SamplingRate = 1000000.0 }\n" \ + " T0 = {\n" \ + " Type = uint64\n" \ + " }\n" \ + " Ch1 = {\n" \ + " Type = int16\n" \ + " NumberOfDimensions = 1\n" \ + " NumberOfElements = 1000\n" \ + " TimeMode = FirstSample\n" \ + " TimeSignal = T0\n" \ + " SamplingRate = 1000000.0\n" \ + " }\n" \ + " Ch2 = {\n" \ + " Type = int16\n" \ + " NumberOfDimensions = 1\n" \ + " NumberOfElements = 1000\n" \ + " TimeMode = FirstSample\n" \ + " TimeSignal = T0\n" \ + " SamplingRate = 1000000.0\n" \ + " }\n" \ + " Ch3 = {\n" \ + " Type = int16\n" \ + " NumberOfDimensions = 1\n" \ + " NumberOfElements = 1000\n" \ + " TimeMode = FirstSample\n" \ + " TimeSignal = T0\n" \ + " SamplingRate = 1000000.0\n" \ + " }\n" \ + " Ch4 = {\n" \ + " Type = int16\n" \ + " NumberOfDimensions = 1\n" \ + " NumberOfElements = 1000\n" \ + " TimeMode = FirstSample\n" \ + " TimeSignal = T0\n" \ + " SamplingRate = 1000000.0\n" \ + " }\n" \ " }\n" #define HF_TAIL_BLOCK \ - " +Timings = { Class = TimingDataSource }\n" \ + " +Timings = {\n" \ + " Class = TimingDataSource\n" \ + " }\n" \ " }\n" \ " +States = {\n" \ " Class = ReferenceContainer\n" \ @@ -1194,36 +1281,57 @@ bool UDPStreamerTest::TestIsClientConnected_InitiallyFalse() { " Class = RealTimeState\n" \ " +Threads = {\n" \ " Class = ReferenceContainer\n" \ - " +Thread1 = { Class = RealTimeThread; Functions = { Writer } }\n" \ + " +Thread1 = {\n" \ + " Class = RealTimeThread\n" \ + " Functions = { Writer }\n" \ + " }\n" \ " }\n" \ " }\n" \ " }\n" \ - " +Scheduler = { Class = GAMScheduler; TimingDataSource = Timings }\n" \ + " +Scheduler = {\n" \ + " Class = GAMScheduler\n" \ + " TimingDataSource = Timings\n" \ + " }\n" \ "}\n" static const MARTe::char8 *const HF_CFG_ALLOC = - "+Test = { Class = RealTimeApplication\n" + "+Test = {\n" + " Class = RealTimeApplication\n" HF_FUNCTIONS_BLOCK - " +Data = { Class = ReferenceContainer\n" - " +Streamer = { Class = UDPStreamer; Port = 44650; MaxPayloadSize = 1400\n" + " +Data = {\n" + " Class = ReferenceContainer\n" + " +Streamer = {\n" + " Class = UDPStreamer\n" + " Port = 44650\n" + " MaxPayloadSize = 1400\n" HF_SIGNALS_BLOCK " }\n" HF_TAIL_BLOCK; static const MARTe::char8 *const HF_CFG_FRAG = - "+Test = { Class = RealTimeApplication\n" + "+Test = {\n" + " Class = RealTimeApplication\n" HF_FUNCTIONS_BLOCK - " +Data = { Class = ReferenceContainer\n" - " +Streamer = { Class = UDPStreamer; Port = 44651; MaxPayloadSize = 1400\n" + " +Data = {\n" + " Class = ReferenceContainer\n" + " +Streamer = {\n" + " Class = UDPStreamer\n" + " Port = 44651\n" + " MaxPayloadSize = 1400\n" HF_SIGNALS_BLOCK " }\n" HF_TAIL_BLOCK; static const MARTe::char8 *const HF_CFG_INTEGRITY = - "+Test = { Class = RealTimeApplication\n" + "+Test = {\n" + " Class = RealTimeApplication\n" HF_FUNCTIONS_BLOCK - " +Data = { Class = ReferenceContainer\n" - " +Streamer = { Class = UDPStreamer; Port = 44653; MaxPayloadSize = 1400\n" + " +Data = {\n" + " Class = ReferenceContainer\n" + " +Streamer = {\n" + " Class = UDPStreamer\n" + " Port = 44653\n" + " MaxPayloadSize = 1400\n" HF_SIGNALS_BLOCK " }\n" HF_TAIL_BLOCK; diff --git a/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.h b/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.h index 331a5b4..0121186 100644 --- a/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.h +++ b/Test/Components/DataSources/UDPStreamer/UDPStreamerTest.h @@ -162,6 +162,21 @@ public: */ bool TestQuantization_Uint8Clamping(); + /** + * @brief Tests PublishingMode = "Auto" with a valid MinRefreshRate. + */ + bool TestInitialise_AutoMode_Valid(); + + /** + * @brief Tests that PublishingMode = "Auto" without MinRefreshRate is rejected. + */ + bool TestInitialise_AutoMode_MissingRefreshRate(); + + /** + * @brief Tests that an unknown PublishingMode string is rejected. + */ + bool TestInitialise_UnknownPublishingMode(); + /** * @brief Tests the GetPort() accessor. */ diff --git a/Test/MARTeApp/run.sh b/Test/MARTeApp/run.sh index 854241d..1a87e62 100755 --- a/Test/MARTeApp/run.sh +++ b/Test/MARTeApp/run.sh @@ -4,14 +4,14 @@ # Usage: # ./run.sh # run MARTe2 app only # ./run.sh --webui # also start the WebUI Go client in the background +# ./run.sh --nativeui # also start the NativeUI ImGui client in the background +# ./run.sh --qtui # also start the NativeUI Qt client in the background # ./run.sh --help # show this message # -# The WebUI is started with three sources: +# The clients are started with three sources: # Streamer @ 127.0.0.1:44500 (scalar signals, PacketTime, 1 kHz) # FastStreamer @ 127.0.0.1:44501 (packed arrays, FirstSample/LastSample, 5 kHz) # FullArrStreamer @ 127.0.0.1:44502 (packed arrays, FullArray, 5 kHz) -# Additional sources and a persistent source list file (--sources-file) can -# be configured directly in the WebUI or by editing this script. # # Prerequisites: # - marte_env.sh must be present in the repo root (sets MARTe2_DIR, etc.) @@ -27,11 +27,15 @@ REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)" # ── Parse arguments ────────────────────────────────────────────────────────── START_WEBUI=0 +START_NATIVEUI=0 +START_QTUI=0 for arg in "$@"; do case "$arg" in - --webui) START_WEBUI=1 ;; + --webui) START_WEBUI=1 ;; + --nativeui) START_NATIVEUI=1 ;; + --qtui) START_QTUI=1 ;; --help|-h) - sed -n '2,15p' "$0" + sed -n '2,21p' "$0" exit 0 ;; esac @@ -84,10 +88,55 @@ WEBUI_DIR="${REPO_ROOT}/Client/WebUI" WEBUI_BIN="${WEBUI_DIR}/udpstreamer-webui" if [ "${START_WEBUI}" -eq 1 ] && [ ! -x "${WEBUI_BIN}" ]; then echo "==> Building WebUI..." - (cd "${WEBUI_DIR}" && go build -o udpstreamer-webui ./...) + (cd "${WEBUI_DIR}" && go build -o udpstreamer-webui ./...) echo "==> WebUI build done." fi +# ── Build NativeUI (ImGui) binary (if requested and not already built) ─────── +NATIVEUI_DIR="${REPO_ROOT}/Client/NativeUI" +NATIVEUI_BUILD="${NATIVEUI_DIR}/build" +NATIVEUI_BIN="${NATIVEUI_BUILD}/daq_viewer" +if [ "${START_NATIVEUI}" -eq 1 ]; then + if [ ! -x "${NATIVEUI_BIN}" ]; then + echo "==> Building NativeUI/ImGui (cmake)..." + cmake -S "${NATIVEUI_DIR}" -B "${NATIVEUI_BUILD}" -DCMAKE_BUILD_TYPE=Release -Wno-dev + cmake --build "${NATIVEUI_BUILD}" --parallel "$(nproc)" + echo "==> NativeUI/ImGui build done." + else + # Rebuild if sources are newer than the binary + if find "${NATIVEUI_DIR}/src" -name '*.cpp' -o -name '*.h' \ + | xargs ls -t 2>/dev/null | head -1 \ + | xargs -I{} test {} -nt "${NATIVEUI_BIN}" 2>/dev/null; then + echo "==> Sources changed — rebuilding NativeUI/ImGui..." + cmake --build "${NATIVEUI_BUILD}" --parallel "$(nproc)" + echo "==> NativeUI/ImGui rebuild done." + fi + fi +fi + +# ── Build NativeUI-Qt binary (if requested and not already built) ───────────── +QTUI_DIR="${REPO_ROOT}/Client/NativeUI-Qt" +QTUI_BUILD="${QTUI_DIR}/build" +QTUI_BIN="${QTUI_BUILD}/daq_viewer" +if [ "${START_QTUI}" -eq 1 ]; then + if [ ! -x "${QTUI_BIN}" ]; then + echo "==> Building NativeUI/Qt (cmake)..." + cmake -S "${QTUI_DIR}" -B "${QTUI_BUILD}" -DCMAKE_BUILD_TYPE=Release -Wno-dev + cmake --build "${QTUI_BUILD}" --parallel "$(nproc)" + echo "==> NativeUI/Qt build done." + else + # Rebuild if sources are newer than the binary + if find "${QTUI_DIR}/src" "${REPO_ROOT}/Client/NativeUI/src" \ + \( -name '*.cpp' -o -name '*.h' \) \ + | xargs ls -t 2>/dev/null | head -1 \ + | xargs -I{} test {} -nt "${QTUI_BIN}" 2>/dev/null; then + echo "==> Sources changed — rebuilding NativeUI/Qt..." + cmake --build "${QTUI_BUILD}" --parallel "$(nproc)" + echo "==> NativeUI/Qt rebuild done." + fi + fi +fi + # ── Set LD_LIBRARY_PATH ─────────────────────────────────────────────────────── COMP="${MARTe2_Components_DIR}/Build/${TARGET}/Components" @@ -105,10 +154,6 @@ ${LD_LIBRARY_PATH}" echo "==> LD_LIBRARY_PATH=${LD_LIBRARY_PATH}" # ── Create class-name symlinks for MARTe2 auto-loader ──────────────────────── -# MARTe2 looks for .so when a class is not yet registered. -# Some components bundle multiple classes into one .so (e.g. WaveformGAM.so -# contains WaveformSin, WaveformChirp, WaveformPointsDef). We need symlinks -# so dlopen(".so") succeeds. WAVEFORM_DIR="${COMP}/GAMs/WaveformGAM" for cls in WaveformSin WaveformChirp WaveformPointsDef; do target="${WAVEFORM_DIR}/${cls}.so" @@ -117,10 +162,10 @@ for cls in WaveformSin WaveformChirp WaveformPointsDef; do fi done - # ── Optionally start WebUI ──────────────────────────────────────────────────── +WEBUI_PID="" if [ "${START_WEBUI}" -eq 1 ]; then - echo "==> Starting WebUI on http://localhost:8080 (source: 127.0.0.1:44500)..." + echo "==> Starting WebUI on http://localhost:8080..." "${WEBUI_BIN}" \ --source "Streamer@127.0.0.1:44500" \ --source "FastStreamer@127.0.0.1:44501" \ @@ -130,6 +175,38 @@ if [ "${START_WEBUI}" -eq 1 ]; then echo "==> WebUI PID ${WEBUI_PID}" fi +# ── Optionally start NativeUI (ImGui) ──────────────────────────────────────── +NATIVEUI_PID="" +if [ "${START_NATIVEUI}" -eq 1 ]; then + if [ ! -x "${NATIVEUI_BIN}" ]; then + echo "ERROR: NativeUI/ImGui binary not found at ${NATIVEUI_BIN}" >&2 + exit 1 + fi + echo "==> Starting NativeUI/ImGui..." + "${NATIVEUI_BIN}" \ + "Streamer@127.0.0.1:44500" \ + "FastStreamer@127.0.0.1:44501" \ + "FullArrStreamer@127.0.0.1:44502" & + NATIVEUI_PID=$! + echo "==> NativeUI/ImGui PID ${NATIVEUI_PID}" +fi + +# ── Optionally start NativeUI-Qt ────────────────────────────────────────────── +QTUI_PID="" +if [ "${START_QTUI}" -eq 1 ]; then + if [ ! -x "${QTUI_BIN}" ]; then + echo "ERROR: NativeUI/Qt binary not found at ${QTUI_BIN}" >&2 + exit 1 + fi + echo "==> Starting NativeUI/Qt..." + "${QTUI_BIN}" \ + "Streamer@127.0.0.1:44500" \ + "FastStreamer@127.0.0.1:44501" \ + "FullArrStreamer@127.0.0.1:44502" & + QTUI_PID=$! + echo "==> NativeUI/Qt PID ${QTUI_PID}" +fi + # ── Launch MARTe2 application ───────────────────────────────────────────────── MARTE_APP="${MARTe2_DIR}/Build/${TARGET}/App/MARTeApp.ex" CFG="${SCRIPT_DIR}/TestApp.cfg" @@ -149,9 +226,9 @@ echo "" cleanup() { echo "" echo "==> Stopping..." - if [ "${START_WEBUI}" -eq 1 ] && kill -0 "${WEBUI_PID}" 2>/dev/null; then - kill "${WEBUI_PID}" - fi + [ -n "${WEBUI_PID}" ] && kill -0 "${WEBUI_PID}" 2>/dev/null && kill "${WEBUI_PID}" + [ -n "${NATIVEUI_PID}" ] && kill -0 "${NATIVEUI_PID}" 2>/dev/null && kill "${NATIVEUI_PID}" + [ -n "${QTUI_PID}" ] && kill -0 "${QTUI_PID}" 2>/dev/null && kill "${QTUI_PID}" } trap cleanup EXIT INT TERM