fix(daemon): adapt coordinator render deadlines
- commit
- 3398032cc9f96fca5221911cac4e49605cba6616
- author
- brice · · 2026-07-11 17:33 UTC
- committer
- brice · · 2026-07-11 17:33 UTC
- parent
- c1108a4
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
3 files · +33 -3
M internal/usecase/daemon/render.go +16 -1
| @@ -2,6 +2,7 @@ | |||
| 2 | 2 | package daemon | |
| 3 | 3 | ||
| 4 | 4 | import ( | |
| 5 | + | "bytes" | |
| 5 | 6 | "errors" | |
| 6 | 7 | "strconv" | |
| 7 | 8 | ||
| @@ -14,6 +15,14 @@ import ( | |||
| 14 | 15 | "github.com/bnema/vev/pkg/vt" | |
| 15 | 16 | ) | |
| 16 | 17 | ||
| 18 | + | // completedSynchronizedUpdate reports a full DEC 2026 batch contained in one | |
| 19 | + | // PTY read. The VT's final state is inactive in this case, so it cannot be | |
| 20 | + | // inferred from before/after SyncUpdateActive alone. | |
| 21 | + | func completedSynchronizedUpdate(data []byte) bool { | |
| 22 | + | _, tail, found := bytes.Cut(data, []byte(renderer.SyncStartCSI)) | |
| 23 | + | return found && bytes.Contains(tail, []byte(renderer.SyncEndCSI)) | |
| 24 | + | } | |
| 25 | + | ||
| 17 | 26 | func signal(ch chan struct{}) { | |
| 18 | 27 | select { | |
| 19 | 28 | case ch <- struct{}{}: | |
| @@ -46,6 +55,7 @@ func (d *Daemon) ptyReader(sess *session, tb *tab, p *pane) { | |||
| 46 | 55 | p.screen.Write(data) | |
| 47 | 56 | p.refreshTerminalTitleLocked() | |
| 48 | 57 | isSyncing := p.screen.SyncUpdateActive() | |
| 58 | + | completeSyncRead := !wasSyncing && !isSyncing && completedSynchronizedUpdate(data) | |
| 49 | 59 | var syncGen uint64 | |
| 50 | 60 | if wasSyncing != isSyncing { | |
| 51 | 61 | if isSyncing { | |
| @@ -91,7 +101,12 @@ func (d *Daemon) ptyReader(sess *session, tb *tab, p *pane) { | |||
| 91 | 101 | rc.noteSyncEnd(gen) | |
| 92 | 102 | } | |
| 93 | 103 | } | |
| 94 | - | if !isSyncing && wasSyncing == isSyncing { | |
| 104 | + | if completeSyncRead { | |
| 105 | + | // The entire batch completed in this read. Its final screen state | |
| 106 | + | // is ready now, so bypass the bulk debounce without arming a | |
| 107 | + | // watchdog for a batch that no longer exists. | |
| 108 | + | rc.invalidate(renderInvalidation{class: invalidateUrgent, producer: "render.go"}) | |
| 109 | + | } else if !isSyncing && wasSyncing == isSyncing { | |
| 95 | 110 | rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"}) | |
| 96 | 111 | } | |
| 97 | 112 | } | |
| @@ -2,6 +2,7 @@ | |||||
| 2 | package daemon | 2 | package daemon | ||
| 3 | 3 | ||||
| 4 | import ( | 4 | import ( | ||
| 5 | + | "bytes" | |||
| 5 | "errors" | 6 | "errors" | ||
| 6 | "strconv" | 7 | "strconv" | ||
| 7 | 8 | ||||
| @@ -14,6 +15,14 @@ import ( | |||||
| 14 | "github.com/bnema/vev/pkg/vt" | 15 | "github.com/bnema/vev/pkg/vt" | ||
| 15 | ) | 16 | ) | ||
| 16 | 17 | ||||
| 18 | + | // completedSynchronizedUpdate reports a full DEC 2026 batch contained in one | |||
| 19 | + | // PTY read. The VT's final state is inactive in this case, so it cannot be | |||
| 20 | + | // inferred from before/after SyncUpdateActive alone. | |||
| 21 | + | func completedSynchronizedUpdate(data []byte) bool { | |||
| 22 | + | _, tail, found := bytes.Cut(data, []byte(renderer.SyncStartCSI)) | |||
| 23 | + | return found && bytes.Contains(tail, []byte(renderer.SyncEndCSI)) | |||
| 24 | + | } | |||
| 25 | + | ||||
| 17 | func signal(ch chan struct{}) { | 26 | func signal(ch chan struct{}) { | ||
| 18 | select { | 27 | select { | ||
| 19 | case ch <- struct{}{}: | 28 | case ch <- struct{}{}: | ||
| @@ -46,6 +55,7 @@ func (d *Daemon) ptyReader(sess *session, tb *tab, p *pane) { | |||||
| 46 | p.screen.Write(data) | 55 | p.screen.Write(data) | ||
| 47 | p.refreshTerminalTitleLocked() | 56 | p.refreshTerminalTitleLocked() | ||
| 48 | isSyncing := p.screen.SyncUpdateActive() | 57 | isSyncing := p.screen.SyncUpdateActive() | ||
| 58 | + | completeSyncRead := !wasSyncing && !isSyncing && completedSynchronizedUpdate(data) | |||
| 49 | var syncGen uint64 | 59 | var syncGen uint64 | ||
| 50 | if wasSyncing != isSyncing { | 60 | if wasSyncing != isSyncing { | ||
| 51 | if isSyncing { | 61 | if isSyncing { | ||
| @@ -91,7 +101,12 @@ func (d *Daemon) ptyReader(sess *session, tb *tab, p *pane) { | |||||
| 91 | rc.noteSyncEnd(gen) | 101 | rc.noteSyncEnd(gen) | ||
| 92 | } | 102 | } | ||
| 93 | } | 103 | } | ||
| 94 | - | if !isSyncing && wasSyncing == isSyncing { | 104 | + | if completeSyncRead { |
| 105 | + | // The entire batch completed in this read. Its final screen state | |||
| 106 | + | // is ready now, so bypass the bulk debounce without arming a | |||
| 107 | + | // watchdog for a batch that no longer exists. | |||
| 108 | + | rc.invalidate(renderInvalidation{class: invalidateUrgent, producer: "render.go"}) | |||
| 109 | + | } else if !isSyncing && wasSyncing == isSyncing { | |||
| 95 | rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"}) | 110 | rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"}) | ||
| 96 | } | 111 | } | ||
| 97 | } | 112 | } | ||
M internal/usecase/daemon/render_coordinator.go +16 -1
| @@ -102,6 +102,10 @@ type renderCoordinator struct { | |||
| 102 | 102 | pendingReset bool | |
| 103 | 103 | pendingUrgent bool | |
| 104 | 104 | coalesced int | |
| 105 | + | // outputPressure carries recent bulk-burst pressure between completed | |
| 106 | + | // batches. It selects a bounded 8–16ms deadline without letting a busy | |
| 107 | + | // batch perpetually reset its own timer. | |
| 108 | + | outputPressure int | |
| 105 | 109 | ||
| 106 | 110 | // previewWake receives the same coalesced wakes for the legacy single | |
| 107 | 111 | // subscriber. previewWakes tracks picker subscriptions by viewer, so one | |
| @@ -187,8 +191,9 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) { | |||
| 187 | 191 | c.cancelNormalTimerLocked() | |
| 188 | 192 | } | |
| 189 | 193 | gen := c.generation | |
| 190 | - | delay := minOutputRenderDeadline | |
| 194 | + | delay := minOutputRenderDeadline + time.Duration(c.outputPressure)*time.Millisecond | |
| 191 | 195 | if c.pendingUrgent { | |
| 196 | + | // Urgent state always supersedes bulk pressure and is never extended. | |
| 192 | 197 | delay = urgentRenderDeadline | |
| 193 | 198 | } | |
| 194 | 199 | clock := c.opts.clock | |
| @@ -469,6 +474,16 @@ func (c *renderCoordinator) fire(gen uint64, watchdog bool) { | |||
| 469 | 474 | w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, watchdog: watchdog} | |
| 470 | 475 | c.metrics.wakes.Add(1) | |
| 471 | 476 | c.metrics.coalesced.Add(uint64(w.coalesced)) | |
| 477 | + | if !w.urgent { | |
| 478 | + | if w.coalesced > 1 { | |
| 479 | + | c.outputPressure += w.coalesced - 1 | |
| 480 | + | if c.outputPressure > int((maxOutputRenderDeadline-minOutputRenderDeadline)/time.Millisecond) { | |
| 481 | + | c.outputPressure = int((maxOutputRenderDeadline - minOutputRenderDeadline) / time.Millisecond) | |
| 482 | + | } | |
| 483 | + | } else if c.outputPressure > 0 { | |
| 484 | + | c.outputPressure-- | |
| 485 | + | } | |
| 486 | + | } | |
| 472 | 487 | c.cancelNormalTimerLocked() | |
| 473 | 488 | c.pending, c.pendingReset, c.pendingUrgent, c.coalesced, c.armed = false, false, false, 0, false | |
| 474 | 489 | wake, preview := c.opts.wake, c.previewWake | |
| @@ -102,6 +102,10 @@ type renderCoordinator struct { | |||||
| 102 | pendingReset bool | 102 | pendingReset bool | ||
| 103 | pendingUrgent bool | 103 | pendingUrgent bool | ||
| 104 | coalesced int | 104 | coalesced int | ||
| 105 | + | // outputPressure carries recent bulk-burst pressure between completed | |||
| 106 | + | // batches. It selects a bounded 8–16ms deadline without letting a busy | |||
| 107 | + | // batch perpetually reset its own timer. | |||
| 108 | + | outputPressure int | |||
| 105 | 109 | ||||
| 106 | // previewWake receives the same coalesced wakes for the legacy single | 110 | // previewWake receives the same coalesced wakes for the legacy single | ||
| 107 | // subscriber. previewWakes tracks picker subscriptions by viewer, so one | 111 | // subscriber. previewWakes tracks picker subscriptions by viewer, so one | ||
| @@ -187,8 +191,9 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) { | |||||
| 187 | c.cancelNormalTimerLocked() | 191 | c.cancelNormalTimerLocked() | ||
| 188 | } | 192 | } | ||
| 189 | gen := c.generation | 193 | gen := c.generation | ||
| 190 | - | delay := minOutputRenderDeadline | 194 | + | delay := minOutputRenderDeadline + time.Duration(c.outputPressure)*time.Millisecond |
| 191 | if c.pendingUrgent { | 195 | if c.pendingUrgent { | ||
| 196 | + | // Urgent state always supersedes bulk pressure and is never extended. | |||
| 192 | delay = urgentRenderDeadline | 197 | delay = urgentRenderDeadline | ||
| 193 | } | 198 | } | ||
| 194 | clock := c.opts.clock | 199 | clock := c.opts.clock | ||
| @@ -469,6 +474,16 @@ func (c *renderCoordinator) fire(gen uint64, watchdog bool) { | |||||
| 469 | w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, watchdog: watchdog} | 474 | w := renderWake{reset: c.pendingReset, urgent: c.pendingUrgent, coalesced: c.coalesced, watchdog: watchdog} | ||
| 470 | c.metrics.wakes.Add(1) | 475 | c.metrics.wakes.Add(1) | ||
| 471 | c.metrics.coalesced.Add(uint64(w.coalesced)) | 476 | c.metrics.coalesced.Add(uint64(w.coalesced)) | ||
| 477 | + | if !w.urgent { | |||
| 478 | + | if w.coalesced > 1 { | |||
| 479 | + | c.outputPressure += w.coalesced - 1 | |||
| 480 | + | if c.outputPressure > int((maxOutputRenderDeadline-minOutputRenderDeadline)/time.Millisecond) { | |||
| 481 | + | c.outputPressure = int((maxOutputRenderDeadline - minOutputRenderDeadline) / time.Millisecond) | |||
| 482 | + | } | |||
| 483 | + | } else if c.outputPressure > 0 { | |||
| 484 | + | c.outputPressure-- | |||
| 485 | + | } | |||
| 486 | + | } | |||
| 472 | c.cancelNormalTimerLocked() | 487 | c.cancelNormalTimerLocked() | ||
| 473 | c.pending, c.pendingReset, c.pendingUrgent, c.coalesced, c.armed = false, false, false, 0, false | 488 | c.pending, c.pendingReset, c.pendingUrgent, c.coalesced, c.armed = false, false, false, 0, false | ||
| 474 | wake, preview := c.opts.wake, c.previewWake | 489 | wake, preview := c.opts.wake, c.previewWake | ||
M internal/usecase/daemon/snapshot_test.go +1 -1
| @@ -223,7 +223,7 @@ func TestPTYReaderSameReadSynchronizedOutputUsesUrgentCoordinatorDeadline(t *tes | |||
| 223 | 223 | invs := make(chan renderInvalidation, 1) | |
| 224 | 224 | clock := &signalClock{timers: make(chan *signalTimer, 2)} | |
| 225 | 225 | sess.installRenderCoordinator(newRenderCoordinator(renderCoordinatorOptions{ | |
| 226 | - | clock: clock, | |
| 226 | + | clock: clock, | |
| 227 | 227 | onInvalidate: func(inv renderInvalidation) { invs <- inv }, | |
| 228 | 228 | })) | |
| 229 | 229 | d.sessions[sess.id] = sess | |
| @@ -223,7 +223,7 @@ func TestPTYReaderSameReadSynchronizedOutputUsesUrgentCoordinatorDeadline(t *tes | |||||
| 223 | invs := make(chan renderInvalidation, 1) | 223 | invs := make(chan renderInvalidation, 1) | ||
| 224 | clock := &signalClock{timers: make(chan *signalTimer, 2)} | 224 | clock := &signalClock{timers: make(chan *signalTimer, 2)} | ||
| 225 | sess.installRenderCoordinator(newRenderCoordinator(renderCoordinatorOptions{ | 225 | sess.installRenderCoordinator(newRenderCoordinator(renderCoordinatorOptions{ | ||
| 226 | - | clock: clock, | 226 | + | clock: clock, |
| 227 | onInvalidate: func(inv renderInvalidation) { invs <- inv }, | 227 | onInvalidate: func(inv renderInvalidation) { invs <- inv }, | ||
| 228 | })) | 228 | })) | ||
| 229 | d.sessions[sess.id] = sess | 229 | d.sessions[sess.id] = sess | ||