diff --git a/cmd/easyss/main.go b/cmd/easyss/main.go index 9eb0adde..9a3e0dd7 100644 --- a/cmd/easyss/main.go +++ b/cmd/easyss/main.go @@ -294,6 +294,7 @@ func (a *App) statsLoop() { "conns", snap.Conns, "priority_conns", snap.PriorityConns, "bulk_conns", snap.BulkConns, + "conns_status", snap.ConnsStatus, "active_streams", snap.ActiveStreams, "priority_active", snap.PriorityActiveStreams, "bulk_active", snap.BulkActiveStreams, diff --git a/transport/http2/client.go b/transport/http2/client.go index 2f3f1ca5..c12feca9 100644 --- a/transport/http2/client.go +++ b/transport/http2/client.go @@ -8,6 +8,9 @@ import ( "net" "net/http" "runtime" + "slices" + "strconv" + "strings" "sync" "time" @@ -96,6 +99,7 @@ func New(cfg Config) (*HTTP2Transport, error) { slots := make([]*transportSlot, maxSlots) for i := range slots { slots[i] = newSlot(cfg.TLSConfig, timeout, dialCtx, connLifetime) + slots[i].idx = i } sched := newScheduler(maxSlots, slots, threshold, prioritySlots) @@ -304,6 +308,13 @@ func (t *HTTP2Transport) CloseIdle() { } func (t *HTTP2Transport) Stats() transport.TransportStats { + // Hold the scheduler read lock so the snapshot is consistent: shrink + // (swap-remove) and grow mutate liveCount and the live slot range under + // the write lock, so an unlocked render could read a stale liveCount + // and report more conns_status entries than Conns. + t.sched.mu.RLock() + defer t.sched.mu.RUnlock() + live := int(t.sched.liveCount.Load()) ts := transport.TransportStats{ Conns: live, @@ -324,9 +335,77 @@ func (t *HTTP2Transport) Stats() transport.TransportStats { ts.BulkActiveStreams += a } } + ts.ConnsStatus = slotStatusString(t.sched, live) return ts } +// slotStatus derives a live slot's connection status from its health flags. +// Multiple flags are joined with "+" so no state is hidden (a heavy download +// crossing the connection lifetime is both heavy and expiring); a slot with +// no flags is "active". +func slotStatus(s *transportSlot) string { + var parts []string + if s.heavy.Load() > 0 { + parts = append(parts, "heavy") + } + if s.degraded.Load() { + parts = append(parts, "degraded") + } + if s.expiring.Load() { + parts = append(parts, "expiring") + } + if len(parts) == 0 { + return "active" + } + return strings.Join(parts, "+") +} + +// slotStatusString renders every live slot as ":: +// ", wrapped in brackets, e.g. "[0:3:degraded, 1:2:expiring, +// 2:1:active, 3:1:heavy]". Entries are ordered by the stable slot identity +// (retire swap-removes scramble the live order) and then renumbered from 0, +// so the rendered indices are always consecutive with no jumps. An empty +// live set renders as "[]". live must be the liveCount value the caller +// snapshot under the scheduler lock, so the rendered entry count always +// matches Conns. +func slotStatusString(sched *slotScheduler, live int) string { + if live == 0 { + return "[]" + } + type entry struct { + idx int + active int + status string + } + entries := make([]entry, 0, live) + for i := 0; i < live; i++ { + s := sched.slots[i] + entries = append(entries, entry{ + idx: s.idx, + active: int(s.active.Load()), + status: slotStatus(s), + }) + } + // Order by stable slot index regardless of the scrambled live order, + // then number entries 0..n-1 so the output indices never jump. + slices.SortFunc(entries, func(a, b entry) int { return a.idx - b.idx }) + + var b strings.Builder + b.WriteByte('[') + for i, e := range entries { + if i > 0 { + b.WriteString(", ") + } + b.WriteString(strconv.Itoa(i)) + b.WriteByte(':') + b.WriteString(strconv.Itoa(e.active)) + b.WriteByte(':') + b.WriteString(e.status) + } + b.WriteByte(']') + return b.String() +} + func (t *HTTP2Transport) Close() error { t.cancel() live := t.sched.liveCount.Load() diff --git a/transport/http2/consistency_test.go b/transport/http2/consistency_test.go new file mode 100644 index 00000000..6d67050d --- /dev/null +++ b/transport/http2/consistency_test.go @@ -0,0 +1,102 @@ +package http2 + +import ( + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// TestStatsConnsStatusConsistencyUnderShrink stresses the scheduler with a +// shrinker goroutine (simulating closeIdleLoop) while a stats reader +// goroutine (simulating /stats polling) renders conns_status. It fails if the +// rendered entry count ever exceeds the concurrently reported Conns value, +// or if the rendered indices are not consecutive from 0. +func TestStatsConnsStatusConsistencyUnderShrink(t *testing.T) { + slots := make([]*transportSlot, 6) + for i := range slots { + slots[i] = &transportSlot{idx: i} + } + sch := newScheduler(6, slots, 2, 1) + sch.liveCount.Store(6) + + var over atomic.Int64 + var badIdx atomic.Int64 + var reads atomic.Int64 + done := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + for { + select { + case <-done: + return + default: + } + sch.mu.Lock() + sch.shrinkIdleLocked() + sch.mu.Unlock() + // Re-grow like new streams arriving. + sch.grow(false) + } + }() + + go func() { + defer wg.Done() + for { + select { + case <-done: + return + default: + } + live := int(sch.liveCount.Load()) + s := slotStatusString(sch, live) + idxs := parseIndices(s) + if len(idxs) > live { + over.Add(int64(len(idxs) - live)) + } + for k, id := range idxs { + if id != k { + badIdx.Add(1) + break + } + } + reads.Add(1) + } + }() + + time.Sleep(200 * time.Millisecond) + close(done) + wg.Wait() + t.Logf("reads=%d over_by_total=%d bad_idx_total=%d", reads.Load(), over.Load(), badIdx.Load()) + if over.Load() > 0 { + t.Fatalf("conns_status rendered more entries than Conns: over=%d", over.Load()) + } + if badIdx.Load() > 0 { + t.Fatalf("conns_status indices are not consecutive from 0: bad=%d", badIdx.Load()) + } +} + +// parseIndices extracts the leading ":" of each entry, verifying the +// array-like shape at the same time. +func parseIndices(s string) []int { + if s == "[]" { + return nil + } + body := strings.TrimSuffix(strings.TrimPrefix(s, "["), "]") + parts := strings.Split(body, ", ") + idxs := make([]int, 0, len(parts)) + for _, p := range parts { + colon := strings.IndexByte(p, ':') + n, err := strconv.Atoi(p[:colon]) + if err != nil { + panic("malformed entry: " + p) + } + idxs = append(idxs, n) + } + return idxs +} diff --git a/transport/http2/scheduler_test.go b/transport/http2/scheduler_test.go index bdf816b6..fb5f05a8 100644 --- a/transport/http2/scheduler_test.go +++ b/transport/http2/scheduler_test.go @@ -11,7 +11,7 @@ import ( func newTestScheduler(specs ...[2]int32) *slotScheduler { slots := make([]*transportSlot, len(specs)) for i, s := range specs { - slots[i] = &transportSlot{} + slots[i] = &transportSlot{idx: i} slots[i].active.Store(s[0]) slots[i].heavy.Store(s[1]) } diff --git a/transport/http2/slot.go b/transport/http2/slot.go index b115288c..8977e64e 100644 --- a/transport/http2/slot.go +++ b/transport/http2/slot.go @@ -10,6 +10,11 @@ import ( // to a single connection via MaxConnsPerHost=1) plus the state that the // scheduler and the connection lifecycle act on. type transportSlot struct { + // idx is the slot's stable index in the scheduler's pre-allocated + // array, set once at construction. Retire swap-removes slots, so the + // live position and idx diverge; idx is used for stats reporting. + idx int + t *http.Transport active atomic.Int32 heavy atomic.Int32 // number of active heavy streams (>= HeavyStreamThreshold bytes) diff --git a/transport/http2/slot_status_test.go b/transport/http2/slot_status_test.go new file mode 100644 index 00000000..3e599aed --- /dev/null +++ b/transport/http2/slot_status_test.go @@ -0,0 +1,94 @@ +package http2 + +import ( + "testing" +) + +func TestSlotStatus(t *testing.T) { + tests := []struct { + name string + heavy int32 + deg bool + exp bool + expect string + }{ + {name: "no flags is active", expect: "active"}, + {name: "heavy", heavy: 1, expect: "heavy"}, + {name: "degraded", deg: true, expect: "degraded"}, + {name: "expiring", exp: true, expect: "expiring"}, + {name: "heavy and expiring", heavy: 1, exp: true, expect: "heavy+expiring"}, + {name: "degraded and heavy", heavy: 2, deg: true, expect: "heavy+degraded"}, + {name: "all flags", heavy: 1, deg: true, exp: true, expect: "heavy+degraded+expiring"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := &transportSlot{} + s.heavy.Store(tt.heavy) + s.degraded.Store(tt.deg) + s.expiring.Store(tt.exp) + if got := slotStatus(s); got != tt.expect { + t.Fatalf("slotStatus() = %q, want %q", got, tt.expect) + } + }) + } +} + +func TestSlotStatusString(t *testing.T) { + t.Run("empty live set", func(t *testing.T) { + sch := newTestScheduler() + if got := slotStatusString(sch, int(sch.liveCount.Load())); got != "[]" { + t.Fatalf("slotStatusString() = %q, want %q", got, "[]") + } + }) + + t.Run("all active with stream counts", func(t *testing.T) { + sch := newTestScheduler([2]int32{1, 0}, [2]int32{2, 0}) + if got, want := slotStatusString(sch, int(sch.liveCount.Load())), "[0:1:active, 1:2:active]"; got != want { + t.Fatalf("slotStatusString() = %q, want %q", got, want) + } + }) + + t.Run("mixed states", func(t *testing.T) { + sch := newTestScheduler([2]int32{3, 0}, [2]int32{2, 0}, [2]int32{1, 0}, [2]int32{1, 1}) + sch.slots[0].degraded.Store(true) + sch.slots[1].expiring.Store(true) + want := "[0:3:degraded, 1:2:expiring, 2:1:active, 3:1:heavy]" + if got := slotStatusString(sch, int(sch.liveCount.Load())); got != want { + t.Fatalf("slotStatusString() = %q, want %q", got, want) + } + }) + + t.Run("combined flags", func(t *testing.T) { + sch := newTestScheduler([2]int32{1, 1}, [2]int32{0, 0}) + sch.slots[0].expiring.Store(true) + sch.slots[1].degraded.Store(true) + want := "[0:1:heavy+expiring, 1:0:degraded]" + if got := slotStatusString(sch, int(sch.liveCount.Load())); got != want { + t.Fatalf("slotStatusString() = %q, want %q", got, want) + } + }) + + t.Run("respects liveCount bound", func(t *testing.T) { + sch := newTestScheduler([2]int32{1, 0}, [2]int32{2, 0}, [2]int32{3, 0}) + sch.slots[2].degraded.Store(true) + sch.liveCount.Store(2) + want := "[0:1:active, 1:2:active]" + if got := slotStatusString(sch, int(sch.liveCount.Load())); got != want { + t.Fatalf("slotStatusString() = %q, want %q", got, want) + } + }) + + t.Run("scrambled live order is renumbered consecutively", func(t *testing.T) { + // Simulate a swap-remove of slot 1: positions [0,1,2] now hold + // slots with real indices 0,3,2 and liveCount dropped to 3. Entries + // stay ordered by real index but are renumbered 0..n-1, so the + // output indices never jump. + sch := newTestScheduler([2]int32{1, 0}, [2]int32{2, 0}, [2]int32{3, 0}, [2]int32{4, 0}) + sch.slots[1], sch.slots[3] = sch.slots[3], sch.slots[1] + sch.liveCount.Store(3) + want := "[0:1:active, 1:3:active, 2:4:active]" + if got := slotStatusString(sch, int(sch.liveCount.Load())); got != want { + t.Fatalf("slotStatusString() = %q, want %q", got, want) + } + }) +} diff --git a/transport/transport.go b/transport/transport.go index bd375e5f..de1db711 100644 --- a/transport/transport.go +++ b/transport/transport.go @@ -51,6 +51,12 @@ type TransportStats struct { BulkActiveStreams int `json:"bulk_active_streams"` PriorityConns int `json:"priority_conns"` BulkConns int `json:"bulk_conns"` + // ConnsStatus is a compact per-connection status summary, e.g. + // "[0:3:degraded, 1:2:expiring, 2:1:active, 3:1:heavy]". Each element is + // "::" where indices are consecutive from + // 0 (ordered by stable connection identity) and status is one of + // active/heavy/degraded/expiring, with multiple flags joined by "+". + ConnsStatus string `json:"conns_status,omitempty"` } type Transport interface {