65 lines
2.2 KiB
Go
65 lines
2.2 KiB
Go
package client
|
|
|
|
// unackedBuf holds the bytes a stream has sent but the peer has not yet
|
|
// credited — exactly the region a reattach may have to retransmit
|
|
// (PROTOCOL.md §7.5).
|
|
//
|
|
// It needs no cap of its own: credit is only granted as bytes reach the peer's
|
|
// terminal socket, so flow control already bounds the outstanding region to one
|
|
// window. That is what makes byte-exact resumption affordable at all.
|
|
//
|
|
// A read offset rather than a copy-down on every trim. Credit arrives once per
|
|
// half-window, and copying the live remainder each time would add a second
|
|
// per-byte copy to the whole send path; compacting only once the dead prefix
|
|
// dominates makes it amortized O(1).
|
|
type unackedBuf struct {
|
|
buf []byte
|
|
head int // bytes at the front already credited, awaiting reclamation
|
|
baseOff int64 // stream offset of buf[head]
|
|
}
|
|
|
|
// length is how many bytes are still outstanding.
|
|
func (u *unackedBuf) length() int { return len(u.buf) - u.head }
|
|
|
|
// base is the offset of the first byte still held.
|
|
func (u *unackedBuf) base() int64 { return u.baseOff }
|
|
|
|
// end is the offset one past the last byte sent.
|
|
func (u *unackedBuf) end() int64 { return u.baseOff + int64(u.length()) }
|
|
|
|
func (u *unackedBuf) append(p []byte) { u.buf = append(u.buf, p...) }
|
|
|
|
// advance drops everything the peer has credited up to off.
|
|
func (u *unackedBuf) advance(off int64) {
|
|
drop := int(off - u.baseOff)
|
|
if drop <= 0 {
|
|
return
|
|
}
|
|
if n := u.length(); drop > n {
|
|
drop = n // only reachable from a peer crediting bytes it was never sent
|
|
}
|
|
u.head += drop
|
|
u.baseOff += int64(drop)
|
|
switch {
|
|
case u.head == len(u.buf):
|
|
u.buf, u.head = u.buf[:0], 0 // fully drained: restart at the front
|
|
case u.head > len(u.buf)/2:
|
|
u.buf = append(u.buf[:0], u.buf[u.head:]...)
|
|
u.head = 0
|
|
}
|
|
}
|
|
|
|
// from returns the outstanding bytes at and after off, or nil when off falls
|
|
// outside what is still held — which means the peer reported an offset we can no
|
|
// longer satisfy, and the stream cannot be resumed.
|
|
func (u *unackedBuf) from(off int64) []byte {
|
|
skip := off - u.baseOff
|
|
if skip < 0 || skip > int64(u.length()) {
|
|
return nil
|
|
}
|
|
return u.buf[u.head+int(skip):]
|
|
}
|
|
|
|
// reset releases the buffer once a stream can no longer be resumed.
|
|
func (u *unackedBuf) reset() { u.buf, u.head = nil, 0 }
|