crates/gunnar-ssh/src/pipeline.rs
//! **The standing pipeline: warm machinery per class, and an operation is data.**
//!
//! # What this replaces
//!
//! Today a git operation builds its own world. It spawns what it needs,
//! allocates what it needs, opens what it needs, and throws all of it away when
//! the client disconnects. That is per-request *initialisation*, and it is the
//! largest avoidable cost in the server:
//!
//! MEASURED on the reference desktop 2026-08-14, a `linux.git` clone —
//!
//! | | |
//! |---|---:|
//! | peak RSS, gunnar | **2314 MB** |
//! | peak RSS, gitea, identical clone and identical wire bytes | 122 MB |
//! | cycles spent copying bytes that never needed to move | **~20%** |
//! | `kernel_init_pages` — the kernel ZEROING pages for per-op allocations | 4.7% of the top profile |
//!
//! (`vs_forge_clone`; and 39
//! `perf record` captures of `gunnar serve` during the mirror clone.)
//!
//! A pipeline that is already standing pays none of that. The threads exist, the
//! buffers exist, the archive is already mapped; a request adds a **descriptor**
//! to a queue and a worker that is already awake picks it up.
//!
//! ```text
//! at BOOT, once, for the life of the process:
//! pipeline[LsRemote] ─┐
//! pipeline[Fetch] ├─ N workers parked on a condvar │ arena each
//! pipeline[Clone] │
//! pipeline[Push] │
//! pipeline[Gc] ─┘
//!
//! per request:
//! classify ─► admission gate ─► submit(descriptor) ─► a parked worker wakes
//! │
//! └─ DENY "server busy", before a single allocation
//! ```
//!
//! # An operation is DATA, not a program
//!
//! Two `Clone` operations do not start two clone engines. They are two
//! descriptors in one queue, drained by the fan that is already running. This is
//! the distinction the whole design rests on, and it is why `submit` takes a
//! value rather than a closure that captures a world.
//!
//! # Why a condvar and not a channel crate
//!
//! No new dependency, and the shape is the point: a worker parked in
//! [`Condvar::wait`] is a thread in `futex_do_wait`, which is exactly what a
//! hang dump of the live server already showed — **42 threads, 40 of them
//! parked on 40 distinct futexes**. The warm fan is not a new idea being
//! introduced here; it already exists in the process. What is missing is
//! anything handing it work, which is what this module is.
//!
//! # What this module does NOT do
//!
//! It does not make one operation faster. A single `linux.git` clone gets one
//! slot and one worker, and it will take exactly as long as it took before —
//! the 25 minutes measured for that clone is serial work *inside* the operation,
//! and only the engine's own fan-out (znippy's `emit_pack` phase 1) touches
//! that. This module is concurrency **across** operations. Conflating the two
//! is how a "pipeline" gets built and then measured as having changed nothing.
use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::JoinHandle;
use crate::governor::OpClass;
/// How many workers a class's fan is built with.
///
/// Deliberately per-class: `LsRemote` is answered in milliseconds and wants
/// enough workers that a burst never queues, while `Gc` is long, rare, and
/// wants at most a couple so that maintenance cannot crowd out serving.
pub fn default_workers(class: OpClass, cores: usize) -> usize {
match class {
// ── I/O-BOUND, so sized for CONCURRENCY and not for cores ───────────
//
// A git transfer spends its life waiting on a socket, not computing.
// Sizing these fans as if they were CPU-bound is a mistake I made and
// the bench caught: at `(cores/2).clamp(2,8)` a 32-core box served
// `multiuser_scaling`'s 102 concurrent clients with EIGHT workers, and
// **16 of 102 clients did not complete** — not refused, which they
// would have understood, but queued behind eight streaming packs until
// they timed out. Zero `transfer.refused` in the whole run, which is
// what made it look like a mystery rather than a queue.
//
// That was strictly worse than what it replaced: tokio's blocking pool
// would have spawned up to 512 threads and served all 102.
//
// Shedding overload is the ADMISSION GATE's job — CoDel sees the queue
// standing and refuses, in the client's own protocol. The fan's job is
// only to stop unbounded thread growth. So it is wide enough that
// ordinary concurrency is never artificially serialised, and the
// governor decides when there is too much.
OpClass::LsRemote => (cores * 4).clamp(16, 256),
OpClass::Fetch => (cores * 4).clamp(16, 256),
OpClass::Clone => (cores * 4).clamp(16, 256),
OpClass::Push => (cores * 4).clamp(16, 256),
// Maintenance yields to serving, and IS bounded by CPU: a repack is
// real work, not a stream. Two, so one long repack cannot crowd out the
// other repositories.
OpClass::Gc => 2,
}
}
/// What a submission was refused for. Both are "server busy" to the client; the
/// distinction is for the operator reading logs.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SubmitError {
/// The class's queue is full. Backpressure working as designed.
QueueFull,
/// The pipeline is shutting down and will not start new work.
ShuttingDown,
}
impl std::fmt::Display for SubmitError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
SubmitError::QueueFull => {
f.write_str("server busy: the queue for this operation class is full")
}
SubmitError::ShuttingDown => f.write_str("server busy: shutting down"),
}
}
}
struct Queue<T> {
items: Mutex<VecDeque<T>>,
/// Workers park here. A parked worker is a thread in `futex_do_wait`.
ready: Condvar,
stopping: AtomicBool,
depth: usize,
/// Test-only: runs in [`Queue::pop_blocking`] with the lock held, after `stopping` was
/// read and before the worker waits — the window a lost wake-up lives in.
#[cfg(test)]
before_wait: std::sync::OnceLock<Box<dyn Fn() + Send + Sync>>,
}
impl<T> Queue<T> {
fn push(&self, item: T) -> Result<usize, SubmitError> {
let mut q = self.items.lock().expect("pipeline queue poisoned");
// Read under the lock `shutdown` sets it under: nothing is queued after the
// workers were told to stop.
if self.stopping.load(Ordering::Acquire) {
return Err(SubmitError::ShuttingDown);
}
if q.len() >= self.depth {
return Err(SubmitError::QueueFull);
}
q.push_back(item);
let queued = q.len();
drop(q);
// One waiter, not all: waking every worker for one item is a thundering
// herd that costs more than the work.
self.ready.notify_one();
Ok(queued)
}
fn pop_blocking(&self) -> Option<T> {
let mut q = self.items.lock().expect("pipeline queue poisoned");
loop {
if let Some(item) = q.pop_front() {
return Some(item);
}
if self.stopping.load(Ordering::Acquire) {
return None;
}
#[cfg(test)]
if let Some(hook) = self.before_wait.get() {
hook();
}
q = self.ready.wait(q).expect("pipeline queue poisoned");
}
}
}
/// One class's standing fan: N workers, built once, parked until there is work.
pub struct Pipeline<T: Send + 'static> {
class: OpClass,
queue: Arc<Queue<T>>,
workers: Mutex<Vec<JoinHandle<()>>>,
/// Operations that have been handed to a worker. Applied output: it is the
/// only thing that distinguishes "the fan ran it" from "the fan exists".
served: Arc<AtomicU64>,
}
impl<T: Send + 'static> Pipeline<T> {
/// Build the fan and START it. Threads are spawned HERE, at boot — never on
/// the request path. That is the entire point of the type.
pub fn start<F>(class: OpClass, workers: usize, depth: usize, run: F) -> Pipeline<T>
where
F: Fn(T) + Send + Sync + 'static,
{
let queue = Arc::new(Queue {
items: Mutex::new(VecDeque::with_capacity(depth)),
ready: Condvar::new(),
stopping: AtomicBool::new(false),
depth,
#[cfg(test)]
before_wait: std::sync::OnceLock::new(),
});
let served = Arc::new(AtomicU64::new(0));
let run = Arc::new(run);
let mut handles = Vec::with_capacity(workers);
for i in 0..workers {
let q = Arc::clone(&queue);
let run = Arc::clone(&run);
let served = Arc::clone(&served);
let name = format!("gunnar-{class:?}-{i}").to_lowercase();
let h = std::thread::Builder::new()
.name(name)
.spawn(move || {
while let Some(item) = q.pop_blocking() {
served.fetch_add(1, Ordering::Relaxed);
run(item);
}
})
.expect("spawning a pipeline worker at boot");
handles.push(h);
}
Pipeline {
class,
queue,
workers: Mutex::new(handles),
served,
}
}
/// **The same fan, but each worker owns an ARENA built once at boot.**
///
/// This is the half of "zero initialisation" the plain [`start`](Self::start)
/// does not give you. A standing fan removes the thread spawn; it does not
/// remove the buffer. If the work still does `Vec::with_capacity` per
/// operation, the kernel still zeroes fresh pages for it — which is
/// `kernel_init_pages`, **4.7% of the top profile** on a `linux.git` clone,
/// the single largest kernel cost measured there.
///
/// `make` runs **once per worker, at boot**. `run` receives `&mut A`, so the
/// buffer that was grown for the first operation is still there, at capacity,
/// for the ten-thousandth. The caller is responsible for `clear()`-ing
/// rather than reallocating — `clear` keeps capacity, which is the whole
/// point, and is why this takes `&mut A` instead of handing back an owned
/// buffer per call.
#[allow(clippy::type_complexity)]
pub fn start_with_arena<A, M, F>(
class: OpClass,
workers: usize,
depth: usize,
make: M,
run: F,
) -> Pipeline<T>
where
A: Send + 'static,
M: Fn(usize) -> A + Send + Sync + 'static,
F: Fn(&mut A, T) + Send + Sync + 'static,
{
let queue = Arc::new(Queue {
items: Mutex::new(VecDeque::with_capacity(depth)),
ready: Condvar::new(),
stopping: AtomicBool::new(false),
depth,
#[cfg(test)]
before_wait: std::sync::OnceLock::new(),
});
let served = Arc::new(AtomicU64::new(0));
let run = Arc::new(run);
let make = Arc::new(make);
let mut handles = Vec::with_capacity(workers);
for i in 0..workers {
let q = Arc::clone(&queue);
let run = Arc::clone(&run);
let make = Arc::clone(&make);
let served = Arc::clone(&served);
let name = format!("gunnar-{class:?}-{i}").to_lowercase();
let h = std::thread::Builder::new()
.name(name)
.spawn(move || {
// ONCE, at boot, before this worker takes any work.
let mut arena = make(i);
while let Some(item) = q.pop_blocking() {
served.fetch_add(1, Ordering::Relaxed);
run(&mut arena, item);
}
})
.expect("spawning a pipeline worker at boot");
handles.push(h);
}
Pipeline {
class,
queue,
workers: Mutex::new(handles),
served,
}
}
/// Hand an operation to the standing fan. Allocates nothing but the queue
/// slot; spawns nothing.
pub fn submit(&self, item: T) -> Result<usize, SubmitError> {
self.queue.push(item)
}
/// The class this fan serves.
pub fn class(&self) -> OpClass {
self.class
}
/// Operations a worker has actually picked up.
pub fn served(&self) -> u64 {
self.served.load(Ordering::Relaxed)
}
/// Items waiting for a worker.
pub fn queued(&self) -> usize {
self.queue
.items
.lock()
.expect("pipeline queue poisoned")
.len()
}
/// Threads standing in this fan.
pub fn workers(&self) -> usize {
self.workers
.lock()
.expect("pipeline workers poisoned")
.len()
}
/// Stop accepting work, drain what is queued, and JOIN every worker.
///
/// Joining rather than detaching: a worker that outlives the pipeline holds
/// an `Arc` on the queue and on whatever the closure captured, which on a
/// server means a repository handle that never closes.
pub fn shutdown(&self) {
// `stopping` is set UNDER the queue's lock: a worker reads it with the lock held
// and waits (releasing the lock atomically), so the store lands either before its
// read (it returns) or after it waits (the notify wakes it). Set without the lock,
// a store + notify between the read and the wait was lost and join() hung for good
// (stinkies ssh-pipeline-shutdown-hang, 33 min at load ~35).
{
let _q = self.queue.items.lock().expect("pipeline queue poisoned");
self.queue.stopping.store(true, Ordering::Release);
}
self.queue.ready.notify_all();
let mut handles = self.workers.lock().expect("pipeline workers poisoned");
for h in handles.drain(..) {
let _ = h.join();
}
}
}
impl<T: Send + 'static> Drop for Pipeline<T> {
fn drop(&mut self) {
self.shutdown();
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashSet;
use std::sync::mpsc;
use std::time::Duration;
/// ★ THREADS ARE SPAWNED AT BOOT, NOT PER OPERATION.
///
/// The whole claim of this module in one guard: submit N operations and the
/// set of threads that ran them is bounded by the fan's size, not by N.
///
/// Seen RED by removing the standing fan entirely — `start` spawns nothing
/// and `submit` runs the work on a fresh thread, which is the shape this
/// module replaces: **"12 operations ran on 12 distinct threads — the fan is
/// being built per request"**.
///
/// A FIRST attempt at that break spawned an idle thread per submit while
/// leaving the work on the fan, and the guard stayed green — correctly, and
/// it is recorded here because it is the trap: breaking something *near* the
/// property is not breaking the property.
#[test]
fn operations_reuse_the_standing_fan_rather_than_a_thread_each() {
let (tx, rx) = mpsc::channel();
let p = Pipeline::start(OpClass::Clone, 3, 64, move |_op: u32| {
let id = std::thread::current().id();
tx.send(id).expect("collector alive");
});
for i in 0..12 {
p.submit(i).expect("queue has room");
}
let mut seen: HashSet<std::thread::ThreadId> = HashSet::new();
for _ in 0..12 {
seen.insert(rx.recv_timeout(Duration::from_secs(5)).expect("op ran"));
}
assert!(
seen.len() <= 3,
"{} operations ran on {} distinct threads — the fan is being built per request",
12,
seen.len()
);
assert_eq!(p.served(), 12);
}
/// The workers exist BEFORE any operation arrives. A fan that is created
/// lazily on first submit is not standing, it is merely cached.
///
/// Seen RED by the same break: **"the fan has 0 workers before any
/// operation was submitted"**.
#[test]
fn the_fan_is_running_before_the_first_operation() {
let p: Pipeline<u32> = Pipeline::start(OpClass::Fetch, 4, 16, |_| {});
assert_eq!(
p.workers(),
4,
"the fan has {} workers before any operation was submitted",
p.workers()
);
assert_eq!(p.served(), 0, "nothing has been served yet");
}
/// ★ A FULL QUEUE REFUSES. Backpressure is the point of a bounded queue; an
/// unbounded one converts overload into unbounded memory and a slower
/// collapse.
///
/// Seen RED by removing the `q.len() >= self.depth` check: **"the queue
/// accepted 40 items into a 4-deep queue"**.
#[test]
fn a_full_queue_refuses_rather_than_growing() {
// A worker that blocks, so the queue genuinely fills.
// The worker closure must be `Sync`, and a `Receiver` is not — so it
// is shared behind a mutex. One worker, so there is no contention.
let (release, held) = mpsc::channel::<()>();
let held = Arc::new(Mutex::new(held));
let p = Pipeline::start(OpClass::Clone, 1, 4, move |_op: u32| {
// BOUNDED, and that is not tidiness. A bare `recv()` here deadlocks
// the whole test binary when an assertion fires before the release
// loop: the worker blocks forever, `Drop` calls `shutdown()`, which
// JOINS it, and nothing ever returns. Seen — the red proof for the
// guard below hung instead of failing, which is strictly worse than
// a failure because CI reports it as a timeout with no message.
let _ = held
.lock()
.expect("held")
.recv_timeout(Duration::from_secs(10));
});
let mut accepted = 0usize;
let mut refused = 0usize;
for i in 0..40 {
match p.submit(i) {
Ok(_) => accepted += 1,
Err(SubmitError::QueueFull) => refused += 1,
Err(e) => panic!("unexpected refusal: {e}"),
}
}
assert!(
refused > 0,
"the queue accepted {accepted} items into a 4-deep queue"
);
assert!(
accepted <= 4 + 1,
"accepted {accepted}, which is more than the depth plus the one in service"
);
for _ in 0..accepted {
let _ = release.send(());
}
}
/// Shutdown JOINS. A detached worker holds an `Arc` on whatever the closure
/// captured — on a server, a repository handle that never closes.
///
/// Seen RED by replacing `h.join()` with `drop(h)`: the guard below observed
/// **"a worker outlived the pipeline: 3 strong refs on the queue after
/// shutdown"**.
#[test]
fn shutdown_joins_every_worker() {
let p: Pipeline<u32> = Pipeline::start(OpClass::Gc, 3, 8, |_| {});
let q = Arc::clone(&p.queue);
assert!(
Arc::strong_count(&q) > 1,
"workers hold the queue while running"
);
p.shutdown();
assert_eq!(p.workers(), 0, "handles were not drained");
assert_eq!(
Arc::strong_count(&q),
2,
"a worker outlived the pipeline: {} strong refs on the queue after shutdown",
Arc::strong_count(&q)
);
}
/// **A shutdown that lands between a worker's `stopping` check and its wait is not
/// lost** (stinkies ssh-pipeline-shutdown-hang: the suite at load ~35 hung 33 min in
/// `shutdown_joins_every_worker`, the worker parked in `futex_do_wait` for good). The
/// worker is held in exactly that window (lock held, `stopping` read false); shutdown
/// is asked meanwhile; the worker then waits. Shutdown must still end.
#[test]
fn a_shutdown_in_the_window_before_a_worker_waits_still_joins_it() {
let p: Pipeline<u32> = Pipeline::start(OpClass::Gc, 1, 8, |_| {});
let (in_window, window) = mpsc::channel::<()>();
let (go, gate) = mpsc::channel::<()>();
let gate = Mutex::new(Some(gate));
let in_window = Mutex::new(in_window);
let _ = p.queue.before_wait.set(Box::new(move || {
// Only the first pass after the item is held; every later wait goes straight on.
if let Some(g) = gate.lock().unwrap().take() {
let _ = in_window.lock().unwrap().send(());
let _ = g.recv_timeout(Duration::from_secs(10));
}
}));
p.submit(1).expect("queued");
window
.recv_timeout(Duration::from_secs(10))
.expect("the worker reached the window");
let (done, ended) = mpsc::channel::<()>();
std::thread::spawn(move || {
p.shutdown();
let _ = done.send(());
});
// Let the shutdown run its course (or block, if it must take the lock) while the
// worker is still in the window, then let the worker go on to wait.
std::thread::sleep(Duration::from_millis(300));
let _ = go.send(());
assert!(
ended.recv_timeout(Duration::from_secs(10)).is_ok(),
"shutdown never joined: the worker waits on a wake-up that came before it waited"
);
}
/// ★ THE ARENA IS BUILT ONCE PER WORKER, NOT ONCE PER OPERATION.
///
/// A standing fan removes the thread spawn but not the buffer. This asserts
/// the buffer survives between operations at full capacity, which is what
/// makes `kernel_init_pages` — 4.7% of the top profile on a `linux.git`
/// clone — stop happening rather than merely shrink.
///
/// Seen RED by moving the allocation inside `run` (`let mut arena =
/// make(i);` per item, the shape this replaces): **"the arena was rebuilt:
/// 20 constructions for 20 operations across 2 workers"**.
#[test]
fn the_arena_is_built_once_per_worker_and_reused_by_every_operation() {
let built = Arc::new(AtomicU64::new(0));
let b = Arc::clone(&built);
let (tx, rx) = mpsc::channel();
let p = Pipeline::start_with_arena(
OpClass::Clone,
2,
64,
move |_i| {
b.fetch_add(1, Ordering::Relaxed);
Vec::<u8>::with_capacity(4096)
},
move |arena: &mut Vec<u8>, _op: u32| {
let before = arena.capacity();
arena.clear();
arena.extend_from_slice(&[7u8; 1024]);
tx.send((before, arena.capacity()))
.expect("collector alive");
},
);
for i in 0..20 {
p.submit(i).expect("queue has room");
}
let mut reused = 0usize;
for _ in 0..20 {
let (before, _after) = rx.recv_timeout(Duration::from_secs(5)).expect("op ran");
if before >= 4096 {
reused += 1;
}
}
// `<= 2`, not `== 2`: a worker builds its arena when the THREAD starts,
// and 20 fast operations can be drained by the first worker before the
// second is even scheduled. Asserting `== 2` made this guard flake on a
// loaded box — the property is "not one per operation", and pinning it
// to an exact thread count tests the scheduler instead.
let n = built.load(Ordering::Relaxed);
assert!(
(1..=2).contains(&n),
"the arena was rebuilt: {n} constructions for 20 operations across 2 workers"
);
assert_eq!(
reused, 20,
"only {reused} of 20 operations saw a warm arena — capacity is being thrown away"
);
}
/// ★ THE TRANSFER FANS ARE SIZED FOR CONCURRENCY, NOT FOR CORES.
///
/// Seen RED — in the BENCH, not here — by `(cores/2).clamp(2,8)`:
/// **"multiuser_scaling: 16 of 102 user(s) did not complete their ops"**,
/// with **zero** `transfer.refused` in the entire run. Eight workers, 102
/// I/O-bound clients, a 64-deep queue: nobody was turned away, they were
/// made to wait until they timed out. Shedding load is the governor's job;
/// the fan must not serialise ordinary concurrency on its own.
///
/// `gc` is the exception and stays small: a repack is real CPU work, so it
/// is bounded by cores and must not crowd out serving.
#[test]
fn transfer_fans_are_wide_and_gc_stays_small() {
for cores in [1usize, 4, 32, 256] {
assert_eq!(
default_workers(OpClass::Gc, cores),
2,
"gc at {cores} cores"
);
for class in [
OpClass::LsRemote,
OpClass::Fetch,
OpClass::Clone,
OpClass::Push,
] {
let n = default_workers(class, cores);
assert!(
n >= 16,
"{class:?} at {cores} cores got {n} workers — 102 concurrent \
clients against a fan this small queue until they time out"
);
assert!(n <= 256, "{class:?} at {cores} cores is unbounded: {n}");
}
}
// The 32-core box the bench runs on: comfortably over 102 clients.
assert!(
default_workers(OpClass::Clone, 32) >= 102,
"the fan must hold multiuser_scaling's 102 clients without queueing"
);
}
}
// ════════════════════════════════════════════════════════════════════════════
// The five fans, built at boot
// ════════════════════════════════════════════════════════════════════════════
/// One unit of work handed to a standing fan.
///
/// A boxed `FnOnce` and not a pure descriptor, and that is a compromise worth
/// naming: a descriptor would be `operate(op)` in its ideal form, but the git
/// transfer it has to carry is an async future over a live socket, and the
/// honest bridge to an OS thread is a closure that owns it. One box per
/// operation is one allocation — against the ~27.6 M this design exists to
/// remove, and against tokio's blocking pool spawning a THREAD for the same
/// work. It is not zero; it is four orders of magnitude off the thing that hurt.
pub type Job = Box<dyn FnOnce() + Send + 'static>;
/// **Every class's fan, standing, for the life of the process.**
///
/// Built once in [`crate::server::SshServer::new`]. A request classifies, passes
/// the governor, and is submitted here; the thread that runs it was already
/// alive and parked before the client connected.
pub struct Pipelines {
ls_remote: Pipeline<Job>,
fetch: Pipeline<Job>,
clone: Pipeline<Job>,
push: Pipeline<Job>,
gc: Pipeline<Job>,
}
impl Pipelines {
/// Build all five fans. Call this at BOOT, once.
///
/// `depth` is per class, not shared: a queue of clones filling up must not
/// refuse an ls-remote, which is the same head-of-line reasoning the
/// governor's per-class sojourn windows exist for.
pub fn start(cores: usize, depth: usize) -> Pipelines {
fn fan(class: OpClass, cores: usize, depth: usize) -> Pipeline<Job> {
Pipeline::start(class, default_workers(class, cores), depth, |job: Job| {
job()
})
}
Pipelines {
ls_remote: fan(OpClass::LsRemote, cores, depth),
fetch: fan(OpClass::Fetch, cores, depth),
clone: fan(OpClass::Clone, cores, depth),
push: fan(OpClass::Push, cores, depth),
gc: fan(OpClass::Gc, cores, depth),
}
}
fn of(&self, class: OpClass) -> &Pipeline<Job> {
match class {
OpClass::LsRemote => &self.ls_remote,
OpClass::Fetch => &self.fetch,
OpClass::Clone => &self.clone,
OpClass::Push => &self.push,
OpClass::Gc => &self.gc,
}
}
/// Hand an operation to its class's standing fan.
pub fn submit(&self, class: OpClass, job: Job) -> Result<usize, SubmitError> {
self.of(class).submit(job)
}
/// Operations this class's fan has picked up. Applied output — the only
/// thing that says the fan is CARRYING traffic rather than merely existing.
pub fn served(&self, class: OpClass) -> u64 {
self.of(class).served()
}
/// Items waiting for a worker in this class.
pub fn queued(&self, class: OpClass) -> usize {
self.of(class).queued()
}
/// Threads standing in this class's fan.
pub fn workers(&self, class: OpClass) -> usize {
self.of(class).workers()
}
}
/// **The process's fans.** One set, for the whole server, forever.
///
/// A global and not a field, for two reasons that only became visible once the
/// SSH path was wired and the HTTP path was not:
///
/// * **Both transports must use the SAME fans.** `gunnar-http` holds its own
/// `Arc<dyn GitTransport>` and calls `serve()` itself, so a set of fans owned
/// by `SshServer` covered SSH transfers and silently missed every HTTP one —
/// which is most of the `population_*` and `multiuser_scaling` load.
/// * **Two listeners must not mean two fans.** Standing machinery sized to the
/// box is the whole point; one set per listener doubles the threads and
/// halves the meaning of the limit.
///
/// Built on first use, sized from the box, and never rebuilt.
static FANS: std::sync::OnceLock<Pipelines> = std::sync::OnceLock::new();
/// The process's standing fans, building them on first call.
pub fn fans() -> &'static Pipelines {
FANS.get_or_init(|| {
let cores = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(4);
// Queue depth per class. Deep enough that a burst waits rather than
// being refused; the governor's CoDel window decides whether that wait
// is acceptable, not this number.
Pipelines::start(cores, 64)
})
}
#[cfg(test)]
mod pipelines_tests {
use super::*;
use std::time::Duration;
/// ★ ALL FIVE FANS ARE STANDING BEFORE ANY REQUEST ARRIVES, and gc's is the
/// small one.
///
/// Seen RED by building only the class asked for on first submit: **"the
/// clone fan has 0 workers before any request"**.
#[test]
fn every_class_has_a_standing_fan_at_boot() {
let p = Pipelines::start(32, 16);
for class in [
OpClass::LsRemote,
OpClass::Fetch,
OpClass::Clone,
OpClass::Push,
OpClass::Gc,
] {
assert!(
p.workers(class) > 0,
"the {class:?} fan has 0 workers before any request"
);
assert_eq!(p.served(class), 0, "{class:?} served something at boot");
}
assert_eq!(p.workers(OpClass::Gc), 2, "gc must stay small");
assert!(
p.workers(OpClass::LsRemote) > p.workers(OpClass::Gc),
"ls-remote must out-scale gc"
);
}
/// ★ ONE set of fans for the process, however many listeners ask for them.
///
/// Seen RED by returning `Pipelines::start(..)` fresh on each call instead
/// of the `OnceLock`: **"two callers got two different fan sets — SSH and
/// HTTP would each have their own"**.
#[test]
fn every_caller_gets_the_same_process_wide_fans() {
let a = fans();
let b = fans();
assert!(
std::ptr::eq(a, b),
"two callers got two different fan sets — SSH and HTTP would each have their own"
);
}
/// ★ THE COUNTERS REPORT REAL OCCUPANCY, not a constant.
///
/// `served`, `queued` and `workers` existed from the start and NOTHING read
/// them, so the only evidence the fans carried traffic was the absence of
/// refusals — silence, not evidence. Now `server.rs` logs depth on every
/// operation that actually queues, and a counter that is wired to a log is
/// a counter that can lie.
///
/// So this asserts the queue is SEEN NON-EMPTY under back-pressure, which
/// is the only reading that distinguishes "the fan is carrying work" from
/// "the fan exists". A guard that merely called `queued()` and accepted
/// whatever came back would pass against a function returning 0 forever —
/// which is exactly the shape of four dead guards found in this suite
/// today, including a control satisfied by `0 == 0`.
///
/// Seen RED with `queued()` stubbed to `0`: **"the queue never reported a
/// waiting item across 24 submissions against 1 worker — the depth this
/// logs is a constant"**.
#[test]
fn the_occupancy_counters_are_not_constants() {
// One worker, held, so submissions must pile up behind it.
let (release, held) = std::sync::mpsc::channel::<()>();
let held = Arc::new(Mutex::new(held));
let p: Pipeline<u32> = Pipeline::start(OpClass::Clone, 1, 64, move |_| {
// BOUNDED, and that is not tidiness. A bare `recv()` here deadlocks
// the whole test binary when an assertion fires before the release
// loop: the worker blocks forever, `Drop` calls `shutdown()`, which
// JOINS it, and nothing ever returns. Seen — the red proof for the
// guard below hung instead of failing, which is strictly worse than
// a failure because CI reports it as a timeout with no message.
let _ = held
.lock()
.expect("held")
.recv_timeout(Duration::from_secs(10));
});
let mut deepest = 0usize;
for i in 0..24 {
if let Ok(depth) = p.submit(i) {
deepest = deepest.max(depth);
}
}
assert!(
deepest > 1,
"the queue never reported a waiting item across 24 submissions against \
1 worker — the depth this logs is a constant"
);
assert!(
p.queued() > 0,
"queued() says {} with {deepest} known to have piled up",
p.queued()
);
assert_eq!(p.workers(), 1, "workers() lost count of its own fan");
// Let them through, then the served counter must MOVE — a `served` that
// never rises would make the log line above report an idle fan forever.
for _ in 0..24 {
let _ = release.send(());
}
let mut waited = 0;
while p.served() < 24 && waited < 200 {
std::thread::sleep(Duration::from_millis(25));
waited += 1;
}
assert_eq!(
p.served(),
24,
"served() reports {} after 24 jobs were released",
p.served()
);
}
/// A submitted job RUNS on the fan, and lands in the right class.
#[test]
fn a_job_runs_on_its_own_class_fan() {
let p = Pipelines::start(8, 16);
let (tx, rx) = std::sync::mpsc::channel();
p.submit(OpClass::Push, Box::new(move || tx.send(()).unwrap()))
.expect("queue has room");
rx.recv_timeout(std::time::Duration::from_secs(5))
.expect("the job ran on the fan");
assert_eq!(p.served(OpClass::Push), 1);
assert_eq!(p.served(OpClass::Clone), 0, "it ran on the wrong fan");
}
}