package runtime import ( "errors" "sync" ) var ErrSendQueueFull = errors.New("client session send queue is full") type FramePriority uint8 const ( PriorityControl FramePriority = iota + 1 // Ping, Pong, Close, and session fencing. PriorityEssential // Lifecycle, acknowledgement, signal, and loss markers. PriorityData // Replayed output and ordinary durable data. ) type Frame struct { Priority FramePriority Payload []byte } type QueueOptions struct { MaxBytes uint64 ControlReserveBytes uint64 EssentialReserveBytes uint64 ControlBurst uint32 } func (options QueueOptions) Validate() error { if options.MaxBytes == 0 || options.ControlReserveBytes == 0 || options.EssentialReserveBytes == 0 || options.ControlReserveBytes+options.EssentialReserveBytes >= options.MaxBytes || options.ControlBurst == 0 { return errors.New("invalid client send queue options") } return nil } // SendQueue has no fallback goroutine or unbounded channel. Output cannot use // bytes reserved for essential state, and essential state cannot use the final // control reserve needed for heartbeat/close traffic. type SendQueue struct { mu sync.Mutex options QueueOptions bytes uint64 controlRun uint32 control []Frame essential []Frame data []Frame } func NewSendQueue(options QueueOptions) (*SendQueue, error) { if err := options.Validate(); err != nil { return nil, err } return &SendQueue{options: options}, nil } func (queue *SendQueue) Enqueue(frame Frame) error { if frame.Priority < PriorityControl || frame.Priority > PriorityData { return ErrSendQueueFull } copyFrame := Frame{Priority: frame.Priority, Payload: append([]byte(nil), frame.Payload...)} needed := uint64(len(copyFrame.Payload)) queue.mu.Lock() defer queue.mu.Unlock() limit := queue.limitForLocked(copyFrame.Priority) if queue.bytes > limit || needed > limit-queue.bytes { return ErrSendQueueFull } queue.bytes += needed switch copyFrame.Priority { case PriorityControl: queue.control = append(queue.control, copyFrame) case PriorityEssential: queue.essential = append(queue.essential, copyFrame) case PriorityData: queue.data = append(queue.data, copyFrame) } return nil } func (queue *SendQueue) Next() (Frame, bool) { queue.mu.Lock() defer queue.mu.Unlock() var selected *Frame if len(queue.control) > 0 && (queue.controlRun < queue.options.ControlBurst || (len(queue.essential) == 0 && len(queue.data) == 0)) { selected = &queue.control[0] queue.control = queue.control[1:] queue.controlRun++ } else if len(queue.essential) > 0 { selected = &queue.essential[0] queue.essential = queue.essential[1:] queue.controlRun = 0 } else if len(queue.data) > 0 { selected = &queue.data[0] queue.data = queue.data[1:] queue.controlRun = 0 } else if len(queue.control) > 0 { selected = &queue.control[0] queue.control = queue.control[1:] queue.controlRun++ } if selected == nil { return Frame{}, false } queue.bytes -= uint64(len(selected.Payload)) return *selected, true } func (queue *SendQueue) Bytes() uint64 { queue.mu.Lock() defer queue.mu.Unlock() return queue.bytes } func (queue *SendQueue) limitForLocked(priority FramePriority) uint64 { switch priority { case PriorityControl: return queue.options.MaxBytes case PriorityEssential: return queue.options.MaxBytes - queue.options.ControlReserveBytes default: return queue.options.MaxBytes - queue.options.ControlReserveBytes - queue.options.EssentialReserveBytes } }