I'm not sure what would happen if you used the standard library's channel implementation in the following cases.
let (tx, rx) = std::sync::mpsc::unbounded_channel();
// some code where multiple producers are sending data through tx
tokio::spawn(async move {
while let Some(message) = rx.recv().await {
//do something with message
}
})
let (tx, rx) = std::sync::mpsc::unbounded_channel();
// some code where multiple producers are sending data through tx
tokio::spawn(async move {
while let Some(message) = rx.try_recv().await {
//do something with message }
}
})
In both cases when the channel is empty and the receivers are waiting for some messages, will the Tokio task in case 1 stay in the run_queue
since recv()
is blocking?
On the other hand, since try_recv()
is not blocking, will the Tokio task get off the run queue so that threads can do other task when there are no messages to read?
After digging into source code, my conclusion is that when Tokio channels are waiting for the messages, they are moved off the run queue. Calling recv()
on the receiver will never be a blocking code when using Tokio.
I also had doubt whether futures::stream::stream::poll_next()
would be blocking, but as with Tokio channels, they are awakened when some value is available.