1// Outbound fair scheduling.
2//
3// Scheduler is the outbound companion of the StreamEngine on one physical
4// daemonmux connection. It owns one bounded, independently cancellable FIFO
5// queue per admitted stream and orders the already-encoded application
6// envelopes a future writer drains. It performs no I/O, owns no goroutine, and
7// carries no framing, pump, socket, or listener: it only decides which stream's
8// next immutable envelope is written next.
9//
10// A frame is enqueued only for a stream the engine has admitted and not yet
11// terminalized (an opening or open stream), or, when no engine gate is
12// installed, for any nonzero physical identity the caller already admitted. A
13// closing stream (an orderly inbound Close with unread data) admits no new
14// frame even though it is not yet terminal. Two exceptions are deliberate.
15// Exactly one terminal Close or Reset is accepted for a known identity the
16// engine has already settled, bypassing normal admission, so an overflow or
17// peer-reset path can still signal the peer. And Refuse schedules exactly one
18// Mux Refused for a fresh identity the engine rejected before recording it, so
19// a refused Open is still answered; that refusal seals its own record. A
20// duplicate terminal, a duplicate refusal, and data for a settled, unadmitted,
21// or closing identity stay refused while the settling record is retained (see
22// below). Data frames are admitted only after the writer spent the stream's
23// send credit (see StreamEngine), so each stream's queued data never exceeds
24// its window and a slow peer slows the writer rather than filling the queue.
25// All streams together queue at most the negotiated aggregate bytes; a frame
26// that would exceed it discards exactly its own stream queue, frees that
27// budget, and leaves every sibling untouched. With the default ceilings the
28// stream windows fit inside the aggregate, so that bound is a safety net, not
29// a flow-control path.
30//
31// An overflowed stream is not sealed outright: the discard settles its data
32// but leaves its record open for exactly one terminal Close or Reset, so the
33// caller can still signal the peer that the stream was aborted. That terminal
34// is always retained: it is inserted without re-running the overflow check, so
35// it may exceed the aggregate data budget by at most one bounded terminal
36// envelope per live or retained overflowed stream. That bounded excess is
37// peer notification, never application data: the data bounds above still hold
38// for every non-terminal frame. Until that one terminal is enqueued the record
39// refuses every non-terminal frame with ErrSchedulerFull; the terminal seals
40// it, after which every later frame is refused (a duplicate terminal with
41// ErrSchedulerSettled).
42//
43// Across active streams the scheduler runs byte-deficit round robin with a
44// 64 KiB quantum (SchedulerQuantum): active streams are visited in turn, each
45// visit tops up the stream's deficit by one quantum unconditionally and caps
46// it at one quantum plus the largest envelope the codec admits, and its oldest
47// frame is written once the head fits the accumulated deficit. A head frame
48// larger than the quantum therefore accumulates deficit across visits instead
49// of stalling the ring, so every admitted frame up to the 1 MiB application
50// ceiling makes bounded progress (at most one quantum of accumulation per
51// visit) and can never spin the scheduler. Per-stream order is always
52// preserved: the scheduler only ever writes a stream's oldest queued frame, so
53// a Close never overtakes earlier accepted data of its own stream. Control
54// frames carry a small bounded priority (MaxMuxSchedulerControlBurst): they
55// may be written ahead of data frames of other streams, but the burst is
56// capped so control metadata can never starve data. A single hot stream
57// therefore never starves the others, and the schedule is deterministic: the
58// same enqueue sequence always yields the same dequeue order.
59//
60// Settled stream records are bounded: the scheduler retains at most
61// MaxRetiredSchedulerRecords sealed-or-overflowed, empty records and evicts
62// the oldest first, so a long-lived connection cannot accumulate an unbounded
63// record table. The ErrSchedulerSettled refusals above are guaranteed within
64// exactly that retention window. Once a record is evicted its identity falls
65// back to the engine gate: an identity the engine gate no longer tracks fails
66// that check and is reported unadmitted (ErrSchedulerUnadmitted), while a late
67// terminal that is still accepted is harmlessly discarded by the peer's
68// stream fencing, which ignores a Close or Reset for an already terminal
69// stream. An overflowed record that accepts its one terminal after being
70// listed keeps that queued frame past eviction, but draining the frame
71// re-lists the record so a later eviction reclaims it and retention stays
72// bounded.
73//
74// Reset settles one stream locally: it discards that stream's queued frames,
75// frees their bytes immediately, and refuses every later frame for that
76// identity. It never touches the physical connection or a sibling stream, so
77// cancelling one stream is isolated. Every dequeued OutboundEnvelope carries
78// the complete serialized directional envelope in Bytes, owned by the caller
79// and never reused or mutated by the scheduler, together with the committed
80// client or server message.
81package daemonmux
82
83import (
84 "errors"
85 "sync"
86)
87
88const (
89 // SchedulerQuantum is the byte-deficit round-robin quantum: 64 KiB, the
90 // largest opaque stream chunk the daemonmux codec admits. A stream with a
91 // head frame at or below the quantum is served once per visit; a larger
92 // frame accumulates one quantum per visit up to the overflow cap and is
93 // served once its deficit covers it.
94 SchedulerQuantum = int64(MaxMuxChunkBytes)
95
96 // MaxRetiredSchedulerRecords bounds the settled scheduler records retained
97 // so a duplicate terminal frame is still refused and a repeated Reset
98 // stays idempotent; that refusal is guaranteed only within this retention
99 // window, and an evicted identity may degrade to unadmitted. An overflowed,
100 // not-yet-sealed record is retained the same way. It mirrors the negotiated
101 // stream ceiling and is evicted oldest-first.
102 MaxRetiredSchedulerRecords = 128
103
104 // MaxMuxSchedulerControlBurst bounds how many control frames the
105 // scheduler writes ahead of data before it forces a data frame. It keeps
106 // control priority small and bounded so control metadata can never starve
107 // data.
108 MaxMuxSchedulerControlBurst = 4
109)
110
111// Outbound scheduling sentinels. They are typed so a caller can classify a
112// refused frame without matching on message text.
113var (
114 // ErrSchedulerUnadmitted reports an outbound frame for a physical stream
115 // the admission gate has not admitted or has already terminalized. An
116 // identity whose settled scheduler record was evicted, and which the engine
117 // gate no longer tracks, degrades to this refusal (see ErrSchedulerSettled).
118 ErrSchedulerUnadmitted = errors.New("daemonmux: stream is not admitted for scheduling")
119 // ErrSchedulerSettled reports an outbound frame for a stream the scheduler
120 // already sealed: a queued terminal Close or Reset, a local Reset, or a
121 // record that took the one terminal a queue overflow leaves open.
122 //
123 // The refusal is guaranteed only while the settled record is retained
124 // (bounded by MaxRetiredSchedulerRecords). Once the record is evicted the
125 // identity may instead be unadmitted (ErrSchedulerUnadmitted), or a late
126 // terminal may be accepted once more and harmlessly discarded by the peer's
127 // stream fencing.
128 ErrSchedulerSettled = errors.New("daemonmux: stream is settled for scheduling")
129 // ErrSchedulerFull reports an outbound frame that would exceed the
130 // aggregate byte bound. The offending stream's queue is discarded and its
131 // budget released. It also reports a non-terminal frame for a stream that already
132 // overflowed and is awaiting its one terminal Close or Reset: the data was
133 // discarded and the record is not yet sealed.
134 ErrSchedulerFull = errors.New("daemonmux: scheduler queue overflow")
135)
136
137// FrameClass classifies one outbound frame for scheduling priority. Control
138// frames may be written ahead of data frames under a bounded burst; data
139// frames carry the opaque stream payload.
140type FrameClass uint8
141
142const (
143 // FrameData is an opaque stream chunk: a daemonmux Data payload.
144 FrameData FrameClass = iota + 1
145 // FrameControl is connection or stream metadata: Open, Opened, Refused,
146 // Close, or Reset.
147 FrameControl
148)
149
150func (c FrameClass) String() string {
151 switch c {
152 case FrameData:
153 return "data"
154 case FrameControl:
155 return "control"
156 default:
157 return "unknown"
158 }
159}
160
161// EnvelopeDirection names the daemonmux direction of one outbound envelope.
162// DirectionClient is broker-to-daemon (the codec's client direction);
163// DirectionServer is daemon-to-broker (the codec's server direction).
164type EnvelopeDirection uint8
165
166const (
167 // DirectionClient is the broker-to-daemon direction.
168 DirectionClient EnvelopeDirection = iota + 1
169 // DirectionServer is the daemon-to-broker direction.
170 DirectionServer
171)
172
173func (d EnvelopeDirection) String() string {
174 switch d {
175 case DirectionClient:
176 return "client"
177 case DirectionServer:
178 return "server"
179 default:
180 return "unknown"
181 }
182}
183
184// OutboundEnvelope is one immutable outbound frame ready for a future writer.
185// Bytes is the complete serialized directional envelope, owned by the caller
186// and never reused or mutated by the scheduler. Exactly one of Client and
187// Server is non-nil and matches Direction; it is a scheduler-owned committed
188// copy, so a caller that mutates its own message after enqueue never changes a
189// dequeued value. Class is the scheduling class the frame was enqueued under.
190type OutboundEnvelope struct {
191 Physical PhysicalStreamID
192 Direction EnvelopeDirection
193 Class FrameClass
194 Bytes []byte
195 Client ClientMessage
196 Server ServerMessage
197}
198
199// schedFrame is one queued outbound frame. Its encoded bytes are the
200// immutable carriage the future writer drains; the committed message is
201// retained for the caller that wants the typed value.
202type schedFrame struct {
203 direction EnvelopeDirection
204 class FrameClass
205 bytes []byte
206 seals bool
207 client ClientMessage
208 server ServerMessage
209}
210
211// schedStream is one tracked outbound stream. It is the scheduling view of a
212// stream the engine admitted: the bounded FIFO queue, its charged bytes, the
213// byte-deficit account, whether it still participates in the active ring, and
214// whether it is sealed against further frames. Every field is guarded by the
215// owning Scheduler's lock.
216type schedStream struct {
217 physical PhysicalStreamID
218 queue []schedFrame
219 bytes int
220 deficit int64
221 sealed bool
222 // overflowed marks a stream whose queue overflowed and was discarded: it
223 // is settled for data but not yet sealed, so exactly one terminal Close or
224 // Reset may still be enqueued to signal the peer; that one terminal is
225 // always retained and is the only frame that may carry the aggregate
226 // charge past MaxMuxAggregateBytes (by at most one bounded envelope).
227 overflowed bool
228 active bool
229 // retired marks the record as listed for bounded eviction once it is
230 // sealed and empty; it is set at most once.
231 retired bool
232}
233
234// Scheduler is the outbound fair-scheduling authority of one physical
235// daemonmux connection. It is safe for concurrent use and owns no I/O. The
236// optional StreamEngine gate reads admission; a nil engine leaves admission to
237// the caller.
238type Scheduler struct {
239 mu sync.Mutex
240 engine *StreamEngine
241 streams map[PhysicalStreamID]*schedStream
242 active []*schedStream
243 retired []*schedStream
244 aggregate int
245 // controlBurst is the remaining number of control frames that may be
246 // written ahead of data. It refills to MaxMuxSchedulerControlBurst after
247 // every data frame.
248 controlBurst int
249
250 maxEnvelopeBytes uint64
251 maxChunkBytes uint64
252
253 // Bounds and quantum are constructed from the package defaults and are
254 // owned by the scheduler; they are fields so a test can tighten them.
255 maxAggregateBytes int
256 quantum int64
257}
258
259// NewSchedulerWithCeilings returns an empty outbound scheduler whose envelope,
260// chunk, and aggregate budgets enforce the given negotiated MuxCeilings. The
261// value is validated and copied, so the ceilings a scheduler enforces are
262// fixed for its lifetime. An invalid advertisement is refused with
263// ErrInvalidCeilings.
264func NewSchedulerWithCeilings(engine *StreamEngine, ceilings MuxCeilings) (*Scheduler, error) {
265 if err := ceilings.Validate(); err != nil {
266 return nil, err
267 }
268 return newScheduler(engine, ceilings), nil
269}
270
271// newScheduler builds a scheduler from already-validated ceilings. The
272// exported constructors validate first; NewScheduler delegates with the
273// package defaults, which are valid by construction.
274func newScheduler(engine *StreamEngine, ceilings MuxCeilings) *Scheduler {
275 return &Scheduler{
276 engine: engine,
277 streams: make(map[PhysicalStreamID]*schedStream, int(ceilings.MaxStreams)),
278 maxEnvelopeBytes: ceilings.MaxReceiveEnvelopeBytes,
279 maxChunkBytes: ceilings.StreamChunkLimit,
280 maxAggregateBytes: int(ceilings.MaxAggregateBytes),
281 quantum: SchedulerQuantum,
282 controlBurst: MaxMuxSchedulerControlBurst,
283 }
284}
285
286// EnqueueClient encodes and enqueues one broker-to-daemon envelope.
287func (s *Scheduler) EnqueueClient(message ClientMessage) error {
288 if s == nil || message == nil {
289 return ErrInvalidMessage
290 }
291 physical, class, err := clientFrame(message)
292 if err != nil {
293 return err
294 }
295 raw, err := EncodeClient(message, s.maxEnvelopeBytes, s.maxChunkBytes)
296 if err != nil {
297 return err
298 }
299 return s.enqueue(schedFrame{
300 direction: DirectionClient,
301 class: class,
302 bytes: raw,
303 seals: clientSeals(message),
304 client: cloneClientMessage(message),
305 }, physical)
306}
307
308// Refuse enqueues exactly one Mux Refused for a fresh inbound Open the engine
309// refused before it recorded any stream. It is the deliberate second admission
310// exception, alongside the one terminal Close or Reset a settled identity
311// accepts: a Refused is final for its physical identity (the daemon never sends
312// another frame for a stream it never admitted), so the record is sealed and
313// bounded by the same retired-record eviction as every other settled stream.
314//
315// The frame bypasses the engine gate because a rejected Open leaves no engine
316// record to confirm; the caller is the pump, and it only reaches here for an
317// identity above the engine's high-water mark that the engine refused. A
318// duplicate Refused, or any frame after the refusal drained, is refused with
319// ErrSchedulerSettled.
320func (s *Scheduler) Refuse(message Refused) error {
321 if s == nil {
322 return ErrInvalidMessage
323 }
324 physical, class, err := serverFrame(message)
325 if err != nil {
326 return err
327 }
328 raw, err := EncodeServer(message, s.maxEnvelopeBytes, s.maxChunkBytes)
329 if err != nil {
330 return err
331 }
332 if err := physical.Validate(); err != nil {
333 return err
334 }
335 frame := schedFrame{
336 direction: DirectionServer,
337 class: class,
338 bytes: raw,
339 seals: true,
340 server: cloneServerMessage(message),
341 }
342 s.mu.Lock()
343 defer s.mu.Unlock()
344 stream := s.streams[physical]
345 if stream != nil {
346 if stream.sealed || stream.overflowed {
347 return ErrSchedulerSettled
348 }
349 } else {
350 stream = &schedStream{physical: physical}
351 s.streams[physical] = stream
352 }
353 if s.overflowLocked(len(frame.bytes)) {
354 s.overflowLockedStream(stream)
355 return ErrSchedulerFull
356 }
357 s.insertLocked(stream, frame)
358 return nil
359}
360
361// EnqueueServer encodes and enqueues one daemon-to-broker envelope.
362func (s *Scheduler) EnqueueServer(message ServerMessage) error {
363 if s == nil || message == nil {
364 return ErrInvalidMessage
365 }
366 physical, class, err := serverFrame(message)
367 if err != nil {
368 return err
369 }
370 raw, err := EncodeServer(message, s.maxEnvelopeBytes, s.maxChunkBytes)
371 if err != nil {
372 return err
373 }
374 return s.enqueue(schedFrame{
375 direction: DirectionServer,
376 class: class,
377 bytes: raw,
378 seals: serverSeals(message),
379 server: cloneServerMessage(message),
380 }, physical)
381}
382
383// enqueue inserts one already-encoded frame under the scheduler lock. It
384// confirms admission, then charges the frame against both byte budgets before
385// the queue adopts it. A frame that would overflow any bound discards its own
386// stream queue and frees that budget without touching a sibling. A terminal
387// frame is the one exception to admission: it is accepted exactly once for a
388// known identity, so an overflow or reset path can still signal the peer even
389// after the engine settled the stream. Data for such an identity stays
390// refused, and a duplicate terminal is refused with ErrSchedulerSettled.
391func (s *Scheduler) enqueue(frame schedFrame, physical PhysicalStreamID) error {
392 if err := physical.Validate(); err != nil {
393 return ErrInvalidStream
394 }
395 s.mu.Lock()
396 defer s.mu.Unlock()
397 stream := s.streams[physical]
398 if stream != nil {
399 if stream.sealed {
400 // A duplicate terminal, and every frame after a terminal, is
401 // refused: the segment is already settled.
402 return ErrSchedulerSettled
403 }
404 if stream.overflowed {
405 // The queue overflowed and was discarded, but the record is not yet
406 // sealed: exactly one terminal Close or Reset may still signal the
407 // peer. Any other frame is refused with ErrSchedulerFull; the
408 // terminal is inserted without re-running the overflow check because
409 // the queue is already empty and its bytes released. It is therefore
410 // the one frame that may carry the aggregate charge past
411 // MaxMuxAggregateBytes, by at most one bounded terminal per
412 // overflowed stream, so peer notification is never lost.
413 if !frame.seals {
414 return ErrSchedulerFull
415 }
416 s.insertLocked(stream, frame)
417 return nil
418 }
419 if !frame.seals && !s.admittedLocked(physical) {
420 return ErrSchedulerUnadmitted
421 }
422 } else {
423 if !frame.seals {
424 if !s.admittedLocked(physical) {
425 return ErrSchedulerUnadmitted
426 }
427 } else if !s.knownLocked(physical) {
428 // A terminal frame still needs a known identity: it may bypass a
429 // settled admission, never a stream that was never admitted.
430 return ErrSchedulerUnadmitted
431 }
432 stream = &schedStream{physical: physical}
433 s.streams[physical] = stream
434 }
435 if s.overflowLocked(len(frame.bytes)) {
436 s.overflowLockedStream(stream)
437 return ErrSchedulerFull
438 }
439 s.insertLocked(stream, frame)
440 return nil
441}
442
443// insertLocked charges one already-admitted frame against both byte budgets
444// and appends it to the stream's queue, sealing the record when the frame is a
445// terminal Close or Reset and entering the stream into the active ring. The
446// caller has already tested every bound, except for the one terminal an
447// overflowed stream still retains: that frame is charged here deliberately, so
448// peer notification survives the overflow at the cost of at most one bounded
449// envelope above the aggregate data budget.
450func (s *Scheduler) insertLocked(stream *schedStream, frame schedFrame) {
451 n := len(frame.bytes)
452 s.aggregate += n
453 stream.bytes += n
454 stream.queue = append(stream.queue, frame)
455 if frame.seals {
456 stream.sealed = true
457 }
458 if !stream.active {
459 stream.active = true
460 s.active = append(s.active, stream)
461 }
462}
463
464// overflowLockedStream settles one stream whose frame exceeded a bound: it
465// discards the queued frames, frees their budget, and marks the record as
466// overflowed rather than sealed, so exactly one terminal Close or Reset may
467// still be enqueued. The record is listed for bounded eviction while it stays
468// empty and unsettled, so a long-lived connection cannot accumulate overflow
469// records.
470func (s *Scheduler) overflowLockedStream(stream *schedStream) {
471 s.discardLocked(stream)
472 stream.overflowed = true
473 s.listRetiredLocked(stream)
474}
475
476// Dequeue removes and returns the next ready outbound envelope, or false when
477// no stream has a queued frame. The returned Bytes are immutable and owned by
478// the caller.
479func (s *Scheduler) Dequeue() (OutboundEnvelope, bool) {
480 if s == nil {
481 return OutboundEnvelope{}, false
482 }
483 s.mu.Lock()
484 defer s.mu.Unlock()
485 if len(s.active) == 0 {
486 return OutboundEnvelope{}, false
487 }
488 if s.controlBurst > 0 {
489 if index := s.firstControlLocked(); index >= 0 {
490 s.controlBurst--
491 return s.serveIndexLocked(index), true
492 }
493 }
494 envelope := s.serveDRRLocked()
495 if envelope.Class == FrameData {
496 s.controlBurst = MaxMuxSchedulerControlBurst
497 }
498 return envelope, true
499}
500
501// Reset settles one stream locally: it discards the stream's queued frames,
502// frees their bytes immediately, and refuses every later frame for that
503// identity. It reports whether it settled a stream that was not already
504// settled, so a repeated cancel is idempotent. It never touches the physical
505// connection or a sibling stream.
506func (s *Scheduler) Reset(physical PhysicalStreamID) bool {
507 if s == nil {
508 return false
509 }
510 s.mu.Lock()
511 defer s.mu.Unlock()
512 stream := s.streams[physical]
513 if stream == nil {
514 if !s.admittedLocked(physical) {
515 return false
516 }
517 stream = &schedStream{physical: physical, sealed: true}
518 s.streams[physical] = stream
519 s.listRetiredLocked(stream)
520 return true
521 }
522 if stream.sealed && len(stream.queue) == 0 {
523 return false
524 }
525 s.discardLocked(stream)
526 stream.sealed = true
527 s.listRetiredLocked(stream)
528 return true
529}
530
531// AggregateBytes returns the outbound bytes currently charged across every
532// stream. Application data never exceeds MaxMuxAggregateBytes; the only
533// possible excess is the terminal an overflowed stream retains, at most one
534// bounded envelope per live or retained overflowed stream, which is what keeps
535// peer notification possible.
536func (s *Scheduler) AggregateBytes() int {
537 if s == nil {
538 return 0
539 }
540 s.mu.Lock()
541 defer s.mu.Unlock()
542 return s.aggregate
543}
544
545// QueuedFrames returns the number of frames queued for one stream.
546func (s *Scheduler) QueuedFrames(physical PhysicalStreamID) int {
547 if s == nil {
548 return 0
549 }
550 s.mu.Lock()
551 defer s.mu.Unlock()
552 if stream := s.streams[physical]; stream != nil {
553 return len(stream.queue)
554 }
555 return 0
556}
557
558// QueuedBytes returns the outbound bytes currently queued for one stream.
559func (s *Scheduler) QueuedBytes(physical PhysicalStreamID) int {
560 if s == nil {
561 return 0
562 }
563 s.mu.Lock()
564 defer s.mu.Unlock()
565 if stream := s.streams[physical]; stream != nil {
566 return stream.bytes
567 }
568 return 0
569}
570
571// Active returns the number of streams with a queued frame.
572func (s *Scheduler) Active() int {
573 if s == nil {
574 return 0
575 }
576 s.mu.Lock()
577 defer s.mu.Unlock()
578 return len(s.active)
579}
580
581// Pending reports whether any stream has a queued frame.
582func (s *Scheduler) Pending() bool { return s.Active() > 0 }
583
584// admittedLocked reports whether the admission gate permits scheduling for one
585// physical stream: a nil engine admits any nonzero identity, and an installed
586// engine admits a stream it tracks in opening or open state. A closing stream
587// (an orderly inbound Close with unread data) accepts no new outbound frame,
588// so it is not admitted; its one terminal Close or Reset still bypasses the
589// gate through knownLocked.
590func (s *Scheduler) admittedLocked(physical PhysicalStreamID) bool {
591 if physical == 0 {
592 return false
593 }
594 if s.engine == nil {
595 return true
596 }
597 status, ok := s.engine.Status(physical)
598 return ok && (status.State == StreamOpening || status.State == StreamOpen)
599}
600
601// knownLocked reports whether one physical stream identity is known to the
602// admission gate at all: a nil engine trusts any nonzero identity, and an
603// installed engine knows a stream it tracks in any state, including a settled
604// one. It backs the single terminal-frame exception: a terminal Close or Reset
605// may be enqueued once for a settled identity to signal the peer.
606func (s *Scheduler) knownLocked(physical PhysicalStreamID) bool {
607 if physical == 0 {
608 return false
609 }
610 if s.engine == nil {
611 return true
612 }
613 _, ok := s.engine.Status(physical)
614 return ok
615}
616
617// listRetiredLocked records one settled or overflowed, empty stream for
618// bounded eviction and evicts the oldest retired record past
619// MaxRetiredSchedulerRecords. An evicted identity falls back to the engine
620// gate for admission, so the ErrSchedulerSettled refusal is guaranteed only
621// within this retention window and a late terminal may instead be accepted
622// once more and harmlessly discarded by the peer's stream fencing. A record
623// is listed at most once and only while it is
624// sealed or overflowed and empty; a still-draining record stays until it
625// empties, and an overflowed record that later takes its one terminal is
626// sealed in place. A record that took that terminal after it was listed keeps
627// its queued frame past eviction instead of being forgotten: eviction clears
628// its retired flag, so draining its last frame re-lists it and a later
629// eviction reclaims the record rather than orphaning it in the stream table.
630func (s *Scheduler) listRetiredLocked(stream *schedStream) {
631 if stream.retired || !(stream.sealed || stream.overflowed) || len(stream.queue) != 0 {
632 return
633 }
634 stream.retired = true
635 s.retired = append(s.retired, stream)
636 for len(s.retired) > MaxRetiredSchedulerRecords {
637 evicted := s.retired[0]
638 s.retired = s.retired[1:]
639 if (evicted.sealed || evicted.overflowed) && len(evicted.queue) == 0 {
640 delete(s.streams, evicted.physical)
641 continue
642 }
643 // The terminal this record accepted after it was listed is still
644 // queued, so the record cannot be reclaimed yet. Clear the flag so
645 // draining its last frame re-lists it for a later eviction.
646 evicted.retired = false
647 }
648}
649
650// overflowLocked reports whether one frame would exceed the aggregate byte
651// bound.
652func (s *Scheduler) overflowLocked(n int) bool {
653 return s.aggregate+n > s.maxAggregateBytes
654}
655
656// discardLocked drops every queued frame of one stream and returns its bytes
657// to the aggregate budget. The stream record stays so a repeat cancel is
658// idempotent and a later frame is refused.
659func (s *Scheduler) discardLocked(stream *schedStream) {
660 s.aggregate -= stream.bytes
661 stream.bytes = 0
662 for i := range stream.queue {
663 stream.queue[i] = schedFrame{}
664 }
665 stream.queue = nil
666 if stream.active {
667 s.removeActiveLocked(stream)
668 }
669}
670
671// removeActiveLocked takes one stream out of the active ring.
672func (s *Scheduler) removeActiveLocked(stream *schedStream) {
673 for i, candidate := range s.active {
674 if candidate != stream {
675 continue
676 }
677 last := len(s.active) - 1
678 copy(s.active[i:], s.active[i+1:])
679 s.active[last] = nil
680 s.active = s.active[:last]
681 break
682 }
683 stream.active = false
684}
685
686// firstControlLocked returns the index of the first active stream whose oldest
687// frame is control, or -1 when none is. It prefers the head position in the
688// ring so the control priority stays round-robin.
689func (s *Scheduler) firstControlLocked() int {
690 for i, stream := range s.active {
691 if stream.queue[0].class == FrameControl {
692 return i
693 }
694 }
695 return -1
696}
697
698// serveIndexLocked writes the oldest frame of the active stream at index. The
699// stream's byte-deficit account is topped up and charged before the frame is
700// removed, and the stream rotates to the back of the ring.
701func (s *Scheduler) serveIndexLocked(index int) OutboundEnvelope {
702 stream := s.active[index]
703 s.topUpLocked(stream)
704 stream.deficit -= int64(len(stream.queue[0].bytes))
705 s.rotateToBackLocked(index)
706 return s.popHeadLocked(stream)
707}
708
709// serveDRRLocked runs one byte-deficit round-robin step: it visits active
710// streams in ring order, tops up and tests the head of each, and writes the
711// first head that fits its accumulated deficit. When no head fits in a pass,
712// every stream was topped up, so another pass is guaranteed to make progress.
713func (s *Scheduler) serveDRRLocked() OutboundEnvelope {
714 for {
715 for i, stream := range s.active {
716 s.topUpLocked(stream)
717 size := int64(len(stream.queue[0].bytes))
718 if size <= stream.deficit {
719 stream.deficit -= size
720 s.rotateToBackLocked(i)
721 return s.popHeadLocked(stream)
722 }
723 }
724 }
725}
726
727// topUpLocked adds one quantum to a stream's deficit unconditionally and caps
728// the result at one quantum plus the largest frame the codec admits
729// (MaxMuxApplicationBytes). Unconditional top-up is what guarantees progress:
730// a head frame larger than the quantum accumulates one quantum per visit until
731// it fits, while the cap bounds the deficit a stream can bank and keeps the
732// schedule fair.
733func (s *Scheduler) topUpLocked(stream *schedStream) {
734 stream.deficit += s.quantum
735 if maxDeficit := s.quantum + int64(min(s.maxEnvelopeBytes, MaxMuxApplicationBytes)); stream.deficit > maxDeficit {
736 stream.deficit = maxDeficit
737 }
738}
739
740// rotateToBackLocked moves the active stream at index to the back of the ring,
741// preserving the relative order of every other stream and allocating nothing.
742func (s *Scheduler) rotateToBackLocked(index int) {
743 last := len(s.active) - 1
744 if index == last {
745 return
746 }
747 stream := s.active[index]
748 copy(s.active[index:], s.active[index+1:])
749 s.active[last] = stream
750}
751
752// popHeadLocked removes and returns the head frame of a stream that the caller
753// already rotated to the back of the ring, releasing its bytes from both
754// budgets. An emptied stream leaves the ring.
755func (s *Scheduler) popHeadLocked(stream *schedStream) OutboundEnvelope {
756 frame := stream.queue[0]
757 stream.queue[0] = schedFrame{}
758 stream.queue = stream.queue[1:]
759 n := len(frame.bytes)
760 stream.bytes -= n
761 s.aggregate -= n
762 if len(stream.queue) == 0 {
763 last := len(s.active) - 1
764 s.active[last] = nil
765 s.active = s.active[:last]
766 stream.active = false
767 s.listRetiredLocked(stream)
768 }
769 return OutboundEnvelope{
770 Physical: stream.physical,
771 Direction: frame.direction,
772 Class: frame.class,
773 Bytes: frame.bytes,
774 Client: frame.client,
775 Server: frame.server,
776 }
777}
778
779// clientSeals reports whether a client message ends its stream for scheduling:
780// a Close or Reset frame is the last frame the stream may enqueue, so a later
781// frame is refused while the terminal frame drains in per-stream order.
782func clientSeals(message ClientMessage) bool {
783 switch message.(type) {
784 case Close, *Close, Reset, *Reset:
785 return true
786 default:
787 return false
788 }
789}
790
791// serverSeals reports whether a server message ends its stream for scheduling.
792func serverSeals(message ServerMessage) bool {
793 switch message.(type) {
794 case Close, *Close, Reset, *Reset:
795 return true
796 default:
797 return false
798 }
799}
800
801// cloneClientMessage returns a committed copy of one client message. Byte and
802// string slices the caller still owns are copied so a dequeued envelope never
803// aliases caller memory; every other field is a value type and is copied by
804// assignment. Unknown concrete types are returned unchanged (they were already
805// refused by the codec).
806func cloneClientMessage(message ClientMessage) ClientMessage {
807 switch m := message.(type) {
808 case Data:
809 return Data{Physical: m.Physical, Data: append([]byte(nil), m.Data...)}
810 case *Data:
811 if m == nil {
812 return m
813 }
814 committed := *m
815 committed.Data = append([]byte(nil), m.Data...)
816 return &committed
817 case Open:
818 return m.clone()
819 case *Open:
820 if m == nil {
821 return m
822 }
823 committed := m.clone()
824 return &committed
825 default:
826 return message
827 }
828}
829
830// clone returns an Open whose environment slice is owned by the scheduler.
831func (m Open) clone() Open {
832 m.Env = append([]string(nil), m.Env...)
833 return m
834}
835
836// cloneServerMessage returns a committed copy of one server message under the
837// same rule as cloneClientMessage: only the opaque data payload can alias
838// caller memory.
839func cloneServerMessage(message ServerMessage) ServerMessage {
840 switch m := message.(type) {
841 case Data:
842 return Data{Physical: m.Physical, Data: append([]byte(nil), m.Data...)}
843 case *Data:
844 if m == nil {
845 return m
846 }
847 committed := *m
848 committed.Data = append([]byte(nil), m.Data...)
849 return &committed
850 default:
851 return message
852 }
853}
854
855// clientFrame resolves the physical stream identity and scheduling class of one
856// broker-to-daemon message before it is encoded. Unknown concrete types fail
857// closed with ErrWrongDirection, mirroring the codec.
858func clientFrame(message ClientMessage) (PhysicalStreamID, FrameClass, error) {
859 switch m := message.(type) {
860 case Open:
861 return m.Ref.Physical, FrameControl, nil
862 case *Open:
863 if m == nil {
864 return 0, 0, ErrInvalidMessage
865 }
866 return clientFrame(*m)
867 case Data:
868 return m.Physical, FrameData, nil
869 case *Data:
870 if m == nil {
871 return 0, 0, ErrInvalidMessage
872 }
873 return clientFrame(*m)
874 case Close:
875 return m.Physical, FrameControl, nil
876 case *Close:
877 if m == nil {
878 return 0, 0, ErrInvalidMessage
879 }
880 return clientFrame(*m)
881 case Reset:
882 return m.Physical, FrameControl, nil
883 case *Reset:
884 if m == nil {
885 return 0, 0, ErrInvalidMessage
886 }
887 return clientFrame(*m)
888 case WindowUpdate:
889 return m.Physical, FrameControl, nil
890 case *WindowUpdate:
891 if m == nil {
892 return 0, 0, ErrInvalidMessage
893 }
894 return clientFrame(*m)
895 default:
896 return 0, 0, ErrWrongDirection
897 }
898}
899
900// serverFrame resolves the physical stream identity and scheduling class of one
901// daemon-to-broker message before it is encoded.
902func serverFrame(message ServerMessage) (PhysicalStreamID, FrameClass, error) {
903 switch m := message.(type) {
904 case Opened:
905 return m.Ref.Physical, FrameControl, nil
906 case *Opened:
907 if m == nil {
908 return 0, 0, ErrInvalidMessage
909 }
910 return serverFrame(*m)
911 case Refused:
912 return m.Ref.Physical, FrameControl, nil
913 case *Refused:
914 if m == nil {
915 return 0, 0, ErrInvalidMessage
916 }
917 return serverFrame(*m)
918 case Data:
919 return m.Physical, FrameData, nil
920 case *Data:
921 if m == nil {
922 return 0, 0, ErrInvalidMessage
923 }
924 return serverFrame(*m)
925 case Close:
926 return m.Physical, FrameControl, nil
927 case *Close:
928 if m == nil {
929 return 0, 0, ErrInvalidMessage
930 }
931 return serverFrame(*m)
932 case Reset:
933 return m.Physical, FrameControl, nil
934 case *Reset:
935 if m == nil {
936 return 0, 0, ErrInvalidMessage
937 }
938 return serverFrame(*m)
939 case WindowUpdate:
940 return m.Physical, FrameControl, nil
941 case *WindowUpdate:
942 if m == nil {
943 return 0, 0, ErrInvalidMessage
944 }
945 return serverFrame(*m)
946 default:
947 return 0, 0, ErrWrongDirection
948 }
949}