feat(daemon): bound unacknowledged Output bytes alongside the frame window
- commit
- e3ca8e16d97cc9b2ccecdd2a706239e64f68d65c
- author
- brice · · 2026-10-08 14:08 UTC
- committer
- brice · · 2026-10-08 15:05 UTC
- parent
- 3a23d13
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
6 files · +194 -11
M internal/protocol/session.go +7 -0
| @@ -33,6 +33,13 @@ const MaxOutputDataLen = (16 << 20) - 1 - 55 - (8 + 16 + 2 + 64 + 1 + 2 + 128 + | |||
| 33 | 33 | // MaxOutputWindow caps output states sent before client acknowledgement. | |
| 34 | 34 | const MaxOutputWindow = 8 | |
| 35 | 35 | ||
| 36 | + | // MaxOutputWindowBytes caps the Output data a daemon keeps unacknowledged per | |
| 37 | + | // attachment, state frames and side effects together. A full window of large | |
| 38 | + | // frames would otherwise overrun every bounded buffer between daemon and | |
| 39 | + | // terminal. The daemon may always send one frame when none is unacknowledged, | |
| 40 | + | // so a single frame up to MaxOutputDataLen still makes progress. | |
| 41 | + | const MaxOutputWindowBytes = 2 << 20 | |
| 42 | + | ||
| 36 | 43 | // ConnectionCapabilities describes transport behavior relevant to session | |
| 37 | 44 | // flow without naming a concrete carriage. | |
| 38 | 45 | type ConnectionCapabilities struct { | |
| @@ -33,6 +33,13 @@ const MaxOutputDataLen = (16 << 20) - 1 - 55 - (8 + 16 + 2 + 64 + 1 + 2 + 128 + | |||||
| 33 | // MaxOutputWindow caps output states sent before client acknowledgement. | 33 | // MaxOutputWindow caps output states sent before client acknowledgement. | ||
| 34 | const MaxOutputWindow = 8 | 34 | const MaxOutputWindow = 8 | ||
| 35 | 35 | ||||
| 36 | + | // MaxOutputWindowBytes caps the Output data a daemon keeps unacknowledged per | |||
| 37 | + | // attachment, state frames and side effects together. A full window of large | |||
| 38 | + | // frames would otherwise overrun every bounded buffer between daemon and | |||
| 39 | + | // terminal. The daemon may always send one frame when none is unacknowledged, | |||
| 40 | + | // so a single frame up to MaxOutputDataLen still makes progress. | |||
| 41 | + | const MaxOutputWindowBytes = 2 << 20 | |||
| 42 | + | ||||
| 36 | // ConnectionCapabilities describes transport behavior relevant to session | 43 | // ConnectionCapabilities describes transport behavior relevant to session | ||
| 37 | // flow without naming a concrete carriage. | 44 | // flow without naming a concrete carriage. | ||
| 38 | type ConnectionCapabilities struct { | 45 | type ConnectionCapabilities struct { | ||
M internal/usecase/daemon/attachment_output.go +99 -10
| @@ -23,14 +23,20 @@ import ( | |||
| 23 | 23 | // Callers serialize access with attachedClient.sendMu. A prepared transaction | |
| 24 | 24 | // must be sent or committed as a no-op before the next transaction is prepared. | |
| 25 | 25 | type attachmentOutput struct { | |
| 26 | - | renderer *renderer.Renderer | |
| 27 | - | epoch uint64 | |
| 28 | - | next uint64 | |
| 29 | - | acked uint64 | |
| 30 | - | generation uint64 | |
| 31 | - | maxOutstanding uint64 | |
| 32 | - | outstandingAtomic atomic.Uint64 | |
| 33 | - | maxOutstandingAtomic atomic.Uint64 | |
| 26 | + | renderer *renderer.Renderer | |
| 27 | + | epoch uint64 | |
| 28 | + | next uint64 | |
| 29 | + | acked uint64 | |
| 30 | + | generation uint64 | |
| 31 | + | maxOutstanding uint64 | |
| 32 | + | outstandingAtomic atomic.Uint64 | |
| 33 | + | maxOutstandingAtomic atomic.Uint64 | |
| 34 | + | // unackedBytes charges Output data sent and not yet acknowledged, keyed by | |
| 35 | + | // the state an ACK must reach to release it. Side-effect output carries no | |
| 36 | + | // state of its own, so it is charged to the next state. Guarded by the view | |
| 37 | + | // lock; unackedBytesAtomic mirrors the total for lock-free capacity reads. | |
| 38 | + | unackedBytes []unackedOutputBytes | |
| 39 | + | unackedBytesAtomic atomic.Uint64 | |
| 34 | 40 | forceSnapshot bool | |
| 35 | 41 | initialized bool | |
| 36 | 42 | viewPublication uint64 | |
| @@ -43,6 +49,51 @@ type attachmentOutput struct { | |||
| 43 | 49 | stateMu sync.Mutex | |
| 44 | 50 | } | |
| 45 | 51 | ||
| 52 | + | // unackedOutputBytes is the Output data released once the client acknowledges | |
| 53 | + | // state. | |
| 54 | + | type unackedOutputBytes struct { | |
| 55 | + | state uint64 | |
| 56 | + | bytes uint64 | |
| 57 | + | } | |
| 58 | + | ||
| 59 | + | // chargeBytesLocked records n bytes released once state is acknowledged. | |
| 60 | + | // Caller holds the view lock. | |
| 61 | + | func (s *attachmentOutput) chargeBytesLocked(state uint64, n int) { | |
| 62 | + | if n <= 0 { | |
| 63 | + | return | |
| 64 | + | } | |
| 65 | + | if last := len(s.unackedBytes) - 1; last >= 0 && s.unackedBytes[last].state == state { | |
| 66 | + | s.unackedBytes[last].bytes += uint64(n) | |
| 67 | + | } else { | |
| 68 | + | s.unackedBytes = append(s.unackedBytes, unackedOutputBytes{state: state, bytes: uint64(n)}) | |
| 69 | + | } | |
| 70 | + | s.unackedBytesAtomic.Add(uint64(n)) | |
| 71 | + | } | |
| 72 | + | ||
| 73 | + | // releaseBytesLocked drops every charge an acknowledgement of state covers. | |
| 74 | + | // Caller holds the view lock. | |
| 75 | + | func (s *attachmentOutput) releaseBytesLocked(state uint64) { | |
| 76 | + | released, kept := uint64(0), 0 | |
| 77 | + | for _, charge := range s.unackedBytes { | |
| 78 | + | if charge.state <= state { | |
| 79 | + | released += charge.bytes | |
| 80 | + | continue | |
| 81 | + | } | |
| 82 | + | s.unackedBytes[kept] = charge | |
| 83 | + | kept++ | |
| 84 | + | } | |
| 85 | + | clear(s.unackedBytes[kept:]) | |
| 86 | + | s.unackedBytes = s.unackedBytes[:kept] | |
| 87 | + | s.unackedBytesAtomic.Add(-released) | |
| 88 | + | } | |
| 89 | + | ||
| 90 | + | // resetBytesLocked forgets every charge: a new epoch acknowledges nothing the | |
| 91 | + | // old one sent. Caller holds the view lock. | |
| 92 | + | func (s *attachmentOutput) resetBytesLocked() { | |
| 93 | + | s.unackedBytes = s.unackedBytes[:0] | |
| 94 | + | s.unackedBytesAtomic.Store(0) | |
| 95 | + | } | |
| 96 | + | ||
| 46 | 97 | // reconfigureAttachmentOutput applies a replacement terminal's declared | |
| 47 | 98 | // capabilities to attachment-owned output state. Caller holds d.mu. | |
| 48 | 99 | func (d *Daemon) reconfigureAttachmentOutput(sess *session, ac *attachedClient, h protocol.Hello) { | |
| @@ -169,7 +220,10 @@ func (o *attachmentOutput) prepareFrame(d *Daemon, state *capturedRenderState, f | |||
| 169 | 220 | ansi.viewRevision = state.view.revision | |
| 170 | 221 | cursor := o.prepareCursorTail(desired, len(ansi.data) > 0) | |
| 171 | 222 | graphicsReset := reset || o.forceSnapshot || !o.initialized | |
| 172 | - | graphicsBudget := max(protocol.MaxOutputDataLen-len(ansi.data)-len(cursor.data), 0) | |
| 223 | + | // Images are optional decoration: they share the frame's data limit and the | |
| 224 | + | // attachment's unacknowledged byte window, and an upload that fits neither | |
| 225 | + | // waits for a later frame instead of overrunning buffers toward the client. | |
| 226 | + | graphicsBudget := max(min(protocol.MaxOutputDataLen, o.unackedBytesBudget())-len(ansi.data)-len(cursor.data), 0) | |
| 173 | 227 | preparedGraphics, err := graphicsOutputDataWithDaemonLimit(d, state, o.attachment, graphicsReset, graphicsBudget) | |
| 174 | 228 | if err != nil { | |
| 175 | 229 | return nil, err | |
| @@ -374,6 +428,7 @@ func (p *preparedOutput) send(data []byte, echoAck uint64, send func(protocol.Ou | |||
| 374 | 428 | p.boundary = protocol.UIReceipt{Epoch: p.epoch, State: p.next, ViewPublication: context.Publication, Outcome: protocol.UIReceiptProcessed} | |
| 375 | 429 | } | |
| 376 | 430 | p.stream.next = p.next | |
| 431 | + | p.stream.chargeBytesLocked(p.next, len(data)) | |
| 377 | 432 | p.stream.publishOutstanding() | |
| 378 | 433 | p.sent = true | |
| 379 | 434 | return nil | |
| @@ -510,6 +565,17 @@ func (s *attachmentOutput) sideEffectLocked(data []byte, echoAck uint64) (protoc | |||
| 510 | 565 | return marshalOutputState(data, epoch, 0, 0, echoAck, viewRevision, size, false, nil) | |
| 511 | 566 | } | |
| 512 | 567 | ||
| 568 | + | // sideEffectSentLocked charges a side effect that reached the transport. It has | |
| 569 | + | // no state of its own, so the ACK of the next state frame releases it: every | |
| 570 | + | // state frame is sent after it on the same ordered stream. Caller holds the view | |
| 571 | + | // lock that built output, so the epoch cannot have moved. | |
| 572 | + | func (s *attachmentOutput) sideEffectSentLocked(output protocol.Output) { | |
| 573 | + | if s == nil || output.Epoch != s.currentEpochLocked() { | |
| 574 | + | return | |
| 575 | + | } | |
| 576 | + | s.chargeBytesLocked(s.next+1, len(output.Data)) | |
| 577 | + | } | |
| 578 | + | ||
| 513 | 579 | func (s *attachmentOutput) ack(epoch, state uint64) bool { | |
| 514 | 580 | if s == nil { | |
| 515 | 581 | return false | |
| @@ -520,6 +586,7 @@ func (s *attachmentOutput) ack(epoch, state uint64) bool { | |||
| 520 | 586 | return false | |
| 521 | 587 | } | |
| 522 | 588 | s.acked = state | |
| 589 | + | s.releaseBytesLocked(state) | |
| 523 | 590 | s.publishOutstanding() | |
| 524 | 591 | return true | |
| 525 | 592 | } | |
| @@ -556,6 +623,7 @@ func (s *attachmentOutput) rebaseLocked() { | |||
| 556 | 623 | s.epoch = epoch + 1 | |
| 557 | 624 | s.next = 0 | |
| 558 | 625 | s.acked = 0 | |
| 626 | + | s.resetBytesLocked() | |
| 559 | 627 | s.generation++ | |
| 560 | 628 | s.publishOutstanding() | |
| 561 | 629 | s.maxOutstandingAtomic.Store(s.maxOutstanding) | |
| @@ -591,6 +659,11 @@ func (s *attachmentOutput) setWindow(window uint8) { | |||
| 591 | 659 | s.maxOutstandingAtomic.Store(s.maxOutstanding) | |
| 592 | 660 | } | |
| 593 | 661 | ||
| 662 | + | // atCapacity reports whether the next state frame must wait for an ACK: the | |
| 663 | + | // frame window is full, or unacknowledged Output data reached | |
| 664 | + | // protocol.MaxOutputWindowBytes while a state frame is in flight. With no state | |
| 665 | + | // frame in flight the next frame is always allowed, since only its ACK can | |
| 666 | + | // release bytes charged by side effects. | |
| 594 | 667 | func (s *attachmentOutput) atCapacity() bool { | |
| 595 | 668 | if s == nil { | |
| 596 | 669 | return false | |
| @@ -599,7 +672,23 @@ func (s *attachmentOutput) atCapacity() bool { | |||
| 599 | 672 | if maxOutstanding == 0 { | |
| 600 | 673 | maxOutstanding = s.maxOutstanding | |
| 601 | 674 | } | |
| 602 | - | return s.outstandingAtomic.Load() >= maxOutstanding | |
| 675 | + | outstanding := s.outstandingAtomic.Load() | |
| 676 | + | return outstanding >= maxOutstanding || | |
| 677 | + | (outstanding > 0 && s.unackedBytesAtomic.Load() >= protocol.MaxOutputWindowBytes) | |
| 678 | + | } | |
| 679 | + | ||
| 680 | + | // unackedBytesBudget is the Output data that may still be sent before the byte | |
| 681 | + | // window is full. It is advisory: a frame may exceed it when no state frame is | |
| 682 | + | // in flight. | |
| 683 | + | func (s *attachmentOutput) unackedBytesBudget() int { | |
| 684 | + | if s == nil { | |
| 685 | + | return protocol.MaxOutputDataLen | |
| 686 | + | } | |
| 687 | + | used := s.unackedBytesAtomic.Load() | |
| 688 | + | if used >= protocol.MaxOutputWindowBytes { | |
| 689 | + | return 0 | |
| 690 | + | } | |
| 691 | + | return int(protocol.MaxOutputWindowBytes - used) | |
| 603 | 692 | } | |
| 604 | 693 | ||
| 605 | 694 | // committedState reports the latest committed output state number. It takes | |
| @@ -23,14 +23,20 @@ import ( | |||||
| 23 | // Callers serialize access with attachedClient.sendMu. A prepared transaction | 23 | // Callers serialize access with attachedClient.sendMu. A prepared transaction | ||
| 24 | // must be sent or committed as a no-op before the next transaction is prepared. | 24 | // must be sent or committed as a no-op before the next transaction is prepared. | ||
| 25 | type attachmentOutput struct { | 25 | type attachmentOutput struct { | ||
| 26 | - | renderer *renderer.Renderer | 26 | + | renderer *renderer.Renderer |
| 27 | - | epoch uint64 | 27 | + | epoch uint64 |
| 28 | - | next uint64 | 28 | + | next uint64 |
| 29 | - | acked uint64 | 29 | + | acked uint64 |
| 30 | - | generation uint64 | 30 | + | generation uint64 |
| 31 | - | maxOutstanding uint64 | 31 | + | maxOutstanding uint64 |
| 32 | - | outstandingAtomic atomic.Uint64 | 32 | + | outstandingAtomic atomic.Uint64 |
| 33 | - | maxOutstandingAtomic atomic.Uint64 | 33 | + | maxOutstandingAtomic atomic.Uint64 |
| 34 | + | // unackedBytes charges Output data sent and not yet acknowledged, keyed by | |||
| 35 | + | // the state an ACK must reach to release it. Side-effect output carries no | |||
| 36 | + | // state of its own, so it is charged to the next state. Guarded by the view | |||
| 37 | + | // lock; unackedBytesAtomic mirrors the total for lock-free capacity reads. | |||
| 38 | + | unackedBytes []unackedOutputBytes | |||
| 39 | + | unackedBytesAtomic atomic.Uint64 | |||
| 34 | forceSnapshot bool | 40 | forceSnapshot bool | ||
| 35 | initialized bool | 41 | initialized bool | ||
| 36 | viewPublication uint64 | 42 | viewPublication uint64 | ||
| @@ -43,6 +49,51 @@ type attachmentOutput struct { | |||||
| 43 | stateMu sync.Mutex | 49 | stateMu sync.Mutex | ||
| 44 | } | 50 | } | ||
| 45 | 51 | ||||
| 52 | + | // unackedOutputBytes is the Output data released once the client acknowledges | |||
| 53 | + | // state. | |||
| 54 | + | type unackedOutputBytes struct { | |||
| 55 | + | state uint64 | |||
| 56 | + | bytes uint64 | |||
| 57 | + | } | |||
| 58 | + | ||||
| 59 | + | // chargeBytesLocked records n bytes released once state is acknowledged. | |||
| 60 | + | // Caller holds the view lock. | |||
| 61 | + | func (s *attachmentOutput) chargeBytesLocked(state uint64, n int) { | |||
| 62 | + | if n <= 0 { | |||
| 63 | + | return | |||
| 64 | + | } | |||
| 65 | + | if last := len(s.unackedBytes) - 1; last >= 0 && s.unackedBytes[last].state == state { | |||
| 66 | + | s.unackedBytes[last].bytes += uint64(n) | |||
| 67 | + | } else { | |||
| 68 | + | s.unackedBytes = append(s.unackedBytes, unackedOutputBytes{state: state, bytes: uint64(n)}) | |||
| 69 | + | } | |||
| 70 | + | s.unackedBytesAtomic.Add(uint64(n)) | |||
| 71 | + | } | |||
| 72 | + | ||||
| 73 | + | // releaseBytesLocked drops every charge an acknowledgement of state covers. | |||
| 74 | + | // Caller holds the view lock. | |||
| 75 | + | func (s *attachmentOutput) releaseBytesLocked(state uint64) { | |||
| 76 | + | released, kept := uint64(0), 0 | |||
| 77 | + | for _, charge := range s.unackedBytes { | |||
| 78 | + | if charge.state <= state { | |||
| 79 | + | released += charge.bytes | |||
| 80 | + | continue | |||
| 81 | + | } | |||
| 82 | + | s.unackedBytes[kept] = charge | |||
| 83 | + | kept++ | |||
| 84 | + | } | |||
| 85 | + | clear(s.unackedBytes[kept:]) | |||
| 86 | + | s.unackedBytes = s.unackedBytes[:kept] | |||
| 87 | + | s.unackedBytesAtomic.Add(-released) | |||
| 88 | + | } | |||
| 89 | + | ||||
| 90 | + | // resetBytesLocked forgets every charge: a new epoch acknowledges nothing the | |||
| 91 | + | // old one sent. Caller holds the view lock. | |||
| 92 | + | func (s *attachmentOutput) resetBytesLocked() { | |||
| 93 | + | s.unackedBytes = s.unackedBytes[:0] | |||
| 94 | + | s.unackedBytesAtomic.Store(0) | |||
| 95 | + | } | |||
| 96 | + | ||||
| 46 | // reconfigureAttachmentOutput applies a replacement terminal's declared | 97 | // reconfigureAttachmentOutput applies a replacement terminal's declared | ||
| 47 | // capabilities to attachment-owned output state. Caller holds d.mu. | 98 | // capabilities to attachment-owned output state. Caller holds d.mu. | ||
| 48 | func (d *Daemon) reconfigureAttachmentOutput(sess *session, ac *attachedClient, h protocol.Hello) { | 99 | func (d *Daemon) reconfigureAttachmentOutput(sess *session, ac *attachedClient, h protocol.Hello) { | ||
| @@ -169,7 +220,10 @@ func (o *attachmentOutput) prepareFrame(d *Daemon, state *capturedRenderState, f | |||||
| 169 | ansi.viewRevision = state.view.revision | 220 | ansi.viewRevision = state.view.revision | ||
| 170 | cursor := o.prepareCursorTail(desired, len(ansi.data) > 0) | 221 | cursor := o.prepareCursorTail(desired, len(ansi.data) > 0) | ||
| 171 | graphicsReset := reset || o.forceSnapshot || !o.initialized | 222 | graphicsReset := reset || o.forceSnapshot || !o.initialized | ||
| 172 | - | graphicsBudget := max(protocol.MaxOutputDataLen-len(ansi.data)-len(cursor.data), 0) | 223 | + | // Images are optional decoration: they share the frame's data limit and the |
| 224 | + | // attachment's unacknowledged byte window, and an upload that fits neither | |||
| 225 | + | // waits for a later frame instead of overrunning buffers toward the client. | |||
| 226 | + | graphicsBudget := max(min(protocol.MaxOutputDataLen, o.unackedBytesBudget())-len(ansi.data)-len(cursor.data), 0) | |||
| 173 | preparedGraphics, err := graphicsOutputDataWithDaemonLimit(d, state, o.attachment, graphicsReset, graphicsBudget) | 227 | preparedGraphics, err := graphicsOutputDataWithDaemonLimit(d, state, o.attachment, graphicsReset, graphicsBudget) | ||
| 174 | if err != nil { | 228 | if err != nil { | ||
| 175 | return nil, err | 229 | return nil, err | ||
| @@ -374,6 +428,7 @@ func (p *preparedOutput) send(data []byte, echoAck uint64, send func(protocol.Ou | |||||
| 374 | p.boundary = protocol.UIReceipt{Epoch: p.epoch, State: p.next, ViewPublication: context.Publication, Outcome: protocol.UIReceiptProcessed} | 428 | p.boundary = protocol.UIReceipt{Epoch: p.epoch, State: p.next, ViewPublication: context.Publication, Outcome: protocol.UIReceiptProcessed} | ||
| 375 | } | 429 | } | ||
| 376 | p.stream.next = p.next | 430 | p.stream.next = p.next | ||
| 431 | + | p.stream.chargeBytesLocked(p.next, len(data)) | |||
| 377 | p.stream.publishOutstanding() | 432 | p.stream.publishOutstanding() | ||
| 378 | p.sent = true | 433 | p.sent = true | ||
| 379 | return nil | 434 | return nil | ||
| @@ -510,6 +565,17 @@ func (s *attachmentOutput) sideEffectLocked(data []byte, echoAck uint64) (protoc | |||||
| 510 | return marshalOutputState(data, epoch, 0, 0, echoAck, viewRevision, size, false, nil) | 565 | return marshalOutputState(data, epoch, 0, 0, echoAck, viewRevision, size, false, nil) | ||
| 511 | } | 566 | } | ||
| 512 | 567 | ||||
| 568 | + | // sideEffectSentLocked charges a side effect that reached the transport. It has | |||
| 569 | + | // no state of its own, so the ACK of the next state frame releases it: every | |||
| 570 | + | // state frame is sent after it on the same ordered stream. Caller holds the view | |||
| 571 | + | // lock that built output, so the epoch cannot have moved. | |||
| 572 | + | func (s *attachmentOutput) sideEffectSentLocked(output protocol.Output) { | |||
| 573 | + | if s == nil || output.Epoch != s.currentEpochLocked() { | |||
| 574 | + | return | |||
| 575 | + | } | |||
| 576 | + | s.chargeBytesLocked(s.next+1, len(output.Data)) | |||
| 577 | + | } | |||
| 578 | + | ||||
| 513 | func (s *attachmentOutput) ack(epoch, state uint64) bool { | 579 | func (s *attachmentOutput) ack(epoch, state uint64) bool { | ||
| 514 | if s == nil { | 580 | if s == nil { | ||
| 515 | return false | 581 | return false | ||
| @@ -520,6 +586,7 @@ func (s *attachmentOutput) ack(epoch, state uint64) bool { | |||||
| 520 | return false | 586 | return false | ||
| 521 | } | 587 | } | ||
| 522 | s.acked = state | 588 | s.acked = state | ||
| 589 | + | s.releaseBytesLocked(state) | |||
| 523 | s.publishOutstanding() | 590 | s.publishOutstanding() | ||
| 524 | return true | 591 | return true | ||
| 525 | } | 592 | } | ||
| @@ -556,6 +623,7 @@ func (s *attachmentOutput) rebaseLocked() { | |||||
| 556 | s.epoch = epoch + 1 | 623 | s.epoch = epoch + 1 | ||
| 557 | s.next = 0 | 624 | s.next = 0 | ||
| 558 | s.acked = 0 | 625 | s.acked = 0 | ||
| 626 | + | s.resetBytesLocked() | |||
| 559 | s.generation++ | 627 | s.generation++ | ||
| 560 | s.publishOutstanding() | 628 | s.publishOutstanding() | ||
| 561 | s.maxOutstandingAtomic.Store(s.maxOutstanding) | 629 | s.maxOutstandingAtomic.Store(s.maxOutstanding) | ||
| @@ -591,6 +659,11 @@ func (s *attachmentOutput) setWindow(window uint8) { | |||||
| 591 | s.maxOutstandingAtomic.Store(s.maxOutstanding) | 659 | s.maxOutstandingAtomic.Store(s.maxOutstanding) | ||
| 592 | } | 660 | } | ||
| 593 | 661 | ||||
| 662 | + | // atCapacity reports whether the next state frame must wait for an ACK: the | |||
| 663 | + | // frame window is full, or unacknowledged Output data reached | |||
| 664 | + | // protocol.MaxOutputWindowBytes while a state frame is in flight. With no state | |||
| 665 | + | // frame in flight the next frame is always allowed, since only its ACK can | |||
| 666 | + | // release bytes charged by side effects. | |||
| 594 | func (s *attachmentOutput) atCapacity() bool { | 667 | func (s *attachmentOutput) atCapacity() bool { | ||
| 595 | if s == nil { | 668 | if s == nil { | ||
| 596 | return false | 669 | return false | ||
| @@ -599,7 +672,23 @@ func (s *attachmentOutput) atCapacity() bool { | |||||
| 599 | if maxOutstanding == 0 { | 672 | if maxOutstanding == 0 { | ||
| 600 | maxOutstanding = s.maxOutstanding | 673 | maxOutstanding = s.maxOutstanding | ||
| 601 | } | 674 | } | ||
| 602 | - | return s.outstandingAtomic.Load() >= maxOutstanding | 675 | + | outstanding := s.outstandingAtomic.Load() |
| 676 | + | return outstanding >= maxOutstanding || | |||
| 677 | + | (outstanding > 0 && s.unackedBytesAtomic.Load() >= protocol.MaxOutputWindowBytes) | |||
| 678 | + | } | |||
| 679 | + | ||||
| 680 | + | // unackedBytesBudget is the Output data that may still be sent before the byte | |||
| 681 | + | // window is full. It is advisory: a frame may exceed it when no state frame is | |||
| 682 | + | // in flight. | |||
| 683 | + | func (s *attachmentOutput) unackedBytesBudget() int { | |||
| 684 | + | if s == nil { | |||
| 685 | + | return protocol.MaxOutputDataLen | |||
| 686 | + | } | |||
| 687 | + | used := s.unackedBytesAtomic.Load() | |||
| 688 | + | if used >= protocol.MaxOutputWindowBytes { | |||
| 689 | + | return 0 | |||
| 690 | + | } | |||
| 691 | + | return int(protocol.MaxOutputWindowBytes - used) | |||
| 603 | } | 692 | } | ||
| 604 | 693 | ||||
| 605 | // committedState reports the latest committed output state number. It takes | 694 | // committedState reports the latest committed output state number. It takes | ||
M internal/usecase/daemon/attachment_output_test.go +74 -0
| @@ -519,3 +519,77 @@ func TestNormalizeOutputWindow(t *testing.T) { | |||
| 519 | 519 | require.Equal(t, tt.want, normalizeOutputWindow(tt.input)) | |
| 520 | 520 | } | |
| 521 | 521 | } | |
| 522 | + | ||
| 523 | + | // TestAttachmentOutputByteWindow covers the unacknowledged byte budget that | |
| 524 | + | // runs alongside the frame window: it fills on large frames and side effects, | |
| 525 | + | // never blocks the only frame that can release it, and an ACK or a new epoch | |
| 526 | + | // releases exactly what it covers. | |
| 527 | + | func TestAttachmentOutputByteWindow(t *testing.T) { | |
| 528 | + | const big = protocol.MaxOutputWindowBytes / 2 | |
| 529 | + | type step struct { | |
| 530 | + | frame int // state frame of this many data bytes | |
| 531 | + | sideEffect int // side effect of this many data bytes | |
| 532 | + | ack uint64 | |
| 533 | + | rebase bool | |
| 534 | + | } | |
| 535 | + | tests := []struct { | |
| 536 | + | name string | |
| 537 | + | steps []step | |
| 538 | + | wantBytes uint64 | |
| 539 | + | wantCapacity bool | |
| 540 | + | }{ | |
| 541 | + | {name: "small frames stay under the byte window", steps: []step{{frame: 100}, {frame: 100}}, wantBytes: 200}, | |
| 542 | + | {name: "large frames fill the byte window", steps: []step{{frame: big}, {frame: big}}, wantBytes: 2 * big, wantCapacity: true}, | |
| 543 | + | {name: "one oversized frame in flight blocks the next", steps: []step{{frame: protocol.MaxOutputWindowBytes + 1}}, wantBytes: protocol.MaxOutputWindowBytes + 1, wantCapacity: true}, | |
| 544 | + | {name: "side effects alone never block a state frame", steps: []step{{sideEffect: protocol.MaxOutputWindowBytes}}, wantBytes: protocol.MaxOutputWindowBytes}, | |
| 545 | + | {name: "side effects count once a frame is in flight", steps: []step{{frame: 100}, {sideEffect: protocol.MaxOutputWindowBytes}}, wantBytes: protocol.MaxOutputWindowBytes + 100, wantCapacity: true}, | |
| 546 | + | {name: "ack releases the frames it covers", steps: []step{{frame: big}, {frame: big}, {ack: 1}}, wantBytes: big}, | |
| 547 | + | {name: "a side effect is released by the next state's ack", steps: []step{{sideEffect: 50}, {frame: 100}, {ack: 1}}, wantBytes: 0}, | |
| 548 | + | {name: "a new epoch forgets every charge", steps: []step{{frame: big}, {frame: big}, {rebase: true}}, wantBytes: 0}, | |
| 549 | + | } | |
| 550 | + | for _, tt := range tests { | |
| 551 | + | t.Run(tt.name, func(t *testing.T) { | |
| 552 | + | stream := newOutputStateStream() | |
| 553 | + | stream.ack(1, 0) | |
| 554 | + | for _, s := range tt.steps { | |
| 555 | + | switch { | |
| 556 | + | case s.frame > 0: | |
| 557 | + | stream.lockView() | |
| 558 | + | stream.next++ | |
| 559 | + | stream.chargeBytesLocked(stream.next, s.frame) | |
| 560 | + | stream.publishOutstanding() | |
| 561 | + | stream.unlockView() | |
| 562 | + | case s.sideEffect > 0: | |
| 563 | + | out, err := stream.sideEffect(make([]byte, s.sideEffect), 0) | |
| 564 | + | require.NoError(t, err) | |
| 565 | + | stream.lockView() | |
| 566 | + | stream.sideEffectSentLocked(out) | |
| 567 | + | stream.unlockView() | |
| 568 | + | case s.ack > 0: | |
| 569 | + | require.True(t, stream.ack(stream.currentEpoch(), s.ack)) | |
| 570 | + | case s.rebase: | |
| 571 | + | stream.rebase() | |
| 572 | + | } | |
| 573 | + | } | |
| 574 | + | require.Equal(t, tt.wantBytes, stream.unackedBytesAtomic.Load()) | |
| 575 | + | require.Equal(t, tt.wantCapacity, stream.atCapacity()) | |
| 576 | + | }) | |
| 577 | + | } | |
| 578 | + | } | |
| 579 | + | ||
| 580 | + | func TestAttachmentOutputSentFrameChargesItsBytes(t *testing.T) { | |
| 581 | + | stream := newOutputStateStream() | |
| 582 | + | frame := renderer.NewFrame(3, 1) | |
| 583 | + | fillOutputStateRows(frame, []string{"abc"}) | |
| 584 | + | prepared, err := stream.prepareTestOutput(frame, nil, true) | |
| 585 | + | require.NoError(t, err) | |
| 586 | + | require.NoError(t, prepared.send(prepared.data, 0, outputFrameSender(func(wire.Envelope) error { return nil }))) | |
| 587 | + | require.Equal(t, uint64(len(prepared.data)), stream.unackedBytesAtomic.Load()) | |
| 588 | + | require.True(t, stream.ack(stream.currentEpoch(), stream.next)) | |
| 589 | + | require.Zero(t, stream.unackedBytesAtomic.Load()) | |
| 590 | + | ||
| 591 | + | failed, err := stream.prepareTestOutput(frame, nil, true) | |
| 592 | + | require.NoError(t, err) | |
| 593 | + | require.Error(t, failed.send(failed.data, 0, outputFrameSender(func(wire.Envelope) error { return errors.New("send failed") }))) | |
| 594 | + | require.Zero(t, stream.unackedBytesAtomic.Load(), "a frame that never left must not be charged") | |
| 595 | + | } | |
| @@ -519,3 +519,77 @@ func TestNormalizeOutputWindow(t *testing.T) { | |||||
| 519 | require.Equal(t, tt.want, normalizeOutputWindow(tt.input)) | 519 | require.Equal(t, tt.want, normalizeOutputWindow(tt.input)) | ||
| 520 | } | 520 | } | ||
| 521 | } | 521 | } | ||
| 522 | + | ||||
| 523 | + | // TestAttachmentOutputByteWindow covers the unacknowledged byte budget that | |||
| 524 | + | // runs alongside the frame window: it fills on large frames and side effects, | |||
| 525 | + | // never blocks the only frame that can release it, and an ACK or a new epoch | |||
| 526 | + | // releases exactly what it covers. | |||
| 527 | + | func TestAttachmentOutputByteWindow(t *testing.T) { | |||
| 528 | + | const big = protocol.MaxOutputWindowBytes / 2 | |||
| 529 | + | type step struct { | |||
| 530 | + | frame int // state frame of this many data bytes | |||
| 531 | + | sideEffect int // side effect of this many data bytes | |||
| 532 | + | ack uint64 | |||
| 533 | + | rebase bool | |||
| 534 | + | } | |||
| 535 | + | tests := []struct { | |||
| 536 | + | name string | |||
| 537 | + | steps []step | |||
| 538 | + | wantBytes uint64 | |||
| 539 | + | wantCapacity bool | |||
| 540 | + | }{ | |||
| 541 | + | {name: "small frames stay under the byte window", steps: []step{{frame: 100}, {frame: 100}}, wantBytes: 200}, | |||
| 542 | + | {name: "large frames fill the byte window", steps: []step{{frame: big}, {frame: big}}, wantBytes: 2 * big, wantCapacity: true}, | |||
| 543 | + | {name: "one oversized frame in flight blocks the next", steps: []step{{frame: protocol.MaxOutputWindowBytes + 1}}, wantBytes: protocol.MaxOutputWindowBytes + 1, wantCapacity: true}, | |||
| 544 | + | {name: "side effects alone never block a state frame", steps: []step{{sideEffect: protocol.MaxOutputWindowBytes}}, wantBytes: protocol.MaxOutputWindowBytes}, | |||
| 545 | + | {name: "side effects count once a frame is in flight", steps: []step{{frame: 100}, {sideEffect: protocol.MaxOutputWindowBytes}}, wantBytes: protocol.MaxOutputWindowBytes + 100, wantCapacity: true}, | |||
| 546 | + | {name: "ack releases the frames it covers", steps: []step{{frame: big}, {frame: big}, {ack: 1}}, wantBytes: big}, | |||
| 547 | + | {name: "a side effect is released by the next state's ack", steps: []step{{sideEffect: 50}, {frame: 100}, {ack: 1}}, wantBytes: 0}, | |||
| 548 | + | {name: "a new epoch forgets every charge", steps: []step{{frame: big}, {frame: big}, {rebase: true}}, wantBytes: 0}, | |||
| 549 | + | } | |||
| 550 | + | for _, tt := range tests { | |||
| 551 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 552 | + | stream := newOutputStateStream() | |||
| 553 | + | stream.ack(1, 0) | |||
| 554 | + | for _, s := range tt.steps { | |||
| 555 | + | switch { | |||
| 556 | + | case s.frame > 0: | |||
| 557 | + | stream.lockView() | |||
| 558 | + | stream.next++ | |||
| 559 | + | stream.chargeBytesLocked(stream.next, s.frame) | |||
| 560 | + | stream.publishOutstanding() | |||
| 561 | + | stream.unlockView() | |||
| 562 | + | case s.sideEffect > 0: | |||
| 563 | + | out, err := stream.sideEffect(make([]byte, s.sideEffect), 0) | |||
| 564 | + | require.NoError(t, err) | |||
| 565 | + | stream.lockView() | |||
| 566 | + | stream.sideEffectSentLocked(out) | |||
| 567 | + | stream.unlockView() | |||
| 568 | + | case s.ack > 0: | |||
| 569 | + | require.True(t, stream.ack(stream.currentEpoch(), s.ack)) | |||
| 570 | + | case s.rebase: | |||
| 571 | + | stream.rebase() | |||
| 572 | + | } | |||
| 573 | + | } | |||
| 574 | + | require.Equal(t, tt.wantBytes, stream.unackedBytesAtomic.Load()) | |||
| 575 | + | require.Equal(t, tt.wantCapacity, stream.atCapacity()) | |||
| 576 | + | }) | |||
| 577 | + | } | |||
| 578 | + | } | |||
| 579 | + | ||||
| 580 | + | func TestAttachmentOutputSentFrameChargesItsBytes(t *testing.T) { | |||
| 581 | + | stream := newOutputStateStream() | |||
| 582 | + | frame := renderer.NewFrame(3, 1) | |||
| 583 | + | fillOutputStateRows(frame, []string{"abc"}) | |||
| 584 | + | prepared, err := stream.prepareTestOutput(frame, nil, true) | |||
| 585 | + | require.NoError(t, err) | |||
| 586 | + | require.NoError(t, prepared.send(prepared.data, 0, outputFrameSender(func(wire.Envelope) error { return nil }))) | |||
| 587 | + | require.Equal(t, uint64(len(prepared.data)), stream.unackedBytesAtomic.Load()) | |||
| 588 | + | require.True(t, stream.ack(stream.currentEpoch(), stream.next)) | |||
| 589 | + | require.Zero(t, stream.unackedBytesAtomic.Load()) | |||
| 590 | + | ||||
| 591 | + | failed, err := stream.prepareTestOutput(frame, nil, true) | |||
| 592 | + | require.NoError(t, err) | |||
| 593 | + | require.Error(t, failed.send(failed.data, 0, outputFrameSender(func(wire.Envelope) error { return errors.New("send failed") }))) | |||
| 594 | + | require.Zero(t, stream.unackedBytesAtomic.Load(), "a frame that never left must not be charged") | |||
| 595 | + | } | |||
M internal/usecase/daemon/client.go +8 -1
| @@ -591,6 +591,9 @@ func (d *Daemon) boundedSendOutputErrTransport(ac *attachedClient, b []byte) (po | |||
| 591 | 591 | if err == nil { | |
| 592 | 592 | err = expected.transport.SendOutputSynchronous(output) | |
| 593 | 593 | } | |
| 594 | + | if err == nil { | |
| 595 | + | ac.output.sideEffectSentLocked(output) | |
| 596 | + | } | |
| 594 | 597 | return expected.transport, err | |
| 595 | 598 | } | |
| 596 | 599 | return d.boundedSendWith(expected.transport, func() error { | |
| @@ -605,7 +608,11 @@ func (d *Daemon) boundedSendOutputErrTransport(ac *attachedClient, b []byte) (po | |||
| 605 | 608 | if err != nil { | |
| 606 | 609 | return err | |
| 607 | 610 | } | |
| 608 | - | return expected.transport.SendOutput(output) | |
| 611 | + | if err := expected.transport.SendOutput(output); err != nil { | |
| 612 | + | return err | |
| 613 | + | } | |
| 614 | + | ac.output.sideEffectSentLocked(output) | |
| 615 | + | return nil | |
| 609 | 616 | }) | |
| 610 | 617 | } | |
| 611 | 618 | ||
| @@ -591,6 +591,9 @@ func (d *Daemon) boundedSendOutputErrTransport(ac *attachedClient, b []byte) (po | |||||
| 591 | if err == nil { | 591 | if err == nil { | ||
| 592 | err = expected.transport.SendOutputSynchronous(output) | 592 | err = expected.transport.SendOutputSynchronous(output) | ||
| 593 | } | 593 | } | ||
| 594 | + | if err == nil { | |||
| 595 | + | ac.output.sideEffectSentLocked(output) | |||
| 596 | + | } | |||
| 594 | return expected.transport, err | 597 | return expected.transport, err | ||
| 595 | } | 598 | } | ||
| 596 | return d.boundedSendWith(expected.transport, func() error { | 599 | return d.boundedSendWith(expected.transport, func() error { | ||
| @@ -605,7 +608,11 @@ func (d *Daemon) boundedSendOutputErrTransport(ac *attachedClient, b []byte) (po | |||||
| 605 | if err != nil { | 608 | if err != nil { | ||
| 606 | return err | 609 | return err | ||
| 607 | } | 610 | } | ||
| 608 | - | return expected.transport.SendOutput(output) | 611 | + | if err := expected.transport.SendOutput(output); err != nil { |
| 612 | + | return err | |||
| 613 | + | } | |||
| 614 | + | ac.output.sideEffectSentLocked(output) | |||
| 615 | + | return nil | |||
| 609 | }) | 616 | }) | ||
| 610 | } | 617 | } | ||
| 611 | 618 | ||||
M internal/usecase/daemon/clipboard.go +3 -0
| @@ -293,6 +293,9 @@ func (d *Daemon) boundedSendClipboardForward(item clipboardForward, ticket *atta | |||
| 293 | 293 | err = expected.transport.SendOutput(output) | |
| 294 | 294 | } | |
| 295 | 295 | } | |
| 296 | + | if err == nil { | |
| 297 | + | ac.output.sideEffectSentLocked(output) | |
| 298 | + | } | |
| 296 | 299 | if err != nil { | |
| 297 | 300 | ticket.reportTransportFailure(expected) | |
| 298 | 301 | } | |
| @@ -293,6 +293,9 @@ func (d *Daemon) boundedSendClipboardForward(item clipboardForward, ticket *atta | |||||
| 293 | err = expected.transport.SendOutput(output) | 293 | err = expected.transport.SendOutput(output) | ||
| 294 | } | 294 | } | ||
| 295 | } | 295 | } | ||
| 296 | + | if err == nil { | |||
| 297 | + | ac.output.sideEffectSentLocked(output) | |||
| 298 | + | } | |||
| 296 | if err != nil { | 299 | if err != nil { | ||
| 297 | ticket.reportTransportFailure(expected) | 300 | ticket.reportTransportFailure(expected) | ||
| 298 | } | 301 | } | ||
M internal/usecase/daemon/graphics_output.go +3 -0
| @@ -943,6 +943,9 @@ func (d *Daemon) cleanupAttachmentOutput(ac *attachedClient) error { | |||
| 943 | 943 | _ = ac.closeCapturedTransport(expected.transport) | |
| 944 | 944 | } | |
| 945 | 945 | } | |
| 946 | + | if sendErr == nil { | |
| 947 | + | ac.output.sideEffectSentLocked(output) | |
| 948 | + | } | |
| 946 | 949 | ac.output.unlockView() | |
| 947 | 950 | if sendErr == nil { | |
| 948 | 951 | prepared.commit() | |
| @@ -943,6 +943,9 @@ func (d *Daemon) cleanupAttachmentOutput(ac *attachedClient) error { | |||||
| 943 | _ = ac.closeCapturedTransport(expected.transport) | 943 | _ = ac.closeCapturedTransport(expected.transport) | ||
| 944 | } | 944 | } | ||
| 945 | } | 945 | } | ||
| 946 | + | if sendErr == nil { | |||
| 947 | + | ac.output.sideEffectSentLocked(output) | |||
| 948 | + | } | |||
| 946 | ac.output.unlockView() | 949 | ac.output.unlockView() | ||
| 947 | if sendErr == nil { | 950 | if sendErr == nil { | ||
| 948 | prepared.commit() | 951 | prepared.commit() | ||