orchard

  1package main
  2
  3import (
  4	"context"
  5	"sync"
  6	"time"
  7)
  8
  9// One turn runs at a time.
 10//
 11// The model server is started with --parallel 1, because Qwen3.5's hybrid
 12// attention corrupts context checkpoints under multi-slot load, and there is one
 13// GPU behind it. Two people asking at once would otherwise interleave into one
 14// slot and both wait longer than if they had taken turns.
 15//
 16// So they take turns, and the waiting is shown rather than hidden: a tab that
 17// sits on "thinking" for ninety seconds because another one is ahead looks
 18// broken, while one that says it is second in line looks like a queue.
 19//
 20// Nothing about who is waiting is recorded. A ticket is a position and a
 21// channel, and it exists only while its request does.
 22type Queue struct {
 23	mu      sync.Mutex
 24	waiting []*ticket
 25	running bool
 26}
 27
 28type ticket struct {
 29	ready chan struct{}
 30	done  bool
 31}
 32
 33// QueueState is what a waiting page is told: how far back it is and how many
 34// are behind it. Positions only, never identities.
 35type QueueState struct {
 36	Position int `json:"position"` // 1 means next, 0 means running now
 37	Ahead    int `json:"ahead"`
 38	Total    int `json:"total"`
 39}
 40
 41func NewQueue() *Queue { return &Queue{} }
 42
 43// Enter joins the queue and blocks until this caller's turn, reporting position
 44// changes to onWait while it waits. The returned release must be called.
 45//
 46// A caller whose context is cancelled leaves the queue. Here that means a turn
 47// that was stopped or timed out, not a closed tab, since a turn outlives its
 48// tab and keeps its place.
 49func (q *Queue) Enter(ctx context.Context, onWait func(QueueState)) (release func(), ok bool) {
 50	t := &ticket{ready: make(chan struct{}, 1)}
 51
 52	q.mu.Lock()
 53	q.waiting = append(q.waiting, t)
 54	first := !q.running && len(q.waiting) == 1
 55	if first {
 56		q.running = true
 57		q.waiting = q.waiting[1:]
 58		t.done = true
 59	}
 60	q.mu.Unlock()
 61
 62	if first {
 63		return q.finish, true
 64	}
 65
 66	// Report the starting position immediately, so a waiting page says so
 67	// rather than showing nothing until the first tick.
 68	if onWait != nil {
 69		onWait(q.stateOf(t))
 70	}
 71
 72	tick := time.NewTicker(time.Second)
 73	defer tick.Stop()
 74	for {
 75		select {
 76		case <-t.ready:
 77			return q.finish, true
 78		case <-ctx.Done():
 79			q.drop(t)
 80			return func() {}, false
 81		case <-tick.C:
 82			if onWait != nil {
 83				onWait(q.stateOf(t))
 84			}
 85		}
 86	}
 87}
 88
 89func (q *Queue) stateOf(t *ticket) QueueState {
 90	q.mu.Lock()
 91	defer q.mu.Unlock()
 92	pos := 0
 93	for i, w := range q.waiting {
 94		if w == t {
 95			pos = i + 1
 96			break
 97		}
 98	}
 99	return QueueState{Position: pos, Ahead: pos - 1, Total: len(q.waiting)}
100}
101
102// finish hands the slot to whoever is next.
103func (q *Queue) finish() {
104	q.mu.Lock()
105	defer q.mu.Unlock()
106	if len(q.waiting) == 0 {
107		q.running = false
108		return
109	}
110	next := q.waiting[0]
111	q.waiting = q.waiting[1:]
112	next.done = true
113	// Buffered, so this never blocks on a receiver that has gone away.
114	next.ready <- struct{}{}
115}
116
117// drop removes a ticket that gave up before its turn.
118//
119// There is a race worth naming: the context can end in the same moment finish
120// promotes this ticket, in which case the slot is already held by a caller that
121// will never use it and has to be handed on, or the person behind waits
122// forever.
123func (q *Queue) drop(t *ticket) {
124	q.mu.Lock()
125	for i, w := range q.waiting {
126		if w == t {
127			q.waiting = append(q.waiting[:i], q.waiting[i+1:]...)
128			q.mu.Unlock()
129			return
130		}
131	}
132	promoted := t.done
133	q.mu.Unlock()
134
135	if promoted {
136		q.finish()
137	}
138}
139
140// Depth is how many are waiting, for the page to show before anything is asked.
141func (q *Queue) Depth() (waiting int, running bool) {
142	q.mu.Lock()
143	defer q.mu.Unlock()
144	return len(q.waiting), q.running
145}