rust/comprehensive-rust
1
0
Fork 0
mirror of https://github.com/google/comprehensive-rust.git synced 2025-05-23 10:50:18 +02:00
Code Issues Releases Activity

Simplify pin example slightly

Browse Source 
This reduces the vertical space needed by the example, thus making it
easier to keep on the screen while refactoring it.
This commit is contained in:
Martin Geisler 2024-06-20 00:46:36 +02:00
parent 317575131e
commit c4958b67de
1 changed files with 9 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/concurrency/async-pitfalls/pin.rs
View File

@ -1,25 +1,20 @@
use anyhow::Result;
use tokio::sync::{mpsc, oneshot}; use tokio::sync::{mpsc, oneshot};
use tokio::task::spawn;
use tokio::time::{sleep, Duration}; use tokio::time::{sleep, Duration};
// A work item. In this case, just sleep for the given time and respond
// with a message on the `respond_on` channel.
#[derive(Debug)] #[derive(Debug)]
struct Work { struct Work {
input: u32, input: u32,
respond_on: oneshot::Sender<u32>, respond_on: oneshot::Sender<u32>,
} }
// A worker which listens for work on a queue and performs it.
async fn worker(mut work_queue: mpsc::Receiver<Work>) { async fn worker(mut work_queue: mpsc::Receiver<Work>) {
let mut _iterations = 0; let mut _iterations = 0;
loop { loop {
tokio::select! { tokio::select! {
Some(work) = work_queue.recv() => { Some(work) = work_queue.recv() => {
sleep(Duration::from_millis(10)).await; // Pretend to work. sleep(Duration::from_millis(10)).await; // Pretend to work.
work.respond_on work.respond_on.send(work.input * 1000).unwrap();
.send(work.input * 1000)
.expect("failed to send response");
_iterations += 1; _iterations += 1;
} }
// TODO: report number of iterations every 100ms // TODO: report number of iterations every 100ms
@ -27,22 +22,19 @@ async fn worker(mut work_queue: mpsc::Receiver<Work>) {
} }
} }
// A requester which requests work and waits for it to complete. async fn do_work(work_queue: &mpsc::Sender<Work>, input: u32) -> Result<u32> {
async fn do_work(work_queue: &mpsc::Sender<Work>, input: u32) -> u32 {
let (tx, rx) = oneshot::channel(); let (tx, rx) = oneshot::channel();
work_queue work_queue.send(Work { input, respond_on: tx }).await?;
.send(Work { input, respond_on: tx }) Ok(rx.await?)
.await
.expect("failed to send on work queue");
rx.await.expect("failed waiting for response")
} }
#[tokio::main] #[tokio::main]
async fn main() { async fn main() -> Result<()> {
let (tx, rx) = mpsc::channel(10); let (tx, rx) = mpsc::channel(10);
spawn(worker(rx)); tokio::spawn(worker(rx));
for i in 0..100 { for i in 0..100 {
let resp = do_work(&tx, i).await; let resp = do_work(&tx, i).await?;
println!("work result for iteration {i}: {resp}"); println!("work result for iteration {i}: {resp}");
} }
Ok(())
} }