diff --git a/cmd/skywire-cli/commands/proxy/mux_ops.go b/cmd/skywire-cli/commands/proxy/mux_ops.go index 4ddba70d9a..5495e782d7 100644 --- a/cmd/skywire-cli/commands/proxy/mux_ops.go +++ b/cmd/skywire-cli/commands/proxy/mux_ops.go @@ -257,7 +257,7 @@ Example: } var muxModeCmd = &cobra.Command{ - Use: "mode ", + Use: "mode ", Short: "Change mux scheduler weighting at runtime", Long: `Set the mux transport-selection mode for the visor. @@ -272,12 +272,22 @@ var muxModeCmd = &cobra.Command{ and a slow one carries little — the thin-spread aggregation mode. A just-promoted leg starts at a small cold-leg floor share and ramps as its goodput proves out. + ecf - Earliest Completion First: predictive hold-back. Sends on + the fastest leg while it has send capacity and only spills + onto a slower leg when that leg would deliver its frame + sooner than the fast leg can drain its own backlog — + otherwise it holds the frame on the fast leg. Unlike + capacity (which still sprays a share onto slow legs and + head-of-line-stalls the reorder buffer on them), ECF + aggregates across heterogeneous legs without paying the + slow-leg HoL cost. Affects every active and future mux'd route group on this visor IMMEDIATELY (the router re-applies the mode to live route groups). The setting persists to skywire-config.json so it survives restart. Example: + skywire cli proxy mux mode ecf # predictive earliest-completion-first skywire cli proxy mux mode capacity # goodput-weighted thin spread skywire cli proxy mux info --watch 1s skywire cli proxy mux mode auto # back to latency-weighted`, @@ -286,9 +296,9 @@ Example: Run: func(cmd *cobra.Command, args []string) { mode := args[0] switch mode { - case "auto", "equal", "capacity": + case "auto", "equal", "capacity", "ecf": default: - internal.PrintFatalError(cmd.Flags(), fmt.Errorf("mode must be 'auto', 'equal', or 'capacity', got %q", mode)) + internal.PrintFatalError(cmd.Flags(), fmt.Errorf("mode must be 'auto', 'equal', 'capacity', or 'ecf', got %q", mode)) } rpcClient, err := clirpc.Client(cmd.Flags()) if err != nil { diff --git a/pkg/router/dial_hook.go b/pkg/router/dial_hook.go index 21481f68d1..60e1fe6069 100644 --- a/pkg/router/dial_hook.go +++ b/pkg/router/dial_hook.go @@ -254,6 +254,17 @@ const ( // for aggregating bandwidth across a disjoint multi-leg route. // This is the adaptive default's bulk-spread mode. DistributionCapacity + // DistributionECF is the predictive Earliest-Completion-First + // hold-back scheduler (router.WeightModeECF). Unlike + // DistributionCapacity, which sprays a fraction of frames onto + // slower legs in proportion to their goodput — and so keeps + // head-of-line-stalling the in-order reorder buffer on those slow + // legs — ECF sends on the fastest leg while it has send capacity + // and only spills onto a slower leg when the slow leg would deliver + // its frame sooner than the fast leg can drain its own backlog. It + // is the mode that actually aggregates across heterogeneous legs + // without paying the slow-leg HoL cost. Per-packet, O(legs). + DistributionECF ) // LegChangeHook is an optional hook fired by the route group diff --git a/pkg/router/policy/distribution.go b/pkg/router/policy/distribution.go index b774357ee6..c114e3355e 100644 --- a/pkg/router/policy/distribution.go +++ b/pkg/router/policy/distribution.go @@ -48,6 +48,8 @@ func ParseDistribution(s string) (router.DistributionConfig, error) { return router.DistributionConfig{Mode: router.DistributionLatencyAdaptive}, nil case "capacity", "capacity-weighted": return router.DistributionConfig{Mode: router.DistributionCapacity}, nil + case "ecf": + return router.DistributionConfig{Mode: router.DistributionECF}, nil } // Prefix-based. diff --git a/pkg/router/policy/presethook/presethook.go b/pkg/router/policy/presethook/presethook.go index a92c26de1c..d336bfd560 100644 --- a/pkg/router/policy/presethook/presethook.go +++ b/pkg/router/policy/presethook/presethook.go @@ -243,6 +243,8 @@ func distributionFor(desc string) router.DistributionConfig { return router.DistributionConfig{Mode: router.DistributionAuto} case "capacity": return router.DistributionConfig{Mode: router.DistributionCapacity} + case "ecf": + return router.DistributionConfig{Mode: router.DistributionECF} default: return router.DistributionConfig{Mode: router.DistributionUnset} } diff --git a/pkg/router/route_group.go b/pkg/router/route_group.go index 15a5d0c84c..14171c87d1 100644 --- a/pkg/router/route_group.go +++ b/pkg/router/route_group.go @@ -1129,6 +1129,8 @@ func (rg *RouteGroup) applyDistribution(cfg DistributionConfig) { wm = WeightModeLatencyAdaptive case DistributionCapacity: wm = WeightModeCapacity + case DistributionECF: + wm = WeightModeECF case DistributionDSCPPriority: wm = WeightModeDSCPPriority rg.mux.tpSelector.SetDSCPThreshold(cfg.DSCPThreshold) diff --git a/pkg/router/route_mux.go b/pkg/router/route_mux.go index cdca36edf8..a57fdb1b40 100644 --- a/pkg/router/route_mux.go +++ b/pkg/router/route_mux.go @@ -94,6 +94,16 @@ type legCounters struct { lastRateNano int64 goodputUpBps float64 goodputDownBps float64 + // ECF (WeightModeECF) per-leg state, maintained by rebuildWeights' ECF + // branch (under legMu, off the data path). ecfLastSentBytes snapshots the + // sent counter at the previous ECF refresh so the delta over the refresh + // window is this leg's send rate (kept separate from lastTotalBytes / + // lastRateSentBytes so the ECF sampler never disturbs the capacity or + // telemetry samplers). ecfRttMs / ecfJitterMs are the EWMA'd mean RTT and + // jitter (sigma) the ECF predicate consumes. + ecfLastSentBytes uint64 + ecfRttMs float64 + ecfJitterMs float64 } // routeMux encapsulates route multiplexing state and logic. @@ -160,6 +170,11 @@ type routeMux struct { // Guarded by legMu. See docs/warm_standby_legs_rfc.md. standby []bool + // ecfLastRebuildNano is the wall-clock (UnixNano) of the previous ECF-state + // refresh, used to turn each leg's sent-byte delta into a bytes/sec rate. + // Touched only under legMu in rebuildWeights' ECF branch. + ecfLastRebuildNano int64 + // standbyNewLegs makes every NEWLY-grown aux leg (index > 0) enter the // warm-standby pool instead of going straight into the active send set. // The primary leg (index 0) is never affected. Set only when a promoting @@ -246,7 +261,8 @@ func (m *routeMux) selectTransport(tps []*transport.ManagedTransport, fwd []rout case WeightModeSizeThreshold, WeightModeSticky5Tuple, WeightModeLatencyAdaptive, - WeightModeDSCPPriority: + WeightModeDSCPPriority, + WeightModeECF: idx := m.tpSelector.SelectForPayload(payload) if idx < len(tps) { tp := tps[idx] @@ -855,5 +871,67 @@ func (m *routeMux) rebuildWeights(tps []*transport.ManagedTransport) { m.legMu.Unlock() m.tpSelector.SetCapacityWeights(weights) } + // ECF mode: build the per-leg {rate, RTT, jitter, ready, BDP} snapshot the + // predictive scheduler reasons over. Rate is the sent-byte delta over the + // refresh window (computed here, not from snapshotLegs, so it works even + // when nothing is observing the telemetry page). RTT is the leg's first-hop + // transport latency (tp.GetLatency(), ms) — the end-to-end route latency + // would be more accurate but is not reachable from the mux; noted as a + // follow-up. Jitter is an EWMA of |RTT-mean|, the ECF sigma margin. + if m.tpSelector.Mode() == WeightModeECF { + m.legMu.Lock() + now := time.Now().UnixNano() + var elapsed float64 + if m.ecfLastRebuildNano != 0 { + elapsed = float64(now-m.ecfLastRebuildNano) / float64(time.Second) + } + states := make([]ecfLegState, len(m.legs)) + for i, lc := range m.legs { + if lc == nil { + continue + } + // Send rate over the refresh window (bytes/sec). + sent := atomic.LoadUint64(&lc.sentBytes) + var rate float64 + if elapsed > 0 { + rate = float64(byteDelta(sent, lc.ecfLastSentBytes)) / elapsed + } + lc.ecfLastSentBytes = sent + // RTT EWMA + jitter (sigma) EWMA. + var rttMs float64 + if i < len(tps) && tps[i] != nil { + rttMs = tps[i].GetLatency() + } + if rttMs > 0 { + if lc.ecfRttMs == 0 { + lc.ecfRttMs = rttMs + } else { + dev := rttMs - lc.ecfRttMs + if dev < 0 { + dev = -dev + } + lc.ecfJitterMs = ecfJitterAlpha*dev + (1-ecfJitterAlpha)*lc.ecfJitterMs + lc.ecfRttMs = ecfRttAlpha*rttMs + (1-ecfRttAlpha)*lc.ecfRttMs + } + } + ready := true + if i < len(m.standby) && m.standby[i] { + ready = false + } + if i < len(m.ready) && !m.ready[i] { + ready = false + } + states[i] = ecfLegState{ + rttMs: lc.ecfRttMs, + jitterMs: lc.ecfJitterMs, + rateBps: rate, + cwndBytes: rate * lc.ecfRttMs / 1000.0, + ready: ready, + } + } + m.ecfLastRebuildNano = now + m.legMu.Unlock() + m.tpSelector.SetECFState(states) + } m.tpSelector.Rebuild(tps) } diff --git a/pkg/router/transport_selector.go b/pkg/router/transport_selector.go index bb92c26ccd..46325a0a76 100644 --- a/pkg/router/transport_selector.go +++ b/pkg/router/transport_selector.go @@ -4,6 +4,7 @@ package router import ( "sync" "sync/atomic" + "time" "github.com/skycoin/skywire/pkg/transport" ) @@ -58,8 +59,72 @@ const ( // capacity fills each leg toward its bandwidth — the mode that // actually aggregates throughput across a disjoint mux. WeightModeCapacity + // WeightModeECF is a PREDICTIVE hold-back scheduler adapted from ECF + // (Earliest Completion First, Lim et al., CoNEXT 2017). Capacity/auto + // weighting keeps assigning a share of in-order frames to slower legs; in + // a reorder buffer that must deliver strictly in order, every frame on a + // slow leg is a head-of-line stall, so goodput/latency-proportional + // spraying does NOT aggregate under path heterogeneity (~25% of ideal in + // the MPTCP/MP-QUIC literature). ECF instead sends on the fastest leg + // while it has send capacity, and spills onto a slower leg ONLY when the + // slow leg would deliver its frame sooner than the fast leg can drain its + // own backlog — otherwise it holds the frame on the fast leg. skywire has + // no TCP cwnd, so the cwnd/"has capacity" test is adapted to a per-leg + // bandwidth-delay-product (rate*RTT) and a selector-maintained in-flight + // byte estimate; see ecfPick and SelectECF. This is the aggregation mode + // that stops a heterogeneous mux from HoL-stalling on its slow legs. + WeightModeECF ) +// ECF (WeightModeECF) tuning constants. Starting values — expect a live-tuning +// pass. See ecfPick for how each is used. +const ( + // ecfBeta is the ECF hysteresis constant. Once the scheduler decides to + // hold a frame on the fast leg (waiting latched), it inflates the slow + // leg's delivery estimate by (1+ecfBeta) on the next pick, so a marginal + // leg does not flap in and out of the spill set packet-to-packet. + ecfBeta = 0.25 + // ecfDefaultFrameBytes is the in-flight increment charged to a leg for a + // frame whose size the caller did not supply (control/handshake frames + // never reach the ECF pick, so this is only a defensive fallback). + ecfDefaultFrameBytes = 1024 + // ecfRttAlpha / ecfJitterAlpha weight the newest sample in the per-leg + // mean-RTT and jitter (sigma) EWMAs the mux maintains for ECF (see + // route_mux.go rebuildWeights). Jitter is the ECF sigma margin. + ecfRttAlpha = 0.3 + ecfJitterAlpha = 0.3 +) + +// ecfLegState is the per-leg snapshot the ECF scheduler reasons over — one +// entry per leg index (parallel to the route group's tps[]). The rate/RTT/ +// jitter/ready fields are refreshed by the mux via SetECFState on the +// rebuildWeights cadence; inflightBytes is maintained by the selector itself +// between refreshes (incremented per selected frame, drained by rate over +// wall-clock), so it survives a refresh (SetECFState carries it forward). +type ecfLegState struct { + // rttMs is the leg's mean round-trip latency estimate in ms (EWMA of + // tp.GetLatency()); 0 = unknown (deprioritized in the fast-leg pick). + rttMs float64 + // jitterMs is the ECF sigma: an EWMA of |sample-mean| RTT deviation, used + // as the inter-leg jitter margin `d` in the hold-back predicate. + jitterMs float64 + // rateBps is the leg's recent send goodput in bytes/sec (from the mux's + // per-leg sent-byte delta over the refresh window); 0 = unknown. + rateBps float64 + // cwndBytes is the ECF cwnd substitute: the leg's bandwidth-delay product + // (rateBps * rttMs/1000) — the bytes it can hold in flight in one RTT. + // 0 when rate or RTT is unknown, which ecfSaturated treats as "unlimited" + // so a cold leg is used (and thus measured) rather than assumed full. + cwndBytes float64 + // ready is true when the leg may be selected for sending (alive, its rule + // confirmed, not a warm standby). selectTransport re-validates readiness + // after the pick, so this is an optimization, not the safety gate. + ready bool + // inflightBytes is the selector's estimate of bytes sent-but-not-yet- + // delivered on this leg. NOT set by SetECFState — carried across refreshes. + inflightBytes float64 +} + // transportSelector implements weighted transport selection based on latency. // Faster transports (lower latency) get proportionally more packets. // Falls back to equal-weight round-robin when latency data is unavailable. @@ -95,6 +160,21 @@ type transportSelector struct { // WeightModeCapacity. Set by the mux via SetCapacityWeights // before each Rebuild. Empty / all-zero → equal bootstrap. capacityWeights []float64 + // ecfLegs holds the per-leg ECF state used by WeightModeECF, one entry + // per leg index. Refreshed by the mux via SetECFState; the inflightBytes + // field is maintained by SelectECF between refreshes. All ECF state is + // guarded by ts.mu — SelectECF takes the write lock because it mutates + // inflightBytes (per-packet picks are already serialized per route group + // by the caller's rg.mu, so the lock is uncontended except vs the ~5s + // SetECFState refresh). + ecfLegs []ecfLegState + // ecfWaiting latches ECF's hold-back hysteresis (the `waiting` state in + // the paper): true after a pick chose to hold on the fast leg, so the + // next pick inflates the slow-leg delivery estimate by (1+ecfBeta). + ecfWaiting bool + // ecfLastNano is the wall-clock (UnixNano) of the previous SelectECF call, + // used to drain each leg's inflightBytes by its rate over the elapsed gap. + ecfLastNano int64 } func newTransportSelector() *transportSelector { @@ -176,10 +256,30 @@ func (m WeightMode) String() string { return "dscp-priority" case WeightModeCapacity: return "capacity" + case WeightModeECF: + return "ecf" } return "unknown" } +// SetECFState stores the per-leg ECF state used by WeightModeECF. Caller (the +// mux) recomputes rate/RTT/jitter/ready each refresh and calls Rebuild +// afterwards. The selector-maintained inflightBytes estimate is carried +// forward per leg index across refreshes so a refresh does not reset the +// in-flight accounting. A copy is kept so the caller may reuse its slice. +func (ts *transportSelector) SetECFState(states []ecfLegState) { + ts.mu.Lock() + ns := make([]ecfLegState, len(states)) + copy(ns, states) + for i := range ns { + if i < len(ts.ecfLegs) { + ns[i].inflightBytes = ts.ecfLegs[i].inflightBytes + } + } + ts.ecfLegs = ns + ts.mu.Unlock() +} + // Rebuild recomputes the selection schedule from the current transport latencies. // Called periodically (e.g., every keep-alive cycle) and when transports change. // tps must not be modified concurrently (caller holds RouteGroup.mu or equivalent). @@ -372,7 +472,10 @@ func (ts *transportSelector) Rebuild(tps []*transport.ManagedTransport) { // GetLatency. The primary schedule is also populated so a // caller using Select() (no payload) still gets a reasonable // leg. - if ts.mode == WeightModeSticky5Tuple || ts.mode == WeightModeLatencyAdaptive { + // ECF also builds the live-leg arrays: SelectECF reasons over ecfLegs + // (set separately via SetECFState) but the mirrored schedule is the + // Select() fallback for control/handshake frames that carry no payload. + if ts.mode == WeightModeSticky5Tuple || ts.mode == WeightModeLatencyAdaptive || ts.mode == WeightModeECF { live := make([]int, 0, n) liveTps := make([]*transport.ManagedTransport, 0, n) for i, tp := range tps { @@ -549,6 +652,8 @@ func (ts *transportSelector) SelectForPayload(p []byte) int { return live[h%uint32(len(live))] //nolint:gosec case WeightModeLatencyAdaptive: return pickLowestLatency(live, liveTps) + case WeightModeECF: + return ts.SelectECF(len(p)) case WeightModeDSCPPriority: if isIPv4DSCPGE(p, dscpThreshold) { return 0 @@ -613,6 +718,181 @@ func isIPv4DSCPGE(p []byte, threshold int) bool { return dscp >= threshold } +// SelectECF returns the leg index for the next DATA frame of the given size +// under the ECF predictive hold-back scheduler (WeightModeECF). It drains each +// leg's in-flight estimate for the time elapsed since the previous pick, runs +// the ecfPick decision, and charges the chosen leg with this frame's size. +// Takes the write lock because it mutates per-leg inflight state; per-packet +// picks for a given route group are already serialized by the caller's rg.mu, +// so the only contention is against the periodic SetECFState refresh. +// +// Falls back to the schedule-based pick (mirrored live legs) when there is no +// ECF state yet (bootstrap) or no leg is ready. +func (ts *transportSelector) SelectECF(size int) int { + ts.mu.Lock() + defer ts.mu.Unlock() + + if len(ts.ecfLegs) == 0 { + return ts.scheduleIndexLocked() + } + + // Drain inflight by rate*elapsed since the last pick. Because the leg + // transports are reliable, a frame put on leg i is delivered ~rttᵢ later; + // draining at the leg's send rate approximates that clearance without a + // per-frame ack signal (skywire has no per-leg ack attribution). + now := time.Now().UnixNano() + if ts.ecfLastNano != 0 { + dt := float64(now-ts.ecfLastNano) / float64(time.Second) + if dt > 0 { + for i := range ts.ecfLegs { + ts.ecfLegs[i].inflightBytes -= ts.ecfLegs[i].rateBps * dt + if ts.ecfLegs[i].inflightBytes < 0 { + ts.ecfLegs[i].inflightBytes = 0 + } + } + } + } + ts.ecfLastNano = now + + idx := ecfPick(ts.ecfLegs, ts.ecfWaiting, &ts.ecfWaiting) + if idx < 0 { + return ts.scheduleIndexLocked() + } + if size <= 0 { + size = ecfDefaultFrameBytes + } + ts.ecfLegs[idx].inflightBytes += float64(size) + return idx +} + +// scheduleIndexLocked returns the next schedule-based leg index. Caller holds +// ts.mu (the atomic counter is still used so it stays consistent with Select()). +func (ts *transportSelector) scheduleIndexLocked() int { + if len(ts.schedule) == 0 { + return 0 + } + idx := atomic.AddUint32(&ts.counter, 1) - 1 + return ts.schedule[idx%uint32(len(ts.schedule))] //nolint:gosec +} + +// ecfPick is the pure ECF (Earliest Completion First) decision, in FILTER form. +// Given per-leg state and the latched `waiting` hysteresis flag, it returns the +// leg index to send the next frame on, or -1 when no leg is ready (caller falls +// back to the schedule). waitOut, if non-nil, receives the updated `waiting` +// latch. +// +// Adaptation of ECF to skywire's no-cwnd model (every substitution called out): +// - "xf has send capacity now" → inflightBytes < cwndBytes (ecfSaturated). +// - CWND_f (bytes/RTT) → cwndBytes, the leg's bandwidth-delay +// product rate*RTT. Unknown capacity (cold leg) is treated as unlimited so +// the leg is used and thus measured, never assumed full. +// - k (backlog not yet drained) → inflightBytes on the fast leg xf. +// - sigma_f / sigma_s (RTT σ) → per-leg jitterMs (EWMA of |sample-mean|). +// +// The governing rule is the paper's: if the fast leg can drain its whole +// backlog before the slow leg delivers even one frame, do not use the slow leg. +// This is the FILTER variant — where full ECF would return NO-LEG and idle +// briefly waiting for xf, this instead returns xf (send on the fast leg now). +// The frame then queues on xf's reliable transport rather than being held by +// the scheduler, which keeps the send path non-blocking. The full wait/idle +// variant is a follow-up (it needs the send path to handle a "hold this frame" +// return without dropping or busy-spinning). +func ecfPick(legs []ecfLegState, waiting bool, waitOut *bool) int { + setWait := func(v bool) { + if waitOut != nil { + *waitOut = v + } + } + + // xf = ready leg with the smallest (positive) RTT. Ties and unknown-RTT + // legs fall back to lowest index (leg 0 is the primary/fastest leg). + xf := -1 + for i := range legs { + if !legs[i].ready { + continue + } + if xf < 0 || ecfBetterRTT(legs[i], legs[xf]) { + xf = i + } + } + if xf < 0 { + return -1 + } + + // Fast leg still has send capacity (or unknown capacity → cold start): + // send on it. This is ECF's first branch and the common case — as long as + // the fastest leg is not saturated, nothing spills to a slower leg. + if !ecfSaturated(legs[xf]) { + setWait(false) + return xf + } + + // Fast leg saturated: find the next-best ready leg that still has capacity + // (xs). If none can take more, stay on xf (its transport queues the frame). + xs := -1 + for i := range legs { + if i == xf || !legs[i].ready || ecfSaturated(legs[i]) { + continue + } + if xs < 0 || ecfBetterRTT(legs[i], legs[xs]) { + xs = i + } + } + if xs < 0 { + setWait(false) + return xf + } + + // ECF hold-back predicate. n = how many xf-RTTs to drain the fast leg's + // backlog; if xf clears that backlog before xs delivers even one frame + // (its RTT plus the jitter margin d), hold on xf instead of spilling. + rttF, rttS := legs[xf].rttMs, legs[xs].rttMs + n := 1.0 + if legs[xf].cwndBytes > 0 { + n = 1 + legs[xf].inflightBytes/legs[xf].cwndBytes + } + d := legs[xf].jitterMs + if legs[xs].jitterMs > d { + d = legs[xs].jitterMs + } + hyst := 1.0 + if waiting { + hyst = 1 + ecfBeta + } + if n*rttF < hyst*(rttS+d) { + // Fast leg wins the race: hold the frame on xf (filter variant — the + // paper would idle here; latch waiting for hysteresis on the next pick). + setWait(true) + return xf + } + setWait(false) + return xs +} + +// ecfSaturated reports whether a leg is carrying a full bandwidth-delay product +// of un-delivered bytes (no send capacity right now). A leg whose capacity is +// unknown (no rate/RTT sample yet) is never saturated, so a cold leg is used +// and measured instead of being assumed full. +func ecfSaturated(l ecfLegState) bool { + if l.cwndBytes <= 0 { + return false + } + return l.inflightBytes >= l.cwndBytes +} + +// ecfBetterRTT reports whether leg a is a better (lower-RTT) fast-leg candidate +// than leg b. A leg with unknown RTT (0) is the worst candidate — it only wins +// when the incumbent is also unknown, which the caller's index order breaks. +func ecfBetterRTT(a, b ecfLegState) bool { + if a.rttMs <= 0 { + return false + } + if b.rttMs <= 0 { + return true + } + return a.rttMs < b.rttMs +} + // pickLowestLatency returns the live-leg index with the smallest // GetLatency() value. Legs reporting 0 (unknown) are deprioritized // — they only win when no leg has a measurement. Linear scan; diff --git a/pkg/router/transport_selector_ecf_test.go b/pkg/router/transport_selector_ecf_test.go new file mode 100644 index 0000000000..fd1cbbe46b --- /dev/null +++ b/pkg/router/transport_selector_ecf_test.go @@ -0,0 +1,222 @@ +package router + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// leg is a small ecfLegState constructor for the table tests. +func ecfLeg(rttMs, jitterMs, cwndBytes, inflightBytes float64, ready bool) ecfLegState { + return ecfLegState{ + rttMs: rttMs, + jitterMs: jitterMs, + cwndBytes: cwndBytes, + inflightBytes: inflightBytes, + ready: ready, + } +} + +// TestECFPick is the deterministic table-driven test of the ECF hold-back +// predicate (the pure ecfPick decision). It feeds synthetic legs + a backlog +// (encoded as the fast leg's inflightBytes) and asserts which leg is chosen. +func TestECFPick(t *testing.T) { + tests := []struct { + name string + legs []ecfLegState + waiting bool + want int + wantW bool // expected waiting latch afterwards + }{ + { + name: "no ready leg returns -1", + legs: []ecfLegState{ + ecfLeg(10, 0, 1000, 0, false), + ecfLeg(100, 0, 1000, 0, false), + }, + want: -1, + }, + { + name: "single ready leg is chosen", + legs: []ecfLegState{ + ecfLeg(10, 0, 1000, 5000, true), // saturated but only option + }, + want: 0, + wantW: false, + }, + { + name: "fast leg not saturated: use it, ignore slow leg", + legs: []ecfLegState{ + ecfLeg(10, 0, 10000, 0, true), // room to spare + ecfLeg(100, 0, 10000, 0, true), + }, + want: 0, + wantW: false, + }, + { + name: "unknown-capacity fast leg (cold start) is used", + legs: []ecfLegState{ + ecfLeg(10, 0, 0, 0, true), // cwnd unknown -> never saturated + ecfLeg(100, 0, 10000, 0, true), + }, + want: 0, + wantW: false, + }, + { + // CANONICAL CASE: one fast leg + one 10x-RTT slow leg under + // backlog. The fast leg is saturated (backlog present) but can + // drain it far sooner than the slow leg delivers even one frame, + // so ECF must NOT schedule onto the slow leg. + name: "fast + 10x-RTT slow under moderate backlog: hold on fast, not slow", + legs: []ecfLegState{ + ecfLeg(10, 0, 10000, 20000, true), // saturated, k=20000, n=3, n*rttF=30 + ecfLeg(100, 0, 1e9, 0, true), // slow leg, rttS=100 + }, + want: 0, // 30 < 100 -> hold on fast + wantW: true, + }, + { + // Same topology but a HUGE backlog: now the fast leg would take + // longer to drain than the slow leg needs to deliver one frame, + // so spilling to the slow leg is the earliest-completion choice. + name: "fast + 10x-RTT slow under huge backlog: spill to slow", + legs: []ecfLegState{ + ecfLeg(10, 0, 10000, 200000, true), // k=200000, n=21, n*rttF=210 + ecfLeg(100, 0, 1e9, 0, true), // rttS=100 + }, + want: 1, // 210 >= 100 -> spill + wantW: false, + }, + { + // Jitter margin d widens the slow leg's effective delivery time, + // making the hold decision stickier. With d=60, rttS+d=160 and + // n*rttF=150 (k=140000 -> n=15) stays under it -> hold on fast. + name: "jitter margin keeps a borderline frame on the fast leg", + legs: []ecfLegState{ + ecfLeg(10, 0, 10000, 140000, true), // n=15, n*rttF=150 + ecfLeg(100, 60, 1e9, 0, true), // rttS=100, jitter 60 -> d=60 + }, + want: 0, // 150 < 160 -> hold + wantW: true, + }, + { + // Two ready spill candidates: the saturated fast leg spills to the + // lower-RTT of the two remaining legs (xs = smallest RTT with + // capacity), here leg 1 (rtt 50) over leg 2 (rtt 80). + name: "spill picks the lowest-RTT candidate with capacity", + legs: []ecfLegState{ + ecfLeg(10, 0, 1000, 500000, true), // hopelessly backlogged -> spill + ecfLeg(50, 0, 1e9, 0, true), + ecfLeg(80, 0, 1e9, 0, true), + }, + want: 1, + wantW: false, + }, + { + // A saturated fast leg with NO spill target that has capacity + // stays on the fast leg (filter variant never returns NO-LEG). + name: "all legs saturated: stay on fast leg", + legs: []ecfLegState{ + ecfLeg(10, 0, 1000, 500000, true), + ecfLeg(100, 0, 1000, 500000, true), + }, + want: 0, + wantW: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + w := tt.waiting + got := ecfPick(tt.legs, tt.waiting, &w) + assert.Equal(t, tt.want, got, "chosen leg") + if tt.want >= 0 { + assert.Equal(t, tt.wantW, w, "waiting latch") + } + }) + } +} + +// TestECFPick_Hysteresis verifies the beta hysteresis: once waiting is latched, +// the slow leg's delivery estimate is inflated by (1+ecfBeta), so a frame that +// would spill when not waiting is instead held when waiting. +func TestECFPick_Hysteresis(t *testing.T) { + // Tune so the predicate is just BELOW the spill boundary once hysteresis + // applies but just ABOVE it without. n*rttF = 105 (k=95000 -> n=10.5). + // rttS+d = 100. Without hysteresis: 105 >= 100 -> spill (leg 1). + // With hysteresis: 105 < 1.25*100=125 -> hold (leg 0). + legs := []ecfLegState{ + ecfLeg(10, 0, 10000, 95000, true), + ecfLeg(100, 0, 1e9, 0, true), + } + + got := ecfPick(legs, false, nil) + assert.Equal(t, 1, got, "without hysteresis: spill to slow leg") + + got = ecfPick(legs, true, nil) + assert.Equal(t, 0, got, "with hysteresis latched: hold on fast leg") +} + +// TestSelectECF_Integration drives the selector end-to-end through SetECFState +// + SelectForPayload, confirming the mode routes via the ECF pick and that the +// inflight accounting charges the chosen leg. +func TestSelectECF_Integration(t *testing.T) { + ts := newTransportSelector() + ts.SetMode(WeightModeECF) + + // Two legs; the fast leg (0) has ample capacity, so every DATA frame + // should land on it until it saturates. + ts.SetECFState([]ecfLegState{ + {rttMs: 10, cwndBytes: 1e9, rateBps: 0, ready: true}, + {rttMs: 100, cwndBytes: 1e9, rateBps: 0, ready: true}, + }) + + payload := make([]byte, 512) + for i := 0; i < 50; i++ { + idx := ts.SelectForPayload(payload) + require.Equal(t, 0, idx, "fast leg with capacity should always win") + } + + // The fast leg should now show accumulated inflight; the slow leg none. + ts.mu.RLock() + fastInflight := ts.ecfLegs[0].inflightBytes + slowInflight := ts.ecfLegs[1].inflightBytes + ts.mu.RUnlock() + assert.Greater(t, fastInflight, 0.0, "chosen leg accrues inflight bytes") + assert.Equal(t, 0.0, slowInflight, "unchosen leg accrues nothing") +} + +// TestSelectECF_SpillsWhenFastSaturated confirms that once the fast leg is +// modeled as saturated with a large backlog, the selector spills DATA frames +// onto the slower ready leg rather than head-of-line-stalling on the fast one. +func TestSelectECF_SpillsWhenFastSaturated(t *testing.T) { + ts := newTransportSelector() + ts.SetMode(WeightModeECF) + + // Fast leg saturated with a huge backlog (rate 0 so time-drain won't clear + // it), small cwnd; slow leg has capacity. n*rttF = (1+200000/1000)*10 huge + // >> rttS -> spill. + ts.SetECFState([]ecfLegState{ + {rttMs: 10, cwndBytes: 1000, inflightBytes: 200000, rateBps: 0, ready: true}, + {rttMs: 100, cwndBytes: 1e9, rateBps: 0, ready: true}, + }) + + idx := ts.SelectForPayload(make([]byte, 512)) + assert.Equal(t, 1, idx, "saturated fast leg under huge backlog spills to slow leg") +} + +// TestSelectECF_BootstrapFallsBackToSchedule confirms that with no ECF state +// the selector falls back to the schedule (never panics / returns garbage). +func TestSelectECF_BootstrapFallsBackToSchedule(t *testing.T) { + ts := newTransportSelector() + ts.SetMode(WeightModeECF) + // No SetECFState call: ecfLegs empty. Provide a schedule via a fake rebuild + // by directly seeding the schedule the way Rebuild would for live legs. + ts.mu.Lock() + ts.schedule = []int{0, 1} + ts.mu.Unlock() + + idx := ts.SelectECF(512) + assert.Contains(t, []int{0, 1}, idx, "bootstrap falls back to a schedule index") +} diff --git a/pkg/visor/api_transport.go b/pkg/visor/api_transport.go index ec240a560a..a229c64c3d 100644 --- a/pkg/visor/api_transport.go +++ b/pkg/visor/api_transport.go @@ -174,8 +174,10 @@ func (v *Visor) SetMuxMode(mode string) error { m = router.WeightModeEqual case "capacity": m = router.WeightModeCapacity + case "ecf": + m = router.WeightModeECF default: - return fmt.Errorf("unknown mux mode %q (use \"auto\", \"equal\", or \"capacity\")", mode) + return fmt.Errorf("unknown mux mode %q (use \"auto\", \"equal\", \"capacity\", or \"ecf\")", mode) } v.router.SetMuxMode(m) v.log.Infof("SetMuxMode: %v", mode)