fix(stats): sample stalled command queue occupancy each period
- commit
- cbbf1eee32587cadd7599fb8965658ae4dc51825
- author
- brice · · 2026-10-19 18:48 UTC
- parent
- d2670da
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
5 files · +58 -2
M internal/adapters/wayland/commands_test.go +2 -2
| @@ -258,8 +258,8 @@ func TestCommandRingStats(t *testing.T) { | |||
| 258 | 258 | if st.maxLen != 3 || st.merged != 2 || st.maxPushBlock != 0 { | |
| 259 | 259 | t.Fatalf("stats = %+v, want maxLen 3, merged 2, no block", st) | |
| 260 | 260 | } | |
| 261 | - | if st := r.takeStats(); st != (ringStats{}) { | |
| 262 | - | t.Fatalf("stats after take = %+v, want zero", st) | |
| 261 | + | if st := r.takeStats(); st != (ringStats{maxLen: 3}) { | |
| 262 | + | t.Fatalf("stats after take = %+v, want current occupancy only", st) | |
| 263 | 263 | } | |
| 264 | 264 | release.do() | |
| 265 | 265 | } | |
| @@ -258,8 +258,8 @@ func TestCommandRingStats(t *testing.T) { | |||||
| 258 | if st.maxLen != 3 || st.merged != 2 || st.maxPushBlock != 0 { | 258 | if st.maxLen != 3 || st.merged != 2 || st.maxPushBlock != 0 { | ||
| 259 | t.Fatalf("stats = %+v, want maxLen 3, merged 2, no block", st) | 259 | t.Fatalf("stats = %+v, want maxLen 3, merged 2, no block", st) | ||
| 260 | } | 260 | } | ||
| 261 | - | if st := r.takeStats(); st != (ringStats{}) { | 261 | + | if st := r.takeStats(); st != (ringStats{maxLen: 3}) { |
| 262 | - | t.Fatalf("stats after take = %+v, want zero", st) | 262 | + | t.Fatalf("stats after take = %+v, want current occupancy only", st) |
| 263 | } | 263 | } | ||
| 264 | release.do() | 264 | release.do() | ||
| 265 | } | 265 | } | ||
M internal/adapters/wayland/postring.go +1 -0
| @@ -48,6 +48,7 @@ type ringStats struct { | |||
| 48 | 48 | func (r *postRing[T]) takeStats() ringStats { | |
| 49 | 49 | r.mu.Lock() | |
| 50 | 50 | st := r.stats | |
| 51 | + | st.maxLen = max(st.maxLen, len(r.ring)) | |
| 51 | 52 | r.stats = ringStats{} | |
| 52 | 53 | r.mu.Unlock() | |
| 53 | 54 | return st | |
| @@ -48,6 +48,7 @@ type ringStats struct { | |||||
| 48 | func (r *postRing[T]) takeStats() ringStats { | 48 | func (r *postRing[T]) takeStats() ringStats { | ||
| 49 | r.mu.Lock() | 49 | r.mu.Lock() | ||
| 50 | st := r.stats | 50 | st := r.stats | ||
| 51 | + | st.maxLen = max(st.maxLen, len(r.ring)) | |||
| 51 | r.stats = ringStats{} | 52 | r.stats = ringStats{} | ||
| 52 | r.mu.Unlock() | 53 | r.mu.Unlock() | ||
| 53 | return st | 54 | return st | ||
M internal/adapters/wayland/postring_test.go +34 -0
| @@ -207,3 +207,37 @@ func waitPushWaiting(t *testing.T) { | |||
| 207 | 207 | runtime.Gosched() | |
| 208 | 208 | } | |
| 209 | 209 | } | |
| 210 | + | ||
| 211 | + | func TestPostRingStatsSampleStalledOccupancy(t *testing.T) { | |
| 212 | + | s, _, _, _, _, _ := presentServer(t) | |
| 213 | + | release := make(chan struct{}) | |
| 214 | + | stalled := make(chan struct{}) | |
| 215 | + | s.display.Post(func() { close(stalled); <-release }) | |
| 216 | + | <-stalled | |
| 217 | + | // Release even if an assertion fails, so server cleanup cannot deadlock. | |
| 218 | + | released := false | |
| 219 | + | defer func() { | |
| 220 | + | if !released { | |
| 221 | + | close(release) | |
| 222 | + | } | |
| 223 | + | }() | |
| 224 | + | r := newPostRing(s.display, 2, func(int) {}) | |
| 225 | + | for i := range 2 { | |
| 226 | + | if !r.Push(i) { | |
| 227 | + | t.Fatal("push refused") | |
| 228 | + | } | |
| 229 | + | } | |
| 230 | + | for range 2 { | |
| 231 | + | if got := r.takeStats().maxLen; got != 2 { | |
| 232 | + | t.Fatalf("occupancy %d, want 2", got) | |
| 233 | + | } | |
| 234 | + | } | |
| 235 | + | close(release) | |
| 236 | + | released = true | |
| 237 | + | if !s.display.Do(func() {}) { | |
| 238 | + | t.Fatal("display stopped") | |
| 239 | + | } | |
| 240 | + | if got := r.takeStats().maxLen; got != 0 { | |
| 241 | + | t.Fatalf("recovered occupancy %d", got) | |
| 242 | + | } | |
| 243 | + | } | |
| @@ -207,3 +207,37 @@ func waitPushWaiting(t *testing.T) { | |||||
| 207 | runtime.Gosched() | 207 | runtime.Gosched() | ||
| 208 | } | 208 | } | ||
| 209 | } | 209 | } | ||
| 210 | + | ||||
| 211 | + | func TestPostRingStatsSampleStalledOccupancy(t *testing.T) { | |||
| 212 | + | s, _, _, _, _, _ := presentServer(t) | |||
| 213 | + | release := make(chan struct{}) | |||
| 214 | + | stalled := make(chan struct{}) | |||
| 215 | + | s.display.Post(func() { close(stalled); <-release }) | |||
| 216 | + | <-stalled | |||
| 217 | + | // Release even if an assertion fails, so server cleanup cannot deadlock. | |||
| 218 | + | released := false | |||
| 219 | + | defer func() { | |||
| 220 | + | if !released { | |||
| 221 | + | close(release) | |||
| 222 | + | } | |||
| 223 | + | }() | |||
| 224 | + | r := newPostRing(s.display, 2, func(int) {}) | |||
| 225 | + | for i := range 2 { | |||
| 226 | + | if !r.Push(i) { | |||
| 227 | + | t.Fatal("push refused") | |||
| 228 | + | } | |||
| 229 | + | } | |||
| 230 | + | for range 2 { | |||
| 231 | + | if got := r.takeStats().maxLen; got != 2 { | |||
| 232 | + | t.Fatalf("occupancy %d, want 2", got) | |||
| 233 | + | } | |||
| 234 | + | } | |||
| 235 | + | close(release) | |||
| 236 | + | released = true | |||
| 237 | + | if !s.display.Do(func() {}) { | |||
| 238 | + | t.Fatal("display stopped") | |||
| 239 | + | } | |||
| 240 | + | if got := r.takeStats().maxLen; got != 0 { | |||
| 241 | + | t.Fatalf("recovered occupancy %d", got) | |||
| 242 | + | } | |||
| 243 | + | } | |||
M internal/core/outbox.go +1 -0
| @@ -61,6 +61,7 @@ func (o *outbox) takeStats() CommandStats { | |||
| 61 | 61 | o.mu.Lock() | |
| 62 | 62 | defer o.mu.Unlock() | |
| 63 | 63 | s := o.stats | |
| 64 | + | s.MaxBacklog = max(s.MaxBacklog, o.pushed-o.sent) | |
| 64 | 65 | o.stats = CommandStats{} | |
| 65 | 66 | return s | |
| 66 | 67 | } | |
| @@ -61,6 +61,7 @@ func (o *outbox) takeStats() CommandStats { | |||||
| 61 | o.mu.Lock() | 61 | o.mu.Lock() | ||
| 62 | defer o.mu.Unlock() | 62 | defer o.mu.Unlock() | ||
| 63 | s := o.stats | 63 | s := o.stats | ||
| 64 | + | s.MaxBacklog = max(s.MaxBacklog, o.pushed-o.sent) | |||
| 64 | o.stats = CommandStats{} | 65 | o.stats = CommandStats{} | ||
| 65 | return s | 66 | return s | ||
| 66 | } | 67 | } | ||
M internal/core/outbox_test.go +20 -0
| @@ -210,3 +210,23 @@ func TestOutboxQueueWaitIsPerEntry(t *testing.T) { | |||
| 210 | 210 | } | |
| 211 | 211 | }) | |
| 212 | 212 | } | |
| 213 | + | ||
| 214 | + | // A stalled sender must remain visible in every reporting period, including | |
| 215 | + | // when it has taken the queue into its private delivery batch. | |
| 216 | + | func TestOutboxStatsSampleOutstandingBacklog(t *testing.T) { | |
| 217 | + | o := newOutbox() | |
| 218 | + | out := make(chan ports.ClientCommand) | |
| 219 | + | for range outboxLimit { | |
| 220 | + | o.push(out, ports.FocusWindow{ID: 1}, ports.SecurityState{}, false) | |
| 221 | + | } | |
| 222 | + | batch := o.take() | |
| 223 | + | for range 2 { | |
| 224 | + | if got := o.takeStats().MaxBacklog; got != outboxLimit { | |
| 225 | + | t.Fatalf("backlog %d, want %d", got, outboxLimit) | |
| 226 | + | } | |
| 227 | + | } | |
| 228 | + | o.recycle(batch) | |
| 229 | + | if got := o.takeStats().MaxBacklog; got != 0 { | |
| 230 | + | t.Fatalf("recovered backlog %d", got) | |
| 231 | + | } | |
| 232 | + | } | |
| @@ -210,3 +210,23 @@ func TestOutboxQueueWaitIsPerEntry(t *testing.T) { | |||||
| 210 | } | 210 | } | ||
| 211 | }) | 211 | }) | ||
| 212 | } | 212 | } | ||
| 213 | + | ||||
| 214 | + | // A stalled sender must remain visible in every reporting period, including | |||
| 215 | + | // when it has taken the queue into its private delivery batch. | |||
| 216 | + | func TestOutboxStatsSampleOutstandingBacklog(t *testing.T) { | |||
| 217 | + | o := newOutbox() | |||
| 218 | + | out := make(chan ports.ClientCommand) | |||
| 219 | + | for range outboxLimit { | |||
| 220 | + | o.push(out, ports.FocusWindow{ID: 1}, ports.SecurityState{}, false) | |||
| 221 | + | } | |||
| 222 | + | batch := o.take() | |||
| 223 | + | for range 2 { | |||
| 224 | + | if got := o.takeStats().MaxBacklog; got != outboxLimit { | |||
| 225 | + | t.Fatalf("backlog %d, want %d", got, outboxLimit) | |||
| 226 | + | } | |||
| 227 | + | } | |||
| 228 | + | o.recycle(batch) | |||
| 229 | + | if got := o.takeStats().MaxBacklog; got != 0 { | |||
| 230 | + | t.Fatalf("recovered backlog %d", got) | |||
| 231 | + | } | |||
| 232 | + | } | |||