package main

import ( "container/heap" "fmt" "sync" "time" )

// Priority levels. Lower number = higher priority (this is a MIN-heap, // so the smallest value comes out first). const ( High = 0 Med = 1 Low = 2 )

func priorityName(p int) string { switch p { case High: return "HIGH" case Med: return "MED" case Low: return "LOW" } return "?" }

// Job is a unit of work with a priority. type Job struct { name string priority int seq int // submission order: FIFO tie-breaker within the same priority }

// ---------------------------------------------------------------------------- // MIN-HEAP via container/heap. // A heap is a binary tree where every parent <= its children, so the minimum // is always at the root. Push and Pop are O(log n). It is NOT fully sorted; // it only guarantees the min is on top, which is exactly what we need. // ----------------------------------------------------------------------------

type JobHeap []Job

func (h JobHeap) Len() int { return len(h) }

func (h JobHeap) Less(i, j int) bool { // Primary order: lower priority number first (HIGH before LOW). if h[i].priority != h[j].priority { return h[i].priority < h[j].priority } // Tie-breaker: earlier submission first (FIFO within the same priority). return h[i].seq < h[j].seq }

func (h JobHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }

func (h *JobHeap) Push(x any) { *h = append(*h, x.(Job)) }

func (h *JobHeap) Pop() any { old := *h n := len(old) job := old[n-1] *h = old[:n-1] return job }

// ---------------------------------------------------------------------------- // SCHEDULER: the heap holds the jobs, sync.Cond gives workers the // "sleep until there is work" behaviour that a channel would give for free // but a plain data structure does not have. // ----------------------------------------------------------------------------

type PriorityScheduler struct { mu sync.Mutex cond *sync.Cond jobs JobHeap seq int closing bool }

func NewPriorityScheduler() *PriorityScheduler { s := &PriorityScheduler{} s.cond = sync.NewCond(&s.mu) // the Cond is built on top of the same mutex return s }

// Submit pushes a job (ordered by priority) and wakes ONE sleeping worker. func (s *PriorityScheduler) Submit(name string, priority int) { s.mu.Lock() s.seq++ heap.Push(&s.jobs, Job{name: name, priority: priority, seq: s.seq}) s.mu.Unlock() s.cond.Signal() // wake a single waiting worker (not all of them) }

// Close asks workers to drain what's left and then exit. func (s *PriorityScheduler) Close() { s.mu.Lock() s.closing = true s.mu.Unlock() s.cond.Broadcast() // wake ALL workers so each can see closing and exit }

func (s *PriorityScheduler) worker(id int, wg *sync.WaitGroup) { defer wg.Done() for { s.mu.Lock() // IMPORTANT: Wait() inside a for-loop, not an if. On wake we must // re-check the condition, because another worker may have taken the // job between the Signal and us re-acquiring the lock. for s.jobs.Len() == 0 && !s.closing { s.cond.Wait() // atomically: unlock mu, park goroutine, re-lock on wake } if s.jobs.Len() == 0 && s.closing { s.mu.Unlock() return // nothing left and we're closing -> exit } job := heap.Pop(&s.jobs).(Job) // take the most-prioritary job, O(log n) s.mu.Unlock()# go-priority-scheduler

A minimal, runnable priority job scheduler in Go: a worker pool that always runs the most-prioritary pending job first, built on a min-heap for ordering and a sync.Cond to park idle workers without busy-looping.

I wrote this as a reference for my future self, to remember two things at once: how a min-heap serves work by priority, and the one situation where sync.Cond is actually the right tool instead of a channel.

The problem

A normal work queue is FIFO: first in, first out. But sometimes jobs aren't equal — a "recompute live odds, match starts in 10s" job must run before a "generate the monthly report" job, even if the report was submitted earlier. A priority scheduler serves the highest-priority job available whenever a worker becomes free, regardless of arrival order.

Why a min-heap

A heap is a binary tree where every parent <= its children, so the minimum is always at the root. Push and Pop are O(log n). It is not fully sorted — it only guarantees the min is on top, which is exactly what we need: "give me the most-prioritary job", nothing more. We map HIGH=0, MED=1, LOW=2 so the min-heap pops HIGH first. Go's standard container/heap provides it; we just implement Len/Less/Swap/Push/Pop. Less also uses a submission sequence as a tie-breaker, giving FIFO order within the same priority.

Why sync.Cond and not a channel

A channel is the usual way to hand work to goroutines, and it gives sleeping for free: job := <-ch parks the goroutine when the channel is empty and the runtime wakes it when something arrives. But a channel is a FIFO queue — it cannot order by priority.

We need a heap for priority, and a heap is just a data structure: it has no built-in way to block a goroutine until work exists. So when a worker pops an empty heap, there is no magic <- to sleep on. The naive fix is a busy-loop (pop, empty, pop, empty, ...) which burns 100% CPU on a core doing nothing.

sync.Cond fills exactly that gap. Workers call cond.Wait() when the heap is empty; Wait() atomically releases the mutex, parks the goroutine (0% CPU), and re-acquires the mutex on wake. Submit calls cond.Signal() to wake a single sleeping worker when a job arrives. This is the rare case where sync.Cond beats a channel: you need a custom data structure (a heap) and sleep/wake semantics, and a channel can't order by priority.

Two details that always come up

• Wait() goes inside a for, not an if. On wake, a worker must re-check the condition, because another worker may have taken the job between the Signal and this worker re-acquiring the lock. for heap.Len() == 0 — if still empty on wake, go back to sleep. This guards against lost races and spurious wakeups.
• Execute the job outside the lock. The mutex only protects access to the heap. Running the job while holding it would block every other worker for the whole duration of the job.

Alternative: 3 FIFO channels

If you have only a few fixed priority levels (say 3), you can skip the heap entirely and use one channel per level, with a worker select that tries HIGH first, then MED, then LOW. That recovers the free sleep/wake of channels. Use channels when priorities are few and fixed; use a heap when the priority is an arbitrary value (a deadline, a timestamp, a number from 1 to 1000) where you can't have one channel per possible value.

Distributed version

For a scheduler shared across many machines, the in-memory heap becomes a Redis Sorted Set (ZADD with the priority as score, ZPOPMIN to take the most-prioritary job atomically).

Run it

go run .
# the ordered writes / shared state are race-free:
go run -race .

Sample output

Submitting 7 jobs in scrambled priority order (no workers running yet):

Starting 1 worker -> it drains the heap in PRIORITY order (HIGH -> MED -> LOW):
[worker 1] running job-B  (HIGH)
[worker 1] running job-D  (HIGH)
[worker 1] running job-G  (HIGH)
[worker 1] running job-C  (MED)
[worker 1] running job-F  (MED)
[worker 1] running job-A  (LOW)
[worker 1] running job-E  (LOW)

Heap empty: the worker is now sleeping on cond.Wait() (0% CPU, no busy-loop).
Submitting one more HIGH job after a 1s pause...
[worker 1] running job-H  (HIGH)
Done.

Jobs were submitted mixed (Low, High, Med, High, Low, Med, High) but come out strictly by priority, and in submission order within each priority.

	// Execute OUTSIDE the lock so other workers aren't blocked while
	// this one runs a (possibly slow) job.
	fmt.Printf("[worker %d] running %-6s (%s)\n", id, job.name, priorityName(job.priority))
	time.Sleep(150 * time.Millisecond) // simulate work
}

}

func main() { s := NewPriorityScheduler()

// Submit everything first (no workers yet) so the priority ordering is
// easy to see when a worker drains the heap.
fmt.Println("Submitting 7 jobs in scrambled priority order (no workers running yet):")
s.Submit("job-A", Low)
s.Submit("job-B", High)
s.Submit("job-C", Med)
s.Submit("job-D", High)
s.Submit("job-E", Low)
s.Submit("job-F", Med)
s.Submit("job-G", High)

fmt.Println("\nStarting 1 worker -> it drains the heap in PRIORITY order (HIGH -> MED -> LOW):")
var wg sync.WaitGroup
wg.Add(1)
go s.worker(1, &wg)

time.Sleep(1500 * time.Millisecond) // let the worker drain the 7 jobs

// Heap is now empty: the worker is parked on cond.Wait(), using 0% CPU.
// No busy-loop asking "any work? any work?".
fmt.Println("\nHeap empty: the worker is now sleeping on cond.Wait() (0% CPU, no busy-loop).")
fmt.Println("Submitting one more HIGH job after a 1s pause...")
time.Sleep(1 * time.Second)
s.Submit("job-H", High) // Signal() wakes the sleeping worker

time.Sleep(500 * time.Millisecond)

s.Close() // tell the worker to finish and exit
wg.Wait()
fmt.Println("Done.")

}