martino/MARTe_IO_Components
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Faster implementation with binary websocket

Browse Source
This commit is contained in:
Martino Ferrari
2026-05-21 15:48:08 +02:00
parent 62545503c4
commit 3315c02282
4 changed files with 658 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Client/WebUI/hub.go
+240 -12
View File
@@ -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()