diff --git a/Dockerfile b/Dockerfile index 094eb45..0bc97c0 100644 --- a/Dockerfile +++ b/Dockerfile @@ -37,7 +37,7 @@ FROM golang:1.25-bookworm AS builder # `main` (or that commit). Pinned to a fixed SHA so release images are # reproducible rather than tracking a floating branch. ARG SAMOYED_REPO=https://github.com/M0LTE/samoyed.git -ARG SAMOYED_REF=6b4f5c7aef633041cb2e55ab063f29ff0bacbefa +ARG SAMOYED_REF=7fdd617fe4acbb67d746b756086db49c716def84 RUN apt-get update && apt-get install -y --no-install-recommends \ git make pkg-config gcc libc6-dev \ @@ -71,6 +71,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ libbsd0 libgps28 libasound2 libjack-jackd2-0 libpulse0 \ direwolf \ ca-certificates \ + libcap2-bin \ && rm -rf /var/lib/apt/lists/* /var/cache/apt/* /var/log/* # binaries @@ -87,6 +88,17 @@ EXPOSE 8080 8001 8002 # A non-root user for the running process. RUN useradd --system --no-create-home --shell /usr/sbin/nologin sim + +# -rt-priority renices the router and its TNC children, which needs +# CAP_SYS_NICE — but the container runs as the non-root `sim` user, and +# docker's --cap-add only populates the *bounding* set (a non-root process +# doesn't inherit it). File capabilities on the binaries grant it to the +# process directly; still inert unless the container ALSO gets +# --cap-add SYS_NICE (file caps can't exceed the bounding set), and the +# binaries only use it when -rt-priority is passed. +RUN setcap cap_sys_nice+ep /usr/local/bin/sim-web \ + && setcap cap_sys_nice+ep /usr/local/bin/sim-router + USER sim ENTRYPOINT ["/usr/local/bin/sim-web", "-addr", ":8080", "-config", "/etc/sim/network.yaml"] diff --git a/README.md b/README.md index 7a5c176..e9e3678 100644 --- a/README.md +++ b/README.md @@ -89,6 +89,30 @@ expected steady state right after startup. Override (`docker run ... ghcr.io/.../net-sim:main` with extra args) if you'd rather drive Start/Stop manually from your test harness via `POST /api/start`. +### Smoother audio under host load (`-rt-priority`) + +The router paces audio with 10 ms tickers and every TNC child runs a +software demodulator; on a busy shared host, scheduler jitter glitches +both (lost ticks → choppy RX audio → decode failures that look like RF +problems). `sim-router -rt-priority` renices the router *and* each +spawned TNC child to `-10`. Deliberately plain niceness, not +`SCHED_FIFO` — a real-time policy could starve the host; niceness is +enough to keep the tickers honest. It's best-effort: without +`CAP_SYS_NICE` you get a one-line warning and the simulation carries on +at normal priority. + +Granting the capability in Docker (`--cap-add SYS_NICE`), or in compose: + +```yaml +services: + net-sim: + image: ghcr.io/packethacking/net-sim:main + cap_add: [SYS_NICE] +``` + + +> **Nested-container hosts:** on a host that is itself an unprivileged container (e.g. Docker inside an unprivileged Proxmox LXC), the kernel checks CAP_SYS_NICE against the *init* user namespace, so negative nice is unavailable to anything inside — even container root. The flag then logs its one-line warning and the sim runs at normal priority; everything else is unaffected. + ## Quick install (curl | sudo bash) On a fresh Debian 12 / Ubuntu 24.04+ host (LXC, VM, bare metal — anywhere @@ -206,7 +230,8 @@ Each demo prints its KISS port assignments at startup. ```yaml mixer_mode: fm_capture # fm_capture (default) | linear_sum (stub) capture_db: 6.0 # FM capture ratio -collision_mode: silence # silence (default) | sum (stub) | noise (stub) +collision_mode: silence # silence (default) | noise (FM garble) | sum (stub) +time_scale: 1.0 # run N x faster than wall clock (>= 1.0; see below) nodes: - id: a @@ -226,12 +251,54 @@ nodes: links: # Directional. Both endpoints must use compatible modem configs. - { from: a.vhf, to: b.vhf, loss_db: 0 } - - { from: b.vhf, to: a.vhf, loss_db: 0 } + - { from: b.vhf, to: a.vhf, loss_db: 0, squelch_open_ms: 50 } ``` +Per-link `squelch_open_ms` (optional, `0..500`, default `0`) models the +receiving radio's squelch / carrier-detect opening delay: the first N ms +of every transmission heard via that link are delivered as silence (the +carrier is still on the air for capture/collision purposes — only the +audio is muted while the squelch opens). Real FM receivers take tens of +milliseconds to open squelch and settle the discriminator, which is +exactly why KISS TXDELAY exists; with the default `0` a tiny TXDELAY +looks fine in simulation when it wouldn't be on air. + Strict parsing: any unknown key inside a `modem:` block (e.g. `baud_rate` when you meant `baud`) is an error at startup, not a silent default. +### time_scale — faster-than-real-time simulation + +> **TNC pacing does not scale (measured):** samoyed paces its transmissions in +> wall-clock time (the real-airtime sleep before PTT release), so at +> `time_scale > 1` the TNC transmits in real time while the channel runs N× +> faster — TX throughput stays wall-clock-bound and ACKMODE echoes arrive N× +> "late" relative to a host whose protocol timers are scaled to match. In +> practice `time_scale` is currently only sound for receive-path/mixer +> experiments; ACKMODE pacing or throughput measurements need `time_scale: 1` +> until the TNC grows a matching speed factor (tracked upstream). + +`time_scale: N` (or the `-time-scale N` flag on `sim-router`, which +overrides the config) runs the whole simulation N× faster than wall +clock: the router divides every pacing interval by N — the 10 ms +per-block RX ticker, the composite recorder's ticker, and the TX +watchdog's tick and silence window (silence detection has to scale with +the audio rate or `tx_end` events would fire mid-transmission). A 60 s +exchange completes in 60/N wall-clock seconds; recordings still come out +as normal 44.1 kHz files whose time axis is *sim* time. + +**Fidelity caveat — read before trusting numbers from a scaled run.** +Only the router's clocks scale. The TNC child processes (samoyed / +direwolf) still run their own wall-clock behaviours — CSMA persist and +slottime waits, DCD hang times, any internal timeouts — which means at +`time_scale: 4` a TNC's 100 ms slottime is effectively 400 ms of sim +time. `time_scale > 1` is therefore an **accelerated-testing mode** (get +through a long soak/protocol exchange quickly), *not* a calibrated CSMA +/ channel-access simulation; for timing-sensitive contention studies run +at `1.0`. Hosts driving the KISS ports must also scale their own +protocol timers (T1/T2 etc.) by N, or their retries will fire N× too +early in sim time. Large factors are also bounded by CPU: every TNC +demodulator must keep up with N× real-time audio. + ### TNC backend per port Each port chooses which TNC implementation runs the modem: @@ -404,12 +471,27 @@ for each rx block per receiving port: if len(active) >= 2: sort by RX level; margin = strongest − next if margin >= capture_db: output strongest, attenuated (capture) - else: collision garbage (silence in v1) + else: collision garbage (per collision_mode) ``` This is what makes the `hidden-node` and `hidden-node-capture` demos do qualitatively different things from the same code. +What "collision garbage" sounds like is selectable via `collision_mode`: + +- `silence` (default) — clean digital silence. Simple and backwards + compatible, but unrealistically *clean*: it hands receiving modems a + perfectly quiet channel at exactly the moment a real one would be full + of noise. +- `noise` — gaussian garble at an RMS matching the strongest colliding + signal's post-loss level. A real FM discriminator outputs loud garble + (the heterodyne beat between the carriers plus wideband noise) when + two comparable carriers collide, at signal-comparable amplitude — so a + hot collision is loud garble, a weak distant one quiet garble. Use + this to exercise modem false-sync / DCD behaviour that the silence + model can't. +- `sum` — accepted but still a stub (behaves as silence). + ## Known limitations (samoyed-side, expected to be fixed upstream) This is a gap in the current samoyed build that affects what you can @@ -450,8 +532,9 @@ round-trip test lives in `internal/tnc/ackmode_test.go`. - BER / FER reporting beyond the basic frame counters demonstrable from KISS sniffing. - SSB modelling, AGC, pre/de-emphasis, multipath, Doppler, fading. -- `linear_sum` / `sum` / `noise` mixer modes — accepted in the YAML, only - `fm_capture` + `silence` are functional in v1. +- `linear_sum` / `sum` mixer modes — accepted in the YAML, only + `fm_capture` is functional (`collision_mode: silence` and `noise` both + work; `sum` is a stub). - Modem modes beyond what samoyed currently supports. ## Layout diff --git a/cmd/sim-router/main.go b/cmd/sim-router/main.go index 2639083..c2c2596 100644 --- a/cmd/sim-router/main.go +++ b/cmd/sim-router/main.go @@ -32,6 +32,8 @@ func main() { workDir := flag.String("workdir", "", "scratch dir for per-port config files / FIFOs (default: a unique subdir of $TMPDIR)") recordDir := flag.String("record", "", "if set, record all per-port TX and RX audio to a timestamped subdirectory of this path") composite := flag.String("composite", "", "comma-separated transmitter ports (e.g. a.vhf,b.vhf) to composite into one real-time, sample-aligned WAV (one TX per channel — stereo for two). Requires -record for the output dir") + timeScale := flag.Float64("time-scale", 0, "run the simulation N x faster than wall clock (>= 1.0; overrides the config's time_scale; see README for the fidelity caveat)") + rtPriority := flag.Bool("rt-priority", false, "renice sim-router and every TNC child to -10 for smoother audio pacing under host load (best-effort; needs CAP_SYS_NICE)") flag.Parse() if *cfgPath == "" { @@ -62,6 +64,19 @@ func main() { logger.Error("load config", "path", *cfgPath, "err", err) os.Exit(1) } + if *timeScale != 0 { + if *timeScale < 1 { + logger.Error("-time-scale must be >= 1.0 (slower-than-real-time is not supported)", "got", *timeScale) + os.Exit(2) + } + cfg.TimeScale = *timeScale + } + if cfg.TimeScale > 1 { + // Accelerated-testing mode, not a calibrated CSMA simulation: the + // TNC children's wall-clock waits (persist/slottime, timeouts) do + // NOT scale. See README "time_scale". + logger.Warn("time_scale active — TNC CSMA timing does not scale; hosts must scale their own protocol timers", "time_scale", cfg.TimeScale) + } samoyedBin, samoyedErr := resolveSamoyed(*samoyedPath) direwolfBin, direwolfErr := resolveDirewolf(*direwolfPath) @@ -95,6 +110,7 @@ func main() { Logger: logger, RecordDir: *recordDir, RecordOnStart: *recordDir != "", + RTPriority: *rtPriority, }) if err != nil { logger.Error("start router", "err", err) diff --git a/cmd/sim-web/main.go b/cmd/sim-web/main.go index c49e6e8..8c56d15 100644 --- a/cmd/sim-web/main.go +++ b/cmd/sim-web/main.go @@ -72,6 +72,7 @@ func main() { workDir := flag.String("workdir", "", "scratch dir for per-port config files / FIFOs (default: temp)") autostart := flag.Bool("autostart", false, "start the router immediately on launch") recordDir := flag.String("record", "", "if set, enables the Record toggle and Composite recording panel in the UI; recordings land under this path") + rtPriority := flag.Bool("rt-priority", false, "renice the router and its TNC children for jitter-free pacing (needs CAP_SYS_NICE)") flag.Parse() logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) @@ -97,6 +98,7 @@ func main() { direwolfBin: direwolfBin, workDir: *workDir, recordBase: *recordDir, + rtPriority: *rtPriority, logger: logger, eventBus: events.NewBus(), audioTap: audio.NewTap(), @@ -179,6 +181,7 @@ type app struct { direwolfBin string workDir string recordBase string // -record DIR; "" means feature disabled + rtPriority bool // -rt-priority: renice router + TNC children at start logger *slog.Logger tmpl *template.Template mapTmpl *template.Template @@ -790,6 +793,7 @@ func (a *app) start() error { EventBus: a.eventBus, AudioTap: a.audioTap, Observer: a.observer, + RTPriority: a.rtPriority, }) if err != nil { cancel() diff --git a/internal/audio/format.go b/internal/audio/format.go index ef03311..10aaaf0 100644 --- a/internal/audio/format.go +++ b/internal/audio/format.go @@ -64,6 +64,23 @@ func (b Block) IsSilence() bool { return true } +// RMS returns the root-mean-square sample amplitude of the block +// (0..32768). Used by the collision "noise" model to size the garble to +// the colliding signals' level. +func (b Block) RMS() float64 { + var sum float64 + n := 0 + for i := 0; i+1 < len(b); i += 2 { + s := int16(uint16(b[i]) | uint16(b[i+1])<<8) + sum += float64(s) * float64(s) + n++ + } + if n == 0 { + return 0 + } + return math.Sqrt(sum / float64(n)) +} + // PeakAbs returns the peak absolute amplitude in the block (0..32768). // Used as a rough RX-level estimator for the capture-effect mixer. func (b Block) PeakAbs() int { diff --git a/internal/audio/mixer.go b/internal/audio/mixer.go index bc420b6..be061e3 100644 --- a/internal/audio/mixer.go +++ b/internal/audio/mixer.go @@ -101,16 +101,50 @@ func (m *Mixer) Mix(active []ActiveTX) (Block, MixDecision) { return m.collision(sorted), MixCollision } -func (m *Mixer) collision(_ []ActiveTX) Block { +// collision renders the receiver's output when 2+ signals land inside the +// capture margin and none of them takes the demodulator. sorted is the +// active set ordered strongest-first (by receiver-side Level). +func (m *Mixer) collision(sorted []ActiveTX) Block { switch m.CollisionMode { - case "sum", "noise": - // stubs only — silence is the simplest defensible model and the - // only branch v1 needs working + case "noise": + // FM capture below threshold does NOT yield silence: with two + // comparable carriers on channel, the discriminator output is + // loud garble — the heterodyne beat between the carriers plus + // wideband noise — at an amplitude comparable to the signals + // themselves. Pure digital silence is unrealistically clean: it + // gives receiving modems a perfectly quiet channel exactly when + // a real one would be hammering their DCD and false-sync + // behaviour. Model the garble as gaussian noise with RMS equal + // to the strongest signal's post-loss level, so a hot collision + // is loud garble and a weak distant one is quiet garble. + return m.noiseBlock(sorted[0].Block.RMS() * AmplitudeFromDB(sorted[0].LossDB)) + case "sum": + // stub — behaves as silence until the SSB path needs it return Silence() } + // "silence" (the default): the v1 model, kept for backwards + // compatibility. return Silence() } +// noiseBlock returns a fresh block of gaussian noise with the given RMS +// amplitude (sigma, in sample units, 0..32767). Routed through AddNoise so +// the RNG locking and int16 clamping live in one place. +func (m *Mixer) noiseBlock(sigma float64) Block { + b := Silence() + if sigma <= 0 { + return b + } + // AddNoise expresses level as positive dB below full-scale; convert, + // clamping at "full-scale garble" for sigma at/above full-scale. + noiseDB := 20 * math.Log10(float64(math.MaxInt16)/sigma) + if noiseDB <= 0 { + noiseDB = 0.01 + } + m.AddNoise(b, noiseDB) + return b +} + func (m *Mixer) linearSum(active []ActiveTX) Block { if len(active) == 0 { return Silence() diff --git a/internal/audio/mixer_test.go b/internal/audio/mixer_test.go index 7068b8d..1812e29 100644 --- a/internal/audio/mixer_test.go +++ b/internal/audio/mixer_test.go @@ -88,6 +88,83 @@ func TestMixCollisionMarginAtThreshold(t *testing.T) { } } +func TestMixCollisionNoiseProducesGarble(t *testing.T) { + m := NewMixer(6, false, "noise") + a := mkBlock(20000) + b := mkBlock(20000) + // Both 6 dB down → margin = 0 < capture threshold → collision. + // Strongest post-loss RMS = 20000 × 10^(-6/20) ≈ 10024; the garble + // must come out at a comparable level, not silence and not full-scale. + out, dec := m.Mix([]ActiveTX{ + {Block: a, LossDB: 6}, + {Block: b, LossDB: 6}, + }) + if dec != MixCollision { + t.Fatalf("decision = %d, want MixCollision", dec) + } + if out.IsSilence() { + t.Fatal("noise-mode collision must not be silent") + } + got := out.RMS() + want := 20000 * AmplitudeFromDB(6) + if got < 0.7*want || got > 1.3*want { + t.Errorf("garble rms = %.0f, want within 30%% of strongest post-loss level %.0f", got, want) + } +} + +func TestMixCollisionNoiseLevelTracksStrongest(t *testing.T) { + m := NewMixer(6, false, "noise") + a := mkBlock(20000) + b := mkBlock(20000) + // A distant collision (both 30 dB down) is QUIET garble. + out, dec := m.Mix([]ActiveTX{ + {Block: a, LossDB: 30}, + {Block: b, LossDB: 30}, + }) + if dec != MixCollision { + t.Fatalf("decision = %d, want MixCollision", dec) + } + got := out.RMS() + want := 20000 * AmplitudeFromDB(30) // ≈ 632 + if got < 0.7*want || got > 1.3*want { + t.Errorf("weak-collision garble rms = %.0f, want within 30%% of %.0f", got, want) + } +} + +// Capture (margin >= capture_db) must be byte-identical between noise +// mode and the default — the noise model only changes what a *collision* +// sounds like. +func TestMixCaptureUnchangedInNoiseMode(t *testing.T) { + strong := mkBlock(20000) + weak := mkBlock(5000) + active := []ActiveTX{ + {Block: weak, LossDB: 12}, + {Block: strong, LossDB: 0}, + } + noiseOut, noiseDec := NewMixer(6, false, "noise").Mix(active) + silenceOut, silenceDec := NewMixer(6, false, "silence").Mix(active) + if noiseDec != MixCapture || silenceDec != MixCapture { + t.Fatalf("decisions = %d / %d, want MixCapture for both", noiseDec, silenceDec) + } + if string(noiseOut) != string(silenceOut) { + t.Error("capture output differs between noise and silence collision modes") + } +} + +// Single-source mixing must also be untouched by the collision mode. +func TestMixSingleUnchangedInNoiseMode(t *testing.T) { + src := mkBlock(10000) + active := []ActiveTX{{Block: src, LossDB: 6}} + noiseOut, noiseDec := NewMixer(6, false, "noise").Mix(active) + silenceOut, silenceDec := NewMixer(6, false, "silence").Mix(active) + if noiseDec != MixSingle || silenceDec != MixSingle { + t.Fatalf("decisions = %d / %d, want MixSingle for both", noiseDec, silenceDec) + } + if string(noiseOut) != string(silenceOut) { + t.Error("single-source output differs between noise and silence collision modes") + } +} + func TestAmplitudeFromDB(t *testing.T) { cases := []struct { lossDB float64 diff --git a/internal/config/config.go b/internal/config/config.go index 29f6fec..497fd52 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -123,6 +123,15 @@ type Link struct { // NoiseDB is the white-noise floor on this link, expressed as dB // below full-scale (positive = quieter; 0 = no noise). Phase 4. NoiseDB float64 `yaml:"noise_db,omitempty"` + + // SquelchOpenMS mutes the first N milliseconds of every transmission + // as heard via this link, modelling the receiving radio's squelch / + // carrier-detect opening delay (tens of ms on real FM gear, plus + // discriminator settle). This is exactly the on-air reason KISS + // TXDELAY exists; without it a tiny TXDELAY looks fine in simulation + // when it wouldn't be on air. 0 (the default) = squelch opens + // instantly — previous behaviour. Valid range 0..500. + SquelchOpenMS float64 `yaml:"squelch_open_ms,omitempty"` } // MixerMode selects the receiver-side mixing model. @@ -138,9 +147,9 @@ const ( type CollisionMode string const ( - CollisionSilence CollisionMode = "silence" // v1 default + CollisionSilence CollisionMode = "silence" // default — clean digital silence CollisionSum CollisionMode = "sum" // stub - CollisionNoise CollisionMode = "noise" // stub + CollisionNoise CollisionMode = "noise" // gaussian garble at the strongest signal's level ) // Config is the whole topology file. @@ -156,6 +165,21 @@ type Config struct { // global floor (back to the old per-link-only behaviour). DefaultNoiseDB float64 `yaml:"default_noise_db,omitempty"` + // TimeScale runs the simulation N× faster than wall clock (default + // 1.0 = real time; values < 1.0 are rejected). The router divides + // every wall-clock pacing interval by this factor: the rxFeeder's + // block ticker, the composite recorder's ticker, and the TX + // watchdog's tick + silence window (silence detection must scale + // with the audio rate or tx_end fires mid-transmission). + // + // Fidelity caveat: only the *router's* clocks scale. The TNC child + // processes' own wall-clock behaviours — CSMA persist/slottime + // waits, any internal timeouts — do NOT scale, so time_scale > 1 is + // an accelerated-testing mode, not a calibrated CSMA simulation. + // Hosts driving the KISS ports must scale their own protocol timers + // (T1/T2) to match, or retries will fire N× early in sim time. + TimeScale float64 `yaml:"time_scale,omitempty"` + Nodes []Node `yaml:"nodes"` Links []Link `yaml:"links"` } @@ -204,6 +228,9 @@ func (c *Config) applyDefaults() { if c.CollisionMode == "" { c.CollisionMode = CollisionSilence } + if c.TimeScale == 0 { + c.TimeScale = 1.0 + } } // PortRef is a (node, port) handle resolved from a "node.port" string. @@ -233,6 +260,9 @@ func (c *Config) Validate() error { if c.CaptureDB < 0 { return fmt.Errorf("config: capture_db must be >= 0, got %g", c.CaptureDB) } + if c.TimeScale < 1 { + return fmt.Errorf("config: time_scale must be >= 1.0, got %g (slower-than-real-time is not supported)", c.TimeScale) + } // Build a lookup table and check ID uniqueness. type slot struct { @@ -312,6 +342,9 @@ func (c *Config) Validate() error { if l.LossDB < 0 { return fmt.Errorf("config: link %s -> %s: loss_db must be >= 0", fromRef, toRef) } + if l.SquelchOpenMS < 0 || l.SquelchOpenMS > 500 { + return fmt.Errorf("config: link %s -> %s: squelch_open_ms must be in 0..500, got %g", fromRef, toRef, l.SquelchOpenMS) + } key := fromRef.String() + "->" + toRef.String() if seenLink[key] { return fmt.Errorf("config: duplicate link %s", key) diff --git a/internal/config/config_test.go b/internal/config/config_test.go index eae51fd..6f0db62 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -164,6 +164,113 @@ links: } } +func TestTimeScaleDefaultsToRealTime(t *testing.T) { + yaml := ` +nodes: + - id: a + ports: + - id: vhf + modem: { mode: afsk1200 } + kiss_port: 8001 +links: [] +` + cfg, err := Parse(strings.NewReader(yaml)) + if err != nil { + t.Fatalf("expected ok, got %v", err) + } + if cfg.TimeScale != 1.0 { + t.Errorf("time_scale default = %g, want 1.0", cfg.TimeScale) + } +} + +func TestTimeScaleAccepted(t *testing.T) { + yaml := ` +time_scale: 8.0 +nodes: + - id: a + ports: + - id: vhf + modem: { mode: afsk1200 } + kiss_port: 8001 +links: [] +` + cfg, err := Parse(strings.NewReader(yaml)) + if err != nil { + t.Fatalf("expected ok, got %v", err) + } + if cfg.TimeScale != 8.0 { + t.Errorf("time_scale = %g, want 8.0", cfg.TimeScale) + } +} + +func TestTimeScaleRejectsSlowerThanRealTime(t *testing.T) { + for _, scale := range []string{"0.5", "-2"} { + yaml := ` +time_scale: ` + scale + ` +nodes: + - id: a + ports: + - id: vhf + modem: { mode: afsk1200 } + kiss_port: 8001 +links: [] +` + _, err := Parse(strings.NewReader(yaml)) + if err == nil || !strings.Contains(err.Error(), "time_scale") { + t.Fatalf("time_scale=%s: expected time_scale error, got %v", scale, err) + } + } +} + +func TestSquelchOpenMSAccepted(t *testing.T) { + yaml := ` +nodes: + - id: a + ports: + - id: vhf + modem: { mode: afsk1200 } + kiss_port: 8001 + - id: b + ports: + - id: vhf + modem: { mode: afsk1200 } + kiss_port: 8002 +links: + - { from: a.vhf, to: b.vhf, loss_db: 0, squelch_open_ms: 50 } +` + cfg, err := Parse(strings.NewReader(yaml)) + if err != nil { + t.Fatalf("expected ok, got %v", err) + } + if cfg.Links[0].SquelchOpenMS != 50 { + t.Errorf("squelch_open_ms = %g, want 50", cfg.Links[0].SquelchOpenMS) + } +} + +func TestSquelchOpenMSRange(t *testing.T) { + for _, v := range []string{"-1", "501"} { + yaml := ` +nodes: + - id: a + ports: + - id: vhf + modem: { mode: afsk1200 } + kiss_port: 8001 + - id: b + ports: + - id: vhf + modem: { mode: afsk1200 } + kiss_port: 8002 +links: + - { from: a.vhf, to: b.vhf, loss_db: 0, squelch_open_ms: ` + v + ` } +` + _, err := Parse(strings.NewReader(yaml)) + if err == nil || !strings.Contains(err.Error(), "squelch_open_ms") { + t.Fatalf("squelch_open_ms=%s: expected range error, got %v", v, err) + } + } +} + func TestSelfLoopRejected(t *testing.T) { yaml := ` nodes: diff --git a/internal/router/composite.go b/internal/router/composite.go index a2fb16c..e93a440 100644 --- a/internal/router/composite.go +++ b/internal/router/composite.go @@ -36,6 +36,13 @@ type compositeRecorder struct { logger *slog.Logger started time.Time + // timeScale is the simulation's acceleration factor; the pacing + // ticker runs at scaled(blockPeriod, timeScale) so the composite + // stays sample-aligned with the (faster-than-wall-clock) sim + // timeline. The WAV itself is still a SampleRate file — its time + // axis is sim time, not wall-clock time. + timeScale float64 + idx map[config.PortRef]int // port → channel index queues []chan audio.Block // one per channel; single producer each writer *audio.MultiWAVWriter @@ -75,7 +82,7 @@ type CompositeStatus struct { Err string } -func newCompositeRecorder(base string, channels []config.PortRef, logger *slog.Logger) (*compositeRecorder, error) { +func newCompositeRecorder(base string, channels []config.PortRef, timeScale float64, logger *slog.Logger) (*compositeRecorder, error) { if base == "" { return nil, errors.New("composite: no record base dir configured") } @@ -101,15 +108,16 @@ func newCompositeRecorder(base string, channels []config.PortRef, logger *slog.L } cr := &compositeRecorder{ - path: path, - channels: append([]config.PortRef(nil), channels...), - logger: logger, - started: time.Now(), - idx: idx, - queues: make([]chan audio.Block, len(channels)), - writer: w, - done: make(chan struct{}), - stopped: make(chan struct{}), + path: path, + channels: append([]config.PortRef(nil), channels...), + logger: logger, + started: time.Now(), + timeScale: timeScale, + idx: idx, + queues: make([]chan audio.Block, len(channels)), + writer: w, + done: make(chan struct{}), + stopped: make(chan struct{}), } for i := range cr.queues { cr.queues[i] = make(chan audio.Block, compositeQueueBlocks) @@ -153,8 +161,7 @@ func (cr *compositeRecorder) feed(ref config.PortRef, blk audio.Block) { // and every channel stays sample-aligned. func (cr *compositeRecorder) run() { defer close(cr.stopped) - period := time.Duration(audio.BlockSamples) * time.Second / audio.SampleRate - ticker := time.NewTicker(period) + ticker := time.NewTicker(scaled(blockPeriod, cr.timeScale)) defer ticker.Stop() for { select { @@ -312,7 +319,7 @@ func (r *Router) StartCompositeRecording(ports []config.PortRef) (string, error) } } - cr, err := newCompositeRecorder(r.opts.RecordDir, ports, r.logger) + cr, err := newCompositeRecorder(r.opts.RecordDir, ports, r.cfg.TimeScale, r.logger) if err != nil { return "", err } diff --git a/internal/router/linkqueue_test.go b/internal/router/linkqueue_test.go index a9aa9f6..affb7d5 100644 --- a/internal/router/linkqueue_test.go +++ b/internal/router/linkqueue_test.go @@ -18,7 +18,7 @@ func TestLinkQueueDeliversLongBurstGapFree(t *testing.T) { q := newLinkQueue( config.PortRef{NodeID: "a", PortID: "vhf"}, config.PortRef{NodeID: "b", PortID: "vhf"}, - 0, 0, + 0, 0, 0, txSilenceWindow, ) // ~10 s of audio — well past the old 3 s cap that used to drop. diff --git a/internal/router/priority.go b/internal/router/priority.go new file mode 100644 index 0000000..10baa8a --- /dev/null +++ b/internal/router/priority.go @@ -0,0 +1,39 @@ +package router + +import ( + "errors" + "log/slog" + "syscall" +) + +// rtNice is the niceness applied to sim-router and its TNC children when +// Options.RTPriority is set. The 10 ms pacing tickers and the children's +// software demodulators glitch under shared host load; a modest negative +// niceness keeps them ahead of batch work without the host-starvation +// risk of a real-time policy — deliberately plain niceness, not +// SCHED_FIFO. +const rtNice = -10 + +// setPriority is syscall.Setpriority(PRIO_PROCESS, ...), indirected so +// tests can stub it (raising priority needs CAP_SYS_NICE, which tests +// don't have and shouldn't need). +var setPriority = func(pid, nice int) error { + return syscall.Setpriority(syscall.PRIO_PROCESS, pid, nice) +} + +// applyRTPriority renices one process (pid 0 = the calling process) to +// rtNice. Best-effort by design: without CAP_SYS_NICE the kernel returns +// EPERM/EACCES, which earns a single clear warning and the simulation +// carries on at normal priority. +func applyRTPriority(logger *slog.Logger, target string, pid int) { + err := setPriority(pid, rtNice) + if err == nil { + return + } + if errors.Is(err, syscall.EPERM) || errors.Is(err, syscall.EACCES) { + logger.Warn("cannot raise scheduling priority — run with CAP_SYS_NICE (docker: --cap-add SYS_NICE / compose: cap_add: [SYS_NICE])", + "target", target, "pid", pid, "nice", rtNice) + return + } + logger.Warn("setpriority failed", "target", target, "pid", pid, "nice", rtNice, "err", err) +} diff --git a/internal/router/priority_test.go b/internal/router/priority_test.go new file mode 100644 index 0000000..265c2a7 --- /dev/null +++ b/internal/router/priority_test.go @@ -0,0 +1,128 @@ +package router + +import ( + "bytes" + "context" + "log/slog" + "strings" + "sync" + "syscall" + "testing" + + "github.com/packethacking/net-sim/internal/config" +) + +// stubSetPriority replaces the setpriority syscall hook for the duration +// of a test, recording every call. Restores the real one on cleanup. +func stubSetPriority(t *testing.T, err error) *priorityCalls { + t.Helper() + calls := &priorityCalls{} + orig := setPriority + setPriority = func(pid, nice int) error { + calls.mu.Lock() + calls.pids = append(calls.pids, pid) + calls.nices = append(calls.nices, nice) + calls.mu.Unlock() + return err + } + t.Cleanup(func() { setPriority = orig }) + return calls +} + +type priorityCalls struct { + mu sync.Mutex + pids []int + nices []int +} + +func TestApplyRTPriorityInvokesSetpriority(t *testing.T) { + calls := stubSetPriority(t, nil) + applyRTPriority(quietLogger(), "a.vhf", 4242) + if len(calls.pids) != 1 || calls.pids[0] != 4242 { + t.Fatalf("setpriority pids = %v, want [4242]", calls.pids) + } + if calls.nices[0] != rtNice { + t.Errorf("niceness = %d, want %d", calls.nices[0], rtNice) + } +} + +// TestApplyRTPriorityEPermWarnsAndContinues: without CAP_SYS_NICE the +// kernel returns EPERM; that must produce a single actionable warning +// (mentioning the capability) and no failure. +func TestApplyRTPriorityEPermWarnsAndContinues(t *testing.T) { + stubSetPriority(t, syscall.EPERM) + var buf bytes.Buffer + logger := slog.New(slog.NewTextHandler(&buf, nil)) + applyRTPriority(logger, "sim-router", 0) // must not panic / propagate + out := buf.String() + if !strings.Contains(out, "CAP_SYS_NICE") { + t.Errorf("EPERM warning should tell the operator about CAP_SYS_NICE; got %q", out) + } + if !strings.Contains(out, "level=WARN") { + t.Errorf("expected a WARN-level log line, got %q", out) + } +} + +// TestStartHonoursRTPriorityOption: the Options.RTPriority flag must +// reach the setpriority hook for the router's own process (pid 0) before +// children spawn. Children can't be started in a unit test (no TNC +// binary), so Start is allowed to fail afterwards — by then the self +// renice has either happened or the plumbing is broken. +func TestStartHonoursRTPriorityOption(t *testing.T) { + calls := stubSetPriority(t, nil) + cfg := &config.Config{ + TimeScale: 1, + Nodes: []config.Node{{ID: "a", Ports: []config.Port{{ + ID: "vhf", + Modem: config.Modem{Mode: config.ModeAFSK1200}, + KissPort: 18001, + }}}}, + } + _, err := Start(context.Background(), cfg, Options{ + RTPriority: true, + SamoyedBin: "/nonexistent/samoyed-direwolf", + WorkDir: t.TempDir(), + Logger: quietLogger(), + }) + if err == nil { + t.Fatal("Start should fail without a TNC binary") + } + if len(calls.pids) != 1 || calls.pids[0] != 0 { + t.Fatalf("setpriority pids = %v, want [0] (self renice before child spawn)", calls.pids) + } +} + +// TestStartSkipsPriorityWhenUnset: default behaviour leaves scheduling +// priority untouched. +func TestStartSkipsPriorityWhenUnset(t *testing.T) { + calls := stubSetPriority(t, nil) + cfg := &config.Config{ + TimeScale: 1, + Nodes: []config.Node{{ID: "a", Ports: []config.Port{{ + ID: "vhf", + Modem: config.Modem{Mode: config.ModeAFSK1200}, + KissPort: 18002, + }}}}, + } + _, err := Start(context.Background(), cfg, Options{ + SamoyedBin: "/nonexistent/samoyed-direwolf", + WorkDir: t.TempDir(), + Logger: quietLogger(), + }) + if err == nil { + t.Fatal("Start should fail without a TNC binary") + } + if len(calls.pids) != 0 { + t.Fatalf("setpriority called %d times, want 0 when RTPriority unset", len(calls.pids)) + } +} + +func TestApplyRTPriorityOtherErrorWarns(t *testing.T) { + stubSetPriority(t, syscall.ESRCH) + var buf bytes.Buffer + logger := slog.New(slog.NewTextHandler(&buf, nil)) + applyRTPriority(logger, "a.vhf", 99999) + if !strings.Contains(buf.String(), "setpriority failed") { + t.Errorf("expected a generic failure warning, got %q", buf.String()) + } +} diff --git a/internal/router/router.go b/internal/router/router.go index a96cac2..b7f1f00 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -31,6 +31,7 @@ import ( "fmt" "io" "log/slog" + "math" "net" "sync" "sync/atomic" @@ -55,8 +56,35 @@ const txActiveThreshold = 1500 // would never advance after the last audible block. 200 ms is long // enough to bridge the inter-frame gaps inside a multi-frame burst // without falsely closing keying, short enough to mark "key up" cleanly. +// +// At time_scale > 1 the audio (and hence the inter-frame gaps) runs +// proportionally faster, so the watchdog applies scaled(txSilenceWindow) +// — an unscaled window would bridge real inter-transmission gaps and a +// scaled audio stream would otherwise fire tx_end mid-transmission. const txSilenceWindow = 200 * time.Millisecond +// txWatchdogTick is how often the watchdog re-checks staleness at +// time_scale 1. Scaled down with time_scale so end-of-keying detection +// keeps the same sim-time resolution. +const txWatchdogTick = 50 * time.Millisecond + +// blockPeriod is the wall-clock duration of one audio block at +// time_scale 1: BlockSamples / SampleRate = 10 ms. The rxFeeder and the +// composite recorder pace themselves at scaled(blockPeriod). +const blockPeriod = time.Duration(audio.BlockSamples) * time.Second / audio.SampleRate + +// scaled divides a real-time pacing interval by the simulation's +// time_scale factor: at scale N the simulation runs N× faster than wall +// clock, so every wall-clock interval the router waits on shrinks by N. +// Scales <= 1 (including the zero value of an unvalidated config) leave +// the duration untouched. +func scaled(d time.Duration, timeScale float64) time.Duration { + if timeScale <= 1 { + return d + } + return time.Duration(float64(d) / timeScale) +} + // Options configures a Router. All paths default to "look up on $PATH". type Options struct { SamoyedBin string // path to samoyed-direwolf @@ -93,6 +121,13 @@ type Options struct { // from each TNC's stderr stream. Cheap; the parser only matches // lines containing "[0L]" so the overhead on busy logs is low. Observer *Observer + + // RTPriority, if set, renices the router's own process and every + // spawned TNC child to rtNice (-10) so the 10 ms pacing tickers and + // the children's demodulators don't glitch under shared host load. + // Best-effort: needs CAP_SYS_NICE, otherwise a one-line warning is + // logged and the simulation runs at normal priority. See priority.go. + RTPriority bool } // Router is the running simulator. @@ -153,6 +188,16 @@ type txTracker struct { // real keyup is at most a few seconds. const maxQueueBlocks = 60 * audio.SampleRate / audio.BlockSamples +// queuedBlock is one FIFO entry: the PCM block plus a start-of-transmission +// marker set by push when the block began a new keyup (see push for how the +// boundary is detected). Carrying the boundary on the block itself keeps the +// pop side timing-independent: the squelch mute is counted in blocks of the +// transmission, not in wall clock. +type queuedBlock struct { + blk audio.Block + sot bool // start of transmission +} + // linkQueue is one source→destination link's audio buffer: a non-dropping FIFO // of PCM blocks the source has transmitted, drained by the destination's // rxFeeder at the channel sample rate (so the receiver hears the carrier in @@ -167,17 +212,48 @@ const maxQueueBlocks = 60 * audio.SampleRate / audio.BlockSamples // channel. Memory is bounded in practice (a keyup is finite; the backing array // is released once drained); only a pathological runaway past maxQueueBlocks // ever drops. +// +// The linkQueue is also where the squelch_open_ms model lives — it is the +// per-directed-link seam the squelch belongs to (squelch is a property of the +// destination receiver hearing this particular carrier): pop substitutes +// silence for the first squelchBlocks blocks of every transmission. type linkQueue struct { src, dst config.PortRef loss float64 noise float64 - mu sync.Mutex - buf []audio.Block // FIFO; index 0 = oldest + // squelchBlocks is the number of leading blocks of every transmission + // delivered as silence (squelch_open_ms rounded up to whole blocks). + // Counted in blocks — i.e. in sim-time audio — so the muted span is + // the same amount of *audio* regardless of time_scale. 0 = squelch + // opens instantly (default; pop is then byte-transparent). + squelchBlocks int + + // silenceWindow is the push-side wall-clock idle gap that separates + // two transmissions. It mirrors the txWatchdog's rule exactly (the + // time-scaled txSilenceWindow): the source TNC bursts a keyup faster + // than real time but pauses between *frames inside one keyup* show up + // as short gaps on its TX stream, and the watchdog already answers + // "is this still the same transmission?" with this window. Reusing it + // here — rather than, say, a pop-side "queue drained" test, which + // would misfire whenever the rxFeeder catches up with a still-keyed + // source — keeps the two transmission-boundary definitions identical. + silenceWindow time.Duration + now func() time.Time // stubbed by tests + + mu sync.Mutex + buf []queuedBlock // FIFO; index 0 = oldest + lastPush time.Time + muteRemaining int // blocks of the current transmission still to mute } -func newLinkQueue(src, dst config.PortRef, loss, noise float64) *linkQueue { - return &linkQueue{src: src, dst: dst, loss: loss, noise: noise} +func newLinkQueue(src, dst config.PortRef, loss, noise float64, squelchBlocks int, silenceWindow time.Duration) *linkQueue { + return &linkQueue{ + src: src, dst: dst, loss: loss, noise: noise, + squelchBlocks: squelchBlocks, + silenceWindow: silenceWindow, + now: time.Now, + } } // push appends a block to the link's buffer. It never blocks and — modelling a @@ -185,31 +261,62 @@ func newLinkQueue(src, dst config.PortRef, loss, noise float64) *linkQueue { // source TNC has run ahead of real time. The only drop is the maxQueueBlocks // safety valve, which signals a stalled rxFeeder rather than normal operation, // so it is logged. +// +// The block is tagged start-of-transmission when it is the first push after a +// silenceWindow-sized idle gap (the same idle→active transition the txTracker +// uses) — that tag is what re-arms the squelch mute on the pop side. func (q *linkQueue) push(blk audio.Block, logger *slog.Logger) { q.mu.Lock() defer q.mu.Unlock() + now := q.now() + sot := q.lastPush.IsZero() || now.Sub(q.lastPush) >= q.silenceWindow + q.lastPush = now if len(q.buf) >= maxQueueBlocks { - q.buf[0] = nil + q.buf[0] = queuedBlock{} q.buf = q.buf[1:] logger.Warn("audio queue safety cap reached (rxFeeder stalled?)", "from", q.src, "to", q.dst) } - q.buf = append(q.buf, blk) + q.buf = append(q.buf, queuedBlock{blk: blk, sot: sot}) } // pop returns one block if immediately available (FIFO, oldest first). +// +// While the squelch is opening (the first squelchBlocks blocks of each +// transmission), pop delivers silence INSTEAD OF the block — not nothing: +// the carrier is on the air and the mixer must still see it for capture / +// collision decisions; it is only the receiver's audio that stays muted +// until the squelch opens. The original block is never modified (it is +// shared with every other link the source fans out to). func (q *linkQueue) pop() (audio.Block, bool) { q.mu.Lock() defer q.mu.Unlock() if len(q.buf) == 0 { return nil, false } - blk := q.buf[0] - q.buf[0] = nil // release the reference + qb := q.buf[0] + q.buf[0] = queuedBlock{} // release the reference q.buf = q.buf[1:] if len(q.buf) == 0 { q.buf = nil // release the backing array once drained } - return blk, true + if qb.sot { + q.muteRemaining = q.squelchBlocks + } + if q.muteRemaining > 0 { + q.muteRemaining-- + return audio.Silence(), true + } + return qb.blk, true +} + +// squelchBlocksFor converts a link's squelch_open_ms into a whole number of +// audio blocks, rounding up so a configured delay is never under-delivered. +func squelchBlocksFor(ms float64) int { + if ms <= 0 { + return 0 + } + const blockMS = float64(audio.BlockSamples) * 1000 / audio.SampleRate + return int(math.Ceil(ms / blockMS)) } // Start spawns all samoyed children and begins routing audio. @@ -245,11 +352,20 @@ func Start(ctx context.Context, cfg *config.Config, opts Options) (*Router, erro for _, l := range cfg.Links { fr, _ := parsePortRef(l.From) to, _ := parsePortRef(l.To) - q := newLinkQueue(fr, to, l.LossDB, l.NoiseDB) + // The transmission-boundary window scales with time_scale just + // like the txWatchdog's: at scale N the audio (and its gaps) + // arrives N× faster. + q := newLinkQueue(fr, to, l.LossDB, l.NoiseDB, + squelchBlocksFor(l.SquelchOpenMS), scaled(txSilenceWindow, cfg.TimeScale)) r.linkQueues[fr] = append(r.linkQueues[fr], q) r.rxLinks[to] = append(r.rxLinks[to], q) } + if opts.RTPriority { + // pid 0 = this process (the router and all its pacing tickers). + applyRTPriority(r.logger, "sim-router", 0) + } + udpPort := opts.StartingRxAudioPort for _, n := range cfg.Nodes { for _, p := range n.Ports { @@ -284,6 +400,9 @@ func Start(ctx context.Context, cfg *config.Config, opts Options) (*Router, erro return nil, fmt.Errorf("start %s: %w", ref, err) } r.children[ref] = child + if opts.RTPriority { + applyRTPriority(r.logger, ref.String(), child.Pid()) + } fields := []any{ "node", n.ID, "port", p.ID, "tnc", string(backend), @@ -511,8 +630,9 @@ func (r *Router) txWatchdog(ctx context.Context, ref config.PortRef) { if tt == nil { return } - tick := time.NewTicker(50 * time.Millisecond) + tick := time.NewTicker(scaled(txWatchdogTick, r.cfg.TimeScale)) defer tick.Stop() + window := scaled(txSilenceWindow, r.cfg.TimeScale) emitEnd := func() { // Also fires once during shutdown for any port that was keyed at // the moment ctx cancelled — visualiser then doesn't end with @@ -536,22 +656,21 @@ func (r *Router) txWatchdog(ctx context.Context, ref config.PortRef) { continue } last := tt.lastBusyNanos.Load() - if time.Since(time.Unix(0, last)) >= txSilenceWindow { + if time.Since(time.Unix(0, last)) >= window { emitEnd() } } } -// rxFeeder writes one audio block per BlockSamples / SampleRate seconds to -// this port's samoyed stdin. The block is the mixer's verdict on every +// rxFeeder writes one audio block per blockPeriod (divided by time_scale) +// to this port's samoyed stdin. The block is the mixer's verdict on every // active TX reaching this port through the topology. // // "Active" simply means a block is available on that link's queue. Per // PLAN Phase 3 self-mute, a port never hears its own TX — that's enforced // by config validation rejecting self-loops, so it's a no-op here. func (r *Router) rxFeeder(ctx context.Context, dst config.PortRef, stdin io.Writer) { - period := time.Duration(audio.BlockSamples) * time.Second / audio.SampleRate - ticker := time.NewTicker(period) + ticker := time.NewTicker(scaled(blockPeriod, r.cfg.TimeScale)) defer ticker.Stop() links := r.rxLinks[dst] // links *into* this destination diff --git a/internal/router/squelch_test.go b/internal/router/squelch_test.go new file mode 100644 index 0000000..125aa20 --- /dev/null +++ b/internal/router/squelch_test.go @@ -0,0 +1,182 @@ +package router + +import ( + "testing" + "time" + + "github.com/packethacking/net-sim/internal/audio" + "github.com/packethacking/net-sim/internal/config" +) + +// newSquelchQueue builds a linkQueue with a stubbed clock so transmission +// boundaries can be driven deterministically (no sleeps). The returned +// advance function moves the fake clock forward. +func newSquelchQueue(squelchBlocks int) (q *linkQueue, advance func(time.Duration)) { + q = newLinkQueue( + config.PortRef{NodeID: "a", PortID: "vhf"}, + config.PortRef{NodeID: "b", PortID: "vhf"}, + 0, 0, squelchBlocks, txSilenceWindow, + ) + now := time.Unix(1000, 0) + q.now = func() time.Time { return now } + return q, func(d time.Duration) { now = now.Add(d) } +} + +// markedBlock returns a block whose first sample carries marker (nonzero). +func markedBlock(marker int) audio.Block { + blk := make(audio.Block, audio.BlockBytes) + blk[0] = byte(marker) + blk[1] = byte(marker >> 8) + return blk +} + +// pushBurst pushes n marked blocks (markers base..base+n-1) with 1 ms +// between pushes — well inside the silence window, i.e. one transmission. +func pushBurst(t *testing.T, q *linkQueue, advance func(time.Duration), base, n int) { + t.Helper() + for i := 0; i < n; i++ { + q.push(markedBlock(base+i), quietLogger()) + advance(time.Millisecond) + } +} + +// popAll drains the queue, returning each block's first-sample marker +// (0 = silence). +func popAll(t *testing.T, q *linkQueue) []int { + t.Helper() + var out []int + for { + blk, ok := q.pop() + if !ok { + return out + } + out = append(out, int(blk[0])|int(blk[1])<<8) + } +} + +// TestSquelchMutesStartOfTransmission: squelch_open_ms=50 (5 blocks at +// 10 ms/block) mutes the first 5 blocks of a burst; the remainder comes +// through intact. +func TestSquelchMutesStartOfTransmission(t *testing.T) { + const squelchBlocks = 5 // squelchBlocksFor(50) + q, advance := newSquelchQueue(squelchBlocks) + pushBurst(t, q, advance, 100, 20) + + got := popAll(t, q) + if len(got) != 20 { + t.Fatalf("popped %d blocks, want 20 (squelch must mute, never drop)", len(got)) + } + for i := 0; i < squelchBlocks; i++ { + if got[i] != 0 { + t.Errorf("block %d = marker %d, want silence while squelch opens", i, got[i]) + } + } + for i := squelchBlocks; i < 20; i++ { + if got[i] != 100+i { + t.Errorf("block %d = marker %d, want %d (audio after squelch opens must be intact)", i, got[i], 100+i) + } + } +} + +// TestSquelchRearmsAfterIdleGap: a second transmission after a +// silence-window-sized idle gap is muted again from its first block. +func TestSquelchRearmsAfterIdleGap(t *testing.T) { + q, advance := newSquelchQueue(5) + pushBurst(t, q, advance, 100, 8) + advance(txSilenceWindow + 50*time.Millisecond) // key-up gap → new transmission + pushBurst(t, q, advance, 200, 8) + + got := popAll(t, q) + if len(got) != 16 { + t.Fatalf("popped %d blocks, want 16", len(got)) + } + // First transmission: 5 muted, 3 intact. + for i := 0; i < 5; i++ { + if got[i] != 0 { + t.Errorf("tx1 block %d = %d, want silence", i, got[i]) + } + } + for i := 5; i < 8; i++ { + if got[i] != 100+i { + t.Errorf("tx1 block %d = %d, want %d", i, got[i], 100+i) + } + } + // Second transmission: muted again from its first block. + for i := 0; i < 5; i++ { + if got[8+i] != 0 { + t.Errorf("tx2 block %d = %d, want silence (squelch must re-arm per transmission)", i, got[8+i]) + } + } + for i := 5; i < 8; i++ { + if got[8+i] != 200+i { + t.Errorf("tx2 block %d = %d, want %d", i, got[8+i], 200+i) + } + } +} + +// TestSquelchNotRearmedWithinBurst: a short push gap (an inter-frame pause +// inside one keyup, below the silence window) must not re-mute — same rule +// the txWatchdog uses to keep a multi-frame burst as one keying. +func TestSquelchNotRearmedWithinBurst(t *testing.T) { + q, advance := newSquelchQueue(3) + pushBurst(t, q, advance, 100, 5) + advance(txSilenceWindow / 2) // inter-frame gap, still the same keyup + pushBurst(t, q, advance, 200, 5) + + got := popAll(t, q) + if len(got) != 10 { + t.Fatalf("popped %d blocks, want 10", len(got)) + } + for i := 3; i < 5; i++ { + if got[i] != 100+i { + t.Errorf("block %d = %d, want %d", i, got[i], 100+i) + } + } + for i := 0; i < 5; i++ { + if got[5+i] != 200+i { + t.Errorf("second-frame block %d = %d, want %d (must NOT be re-muted mid-keyup)", i, got[5+i], 200+i) + } + } +} + +// TestSquelchZeroIsTransparent: the default (squelch_open_ms: 0) delivers +// every block byte-identically — today's behaviour exactly. +func TestSquelchZeroIsTransparent(t *testing.T) { + q, advance := newSquelchQueue(0) + pushBurst(t, q, advance, 1, 6) + advance(txSilenceWindow * 2) + pushBurst(t, q, advance, 100, 6) + + got := popAll(t, q) + want := []int{1, 2, 3, 4, 5, 6, 100, 101, 102, 103, 104, 105} + if len(got) != len(want) { + t.Fatalf("popped %d blocks, want %d", len(got), len(want)) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("block %d = %d, want %d (squelch 0 must be transparent)", i, got[i], want[i]) + } + } +} + +// TestSquelchBlocksFor pins the ms→blocks conversion (10 ms blocks, +// rounded up). +func TestSquelchBlocksFor(t *testing.T) { + cases := []struct { + ms float64 + want int + }{ + {0, 0}, + {-5, 0}, + {1, 1}, + {10, 1}, + {50, 5}, + {55, 6}, + {500, 50}, + } + for _, c := range cases { + if got := squelchBlocksFor(c.ms); got != c.want { + t.Errorf("squelchBlocksFor(%g) = %d, want %d", c.ms, got, c.want) + } + } +} diff --git a/internal/router/timescale_test.go b/internal/router/timescale_test.go new file mode 100644 index 0000000..8e66d0a --- /dev/null +++ b/internal/router/timescale_test.go @@ -0,0 +1,119 @@ +package router + +import ( + "bytes" + "context" + "sync" + "testing" + "time" + + "github.com/packethacking/net-sim/internal/audio" + "github.com/packethacking/net-sim/internal/config" +) + +// TestScaledPeriods pins the arithmetic every paced loop relies on: the +// rxFeeder/composite block ticker, the txWatchdog tick, and the +// txSilenceWindow all divide by time_scale, and scale <= 1 (including +// the zero value of a hand-built config) is a no-op. +func TestScaledPeriods(t *testing.T) { + cases := []struct { + d time.Duration + scale float64 + want time.Duration + }{ + {blockPeriod, 1, 10 * time.Millisecond}, + {blockPeriod, 10, time.Millisecond}, + {blockPeriod, 0, 10 * time.Millisecond}, // unvalidated zero config = real time + {txSilenceWindow, 1, 200 * time.Millisecond}, + {txSilenceWindow, 4, 50 * time.Millisecond}, + {txWatchdogTick, 5, 10 * time.Millisecond}, + {txWatchdogTick, 0.5, 50 * time.Millisecond}, // sub-1 scales are clamped to real time + } + for _, c := range cases { + if got := scaled(c.d, c.scale); got != c.want { + t.Errorf("scaled(%v, %g) = %v, want %v", c.d, c.scale, got, c.want) + } + } +} + +// collectWriter accumulates rxFeeder output and closes done once target +// bytes have arrived, so the test can stop the feeder without sleeping. +type collectWriter struct { + mu sync.Mutex + buf []byte + target int + done chan struct{} + once sync.Once +} + +func (w *collectWriter) Write(p []byte) (int, error) { + w.mu.Lock() + w.buf = append(w.buf, p...) + n := len(w.buf) + w.mu.Unlock() + if n >= w.target { + w.once.Do(func() { close(w.done) }) + } + return len(p), nil +} + +// runRxFeederSequence pre-loads one link queue with nBlocks marked blocks, +// runs the real rxFeeder against it at the given time_scale, and returns +// the first nBlocks of delivered audio. +func runRxFeederSequence(t *testing.T, timeScale float64, nBlocks int) []byte { + t.Helper() + src := config.PortRef{NodeID: "a", PortID: "vhf"} + dst := config.PortRef{NodeID: "b", PortID: "vhf"} + q := newLinkQueue(src, dst, 0, 0, 0, txSilenceWindow) + + r := &Router{ + cfg: &config.Config{TimeScale: timeScale}, + mixer: audio.NewMixer(6, false, "silence"), + logger: quietLogger(), + rxLinks: map[config.PortRef][]*linkQueue{dst: {q}}, + } + + for i := 0; i < nBlocks; i++ { + blk := make(audio.Block, audio.BlockBytes) + // Unique nonzero marker in the first sample (LE int16). + blk[0] = byte(i + 1) + blk[1] = byte((i + 1) >> 8) + q.push(blk, r.logger) + } + + w := &collectWriter{target: nBlocks * audio.BlockBytes, done: make(chan struct{})} + ctx, cancel := context.WithCancel(context.Background()) + feederDone := make(chan struct{}) + go func() { + r.rxFeeder(ctx, dst, w) + close(feederDone) + }() + select { + case <-w.done: + case <-time.After(10 * time.Second): + t.Fatal("rxFeeder never delivered the expected blocks") + } + cancel() + <-feederDone + return w.buf[:nBlocks*audio.BlockBytes] +} + +// TestRxFeederSequenceIdenticalAcrossTimeScales: time_scale changes only +// the pacing of delivery, never its content — the block SEQUENCE a +// receiver hears must be byte-identical at scale 1 and at a large scale. +func TestRxFeederSequenceIdenticalAcrossTimeScales(t *testing.T) { + const nBlocks = 8 + realTime := runRxFeederSequence(t, 1, nBlocks) + accelerated := runRxFeederSequence(t, 20, nBlocks) + if !bytes.Equal(realTime, accelerated) { + t.Fatal("block sequence differs between time_scale 1 and 20 — scaling must affect timing only") + } + // And the markers really came through in FIFO order. + for i := 0; i < nBlocks; i++ { + off := i * audio.BlockBytes + got := int(realTime[off]) | int(realTime[off+1])<<8 + if got != i+1 { + t.Fatalf("block %d carries marker %d, want %d", i, got, i+1) + } + } +} diff --git a/internal/tnc/tnc.go b/internal/tnc/tnc.go index 3da59b1..9ef1ed7 100644 --- a/internal/tnc/tnc.go +++ b/internal/tnc/tnc.go @@ -90,6 +90,15 @@ func (c *Child) TXAudio() io.Reader { return c.txReader } // Spec returns the launch spec. func (c *Child) Spec() Spec { return c.spec } +// Pid returns the child's OS process id, or 0 if it never started. Used +// by the router's -rt-priority plumbing to renice spawned TNCs. +func (c *Child) Pid() int { + if c.cmd != nil && c.cmd.Process != nil { + return c.cmd.Process.Pid + } + return 0 +} + // Wait blocks until the child exits. func (c *Child) Wait() error { return <-c.exitC }