crates/gunnar-ui-server/src/readme/queue.rs
//! **The bounded door every typeset goes through.**
//!
//! ```text
//! request ─▶ admitted? ──no──▶ 503, at once
//! │yes
//! ▼
//! a worker ──waited too long──▶ 503, with a Retry-After
//! │
//! ▼
//! spawn_blocking ──took too long──▶ 504
//! │
//! ▼
//! the page
//! ```
//!
//! # Why a queue at all, when the cache already exists
//!
//! The README cache makes the SECOND reader of a document free. It does nothing
//! at all for the first reader of a THOUSAND documents, and that is the request
//! pattern a stranger with a for-loop has: `/render?path=` names any file in any
//! public tree, every one a cache miss, every miss a typst compile measured in
//! tens to hundreds of milliseconds of one core. `spawn_blocking` moves that off
//! the runtime's workers, which is why a burst does not stall the event loop —
//! but tokio's blocking pool is **512 threads by default and grows to meet
//! demand**, so what it does instead is hand a stranger every core on the box
//! and several hundred typst arenas' worth of memory.
//!
//! So the bound is here, in front of the pool, and it is two numbers rather than
//! one:
//!
//! * **how many may RUN** — [`Queue::workers`], the box's parallelism, because a
//! typeset is pure compute and more of them at once buys nothing;
//! * **how many may be in the building at all** — running plus waiting. Past
//! that the answer is an immediate 503. A waiter costs almost nothing to hold,
//! which is exactly why an unbounded line of them is a way to run the console
//! out of memory without ever compiling anything.
//!
//! # Why the permit is moved INTO the blocking closure
//!
//! Because a typeset cannot be cancelled. Dropping the `JoinHandle` after the
//! deadline does not stop the thread; typst keeps going until it is done. A
//! permit released when the *waiter* gave up would let the next request start
//! while the abandoned compile still had a core, and under sustained timeouts
//! the bound would drift to meaninglessness — the failure mode of every
//! "timeout the acquire" queue in front of uncancellable work. The permit rides
//! into the closure and is dropped by the thread that finishes the job, so what
//! is counted is what is actually running.
//!
//! # Why it is a process-wide handle and not a field on `App`
//!
//! What this bounds is the BOX's cores, and the box is the process. Two
//! consoles' worth of `App` in one process is not a thing that happens, and a
//! per-`App` queue would have meant every test fixture in the tree growing a
//! field for a limit it never reaches. [`shared`] is read from
//! [`crate::api::readme`], [`crate::api::render`] and nowhere else; the tests
//! below build their own [`Queue`] with numbers small enough to see the edge.
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use crate::grpc::Fail;
/// How long a request will WAIT for a worker before it is refused.
///
/// Short on purpose: this is time a browser spends with nothing on the screen,
/// and the honest answer to "the console is busy" is faster than the polite
/// one. A README that is already cached never gets here.
pub const DEFAULT_WAIT: Duration = Duration::from_secs(5);
/// How long one typeset may RUN before the reader is told it will not arrive.
///
/// Well past a normal compile — a 32 KB README typesets in tens of milliseconds
/// — so reaching this means a document that is pathological rather than large,
/// and the size ceiling in [`crate::readme::MAX_SVG_BYTES`] catches large.
pub const DEFAULT_DEADLINE: Duration = Duration::from_secs(30);
/// How many may wait per worker before the queue says no.
///
/// Four deep is a queue that absorbs a page of panels arriving together and
/// refuses a crawler. It is not a tuning knob anybody has needed; the number is
/// here to be read, not turned.
pub const QUEUED_PER_WORKER: usize = 4;
/// Why a typeset did not happen.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Refused {
/// The queue was full when the request arrived. Nothing was compiled and
/// nothing waited.
Full { admitted: usize },
/// It waited for a worker and did not get one.
Waited { waited: Duration },
/// A worker took it and did not finish in time. **The compile is still
/// running**; what ended is this request's interest in it.
TooSlow { deadline: Duration },
/// The blocking task did not come back at all — a panic in the typesetter.
Crashed { why: String },
}
impl Refused {
/// The refusal as the console's wire failure, so it reaches a browser with
/// the same shape as everything else `api` answers.
///
/// `unavailable` (503) for the three that mean *ask again*, and
/// `deadlineexceeded` (504) for the one that means *this document did not
/// typeset in the time a page is allowed to take*. Both come out of
/// [`Fail::http_status`], which already knows those two words, rather than
/// out of a status code invented here.
pub fn fail(&self) -> Fail {
let (code, message) = match self {
Refused::Full { admitted } => (
"unavailable",
format!(
"this console is typesetting as many documents as it will at once \
({admitted} in hand); try again in a moment"
),
),
Refused::Waited { waited } => (
"unavailable",
format!(
"the typesetter did not free up within {}s; try again in a moment",
waited.as_secs()
),
),
Refused::TooSlow { deadline } => (
"deadlineexceeded",
format!(
"this document was still typesetting after {}s, so the console stopped \
waiting for it",
deadline.as_secs()
),
),
Refused::Crashed { why } => (
"unavailable",
format!("the typesetter did not finish: {why}"),
),
};
Fail {
code: code.to_owned(),
message,
}
}
}
impl std::fmt::Display for Refused {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.fail().message)
}
}
/// The door.
#[derive(Debug)]
pub struct Queue {
/// How many typesets may be running at once.
running: Arc<Semaphore>,
/// How many may be here at all: running plus waiting.
admitted: Arc<Semaphore>,
workers: usize,
wait: Duration,
deadline: Duration,
}
impl Queue {
/// A queue with `workers` running at once and `QUEUED_PER_WORKER` waiting
/// behind each.
pub fn new(workers: usize, wait: Duration, deadline: Duration) -> Queue {
let workers = workers.max(1);
Queue {
running: Arc::new(Semaphore::new(workers)),
admitted: Arc::new(Semaphore::new(workers * (1 + QUEUED_PER_WORKER))),
workers,
wait,
deadline,
}
}
/// The process's queue: one worker per core the runtime can see, unless
/// `GUNNAR_UI_TYPESETTERS` says otherwise.
///
/// An unreadable or zero value is the default rather than a refusal to
/// start: this bounds a cache miss, and a console that would not boot over a
/// typo in a tuning knob is worse than one that typesets on every core.
pub fn from_environment() -> Queue {
let workers = std::env::var("GUNNAR_UI_TYPESETTERS")
.ok()
.and_then(|value| value.trim().parse::<usize>().ok())
.filter(|workers| *workers > 0)
.unwrap_or_else(|| {
std::thread::available_parallelism().map_or(2, std::num::NonZeroUsize::get)
});
Queue::new(workers, DEFAULT_WAIT, DEFAULT_DEADLINE)
}
/// How many typesets may run at once here.
pub fn workers(&self) -> usize {
self.workers
}
/// How many may be in the building — running plus waiting — before the
/// queue refuses.
pub fn capacity(&self) -> usize {
self.workers * (1 + QUEUED_PER_WORKER)
}
/// Run one typeset, or say why not.
///
/// `job` is the synchronous, CPU-bound compile. It runs on
/// `spawn_blocking`, holding one permit from the moment it starts until the
/// moment it ends — see the module docs for why the permit travels with the
/// work rather than with the waiter.
///
/// # Errors
///
/// [`Refused`], which every caller turns into a 503 or a 504 through
/// [`Refused::fail`]. It is never a fallback to a cheaper grade: a console
/// under load must not answer a burst by doing MORE work per request.
pub async fn typeset<T, F>(&self, job: F) -> Result<T, Refused>
where
F: FnOnce() -> T + Send + 'static,
T: Send + 'static,
{
// The door, and it does not block: a full queue is answered now.
let Ok(_admitted) = self.admitted.clone().try_acquire_owned() else {
return Err(Refused::Full {
admitted: self.capacity(),
});
};
let worker: OwnedSemaphorePermit =
match tokio::time::timeout(self.wait, self.running.clone().acquire_owned()).await {
Ok(Ok(permit)) => permit,
// A closed semaphore cannot happen — nothing closes it — but the
// safe answer costs one arm and an `expect` here would be a
// panic on a path a burst reaches.
Ok(Err(_)) => {
return Err(Refused::Waited { waited: self.wait });
}
Err(_) => return Err(Refused::Waited { waited: self.wait }),
};
let handle = tokio::task::spawn_blocking(move || {
// Dropped by the thread that did the work, however this request
// ends.
let _worker = worker;
job()
});
match tokio::time::timeout(self.deadline, handle).await {
Ok(Ok(done)) => Ok(done),
Ok(Err(e)) => Err(Refused::Crashed { why: e.to_string() }),
Err(_) => Err(Refused::TooSlow {
deadline: self.deadline,
}),
}
}
}
/// The process's typeset queue.
pub fn shared() -> &'static Queue {
static SHARED: OnceLock<Queue> = OnceLock::new();
#[cfg(not(test))]
let make = Queue::from_environment;
// Under `cargo test` every README test in the crate typesets through this
// one queue at once, on a debug build of the typesetter, on whatever else
// the host is running: the production wait and deadline would then measure
// the host, not the console. The edges themselves are held by this module's
// own tests, on a queue small enough to reach them.
#[cfg(test)]
let make = || Queue::new(Queue::from_environment().workers(), Duration::from_secs(300), Duration::from_secs(600));
SHARED.get_or_init(make)
}
#[cfg(test)]
mod tests {
use super::*;
fn small() -> Queue {
Queue::new(1, Duration::from_millis(200), Duration::from_millis(400))
}
/// ★ **The bound is real: past it the answer is a refusal and not a
/// longer line.**
///
/// Red-first, and it is the whole reason this module exists: with every
/// typeset going straight to `spawn_blocking`, a hundred anonymous requests
/// for a hundred uncached documents became a hundred typst compiles on a
/// pool that grows to 512 threads. Nothing refused, nothing queued, the box
/// saturated by a for-loop.
///
/// One worker and `QUEUED_PER_WORKER` behind it, so the edge is countable:
/// the first `capacity()` are taken and the next one is refused **without
/// waiting**.
#[tokio::test]
async fn past_its_bound_the_queue_refuses_rather_than_queues() {
let queue = Arc::new(small());
let (release, held) = tokio::sync::oneshot::channel::<()>();
let held = Arc::new(std::sync::Mutex::new(Some(held)));
// Fill it: one running (and parked inside the job), the rest waiting.
let mut taken = Vec::new();
for _ in 0..queue.capacity() {
let queue = queue.clone();
let held = held.clone();
taken.push(tokio::spawn(async move {
queue
.typeset(move || {
// Only the first job gets the receiver; the others never
// start, because there is one worker.
if let Some(wait) = held.lock().expect("the gate").take() {
let _ = wait.blocking_recv();
}
"typeset"
})
.await
}));
}
// Let the first one reach the blocking pool and the rest reach the
// line, so the next arrival meets a full queue rather than a race.
tokio::time::sleep(Duration::from_millis(50)).await;
let refused = queue.typeset(|| "typeset").await;
assert_eq!(
refused,
Err(Refused::Full {
admitted: queue.capacity()
}),
"the queue took one past its bound"
);
assert_eq!(
refused.unwrap_err().fail().http_status(),
http::StatusCode::SERVICE_UNAVAILABLE
);
let _ = release.send(());
for task in taken {
let _ = task.await.expect("the task");
}
}
/// ★ **A document that will not finish is answered, not held.** The
/// deadline is a 504 with the number in it, and the abandoned compile keeps
/// its permit until it actually ends — which is what stops a timeout from
/// quietly widening the bound.
#[tokio::test]
async fn a_typeset_past_the_deadline_is_a_gateway_timeout() {
let queue = Queue::new(1, Duration::from_millis(100), Duration::from_millis(80));
let refused = queue
.typeset(|| {
std::thread::sleep(Duration::from_millis(400));
"eventually"
})
.await
.expect_err("it must not wait for this one");
assert!(matches!(refused, Refused::TooSlow { .. }), "{refused:?}");
assert_eq!(
refused.fail().http_status(),
http::StatusCode::GATEWAY_TIMEOUT
);
assert!(refused.fail().message.contains('s'), "{refused}");
// The worker is still busy, so the next arrival waits for it and is
// refused rather than served — the permit did not come back when this
// request gave up.
let next = queue.typeset(|| "typeset").await;
assert!(
matches!(next, Err(Refused::Waited { .. })),
"an abandoned compile released its permit early: {next:?}"
);
}
/// The ordinary path: what goes in comes out, and the queue empties.
#[tokio::test]
async fn a_typeset_that_fits_is_simply_run() {
let queue = small();
for _ in 0..3 {
assert_eq!(
queue.typeset(|| 7 * 6).await.expect("it runs"),
42,
"a queue that does not empty is a queue that stops working"
);
}
}
/// The knob is read, and a nonsense value is not a reason to refuse to
/// start.
#[test]
fn the_worker_count_has_a_floor_and_a_default() {
assert_eq!(Queue::new(0, DEFAULT_WAIT, DEFAULT_DEADLINE).workers(), 1);
assert_eq!(
Queue::new(3, DEFAULT_WAIT, DEFAULT_DEADLINE).capacity(),
3 * (1 + QUEUED_PER_WORKER)
);
// Whatever the box is, the process queue is a queue.
assert!(Queue::from_environment().workers() >= 1);
}
}