1package brokeripc
2
3import (
4 "io"
5 "sync"
6
7 "github.com/bnema/vev/internal/adapters/brokerwire"
8)
9
10// streamPipe is the private ordered, credit-controlled byte channel between one
11// broker wire session and the sessionwire framing of one logical stream.
12//
13// Each direction of a stream has a brokerwire.StreamWindowBytes credit window.
14// The sender spends brokerwire.StreamChunkCredit per data frame before writing
15// it and waits while it has too little, so a slow consumer slows its own
16// sender instead of failing the stream. The receiver returns the credit of
17// every frame its consumer has fully read with one StreamWindowUpdate per
18// batch. Because a conforming peer never exceeds the window, the wire reader
19// never blocks: it appends each frame to the queue, and a frame beyond the
20// granted credit is a peer violation (ErrStreamCredit) that settles only that
21// stream. Close releases the reader, a writer waiting for credit, and the
22// writer's peer promptly.
23type streamPipe struct {
24 mu sync.Mutex
25 queue []pipeChunk
26 closed bool
27 err error
28
29 // recvUsed is the credit held by queued frames, recvReturn the credit of
30 // fully read frames not yet granted back, and returnAt the batch size.
31 recvUsed uint64
32 recvReturn uint64
33 returnAt uint64
34 // sendCredit is what the peer still allows this side to send.
35 sendCredit uint64
36 creditWake chan struct{}
37
38 wake chan struct{}
39 closedCh chan struct{}
40
41 ceiling int
42 sendFrame func([]byte) error
43 grant func(uint64) error
44}
45
46// pipeChunk is one queued inbound frame and the credit it holds until it is
47// fully read.
48type pipeChunk struct {
49 data []byte
50 cost uint64
51}
52
53// newStreamPipe builds one stream carriage. Write splits outbound bytes into
54// frames of at most ceiling bytes, the negotiated stream chunk limit, and hands
55// each to sendFrame once its credit is reserved; grant returns consumed inbound
56// credit to the peer. Neither hook may be nil.
57func newStreamPipe(ceiling int, sendFrame func([]byte) error, grant func(uint64) error) *streamPipe {
58 window := brokerwire.StreamWindowBytes
59 return &streamPipe{
60 returnAt: streamCreditReturnAt(window),
61 sendCredit: window,
62 creditWake: make(chan struct{}),
63 wake: make(chan struct{}, 1),
64 closedCh: make(chan struct{}),
65 ceiling: ceiling,
66 sendFrame: sendFrame,
67 grant: grant,
68 }
69}
70
71// streamCreditReturnAt is the consumed credit a receiver batches before it
72// grants it back: half the window, lowered so a sender blocked on a maximum
73// frame is always released. A sender waits only once more than
74// window-StreamChunkCredit(max) is unreturned, so granting at or below that
75// point can never deadlock.
76func streamCreditReturnAt(window uint64) uint64 {
77 cost := brokerwire.StreamChunkCredit(int(brokerwire.MaxStreamChunkBytes))
78 return max(min(window/2, window-cost+1), 1)
79}
80
81// Read serves the next queued frame, blocking until one arrives or the pipe
82// closes. A partial copy retains the unread remainder for the next call. The
83// credit of each fully read frame is returned to the peer in batches.
84func (p *streamPipe) Read(b []byte) (int, error) {
85 if len(b) == 0 {
86 return 0, nil
87 }
88 for {
89 p.mu.Lock()
90 if len(p.queue) > 0 {
91 chunk := &p.queue[0]
92 n := copy(b, chunk.data)
93 var grant uint64
94 if n == len(chunk.data) {
95 p.recvUsed -= chunk.cost
96 p.recvReturn += chunk.cost
97 p.queue[0] = pipeChunk{}
98 p.queue = p.queue[1:]
99 if p.recvReturn >= p.returnAt && !p.closed {
100 grant, p.recvReturn = p.recvReturn, 0
101 }
102 } else {
103 chunk.data = chunk.data[n:]
104 }
105 p.mu.Unlock()
106 if grant > 0 {
107 // A failed grant means the connection itself is failing; its
108 // terminal path settles this stream, so there is nothing to retry.
109 _ = p.grant(grant)
110 }
111 return n, nil
112 }
113 if p.closed {
114 err := p.err
115 p.mu.Unlock()
116 if err != nil {
117 return 0, err
118 }
119 return 0, io.EOF
120 }
121 p.mu.Unlock()
122 select {
123 case <-p.wake:
124 case <-p.closedCh:
125 }
126 }
127}
128
129// Write sends b in order as frames under the chunk ceiling, waiting for the
130// peer's credit before each one: a peer that is not reading slows this writer
131// instead of losing the stream. It returns ErrStreamGone once the pipe reached
132// its terminal outcome.
133func (p *streamPipe) Write(b []byte) (int, error) {
134 if p.ceiling <= 0 {
135 return 0, ErrConfig
136 }
137 p.mu.Lock()
138 closed := p.closed
139 p.mu.Unlock()
140 if closed {
141 return 0, ErrStreamGone
142 }
143 for data := b; len(data) > 0; {
144 n := min(len(data), p.ceiling)
145 if err := p.reserve(n); err != nil {
146 return 0, err
147 }
148 if err := p.sendFrame(data[:n]); err != nil {
149 return 0, err
150 }
151 data = data[n:]
152 }
153 return len(b), nil
154}
155
156// reserve spends the send credit of one outbound frame of n bytes, waiting for
157// the peer to grant more when it is short. It returns ErrStreamGone once the
158// pipe closes, so a writer waiting for credit is always released.
159func (p *streamPipe) reserve(n int) error {
160 cost := brokerwire.StreamChunkCredit(n)
161 for {
162 p.mu.Lock()
163 if p.closed {
164 p.mu.Unlock()
165 return ErrStreamGone
166 }
167 if p.sendCredit >= cost {
168 p.sendCredit -= cost
169 p.mu.Unlock()
170 return nil
171 }
172 wake := p.creditWake
173 p.mu.Unlock()
174 select {
175 case <-wake:
176 case <-p.closedCh:
177 }
178 }
179}
180
181// granted adds credit the peer returned. Credit beyond the window is a peer
182// flow-control violation.
183func (p *streamPipe) granted(credit uint64) error {
184 p.mu.Lock()
185 defer p.mu.Unlock()
186 if p.closed {
187 return nil
188 }
189 if credit > brokerwire.StreamWindowBytes-p.sendCredit {
190 return ErrStreamCredit
191 }
192 p.sendCredit += credit
193 close(p.creditWake)
194 p.creditWake = make(chan struct{})
195 return nil
196}
197
198// deliver accepts one inbound frame from the connection reader. It returns
199// ErrStreamCredit when the peer sent beyond the credit it was granted and
200// ErrStreamGone once the pipe is closed; the caller settles the affected stream
201// in both cases. The pipe takes ownership of chunk: callers pass freshly
202// decoded bytes and never touch them again.
203func (p *streamPipe) deliver(chunk []byte) error {
204 if len(chunk) == 0 {
205 return nil
206 }
207 cost := brokerwire.StreamChunkCredit(len(chunk))
208 p.mu.Lock()
209 if p.closed {
210 p.mu.Unlock()
211 return ErrStreamGone
212 }
213 if p.recvUsed+p.recvReturn+cost > brokerwire.StreamWindowBytes {
214 p.mu.Unlock()
215 return ErrStreamCredit
216 }
217 p.queue = append(p.queue, pipeChunk{data: chunk, cost: cost})
218 p.recvUsed += cost
219 p.mu.Unlock()
220 select {
221 case p.wake <- struct{}{}:
222 default:
223 }
224 return nil
225}
226
227// closeWith settles the pipe exactly once with cause. A nil cause is an orderly
228// close; every blocked Read observes it promptly. On a nil cause, any frames
229// already queued are kept: Read serves them first and only returns io.EOF once
230// the queue is drained, so a reply that arrived just before an orderly peer
231// close is never discarded underneath its reader. An error cause is a failure:
232// the queue is discarded, since its content can no longer be trusted to reach
233// a consumer that is being torn down for cause.
234func (p *streamPipe) closeWith(cause error) {
235 p.mu.Lock()
236 if p.closed {
237 p.mu.Unlock()
238 return
239 }
240 p.closed = true
241 p.err = cause
242 if cause != nil {
243 p.queue = nil
244 p.recvUsed = 0
245 }
246 p.mu.Unlock()
247 close(p.closedCh)
248 select {
249 case p.wake <- struct{}{}:
250 default:
251 }
252}
253
254// Close settles the pipe as an orderly close.
255func (p *streamPipe) Close() error {
256 p.closeWith(nil)
257 return nil
258}