Unbounded Queues Starve Event Loops: Why Backpressure Isn't Enough
Starvation in an event-driven renderer can't be fixed by backpressure alone...
The Symptom
I’m building a terminal emulator in Rust: winit for the window, wgpu for rendering, a pty for the shell. It works, until I ran yes inside the terminal and the screen freezes. The child is clearly still alive (CPU is spinning, data is being consumed), but the window never updates…
Not just yes: any flood like cat something large.
Minimal Reproduction
The terminal does roughly two things per event: parse pty data, then request a redraw. So the minimal case is: some thread floods the event loop with user events, every user event asks for a redraw, and we count how often RedrawRequested actually arrives.
Also the following code is in this post. Rendering part I did not show because it’s not relevant and
this is an actually huge work
use std::thread;
use winit::{
application::ApplicationHandler,
event::WindowEvent,
event_loop::{ActiveEventLoop, EventLoop},
window::{Window, WindowId},
};
enum AppEvent {
Data,
}
struct App {
window: Option<Window>,
events: u64,
redraws: u64,
}
impl ApplicationHandler<AppEvent> for App {
fn resumed(&mut self, event_loop: &ActiveEventLoop) {
let window = event_loop
.create_window(Window::default_attributes().with_title("MRE"))
.unwrap();
window.request_redraw();
self.window = Some(window);
}
/// Called when the user event(here is `AppEvent`) is received.
fn user_event(&mut self, _: &ActiveEventLoop, _: AppEvent) {
self.events += 1;
if self.events % 100000 == 0 {
println!(
"user_event = {}, redraw = {}",
self.events, self.redraws
);
}
if let Some(window) = &self.window {
window.request_redraw();
}
}
fn window_event(
&mut self,
event_loop: &ActiveEventLoop,
_: WindowId,
event: WindowEvent,
) {
match event {
WindowEvent::RedrawRequested => {
self.redraws += 1;
if self.events % 100000 == 0 {
println!("Redraw {}", self.redraws);
}
if let Some(window) = &self.window {
window.request_redraw();
}
}
WindowEvent::CloseRequested => event_loop.exit(),
_ => {}
}
}
}
fn main() {
let event_loop = EventLoop::<AppEvent>::with_user_event()
.build()
.unwrap();
let proxy = event_loop.create_proxy();
// The producer
thread::spawn(move || loop {
proxy.send_event(AppEvent::Data).ok().unwrap();
});
let mut app = App {
window: None,
events: 0,
redraws: 0,
};
event_loop.run_app(&mut app).unwrap();
}The producer thread is brutal: loop { proxy.send_event(...) }, exactly what a flooding kernel does to a pty reader thread.
Running it gives:
user_event = 100000, redraw = 0
user_event = 200000, redraw = 0
...The Obvious Fix (and the Trap It Hides)
What make this fatal when the producer outruns the consumer is: The user-event queue is unbounded. So the classic fix is the one I wrote about in this post: a bounded buffer with blocking sends.
let (tx, rx) = std::sync::mpsc::sync_channel::<Vec<u8>>(32);Now the pty reader thread blocks when the channel is full, and in a terminal this is better than the abstract case: the reader stops reading the pty master, the kernel’s pty buffer fills up, and the child process’s write() blocks. Backpressure all the way back to the writer, not just to the reader thread. No more unbounded queue growth.
Actually, let me rewind and ask you the question I should have asked myself before writing:
If you’re adding this
recv, where would you put it?
Think about it. You need a loop that takes data out of the channel, and there is exactly one correct form for it:
while let Ok(event) = rx.try_recv() { // blocking
process(event);
}…and exactly one place it can live: a thread that can block. But now, the winit main thread is not a thread that can block. It is an event-driven dispatcher: poll for events → dispatch callbacks → poll again, forever. Park a recv() on it and the window stops answering to anything.
So I did what felt natural: translated the blocking recv into a non-blocking loop inside a callback — the “drain until empty” you saw above. And it froze everything, not just rendering: keyboard, resize, close, all of it. We hadn’t changed the structure at all. We had moved the unbounded loop into the one place where it is forbidden: a callback of an event-driven dispatcher. The callback runs inside winit’s own user-event drain, and like every other event is only dispatched after that drain ends:
// winit's single_iteration
while let Ok(event) = self.user_receiver.try_recv() {
callback(Event::UserEvent(event), ...); // our drain is in here
}
// RedrawRequested is only reached after this loop endsThe Fix: Three Mechanisms, Each with One Job
1. Bounded channel
As above: the pty data travels over a bounded sync_channel; the reader blocks when full; the kernel buffer and the child process absorb the pressure. Memory is bounded, and any drain is guaranteed to terminate.
2. Wake latch
EventLoopProxy is still the only way to wake the event loop from another thread — and it is still unbounded. So we stop sending one notification per chunk. We send one per batch, deduplicated by an atomic flag:
// producer side, after enqueueing a chunk into the bounded channel:
if !wake_pending.swap(true, Ordering::AcqRel) {
proxy.send_event(AppEvent::Wake);
}swap(true) atomically sets the flag and returns its previous value. Only the sender that sees false actually sends Wake. Everyone else skips: a wake is already on its way, and when the consumer runs it will drain everything currently in the channel, including the chunks these senders just enqueued.
// consumer side, upon receiving Wake:
self.wake_pending.store(false, Ordering::Release); // consumed — arm the next oneThe unbounded queue is still unbounded, but we just never put more than ~1 item into it at a time. The notification rate is decoupled from the data rate.
3. Frame-paced drain
The drain does not happen in user_event (that callback runs inside winit’s own unbounded user-event drain) and it is not “until empty”. It happens in RedrawRequested, capped per frame:
// inside RedrawRequested:
let mut drained = 0;
while drained < DRAIN_BATCH && let Ok(event) = rx.try_recv() {
terminal.append_bytes(&event);
drained += 1;
}
if drained == DRAIN_BATCH {
window.request_redraw(); // more data pending — keep the chain alive
}So the fixed minimal reproduction, end to end:
use std::{
sync::{
atomic::{AtomicBool, Ordering},
mpsc::sync_channel,
Arc,
},
thread,
};
use winit::{
application::ApplicationHandler,
event::WindowEvent,
event_loop::{ActiveEventLoop, EventLoop},
window::{Window, WindowId},
};
/// The producer can only push this many chunks before it has to wait for
/// the consumer — memory stays bounded, and any drain terminates.
const CHANNEL_CAPACITY: usize = 32;
/// Chunks ingested per render frame. When the drain hits this limit we
/// re-request a redraw (chain); when the channel empties the chain stops.
const DRAIN_BATCH: usize = 32;
enum AppEvent {
/// "The channel is non-empty." Sent at most one per batch.
Wake,
}
struct App {
window: Option<Window>,
events: std::sync::mpsc::Receiver<()>,
wake_pending: Arc<AtomicBool>,
drained: u64,
redraws: u64,
}
impl ApplicationHandler<AppEvent> for App {
fn resumed(&mut self, event_loop: &ActiveEventLoop) {
let window = event_loop
.create_window(Window::default_attributes().with_title("MRE fixed"))
.unwrap();
window.request_redraw();
self.window = Some(window);
}
fn user_event(&mut self, _: &ActiveEventLoop, _: AppEvent) {
// Consume the wake token so the next producer send can arm a fresh
// one. The actual drain happens in RedrawRequested.
self.wake_pending.store(false, Ordering::Release);
if let Some(window) = &self.window {
window.request_redraw();
}
}
fn window_event(
&mut self,
event_loop: &ActiveEventLoop,
_: WindowId,
event: WindowEvent,
) {
match event {
WindowEvent::RedrawRequested => {
self.redraws += 1;
// Ingest at most one batch per frame. The producer can
// never outrun this loop because the channel is bounded.
let mut batch = 0;
while batch < DRAIN_BATCH && self.events.try_recv().is_ok() {
batch += 1;
}
self.drained += batch as u64;
// More data pending -> keep the redraw chain alive so the
// render loop paces the drain.
if batch == DRAIN_BATCH {
if let Some(window) = &self.window {
window.request_redraw();
}
}
if self.redraws % 1000 == 0 {
println!("redraw = {}, drained = {}", self.redraws, self.drained);
}
}
WindowEvent::CloseRequested => event_loop.exit(),
_ => {}
}
}
}
fn main() {
let event_loop = EventLoop::<AppEvent>::with_user_event()
.build()
.unwrap();
let proxy = event_loop.create_proxy();
// The flood: instead of one winit user event per chunk, the producer
// pushes into a bounded channel and blocks when full (backpressure).
let (tx, rx) = sync_channel::<()>(CHANNEL_CAPACITY);
let wake_pending = Arc::new(AtomicBool::new(false));
{
let proxy = proxy.clone();
let wake_pending = Arc::clone(&wake_pending);
thread::spawn(move || loop {
// Blocking send: stops the flood from ever outrunning the
// consumer, which is what keeps every drain terminating.
if tx.send(()).is_err() {
break;
}
// Latch: only wake the event loop if no wake is in flight.
if !wake_pending.swap(true, Ordering::AcqRel) {
let _ = proxy.send_event(AppEvent::Wake);
}
});
}
let mut app = App {
window: None,
events: rx,
wake_pending,
drained: 0,
redraws: 0,
};
event_loop.run_app(&mut app).unwrap();
}Running the fixed reproduction under maximum flood:
redraw = 1000, drained = 31860
redraw = 2000, drained = 63701
redraw = 3000, drained = 95561
redraw = 4000, drained = 127561
...Off Topic
This post is a direct sequel to When the Producer Is Faster Than the Consumer in Producer–Consumer. There, the two strategies were block the producer and drop the data. Here is the third one(the core is still backpressure): decouple the notification rate from the data rate, and let the consumer’s frame loop do the pacing.
Personal experience and knowledge - Issues are welcome