When the Producer Is Faster Than the Consumer in Producer–Consumer
Loading...
Syome Back to Posts

When the Producer Is Faster Than the Consumer in Producer–Consumer

The producer outpaces the consumer, you should do something to calm it down...

Producer–Consumer Model

This part is the basic knowledge’s introduction. If you are familiar with this Producer–Consumer Model, you can skip it.

Overall

In the producer–consumer model, a producer generates data and a consumer consumes data. The producer and consumer are independent of each other. The producer generates data and stores it in a buffer. The consumer consumes data from the buffer. The producer and consumer are independent of each other.

Here is a simple example of the producer–consumer model in Rust:

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    // Create a channel
    let (tx, rx) = mpsc::channel();

    // Producer
    let producer = thread::spawn(move || {
        for i in 1..=5 {
            println!("Produced: {}", i);
            tx.send(i).unwrap();
            thread::sleep(Duration::from_millis(500));
        }
    });

    // Consumer
    let consumer = thread::spawn(move || {
        while let Ok(value) = rx.recv() {
            println!("Consumed: {}", value);
        }
        println!("Producer finished.");
    });

    producer.join().unwrap();
    consumer.join().unwrap();
}

In this example, the producer generates data and sends it to the consumer through a channel. The consumer consumes data from the channel. The producer and consumer are independent of each other. And the so-called buffer is mpsc::channel itself.

Buffer’s Bound

In the previous example, the buffer is unbounded, which means that the producer can generate data without any limit until OOM.

To solve this problem, we can use a bounded buffer. In Rust’s mpsc, there is an implementation:

let (tx, rx) = mpsc::sync_channel(100); // Creates a new synchronous, bounded(100) channel.

In this case, the producer’s thread will be blocked when the buffer is full until the consumer consumes data.

This strategy is called backpressure.

Common Strategies

While producer produces data in a faster pace:

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    // Create a channel
    let (tx, rx) = mpsc::channel();

    // Producer
    let producer = thread::spawn(move || {
        for i in 1..=1000 /* Change this number to larger it will be interesting XD */ {
            println!("Produced: {}", i);
            tx.send(i).unwrap();
        }
    });

    // Consumer
    let consumer = thread::spawn(move || {
        while let Ok(value) = rx.recv() {
            println!("Consumed: {}", value);
            thread::sleep(Duration::from_millis(1000));
        }
        println!("Producer finished.");
    });

    producer.join().unwrap();
    consumer.join().unwrap();
}

Run this, and you’ll see something that looks fine at first glance. All 1000 “Produced” lines fly by almost instantly. Then, over the next 1000 seconds, the “Consumed” lines trickle out one by one. No crash. No panic. No data loss. Every single item eventually gets consumed.

But here is the issue: at the moment the producer finishes — roughly 1 second in — the other ~999 items that haven’t been consumed are just sitting in memory.

for i in 1..=10000000

Backpressure

One way to solve this problem is this strategy. We can use a bounded buffer like it is mentioned in the previous section.

let (tx, rx) = mpsc::sync_channel(100);

Two costs come with this:

  1. Delay. This is the same shape as TCP’s head-of-line blocking in RTC: ordering is guaranteed, so one slow step stalls everything behind it, even data that’s ready to go. Fine for an event stream. Actively harmful for a state stream (like video frames), which is why RTC runs over UDP instead.

    ADs: About WebRTC, you can check this post written by me.

  2. Blocking the calling thread. send() doesn’t return until there’s room. If that thread has other jobs — accepting new connections, handling other users’ requests — those stop too, for however long send() stays blocked.

So the trade is: bounded memory, in exchange for an unpredictable producer stall. Worth it depends on whether the producer can afford to stall, which is exactly what we look at next.

Dropping(Overwriting) the Data

Backpressure via blocking assumes one thing: the producer can afford to wait. But sometimes it’s on the hot path of a user request, or it’s a hardware interrupt handler, or slowing it down would stall something upstream that matters more than this data does.

use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

fn main() {
    let latest = Arc::new(Mutex::new(None));

    let producer_latest = Arc::clone(&latest);
    let producer = thread::spawn(move || {
        for i in 1..=1000 {
            println!("Produced: {}", i);
            *producer_latest.lock().unwrap() = Some(i);
        }
    });

    let consumer_latest = Arc::clone(&latest);
    let consumer = thread::spawn(move || {
        loop {
            thread::sleep(Duration::from_millis(1000));
            match consumer_latest.lock().unwrap().take() {
                Some(value) => println!("Consumed: {}", value),
                None => continue,
            }
        }
    });

    producer.join().unwrap();
    consumer.join().unwrap();
}

Instead of queuing every produced value, it stores only the latest one. Absolutely the dropping principle can be re-written by yourself. If the producer generates multiple values before the consumer wakes up, all previous values are simply overwritten. In other words, the system deliberately sacrifices completeness to keep memory usage constant and to ensure the producer never has to wait.

The downside is obvious: intermediate updates are lost forever. For applications where every update carries meaningful information, this is too aggressive. But is there a middle ground—one that avoids unbounded queues without simply throwing data away?

Looking Beyond the Queue

The previous two strategies treat the queue as a black box. They either slow down the producer or discard queued data. But some systems exploit something the queue itself doesn’t know: the meaning of the data.

One place where I encountered this idea was while experimenting with winit:

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();
}

At first glance, this looks like another producer-consumer problem: the producer is simply generating events faster than the consumer can process them. But there is an important observation here:

Every UserEvent eventually performs the same operation: window.request_redraw().

The queue has no idea that these requests are semantically equivalent—it simply delivers every message in order. Whether those messages can be merged is no longer a property of the queue, but of the application itself.

The producer–consumer model is often introduced as a synchronization pattern. In practice, however, the difficult part is rarely synchronization itself—it is deciding what should happen when production and consumption run at different speeds.

Every solution is ultimately a trade-off between latency, memory usage, throughput, and correctness.

I’m still writing smth about starvation with that winit example 🤯

Personal experience and knowledge - Issues are welcome