I have a regular multithreaded app (not async) with multiple producers and one consumer. The problem is that the producers all need different synchronization methods – some use sync channels, some use async channels, some directly modify shared state (behind a mutex). How can I make my producer thread wait for all those events simultaneously?
If this were in async land, it would be as simple as a single select! call. Which made me think – all of my channels can be polled, all I need to do is have some way to notify the consumer thread that it should wake up and check the output of all the producers (there aren't very many of them so this should be cheap). How can I implement this notification though?
My first thought was a condvar, but it seems like I wouldn't need the associated mutex for anything, which feels like I'm doing something wrong. My other thought was an mpsc channel with capacity of 1, the producers will try_send (and ignore the error if the channel is already full), the consumer will block on recv. I think this should work? But it also seems a little wrong for some reason. Is there an idiomatic solution that I'm missing?
Some code for reference:
fn producer1(chan: Sender<Data>, notif: SyncSender<()>) {
loop {
let data = produce();
chan.send(data).unwrap();
let _ = notif.try_send(());
}
}
fn producer2(state: &Mutex<State>, notif: SyncSender<()>) {
loop {
let mut state = state.lock().unwrap();
modify(&mut state);
let _ = notif.try_send(());
}
}
fn consumer(chan: Receiver<Data>, state: &Mutex<State>, notif: Receiver<()>) {
loop {
notif.recv().unwrap();
if let Ok(data) = chan.try_recv() {
// ...
}
let state = state.lock().unwrap();
if has_changed(&*state) {
// ...
}
}
}
You’re not missing anything fundamental. “Heterogenous select” is one of the big advantages of async, whereas blocking function calls, by definition, each tie up a whole thread with their own particular kind of thing they are waiting for. Here are some possible solutions — all of which are some variety of converting everything to be compatible with a single blocking function:
Add a dash of async without changing your whole architecture:
Use async-compatible channels like flume, adapting other things to them when necessary.
Use pollster (the smallest possible async executor) to block on a select!() on all of your channels and other things.
Adapt all your event sources to the same variety of channels; flume and crossbeam both have select operations for multiple channels. But this may require using more threads than the async approach.
Adapt all your event sources to be files/pipes/sockets and use an IO select.
Here's an example of a single consumer thread which waits for multiple queues. There are four crossbeam_channel queues feeding into this loop. Note the select!, which has four inputs and a timer.
use crossbeam_channel::{select, Receiver, Sender};
...
pub fn run(&mut self, ui_to_client: Receiver<UiToClient>) -> Result<(), Error> {
while !self.shared.is_aborting() {
// Quit if some thread panics
// Pre-select, so UDP messages get priority.
// Errors fall through and are handled below.
if let Ok(incoming_udp) = self.shared.udp_receiver.try_recv() {
self.handle_incoming_udp(&incoming_udp);
continue; // keep doing UDPs first
}
// Main select, will block.
select! {
recv(self.shared.udp_receiver) -> incoming_udp => self.handle_incoming_udp(&incoming_udp?),
recv(ui_to_client) -> command_to_client => {
match command_to_client {
Ok(cmd) => {
match &cmd {
UiToClient::Shutdown => {
break;
}, // shutdown signal
_ => self.handle_incoming_command(&cmd)
}
}
Err(_) => { break; } // alternate shutdown signal
}
}
recv(self.shared.client_to_client_receiver) -> command_to_client => self.handle_client_to_client(&command_to_client?),
recv(self.shared.server_to_client_receiver) -> server_to_client => self.handle_server_to_client(&server_to_client?),
default(Self::TICK_INTERVAL) => self.tick() // nothing is happening
}
}
log::warn!("Exiting main message read loop.");
self.shutdown()?;
Ok(())
Depending on the nature of your app, you can also use (maybe unpopular) polling with some timeout management:
fn consumer(...) {
let mut delay = Duration::from_millis(1);
loop {
let mut processed = 0;
if let Ok(msg) = try_receive(...) {
processed += 1;
handle_message(msg);
}
if state_has_changed(....) {
processed += 1;
use_state(...);
}
if processed == 0 {
thread::sleep(delay);
delay = (delay * 2).min(Duration::from_millis(250));
} else {
delay = Duration::from_millis(1);
}
}
}
You can play with additional timers, for example when checking for changed state is expensive, only check once a seconds etc. Or for channels you can use while let Ok(msg) = ... loops to receive in batches etc. You can limit the batch size, if needed. There are many variants possible.
This approach does not fit everywhere and sometimes it is seen as old-fashioned. But when it fits, it is simple an yet very powerful. You can control behavior very fine grained. A lot of mission critical software is written in this style, from tiny microcontrollers to large machines, and financial apps, to aerospace applications.
Yeah, that's currently my backup solution. But I'd be either adding unnecessary processing latency (if I choose a long sleep interval) or wasting a lot of system resources (if I choose a short sleep interval). My app doesn't need to process a whole lot of messages, it's the latency that's more important to me so this doesn't seem like a great fit.
Good to know that I'm not completely off here. I really don't feel like converting parts of this code to async and using IO select also seems a little far fetched. I'll look into flume and crossbeam though, thanks!
You can use eventfd on Linux to wait for arbitrary events with epoll. But with crossbeam depending on what you are waiting for you can just spawn an OS thread that sends when a condition is ready.
Using a condvar turned out to be the simplest solution in my case. I’ve added a simple u64 bitmask where every producer sets their corresponding bit when they’re ready and signal the consumer via a condvar. This way the consumer always knows exactly which data is ready and doesn’t have to poll every producer.