fix: defer sync update rendering
- commit
- 00a0215b90279b23e639d075fca879bb16af20c9
- author
- brice · · 2026-07-02 16:42 UTC
- parent
- e65903a
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +185 -25
M docs/performance.md +4 -3
| @@ -20,11 +20,12 @@ Full end-to-end vev-vs-tmux workload baselines were not run for this change beca | |||
| 20 | 20 | 4. Capture client/daemon syscall summaries with `strace -c` (or platform equivalent), bytes written per rendered frame, and heap profiles or `-benchmem` allocation counts. | |
| 21 | 21 | 5. Store raw command lines and outputs beside this document before comparing. | |
| 22 | 22 | ||
| 23 | - | Microbenchmarks executed locally for this review using `go test ./internal/adapters/ipc -run '^$' -bench=BenchmarkTransportRecvReuse -benchmem`: | |
| 23 | + | Microbenchmarks executed locally for this review using `go test ./internal/adapters/ipc -run '^$' -bench=. -benchmem`: | |
| 24 | 24 | ||
| 25 | - | | benchmark | result | | |
| 25 | + | | benchmark | methodology | | |
| 26 | 26 | | --- | --- | | |
| 27 | - | | BenchmarkTransportRecvReuse | recorded during review; isolates the receiver loop by reusing both transports and moving frame encoding/sending to a helper goroutine | | |
| 27 | + | | BenchmarkTransportSend | writes encoded frames to an in-memory discard `net.Conn`; allocations reported here are Send-side frame assembly costs. | | |
| 28 | + | | BenchmarkTransportRecvReuse | reads a pre-encoded frame from a looping in-memory reader; no `Send` call, goroutine, channel, or frame production work runs inside the measured loop, so allocations reported here are Recv-side costs only. | | |
| 28 | 29 | ||
| 29 | 30 | | workload | vev baseline | tmux baseline | notes | | |
| 30 | 31 | | --- | --- | --- | --- | | |
| @@ -20,11 +20,12 @@ Full end-to-end vev-vs-tmux workload baselines were not run for this change beca | |||||
| 20 | 4. Capture client/daemon syscall summaries with `strace -c` (or platform equivalent), bytes written per rendered frame, and heap profiles or `-benchmem` allocation counts. | 20 | 4. Capture client/daemon syscall summaries with `strace -c` (or platform equivalent), bytes written per rendered frame, and heap profiles or `-benchmem` allocation counts. | ||
| 21 | 5. Store raw command lines and outputs beside this document before comparing. | 21 | 5. Store raw command lines and outputs beside this document before comparing. | ||
| 22 | 22 | ||||
| 23 | - | Microbenchmarks executed locally for this review using `go test ./internal/adapters/ipc -run '^$' -bench=BenchmarkTransportRecvReuse -benchmem`: | 23 | + | Microbenchmarks executed locally for this review using `go test ./internal/adapters/ipc -run '^$' -bench=. -benchmem`: |
| 24 | 24 | ||||
| 25 | - | | benchmark | result | | 25 | + | | benchmark | methodology | |
| 26 | | --- | --- | | 26 | | --- | --- | | ||
| 27 | - | | BenchmarkTransportRecvReuse | recorded during review; isolates the receiver loop by reusing both transports and moving frame encoding/sending to a helper goroutine | | 27 | + | | BenchmarkTransportSend | writes encoded frames to an in-memory discard `net.Conn`; allocations reported here are Send-side frame assembly costs. | |
| 28 | + | | BenchmarkTransportRecvReuse | reads a pre-encoded frame from a looping in-memory reader; no `Send` call, goroutine, channel, or frame production work runs inside the measured loop, so allocations reported here are Recv-side costs only. | | |||
| 28 | 29 | ||||
| 29 | | workload | vev baseline | tmux baseline | notes | | 30 | | workload | vev baseline | tmux baseline | notes | | ||
| 30 | | --- | --- | --- | --- | | 31 | | --- | --- | --- | --- | | ||
M internal/adapters/ipc/transport_benchmark_test.go +70 -20
| @@ -1,38 +1,88 @@ | |||
| 1 | 1 | package ipc | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | + | "encoding/binary" | |
| 5 | + | "io" | |
| 4 | 6 | "net" | |
| 5 | 7 | "testing" | |
| 8 | + | "time" | |
| 6 | 9 | ||
| 7 | 10 | "github.com/bnema/vev/internal/ports" | |
| 8 | 11 | ) | |
| 9 | 12 | ||
| 10 | - | func BenchmarkTransportRecvReuse(b *testing.B) { | |
| 11 | - | c1, c2 := net.Pipe() | |
| 12 | - | defer func() { _ = c1.Close() }() | |
| 13 | - | defer func() { _ = c2.Close() }() | |
| 14 | - | recv := NewTransport(c2) | |
| 15 | - | send := NewTransport(c1) | |
| 13 | + | func BenchmarkTransportSend(b *testing.B) { | |
| 14 | + | conn := discardConn{} | |
| 15 | + | tr := NewTransport(conn) | |
| 16 | 16 | payload := []byte("payload payload payload") | |
| 17 | - | frames := make(chan ports.Frame, 1024) | |
| 18 | - | done := make(chan struct{}) | |
| 19 | - | go func() { | |
| 20 | - | defer close(done) | |
| 21 | - | for frame := range frames { | |
| 22 | - | if err := send.Send(frame); err != nil { | |
| 23 | - | return | |
| 24 | - | } | |
| 17 | + | frame := ports.Frame{Type: ports.MsgOutput, Payload: payload} | |
| 18 | + | ||
| 19 | + | b.ReportAllocs() | |
| 20 | + | for b.Loop() { | |
| 21 | + | if err := tr.Send(frame); err != nil { | |
| 22 | + | b.Fatal(err) | |
| 25 | 23 | } | |
| 26 | - | }() | |
| 24 | + | } | |
| 25 | + | } | |
| 26 | + | ||
| 27 | + | func BenchmarkTransportRecvReuse(b *testing.B) { | |
| 28 | + | payload := []byte("payload payload payload") | |
| 29 | + | encoded := encodeBenchmarkFrame(ports.Frame{Type: ports.MsgOutput, Payload: payload}) | |
| 30 | + | recv := &unixTransport{conn: &loopingReaderConn{data: encoded}} | |
| 31 | + | ||
| 27 | 32 | b.ReportAllocs() | |
| 28 | - | b.ResetTimer() | |
| 29 | 33 | for b.Loop() { | |
| 30 | - | frames <- ports.Frame{Type: ports.MsgOutput, Payload: payload} | |
| 31 | 34 | if _, err := recv.Recv(); err != nil { | |
| 32 | 35 | b.Fatal(err) | |
| 33 | 36 | } | |
| 34 | 37 | } | |
| 35 | - | b.StopTimer() | |
| 36 | - | close(frames) | |
| 37 | - | <-done | |
| 38 | 38 | } | |
| 39 | + | ||
| 40 | + | func encodeBenchmarkFrame(f ports.Frame) []byte { | |
| 41 | + | n := 1 + len(f.Payload) | |
| 42 | + | buf := make([]byte, frameHeaderLen+n) | |
| 43 | + | binary.BigEndian.PutUint32(buf[:frameHeaderLen], uint32(n)) | |
| 44 | + | buf[frameHeaderLen] = byte(f.Type) | |
| 45 | + | copy(buf[frameHeaderLen+1:], f.Payload) | |
| 46 | + | return buf | |
| 47 | + | } | |
| 48 | + | ||
| 49 | + | type loopingReaderConn struct { | |
| 50 | + | data []byte | |
| 51 | + | off int | |
| 52 | + | } | |
| 53 | + | ||
| 54 | + | func (c *loopingReaderConn) Read(p []byte) (int, error) { | |
| 55 | + | if len(c.data) == 0 { | |
| 56 | + | return 0, io.EOF | |
| 57 | + | } | |
| 58 | + | for i := range p { | |
| 59 | + | p[i] = c.data[c.off] | |
| 60 | + | c.off = (c.off + 1) % len(c.data) | |
| 61 | + | } | |
| 62 | + | return len(p), nil | |
| 63 | + | } | |
| 64 | + | ||
| 65 | + | func (c *loopingReaderConn) Write([]byte) (int, error) { return 0, io.ErrClosedPipe } | |
| 66 | + | func (c *loopingReaderConn) Close() error { return nil } | |
| 67 | + | func (c *loopingReaderConn) LocalAddr() net.Addr { return nil } | |
| 68 | + | func (c *loopingReaderConn) RemoteAddr() net.Addr { return nil } | |
| 69 | + | func (c *loopingReaderConn) SetDeadline(time.Time) error { | |
| 70 | + | return nil | |
| 71 | + | } | |
| 72 | + | func (c *loopingReaderConn) SetReadDeadline(time.Time) error { | |
| 73 | + | return nil | |
| 74 | + | } | |
| 75 | + | func (c *loopingReaderConn) SetWriteDeadline(time.Time) error { | |
| 76 | + | return nil | |
| 77 | + | } | |
| 78 | + | ||
| 79 | + | type discardConn struct{} | |
| 80 | + | ||
| 81 | + | func (discardConn) Read([]byte) (int, error) { return 0, io.ErrClosedPipe } | |
| 82 | + | func (discardConn) Write(p []byte) (int, error) { return len(p), nil } | |
| 83 | + | func (discardConn) Close() error { return nil } | |
| 84 | + | func (discardConn) LocalAddr() net.Addr { return nil } | |
| 85 | + | func (discardConn) RemoteAddr() net.Addr { return nil } | |
| 86 | + | func (discardConn) SetDeadline(time.Time) error { return nil } | |
| 87 | + | func (discardConn) SetReadDeadline(time.Time) error { return nil } | |
| 88 | + | func (discardConn) SetWriteDeadline(time.Time) error { return nil } | |
| @@ -1,38 +1,88 @@ | |||||
| 1 | package ipc | 1 | package ipc | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | + | "encoding/binary" | |||
| 5 | + | "io" | |||
| 4 | "net" | 6 | "net" | ||
| 5 | "testing" | 7 | "testing" | ||
| 8 | + | "time" | |||
| 6 | 9 | ||||
| 7 | "github.com/bnema/vev/internal/ports" | 10 | "github.com/bnema/vev/internal/ports" | ||
| 8 | ) | 11 | ) | ||
| 9 | 12 | ||||
| 10 | - | func BenchmarkTransportRecvReuse(b *testing.B) { | 13 | + | func BenchmarkTransportSend(b *testing.B) { |
| 11 | - | c1, c2 := net.Pipe() | 14 | + | conn := discardConn{} |
| 12 | - | defer func() { _ = c1.Close() }() | 15 | + | tr := NewTransport(conn) |
| 13 | - | defer func() { _ = c2.Close() }() | |||
| 14 | - | recv := NewTransport(c2) | |||
| 15 | - | send := NewTransport(c1) | |||
| 16 | payload := []byte("payload payload payload") | 16 | payload := []byte("payload payload payload") | ||
| 17 | - | frames := make(chan ports.Frame, 1024) | 17 | + | frame := ports.Frame{Type: ports.MsgOutput, Payload: payload} |
| 18 | - | done := make(chan struct{}) | 18 | + | |
| 19 | - | go func() { | 19 | + | b.ReportAllocs() |
| 20 | - | defer close(done) | 20 | + | for b.Loop() { |
| 21 | - | for frame := range frames { | 21 | + | if err := tr.Send(frame); err != nil { |
| 22 | - | if err := send.Send(frame); err != nil { | 22 | + | b.Fatal(err) |
| 23 | - | return | |||
| 24 | - | } | |||
| 25 | } | 23 | } | ||
| 26 | - | }() | 24 | + | } |
| 25 | + | } | |||
| 26 | + | ||||
| 27 | + | func BenchmarkTransportRecvReuse(b *testing.B) { | |||
| 28 | + | payload := []byte("payload payload payload") | |||
| 29 | + | encoded := encodeBenchmarkFrame(ports.Frame{Type: ports.MsgOutput, Payload: payload}) | |||
| 30 | + | recv := &unixTransport{conn: &loopingReaderConn{data: encoded}} | |||
| 31 | + | ||||
| 27 | b.ReportAllocs() | 32 | b.ReportAllocs() | ||
| 28 | - | b.ResetTimer() | |||
| 29 | for b.Loop() { | 33 | for b.Loop() { | ||
| 30 | - | frames <- ports.Frame{Type: ports.MsgOutput, Payload: payload} | |||
| 31 | if _, err := recv.Recv(); err != nil { | 34 | if _, err := recv.Recv(); err != nil { | ||
| 32 | b.Fatal(err) | 35 | b.Fatal(err) | ||
| 33 | } | 36 | } | ||
| 34 | } | 37 | } | ||
| 35 | - | b.StopTimer() | |||
| 36 | - | close(frames) | |||
| 37 | - | <-done | |||
| 38 | } | 38 | } | ||
| 39 | + | ||||
| 40 | + | func encodeBenchmarkFrame(f ports.Frame) []byte { | |||
| 41 | + | n := 1 + len(f.Payload) | |||
| 42 | + | buf := make([]byte, frameHeaderLen+n) | |||
| 43 | + | binary.BigEndian.PutUint32(buf[:frameHeaderLen], uint32(n)) | |||
| 44 | + | buf[frameHeaderLen] = byte(f.Type) | |||
| 45 | + | copy(buf[frameHeaderLen+1:], f.Payload) | |||
| 46 | + | return buf | |||
| 47 | + | } | |||
| 48 | + | ||||
| 49 | + | type loopingReaderConn struct { | |||
| 50 | + | data []byte | |||
| 51 | + | off int | |||
| 52 | + | } | |||
| 53 | + | ||||
| 54 | + | func (c *loopingReaderConn) Read(p []byte) (int, error) { | |||
| 55 | + | if len(c.data) == 0 { | |||
| 56 | + | return 0, io.EOF | |||
| 57 | + | } | |||
| 58 | + | for i := range p { | |||
| 59 | + | p[i] = c.data[c.off] | |||
| 60 | + | c.off = (c.off + 1) % len(c.data) | |||
| 61 | + | } | |||
| 62 | + | return len(p), nil | |||
| 63 | + | } | |||
| 64 | + | ||||
| 65 | + | func (c *loopingReaderConn) Write([]byte) (int, error) { return 0, io.ErrClosedPipe } | |||
| 66 | + | func (c *loopingReaderConn) Close() error { return nil } | |||
| 67 | + | func (c *loopingReaderConn) LocalAddr() net.Addr { return nil } | |||
| 68 | + | func (c *loopingReaderConn) RemoteAddr() net.Addr { return nil } | |||
| 69 | + | func (c *loopingReaderConn) SetDeadline(time.Time) error { | |||
| 70 | + | return nil | |||
| 71 | + | } | |||
| 72 | + | func (c *loopingReaderConn) SetReadDeadline(time.Time) error { | |||
| 73 | + | return nil | |||
| 74 | + | } | |||
| 75 | + | func (c *loopingReaderConn) SetWriteDeadline(time.Time) error { | |||
| 76 | + | return nil | |||
| 77 | + | } | |||
| 78 | + | ||||
| 79 | + | type discardConn struct{} | |||
| 80 | + | ||||
| 81 | + | func (discardConn) Read([]byte) (int, error) { return 0, io.ErrClosedPipe } | |||
| 82 | + | func (discardConn) Write(p []byte) (int, error) { return len(p), nil } | |||
| 83 | + | func (discardConn) Close() error { return nil } | |||
| 84 | + | func (discardConn) LocalAddr() net.Addr { return nil } | |||
| 85 | + | func (discardConn) RemoteAddr() net.Addr { return nil } | |||
| 86 | + | func (discardConn) SetDeadline(time.Time) error { return nil } | |||
| 87 | + | func (discardConn) SetReadDeadline(time.Time) error { return nil } | |||
| 88 | + | func (discardConn) SetWriteDeadline(time.Time) error { return nil } | |||
M internal/usecase/daemon/daemon.go +15 -2
| @@ -26,6 +26,7 @@ | |||
| 26 | 26 | package daemon | |
| 27 | 27 | ||
| 28 | 28 | import ( | |
| 29 | + | "bytes" | |
| 29 | 30 | "context" | |
| 30 | 31 | "errors" | |
| 31 | 32 | "fmt" | |
| @@ -951,12 +952,13 @@ func (d *Daemon) ptyReader(sess *session, win *window) { | |||
| 951 | 952 | for { | |
| 952 | 953 | n, err := win.pty.Read(buf) | |
| 953 | 954 | if n > 0 { | |
| 955 | + | data := buf[:n] | |
| 954 | 956 | win.mu.Lock() | |
| 955 | 957 | wasSyncing := win.screen.SyncUpdateActive() | |
| 956 | - | win.screen.Write(buf[:n]) | |
| 958 | + | win.screen.Write(data) | |
| 957 | 959 | isSyncing := win.screen.SyncUpdateActive() | |
| 958 | 960 | win.mu.Unlock() | |
| 959 | - | if wasSyncing && !isSyncing { | |
| 961 | + | if (wasSyncing && !isSyncing) || (!isSyncing && syncUpdateEndIn(data)) { | |
| 960 | 962 | signal(win.flush) | |
| 961 | 963 | continue | |
| 962 | 964 | } | |
| @@ -1050,9 +1052,20 @@ func windowDone(win *window) <-chan struct{} { | |||
| 1050 | 1052 | return win.ctx.Done() | |
| 1051 | 1053 | } | |
| 1052 | 1054 | ||
| 1055 | + | func syncUpdateEndIn(data []byte) bool { | |
| 1056 | + | return bytes.Contains(data, []byte("\x1b[?2026l")) | |
| 1057 | + | } | |
| 1058 | + | ||
| 1053 | 1059 | // render paints the current client, or (when detached) just clears accumulated | |
| 1054 | 1060 | // damage so it never grows unbounded while headless. | |
| 1055 | 1061 | func (d *Daemon) render(sess *session, win *window) { | |
| 1062 | + | win.mu.Lock() | |
| 1063 | + | if win.screen.SyncUpdateActive() { | |
| 1064 | + | win.mu.Unlock() | |
| 1065 | + | return | |
| 1066 | + | } | |
| 1067 | + | win.mu.Unlock() | |
| 1068 | + | ||
| 1056 | 1069 | sess.mu.Lock() | |
| 1057 | 1070 | ac := sess.client | |
| 1058 | 1071 | active := sess.active >= 0 && sess.active < len(sess.windows) && sess.windows[sess.active] == win | |
| @@ -26,6 +26,7 @@ | |||||
| 26 | package daemon | 26 | package daemon | ||
| 27 | 27 | ||||
| 28 | import ( | 28 | import ( | ||
| 29 | + | "bytes" | |||
| 29 | "context" | 30 | "context" | ||
| 30 | "errors" | 31 | "errors" | ||
| 31 | "fmt" | 32 | "fmt" | ||
| @@ -951,12 +952,13 @@ func (d *Daemon) ptyReader(sess *session, win *window) { | |||||
| 951 | for { | 952 | for { | ||
| 952 | n, err := win.pty.Read(buf) | 953 | n, err := win.pty.Read(buf) | ||
| 953 | if n > 0 { | 954 | if n > 0 { | ||
| 955 | + | data := buf[:n] | |||
| 954 | win.mu.Lock() | 956 | win.mu.Lock() | ||
| 955 | wasSyncing := win.screen.SyncUpdateActive() | 957 | wasSyncing := win.screen.SyncUpdateActive() | ||
| 956 | - | win.screen.Write(buf[:n]) | 958 | + | win.screen.Write(data) |
| 957 | isSyncing := win.screen.SyncUpdateActive() | 959 | isSyncing := win.screen.SyncUpdateActive() | ||
| 958 | win.mu.Unlock() | 960 | win.mu.Unlock() | ||
| 959 | - | if wasSyncing && !isSyncing { | 961 | + | if (wasSyncing && !isSyncing) || (!isSyncing && syncUpdateEndIn(data)) { |
| 960 | signal(win.flush) | 962 | signal(win.flush) | ||
| 961 | continue | 963 | continue | ||
| 962 | } | 964 | } | ||
| @@ -1050,9 +1052,20 @@ func windowDone(win *window) <-chan struct{} { | |||||
| 1050 | return win.ctx.Done() | 1052 | return win.ctx.Done() | ||
| 1051 | } | 1053 | } | ||
| 1052 | 1054 | ||||
| 1055 | + | func syncUpdateEndIn(data []byte) bool { | |||
| 1056 | + | return bytes.Contains(data, []byte("\x1b[?2026l")) | |||
| 1057 | + | } | |||
| 1058 | + | ||||
| 1053 | // render paints the current client, or (when detached) just clears accumulated | 1059 | // render paints the current client, or (when detached) just clears accumulated | ||
| 1054 | // damage so it never grows unbounded while headless. | 1060 | // damage so it never grows unbounded while headless. | ||
| 1055 | func (d *Daemon) render(sess *session, win *window) { | 1061 | func (d *Daemon) render(sess *session, win *window) { | ||
| 1062 | + | win.mu.Lock() | |||
| 1063 | + | if win.screen.SyncUpdateActive() { | |||
| 1064 | + | win.mu.Unlock() | |||
| 1065 | + | return | |||
| 1066 | + | } | |||
| 1067 | + | win.mu.Unlock() | |||
| 1068 | + | ||||
| 1056 | sess.mu.Lock() | 1069 | sess.mu.Lock() | ||
| 1057 | ac := sess.client | 1070 | ac := sess.client | ||
| 1058 | active := sess.active >= 0 && sess.active < len(sess.windows) && sess.windows[sess.active] == win | 1071 | active := sess.active >= 0 && sess.active < len(sess.windows) && sess.windows[sess.active] == win | ||
M internal/usecase/daemon/daemon_test.go +96 -0
| @@ -1338,6 +1338,102 @@ func TestSendErrorKillsEphemeral(t *testing.T) { | |||
| 1338 | 1338 | d.waitNotifies() | |
| 1339 | 1339 | } | |
| 1340 | 1340 | ||
| 1341 | + | func TestSchedulerDefersPendingDirtyTimerDuringSynchronizedUpdate(t *testing.T) { | |
| 1342 | + | mc := portsmocks.NewMockClock(t) | |
| 1343 | + | mc.EXPECT().Now().Return(time.Now()).Maybe() | |
| 1344 | + | mt := portsmocks.NewMockTimer(t) | |
| 1345 | + | timerCh := make(chan time.Time, 1) | |
| 1346 | + | mc.EXPECT().NewTimer(mock.Anything).Return(mt).Maybe() | |
| 1347 | + | mt.EXPECT().C().Return(timerCh).Maybe() | |
| 1348 | + | mt.EXPECT().Stop().Return(true).Maybe() | |
| 1349 | + | ||
| 1350 | + | var outputs atomic.Int32 | |
| 1351 | + | gotOutput := make(chan struct{}, 1) | |
| 1352 | + | tr := portsmocks.NewMockTransport(t) | |
| 1353 | + | tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error { | |
| 1354 | + | if f.Type == ports.MsgOutput { | |
| 1355 | + | outputs.Add(1) | |
| 1356 | + | select { | |
| 1357 | + | case gotOutput <- struct{}{}: | |
| 1358 | + | default: | |
| 1359 | + | } | |
| 1360 | + | } | |
| 1361 | + | return nil | |
| 1362 | + | }).Maybe() | |
| 1363 | + | ||
| 1364 | + | d := New(nil, mc, slog.New(slog.NewTextHandler(io.Discard, nil))) | |
| 1365 | + | sctx, cancel := context.WithCancel(context.Background()) | |
| 1366 | + | win := &window{screen: vt.NewScreen(20, 5), dirty: make(chan struct{}, 1), flush: make(chan struct{}, 1), size: domain.Size{Cols: 20, Rows: 5}, ctx: sctx, cancel: cancel} | |
| 1367 | + | ac := &attachedClient{tr: tr, rend: renderer.New(renderer.Capabilities{})} | |
| 1368 | + | sess := &session{id: "sync", name: "sync", windows: []*window{win}, ctx: sctx, cancel: cancel, client: ac} | |
| 1369 | + | ||
| 1370 | + | win.mu.Lock() | |
| 1371 | + | win.screen.Write([]byte("before")) | |
| 1372 | + | win.mu.Unlock() | |
| 1373 | + | ||
| 1374 | + | d.sessWg.Add(1) | |
| 1375 | + | go d.scheduler(sess, win) | |
| 1376 | + | win.dirty <- struct{}{} | |
| 1377 | + | ||
| 1378 | + | win.mu.Lock() | |
| 1379 | + | win.screen.Write([]byte("\x1b[?2026hafter")) | |
| 1380 | + | win.mu.Unlock() | |
| 1381 | + | timerCh <- time.Now() | |
| 1382 | + | ||
| 1383 | + | select { | |
| 1384 | + | case <-gotOutput: | |
| 1385 | + | t.Fatal("scheduler rendered while synchronized update was active") | |
| 1386 | + | case <-time.After(20 * time.Millisecond): | |
| 1387 | + | } | |
| 1388 | + | require.Equal(t, int32(0), outputs.Load()) | |
| 1389 | + | ||
| 1390 | + | win.mu.Lock() | |
| 1391 | + | win.screen.Write([]byte(" done\x1b[?2026l")) | |
| 1392 | + | win.mu.Unlock() | |
| 1393 | + | win.flush <- struct{}{} | |
| 1394 | + | select { | |
| 1395 | + | case <-gotOutput: | |
| 1396 | + | case <-time.After(time.Second): | |
| 1397 | + | t.Fatal("sync end did not flush deferred damage") | |
| 1398 | + | } | |
| 1399 | + | ||
| 1400 | + | cancel() | |
| 1401 | + | d.sessWg.Wait() | |
| 1402 | + | } | |
| 1403 | + | ||
| 1404 | + | func TestPTYReaderSameReadSynchronizedUpdateStartEndFlushesImmediately(t *testing.T) { | |
| 1405 | + | d := newTestDaemon(t, nil, stubClock{}) | |
| 1406 | + | p := portsmocks.NewMockPTY(t) | |
| 1407 | + | chunks := [][]byte{[]byte("\x1b[?2026hhello\x1b[?2026l")} | |
| 1408 | + | p.EXPECT().Read(mock.Anything).RunAndReturn(func(buf []byte) (int, error) { | |
| 1409 | + | if len(chunks) == 0 { | |
| 1410 | + | return 0, io.EOF | |
| 1411 | + | } | |
| 1412 | + | n := copy(buf, chunks[0]) | |
| 1413 | + | chunks = chunks[1:] | |
| 1414 | + | return n, nil | |
| 1415 | + | }) | |
| 1416 | + | p.EXPECT().Close().Return(nil).Maybe() | |
| 1417 | + | ||
| 1418 | + | sctx, cancel := context.WithCancel(context.Background()) | |
| 1419 | + | win := &window{pty: p, screen: vt.NewScreen(20, 5), dirty: make(chan struct{}, 1), flush: make(chan struct{}, 1), size: domain.Size{Cols: 20, Rows: 5}, ctx: sctx, cancel: cancel} | |
| 1420 | + | sess := &session{id: "sync", name: "sync", windows: []*window{win}, ctx: sctx, cancel: cancel} | |
| 1421 | + | d.sessions[sess.id] = sess | |
| 1422 | + | d.sessWg.Add(1) | |
| 1423 | + | d.ptyReader(sess, win) | |
| 1424 | + | ||
| 1425 | + | select { | |
| 1426 | + | case <-win.dirty: | |
| 1427 | + | t.Fatal("same-read synchronized update end should request flush, not dirty") | |
| 1428 | + | default: | |
| 1429 | + | } | |
| 1430 | + | select { | |
| 1431 | + | case <-win.flush: | |
| 1432 | + | case <-time.After(time.Second): | |
| 1433 | + | t.Fatal("same-read synchronized update end did not request immediate flush") | |
| 1434 | + | } | |
| 1435 | + | } | |
| 1436 | + | ||
| 1341 | 1437 | func TestPTYReaderDefersDirtyDuringSynchronizedUpdateAndFlushesAtEnd(t *testing.T) { | |
| 1342 | 1438 | d := newTestDaemon(t, nil, stubClock{}) | |
| 1343 | 1439 | p := portsmocks.NewMockPTY(t) | |
| @@ -1338,6 +1338,102 @@ func TestSendErrorKillsEphemeral(t *testing.T) { | |||||
| 1338 | d.waitNotifies() | 1338 | d.waitNotifies() | ||
| 1339 | } | 1339 | } | ||
| 1340 | 1340 | ||||
| 1341 | + | func TestSchedulerDefersPendingDirtyTimerDuringSynchronizedUpdate(t *testing.T) { | |||
| 1342 | + | mc := portsmocks.NewMockClock(t) | |||
| 1343 | + | mc.EXPECT().Now().Return(time.Now()).Maybe() | |||
| 1344 | + | mt := portsmocks.NewMockTimer(t) | |||
| 1345 | + | timerCh := make(chan time.Time, 1) | |||
| 1346 | + | mc.EXPECT().NewTimer(mock.Anything).Return(mt).Maybe() | |||
| 1347 | + | mt.EXPECT().C().Return(timerCh).Maybe() | |||
| 1348 | + | mt.EXPECT().Stop().Return(true).Maybe() | |||
| 1349 | + | ||||
| 1350 | + | var outputs atomic.Int32 | |||
| 1351 | + | gotOutput := make(chan struct{}, 1) | |||
| 1352 | + | tr := portsmocks.NewMockTransport(t) | |||
| 1353 | + | tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f ports.Frame) error { | |||
| 1354 | + | if f.Type == ports.MsgOutput { | |||
| 1355 | + | outputs.Add(1) | |||
| 1356 | + | select { | |||
| 1357 | + | case gotOutput <- struct{}{}: | |||
| 1358 | + | default: | |||
| 1359 | + | } | |||
| 1360 | + | } | |||
| 1361 | + | return nil | |||
| 1362 | + | }).Maybe() | |||
| 1363 | + | ||||
| 1364 | + | d := New(nil, mc, slog.New(slog.NewTextHandler(io.Discard, nil))) | |||
| 1365 | + | sctx, cancel := context.WithCancel(context.Background()) | |||
| 1366 | + | win := &window{screen: vt.NewScreen(20, 5), dirty: make(chan struct{}, 1), flush: make(chan struct{}, 1), size: domain.Size{Cols: 20, Rows: 5}, ctx: sctx, cancel: cancel} | |||
| 1367 | + | ac := &attachedClient{tr: tr, rend: renderer.New(renderer.Capabilities{})} | |||
| 1368 | + | sess := &session{id: "sync", name: "sync", windows: []*window{win}, ctx: sctx, cancel: cancel, client: ac} | |||
| 1369 | + | ||||
| 1370 | + | win.mu.Lock() | |||
| 1371 | + | win.screen.Write([]byte("before")) | |||
| 1372 | + | win.mu.Unlock() | |||
| 1373 | + | ||||
| 1374 | + | d.sessWg.Add(1) | |||
| 1375 | + | go d.scheduler(sess, win) | |||
| 1376 | + | win.dirty <- struct{}{} | |||
| 1377 | + | ||||
| 1378 | + | win.mu.Lock() | |||
| 1379 | + | win.screen.Write([]byte("\x1b[?2026hafter")) | |||
| 1380 | + | win.mu.Unlock() | |||
| 1381 | + | timerCh <- time.Now() | |||
| 1382 | + | ||||
| 1383 | + | select { | |||
| 1384 | + | case <-gotOutput: | |||
| 1385 | + | t.Fatal("scheduler rendered while synchronized update was active") | |||
| 1386 | + | case <-time.After(20 * time.Millisecond): | |||
| 1387 | + | } | |||
| 1388 | + | require.Equal(t, int32(0), outputs.Load()) | |||
| 1389 | + | ||||
| 1390 | + | win.mu.Lock() | |||
| 1391 | + | win.screen.Write([]byte(" done\x1b[?2026l")) | |||
| 1392 | + | win.mu.Unlock() | |||
| 1393 | + | win.flush <- struct{}{} | |||
| 1394 | + | select { | |||
| 1395 | + | case <-gotOutput: | |||
| 1396 | + | case <-time.After(time.Second): | |||
| 1397 | + | t.Fatal("sync end did not flush deferred damage") | |||
| 1398 | + | } | |||
| 1399 | + | ||||
| 1400 | + | cancel() | |||
| 1401 | + | d.sessWg.Wait() | |||
| 1402 | + | } | |||
| 1403 | + | ||||
| 1404 | + | func TestPTYReaderSameReadSynchronizedUpdateStartEndFlushesImmediately(t *testing.T) { | |||
| 1405 | + | d := newTestDaemon(t, nil, stubClock{}) | |||
| 1406 | + | p := portsmocks.NewMockPTY(t) | |||
| 1407 | + | chunks := [][]byte{[]byte("\x1b[?2026hhello\x1b[?2026l")} | |||
| 1408 | + | p.EXPECT().Read(mock.Anything).RunAndReturn(func(buf []byte) (int, error) { | |||
| 1409 | + | if len(chunks) == 0 { | |||
| 1410 | + | return 0, io.EOF | |||
| 1411 | + | } | |||
| 1412 | + | n := copy(buf, chunks[0]) | |||
| 1413 | + | chunks = chunks[1:] | |||
| 1414 | + | return n, nil | |||
| 1415 | + | }) | |||
| 1416 | + | p.EXPECT().Close().Return(nil).Maybe() | |||
| 1417 | + | ||||
| 1418 | + | sctx, cancel := context.WithCancel(context.Background()) | |||
| 1419 | + | win := &window{pty: p, screen: vt.NewScreen(20, 5), dirty: make(chan struct{}, 1), flush: make(chan struct{}, 1), size: domain.Size{Cols: 20, Rows: 5}, ctx: sctx, cancel: cancel} | |||
| 1420 | + | sess := &session{id: "sync", name: "sync", windows: []*window{win}, ctx: sctx, cancel: cancel} | |||
| 1421 | + | d.sessions[sess.id] = sess | |||
| 1422 | + | d.sessWg.Add(1) | |||
| 1423 | + | d.ptyReader(sess, win) | |||
| 1424 | + | ||||
| 1425 | + | select { | |||
| 1426 | + | case <-win.dirty: | |||
| 1427 | + | t.Fatal("same-read synchronized update end should request flush, not dirty") | |||
| 1428 | + | default: | |||
| 1429 | + | } | |||
| 1430 | + | select { | |||
| 1431 | + | case <-win.flush: | |||
| 1432 | + | case <-time.After(time.Second): | |||
| 1433 | + | t.Fatal("same-read synchronized update end did not request immediate flush") | |||
| 1434 | + | } | |||
| 1435 | + | } | |||
| 1436 | + | ||||
| 1341 | func TestPTYReaderDefersDirtyDuringSynchronizedUpdateAndFlushesAtEnd(t *testing.T) { | 1437 | func TestPTYReaderDefersDirtyDuringSynchronizedUpdateAndFlushesAtEnd(t *testing.T) { | ||
| 1342 | d := newTestDaemon(t, nil, stubClock{}) | 1438 | d := newTestDaemon(t, nil, stubClock{}) | ||
| 1343 | p := portsmocks.NewMockPTY(t) | 1439 | p := portsmocks.NewMockPTY(t) | ||