diff --git a/Client/WebUI/hub.go b/Client/WebUI/hub.go index 5e9edde..81b7e2f 100644 --- a/Client/WebUI/hub.go +++ b/Client/WebUI/hub.go @@ -1,6 +1,7 @@ package main import ( + "encoding/binary" "encoding/json" "log" "math" @@ -15,10 +16,15 @@ import ( // ─── WebSocket client ───────────────────────────────────────────────────────── +type wsMessage struct { + msgType int + data []byte +} + type wsClient struct { hub *Hub conn *websocket.Conn - send chan []byte + send chan wsMessage } func (c *wsClient) writePump() { @@ -34,7 +40,7 @@ func (c *wsClient) writePump() { c.conn.WriteMessage(websocket.CloseMessage, []byte{}) return } - if err := c.conn.WriteMessage(websocket.TextMessage, msg); err != nil { + if err := c.conn.WriteMessage(msg.msgType, msg.data); err != nil { return } case <-pingTicker.C: @@ -69,7 +75,7 @@ func (c *wsClient) readPump() { case "ping": resp, _ := json.Marshal(map[string]string{"type": "pong"}) select { - case c.send <- resp: + case c.send <- wsMessage{websocket.TextMessage, resp}: default: } case "addSource": @@ -167,8 +173,8 @@ func NewHub() *Hub { clients: make(map[*wsClient]bool), register: make(chan *wsClient, 8), unregister: make(chan *wsClient, 8), - broadcastCh: make(chan []byte, 64), - dataCh: make(chan taggedSample, 256), + broadcastCh: make(chan []byte, 256), + dataCh: make(chan taggedSample, 65536), // large buffer: absorbs bursts at high sample rates commandCh: make(chan hubCmd, 64), rings: make(map[string]*sigRing), statsMap: make(map[string]*SourceStat), @@ -298,7 +304,7 @@ func (h *Hub) HandleWebSocket(w http.ResponseWriter, r *http.Request) { log.Printf("ws upgrade: %v", err) return } - c := &wsClient{hub: h, conn: conn, send: make(chan []byte, 64)} + c := &wsClient{hub: h, conn: conn, send: make(chan wsMessage, 64)} h.register <- c go c.writePump() go c.readPump() @@ -322,7 +328,7 @@ func buildSourcesMsg(sm map[string]*sourceHubState) []byte { // Run is the hub's main goroutine. Must be started with go hub.Run(). func (h *Hub) Run() { - ticker := time.NewTicker(time.Second / 30) + ticker := time.NewTicker(time.Second / 20) defer ticker.Stop() statsTicker := time.NewTicker(time.Second) @@ -345,11 +351,11 @@ func (h *Hub) Run() { h.clients[c] = true // Send current state to the new client. if sourcesMsg != nil { - select { case c.send <- sourcesMsg: default: } + select { case c.send <- wsMessage{websocket.TextMessage, sourcesMsg}: default: } } for _, src := range sourcesMap { if src.configJS != nil { - select { case c.send <- src.configJS: default: } + select { case c.send <- wsMessage{websocket.TextMessage, src.configJS}: default: } } } @@ -361,7 +367,7 @@ func (h *Hub) Run() { case msg := <-h.broadcastCh: for c := range h.clients { - select { case c.send <- msg: default: } + select { case c.send <- wsMessage{websocket.TextMessage, msg}: default: } } case cmd := <-h.commandCh: @@ -473,10 +479,15 @@ func (h *Hub) Run() { pending[srcID] = pending[srcID][:0] continue } - msg := h.buildDataMessageForSource(src, samples) + msg := h.buildBinaryDataMessageForSource(src, samples) pending[srcID] = pending[srcID][:0] if msg != nil { - h.broadcast(msg) + for c := range h.clients { + select { + case c.send <- wsMessage{websocket.BinaryMessage, msg}: + default: + } + } } } @@ -763,6 +774,223 @@ func (h *Hub) buildDataMessageForSource(src *sourceHubState, batch []DataSample) return result } +// buildBinaryDataMessageForSource encodes a batch of samples as a compact binary +// frame for WebSocket binary messages. Skips the JSON overhead entirely. +// +// Wire format (little-endian): +// +// uint8 version (1) +// uint8 source ID length +// UTF-8 source ID +// uint32 number of signals +// for each signal: +// uint16 key length +// UTF-8 key (relative to source, e.g. "sigName" not "s1:sigName") +// uint32 pair count N +// float64[N] t values +// float64[N] v values +func (h *Hub) buildBinaryDataMessageForSource(src *sourceHubState, batch []DataSample) []byte { + if len(batch) == 0 { + return nil + } + if src.configSeq != src.configSeqAtCalib { + src.configSeqAtCalib = src.configSeq + src.timeSigCalib = make(map[string]float64) + } + + sigs := src.signals + pfx := src.id + ":" + + // ---- Phase 1: collect (t,v) for each signal (same logic as JSON path) ---- + + type pairBuf struct { + t, v []float64 + } + pairs := make(map[string]pairBuf, len(sigs)*2) + + for _, sig := range sigs { + n := sig.NumElements() + + switch { + case n > 1 && (sig.TimeMode == TimeModeFirstSample || sig.TimeMode == TimeModeLastSample): + hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs) + var timeSigName string + timerToSec := 1e-6 + if hasTimeSig { + ts := sigs[sig.TimeSignalIdx] + timeSigName = ts.Name + if ts.TypeCode == 6 { + timerToSec = 1e-9 + } + } + dt := 0.0 + if sig.SamplingRate > 0 { + dt = 1.0 / sig.SamplingRate + } + allT := make([]float64, 0, len(batch)*n) + allV := make([]float64, 0, len(batch)*n) + for _, s := range batch { + vals, ok := s.Values[sig.Name] + if !ok || len(vals) < n { + continue + } + var anchorTime float64 + anchorIsFirstSample := sig.TimeMode == TimeModeFirstSample + if hasTimeSig { + tVals, tOk := s.Values[timeSigName] + if tOk && len(tVals) >= 1 { + timerS := tVals[0] * timerToSec + wallT := float64(s.WallTime.UnixNano()) / 1e9 + if _, exists := src.timeSigCalib[timeSigName]; !exists { + src.timeSigCalib[timeSigName] = wallT - timerS + } + anchorTime = src.timeSigCalib[timeSigName] + timerS + } else { + anchorTime = float64(s.WallTime.UnixNano()) / 1e9 + anchorIsFirstSample = false + } + } else { + anchorTime = float64(s.WallTime.UnixNano()) / 1e9 + anchorIsFirstSample = false + } + for k := 0; k < n; k++ { + var t float64 + if anchorIsFirstSample { + t = anchorTime + float64(k)*dt + } else { + t = anchorTime - float64(n-1-k)*dt + } + allT = append(allT, t) + 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) + } + // Decimate for push. + decimT, decimV := lttbDecimate(allT, allV, maxPushPoints) + pairs[sig.Name] = pairBuf{t: decimT, v: decimV} + + case n > 1 && sig.TimeMode == TimeModeFullArray: + hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs) + var timeSigName string + timerToSec := 1e-6 + if hasTimeSig { + ts := sigs[sig.TimeSignalIdx] + timeSigName = ts.Name + if ts.TypeCode == 6 { + timerToSec = 1e-9 + } + } + allT := make([]float64, 0, len(batch)*n) + allV := make([]float64, 0, len(batch)*n) + for _, s := range batch { + vals, ok := s.Values[sig.Name] + if !ok || len(vals) < n { + continue + } + if hasTimeSig { + tVals, tOk := s.Values[timeSigName] + if tOk && len(tVals) >= n { + if _, exists := src.timeSigCalib[timeSigName]; !exists { + wallT := float64(s.WallTime.UnixNano()) / 1e9 + src.timeSigCalib[timeSigName] = wallT - tVals[0]*timerToSec + } + calib := src.timeSigCalib[timeSigName] + for k := 0; k < n; k++ { + allT = append(allT, calib+tVals[k]*timerToSec) + allV = append(allV, vals[k]) + } + continue + } + } + wallT := float64(s.WallTime.UnixNano()) / 1e9 + for k := 0; k < n; k++ { + allT = append(allT, wallT) + allV = append(allV, vals[k]) + } + } + ringT, ringV := lttbDecimate(allT, allV, maxRingPoints) + if rb := h.getRing(pfx + sig.Name); rb != nil { + rb.write(ringT, ringV) + } + decimT, decimV := lttbDecimate(allT, allV, maxPushPoints) + pairs[sig.Name] = pairBuf{t: decimT, v: decimV} + + case n == 1: + ts := make([]float64, 0, len(batch)) + vs := make([]float64, 0, len(batch)) + for _, s := range batch { + vals, ok := s.Values[sig.Name] + if !ok || len(vals) < 1 { + continue + } + 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) + } + pairs[sig.Name] = pairBuf{t: ts, v: vs} + + default: + for i := 0; i < n; i++ { + key := arrayKey(sig.Name, i) + ts := make([]float64, 0, len(batch)) + vs := make([]float64, 0, len(batch)) + for _, s := range batch { + vals, ok := s.Values[sig.Name] + if !ok || len(vals) <= i { + continue + } + 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) + } + pairs[key] = pairBuf{t: ts, v: vs} + } + } + } + + // ---- Phase 2: compute total size for pre-allocation ---- + 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 + } + + buf := make([]byte, totalSize) + buf[0] = 1 // version + buf[1] = byte(len(src.id)) + copy(buf[2:], src.id) + off := 2 + len(src.id) + binary.LittleEndian.PutUint32(buf[off:], uint32(len(pairs))) + off += 4 + + for key, p := range pairs { + binary.LittleEndian.PutUint16(buf[off:], uint16(len(key))) + off += 2 + copy(buf[off:], key) + 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 + } + } + + return buf +} + // RecordDataFragment is called by UDPClient for every incoming DATA datagram. func (h *Hub) RecordDataFragment(sourceID string, counter uint32, nBytes int, arrivalNs int64, complete bool) { h.statsMu.RLock() diff --git a/Client/WebUI/static/app.js b/Client/WebUI/static/app.js index 98c189b..a4cfe12 100644 --- a/Client/WebUI/static/app.js +++ b/Client/WebUI/static/app.js @@ -99,6 +99,7 @@ function trigPostSec() { return trig.windowSec * (100 - trig.prePercent) / 100; let ws = null, wsBackoff = 1000; function connectWS() { ws = new WebSocket('ws://' + location.host + '/ws'); + ws.binaryType = 'arraybuffer'; ws.onopen = () => { wsBackoff = 1000; setStatus('orange', 'Connected – waiting for data'); }; ws.onclose = () => { setStatus('red', 'Disconnected (reconnecting…)'); @@ -107,6 +108,7 @@ function connectWS() { }; ws.onerror = () => { }; ws.onmessage = evt => { + if (evt.data instanceof ArrayBuffer) { onBinaryData(evt.data); return; } let msg; try { msg = JSON.parse(evt.data); } catch { return; } if (msg.type === 'sources') onSources(msg); else if (msg.type === 'config') onConfig(msg); @@ -200,6 +202,8 @@ function onData(msg) { const len = Math.min(sd.t.length, sd.v.length); for (let i = 0; i < len; i++) pushBuffer(buf, sd.t[i], sd.v[i]); }); + // Increment data generation counter so render loop knows data changed + _dataGen++; if (trig.enabled && trig.armed && trig.signal) checkTrigger(sigs); if (trig.enabled && trig.collecting && (Date.now() / 1000) >= trig.trigTime + trigPostSec()) finaliseTriggerCapture(); @@ -212,8 +216,81 @@ function onData(msg) { } /* ════════════════════════════════════════════════════════════════ - Trigger logic + Binary data handler — parses compact binary frames from Go backend. + Wire format (little-endian): + uint8 version (1) + uint8 sourceIdLen + UTF-8 sourceId + uint32 numSignals + for each signal: + uint16 keyLen + UTF-8 key (relative to source) + uint32 pairCount N + float64[N] t values + float64[N] v values ════════════════════════════════════════════════════════════════ */ +function onBinaryData(buf) { + lastDataAt = performance.now(); + const dv = new DataView(buf); + let off = 0; + if (dv.getUint8(off) !== 1) return; + off += 1; + + const srcIdLen = dv.getUint8(off); off += 1; + const srcId = new TextDecoder().decode(new Uint8Array(buf, off, srcIdLen)); + off += srcIdLen; + const prefix = srcId + ':'; + + const numSigs = dv.getUint32(off, true); off += 4; + + // Collect trigger-signal values for inline check + let trigVals = null; + + for (let s = 0; s < numSigs; s++) { + const keyLen = dv.getUint16(off, true); off += 2; + const key = new TextDecoder().decode(new Uint8Array(buf, off, keyLen)); + off += keyLen; + const fullKey = prefix + key; + + const n = dv.getUint32(off, true); off += 4; + let bufObj = buffers[fullKey]; + if (!bufObj) { + bufObj = makeBuffer(n > 100 ? TEMPORAL_CAP : DEFAULT_CAP); + buffers[fullKey] = bufObj; + } + + // Read t and v values in one pass (v array starts at off + n*8) + const tOff = off, vOff = off + n * 8; + for (let i = 0; i < n; i++) { + pushBuffer(bufObj, dv.getFloat64(tOff + i * 8, true), dv.getFloat64(vOff + i * 8, true)); + } + off += n * 16; // skip both t and v arrays + + // Capture trigger signal values + if (trig.enabled && trig.armed && fullKey === trig.signal) { + trigVals = { t: new Float64Array(n), v: new Float64Array(n) }; + for (let i = 0; i < n; i++) { + trigVals.t[i] = dv.getFloat64(tOff + i * 8, true); + trigVals.v[i] = dv.getFloat64(vOff + i * 8, true); + } + } + } + + // Trigger check + if (trigVals) checkTrigger(trigVals); + + if (trig.enabled && trig.collecting && (Date.now() / 1000) >= trig.trigTime + trigPostSec()) + finaliseTriggerCapture(); + + if (!trig.enabled) { + _dataGen++; + plots.forEach(p => { + if (globalPause) return; + if (p.traces.some(t => buffers[t] !== undefined)) p.needsRedraw = true; + }); + } +} + function checkTrigger(sigs) { const sd = sigs[trig.signal]; if (!sd || !sd.v || !sd.v.length) return; for (let i = 0; i < sd.v.length; i++) { @@ -747,8 +824,11 @@ function drawCursorLines(u, p) { } // Compute the rolling-window anchor ("newest common timestamp") for a plot. -// Returns the min-of-max timestamp across all sources contributing traces to p, -// so no source shows a blank right edge. Falls back to Date.now()/1000 if no data. +// Returns the min-of-max timestamp across ACTIVE sources contributing traces to p, +// so no live source shows a blank right edge. +// Sources whose newest timestamp lags the fastest source by more than windowSec are +// considered stale (disconnected / from a previous session) and are excluded, so they +// cannot anchor the rolling window far in the past. function computePlotNow(p) { const sourceNewest = {}; p.traces.forEach(key => { @@ -761,7 +841,11 @@ function computePlotNow(p) { if (sourceNewest[srcId] === undefined || t > sourceNewest[srcId]) sourceNewest[srcId] = t; }); const srcVals = Object.values(sourceNewest); - let now = srcVals.length > 0 ? Math.min(...srcVals) : -Infinity; + if (srcVals.length === 0) return Date.now() / 1000; + const globalMax = Math.max(...srcVals); + // Keep only sources that have received data within the last windowSec. + const active = srcVals.filter(t => t >= globalMax - windowSec); + let now = active.length > 0 ? Math.min(...active) : globalMax; if (!isFinite(now)) now = Date.now() / 1000; return now; } @@ -1031,27 +1115,21 @@ function resampleLinear(tSrc, vSrc, tDst) { function buildLiveData(p) { if (p.traces.length === 0) return [new Float64Array(0)]; - // plotNow = min(newest per source) so no source shows a blank right edge. const plotNow = computePlotNow(p); - const t0 = p.xRange ? p.xRange[0] : plotNow - windowSec; const t1 = p.xRange ? p.xRange[1] : plotNow; - // Pixel-adaptive LTTB target: 2× plot width so zooming in automatically - // raises the effective sample cap and reveals full resolution. - const targetPts = Math.max(LTTB_MIN, ((p.uplot ? p.uplot.width : p.div.clientWidth) || 600) * 2); + const isRolling = !p.xRange; // When zoomed, prefer server-fetched hi-res data if it covers this exact range. if (p.xRange) { const zd = zoomData[p.id]; if (zd && Math.abs(zd.t0 - t0) < 1e-9 && Math.abs(zd.t1 - t1) < 1e-9) { - return buildDataFromFetched(p, zd.signals, targetPts); + return buildDataFromFetched(p, zd.signals, Math.max(LTTB_MIN, ((p.uplot ? p.uplot.width : p.div.clientWidth) || 600) * 2)); } } - // Slice all traces once; pick the master time grid using configured samplingRate - // as the primary criterion (unambiguous, independent of buffer fill / trace order). - // Fall back to raw sample count for signals without a configured rate. + // Slice all traces; pick master by sampling rate then count. const slices = {}; let masterKey = p.traces[0], masterCount = -1, masterRate = -1; for (const key of p.traces) { @@ -1069,13 +1147,24 @@ function buildLiveData(p) { if (!masterRaw || masterRaw.t.length === 0) return [new Float64Array(0), ...p.traces.map(() => new Float64Array(0))]; - // Decimate master with pixel-adaptive LTTB, use resulting grid for all others - const dec = lttb(masterRaw.t, masterRaw.v, targetPts); - const sharedT = dec.t; + // 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. + let sharedT, masterV; + if (isRolling) { + 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); + sharedT = dec.t; + masterV = dec.v; + } const yArrays = []; for (const key of p.traces) { - if (key === masterKey) { yArrays.push(dec.v); continue; } + if (key === masterKey) { yArrays.push(masterV); continue; } const sl = slices[key]; if (!sl || sl.t.length === 0) { yArrays.push(new Float64Array(sharedT.length)); continue; } yArrays.push(resampleLinear(sl.t, sl.v, sharedT)); @@ -1759,7 +1848,7 @@ function addPlot() { document.getElementById('plot-grid').appendChild(card); const plotBody = card.querySelector('#pbody-' + id); - const p = { id, traces: [], div: plotBody, needsRedraw: false, xRange: null, uplot: null, ro: null }; + const p = { id, traces: [], div: plotBody, needsRedraw: false, xRange: null, uplot: null, ro: null, lastDataGen: -1 }; plots.push(p); // uPlot creation is handled by applyLayout (batch, after DOM settles). return id; @@ -1822,8 +1911,8 @@ function deletePlot(plotId) { Render loop ════════════════════════════════════════════════════════════════ */ let _dbgTick = 0; +let _dataGen = 0; // incremented each time new data arrives function renderDirtyPlots() { - const inTrigMode = trig.enabled && trig.snapshot !== null; // Diagnostic: every ~5 s print buffer state to the browser console. // Open DevTools → Console to see timestamps and sizes. @@ -1872,8 +1961,8 @@ function renderDirtyPlots() { } // Rolling-window plots: mark dirty every frame for smooth continuous scrolling. - // setScale is called AFTER setData inside the rebuild loop so the viewport and - // data slice are always computed with the same plotNow anchor. + // When no new data arrived since the last render, only advance the viewport + // via setScale instead of rebuilding all data arrays (much cheaper). if (!trig.enabled && !globalPause) { plots.forEach(p => { if (!p.uplot || p.xRange) return; @@ -1883,9 +1972,24 @@ function renderDirtyPlots() { plots.forEach(p => { if (!p.needsRedraw || !p.uplot) return; - p.needsRedraw = false; - const data = buildUPlotData(p, inTrigMode); + const inTrigModeNow = trig.enabled && trig.snapshot !== null; + const isRolling = !trig.enabled && !p.xRange; + + // Fast path: rolling-window plot with no new data — just shift viewport. + 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 }); + zoomGuard = false; + return; + } + + p.needsRedraw = false; + p.lastDataGen = _dataGen; + + const data = buildUPlotData(p, inTrigModeNow); // setData internally triggers the setScale hook in uPlot (it reaffirms the // current scale even with auto:false). Keep zoomGuard raised across the @@ -1896,9 +2000,9 @@ function renderDirtyPlots() { p.uplot.setData(data); // Re-apply the x-scale after setData so the viewport stays correct. - if (trig.enabled && !inTrigMode) { + if (trig.enabled && !inTrigModeNow) { // Armed / waiting for trigger: keep the current scale frozen. - } else if (inTrigMode) { + } else if (inTrigModeNow) { const preS = trig.snapshot._preS !== undefined ? trig.snapshot._preS : trigPreSec(); const postS = trig.snapshot._postS !== undefined ? trig.snapshot._postS : trigPostSec(); p.uplot.setScale('x', { diff --git a/Client/WebUI/static/worker.js b/Client/WebUI/static/worker.js new file mode 100644 index 0000000..95164c6 --- /dev/null +++ b/Client/WebUI/static/worker.js @@ -0,0 +1,283 @@ +'use strict'; +/* ════════════════════════════════════════════════════════════════ + Web Worker – buffer management, binary parsing, LTTB + ════════════════════════════════════════════════════════════════ */ + +const TEMPORAL_CAP = 600_000; +const DEFAULT_CAP = 10_000; + +// Circular buffers: key → {t:Float64Array, v:Float64Array, head, size, cap} +const buffers = {}; + +function makeBuffer(cap) { + return { t: new Float64Array(cap), v: new Float64Array(cap), head: 0, size: 0, cap }; +} +function pushBuffer(buf, t, v) { + buf.t[buf.head] = t; buf.v[buf.head] = v; + buf.head = (buf.head + 1) % buf.cap; + if (buf.size < buf.cap) buf.size++; +} + +// ─── Binary frame parser ───────────────────────────────────────────── +// Format (little-endian): +// uint8 version (1) +// uint8 sourceIdLen +// UTF-8 sourceId +// uint32 numSignals +// for each signal: +// uint16 keyLen +// UTF-8 key (relative to source) +// uint32 pairCount N +// float64[N] t values +// float64[N] v values +function parseBinaryFrame(buf) { + const dv = new DataView(buf); + let off = 0; + + if (dv.getUint8(off) !== 1) { console.warn('[worker] bad binary version'); return; } + off += 1; + + const srcIdLen = dv.getUint8(off); off += 1; + const srcId = new TextDecoder().decode(new Uint8Array(buf, off, srcIdLen)); + off += srcIdLen; + + const prefix = srcId + ':'; + const numSigs = dv.getUint32(off, true); off += 4; + + for (let s = 0; s < numSigs; s++) { + const keyLen = dv.getUint16(off, true); off += 2; + const key = new TextDecoder().decode(new Uint8Array(buf, off, keyLen)); + off += keyLen; + + const fullKey = prefix + key; + const n = dv.getUint32(off, true); off += 4; + + let bufObj = buffers[fullKey]; + if (!bufObj) { + // Auto-create buffer with reasonable capacity + const cap = n > 100 ? TEMPORAL_CAP : DEFAULT_CAP; + bufObj = makeBuffer(cap); + buffers[fullKey] = bufObj; + } + + // Read t values + for (let i = 0; i < n; i++) { + const t = dv.getFloat64(off, true); off += 8; + const v = dv.getFloat64(off + n * 8, true); // v array starts after t array + pushBuffer(bufObj, t, v); + } + off += n * 8; // skip v array (already read inline above) + } +} + +// ─── Range slice from circular buffer ──────────────────────────────── +function getBufferSliceRange(bufObj, t0, t1) { + const { cap, size, head } = bufObj; + if (size === 0) return { t: new Float64Array(0), v: new Float64Array(0) }; + const start = (size === cap) ? head : 0; + const physAt = k => (start + k) % cap; + + let lo = 0, hi = size; + while (lo < hi) { const m = (lo + hi) >>> 1; if (bufObj.t[physAt(m)] < t0) lo = m + 1; else hi = m; } + const kStart = lo; + lo = kStart; hi = size; + while (lo < hi) { const m = (lo + hi) >>> 1; if (bufObj.t[physAt(m)] <= t1) lo = m + 1; else hi = m; } + const kEnd = lo, len = kEnd - kStart; + if (len <= 0) return { t: new Float64Array(0), v: new Float64Array(0) }; + + const outT = new Float64Array(len), outV = new Float64Array(len); + const physStart = physAt(kStart), tail = cap - physStart; + if (tail >= len) { + outT.set(bufObj.t.subarray(physStart, physStart + len)); + outV.set(bufObj.v.subarray(physStart, physStart + len)); + } else { + outT.set(bufObj.t.subarray(physStart, physStart + tail)); + outT.set(bufObj.t.subarray(0, len - tail), tail); + outV.set(bufObj.v.subarray(physStart, physStart + tail)); + outV.set(bufObj.v.subarray(0, len - tail), tail); + } + return { t: outT, v: outV }; +} + +// ─── LTTB decimation ───────────────────────────────────────────────── +function lttb(t, v, threshold) { + const len = t.length; + if (len <= threshold || threshold < 3) return { t, v }; + const outT = new Float64Array(threshold), outV = new Float64Array(threshold); + outT[0] = t[0]; outV[0] = v[0]; + outT[threshold - 1] = t[len - 1]; outV[threshold - 1] = v[len - 1]; + const every = (len - 2) / (threshold - 2); + let a = 0; + for (let i = 0; i < threshold - 2; i++) { + const avgS = Math.floor((i + 1) * every) + 1, avgE = Math.min(Math.floor((i + 2) * every) + 1, len); + let avgT = 0, avgV = 0, n = 0; + for (let j = avgS; j < avgE; j++) { avgT += t[j]; avgV += v[j]; n++; } + if (n) { avgT /= n; avgV /= n; } + const rS = Math.floor(i * every) + 1, rE = Math.min(Math.floor((i + 1) * every) + 1, len); + let maxA = -1, next = rS; + const aT = t[a], aV = v[a]; + for (let j = rS; j < rE; j++) { + const area = Math.abs((aT - avgT) * (v[j] - aV) - (aT - t[j]) * (avgV - aV)); + if (area > maxA) { maxA = area; next = j; } + } + outT[i + 1] = t[next]; outV[i + 1] = v[next]; a = next; + } + return { t: outT, v: outV }; +} + +// ─── Linear resampling ─────────────────────────────────────────────── +function resampleLinear(tSrc, vSrc, tDst) { + const n = tDst.length; + const out = new Float64Array(n); + if (tSrc.length === 0) return out; + if (tSrc.length === 1) { out.fill(vSrc[0]); return out; } + let j = 0; + for (let i = 0; i < n; i++) { + const td = tDst[i]; + while (j < tSrc.length - 2 && tSrc[j + 1] < td) j++; + if (td <= tSrc[0]) { out[i] = vSrc[0]; } + else if (td >= tSrc[tSrc.length - 1]) { out[i] = vSrc[vSrc.length - 1]; } + else { + const t0 = tSrc[j], t1 = tSrc[j + 1]; + const frac = (td - t0) / (t1 - t0); + out[i] = vSrc[j] + frac * (vSrc[j + 1] - vSrc[j]); + } + } + return out; +} + +// ─── Master time grid selection ────────────────────────────────────── +// samplingRates: key → rate (Hz), provided by main thread on init +const samplingRates = {}; + +function pickMasterKey(keys) { + let bestKey = keys[0], bestRate = -1; + for (const k of keys) { + const rate = samplingRates[k] || 0; + if (rate > bestRate) { bestRate = rate; bestKey = k; } + } + return bestKey; +} + +// ─── Build uPlot-compatible data arrays ────────────────────────────── +function buildRenderData(keys, t0, t1, targetPts) { + if (!keys || keys.length === 0) return [new Float64Array(0)]; + + const slices = {}; + let masterKey = pickMasterKey(keys), masterCount = -1; + + for (const key of keys) { + const bufObj = buffers[key]; + if (!bufObj || bufObj.size === 0) continue; + const sl = getBufferSliceRange(bufObj, t0, t1); + slices[key] = sl; + if (sl.t.length > masterCount) { masterCount = sl.t.length; masterKey = key; } + } + + const masterRaw = slices[masterKey]; + if (!masterRaw || masterRaw.t.length === 0) + return [new Float64Array(0), ...keys.map(() => new Float64Array(0))]; + + const dec = lttb(masterRaw.t, masterRaw.v, targetPts); + const sharedT = dec.t; + const yArrays = []; + + for (const key of keys) { + if (key === masterKey) { yArrays.push(dec.v); continue; } + const sl = slices[key]; + if (!sl || sl.t.length === 0) { yArrays.push(new Float64Array(sharedT.length)); continue; } + yArrays.push(resampleLinear(sl.t, sl.v, sharedT)); + } + + const result = [sharedT, ...yArrays]; + // Transfer ownership of the Float64Arrays to main thread + const transferList = result.map(a => a.buffer); + return { data: result, transfer: transferList }; +} + +// ─── Message handler ───────────────────────────────────────────────── +self.onmessage = function(e) { + const msg = e.data; + + switch (msg.type) { + case 'initSignals': { + // {signals: [{key, cap}]} + const sigs = msg.signals || []; + sigs.forEach(s => { + if (!buffers[s.key]) { + buffers[s.key] = makeBuffer(s.cap || DEFAULT_CAP); + } + if (s.samplingRate !== undefined) { + samplingRates[s.key] = s.samplingRate; + } + }); + break; + } + + case 'binaryData': { + // {buffer: ArrayBuffer} — transferred from main thread + parseBinaryFrame(msg.buffer); + self.postMessage({ type: 'dataReady' }); + break; + } + + case 'requestData': { + // {id, t0, t1, targetPts, keys} + const { id, t0, t1, targetPts, keys } = msg; + const { data, transfer } = buildRenderData(keys, t0, t1, targetPts); + self.postMessage({ type: 'renderData', id, data }, transfer); + break; + } + + case 'clearSource': { + const prefix = msg.prefix; + Object.keys(buffers).forEach(k => { + if (k.startsWith(prefix)) delete buffers[k]; + }); + Object.keys(samplingRates).forEach(k => { + if (k.startsWith(prefix)) delete samplingRates[k]; + }); + break; + } + + case 'getBufferNow': { + // Returns newest timestamp across given keys + const keys = msg.keys || []; + let latest = -Infinity; + keys.forEach(key => { + const bufObj = buffers[key]; + if (bufObj && bufObj.size > 0) { + const t = bufObj.t[(bufObj.head - 1 + bufObj.cap) % bufObj.cap]; + if (t > latest) latest = t; + } + }); + self.postMessage({ type: 'bufferNow', id: msg.id, now: isFinite(latest) ? latest : null }); + break; + } + + case 'getBufferForTrig': { + // Returns full buffer contents for a single key (used for trigger check) + const key = msg.key; + const bufObj = buffers[key]; + if (!bufObj || bufObj.size === 0) { + self.postMessage({ type: 'trigBuf', id: msg.id, key, size: 0 }); + break; + } + // Copy out all data + const { cap, size, head } = bufObj; + const start = (size === cap) ? head : 0; + const t = new Float64Array(size), v = new Float64Array(size); + const physAt = k => (start + k) % cap; + for (let i = 0; i < size; i++) { + const p = physAt(i); + t[i] = bufObj.t[p]; + v[i] = bufObj.v[p]; + } + self.postMessage({ + type: 'trigBuf', id: msg.id, key, size, + t, v + }, [t.buffer, v.buffer]); + break; + } + } +}; diff --git a/Client/WebUI/udpclient.go b/Client/WebUI/udpclient.go index 4eb05bc..4ed07ee 100644 --- a/Client/WebUI/udpclient.go +++ b/Client/WebUI/udpclient.go @@ -10,6 +10,7 @@ const ( silenceTimeout = 5 * time.Second reconnectDelay = 2 * time.Second readBufSize = 65536 + udpRcvBufSize = 8 * 1024 * 1024 // 8 MB OS receive buffer — absorbs bursts at high data rates ) // UDPClient manages the UDP connection to one MARTe2 streamer source. @@ -71,6 +72,11 @@ func (u *UDPClient) runSession() error { } defer conn.Close() + // Increase OS receive buffer to reduce kernel-level packet drops at high data rates. + if err := conn.SetReadBuffer(udpRcvBufSize); err != nil { + log.Printf("[%s] udp: SetReadBuffer: %v (proceeding with OS default)", u.sourceID, err) + } + serverAddr, err := net.ResolveUDPAddr("udp4", u.serverAddr) if err != nil { return err