|
I have a service that receives commands on one channel and events on another. I'm using tokio::select! to listen on both simultaneously, but when messages arrive quickly on both channels, some get lost. I checked the buffer sizes and they're not full. use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
let (cmd_tx, mut cmd_rx) = mpsc::channel::<String>(100);
let (evt_tx, mut evt_rx) = mpsc::channel::<String>(100);
let cmd_tx_clone = cmd_tx.clone();
tokio::spawn(async move {
for i in 0..50 {
cmd_tx_clone.send(format!("cmd-{i}")).await.unwrap();
}
});
tokio::spawn(async move {
for i in 0..50 {
evt_tx.send(format!("evt-{i}")).await.unwrap();
}
});
let mut cmd_count = 0u32;
let mut evt_count = 0u32;
let mut cmd_done = false;
let mut evt_done = false;
while !cmd_done || !evt_done {
tokio::select! {
msg = cmd_rx.recv(), if !cmd_done => {
match msg {
Some(_cmd) => {
cmd_count += 1;
}
None => cmd_done = true,
}
}
msg = evt_rx.recv(), if !evt_done => {
match msg {
Some(_evt) => {
evt_count += 1;
}
None => evt_done = true,
}
}
}
}
println!("cmds: {cmd_count}, evts: {evt_count}");
// Now always: cmds: 50, evts: 50
}Every run gives me different numbers for evt_count, always less than 50. The send() calls never fail, so the messages are definitely being sent. I can't figure out where they're being lost. |
Replies: 2 comments 2 replies
|
The program you shared runs forever. Please share the correct example. |
|
The original cmd_tx is never dropped (it lives until the end of main), so cmd_rx.recv() never returns None, cmd_done never becomes true, and the loop never exits. |
The original cmd_tx is never dropped (it lives until the end of main), so cmd_rx.recv() never returns None, cmd_done never becomes true, and the loop never exits.
But apparently that wasnt your problem