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) {
            // ...
        }
    }
}

2 posts - 2 participants

Read full topic