From 6ff8fb5c25cd7675451dd65b5799f60daad89ab1 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Tue, 12 May 2026 10:16:48 +0200 Subject: [PATCH] Improved perfs --- cmd/uopi/main.go | 2 +- internal/api/api.go | 36 +++ internal/broker/broker.go | 80 +++++- internal/broker/broker_stress_test.go | 269 ++++++++++++++++++++ internal/broker/broker_test.go | 62 +++++ internal/config/config.go | 11 +- internal/datasource/stub/stub.go | 53 +++- internal/server/ws.go | 4 +- internal/server/ws_stress_test.go | 343 ++++++++++++++++++++++++++ internal/storage/store.go | 23 +- web/src/GroupsTree.tsx | 327 ++++++++++++++++++++++++ web/src/SignalTree.tsx | 27 +- web/src/lib/types.ts | 6 + web/src/styles.css | 139 +++++++++++ 14 files changed, 1357 insertions(+), 25 deletions(-) create mode 100644 internal/broker/broker_stress_test.go create mode 100644 internal/server/ws_stress_test.go create mode 100644 web/src/GroupsTree.tsx diff --git a/cmd/uopi/main.go b/cmd/uopi/main.go index 5232c04..92ec7cf 100644 --- a/cmd/uopi/main.go +++ b/cmd/uopi/main.go @@ -58,7 +58,7 @@ func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() - brk := broker.New(ctx, log) + brk := broker.New(ctx, log, broker.WithMaxUpdateRate(cfg.Server.MaxUpdateRateHz)) // Stub data source: built-in simulated signals (sine, ramp, noise, setpoint). // Enabled by default; disable via [datasource.stub] enabled = false in config. diff --git a/internal/api/api.go b/internal/api/api.go index 5fcc7fd..ef21f6a 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -42,6 +42,9 @@ func (h *Handler) Register(mux *http.ServeMux, prefix string) { mux.HandleFunc("PUT "+prefix+"/interfaces/{id}", h.updateInterface) mux.HandleFunc("DELETE "+prefix+"/interfaces/{id}", h.deleteInterface) mux.HandleFunc("POST "+prefix+"/interfaces/{id}/clone", h.cloneInterface) + // Signal group tree + mux.HandleFunc("GET "+prefix+"/groups", h.getGroups) + mux.HandleFunc("PUT "+prefix+"/groups", h.putGroups) // Synthetic signal CRUD mux.HandleFunc("GET "+prefix+"/synthetic", h.listSynthetic) mux.HandleFunc("POST "+prefix+"/synthetic", h.createSynthetic) @@ -184,6 +187,39 @@ func (h *Handler) channelFinder(w http.ResponseWriter, r *http.Request) { jsonOK(w, names) } +// ── /groups ─────────────────────────────────────────────────────────────────── + +// getGroups returns the workspace group tree as a JSON array. +func (h *Handler) getGroups(w http.ResponseWriter, r *http.Request) { + data, err := h.store.ReadGroups() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(data) +} + +// putGroups replaces the workspace group tree. The body must be a JSON array. +func (h *Handler) putGroups(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20)) // 1 MB max + if err != nil { + http.Error(w, "read error", http.StatusBadRequest) + return + } + // Validate it is a JSON array (basic sanity check). + var raw []json.RawMessage + if err := json.Unmarshal(body, &raw); err != nil { + http.Error(w, "body must be a JSON array", http.StatusBadRequest) + return + } + if err := h.store.WriteGroups(body); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusNoContent) +} + // ── /interfaces ─────────────────────────────────────────────────────────────── func (h *Handler) listInterfaces(w http.ResponseWriter, r *http.Request) { diff --git a/internal/broker/broker.go b/internal/broker/broker.go index 7cb5b18..2dfe0b2 100644 --- a/internal/broker/broker.go +++ b/internal/broker/broker.go @@ -8,6 +8,7 @@ import ( "fmt" "log/slog" "sync" + "time" "github.com/uopi/uopi/internal/datasource" ) @@ -41,18 +42,38 @@ type Broker struct { sources map[string]datasource.DataSource subs map[SignalRef]*signalSub - log *slog.Logger + maxInterval time.Duration // 0 = unlimited + log *slog.Logger +} + +// Option is a functional option for Broker configuration. +type Option func(*Broker) + +// WithMaxUpdateRate limits fan-out to at most hz updates per second per signal. +// When upstream delivers faster, intermediate values are coalesced: the most +// recent value in each interval is forwarded once the interval elapses. +// A value ≤ 0 disables rate limiting (the default). +func WithMaxUpdateRate(hz float64) Option { + return func(b *Broker) { + if hz > 0 { + b.maxInterval = time.Duration(float64(time.Second) / hz) + } + } } // New creates a Broker whose upstream subscriptions are bound to ctx. // Cancel ctx (or the parent context passed to main) to shut everything down. -func New(ctx context.Context, log *slog.Logger) *Broker { - return &Broker{ +func New(ctx context.Context, log *slog.Logger, opts ...Option) *Broker { + b := &Broker{ ctx: ctx, sources: make(map[string]datasource.DataSource), subs: make(map[SignalRef]*signalSub), log: log, } + for _, opt := range opts { + opt(b) + } + return b } // Register adds a DataSource to the broker. Must be called before Subscribe. @@ -169,7 +190,36 @@ func (b *Broker) unsubscribe(ref SignalRef, ch chan<- Update) { // fanOut reads values from rawCh and dispatches them to all registered clients. // It exits when sub.done is closed or rawCh is closed. +// +// When b.maxInterval > 0, updates that arrive faster than the interval are +// coalesced: only the most recent value in each interval window is forwarded, +// delivered at the end of the interval by a flush ticker. This ensures the +// latest value is always eventually seen even when the source fires rapidly. func (b *Broker) fanOut(ref SignalRef, sub *signalSub, rawCh <-chan datasource.Value) { + maxInterval := b.maxInterval + + var lastSent time.Time + var pending *Update + + // Only allocate a ticker when rate-limiting is configured. + var flushC <-chan time.Time + if maxInterval > 0 { + t := time.NewTicker(maxInterval) + defer t.Stop() + flushC = t.C + } + + dispatch := func(u Update) { + sub.mu.RLock() + for ch := range sub.clients { + select { + case ch <- u: + default: // slow consumer: drop rather than block + } + } + sub.mu.RUnlock() + } + for { select { case v, ok := <-rawCh: @@ -177,15 +227,23 @@ func (b *Broker) fanOut(ref SignalRef, sub *signalSub, rawCh <-chan datasource.V return } update := Update{Ref: ref, Value: v} - sub.mu.RLock() - for ch := range sub.clients { - select { - case ch <- update: - default: - // slow consumer: drop rather than block - } + // Rate-limit: if an interval is configured and the last delivery + // was recent, hold this update as pending (coalesce). + if maxInterval > 0 && !lastSent.IsZero() && time.Since(lastSent) < maxInterval { + pending = &update + continue + } + lastSent = time.Now() + pending = nil + dispatch(update) + + case <-flushC: + // Deliver the most recent coalesced value, if any. + if pending != nil { + lastSent = time.Now() + dispatch(*pending) + pending = nil } - sub.mu.RUnlock() case <-sub.done: return diff --git a/internal/broker/broker_stress_test.go b/internal/broker/broker_stress_test.go new file mode 100644 index 0000000..fe7e064 --- /dev/null +++ b/internal/broker/broker_stress_test.go @@ -0,0 +1,269 @@ +package broker_test + +import ( + "context" + "fmt" + "log/slog" + "runtime" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/uopi/uopi/internal/broker" + "github.com/uopi/uopi/internal/datasource/stub" +) + +// newBrokerN builds a broker backed by a stub with n dynamically generated signals. +func newBrokerN(tb testing.TB, n int) (*broker.Broker, context.CancelFunc) { + tb.Helper() + ctx, cancel := context.WithCancel(context.Background()) + log := slog.Default() + b := broker.New(ctx, log) + ds := stub.NewN(n) + if err := ds.Connect(ctx); err != nil { + cancel() + tb.Fatal(err) + } + b.Register(ds) + return b, cancel +} + +// TestStress_ManySignalsManyClients subscribes 20 clients to 500 signals each, +// runs for 2 seconds, then verifies delivery counts and zero subscription leaks. +func TestStress_ManySignalsManyClients(t *testing.T) { + if testing.Short() { + t.Skip("skipping stress test in short mode") + } + + const ( + nSignals = 500 + nClients = 20 + duration = 2 * time.Second + ) + + goroutinesBefore := runtime.NumGoroutine() + + brk, cancel := newBrokerN(t, nSignals) + defer cancel() + + // Each client has its own channel and subscribes to all nSignals signals. + type clientState struct { + ch chan broker.Update + unsub []func() + } + clients := make([]clientState, nClients) + + for i := range nClients { + ch := make(chan broker.Update, 2048) + unsubs := make([]func(), nSignals) + for j := range nSignals { + ref := broker.SignalRef{DS: "stub", Name: fmt.Sprintf("pv_%d", j)} + unsub, err := brk.Subscribe(ref, ch) + if err != nil { + t.Fatalf("client %d subscribe pv_%d: %v", i, j, err) + } + unsubs[j] = unsub + } + clients[i] = clientState{ch: ch, unsub: unsubs} + } + + if n := brk.ActiveSubscriptions(); n != nSignals { + t.Errorf("expected %d active upstream subscriptions, got %d", nSignals, n) + } + + // Drain updates concurrently for the test duration. + var totalReceived atomic.Int64 + stop := make(chan struct{}) + var wg sync.WaitGroup + + for i := range nClients { + wg.Add(1) + ch := clients[i].ch + go func() { + defer wg.Done() + var n int64 + for { + select { + case <-ch: + n++ + case <-stop: + totalReceived.Add(n) + return + } + } + }() + } + + time.Sleep(duration) + close(stop) + wg.Wait() + + total := totalReceived.Load() + // At 10 Hz for duration seconds: nSignals * nClients * duration.Seconds() * 10 + minExpected := int64(nSignals) * int64(nClients) * int64(duration.Seconds()) * 5 // conservative: 50% + t.Logf("received %d updates (%d clients × %d signals × %.0fs @ 10Hz, min=%d)", + total, nClients, nSignals, duration.Seconds(), minExpected) + if total < minExpected { + t.Errorf("too few updates: got %d, want >= %d", total, minExpected) + } + + // Unsubscribe all clients. + for _, c := range clients { + for _, unsub := range c.unsub { + unsub() + } + } + + // Give the broker time to tear down upstream subscriptions. + time.Sleep(200 * time.Millisecond) + + if n := brk.ActiveSubscriptions(); n != 0 { + t.Errorf("subscription leak: %d active subscriptions remain after all clients unsubscribed", n) + } + + cancel() + time.Sleep(300 * time.Millisecond) + + goroutinesAfter := runtime.NumGoroutine() + leaked := goroutinesAfter - goroutinesBefore + t.Logf("goroutines: before=%d after=%d delta=%d", goroutinesBefore, goroutinesAfter, leaked) + if leaked > 20 { + t.Errorf("goroutine leak: started with %d, ended with %d (%d extra)", goroutinesBefore, goroutinesAfter, leaked) + } +} + +// TestStress_RapidSubscribeUnsubscribe hammers subscribe/unsubscribe on a small +// signal set from many goroutines concurrently to expose races and deadlocks. +func TestStress_RapidSubscribeUnsubscribe(t *testing.T) { + if testing.Short() { + t.Skip("skipping stress test in short mode") + } + + const ( + nSignals = 20 + nGoroutines = 50 + duration = 2 * time.Second + ) + + brk, cancel := newBrokerN(t, nSignals) + defer cancel() + + var wg sync.WaitGroup + stop := make(chan struct{}) + var ops atomic.Int64 + + for range nGoroutines { + wg.Add(1) + go func() { + defer wg.Done() + ch := make(chan broker.Update, 64) + for { + select { + case <-stop: + return + default: + } + // Pick a signal and rapidly subscribe/unsubscribe. + idx := int(ops.Load()) % nSignals + ref := broker.SignalRef{DS: "stub", Name: fmt.Sprintf("pv_%d", idx)} + unsub, err := brk.Subscribe(ref, ch) + if err != nil { + t.Errorf("subscribe pv_%d: %v", idx, err) + return + } + ops.Add(1) + // Drain a bit before unsubscribing. + select { + case <-ch: + case <-time.After(50 * time.Millisecond): + } + unsub() + } + }() + } + + time.Sleep(duration) + close(stop) + wg.Wait() + + t.Logf("completed %d subscribe/unsubscribe cycles in %s", ops.Load(), duration) + + time.Sleep(200 * time.Millisecond) + if n := brk.ActiveSubscriptions(); n != 0 { + t.Errorf("subscription leak after rapid churn: %d remain", n) + } +} + +// TestStress_SharedSignalManyClients verifies that a single high-rate signal +// fans out correctly to many concurrent clients with no drops for fast consumers. +func TestStress_SharedSignalManyClients(t *testing.T) { + if testing.Short() { + t.Skip("skipping stress test in short mode") + } + + const ( + nClients = 50 + duration = 2 * time.Second + ) + + brk, cancel := newBroker(t) // uses stub.New() which has counter_fast at 1 ms + defer cancel() + + ref := broker.SignalRef{DS: "stub", Name: "counter_fast"} + channels := make([]chan broker.Update, nClients) + unsubs := make([]func(), nClients) + + for i := range nClients { + ch := make(chan broker.Update, 4096) + channels[i] = ch + unsub, err := brk.Subscribe(ref, ch) + if err != nil { + t.Fatalf("subscribe client %d: %v", i, err) + } + unsubs[i] = unsub + } + defer func() { + for _, u := range unsubs { + u() + } + }() + + if n := brk.ActiveSubscriptions(); n != 1 { + t.Errorf("expected exactly 1 upstream subscription for shared signal, got %d", n) + } + + var totalReceived atomic.Int64 + stop := make(chan struct{}) + var wg sync.WaitGroup + + for i := range nClients { + wg.Add(1) + ch := channels[i] + go func() { + defer wg.Done() + var n int64 + for { + select { + case <-ch: + n++ + case <-stop: + totalReceived.Add(n) + return + } + } + }() + } + + time.Sleep(duration) + close(stop) + wg.Wait() + + total := totalReceived.Load() + // counter_fast ticks at 1ms → ~1000 Hz → 2000 updates × 50 clients = 100000 expected minimum + minExpected := int64(nClients) * int64(duration.Seconds()) * 500 // 50% delivery ok due to buffering + t.Logf("shared signal: received %d updates across %d clients (min=%d)", total, nClients, minExpected) + if total < minExpected { + t.Errorf("too few updates: got %d, want >= %d", total, minExpected) + } +} diff --git a/internal/broker/broker_test.go b/internal/broker/broker_test.go index 1542345..8272265 100644 --- a/internal/broker/broker_test.go +++ b/internal/broker/broker_test.go @@ -122,6 +122,68 @@ func TestUnknownSignal(t *testing.T) { } } +func TestMaxUpdateRate(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + const maxHz = 5.0 // 5 Hz → 200 ms interval + + ds := stub.New() // counter_fast fires at 1 kHz + if err := ds.Connect(ctx); err != nil { + t.Fatal(err) + } + b := broker.New(ctx, slog.Default(), broker.WithMaxUpdateRate(maxHz)) + b.Register(ds) + + ref := broker.SignalRef{DS: "stub", Name: "counter_fast"} + ch := make(chan broker.Update, 512) + unsub, err := b.Subscribe(ref, ch) + if err != nil { + t.Fatal(err) + } + defer unsub() + + // Collect updates for 2 seconds. + var count int + deadline := time.After(2 * time.Second) +loop: + for { + select { + case <-ch: + count++ + case <-deadline: + break loop + } + } + + // At 5 Hz for 2 s we expect ~10 updates (±3 for timer jitter + coalescing). + t.Logf("MaxUpdateRate: received %d updates at %.0f Hz limit over 2s (expect ~10)", count, maxHz) + if count < 5 || count > 20 { + t.Errorf("expected ~10 updates at 5 Hz, got %d", count) + } +} + +func TestMaxUpdateRate_UnlimitedByDefault(t *testing.T) { + b, cancel := newBroker(t) // no WithMaxUpdateRate → unlimited + defer cancel() + + ref := broker.SignalRef{DS: "stub", Name: "counter_fast"} + ch := make(chan broker.Update, 4096) + unsub, err := b.Subscribe(ref, ch) + if err != nil { + t.Fatal(err) + } + defer unsub() + + // counter_fast fires at 1 kHz; over 500 ms we should see ≥ 400 updates. + time.Sleep(500 * time.Millisecond) + got := len(ch) + t.Logf("Unlimited: received %d updates in 500ms from 1kHz signal", got) + if got < 400 { + t.Errorf("expected >= 400 updates with no rate limit, got %d", got) + } +} + func TestMultipleSignals(t *testing.T) { b, cancel := newBroker(t) defer cancel() diff --git a/internal/config/config.go b/internal/config/config.go index b589ff2..1700aec 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -3,6 +3,7 @@ package config import ( "fmt" "os" + "strconv" "strings" "github.com/BurntSushi/toml" @@ -14,8 +15,9 @@ type Config struct { } type ServerConfig struct { - Listen string `toml:"listen"` - StorageDir string `toml:"storage_dir"` + Listen string `toml:"listen"` + StorageDir string `toml:"storage_dir"` + MaxUpdateRateHz float64 `toml:"max_update_rate_hz"` // 0 = unlimited } type DatasourceConfig struct { @@ -84,6 +86,11 @@ func applyEnv(cfg *Config) { if v := env("UOPI_SERVER_STORAGE_DIR"); v != "" { cfg.Server.StorageDir = v } + if v := env("UOPI_SERVER_MAX_UPDATE_RATE_HZ"); v != "" { + if hz, err := strconv.ParseFloat(v, 64); err == nil { + cfg.Server.MaxUpdateRateHz = hz + } + } if v := env("UOPI_EPICS_CA_ADDR_LIST"); v != "" { cfg.Datasource.EPICS.CAAddrList = v } diff --git a/internal/datasource/stub/stub.go b/internal/datasource/stub/stub.go index 677ada5..6048a3a 100644 --- a/internal/datasource/stub/stub.go +++ b/internal/datasource/stub/stub.go @@ -5,6 +5,7 @@ package stub import ( "context" + "fmt" "math" "math/rand/v2" "sync" @@ -13,11 +14,12 @@ import ( "github.com/uopi/uopi/internal/datasource" ) -const updateInterval = 100 * time.Millisecond // 10 Hz +const updateInterval = 100 * time.Millisecond // 10 Hz default type signalDef struct { - meta datasource.Metadata - fn func(t time.Time) any + meta datasource.Metadata + interval time.Duration // 0 → updateInterval + fn func(t time.Time) any } // Stub is a data source that emits canned signals for development and testing. @@ -96,6 +98,41 @@ func New() *Stub { fn: func(t time.Time) any { return 25.0 }, // default; overwritten by Write }) s.values["setpoint"] = 25.0 + s.register(signalDef{ + meta: datasource.Metadata{ + Name: "counter_fast", Type: datasource.TypeFloat64, + Description: "Fast monotonic counter (1 ms tick) for benchmarks", + }, + interval: time.Millisecond, + fn: func(t time.Time) any { return float64(t.UnixMilli()) }, + }) + return s +} + +// NewN creates a Stub with n dynamically generated signals named "pv_0" … "pv_{n-1}". +// Each signal is a sine wave at a distinct frequency, emitted at 10 Hz. +// Use this for stress and scale testing. +func NewN(n int) *Stub { + s := &Stub{ + signals: make(map[string]signalDef), + values: make(map[string]any), + } + for i := range n { + freq := 0.1 + float64(i)*0.01 // spread frequencies so signals differ + s.register(signalDef{ + meta: datasource.Metadata{ + Name: fmt.Sprintf("pv_%d", i), + Type: datasource.TypeFloat64, + Unit: "V", + DisplayLow: -1, + DisplayHigh: 1, + Description: fmt.Sprintf("Synthetic PV #%d (%.2f Hz sine)", i, freq), + }, + fn: func(t time.Time) any { + return math.Sin(2 * math.Pi * freq * float64(t.UnixNano()) / 1e9) + }, + }) + } return s } @@ -127,16 +164,22 @@ func (s *Stub) GetMetadata(_ context.Context, signal string) (datasource.Metadat return d.meta, nil } -// Subscribe starts a 10 Hz ticker that pushes generated values into ch. +// Subscribe starts a ticker that pushes generated values into ch. +// The tick interval is the signal's configured interval (default 10 Hz). func (s *Stub) Subscribe(ctx context.Context, signal string, ch chan<- datasource.Value) (datasource.CancelFunc, error) { def, ok := s.signals[signal] if !ok { return nil, datasource.ErrNotFound } + interval := def.interval + if interval == 0 { + interval = updateInterval + } + ctx, cancel := context.WithCancel(ctx) go func() { - ticker := time.NewTicker(updateInterval) + ticker := time.NewTicker(interval) defer ticker.Stop() for { select { diff --git a/internal/server/ws.go b/internal/server/ws.go index 7fc7290..953e0df 100644 --- a/internal/server/ws.go +++ b/internal/server/ws.go @@ -105,8 +105,8 @@ func (h *wsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { c := &wsClient{ conn: conn, broker: h.broker, - outCh: make(chan []byte, 128), - updateCh: make(chan broker.Update, 256), + outCh: make(chan []byte, 512), + updateCh: make(chan broker.Update, 1024), subs: make(map[broker.SignalRef]func()), log: h.log, } diff --git a/internal/server/ws_stress_test.go b/internal/server/ws_stress_test.go new file mode 100644 index 0000000..c5b39a3 --- /dev/null +++ b/internal/server/ws_stress_test.go @@ -0,0 +1,343 @@ +package server + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" + + "github.com/uopi/uopi/internal/broker" + "github.com/uopi/uopi/internal/datasource/stub" +) + +// newStressServer creates an httptest.Server backed by a stub with n signals. +// The server and broker contexts are tied to t.Context() and cleaned up automatically. +func newStressServer(t *testing.T, nSignals int) (*httptest.Server, *broker.Broker) { + t.Helper() + ctx := t.Context() + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + + var ds *stub.Stub + if nSignals > 0 { + ds = stub.NewN(nSignals) + } else { + ds = stub.New() + } + if err := ds.Connect(ctx); err != nil { + t.Fatal(err) + } + + brk := broker.New(ctx, log) + brk.Register(ds) + + mux := http.NewServeMux() + mux.Handle("/ws", &wsHandler{broker: brk, log: log}) + + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + return srv, brk +} + +// wsURL converts an http:// test server URL to ws://. +func wsURL(srv *httptest.Server) string { + return strings.ReplaceAll(srv.URL, "http://", "ws://") + "/ws" +} + +// buildSubMsg encodes a subscribe message for n stub PVs named "pv_0"…"pv_{n-1}". +func buildSubMsg(n int) []byte { + sigs := make([]sigRef, n) + for i := range n { + sigs[i] = sigRef{DS: "stub", Name: fmt.Sprintf("pv_%d", i)} + } + data, _ := json.Marshal(inMsg{Type: "subscribe", Signals: sigs}) + return data +} + +// connectAndCount dials the WebSocket, sends subMsg, counts incoming messages until +// ctx is cancelled, adds the count to *total, and signals wg.Done. +func connectAndCount(ctx context.Context, t *testing.T, url string, subMsg []byte, total *atomic.Int64, wg *sync.WaitGroup) { + t.Helper() + defer wg.Done() + + conn, _, err := websocket.Dial(ctx, url, nil) + if err != nil { + t.Errorf("dial: %v", err) + return + } + defer conn.CloseNow() + + if err := conn.Write(ctx, websocket.MessageText, subMsg); err != nil { + t.Errorf("subscribe write: %v", err) + return + } + + var n int64 + for { + _, _, err := conn.Read(ctx) + if err != nil { + break + } + n++ + } + total.Add(n) +} + +// TestStress_100PVs_10Clients verifies the full stack (broker + WS handler) can +// deliver updates from 100 signals to 10 concurrent clients without loss or deadlock. +func TestStress_100PVs_10Clients(t *testing.T) { + if testing.Short() { + t.Skip("skipping stress test in short mode") + } + + const ( + nSignals = 100 + nClients = 10 + duration = 3 * time.Second + ) + + srv, brk := newStressServer(t, nSignals) + url := wsURL(srv) + subMsg := buildSubMsg(nSignals) + + ctx, cancel := context.WithTimeout(t.Context(), duration+5*time.Second) + defer cancel() + + var total atomic.Int64 + var wg sync.WaitGroup + + for range nClients { + wg.Add(1) + // Each client runs for exactly `duration`, then its context expires and + // conn.Read returns an error, causing the goroutine to exit cleanly. + clientCtx, clientCancel := context.WithTimeout(ctx, duration) + go func() { + defer clientCancel() + connectAndCount(clientCtx, t, url, subMsg, &total, &wg) + }() + } + + wg.Wait() + + got := total.Load() + // At 10 Hz, each PV fires ≥ duration*10 times; each client receives all of them. + // min = nClients × nSignals × duration_s × 10 × 0.5 (conservative — meta msgs too) + minExpected := int64(nClients) * int64(nSignals) * int64(duration.Seconds()) * 5 + t.Logf("100PVs/10Clients: received %d updates (min expected %d)", got, minExpected) + if got < minExpected { + t.Errorf("too few updates: got %d, want >= %d", got, minExpected) + } + + // Broker must release all upstream subscriptions after clients disconnect. + time.Sleep(300 * time.Millisecond) + if n := brk.ActiveSubscriptions(); n != 0 { + t.Errorf("subscription leak: %d remain after all clients disconnected", n) + } +} + +// TestStress_500PVs_20Clients is the full-scale scenario: hundreds of signals, +// tens of clients. Verifies no crash, no deadlock, and clean teardown. +func TestStress_500PVs_20Clients(t *testing.T) { + if testing.Short() { + t.Skip("skipping stress test in short mode") + } + + const ( + nSignals = 500 + nClients = 20 + duration = 3 * time.Second + ) + + srv, brk := newStressServer(t, nSignals) + url := wsURL(srv) + subMsg := buildSubMsg(nSignals) + + ctx, cancel := context.WithTimeout(t.Context(), duration+10*time.Second) + defer cancel() + + var total atomic.Int64 + var wg sync.WaitGroup + + for range nClients { + wg.Add(1) + clientCtx, clientCancel := context.WithTimeout(ctx, duration) + go func() { + defer clientCancel() + connectAndCount(clientCtx, t, url, subMsg, &total, &wg) + }() + } + + wg.Wait() + + got := total.Load() + // Conservative: expect at least 20% of the theoretical maximum + minExpected := int64(nClients) * int64(nSignals) * int64(duration.Seconds()) * 2 + t.Logf("500PVs/20Clients: received %d updates (min expected %d)", got, minExpected) + if got < minExpected { + t.Errorf("too few updates: got %d, want >= %d", got, minExpected) + } + + time.Sleep(500 * time.Millisecond) + if n := brk.ActiveSubscriptions(); n != 0 { + t.Errorf("subscription leak: %d remain after all clients disconnected", n) + } +} + +// TestStress_ClientChurn repeatedly connects and disconnects clients while a +// stable set of signals is being delivered to ensure subscribe/unsubscribe under +// load doesn't cause races, panics, or subscription leaks. +func TestStress_ClientChurn(t *testing.T) { + if testing.Short() { + t.Skip("skipping stress test in short mode") + } + + const ( + nSignals = 50 + nParallelConns = 30 + duration = 3 * time.Second + ) + + srv, brk := newStressServer(t, nSignals) + u := wsURL(srv) + subMsg := buildSubMsg(nSignals) + + deadline := time.Now().Add(duration) + var wg sync.WaitGroup + var total atomic.Int64 + + for range nParallelConns { + wg.Add(1) + go func() { + defer wg.Done() + for time.Now().Before(deadline) { + // Each iteration: one short-lived connection (~500 ms). + func() { + connCtx, connCancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer connCancel() + + conn, _, err := websocket.Dial(connCtx, u, nil) + if err != nil { + return // dial failed (server busy) — try next iteration + } + defer conn.CloseNow() + + if err := conn.Write(connCtx, websocket.MessageText, subMsg); err != nil { + return + } + var n int64 + for { + _, _, err := conn.Read(connCtx) + if err != nil { + break + } + n++ + } + total.Add(n) + }() + } + }() + } + + wg.Wait() + + t.Logf("ClientChurn: %d total messages during %s of churn with %d concurrent connections", + total.Load(), duration, nParallelConns) + + time.Sleep(500 * time.Millisecond) + if n := brk.ActiveSubscriptions(); n != 0 { + t.Errorf("subscription leak after churn: %d remain", n) + } +} + +// TestStress_SlowConsumer verifies that a slow-reading client doesn't block +// fast clients or the broker fan-out (drop-on-full behaviour). +func TestStress_SlowConsumer(t *testing.T) { + if testing.Short() { + t.Skip("skipping stress test in short mode") + } + + const ( + nSignals = 100 + duration = 3 * time.Second + ) + + srv, brk := newStressServer(t, nSignals) + url := wsURL(srv) + subMsg := buildSubMsg(nSignals) + + ctx, cancel := context.WithTimeout(t.Context(), duration+5*time.Second) + defer cancel() + + var fastTotal, slowTotal atomic.Int64 + var wg sync.WaitGroup + + // Fast client: reads as quickly as possible. + wg.Add(1) + fastCtx, fastCancel := context.WithTimeout(ctx, duration) + go func() { + defer fastCancel() + connectAndCount(fastCtx, t, url, subMsg, &fastTotal, &wg) + }() + + // Slow client: sleeps 100 ms between each read. + wg.Add(1) + slowCtx, slowCancel := context.WithTimeout(ctx, duration) + go func() { + defer wg.Done() + defer slowCancel() + + conn, _, err := websocket.Dial(slowCtx, url, nil) + if err != nil { + t.Errorf("slow client dial: %v", err) + return + } + defer conn.CloseNow() + + if err := conn.Write(slowCtx, websocket.MessageText, subMsg); err != nil { + t.Errorf("slow client subscribe: %v", err) + return + } + + var n int64 + for { + time.Sleep(100 * time.Millisecond) + _, _, err := conn.Read(slowCtx) + if err != nil { + break + } + n++ + } + slowTotal.Add(n) + }() + + wg.Wait() + + fast := fastTotal.Load() + slow := slowTotal.Load() + t.Logf("SlowConsumer: fast=%d updates, slow=%d updates in %s", fast, slow, duration) + + // Fast client must receive substantially more than slow client. + if fast < slow*3 { + t.Errorf("fast client should receive many more updates than slow client (fast=%d slow=%d)", fast, slow) + } + + // Fast client must have received a meaningful number of updates. + minFast := int64(nSignals) * int64(duration.Seconds()) * 5 + if fast < minFast { + t.Errorf("fast client received too few updates: got %d, want >= %d", fast, minFast) + } + + time.Sleep(300 * time.Millisecond) + if n := brk.ActiveSubscriptions(); n != 0 { + t.Errorf("subscription leak: %d remain", n) + } +} diff --git a/internal/storage/store.go b/internal/storage/store.go index f15b039..525a54b 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -22,9 +22,11 @@ type InterfaceMeta struct { Version int `json:"version"` } -// Store persists interface definitions as XML files under {storageDir}/interfaces/. +// Store persists interface definitions as XML files under {storageDir}/interfaces/ +// and workspace-level JSON blobs directly in {storageDir}. type Store struct { - dir string + rootDir string // {storageDir} — for workspace-level files (groups.json, etc.) + dir string // {storageDir}/interfaces } // New opens (and creates, if needed) the storage directory. @@ -33,7 +35,22 @@ func New(storageDir string) (*Store, error) { if err := os.MkdirAll(dir, 0o755); err != nil { return nil, fmt.Errorf("create interfaces dir: %w", err) } - return &Store{dir: dir}, nil + return &Store{rootDir: storageDir, dir: dir}, nil +} + +// ReadGroups returns the raw JSON array of group tree nodes. +// Returns an empty JSON array if no groups have been saved yet. +func (s *Store) ReadGroups() ([]byte, error) { + data, err := os.ReadFile(filepath.Join(s.rootDir, "groups.json")) + if errors.Is(err, os.ErrNotExist) { + return []byte("[]"), nil + } + return data, err +} + +// WriteGroups persists the group tree. data must be a valid JSON array. +func (s *Store) WriteGroups(data []byte) error { + return os.WriteFile(filepath.Join(s.rootDir, "groups.json"), data, 0o644) } // validateID returns an error if id could be used for path traversal or is diff --git a/web/src/GroupsTree.tsx b/web/src/GroupsTree.tsx new file mode 100644 index 0000000..9074847 --- /dev/null +++ b/web/src/GroupsTree.tsx @@ -0,0 +1,327 @@ +import { h } from 'preact'; +import { useState, useEffect } from 'preact/hooks'; +import type { GroupNode, SignalRef } from './lib/types'; + +interface Props { + onDragStart?: (sig: SignalRef) => void; +} + +// ── Pure tree helpers ────────────────────────────────────────────────────────── + +function genId(): string { + return Math.random().toString(36).slice(2, 10); +} + +/** Insert a child node at the end of the folder at `path` (empty = root). */ +function insertAt(nodes: GroupNode[], path: number[], child: GroupNode): GroupNode[] { + if (path.length === 0) return [...nodes, child]; + return nodes.map((n, i) => { + if (i !== path[0] || n.kind !== 'folder') return n; + return { ...n, children: insertAt(n.children, path.slice(1), child) }; + }); +} + +/** Remove the node at `path`. */ +function removeAt(nodes: GroupNode[], path: number[]): GroupNode[] { + if (path.length === 1) return nodes.filter((_, i) => i !== path[0]); + return nodes.map((n, i) => { + if (i !== path[0] || n.kind !== 'folder') return n; + return { ...n, children: removeAt(n.children, path.slice(1)) }; + }); +} + +/** Update the folder label at `path`. */ +function renameAt(nodes: GroupNode[], path: number[], label: string): GroupNode[] { + if (path.length === 1) { + return nodes.map((n, i) => + i === path[0] && n.kind === 'folder' ? { ...n, label } : n + ); + } + return nodes.map((n, i) => { + if (i !== path[0] || n.kind !== 'folder') return n; + return { ...n, children: renameAt(n.children, path.slice(1), label) }; + }); +} + +// ── Parse signal input ───────────────────────────────────────────────────────── + +/** Parse "ds:name" or bare "name" (defaults to epics). */ +function parseSignal(input: string): { ds: string; name: string } | null { + const s = input.trim(); + if (!s) return null; + const colon = s.indexOf(':'); + if (colon > 0) return { ds: s.slice(0, colon), name: s.slice(colon + 1) }; + return { ds: 'epics', name: s }; +} + +// ── Component ────────────────────────────────────────────────────────────────── + +export default function GroupsTree({ onDragStart }: Props) { + const [nodes, setNodes] = useState([]); + const [loading, setLoading] = useState(true); + const [collapsed, setCollapsed] = useState>(new Set()); + + // What the user is currently adding inline (null = nothing) + const [addingAt, setAddingAt] = useState<{ path: number[]; kind: 'signal' | 'folder' } | null>(null); + const [addInput, setAddInput] = useState(''); + + // Inline folder rename + const [renamingPath, setRenamingPath] = useState(null); + const [renameInput, setRenameInput] = useState(''); + + // ── Load / save ───────────────────────────────────────────────────────────── + + useEffect(() => { + fetch('/api/v1/groups') + .then(r => r.json()) + .then((data: GroupNode[]) => { setNodes(data); setLoading(false); }) + .catch(() => setLoading(false)); + }, []); + + function save(next: GroupNode[]) { + fetch('/api/v1/groups', { + method: 'PUT', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(next), + }).catch(e => console.error('groups save failed:', e)); + } + + function update(next: GroupNode[]) { + setNodes(next); + save(next); + } + + // ── Drag ──────────────────────────────────────────────────────────────────── + + function makeDraggable(ds: string, name: string) { + return { + draggable: true as const, + onDragStart: (e: DragEvent) => { + const ref: SignalRef = { ds, name }; + e.dataTransfer?.setData('application/json', JSON.stringify(ref)); + onDragStart?.(ref); + }, + }; + } + + // ── Inline add ────────────────────────────────────────────────────────────── + + function startAdding(path: number[], kind: 'signal' | 'folder') { + setAddingAt({ path, kind }); + setAddInput(''); + } + + function commitAdd() { + if (!addingAt) return; + const { path, kind } = addingAt; + const val = addInput.trim(); + setAddingAt(null); + setAddInput(''); + if (!val) return; + if (kind === 'folder') { + const node: GroupNode = { kind: 'folder', id: genId(), label: val, children: [] }; + update(insertAt(nodes, path, node)); + } else { + const sig = parseSignal(val); + if (!sig) return; + const node: GroupNode = { kind: 'signal', ds: sig.ds, name: sig.name }; + update(insertAt(nodes, path, node)); + } + } + + function cancelAdd() { + setAddingAt(null); + setAddInput(''); + } + + // ── Rename ────────────────────────────────────────────────────────────────── + + function startRename(path: number[], label: string) { + setRenamingPath(path); + setRenameInput(label); + } + + function commitRename() { + if (!renamingPath || !renameInput.trim()) { cancelRename(); return; } + update(renameAt(nodes, renamingPath, renameInput.trim())); + setRenamingPath(null); + } + + function cancelRename() { + setRenamingPath(null); + setRenameInput(''); + } + + // ── Collapse ──────────────────────────────────────────────────────────────── + + function toggleCollapse(id: string) { + setCollapsed(prev => { + const next = new Set(prev); + if (next.has(id)) next.delete(id); else next.add(id); + return next; + }); + } + + // ── Render ────────────────────────────────────────────────────────────────── + + function renderAddRow(path: number[], kind: 'signal' | 'folder') { + const isActive = addingAt && + addingAt.kind === kind && + JSON.stringify(addingAt.path) === JSON.stringify(path); + if (!isActive) return null; + return ( +
+ setAddInput((e.target as HTMLInputElement).value)} + onKeyDown={(e: KeyboardEvent) => { + if (e.key === 'Enter') commitAdd(); + if (e.key === 'Escape') cancelAdd(); + }} + /> + + +
+ ); + } + + function renderNode(node: GroupNode, path: number[]) { + const pathKey = JSON.stringify(path); + + if (node.kind === 'signal') { + return ( +
+ {node.ds} + {node.name} + +
+ ); + } + + // folder node + const isCollapsed = collapsed.has(node.id); + const isRenaming = renamingPath !== null && JSON.stringify(renamingPath) === pathKey; + + return ( +
+
+ toggleCollapse(node.id)} + > + {isCollapsed ? '▸' : '▾'} + + + {isRenaming ? ( + setRenameInput((e.target as HTMLInputElement).value)} + onKeyDown={(e: KeyboardEvent) => { + if (e.key === 'Enter') commitRename(); + if (e.key === 'Escape') cancelRename(); + }} + onBlur={commitRename} + /> + ) : ( + startRename(path, node.label)} + title="Double-click to rename" + > + {node.label} + + )} + + {node.children.length} + +
+ + + +
+
+ + {!isCollapsed && ( +
+ {node.children.map((child, i) => renderNode(child, [...path, i]))} + {renderAddRow(path, 'signal')} + {renderAddRow(path, 'folder')} +
+ )} +
+ ); + } + + const isAddingRoot = addingAt !== null && addingAt.path.length === 0; + + return ( +
+ {/* Root-level toolbar */} +
+ +
+ + {/* Inline input for new root folder */} + {isAddingRoot && addingAt?.kind === 'folder' && ( +
+ setAddInput((e.target as HTMLInputElement).value)} + onKeyDown={(e: KeyboardEvent) => { + if (e.key === 'Enter') commitAdd(); + if (e.key === 'Escape') cancelAdd(); + }} + /> + + +
+ )} + +
+ {loading ? ( +

Loading groups…

+ ) : nodes.length === 0 && !isAddingRoot ? ( +

No groups yet — click "+ Group" to create one.

+ ) : ( + nodes.map((node, i) => renderNode(node, [i])) + )} +
+
+ ); +} diff --git a/web/src/SignalTree.tsx b/web/src/SignalTree.tsx index 41dda32..da06efe 100644 --- a/web/src/SignalTree.tsx +++ b/web/src/SignalTree.tsx @@ -1,8 +1,11 @@ -import { h } from 'preact'; +import { h, Fragment } from 'preact'; import { useState, useEffect, useRef } from 'preact/hooks'; import type { SignalRef } from './lib/types'; import SyntheticWizard from './SyntheticWizard'; import SyntheticEditor from './SyntheticEditor'; +import GroupsTree from './GroupsTree'; + +type TreeTab = 'sources' | 'groups'; interface SignalInfo { name: string; @@ -41,6 +44,7 @@ interface Props { } export default function SignalTree({ onDragStart, width }: Props) { + const [treeTab, setTreeTab] = useState('sources'); const [collapsed, setCollapsed] = useState(false); const [groups, setGroups] = useState([]); const [customSignals, setCustomSignals] = useState>(loadCustom); @@ -222,6 +226,26 @@ export default function SignalTree({ onDragStart, width }: Props) { {!collapsed && (
+ {/* Tab bar: Sources | Groups */} +
+ + +
+ + {/* Groups tab */} + {treeTab === 'groups' && ( + + )} + + {/* Sources tab contents */} + {treeTab === 'sources' && <> + {/* Search / filter bar */}
@@ -332,6 +356,7 @@ export default function SignalTree({ onDragStart, width }: Props) { }) )}
+ }
)} diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index 2da133e..e16a416 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -40,6 +40,12 @@ export interface HistoryPoint { value: any; } +// A node in the user-defined signal group tree. +// Folders contain children (other folders or signals); signals are leaf nodes. +export type GroupNode = + | { kind: 'folder'; id: string; label: string; children: GroupNode[] } + | { kind: 'signal'; ds: string; name: string }; + // A complete HMI interface definition export interface Interface { id: string; diff --git a/web/src/styles.css b/web/src/styles.css index 819bbf1..7693886 100644 --- a/web/src/styles.css +++ b/web/src/styles.css @@ -1185,6 +1185,145 @@ body { min-height: 0; } +/* ── Signal tree tab bar ──────────────────────────────────────────────────── */ + +.signal-tree-tabs { + display: flex; + border-bottom: 1px solid #2d3748; + flex-shrink: 0; +} + +.stab { + flex: 1; + background: none; + border: none; + border-bottom: 2px solid transparent; + color: #64748b; + cursor: pointer; + font-size: 0.75rem; + font-weight: 600; + padding: 0.35rem 0; + text-transform: uppercase; + letter-spacing: 0.04em; + transition: color 0.15s, border-color 0.15s; +} +.stab:hover { color: #94a3b8; } +.stab-active { color: #e2e8f0; border-bottom-color: #4a9eff; } + +/* ── Groups tree ──────────────────────────────────────────────────────────── */ + +.groups-tree { + display: flex; + flex-direction: column; + flex: 1; + min-height: 0; + overflow: hidden; +} + +.groups-toolbar { + display: flex; + gap: 0.25rem; + padding: 0.3rem 0.5rem; + border-bottom: 1px solid #2d3748; + flex-shrink: 0; +} + +.group-list { overflow-y: auto; flex: 1; padding: 0.25rem 0; } + +.group-add-row { + display: flex; + align-items: center; + gap: 0.25rem; + padding: 0.2rem 0.5rem; +} +.group-add-root { padding: 0.25rem 0.5rem; } +.group-add-input { + flex: 1; + background: #0f1117; + border: 1px solid #4a9eff; + border-radius: 3px; + color: #e2e8f0; + font-size: 0.78rem; + padding: 2px 6px; + min-width: 0; +} + +.group-folder { margin-bottom: 1px; } + +.group-folder-header { + display: flex; + align-items: center; + gap: 0.2rem; + padding: 0.15rem 0.4rem; + cursor: default; + border-radius: 3px; + font-size: 0.8rem; + color: #94a3b8; + min-height: 1.7rem; +} +.group-folder-header:hover { background: #1e293b; color: #cbd5e1; } +.group-folder-header:hover .group-folder-actions { opacity: 1; } + +.group-folder-arrow { font-size: 0.65rem; flex-shrink: 0; width: 0.9rem; text-align: center; cursor: pointer; } +.group-folder-name { flex: 1; font-weight: 600; cursor: pointer; user-select: none; } +.group-folder-count { + font-size: 0.65rem; + color: #475569; + background: #1e293b; + border-radius: 8px; + padding: 0 5px; + min-width: 1.1rem; + text-align: center; +} +.group-folder-actions { + display: flex; + gap: 1px; + opacity: 0; + transition: opacity 0.15s; +} +.group-folder-actions .icon-btn { font-size: 0.65rem; padding: 1px 3px; } + +.group-rename-input { + flex: 1; + background: #0f1117; + border: 1px solid #4a9eff; + border-radius: 3px; + color: #e2e8f0; + font-size: 0.8rem; + font-weight: 600; + padding: 1px 4px; + min-width: 0; +} + +.group-folder-children { + padding-left: 1rem; + border-left: 1px solid #1e293b; + margin-left: 0.7rem; +} + +.group-signal-item { + display: flex; + align-items: center; + gap: 0.25rem; + padding: 0.15rem 0.4rem; + font-size: 0.78rem; + color: #94a3b8; + border-radius: 3px; + cursor: grab; + min-height: 1.5rem; +} +.group-signal-item:hover { background: #1e293b; color: #cbd5e1; } +.group-signal-item:hover .group-node-remove { opacity: 1; } + +.group-signal-ds { + color: #475569; + font-size: 0.7rem; + flex-shrink: 0; +} +.group-signal-name { flex: 1; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } + +.group-node-remove { opacity: 0; transition: opacity 0.15s; margin-left: auto; flex-shrink: 0; } + .signal-tree-search { padding: 0.4rem 0.5rem; border-bottom: 1px solid #2d3748;