orchard

  1package main
  2
  3import (
  4	"context"
  5	"sync"
  6	"time"
  7)
  8
  9// One question 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 page that
 17// sits on "searching" for ninety seconds because someone else 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, which is the common
 47// case: a closed tab should not hold the GPU for the person behind it.
 48func (q *Queue) Enter(ctx context.Context, onWait func(QueueState)) (release func(), ok bool) {
 49	t := &ticket{ready: make(chan struct{}, 1)}
 50
 51	q.mu.Lock()
 52	q.waiting = append(q.waiting, t)
 53	first := !q.running && len(q.waiting) == 1
 54	if first {
 55		q.running = true
 56		q.waiting = q.waiting[1:]
 57		t.done = true
 58	}
 59	q.mu.Unlock()
 60
 61	if first {
 62		return q.finish, true
 63	}
 64
 65	// Report the starting position immediately, so a waiting page says so
 66	// rather than showing nothing until the first tick.
 67	if onWait != nil {
 68		onWait(q.stateOf(t))
 69	}
 70
 71	tick := time.NewTicker(time.Second)
 72	defer tick.Stop()
 73	for {
 74		select {
 75		case <-t.ready:
 76			return q.finish, true
 77		case <-ctx.Done():
 78			q.drop(t)
 79			return func() {}, false
 80		case <-tick.C:
 81			if onWait != nil {
 82				onWait(q.stateOf(t))
 83			}
 84		}
 85	}
 86}
 87
 88func (q *Queue) stateOf(t *ticket) QueueState {
 89	q.mu.Lock()
 90	defer q.mu.Unlock()
 91	pos := 0
 92	for i, w := range q.waiting {
 93		if w == t {
 94			pos = i + 1
 95			break
 96		}
 97	}
 98	return QueueState{Position: pos, Ahead: pos - 1, Total: len(q.waiting)}
 99}
100
101// finish hands the slot to whoever is next.
102func (q *Queue) finish() {
103	q.mu.Lock()
104	defer q.mu.Unlock()
105	if len(q.waiting) == 0 {
106		q.running = false
107		return
108	}
109	next := q.waiting[0]
110	q.waiting = q.waiting[1:]
111	next.done = true
112	// Buffered, so this never blocks on a receiver that has gone away.
113	next.ready <- struct{}{}
114}
115
116// drop removes a ticket that gave up before its turn.
117//
118// There is a race worth naming: the context can end in the same moment finish
119// promotes this ticket, in which case the slot is already held by a caller that
120// will never use it and has to be handed on, or the person behind waits
121// forever.
122func (q *Queue) drop(t *ticket) {
123	q.mu.Lock()
124	for i, w := range q.waiting {
125		if w == t {
126			q.waiting = append(q.waiting[:i], q.waiting[i+1:]...)
127			q.mu.Unlock()
128			return
129		}
130	}
131	promoted := t.done
132	q.mu.Unlock()
133
134	if promoted {
135		q.finish()
136	}
137}
138
139// Depth is how many are waiting, for the page to show before anything is asked.
140func (q *Queue) Depth() (waiting int, running bool) {
141	q.mu.Lock()
142	defer q.mu.Unlock()
143	return len(q.waiting), q.running
144}