Merge branch 'fix/review-outbox' into fix/release-review
18 files · +159 -235
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/imagecapture_test.go +1 -30
| @@ -242,39 +242,10 @@ func noSession(t *testing.T, cc *cclient, ev chan []uint32) { | |||
| 242 | 242 | // client received what followed. | |
| 243 | 243 | func (h *captureHarness) settle(t *testing.T, cc *cclient) { | |
| 244 | 244 | t.Helper() | |
| 245 | - | // The command loop takes commands off the channel before it applies | |
| 246 | - | // them on the display: an empty channel does not mean applied. Commands | |
| 247 | - | // apply in order, so a marker (SlotsPending, a plain flag) seen on the | |
| 248 | - | // display means every command sent before it was applied. | |
| 249 | - | var was bool | |
| 250 | - | if !h.s.display.Do(func() { was = h.s.slotsPending }) { | |
| 251 | - | t.Fatal("display stopped") | |
| 252 | - | } | |
| 253 | - | h.applied(t, !was) | |
| 254 | - | h.applied(t, was) | |
| 245 | + | syncCommands(t, h.s, h.commands, cc.c) | |
| 255 | 246 | cc.roundtrip(t) | |
| 256 | 247 | } | |
| 257 | 248 | ||
| 258 | - | // applied sends SlotsPending{pending} and waits until the display applied it. | |
| 259 | - | func (h *captureHarness) applied(t *testing.T, pending bool) { | |
| 260 | - | t.Helper() | |
| 261 | - | h.commands <- ports.SlotsPending{Pending: pending} | |
| 262 | - | deadline := time.Now().Add(2 * time.Second) | |
| 263 | - | for { | |
| 264 | - | var now bool | |
| 265 | - | if !h.s.display.Do(func() { now = h.s.slotsPending }) { | |
| 266 | - | t.Fatal("display stopped") | |
| 267 | - | } | |
| 268 | - | if now == pending { | |
| 269 | - | return | |
| 270 | - | } | |
| 271 | - | if time.Now().After(deadline) { | |
| 272 | - | t.Fatal("commands not applied") | |
| 273 | - | } | |
| 274 | - | time.Sleep(time.Millisecond) | |
| 275 | - | } | |
| 276 | - | } | |
| 277 | - | ||
| 278 | 249 | func (h *captureHarness) send(cmd ports.ClientCommand) { h.commands <- cmd } | |
| 279 | 250 | ||
| 280 | 251 | // captureShm makes a w x h XRGB8888 wl_buffer. | |
| @@ -242,39 +242,10 @@ func noSession(t *testing.T, cc *cclient, ev chan []uint32) { | |||||
| 242 | // client received what followed. | 242 | // client received what followed. | ||
| 243 | func (h *captureHarness) settle(t *testing.T, cc *cclient) { | 243 | func (h *captureHarness) settle(t *testing.T, cc *cclient) { | ||
| 244 | t.Helper() | 244 | t.Helper() | ||
| 245 | - | // The command loop takes commands off the channel before it applies | 245 | + | syncCommands(t, h.s, h.commands, cc.c) |
| 246 | - | // them on the display: an empty channel does not mean applied. Commands | |||
| 247 | - | // apply in order, so a marker (SlotsPending, a plain flag) seen on the | |||
| 248 | - | // display means every command sent before it was applied. | |||
| 249 | - | var was bool | |||
| 250 | - | if !h.s.display.Do(func() { was = h.s.slotsPending }) { | |||
| 251 | - | t.Fatal("display stopped") | |||
| 252 | - | } | |||
| 253 | - | h.applied(t, !was) | |||
| 254 | - | h.applied(t, was) | |||
| 255 | cc.roundtrip(t) | 246 | cc.roundtrip(t) | ||
| 256 | } | 247 | } | ||
| 257 | 248 | ||||
| 258 | - | // applied sends SlotsPending{pending} and waits until the display applied it. | |||
| 259 | - | func (h *captureHarness) applied(t *testing.T, pending bool) { | |||
| 260 | - | t.Helper() | |||
| 261 | - | h.commands <- ports.SlotsPending{Pending: pending} | |||
| 262 | - | deadline := time.Now().Add(2 * time.Second) | |||
| 263 | - | for { | |||
| 264 | - | var now bool | |||
| 265 | - | if !h.s.display.Do(func() { now = h.s.slotsPending }) { | |||
| 266 | - | t.Fatal("display stopped") | |||
| 267 | - | } | |||
| 268 | - | if now == pending { | |||
| 269 | - | return | |||
| 270 | - | } | |||
| 271 | - | if time.Now().After(deadline) { | |||
| 272 | - | t.Fatal("commands not applied") | |||
| 273 | - | } | |||
| 274 | - | time.Sleep(time.Millisecond) | |||
| 275 | - | } | |||
| 276 | - | } | |||
| 277 | - | ||||
| 278 | func (h *captureHarness) send(cmd ports.ClientCommand) { h.commands <- cmd } | 249 | func (h *captureHarness) send(cmd ports.ClientCommand) { h.commands <- cmd } | ||
| 279 | 250 | ||||
| 280 | // captureShm makes a w x h XRGB8888 wl_buffer. | 251 | // captureShm makes a w x h XRGB8888 wl_buffer. | ||
M internal/adapters/wayland/keyboard_test.go +59 -23
| @@ -129,12 +129,16 @@ type eventsProxy struct { | |||
| 129 | 129 | wlturbo.BaseProxy | |
| 130 | 130 | opcodes []uint16 | |
| 131 | 131 | keymaps []string | |
| 132 | - | locked []uint32 // locked modifiers of each modifiers event | |
| 133 | - | depress []uint32 // depressed modifiers of each modifiers event | |
| 132 | + | locked []uint32 // locked modifiers of each modifiers event | |
| 133 | + | depress []uint32 // depressed modifiers of each modifiers event | |
| 134 | + | repeat chan struct{} // optional protocol barrier notification | |
| 134 | 135 | } | |
| 135 | 136 | ||
| 136 | 137 | func (p *eventsProxy) Dispatch(e *wlturbo.Event) { | |
| 137 | 138 | p.opcodes = append(p.opcodes, e.Opcode) | |
| 139 | + | if e.Opcode == uint16(wayland.KeyboardEventRepeatInfo) && p.repeat != nil { | |
| 140 | + | p.repeat <- struct{}{} | |
| 141 | + | } | |
| 138 | 142 | if e.Opcode == uint16(wayland.KeyboardEventModifiers) { | |
| 139 | 143 | e.Uint32() // serial | |
| 140 | 144 | p.depress = append(p.depress, e.Uint32()) | |
| @@ -267,30 +271,11 @@ func TestSetModifiersOverrideAndRestore(t *testing.T) { | |||
| 267 | 271 | set := make(chan struct{}) | |
| 268 | 272 | s.display.Do(func() { s.seat.modState = ports.ModState{Locked: capsLocked}; id = s.seat.focused; close(set) }) | |
| 269 | 273 | <-set | |
| 270 | - | // A marker command applied on the display means every command sent before | |
| 271 | - | // it was applied (they apply in order); the roundtrip then delivers the | |
| 272 | - | // events they sent. | |
| 273 | - | var pending bool | |
| 274 | 274 | wait := func() { | |
| 275 | 275 | t.Helper() | |
| 276 | - | pending = !pending | |
| 277 | - | commands <- ports.SlotsPending{Pending: pending} | |
| 278 | - | deadline := time.Now().Add(2 * time.Second) | |
| 279 | - | for { | |
| 280 | - | var now bool | |
| 281 | - | if !s.display.Do(func() { now = s.slotsPending }) { | |
| 282 | - | t.Fatal("display stopped") | |
| 283 | - | } | |
| 284 | - | if now == pending { | |
| 285 | - | break | |
| 286 | - | } | |
| 287 | - | if time.Now().After(deadline) { | |
| 288 | - | t.Fatal("commands not applied") | |
| 289 | - | } | |
| 290 | - | time.Sleep(time.Millisecond) | |
| 291 | - | } | |
| 292 | - | roundtrip(t, target) | |
| 276 | + | syncCommands(t, s, commands, target) | |
| 293 | 277 | } | |
| 278 | + | ||
| 294 | 279 | proxy.depress, proxy.locked = nil, nil | |
| 295 | 280 | commands <- ports.SetModifiers{ID: id + 100, Mods: ports.ModCtrl} | |
| 296 | 281 | wait() | |
| @@ -341,3 +326,54 @@ func TestSetModifiersRestoreAdoptsCoreState(t *testing.T) { | |||
| 341 | 326 | t.Fatalf("restore depressed %v locked %v", proxy.depress, proxy.locked) | |
| 342 | 327 | } | |
| 343 | 328 | } | |
| 329 | + | ||
| 330 | + | // syncCommands observes an existing command through the real protocol. A | |
| 331 | + | // repeat_info emitted by SetKeymap means all preceding commands were applied; | |
| 332 | + | // merely emptying the channel or posting a display callback cannot prove that. | |
| 333 | + | func syncCommands(t *testing.T, s *Server, commands chan<- ports.ClientCommand, c *wlturbo.Display) { | |
| 334 | + | t.Helper() | |
| 335 | + | var needsKeymap bool | |
| 336 | + | if !s.display.Do(func() { needsKeymap = s.seat.keymapFD < 0 }) { | |
| 337 | + | t.Fatal("display stopped") | |
| 338 | + | } | |
| 339 | + | if needsKeymap { | |
| 340 | + | keymap := keymapText(t) | |
| 341 | + | if !s.display.Do(func() { s.setKeymap(ports.SetKeymap{Keymap: keymap}) }) { | |
| 342 | + | t.Fatal("display stopped") | |
| 343 | + | } | |
| 344 | + | } | |
| 345 | + | seat := bindVersion(t, c, "wl_seat", 4) | |
| 346 | + | registerProtocol(t, c, seat) | |
| 347 | + | id := c.AllocateID() | |
| 348 | + | p := &eventsProxy{repeat: make(chan struct{}, 2)} | |
| 349 | + | p.SetID(id) | |
| 350 | + | registerWireProxy(c, p) | |
| 351 | + | requestProtocol(t, c, seat, wayland.SeatRequestGetKeyboard, id) | |
| 352 | + | roundtrip(t, c) | |
| 353 | + | select { | |
| 354 | + | case <-p.repeat: | |
| 355 | + | default: | |
| 356 | + | t.Fatal("missing initial repeat_info") | |
| 357 | + | } | |
| 358 | + | var rate, delay int | |
| 359 | + | if !s.display.Do(func() { rate, delay = s.seat.repeatRate, s.seat.repeatDelay }) { | |
| 360 | + | t.Fatal("display stopped") | |
| 361 | + | } | |
| 362 | + | commands <- ports.SetKeymap{RepeatRate: rate, RepeatDelay: delay} | |
| 363 | + | // Dispatch blocks on the protocol socket; close it on timeout to bound | |
| 364 | + | // failure without polling the display or sleeping for command delivery. | |
| 365 | + | timeout := time.AfterFunc(2*time.Second, func() { _ = c.Close() }) | |
| 366 | + | defer timeout.Stop() | |
| 367 | + | for { | |
| 368 | + | select { | |
| 369 | + | case <-p.repeat: | |
| 370 | + | requestProtocol(t, c, id, wayland.KeyboardRequestRelease) | |
| 371 | + | roundtrip(t, c) | |
| 372 | + | return | |
| 373 | + | default: | |
| 374 | + | } | |
| 375 | + | if err := c.Dispatch(); err != nil { | |
| 376 | + | t.Fatal(err) | |
| 377 | + | } | |
| 378 | + | } | |
| 379 | + | } | |
| @@ -129,12 +129,16 @@ type eventsProxy struct { | |||||
| 129 | wlturbo.BaseProxy | 129 | wlturbo.BaseProxy | ||
| 130 | opcodes []uint16 | 130 | opcodes []uint16 | ||
| 131 | keymaps []string | 131 | keymaps []string | ||
| 132 | - | locked []uint32 // locked modifiers of each modifiers event | 132 | + | locked []uint32 // locked modifiers of each modifiers event |
| 133 | - | depress []uint32 // depressed modifiers of each modifiers event | 133 | + | depress []uint32 // depressed modifiers of each modifiers event |
| 134 | + | repeat chan struct{} // optional protocol barrier notification | |||
| 134 | } | 135 | } | ||
| 135 | 136 | ||||
| 136 | func (p *eventsProxy) Dispatch(e *wlturbo.Event) { | 137 | func (p *eventsProxy) Dispatch(e *wlturbo.Event) { | ||
| 137 | p.opcodes = append(p.opcodes, e.Opcode) | 138 | p.opcodes = append(p.opcodes, e.Opcode) | ||
| 139 | + | if e.Opcode == uint16(wayland.KeyboardEventRepeatInfo) && p.repeat != nil { | |||
| 140 | + | p.repeat <- struct{}{} | |||
| 141 | + | } | |||
| 138 | if e.Opcode == uint16(wayland.KeyboardEventModifiers) { | 142 | if e.Opcode == uint16(wayland.KeyboardEventModifiers) { | ||
| 139 | e.Uint32() // serial | 143 | e.Uint32() // serial | ||
| 140 | p.depress = append(p.depress, e.Uint32()) | 144 | p.depress = append(p.depress, e.Uint32()) | ||
| @@ -267,30 +271,11 @@ func TestSetModifiersOverrideAndRestore(t *testing.T) { | |||||
| 267 | set := make(chan struct{}) | 271 | set := make(chan struct{}) | ||
| 268 | s.display.Do(func() { s.seat.modState = ports.ModState{Locked: capsLocked}; id = s.seat.focused; close(set) }) | 272 | s.display.Do(func() { s.seat.modState = ports.ModState{Locked: capsLocked}; id = s.seat.focused; close(set) }) | ||
| 269 | <-set | 273 | <-set | ||
| 270 | - | // A marker command applied on the display means every command sent before | |||
| 271 | - | // it was applied (they apply in order); the roundtrip then delivers the | |||
| 272 | - | // events they sent. | |||
| 273 | - | var pending bool | |||
| 274 | wait := func() { | 274 | wait := func() { | ||
| 275 | t.Helper() | 275 | t.Helper() | ||
| 276 | - | pending = !pending | 276 | + | syncCommands(t, s, commands, target) |
| 277 | - | commands <- ports.SlotsPending{Pending: pending} | |||
| 278 | - | deadline := time.Now().Add(2 * time.Second) | |||
| 279 | - | for { | |||
| 280 | - | var now bool | |||
| 281 | - | if !s.display.Do(func() { now = s.slotsPending }) { | |||
| 282 | - | t.Fatal("display stopped") | |||
| 283 | - | } | |||
| 284 | - | if now == pending { | |||
| 285 | - | break | |||
| 286 | - | } | |||
| 287 | - | if time.Now().After(deadline) { | |||
| 288 | - | t.Fatal("commands not applied") | |||
| 289 | - | } | |||
| 290 | - | time.Sleep(time.Millisecond) | |||
| 291 | - | } | |||
| 292 | - | roundtrip(t, target) | |||
| 293 | } | 277 | } | ||
| 278 | + | ||||
| 294 | proxy.depress, proxy.locked = nil, nil | 279 | proxy.depress, proxy.locked = nil, nil | ||
| 295 | commands <- ports.SetModifiers{ID: id + 100, Mods: ports.ModCtrl} | 280 | commands <- ports.SetModifiers{ID: id + 100, Mods: ports.ModCtrl} | ||
| 296 | wait() | 281 | wait() | ||
| @@ -341,3 +326,54 @@ func TestSetModifiersRestoreAdoptsCoreState(t *testing.T) { | |||||
| 341 | t.Fatalf("restore depressed %v locked %v", proxy.depress, proxy.locked) | 326 | t.Fatalf("restore depressed %v locked %v", proxy.depress, proxy.locked) | ||
| 342 | } | 327 | } | ||
| 343 | } | 328 | } | ||
| 329 | + | ||||
| 330 | + | // syncCommands observes an existing command through the real protocol. A | |||
| 331 | + | // repeat_info emitted by SetKeymap means all preceding commands were applied; | |||
| 332 | + | // merely emptying the channel or posting a display callback cannot prove that. | |||
| 333 | + | func syncCommands(t *testing.T, s *Server, commands chan<- ports.ClientCommand, c *wlturbo.Display) { | |||
| 334 | + | t.Helper() | |||
| 335 | + | var needsKeymap bool | |||
| 336 | + | if !s.display.Do(func() { needsKeymap = s.seat.keymapFD < 0 }) { | |||
| 337 | + | t.Fatal("display stopped") | |||
| 338 | + | } | |||
| 339 | + | if needsKeymap { | |||
| 340 | + | keymap := keymapText(t) | |||
| 341 | + | if !s.display.Do(func() { s.setKeymap(ports.SetKeymap{Keymap: keymap}) }) { | |||
| 342 | + | t.Fatal("display stopped") | |||
| 343 | + | } | |||
| 344 | + | } | |||
| 345 | + | seat := bindVersion(t, c, "wl_seat", 4) | |||
| 346 | + | registerProtocol(t, c, seat) | |||
| 347 | + | id := c.AllocateID() | |||
| 348 | + | p := &eventsProxy{repeat: make(chan struct{}, 2)} | |||
| 349 | + | p.SetID(id) | |||
| 350 | + | registerWireProxy(c, p) | |||
| 351 | + | requestProtocol(t, c, seat, wayland.SeatRequestGetKeyboard, id) | |||
| 352 | + | roundtrip(t, c) | |||
| 353 | + | select { | |||
| 354 | + | case <-p.repeat: | |||
| 355 | + | default: | |||
| 356 | + | t.Fatal("missing initial repeat_info") | |||
| 357 | + | } | |||
| 358 | + | var rate, delay int | |||
| 359 | + | if !s.display.Do(func() { rate, delay = s.seat.repeatRate, s.seat.repeatDelay }) { | |||
| 360 | + | t.Fatal("display stopped") | |||
| 361 | + | } | |||
| 362 | + | commands <- ports.SetKeymap{RepeatRate: rate, RepeatDelay: delay} | |||
| 363 | + | // Dispatch blocks on the protocol socket; close it on timeout to bound | |||
| 364 | + | // failure without polling the display or sleeping for command delivery. | |||
| 365 | + | timeout := time.AfterFunc(2*time.Second, func() { _ = c.Close() }) | |||
| 366 | + | defer timeout.Stop() | |||
| 367 | + | for { | |||
| 368 | + | select { | |||
| 369 | + | case <-p.repeat: | |||
| 370 | + | requestProtocol(t, c, id, wayland.KeyboardRequestRelease) | |||
| 371 | + | roundtrip(t, c) | |||
| 372 | + | return | |||
| 373 | + | default: | |||
| 374 | + | } | |||
| 375 | + | if err := c.Dispatch(); err != nil { | |||
| 376 | + | t.Fatal(err) | |||
| 377 | + | } | |||
| 378 | + | } | |||
| 379 | + | } | |||
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/adapters/wayland/server.go +0 -3
| @@ -162,7 +162,6 @@ type Server struct { | |||
| 162 | 162 | securityLive atomic.Int32 | |
| 163 | 163 | display *server.Display | |
| 164 | 164 | env procEnv | |
| 165 | - | slotsPending bool // core waits for a slot window | |
| 166 | 165 | name string | |
| 167 | 166 | cleanup func() | |
| 168 | 167 | log zerowrap.Logger | |
| @@ -809,8 +808,6 @@ func (s *Server) apply(cmd ports.ClientCommand) { | |||
| 809 | 808 | s.captureState(c) | |
| 810 | 809 | case ports.UserActivity: | |
| 811 | 810 | s.userActivity() | |
| 812 | - | case ports.SlotsPending: | |
| 813 | - | s.slotsPending = c.Pending | |
| 814 | 811 | case ports.ShortcutsInhibitState: | |
| 815 | 812 | s.setShortcutsInhibit(c) | |
| 816 | 813 | case ports.CloseWindow: | |
| @@ -162,7 +162,6 @@ type Server struct { | |||||
| 162 | securityLive atomic.Int32 | 162 | securityLive atomic.Int32 | ||
| 163 | display *server.Display | 163 | display *server.Display | ||
| 164 | env procEnv | 164 | env procEnv | ||
| 165 | - | slotsPending bool // core waits for a slot window | |||
| 166 | name string | 165 | name string | ||
| 167 | cleanup func() | 166 | cleanup func() | ||
| 168 | log zerowrap.Logger | 167 | log zerowrap.Logger | ||
| @@ -809,8 +808,6 @@ func (s *Server) apply(cmd ports.ClientCommand) { | |||||
| 809 | s.captureState(c) | 808 | s.captureState(c) | ||
| 810 | case ports.UserActivity: | 809 | case ports.UserActivity: | ||
| 811 | s.userActivity() | 810 | s.userActivity() | ||
| 812 | - | case ports.SlotsPending: | |||
| 813 | - | s.slotsPending = c.Pending | |||
| 814 | case ports.ShortcutsInhibitState: | 811 | case ports.ShortcutsInhibitState: | ||
| 815 | s.setShortcutsInhibit(c) | 812 | s.setShortcutsInhibit(c) | ||
| 816 | case ports.CloseWindow: | 813 | case ports.CloseWindow: | ||
M internal/adapters/wayland/slot.go +4 -5
| @@ -37,12 +37,11 @@ func (linuxProcEnv) Lookup(pid int, key string) (string, bool) { | |||
| 37 | 37 | } | |
| 38 | 38 | ||
| 39 | 39 | // slotToken returns the SlotEnv value of the client process, if any. It | |
| 40 | - | // reads /proc only while core waits for a slot window. There is no per-client | |
| 41 | - | // cache: libwayland may reuse a client address after a disconnect. | |
| 40 | + | // reads /proc once at toplevel mapping, independently of command application: | |
| 41 | + | // a spawned client can map before the display drains core's commands. Core | |
| 42 | + | // accepts only tokens for pending placements. There is no per-client cache: | |
| 43 | + | // libwayland may reuse a client address after a disconnect. | |
| 42 | 44 | func (s *Server) slotToken(c server.Client) string { | |
| 43 | - | if !s.slotsPending { | |
| 44 | - | return "" | |
| 45 | - | } | |
| 46 | 45 | v, _ := s.env.Lookup(c.PID(), ports.SlotEnv) | |
| 47 | 46 | return v | |
| 48 | 47 | } | |
| @@ -37,12 +37,11 @@ func (linuxProcEnv) Lookup(pid int, key string) (string, bool) { | |||||
| 37 | } | 37 | } | ||
| 38 | 38 | ||||
| 39 | // slotToken returns the SlotEnv value of the client process, if any. It | 39 | // slotToken returns the SlotEnv value of the client process, if any. It | ||
| 40 | - | // reads /proc only while core waits for a slot window. There is no per-client | 40 | + | // reads /proc once at toplevel mapping, independently of command application: |
| 41 | - | // cache: libwayland may reuse a client address after a disconnect. | 41 | + | // a spawned client can map before the display drains core's commands. Core |
| 42 | + | // accepts only tokens for pending placements. There is no per-client cache: | |||
| 43 | + | // libwayland may reuse a client address after a disconnect. | |||
| 42 | func (s *Server) slotToken(c server.Client) string { | 44 | func (s *Server) slotToken(c server.Client) string { | ||
| 43 | - | if !s.slotsPending { | |||
| 44 | - | return "" | |||
| 45 | - | } | |||
| 46 | v, _ := s.env.Lookup(c.PID(), ports.SlotEnv) | 45 | v, _ := s.env.Lookup(c.PID(), ports.SlotEnv) | ||
| 47 | return v | 46 | return v | ||
| 48 | } | 47 | } | ||
M internal/adapters/wayland/slot_test.go +3 -10
| @@ -14,18 +14,11 @@ func TestWindowMappedCarriesSlot(t *testing.T) { | |||
| 14 | 14 | env.EXPECT().Lookup(os.Getpid(), ports.SlotEnv).Return("7", true).Once() | |
| 15 | 15 | s, events, _, dir := keyboardServer(t, func(s *Server) { s.env = env }) | |
| 16 | 16 | mapWindow := toplevelMapper(t, protocolClient(t, s, dir), events) | |
| 17 | - | // No slot pending: /proc is not read (the mock would fail on a call). | |
| 18 | - | if w := mapWindow(); w.Slot != "" || w.PID != os.Getpid() { | |
| 17 | + | // No command needs to reach the display before this fast client maps. | |
| 18 | + | if w := mapWindow(); w.Slot != "7" || w.PID != os.Getpid() { | |
| 19 | 19 | t.Fatalf("slot %q pid %d", w.Slot, w.PID) | |
| 20 | 20 | } | |
| 21 | - | // This tests environment lookup, not command transport. Apply the gate | |
| 22 | - | // synchronously before mapping over the independent protocol connection. | |
| 23 | - | if !s.display.Do(func() { s.apply(ports.SlotsPending{Pending: true}) }) { | |
| 24 | - | t.Fatal("display stopped before applying SlotsPending") | |
| 25 | - | } | |
| 26 | - | if w := mapWindow(); w.Slot != "7" { | |
| 27 | - | t.Fatalf("slot %q", w.Slot) | |
| 28 | - | } | |
| 21 | + | ||
| 29 | 22 | } | |
| 30 | 23 | ||
| 31 | 24 | func TestLinuxProcEnv(t *testing.T) { | |
| @@ -14,18 +14,11 @@ func TestWindowMappedCarriesSlot(t *testing.T) { | |||||
| 14 | env.EXPECT().Lookup(os.Getpid(), ports.SlotEnv).Return("7", true).Once() | 14 | env.EXPECT().Lookup(os.Getpid(), ports.SlotEnv).Return("7", true).Once() | ||
| 15 | s, events, _, dir := keyboardServer(t, func(s *Server) { s.env = env }) | 15 | s, events, _, dir := keyboardServer(t, func(s *Server) { s.env = env }) | ||
| 16 | mapWindow := toplevelMapper(t, protocolClient(t, s, dir), events) | 16 | mapWindow := toplevelMapper(t, protocolClient(t, s, dir), events) | ||
| 17 | - | // No slot pending: /proc is not read (the mock would fail on a call). | 17 | + | // No command needs to reach the display before this fast client maps. |
| 18 | - | if w := mapWindow(); w.Slot != "" || w.PID != os.Getpid() { | 18 | + | if w := mapWindow(); w.Slot != "7" || w.PID != os.Getpid() { |
| 19 | t.Fatalf("slot %q pid %d", w.Slot, w.PID) | 19 | t.Fatalf("slot %q pid %d", w.Slot, w.PID) | ||
| 20 | } | 20 | } | ||
| 21 | - | // This tests environment lookup, not command transport. Apply the gate | 21 | + | |
| 22 | - | // synchronously before mapping over the independent protocol connection. | |||
| 23 | - | if !s.display.Do(func() { s.apply(ports.SlotsPending{Pending: true}) }) { | |||
| 24 | - | t.Fatal("display stopped before applying SlotsPending") | |||
| 25 | - | } | |||
| 26 | - | if w := mapWindow(); w.Slot != "7" { | |||
| 27 | - | t.Fatalf("slot %q", w.Slot) | |||
| 28 | - | } | |||
| 29 | } | 22 | } | ||
| 30 | 23 | ||||
| 31 | func TestLinuxProcEnv(t *testing.T) { | 24 | func TestLinuxProcEnv(t *testing.T) { | ||
M internal/adapters/wayland/xdg.go +6 -1
| @@ -239,7 +239,12 @@ func (x *xdgSurface) GetToplevel(r *xdgshell.Surface, id uint32) { | |||
| 239 | 239 | } else if buffer && !w.mapped && x.acked { | |
| 240 | 240 | w.mapped = true | |
| 241 | 241 | w.gen++ | |
| 242 | - | slot := x.server.slotToken(r.Client()) | |
| 242 | + | // Placement claims are for a new toplevel, not later remaps. | |
| 243 | + | // Ordinary commits, popups and remaps need no /proc lookup. | |
| 244 | + | var slot string | |
| 245 | + | if w.gen == 1 { | |
| 246 | + | slot = x.server.slotToken(r.Client()) | |
| 247 | + | } | |
| 243 | 248 | // Its size comes with the commit, in WindowResized (afterCommit). | |
| 244 | 249 | w.floating = w.floats() | |
| 245 | 250 | var parent ports.WindowID | |
| @@ -239,7 +239,12 @@ func (x *xdgSurface) GetToplevel(r *xdgshell.Surface, id uint32) { | |||||
| 239 | } else if buffer && !w.mapped && x.acked { | 239 | } else if buffer && !w.mapped && x.acked { | ||
| 240 | w.mapped = true | 240 | w.mapped = true | ||
| 241 | w.gen++ | 241 | w.gen++ | ||
| 242 | - | slot := x.server.slotToken(r.Client()) | 242 | + | // Placement claims are for a new toplevel, not later remaps. |
| 243 | + | // Ordinary commits, popups and remaps need no /proc lookup. | |||
| 244 | + | var slot string | |||
| 245 | + | if w.gen == 1 { | |||
| 246 | + | slot = x.server.slotToken(r.Client()) | |||
| 247 | + | } | |||
| 243 | // Its size comes with the commit, in WindowResized (afterCommit). | 248 | // Its size comes with the commit, in WindowResized (afterCommit). | ||
| 244 | w.floating = w.floats() | 249 | w.floating = w.floats() | ||
| 245 | var parent ports.WindowID | 250 | var parent ports.WindowID | ||
M internal/core/core.go +6 -51
| @@ -204,7 +204,6 @@ type sentCache struct { | |||
| 204 | 204 | outputs ports.SetOutputs | |
| 205 | 205 | state ports.State | |
| 206 | 206 | workspaces ports.Workspaces | |
| 207 | - | pending bool // spawn placement pending | |
| 208 | 207 | } | |
| 209 | 208 | ||
| 210 | 209 | func keyName(s string) string { | |
| @@ -364,23 +363,6 @@ func (c *Core) apply(cfg ports.Config) error { | |||
| 364 | 363 | return nil | |
| 365 | 364 | } | |
| 366 | 365 | ||
| 367 | - | // publishPending tells wayland at once when slots start waiting, before | |
| 368 | - | // their windows can map. | |
| 369 | - | func (c *Core) publishPending(ctx context.Context) error { | |
| 370 | - | if p := c.placement.anyPending(); p != c.sent.pending { | |
| 371 | - | // Pending must be in wayland's queue before the spawn that follows. | |
| 372 | - | send := c.command | |
| 373 | - | if p { | |
| 374 | - | send = c.commandFlushed | |
| 375 | - | } | |
| 376 | - | if err := send(ctx, ports.SlotsPending{Pending: p}); err != nil { | |
| 377 | - | return err | |
| 378 | - | } | |
| 379 | - | c.sent.pending = p | |
| 380 | - | } | |
| 381 | - | return nil | |
| 382 | - | } | |
| 383 | - | ||
| 384 | 366 | // spawnSlots starts the slots queued by config. With shown set (the user | |
| 385 | 367 | // just switched to another workspace), it also refills the empty slots of | |
| 386 | 368 | // the workspace now on screen. | |
| @@ -406,13 +388,11 @@ func (c *Core) spawnSlots(ctx context.Context, shown bool) error { | |||
| 406 | 388 | } | |
| 407 | 389 | started[key] = true | |
| 408 | 390 | req := c.spawnSlot(key) | |
| 409 | - | // Wayland must read slot tokens before this window can map. | |
| 410 | - | pendingErr := c.publishPending(ctx) | |
| 411 | - | if pendingErr != nil || c.protectionRequested() { | |
| 391 | + | if err := ctx.Err(); err != nil || c.protectionRequested() { | |
| 412 | 392 | c.placement.dropSlot(key) | |
| 413 | 393 | st.stale = false | |
| 414 | 394 | c.toSpawn = append(c.toSpawn, keys[i:]...) | |
| 415 | - | return pendingErr | |
| 395 | + | return err | |
| 416 | 396 | } | |
| 417 | 397 | if !c.trySpawn(ctx, req) { | |
| 418 | 398 | c.placement.dropSlot(key) | |
| @@ -473,26 +453,6 @@ func (c *Core) command(ctx context.Context, v ports.ClientCommand) error { | |||
| 473 | 453 | return err | |
| 474 | 454 | } | |
| 475 | 455 | ||
| 476 | - | // commandFlushed is command for a command that another channel depends on: | |
| 477 | - | // it returns once v is in the command channel. SlotsPending must reach | |
| 478 | - | // wayland before the spawn whose window it lets wayland claim; the spawn | |
| 479 | - | // goes on its own channel, so queue order alone does not keep them in order. | |
| 480 | - | func (c *Core) commandFlushed(ctx context.Context, v ports.ClientCommand) error { | |
| 481 | - | seq, err := c.queue(ctx, v) | |
| 482 | - | if err != nil || seq == 0 { | |
| 483 | - | return err | |
| 484 | - | } | |
| 485 | - | if err := c.out.flush(ctx, c.ch.Commands, seq); err != nil { | |
| 486 | - | return err | |
| 487 | - | } | |
| 488 | - | // As a blocked send did: the rest of the event must not go on under a | |
| 489 | - | // new epoch. | |
| 490 | - | if c.inputEpochChanged() { | |
| 491 | - | return errSecurityChanged | |
| 492 | - | } | |
| 493 | - | return nil | |
| 494 | - | } | |
| 495 | - | ||
| 496 | 456 | // queue pushes v to the outbox and returns its sequence number (0: not | |
| 497 | 457 | // queued). | |
| 498 | 458 | func (c *Core) queue(ctx context.Context, v ports.ClientCommand) (uint64, error) { | |
| @@ -689,9 +649,6 @@ func (c *Core) publishFrame(ctx context.Context, only *screen) error { | |||
| 689 | 649 | if err := c.spawnEmpty(ctx); err != nil { | |
| 690 | 650 | return err | |
| 691 | 651 | } | |
| 692 | - | if err := c.publishPending(ctx); err != nil { | |
| 693 | - | return err | |
| 694 | - | } | |
| 695 | 652 | focus := c.keyboardFocus() | |
| 696 | 653 | c.advanceFlash(only) | |
| 697 | 654 | scenes := make([]ports.Scene, 0, len(c.screens)) | |
| @@ -1120,7 +1077,7 @@ func (c *Core) Run(ctx context.Context) error { | |||
| 1120 | 1077 | c.startup[i] = slices.Clone(argv) | |
| 1121 | 1078 | } | |
| 1122 | 1079 | c.spawnStartup(ctx) | |
| 1123 | - | if c.spawnSlots(ctx, false) != nil || c.publishPending(ctx) != nil { | |
| 1080 | + | if c.spawnSlots(ctx, false) != nil { | |
| 1124 | 1081 | return nil | |
| 1125 | 1082 | } | |
| 1126 | 1083 | c.publishState() | |
| @@ -1355,13 +1312,11 @@ func (c *Core) spawnEmpty(ctx context.Context) error { | |||
| 1355 | 1312 | } | |
| 1356 | 1313 | ||
| 1357 | 1314 | // dispatch sends reqs to the launcher and reports how many it issued. | |
| 1358 | - | // Wayland reads claim tokens only while it knows spawns are pending, so the | |
| 1359 | - | // flag is published before the first send. From the first request that is | |
| 1360 | - | // not issued (publication failed, or the gate suppressed the send), the | |
| 1361 | - | // rest are not sent and their claims are cancelled. | |
| 1315 | + | // From the first request that is not issued (context ended, or the gate | |
| 1316 | + | // suppressed the send), the rest are not sent and their claims are cancelled. | |
| 1362 | 1317 | func (c *Core) dispatch(ctx context.Context, reqs []ports.SpawnRequest) (issued int, err error) { | |
| 1363 | - | err = c.publishPending(ctx) | |
| 1364 | 1318 | for i, req := range reqs { | |
| 1319 | + | err = ctx.Err() | |
| 1365 | 1320 | if err != nil || !c.trySpawn(ctx, req) { | |
| 1366 | 1321 | for _, suppressed := range reqs[i:] { | |
| 1367 | 1322 | c.cancelClaim(suppressed) | |
| @@ -204,7 +204,6 @@ type sentCache struct { | |||||
| 204 | outputs ports.SetOutputs | 204 | outputs ports.SetOutputs | ||
| 205 | state ports.State | 205 | state ports.State | ||
| 206 | workspaces ports.Workspaces | 206 | workspaces ports.Workspaces | ||
| 207 | - | pending bool // spawn placement pending | |||
| 208 | } | 207 | } | ||
| 209 | 208 | ||||
| 210 | func keyName(s string) string { | 209 | func keyName(s string) string { | ||
| @@ -364,23 +363,6 @@ func (c *Core) apply(cfg ports.Config) error { | |||||
| 364 | return nil | 363 | return nil | ||
| 365 | } | 364 | } | ||
| 366 | 365 | ||||
| 367 | - | // publishPending tells wayland at once when slots start waiting, before | |||
| 368 | - | // their windows can map. | |||
| 369 | - | func (c *Core) publishPending(ctx context.Context) error { | |||
| 370 | - | if p := c.placement.anyPending(); p != c.sent.pending { | |||
| 371 | - | // Pending must be in wayland's queue before the spawn that follows. | |||
| 372 | - | send := c.command | |||
| 373 | - | if p { | |||
| 374 | - | send = c.commandFlushed | |||
| 375 | - | } | |||
| 376 | - | if err := send(ctx, ports.SlotsPending{Pending: p}); err != nil { | |||
| 377 | - | return err | |||
| 378 | - | } | |||
| 379 | - | c.sent.pending = p | |||
| 380 | - | } | |||
| 381 | - | return nil | |||
| 382 | - | } | |||
| 383 | - | ||||
| 384 | // spawnSlots starts the slots queued by config. With shown set (the user | 366 | // spawnSlots starts the slots queued by config. With shown set (the user | ||
| 385 | // just switched to another workspace), it also refills the empty slots of | 367 | // just switched to another workspace), it also refills the empty slots of | ||
| 386 | // the workspace now on screen. | 368 | // the workspace now on screen. | ||
| @@ -406,13 +388,11 @@ func (c *Core) spawnSlots(ctx context.Context, shown bool) error { | |||||
| 406 | } | 388 | } | ||
| 407 | started[key] = true | 389 | started[key] = true | ||
| 408 | req := c.spawnSlot(key) | 390 | req := c.spawnSlot(key) | ||
| 409 | - | // Wayland must read slot tokens before this window can map. | 391 | + | if err := ctx.Err(); err != nil || c.protectionRequested() { |
| 410 | - | pendingErr := c.publishPending(ctx) | |||
| 411 | - | if pendingErr != nil || c.protectionRequested() { | |||
| 412 | c.placement.dropSlot(key) | 392 | c.placement.dropSlot(key) | ||
| 413 | st.stale = false | 393 | st.stale = false | ||
| 414 | c.toSpawn = append(c.toSpawn, keys[i:]...) | 394 | c.toSpawn = append(c.toSpawn, keys[i:]...) | ||
| 415 | - | return pendingErr | 395 | + | return err |
| 416 | } | 396 | } | ||
| 417 | if !c.trySpawn(ctx, req) { | 397 | if !c.trySpawn(ctx, req) { | ||
| 418 | c.placement.dropSlot(key) | 398 | c.placement.dropSlot(key) | ||
| @@ -473,26 +453,6 @@ func (c *Core) command(ctx context.Context, v ports.ClientCommand) error { | |||||
| 473 | return err | 453 | return err | ||
| 474 | } | 454 | } | ||
| 475 | 455 | ||||
| 476 | - | // commandFlushed is command for a command that another channel depends on: | |||
| 477 | - | // it returns once v is in the command channel. SlotsPending must reach | |||
| 478 | - | // wayland before the spawn whose window it lets wayland claim; the spawn | |||
| 479 | - | // goes on its own channel, so queue order alone does not keep them in order. | |||
| 480 | - | func (c *Core) commandFlushed(ctx context.Context, v ports.ClientCommand) error { | |||
| 481 | - | seq, err := c.queue(ctx, v) | |||
| 482 | - | if err != nil || seq == 0 { | |||
| 483 | - | return err | |||
| 484 | - | } | |||
| 485 | - | if err := c.out.flush(ctx, c.ch.Commands, seq); err != nil { | |||
| 486 | - | return err | |||
| 487 | - | } | |||
| 488 | - | // As a blocked send did: the rest of the event must not go on under a | |||
| 489 | - | // new epoch. | |||
| 490 | - | if c.inputEpochChanged() { | |||
| 491 | - | return errSecurityChanged | |||
| 492 | - | } | |||
| 493 | - | return nil | |||
| 494 | - | } | |||
| 495 | - | ||||
| 496 | // queue pushes v to the outbox and returns its sequence number (0: not | 456 | // queue pushes v to the outbox and returns its sequence number (0: not | ||
| 497 | // queued). | 457 | // queued). | ||
| 498 | func (c *Core) queue(ctx context.Context, v ports.ClientCommand) (uint64, error) { | 458 | func (c *Core) queue(ctx context.Context, v ports.ClientCommand) (uint64, error) { | ||
| @@ -689,9 +649,6 @@ func (c *Core) publishFrame(ctx context.Context, only *screen) error { | |||||
| 689 | if err := c.spawnEmpty(ctx); err != nil { | 649 | if err := c.spawnEmpty(ctx); err != nil { | ||
| 690 | return err | 650 | return err | ||
| 691 | } | 651 | } | ||
| 692 | - | if err := c.publishPending(ctx); err != nil { | |||
| 693 | - | return err | |||
| 694 | - | } | |||
| 695 | focus := c.keyboardFocus() | 652 | focus := c.keyboardFocus() | ||
| 696 | c.advanceFlash(only) | 653 | c.advanceFlash(only) | ||
| 697 | scenes := make([]ports.Scene, 0, len(c.screens)) | 654 | scenes := make([]ports.Scene, 0, len(c.screens)) | ||
| @@ -1120,7 +1077,7 @@ func (c *Core) Run(ctx context.Context) error { | |||||
| 1120 | c.startup[i] = slices.Clone(argv) | 1077 | c.startup[i] = slices.Clone(argv) | ||
| 1121 | } | 1078 | } | ||
| 1122 | c.spawnStartup(ctx) | 1079 | c.spawnStartup(ctx) | ||
| 1123 | - | if c.spawnSlots(ctx, false) != nil || c.publishPending(ctx) != nil { | 1080 | + | if c.spawnSlots(ctx, false) != nil { |
| 1124 | return nil | 1081 | return nil | ||
| 1125 | } | 1082 | } | ||
| 1126 | c.publishState() | 1083 | c.publishState() | ||
| @@ -1355,13 +1312,11 @@ func (c *Core) spawnEmpty(ctx context.Context) error { | |||||
| 1355 | } | 1312 | } | ||
| 1356 | 1313 | ||||
| 1357 | // dispatch sends reqs to the launcher and reports how many it issued. | 1314 | // dispatch sends reqs to the launcher and reports how many it issued. | ||
| 1358 | - | // Wayland reads claim tokens only while it knows spawns are pending, so the | 1315 | + | // From the first request that is not issued (context ended, or the gate |
| 1359 | - | // flag is published before the first send. From the first request that is | 1316 | + | // suppressed the send), the rest are not sent and their claims are cancelled. |
| 1360 | - | // not issued (publication failed, or the gate suppressed the send), the | |||
| 1361 | - | // rest are not sent and their claims are cancelled. | |||
| 1362 | func (c *Core) dispatch(ctx context.Context, reqs []ports.SpawnRequest) (issued int, err error) { | 1317 | func (c *Core) dispatch(ctx context.Context, reqs []ports.SpawnRequest) (issued int, err error) { | ||
| 1363 | - | err = c.publishPending(ctx) | |||
| 1364 | for i, req := range reqs { | 1318 | for i, req := range reqs { | ||
| 1319 | + | err = ctx.Err() | |||
| 1365 | if err != nil || !c.trySpawn(ctx, req) { | 1320 | if err != nil || !c.trySpawn(ctx, req) { | ||
| 1366 | for _, suppressed := range reqs[i:] { | 1321 | for _, suppressed := range reqs[i:] { | ||
| 1367 | c.cancelClaim(suppressed) | 1322 | c.cancelClaim(suppressed) | ||
M internal/core/outbox.go +3 -4
| @@ -51,8 +51,7 @@ type CommandStats struct { | |||
| 51 | 51 | // MaxBacklog is the longest queue seen. | |
| 52 | 52 | MaxBacklog uint64 | |
| 53 | 53 | // MaxQueueWait is the longest time a queued command took to reach the | |
| 54 | - | // channel; MaxFlushWait the longest time core waited in flush (a spawn's | |
| 55 | - | // SlotsPending, or a full outbox). | |
| 54 | + | // channel; MaxFlushWait the longest time core waited for outbox room. | |
| 56 | 55 | MaxQueueWait, MaxFlushWait time.Duration | |
| 57 | 56 | } | |
| 58 | 57 | ||
| @@ -61,6 +60,7 @@ func (o *outbox) takeStats() CommandStats { | |||
| 61 | 60 | o.mu.Lock() | |
| 62 | 61 | defer o.mu.Unlock() | |
| 63 | 62 | s := o.stats | |
| 63 | + | s.MaxBacklog = max(s.MaxBacklog, o.pushed-o.sent) | |
| 64 | 64 | o.stats = CommandStats{} | |
| 65 | 65 | return s | |
| 66 | 66 | } | |
| @@ -171,8 +171,7 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta | |||
| 171 | 171 | } | |
| 172 | 172 | ||
| 173 | 173 | // flush waits until the entry seq is in out, or ctx ends. It is for the | |
| 174 | - | // rare command another channel depends on (SlotsPending before a spawn); | |
| 175 | - | // everything else never waits. Without a running sender (core driven | |
| 174 | + | // bounded outbox backpressure. Without a running sender (core driven | |
| 176 | 175 | // without Run), it delivers the queue itself, in order. | |
| 177 | 176 | func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error { | |
| 178 | 177 | var start time.Time | |
| @@ -51,8 +51,7 @@ type CommandStats struct { | |||||
| 51 | // MaxBacklog is the longest queue seen. | 51 | // MaxBacklog is the longest queue seen. | ||
| 52 | MaxBacklog uint64 | 52 | MaxBacklog uint64 | ||
| 53 | // MaxQueueWait is the longest time a queued command took to reach the | 53 | // MaxQueueWait is the longest time a queued command took to reach the | ||
| 54 | - | // channel; MaxFlushWait the longest time core waited in flush (a spawn's | 54 | + | // channel; MaxFlushWait the longest time core waited for outbox room. |
| 55 | - | // SlotsPending, or a full outbox). | |||
| 56 | MaxQueueWait, MaxFlushWait time.Duration | 55 | MaxQueueWait, MaxFlushWait time.Duration | ||
| 57 | } | 56 | } | ||
| 58 | 57 | ||||
| @@ -61,6 +60,7 @@ func (o *outbox) takeStats() CommandStats { | |||||
| 61 | o.mu.Lock() | 60 | o.mu.Lock() | ||
| 62 | defer o.mu.Unlock() | 61 | defer o.mu.Unlock() | ||
| 63 | s := o.stats | 62 | s := o.stats | ||
| 63 | + | s.MaxBacklog = max(s.MaxBacklog, o.pushed-o.sent) | |||
| 64 | o.stats = CommandStats{} | 64 | o.stats = CommandStats{} | ||
| 65 | return s | 65 | return s | ||
| 66 | } | 66 | } | ||
| @@ -171,8 +171,7 @@ func (o *outbox) push(out chan<- ports.ClientCommand, v ports.ClientCommand, sta | |||||
| 171 | } | 171 | } | ||
| 172 | 172 | ||||
| 173 | // flush waits until the entry seq is in out, or ctx ends. It is for the | 173 | // flush waits until the entry seq is in out, or ctx ends. It is for the | ||
| 174 | - | // rare command another channel depends on (SlotsPending before a spawn); | 174 | + | // bounded outbox backpressure. Without a running sender (core driven |
| 175 | - | // everything else never waits. Without a running sender (core driven | |||
| 176 | // without Run), it delivers the queue itself, in order. | 175 | // without Run), it delivers the queue itself, in order. | ||
| 177 | func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error { | 176 | func (o *outbox) flush(ctx context.Context, out chan<- ports.ClientCommand, seq uint64) error { | ||
| 178 | var start time.Time | 177 | var start time.Time | ||
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 | + | } | |||
M internal/core/screens_test.go +0 -3
| @@ -647,9 +647,6 @@ func TestFirstTerminalFollowsSetOutputs(t *testing.T) { | |||
| 647 | 647 | case <-time.After(time.Second): | |
| 648 | 648 | t.Fatal("no output command or spawn") | |
| 649 | 649 | } | |
| 650 | - | if v, ok := receive(t, commands).(ports.SlotsPending); !ok || !v.Pending { | |
| 651 | - | t.Fatalf("expected pending before spawn, got %v", v) | |
| 652 | - | } | |
| 653 | 650 | if req := receive(t, spawn); len(req.Argv) == 0 { | |
| 654 | 651 | t.Fatal(req) | |
| 655 | 652 | } | |
| @@ -647,9 +647,6 @@ func TestFirstTerminalFollowsSetOutputs(t *testing.T) { | |||||
| 647 | case <-time.After(time.Second): | 647 | case <-time.After(time.Second): | ||
| 648 | t.Fatal("no output command or spawn") | 648 | t.Fatal("no output command or spawn") | ||
| 649 | } | 649 | } | ||
| 650 | - | if v, ok := receive(t, commands).(ports.SlotsPending); !ok || !v.Pending { | |||
| 651 | - | t.Fatalf("expected pending before spawn, got %v", v) | |||
| 652 | - | } | |||
| 653 | if req := receive(t, spawn); len(req.Argv) == 0 { | 650 | if req := receive(t, spawn); len(req.Argv) == 0 { | ||
| 654 | t.Fatal(req) | 651 | t.Fatal(req) | ||
| 655 | } | 652 | } | ||
M internal/core/security_review_test.go +2 -7
| @@ -77,7 +77,7 @@ func TestExactEpochAdmissionPreservesProducerModifierState(t *testing.T) { | |||
| 77 | 77 | } | |
| 78 | 78 | } | |
| 79 | 79 | ||
| 80 | - | func TestSlotsPendingBackpressureSuppressesSpawnAndDefersSlots(t *testing.T) { | |
| 80 | + | func TestEpochChangeSuppressesSpawnAndDefersSlots(t *testing.T) { | |
| 81 | 81 | c, _, _ := hiddenCaptureCommands(t) | |
| 82 | 82 | key := slotKey{workspace: "dev", index: 1} | |
| 83 | 83 | c.slots[key] = &slotState{argv: []string{"terminal"}} | |
| @@ -100,10 +100,6 @@ func TestSlotsPendingBackpressureSuppressesSpawnAndDefersSlots(t *testing.T) { | |||
| 100 | 100 | spawn := make(chan ports.SpawnRequest, 2) | |
| 101 | 101 | c.ch.Spawn = spawn | |
| 102 | 102 | err := c.spawnSlots(context.Background(), false) | |
| 103 | - | cmd := reviewReceive(t, commands).(ports.SecurityCommand) | |
| 104 | - | if _, ok := cmd.Command.(ports.SlotsPending); !ok { | |
| 105 | - | t.Fatal(cmd) | |
| 106 | - | } | |
| 107 | 103 | if err != nil { | |
| 108 | 104 | t.Fatal(err) | |
| 109 | 105 | } | |
| @@ -122,7 +118,7 @@ func TestSlotsPendingBackpressureSuppressesSpawnAndDefersSlots(t *testing.T) { | |||
| 122 | 118 | } | |
| 123 | 119 | } | |
| 124 | 120 | ||
| 125 | - | func TestTerminalPendingBackpressureSuppressesSpawnAndRestoresRetry(t *testing.T) { | |
| 121 | + | func TestTerminalEpochChangeSuppressesSpawnAndRestoresRetry(t *testing.T) { | |
| 126 | 122 | c, _, _ := hiddenCaptureCommands(t) | |
| 127 | 123 | c.cur().mon.RemoveWindow(1) | |
| 128 | 124 | c.cur().mon.RemoveWindow(2) | |
| @@ -147,7 +143,6 @@ func TestTerminalPendingBackpressureSuppressesSpawnAndRestoresRetry(t *testing.T | |||
| 147 | 143 | spawn := make(chan ports.SpawnRequest, 1) | |
| 148 | 144 | c.ch.Spawn = spawn | |
| 149 | 145 | err := c.spawnEmpty(context.Background()) | |
| 150 | - | reviewReceive(t, commands) | |
| 151 | 146 | if err != nil { | |
| 152 | 147 | t.Fatal(err) | |
| 153 | 148 | } | |
| @@ -77,7 +77,7 @@ func TestExactEpochAdmissionPreservesProducerModifierState(t *testing.T) { | |||||
| 77 | } | 77 | } | ||
| 78 | } | 78 | } | ||
| 79 | 79 | ||||
| 80 | - | func TestSlotsPendingBackpressureSuppressesSpawnAndDefersSlots(t *testing.T) { | 80 | + | func TestEpochChangeSuppressesSpawnAndDefersSlots(t *testing.T) { |
| 81 | c, _, _ := hiddenCaptureCommands(t) | 81 | c, _, _ := hiddenCaptureCommands(t) | ||
| 82 | key := slotKey{workspace: "dev", index: 1} | 82 | key := slotKey{workspace: "dev", index: 1} | ||
| 83 | c.slots[key] = &slotState{argv: []string{"terminal"}} | 83 | c.slots[key] = &slotState{argv: []string{"terminal"}} | ||
| @@ -100,10 +100,6 @@ func TestSlotsPendingBackpressureSuppressesSpawnAndDefersSlots(t *testing.T) { | |||||
| 100 | spawn := make(chan ports.SpawnRequest, 2) | 100 | spawn := make(chan ports.SpawnRequest, 2) | ||
| 101 | c.ch.Spawn = spawn | 101 | c.ch.Spawn = spawn | ||
| 102 | err := c.spawnSlots(context.Background(), false) | 102 | err := c.spawnSlots(context.Background(), false) | ||
| 103 | - | cmd := reviewReceive(t, commands).(ports.SecurityCommand) | |||
| 104 | - | if _, ok := cmd.Command.(ports.SlotsPending); !ok { | |||
| 105 | - | t.Fatal(cmd) | |||
| 106 | - | } | |||
| 107 | if err != nil { | 103 | if err != nil { | ||
| 108 | t.Fatal(err) | 104 | t.Fatal(err) | ||
| 109 | } | 105 | } | ||
| @@ -122,7 +118,7 @@ func TestSlotsPendingBackpressureSuppressesSpawnAndDefersSlots(t *testing.T) { | |||||
| 122 | } | 118 | } | ||
| 123 | } | 119 | } | ||
| 124 | 120 | ||||
| 125 | - | func TestTerminalPendingBackpressureSuppressesSpawnAndRestoresRetry(t *testing.T) { | 121 | + | func TestTerminalEpochChangeSuppressesSpawnAndRestoresRetry(t *testing.T) { |
| 126 | c, _, _ := hiddenCaptureCommands(t) | 122 | c, _, _ := hiddenCaptureCommands(t) | ||
| 127 | c.cur().mon.RemoveWindow(1) | 123 | c.cur().mon.RemoveWindow(1) | ||
| 128 | c.cur().mon.RemoveWindow(2) | 124 | c.cur().mon.RemoveWindow(2) | ||
| @@ -147,7 +143,6 @@ func TestTerminalPendingBackpressureSuppressesSpawnAndRestoresRetry(t *testing.T | |||||
| 147 | spawn := make(chan ports.SpawnRequest, 1) | 143 | spawn := make(chan ports.SpawnRequest, 1) | ||
| 148 | c.ch.Spawn = spawn | 144 | c.ch.Spawn = spawn | ||
| 149 | err := c.spawnEmpty(context.Background()) | 145 | err := c.spawnEmpty(context.Background()) | ||
| 150 | - | reviewReceive(t, commands) | |||
| 151 | if err != nil { | 146 | if err != nil { | ||
| 152 | t.Fatal(err) | 147 | t.Fatal(err) | ||
| 153 | } | 148 | } | ||
M internal/core/spawn_placement_test.go +1 -10
| @@ -1,7 +1,6 @@ | |||
| 1 | 1 | package core | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | - | "context" | |
| 5 | 4 | "strings" | |
| 6 | 5 | "testing" | |
| 7 | 6 | "time" | |
| @@ -32,15 +31,7 @@ func TestSlotClaimWithoutWorkspace(t *testing.T) { | |||
| 32 | 31 | if s, _ := c.screenOf(21); s == nil { | |
| 33 | 32 | t.Fatal("window did not map normally") | |
| 34 | 33 | } | |
| 35 | - | // Claim status is also reflected on the client command port. | |
| 36 | - | commands := make(chan ports.ClientCommand, 1) | |
| 37 | - | c.ch.Commands = commands | |
| 38 | - | if err := c.publishPending(context.Background()); err != nil { | |
| 39 | - | t.Fatal(err) | |
| 40 | - | } | |
| 41 | - | if got := (<-commands).(ports.SlotsPending); !got.Pending { | |
| 42 | - | t.Fatal("pending claim was not retained") | |
| 43 | - | } | |
| 34 | + | ||
| 44 | 35 | } | |
| 45 | 36 | ||
| 46 | 37 | func TestExpiredTerminalTokenMapsNormally(t *testing.T) { | |
| @@ -1,7 +1,6 @@ | |||||
| 1 | package core | 1 | package core | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | - | "context" | |||
| 5 | "strings" | 4 | "strings" | ||
| 6 | "testing" | 5 | "testing" | ||
| 7 | "time" | 6 | "time" | ||
| @@ -32,15 +31,7 @@ func TestSlotClaimWithoutWorkspace(t *testing.T) { | |||||
| 32 | if s, _ := c.screenOf(21); s == nil { | 31 | if s, _ := c.screenOf(21); s == nil { | ||
| 33 | t.Fatal("window did not map normally") | 32 | t.Fatal("window did not map normally") | ||
| 34 | } | 33 | } | ||
| 35 | - | // Claim status is also reflected on the client command port. | 34 | + | |
| 36 | - | commands := make(chan ports.ClientCommand, 1) | |||
| 37 | - | c.ch.Commands = commands | |||
| 38 | - | if err := c.publishPending(context.Background()); err != nil { | |||
| 39 | - | t.Fatal(err) | |||
| 40 | - | } | |||
| 41 | - | if got := (<-commands).(ports.SlotsPending); !got.Pending { | |||
| 42 | - | t.Fatal("pending claim was not retained") | |||
| 43 | - | } | |||
| 44 | } | 35 | } | ||
| 45 | 36 | ||||
| 46 | func TestExpiredTerminalTokenMapsNormally(t *testing.T) { | 37 | func TestExpiredTerminalTokenMapsNormally(t *testing.T) { | ||
M internal/core/stash_spawn_internal_test.go +17 -58
| @@ -5,7 +5,6 @@ import ( | |||
| 5 | 5 | "errors" | |
| 6 | 6 | "strings" | |
| 7 | 7 | "testing" | |
| 8 | - | "testing/synctest" | |
| 9 | 8 | "time" | |
| 10 | 9 | ||
| 11 | 10 | portsmocks "github.com/bnema/neferwl/internal/mocks/ports" | |
| @@ -40,46 +39,23 @@ func claimToken(req ports.SpawnRequest) string { | |||
| 40 | 39 | return strings.TrimPrefix(req.Env[0], ports.SlotEnv+"=") | |
| 41 | 40 | } | |
| 42 | 41 | ||
| 43 | - | // Wayland reads tokens only while slots are pending, so the flag is in the | |
| 44 | - | // command channel before the spawn can reach the launcher, also when the | |
| 45 | - | // display reads nothing: in a synctest bubble the sender is waited on until | |
| 46 | - | // it blocks on the unread channel, and the spawn channel must still be empty. | |
| 47 | - | func TestSpawnStashPublishesPendingBeforeSpawn(t *testing.T) { | |
| 48 | - | synctest.Test(t, func(t *testing.T) { | |
| 49 | - | now := claimBase | |
| 50 | - | commands := make(chan ports.ClientCommand) | |
| 51 | - | spawn := make(chan ports.SpawnRequest, 1) | |
| 52 | - | c := stashSpawnCore(t, commands, spawn, fixedClock(t, &now)) | |
| 53 | - | ctx, cancel := context.WithCancel(context.Background()) | |
| 54 | - | t.Cleanup(cancel) | |
| 55 | - | c.out.start() | |
| 56 | - | go c.out.run(ctx, commands) | |
| 57 | - | done := make(chan error, 1) | |
| 58 | - | go func() { done <- c.spawnStash(ctx, c.cur().mon.Current(), [][]string{{"firefox"}}, false) }() | |
| 59 | - | synctest.Wait() | |
| 60 | - | if len(spawn) != 0 { | |
| 61 | - | t.Fatal("spawned before the pending flag was delivered") | |
| 62 | - | } | |
| 63 | - | select { | |
| 64 | - | case err := <-done: | |
| 65 | - | t.Fatalf("spawnStash returned before the flag was read: %v", err) | |
| 66 | - | default: | |
| 67 | - | } | |
| 68 | - | if p, ok := (<-commands).(ports.SlotsPending); !ok || !p.Pending { | |
| 69 | - | t.Fatal("first command is not SlotsPending{true}") | |
| 70 | - | } | |
| 71 | - | synctest.Wait() | |
| 72 | - | if err := <-done; err != nil { | |
| 73 | - | t.Fatal(err) | |
| 74 | - | } | |
| 75 | - | if len(spawn) != 1 { | |
| 76 | - | t.Fatal("no spawn after the flag") | |
| 77 | - | } | |
| 78 | - | }) | |
| 42 | + | // A stalled command transport must not delay a placement spawn. | |
| 43 | + | func TestSpawnStashIndependentOfCommandDrain(t *testing.T) { | |
| 44 | + | now := claimBase | |
| 45 | + | commands := make(chan ports.ClientCommand) | |
| 46 | + | spawn := make(chan ports.SpawnRequest, 1) | |
| 47 | + | c := stashSpawnCore(t, commands, spawn, fixedClock(t, &now)) | |
| 48 | + | if err := c.spawnStash(context.Background(), c.cur().mon.Current(), [][]string{{"firefox"}}, false); err != nil { | |
| 49 | + | t.Fatal(err) | |
| 50 | + | } | |
| 51 | + | req := <-spawn | |
| 52 | + | c.placement.place(c, ports.WindowMapped{ID: 5, Slot: claimToken(req)}) | |
| 53 | + | if w := c.cur().mon.Current(); len(w.Stash) != 1 || w.Stash[0].ID != 5 { | |
| 54 | + | t.Fatal("placement lost while commands unread") | |
| 55 | + | } | |
| 79 | 56 | } | |
| 80 | 57 | ||
| 81 | - | // A send the launcher queue suppresses leaves no claim behind, and the | |
| 82 | - | // pending flag falls again at the next publish. | |
| 58 | + | // A send the launcher queue suppresses leaves no claim behind. | |
| 83 | 59 | func TestSpawnStashSuppressedLeavesNoClaim(t *testing.T) { | |
| 84 | 60 | now := claimBase | |
| 85 | 61 | commands := make(chan ports.ClientCommand, 8) | |
| @@ -90,15 +66,6 @@ func TestSpawnStashSuppressedLeavesNoClaim(t *testing.T) { | |||
| 90 | 66 | if c.placement.anyPending() { | |
| 91 | 67 | t.Fatal("suppressed claims stayed pending") | |
| 92 | 68 | } | |
| 93 | - | if v := (<-commands).(ports.SlotsPending); !v.Pending { | |
| 94 | - | t.Fatal("pending not announced") | |
| 95 | - | } | |
| 96 | - | if err := c.publishPending(context.Background()); err != nil { | |
| 97 | - | t.Fatal(err) | |
| 98 | - | } | |
| 99 | - | if v := (<-commands).(ports.SlotsPending); v.Pending { | |
| 100 | - | t.Fatal("stale pending flag not republished") | |
| 101 | - | } | |
| 102 | 69 | } | |
| 103 | 70 | ||
| 104 | 71 | // From the first command that is not issued, the rest are not sent and their | |
| @@ -112,25 +79,17 @@ func TestSpawnStashCancelsClaimsFromFirstUnissued(t *testing.T) { | |||
| 112 | 79 | wantErr error | |
| 113 | 80 | }{ | |
| 114 | 81 | {"queue of one", 1, false, 1, nil}, | |
| 115 | - | {"publication fails", 4, true, 0, context.Canceled}, | |
| 82 | + | {"context cancelled", 4, true, 0, context.Canceled}, | |
| 116 | 83 | } { | |
| 117 | 84 | t.Run(tc.name, func(t *testing.T) { | |
| 118 | 85 | now := claimBase | |
| 119 | - | commands := make(chan ports.ClientCommand) // unread: a send blocks until the context ends | |
| 86 | + | commands := make(chan ports.ClientCommand) // unread: spawns are independent of command delivery | |
| 120 | 87 | spawn := make(chan ports.SpawnRequest, tc.queue) | |
| 121 | 88 | c := stashSpawnCore(t, commands, spawn, fixedClock(t, &now)) | |
| 122 | 89 | ctx, cancel := context.WithCancel(context.Background()) | |
| 123 | 90 | t.Cleanup(cancel) | |
| 124 | 91 | if tc.cancelled { | |
| 125 | 92 | cancel() | |
| 126 | - | } else { | |
| 127 | - | // An unread unbuffered channel would block the flag: read it. | |
| 128 | - | go func() { | |
| 129 | - | select { | |
| 130 | - | case <-commands: | |
| 131 | - | case <-ctx.Done(): | |
| 132 | - | } | |
| 133 | - | }() | |
| 134 | 93 | } | |
| 135 | 94 | err := c.spawnStash(ctx, c.cur().mon.Current(), [][]string{{"a"}, {"b"}}, false) | |
| 136 | 95 | if !errors.Is(err, tc.wantErr) { | |
| @@ -5,7 +5,6 @@ import ( | |||||
| 5 | "errors" | 5 | "errors" | ||
| 6 | "strings" | 6 | "strings" | ||
| 7 | "testing" | 7 | "testing" | ||
| 8 | - | "testing/synctest" | |||
| 9 | "time" | 8 | "time" | ||
| 10 | 9 | ||||
| 11 | portsmocks "github.com/bnema/neferwl/internal/mocks/ports" | 10 | portsmocks "github.com/bnema/neferwl/internal/mocks/ports" | ||
| @@ -40,46 +39,23 @@ func claimToken(req ports.SpawnRequest) string { | |||||
| 40 | return strings.TrimPrefix(req.Env[0], ports.SlotEnv+"=") | 39 | return strings.TrimPrefix(req.Env[0], ports.SlotEnv+"=") | ||
| 41 | } | 40 | } | ||
| 42 | 41 | ||||
| 43 | - | // Wayland reads tokens only while slots are pending, so the flag is in the | 42 | + | // A stalled command transport must not delay a placement spawn. |
| 44 | - | // command channel before the spawn can reach the launcher, also when the | 43 | + | func TestSpawnStashIndependentOfCommandDrain(t *testing.T) { |
| 45 | - | // display reads nothing: in a synctest bubble the sender is waited on until | 44 | + | now := claimBase |
| 46 | - | // it blocks on the unread channel, and the spawn channel must still be empty. | 45 | + | commands := make(chan ports.ClientCommand) |
| 47 | - | func TestSpawnStashPublishesPendingBeforeSpawn(t *testing.T) { | 46 | + | spawn := make(chan ports.SpawnRequest, 1) |
| 48 | - | synctest.Test(t, func(t *testing.T) { | 47 | + | c := stashSpawnCore(t, commands, spawn, fixedClock(t, &now)) |
| 49 | - | now := claimBase | 48 | + | if err := c.spawnStash(context.Background(), c.cur().mon.Current(), [][]string{{"firefox"}}, false); err != nil { |
| 50 | - | commands := make(chan ports.ClientCommand) | 49 | + | t.Fatal(err) |
| 51 | - | spawn := make(chan ports.SpawnRequest, 1) | 50 | + | } |
| 52 | - | c := stashSpawnCore(t, commands, spawn, fixedClock(t, &now)) | 51 | + | req := <-spawn |
| 53 | - | ctx, cancel := context.WithCancel(context.Background()) | 52 | + | c.placement.place(c, ports.WindowMapped{ID: 5, Slot: claimToken(req)}) |
| 54 | - | t.Cleanup(cancel) | 53 | + | if w := c.cur().mon.Current(); len(w.Stash) != 1 || w.Stash[0].ID != 5 { |
| 55 | - | c.out.start() | 54 | + | t.Fatal("placement lost while commands unread") |
| 56 | - | go c.out.run(ctx, commands) | 55 | + | } |
| 57 | - | done := make(chan error, 1) | |||
| 58 | - | go func() { done <- c.spawnStash(ctx, c.cur().mon.Current(), [][]string{{"firefox"}}, false) }() | |||
| 59 | - | synctest.Wait() | |||
| 60 | - | if len(spawn) != 0 { | |||
| 61 | - | t.Fatal("spawned before the pending flag was delivered") | |||
| 62 | - | } | |||
| 63 | - | select { | |||
| 64 | - | case err := <-done: | |||
| 65 | - | t.Fatalf("spawnStash returned before the flag was read: %v", err) | |||
| 66 | - | default: | |||
| 67 | - | } | |||
| 68 | - | if p, ok := (<-commands).(ports.SlotsPending); !ok || !p.Pending { | |||
| 69 | - | t.Fatal("first command is not SlotsPending{true}") | |||
| 70 | - | } | |||
| 71 | - | synctest.Wait() | |||
| 72 | - | if err := <-done; err != nil { | |||
| 73 | - | t.Fatal(err) | |||
| 74 | - | } | |||
| 75 | - | if len(spawn) != 1 { | |||
| 76 | - | t.Fatal("no spawn after the flag") | |||
| 77 | - | } | |||
| 78 | - | }) | |||
| 79 | } | 56 | } | ||
| 80 | 57 | ||||
| 81 | - | // A send the launcher queue suppresses leaves no claim behind, and the | 58 | + | // A send the launcher queue suppresses leaves no claim behind. |
| 82 | - | // pending flag falls again at the next publish. | |||
| 83 | func TestSpawnStashSuppressedLeavesNoClaim(t *testing.T) { | 59 | func TestSpawnStashSuppressedLeavesNoClaim(t *testing.T) { | ||
| 84 | now := claimBase | 60 | now := claimBase | ||
| 85 | commands := make(chan ports.ClientCommand, 8) | 61 | commands := make(chan ports.ClientCommand, 8) | ||
| @@ -90,15 +66,6 @@ func TestSpawnStashSuppressedLeavesNoClaim(t *testing.T) { | |||||
| 90 | if c.placement.anyPending() { | 66 | if c.placement.anyPending() { | ||
| 91 | t.Fatal("suppressed claims stayed pending") | 67 | t.Fatal("suppressed claims stayed pending") | ||
| 92 | } | 68 | } | ||
| 93 | - | if v := (<-commands).(ports.SlotsPending); !v.Pending { | |||
| 94 | - | t.Fatal("pending not announced") | |||
| 95 | - | } | |||
| 96 | - | if err := c.publishPending(context.Background()); err != nil { | |||
| 97 | - | t.Fatal(err) | |||
| 98 | - | } | |||
| 99 | - | if v := (<-commands).(ports.SlotsPending); v.Pending { | |||
| 100 | - | t.Fatal("stale pending flag not republished") | |||
| 101 | - | } | |||
| 102 | } | 69 | } | ||
| 103 | 70 | ||||
| 104 | // From the first command that is not issued, the rest are not sent and their | 71 | // From the first command that is not issued, the rest are not sent and their | ||
| @@ -112,25 +79,17 @@ func TestSpawnStashCancelsClaimsFromFirstUnissued(t *testing.T) { | |||||
| 112 | wantErr error | 79 | wantErr error | ||
| 113 | }{ | 80 | }{ | ||
| 114 | {"queue of one", 1, false, 1, nil}, | 81 | {"queue of one", 1, false, 1, nil}, | ||
| 115 | - | {"publication fails", 4, true, 0, context.Canceled}, | 82 | + | {"context cancelled", 4, true, 0, context.Canceled}, |
| 116 | } { | 83 | } { | ||
| 117 | t.Run(tc.name, func(t *testing.T) { | 84 | t.Run(tc.name, func(t *testing.T) { | ||
| 118 | now := claimBase | 85 | now := claimBase | ||
| 119 | - | commands := make(chan ports.ClientCommand) // unread: a send blocks until the context ends | 86 | + | commands := make(chan ports.ClientCommand) // unread: spawns are independent of command delivery |
| 120 | spawn := make(chan ports.SpawnRequest, tc.queue) | 87 | spawn := make(chan ports.SpawnRequest, tc.queue) | ||
| 121 | c := stashSpawnCore(t, commands, spawn, fixedClock(t, &now)) | 88 | c := stashSpawnCore(t, commands, spawn, fixedClock(t, &now)) | ||
| 122 | ctx, cancel := context.WithCancel(context.Background()) | 89 | ctx, cancel := context.WithCancel(context.Background()) | ||
| 123 | t.Cleanup(cancel) | 90 | t.Cleanup(cancel) | ||
| 124 | if tc.cancelled { | 91 | if tc.cancelled { | ||
| 125 | cancel() | 92 | cancel() | ||
| 126 | - | } else { | |||
| 127 | - | // An unread unbuffered channel would block the flag: read it. | |||
| 128 | - | go func() { | |||
| 129 | - | select { | |||
| 130 | - | case <-commands: | |||
| 131 | - | case <-ctx.Done(): | |||
| 132 | - | } | |||
| 133 | - | }() | |||
| 134 | } | 93 | } | ||
| 135 | err := c.spawnStash(ctx, c.cur().mon.Current(), [][]string{{"a"}, {"b"}}, false) | 94 | err := c.spawnStash(ctx, c.cur().mon.Current(), [][]string{{"a"}, {"b"}}, false) | ||
| 136 | if !errors.Is(err, tc.wantErr) { | 95 | if !errors.Is(err, tc.wantErr) { | ||
M internal/core/stash_spawn_test.go +0 -22
| @@ -29,16 +29,6 @@ func slotToken(t *testing.T, req ports.SpawnRequest) string { | |||
| 29 | 29 | return req.Env[0][len(ports.SlotEnv)+1:] | |
| 30 | 30 | } | |
| 31 | 31 | ||
| 32 | - | // pendingSent drains the command queue and reports the last SlotsPending. | |
| 33 | - | func pendingSent(r *multiRig) (pending, sent bool) { | |
| 34 | - | for len(r.commands) > 0 { | |
| 35 | - | if v, ok := (<-r.commands).(ports.SlotsPending); ok { | |
| 36 | - | pending, sent = v.Pending, true | |
| 37 | - | } | |
| 38 | - | } | |
| 39 | - | return pending, sent | |
| 40 | - | } | |
| 41 | - | ||
| 42 | 32 | func sceneWindow(set []ports.Scene, id ports.WindowID) (ports.SceneWindow, bool) { | |
| 43 | 33 | for _, s := range set { | |
| 44 | 34 | for _, w := range s.Windows { | |
| @@ -62,9 +52,6 @@ func TestStashBindOpensWindowInStash(t *testing.T) { | |||
| 62 | 52 | t.Fatalf("argv %v", req.Argv) | |
| 63 | 53 | } | |
| 64 | 54 | token := slotToken(t, req) | |
| 65 | - | if pending, sent := pendingSent(r); !sent || !pending { | |
| 66 | - | t.Fatal("SlotsPending{true} was not sent") | |
| 67 | - | } | |
| 68 | 55 | r.key(t, "Next", ports.ModAlt) // focus moves to workspace 2 | |
| 69 | 56 | r.client <- ports.WindowMapped{ID: 2, Slot: token} | |
| 70 | 57 | set := receive(t, r.scenes) | |
| @@ -79,9 +66,6 @@ func TestStashBindOpensWindowInStash(t *testing.T) { | |||
| 79 | 66 | if w1, _ := sceneWindow(set, 1); w1.Focused { | |
| 80 | 67 | t.Fatal("tile kept the focus") | |
| 81 | 68 | } | |
| 82 | - | if pending, sent := pendingSent(r); !sent || pending { | |
| 83 | - | t.Fatalf("pending flag not cleared: %v %v", pending, sent) | |
| 84 | - | } | |
| 85 | 69 | // Unstashing tiles it as a column of its own. | |
| 86 | 70 | set = r.key(t, "s", ports.ModAlt|ports.ModShift) | |
| 87 | 71 | if w, ok := sceneWindow(set, 2); !ok || w.Floating || w.Hidden || !w.Focused { | |
| @@ -112,9 +96,6 @@ func TestToggleEmptyStashOpensTerminal(t *testing.T) { | |||
| 112 | 96 | t.Fatalf("argv %v", req.Argv) | |
| 113 | 97 | } | |
| 114 | 98 | token := slotToken(t, req) | |
| 115 | - | if pending, sent := pendingSent(r); !sent || !pending { | |
| 116 | - | t.Fatal("SlotsPending{true} was not sent") | |
| 117 | - | } | |
| 118 | 99 | r.key(t, "s", ports.ModAlt) | |
| 119 | 100 | if len(r.spawn) != 0 { | |
| 120 | 101 | t.Fatal("second terminal while the first is awaited") | |
| @@ -182,9 +163,6 @@ func TestFourFingerSwipeEmptyStashOpensTerminal(t *testing.T) { | |||
| 182 | 163 | t.Fatalf("argv %v", req.Argv) | |
| 183 | 164 | } | |
| 184 | 165 | token := slotToken(t, req) | |
| 185 | - | if pending, sent := pendingSent(r.multiRig); !sent || !pending { | |
| 186 | - | t.Fatal("SlotsPending{true} was not sent") | |
| 187 | - | } | |
| 188 | 166 | r.four(t, 0, 40, 10) | |
| 189 | 167 | if len(r.spawn) != 0 { | |
| 190 | 168 | t.Fatal("second terminal while the first is awaited") | |
| @@ -29,16 +29,6 @@ func slotToken(t *testing.T, req ports.SpawnRequest) string { | |||||
| 29 | return req.Env[0][len(ports.SlotEnv)+1:] | 29 | return req.Env[0][len(ports.SlotEnv)+1:] | ||
| 30 | } | 30 | } | ||
| 31 | 31 | ||||
| 32 | - | // pendingSent drains the command queue and reports the last SlotsPending. | |||
| 33 | - | func pendingSent(r *multiRig) (pending, sent bool) { | |||
| 34 | - | for len(r.commands) > 0 { | |||
| 35 | - | if v, ok := (<-r.commands).(ports.SlotsPending); ok { | |||
| 36 | - | pending, sent = v.Pending, true | |||
| 37 | - | } | |||
| 38 | - | } | |||
| 39 | - | return pending, sent | |||
| 40 | - | } | |||
| 41 | - | ||||
| 42 | func sceneWindow(set []ports.Scene, id ports.WindowID) (ports.SceneWindow, bool) { | 32 | func sceneWindow(set []ports.Scene, id ports.WindowID) (ports.SceneWindow, bool) { | ||
| 43 | for _, s := range set { | 33 | for _, s := range set { | ||
| 44 | for _, w := range s.Windows { | 34 | for _, w := range s.Windows { | ||
| @@ -62,9 +52,6 @@ func TestStashBindOpensWindowInStash(t *testing.T) { | |||||
| 62 | t.Fatalf("argv %v", req.Argv) | 52 | t.Fatalf("argv %v", req.Argv) | ||
| 63 | } | 53 | } | ||
| 64 | token := slotToken(t, req) | 54 | token := slotToken(t, req) | ||
| 65 | - | if pending, sent := pendingSent(r); !sent || !pending { | |||
| 66 | - | t.Fatal("SlotsPending{true} was not sent") | |||
| 67 | - | } | |||
| 68 | r.key(t, "Next", ports.ModAlt) // focus moves to workspace 2 | 55 | r.key(t, "Next", ports.ModAlt) // focus moves to workspace 2 | ||
| 69 | r.client <- ports.WindowMapped{ID: 2, Slot: token} | 56 | r.client <- ports.WindowMapped{ID: 2, Slot: token} | ||
| 70 | set := receive(t, r.scenes) | 57 | set := receive(t, r.scenes) | ||
| @@ -79,9 +66,6 @@ func TestStashBindOpensWindowInStash(t *testing.T) { | |||||
| 79 | if w1, _ := sceneWindow(set, 1); w1.Focused { | 66 | if w1, _ := sceneWindow(set, 1); w1.Focused { | ||
| 80 | t.Fatal("tile kept the focus") | 67 | t.Fatal("tile kept the focus") | ||
| 81 | } | 68 | } | ||
| 82 | - | if pending, sent := pendingSent(r); !sent || pending { | |||
| 83 | - | t.Fatalf("pending flag not cleared: %v %v", pending, sent) | |||
| 84 | - | } | |||
| 85 | // Unstashing tiles it as a column of its own. | 69 | // Unstashing tiles it as a column of its own. | ||
| 86 | set = r.key(t, "s", ports.ModAlt|ports.ModShift) | 70 | set = r.key(t, "s", ports.ModAlt|ports.ModShift) | ||
| 87 | if w, ok := sceneWindow(set, 2); !ok || w.Floating || w.Hidden || !w.Focused { | 71 | if w, ok := sceneWindow(set, 2); !ok || w.Floating || w.Hidden || !w.Focused { | ||
| @@ -112,9 +96,6 @@ func TestToggleEmptyStashOpensTerminal(t *testing.T) { | |||||
| 112 | t.Fatalf("argv %v", req.Argv) | 96 | t.Fatalf("argv %v", req.Argv) | ||
| 113 | } | 97 | } | ||
| 114 | token := slotToken(t, req) | 98 | token := slotToken(t, req) | ||
| 115 | - | if pending, sent := pendingSent(r); !sent || !pending { | |||
| 116 | - | t.Fatal("SlotsPending{true} was not sent") | |||
| 117 | - | } | |||
| 118 | r.key(t, "s", ports.ModAlt) | 99 | r.key(t, "s", ports.ModAlt) | ||
| 119 | if len(r.spawn) != 0 { | 100 | if len(r.spawn) != 0 { | ||
| 120 | t.Fatal("second terminal while the first is awaited") | 101 | t.Fatal("second terminal while the first is awaited") | ||
| @@ -182,9 +163,6 @@ func TestFourFingerSwipeEmptyStashOpensTerminal(t *testing.T) { | |||||
| 182 | t.Fatalf("argv %v", req.Argv) | 163 | t.Fatalf("argv %v", req.Argv) | ||
| 183 | } | 164 | } | ||
| 184 | token := slotToken(t, req) | 165 | token := slotToken(t, req) | ||
| 185 | - | if pending, sent := pendingSent(r.multiRig); !sent || !pending { | |||
| 186 | - | t.Fatal("SlotsPending{true} was not sent") | |||
| 187 | - | } | |||
| 188 | r.four(t, 0, 40, 10) | 166 | r.four(t, 0, 40, 10) | ||
| 189 | if len(r.spawn) != 0 { | 167 | if len(r.spawn) != 0 { | ||
| 190 | t.Fatal("second terminal while the first is awaited") | 168 | t.Fatal("second terminal while the first is awaited") | ||
M internal/ports/client_commands.go +0 -6
| @@ -120,12 +120,6 @@ type SetOutputs struct { | |||
| 120 | 120 | ||
| 121 | 121 | func (SetOutputs) clientCommand() {} | |
| 122 | 122 | ||
| 123 | - | // SlotsPending tells wayland whether a slot waits for its window. Only then | |
| 124 | - | // does it read the SlotEnv of mapping clients. | |
| 125 | - | type SlotsPending struct{ Pending bool } | |
| 126 | - | ||
| 127 | - | func (SlotsPending) clientCommand() {} | |
| 128 | - | ||
| 129 | 123 | // CloseWindow carries core → wayland close requests. | |
| 130 | 124 | type CloseWindow struct{ ID WindowID } | |
| 131 | 125 | ||
| @@ -120,12 +120,6 @@ type SetOutputs struct { | |||||
| 120 | 120 | ||||
| 121 | func (SetOutputs) clientCommand() {} | 121 | func (SetOutputs) clientCommand() {} | ||
| 122 | 122 | ||||
| 123 | - | // SlotsPending tells wayland whether a slot waits for its window. Only then | |||
| 124 | - | // does it read the SlotEnv of mapping clients. | |||
| 125 | - | type SlotsPending struct{ Pending bool } | |||
| 126 | - | ||||
| 127 | - | func (SlotsPending) clientCommand() {} | |||
| 128 | - | ||||
| 129 | // CloseWindow carries core → wayland close requests. | 123 | // CloseWindow carries core → wayland close requests. | ||
| 130 | type CloseWindow struct{ ID WindowID } | 124 | type CloseWindow struct{ ID WindowID } | ||
| 131 | 125 | ||||