| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "sync" |
| 6 | ) |
| 7 | |
| 8 | // The shared background-index budget stays at two concurrent build tasks. |
| 9 | // Slots are handed out by priority: user-requested history pages and recovery |
| 10 | // win before search, and both win before catalog/prefetch metadata work. |
| 11 | // Catalog workers wait at prefetch priority; saturated slots never discard |
| 12 | // pending session metadata work. |
| 13 | type rebuildPriority int |
| 14 | |
| 15 | const ( |
| 16 | rebuildPriorityUser rebuildPriority = iota |
| 17 | rebuildPrioritySearch |
| 18 | rebuildPriorityPrefetch |
| 19 | rebuildPriorityCount |
| 20 | ) |
| 21 | |
| 22 | type rebuildWaiter struct { |
| 23 | notify chan struct{} |
| 24 | granted bool |
| 25 | } |
| 26 | |
| 27 | type rebuildSlots struct { |
| 28 | capacity int |
| 29 | held int |
| 30 | mu sync.Mutex |
| 31 | queues [rebuildPriorityCount][]*rebuildWaiter |
| 32 | } |
| 33 | |
| 34 | func newRebuildSlots(capacity int) *rebuildSlots { |
| 35 | return &rebuildSlots{capacity: capacity} |
| 36 | } |
| 37 | |
| 38 | func (s *rebuildSlots) tryAcquire() bool { |
| 39 | s.mu.Lock() |
| 40 | defer s.mu.Unlock() |
| 41 | if s.held >= s.capacity { |
| 42 | return false |
| 43 | } |
| 44 | s.held++ |
| 45 | return true |
| 46 | } |
| 47 | |
| 48 | // acquire waits for a slot. When one frees, the highest-priority waiter wins; |
| 49 | // within one priority the earliest waiter wins. A granted slot must be |
| 50 | // returned through release, including when ctx was cancelled in the same |
| 51 | // select round that granted it. |
| 52 | func (s *rebuildSlots) acquire(ctx context.Context, prio rebuildPriority) error { |
| 53 | if prio < rebuildPriorityUser || prio >= rebuildPriorityCount { |
| 54 | prio = rebuildPriorityPrefetch |
| 55 | } |
| 56 | s.mu.Lock() |
| 57 | if s.held < s.capacity { |
| 58 | s.held++ |
| 59 | s.mu.Unlock() |
| 60 | return nil |
| 61 | } |
| 62 | w := &rebuildWaiter{notify: make(chan struct{}, 1)} |
| 63 | s.queues[prio] = append(s.queues[prio], w) |
| 64 | s.mu.Unlock() |
| 65 | select { |
| 66 | case <-w.notify: |
| 67 | return nil |
| 68 | case <-ctx.Done(): |
| 69 | s.mu.Lock() |
| 70 | granted := w.granted |
| 71 | if !granted { |
| 72 | queue := s.queues[prio] |
| 73 | for i, candidate := range queue { |
| 74 | if candidate == w { |
| 75 | s.queues[prio] = append(queue[:i:i], queue[i+1:]...) |
| 76 | break |
| 77 | } |
| 78 | } |
| 79 | } |
| 80 | s.mu.Unlock() |
| 81 | if granted { |
| 82 | return nil |
| 83 | } |
| 84 | return ctx.Err() |
| 85 | } |
| 86 | } |
| 87 | |
| 88 | func (s *rebuildSlots) release() { |
| 89 | s.mu.Lock() |
| 90 | s.held-- |
| 91 | if s.held < 0 { |
| 92 | s.held = 0 |
| 93 | } |
| 94 | for offset := range int(rebuildPriorityCount - rebuildPriorityUser) { |
| 95 | prio := rebuildPriorityUser + rebuildPriority(offset) |
| 96 | if len(s.queues[prio]) == 0 { |
| 97 | continue |
| 98 | } |
| 99 | w := s.queues[prio][0] |
| 100 | s.queues[prio] = s.queues[prio][1:] |
| 101 | w.granted = true |
| 102 | w.notify <- struct{}{} |
| 103 | break |
| 104 | } |
| 105 | s.mu.Unlock() |
| 106 | } |
| 107 |