add tokio/sync examples

Author: smallnest

tokio_examples/Cargo.toml

@@ -6,3 +6,4 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+tokio = { version = "1.21.2", features = ["full"] }

tokio_examples/src/lib.rs

@@ -0,0 +1,178 @@
+pub mod primitives;
+
+pub use primitives::*;
+
+use std::process::Stdio;
+
+use tokio;
+use tokio::process::Command;
+use tokio::io::{BufReader, AsyncBufReadExt};
+
+pub fn process() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    let _: Result<(), Box<dyn std::error::Error>> = rt.block_on(async {
+        let mut child = Command::new("echo")
+            .arg("hello")
+            .arg("world")
+            .spawn()
+            .expect("failed to spawn");
+
+        // Await until the command completes
+        let status = child.wait().await?;
+        println!("the command exited with: {}", status);
+
+        Ok(())
+    });
+}
+
+pub fn process2() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    let _: Result<(), Box<dyn std::error::Error>> = rt.block_on(async {
+        let mut cmd = Command::new("cat");
+        cmd.arg("Cargo.toml");
+
+        // Specify that we want the command's standard output piped back to us.
+        // By default, standard input/output/error will be inherited from the
+        // current process (for example, this means that standard input will
+        // come from the keyboard and standard output/error will go directly to
+        // the terminal if this process is invoked from the command line).
+        cmd.stdout(Stdio::piped());
+
+        let mut child = cmd.spawn().expect("failed to spawn command");
+
+        let stdout = child
+            .stdout
+            .take()
+            .expect("child did not have a handle to stdout");
+
+        let mut reader = BufReader::new(stdout).lines();
+
+        // Ensure the child process is spawned in the runtime so it can
+        // make progress on its own while we await for any output.
+        tokio::spawn(async move {
+            let status = child
+                .wait()
+                .await
+                .expect("child process encountered an error");
+
+            println!("child status was: {}", status);
+        });
+
+        while let Some(line) = reader.next_line().await? {
+            println!("Line: {}", line);
+        }
+
+        Ok(())
+    });
+
+
+}
+
+
+pub fn oneshot() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    let _: Result<(), Box<dyn std::error::Error>> = rt.block_on(async {
+        let (tx, rx) = tokio::sync::oneshot::channel::<String>();
+
+        tokio::spawn(async move {
+            tx.send("hello".to_string()).unwrap();
+        });
+
+        let msg = rx.await?;
+        println!("got = {}", msg);
+
+        Ok(())
+    });
+}
+
+pub fn async_with_oneshot() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    async fn some_computation() -> String {
+        "the result of the computation".to_string()
+    }
+
+    let _: Result<(), Box<dyn std::error::Error>> = rt.block_on(async {
+        let join_handle = tokio::spawn(async move {
+            some_computation().await
+        });
+    
+        // Do other work while the computation is happening in the background
+    
+        // Wait for the computation result
+        let res = join_handle.await?;
+        println!("result = {}", res);
+
+        Ok(())
+    });
+}
+
+pub fn mpsc_example() {
+    async fn some_computation(input: u32) -> String {
+        format!("the result of computation {}", input)
+    }
+    
+
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    let _: Result<(), Box<dyn std::error::Error>> = rt.block_on(async {
+        let (tx, mut rx) = tokio::sync::mpsc::channel::<String>(10);
+
+        tokio::spawn(async move {
+            for i in 0..10 {
+                let res = some_computation(i).await;
+                tx.send(res).await.unwrap();
+            }
+        });
+
+        while let Some(res) = rx.recv().await {
+            println!("got = {}", res);
+        }
+
+        Ok(())
+    });
+}
+
+pub fn broadcast_example() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    let _: Result<(), Box<dyn std::error::Error>> = rt.block_on(async {
+        let (tx, mut rx1) = tokio::sync::broadcast::channel::<String>(10);
+        let mut rx2 = tx.subscribe();
+
+        tokio::spawn(async move {
+            tx.send("hello".to_string()).unwrap();
+            tx.send("world".to_string()).unwrap();
+        });
+
+        println!("rx1 = {:?}", rx1.recv().await);
+        println!("rx2 = {:?}", rx2.recv().await);
+        println!("rx2 = {:?}", rx2.recv().await);
+
+        Ok(())
+    });
+    
+}
+
+pub fn watch_example() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    let _: Result<(), Box<dyn std::error::Error>> = rt.block_on(async {
+        let (tx, rx1) = tokio::sync::watch::channel::<String>("hello".to_string());
+        let mut rx2 = tx.subscribe();
+
+        tokio::spawn(async move {
+            tx.send("world".to_string()).unwrap();
+        });
+
+        println!("rx1 = {:?}", *rx1.borrow());
+        println!("rx2 = {:?}", *rx2.borrow());
+        println!("rx2 = {:?}", rx2.changed().await);
+
+        Ok(())
+    });
+
+}

tokio_examples/src/main.rs

@@ -1,3 +1,21 @@
+use tokio_examples::*;
+
 fn main() {
-    println!("Hello, world!");
+    process();
+    process2();
+
+    oneshot();
+    async_with_oneshot();
+    mpsc_example();
+    broadcast_example();
+    watch_example();
+
+    barrier_example();
+    mutex_example();
+    rwlock_example();
+    semaphore_example();
+    semaphore_example2();
+    notify_example();
+    notify_example2();
+    
 }

tokio_examples/src/primitives.rs

@@ -0,0 +1,158 @@
+use std::sync::Arc;
+use tokio::sync::Barrier;
+use tokio::sync::Mutex;
+use tokio::sync::RwLock;
+
+use tokio::sync::Notify;
+use tokio::sync::{Semaphore, TryAcquireError};
+
+pub fn barrier_example() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    rt.block_on(async {
+        let mut handles = Vec::with_capacity(10);
+        let barrier = Arc::new(Barrier::new(10));
+        for _ in 0..10 {
+            let c = barrier.clone();
+            // The same messages will be printed together.
+            // You will NOT see any interleaving.
+            handles.push(tokio::spawn(async move {
+                println!("before wait");
+                let wait_result = c.wait().await;
+                println!("after wait");
+                wait_result
+            }));
+        }
+
+        // Will not resolve until all "after wait" messages have been printed
+        let mut num_leaders = 0;
+        for handle in handles {
+            let wait_result = handle.await.unwrap();
+            if wait_result.is_leader() {
+                num_leaders += 1;
+            }
+        }
+
+        // Exactly one barrier will resolve as the "leader"
+        assert_eq!(num_leaders, 1);
+    });
+}
+
+pub fn mutex_example() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    rt.block_on(async {
+        let data1 = Arc::new(Mutex::new(0));
+        let data2 = Arc::clone(&data1);
+
+        tokio::spawn(async move {
+            let mut lock = data2.lock().await;
+            *lock += 1;
+        });
+
+        let mut lock = data1.lock().await;
+        *lock += 1;
+    });
+}
+
+pub fn rwlock_example() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    rt.block_on(async {
+        let lock = RwLock::new(5);
+
+        // many reader locks can be held at once
+        {
+            let r1 = lock.read().await;
+            let r2 = lock.read().await;
+            assert_eq!(*r1, 5);
+            assert_eq!(*r2, 5);
+        } // read locks are dropped at this point
+
+        // only one write lock may be held, however
+        {
+            let mut w = lock.write().await;
+            *w += 1;
+            assert_eq!(*w, 6);
+        } // write lock is dropped here
+    });
+}
+
+pub fn semaphore_example() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    rt.block_on(async {
+        let semaphore = Semaphore::new(3);
+
+        let _a_permit = semaphore.acquire().await.unwrap();
+        let _two_permits = semaphore.acquire_many(2).await.unwrap();
+
+        assert_eq!(semaphore.available_permits(), 0);
+
+        let permit_attempt = semaphore.try_acquire();
+        assert_eq!(permit_attempt.err(), Some(TryAcquireError::NoPermits));
+    });
+}
+
+pub fn semaphore_example2() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    rt.block_on(async {
+        let semaphore = Arc::new(Semaphore::new(3));
+        let mut join_handles = Vec::new();
+
+        for _ in 0..5 {
+            let permit = semaphore.clone().acquire_owned().await.unwrap();
+            join_handles.push(tokio::spawn(async move {
+                // perform task...
+                // explicitly own `permit` in the task
+                drop(permit);
+            }));
+        }
+
+        for handle in join_handles {
+            handle.await.unwrap();
+        }
+    });
+}
+
+pub fn notify_example() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    rt.block_on(async {
+        let notify = Arc::new(Notify::new());
+        let notify2 = notify.clone();
+
+        let handle = tokio::spawn(async move {
+            notify2.notified().await;
+            println!("received notification");
+        });
+
+        println!("sending notification");
+        notify.notify_one();
+
+        // Wait for task to receive notification.
+        handle.await.unwrap();
+    });
+}
+
+pub fn notify_example2() {
+    let rt = tokio::runtime::Runtime::new().unwrap();
+
+    rt.block_on(async {
+        let notify = Arc::new(Notify::new());
+        let notify2 = notify.clone();
+
+        let notified1 = notify.notified();
+        let notified2 = notify.notified();
+
+        let _handle = tokio::spawn(async move {
+            println!("sending notifications");
+            notify2.notify_waiters();
+        });
+
+        notified1.await;
+        notified2.await;
+        println!("received notifications");
+    });
+}