Rust Concurrency

use std::thread;
use std::time::Duration;

fn main() {
    thread::spawn(|| {                  // 这里创建了一个子线程，spawn 参数是个 closure
        for i in 1..10 {                // 这是子线程执行的代码
            println!("hi number {} from the spawned thread!", i);
            thread::sleep(Duration::from_millis(3));
        }
    });

    for i in 1..5 {                     // 这里主线程执行的代码
        println!("hi number {} from the main thread!", i);
        thread::sleep(Duration::from_millis(3));
    }
}

use std::thread;
use std::time::Duration;

fn main() {
    let handle = thread::spawn(|| {
        for i in 1..10 {
            println!("hi number {} from the spawned thread!", i);
            thread::sleep(Duration::from_millis(3));
        }
    });

    for i in 1..5 {
        println!("hi number {} from the main thread!", i);
        thread::sleep(Duration::from_millis(3));
    }

    handle.join().unwrap();                       // join 表示等待 handle 关联线程执行完
}

use std::thread;

fn main() {
    let i: i32 = 100;
    let v: Vec<i32> = vec![1, 2, 3];

    let handle = thread::spawn(move || {                 // 指定 move，子线程会“复制变量”或者“拿走变量 ownership”
        println!("Here's a int: {:?}", i);               // i32 类型实现了 Copy Trait，i 会直接复制进来
        println!("Here's a vector: {:?}", v);            // Vec<i32> 没有实现 Copy Trait，其 ownership 会转移到子线程
    });

    println!("Main thread, here's a int: {:?}", i);      // 这里还可以使用 i，因为类型 i32 实现了 Copy Trait，它会被复制
    //println!("Main thread, here's a vector: {:?}", v); // 主线程不能使用 v 了，因为类型 Vec<i32> 没有实现 Copy Trait，
                                                         // 从而，ownership 会转移进子线程，主线程中 v 不再可用

    handle.join().unwrap();
}

use std::thread;
use std::time::Duration;
use std::thread::current;

fn main() {
    let builder = thread::Builder::new()
        .name("foo".into())           // 定制线程名称为 foo
        .stack_size(32 * 1024);       // 设置线程栈大小

    let handle = builder.spawn(|| {
        for i in 1..10 {
            println!("hi number {} from the spawned thread {}!", i, current().name().unwrap());
            thread::sleep(Duration::from_millis(3));
        }
    }).unwrap();

    handle.join().unwrap();
}

use std::sync::{Mutex, MutexGuard};

fn main() {                       // 这是个单线程例子，Mutex没啥意义，仅是为了介绍 Mutex API 使用
    let m = Mutex::new(5);

    {
        let mut num: MutexGuard<i32> = m.lock().unwrap();    // 使用数据前需要获取锁
        *num = 6;
    }                             // num 在这里会离开作用域，自动释放锁

    println!("m = {:?}", m);      // 输出 m = Mutex { data: 6 }
}

// From https://riptutorial.com/rust/example/24527/read-write-locks
use std::time::Duration;
use std::thread;
use std::thread::sleep;
use std::sync::{Arc, RwLock };

fn main() {
    // Create an u32 with an inital value of 0
    let initial_value = 0u32;

    // Move the initial value into the read-write lock which is wrapped into an atomic reference
    // counter in order to allow safe sharing.
    let rw_lock = Arc::new(RwLock::new(initial_value));

    let producer_lock = rw_lock.clone();
    let producer_thread = thread::spawn(move || {
        loop {
            if let Ok(mut write_guard) = producer_lock.write() { // 用 write 获得写锁，没拿到锁就 block 直接拿到
                // the returned write_guard implements `Deref` giving us easy access to the target value
                *write_guard += 1;

                sleep(Duration::from_millis(600));
                println!("Updated value: {}", *write_guard);
            }  // write_guard 超过作用域会自动释放锁

            sleep(Duration::from_millis(1000));
        }
    });

    let consumer_id_lock = rw_lock.clone();
    let consumer_id_thread = thread::spawn(move || {
        loop {
            if let Ok(read_guard) = consumer_id_lock.read() { // 用 read 获得读锁，没拿到锁就 block 直接拿到
                // the returned read_guard also implements `Deref`
                println!("Read value: {}", *read_guard);
            }  // read_guard 超过作用域会自动释放锁

            sleep(Duration::from_millis(1000));
        }
    });

    let consumer_square_lock = rw_lock.clone();
    let consumer_square_thread = thread::spawn(move || {
        loop {
            if let Ok(read_guard) = consumer_square_lock.try_read() { // try_read 获得读锁，没拿到锁就返回错
                let value = *read_guard;
                println!("Read value squared: {}", value * value);
            } else { // read_guard 超过作用域会自动释放锁
                println!("The try_read() get read lock failed");
            }

            sleep(Duration::from_millis(1000));
        }
    });

    let _ = producer_thread.join();
    let _ = consumer_id_thread.join();
    let _ = consumer_square_thread.join();
}

use std::sync::Once;

static mut VAL: usize = 0;
static INIT: Once = Once::new();

// Accessing a `static mut` is unsafe much of the time, but if we do so
// in a synchronized fashion (e.g., write once or read all) then we're
// good to go!
//
// This function will only call `expensive_computation` once, and will
// otherwise always return the value returned from the first invocation.
fn get_cached_val() -> usize {
    unsafe {
        INIT.call_once(|| {
            VAL = expensive_computation();
        });
        VAL
    }
}

fn expensive_computation() -> usize {
    // ...
    println!("expensive_computation is called");
    100
}

fn main() {
    println!("{}", get_cached_val());
    println!("{}", get_cached_val());
}

use std::sync::{Arc, Barrier};
use std::thread;

fn main() {
    let nthreads = 10;
    let mut handles = Vec::with_capacity(nthreads);
    let barrier = Arc::new(Barrier::new(nthreads));

    for _ in 0..nthreads {
        let b = Arc::clone(&barrier);
        // The same messages will be printed together.
        // You will NOT see any interleaving.
        handles.push(thread::spawn(move || {
            println!("before wait");
            b.wait();                       // barrier 上调用 wait 表示所有线程都运行到这个点后，才往下运行
            println!("after wait");
        }));
    }

    // Wait for other threads to finish.
    for handle in handles {
        handle.join().unwrap();
    }
}

use std::thread;
use std::sync::mpsc;

fn main() {
    let (tx, rx) = mpsc::channel();           // 创建一个通道，tx 为发送者，rx 为接收者

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();                // 调用 send 方法往通道中发送数据
    });

    let received = rx.recv().unwrap();        // 调用 recv 方法从通道中读数据
    println!("Got: {}", received);            // Got: hi
}

    for received in rx {                      // 不用显式调用 recv 函数，而是将 rx 当作一个迭代器
        println!("Got: {}", received);
    }

use std::net::{TcpListener, TcpStream};
use std::io::{Read, Write};

fn handle_connection(mut stream: TcpStream) {
    let mut buffer = [0; 512];
    stream.read(&mut buffer).unwrap();             // 从流中读取数据
    println!("Get request: [{}]", String::from_utf8_lossy(&buffer[..])); // 打印出请求数据

    let response = "HTTP/1.1 200 OK\r\n\r\nThis is a Rust Web Server";
    stream.write(response.as_bytes()).unwrap();    // 往流中写入数据
    stream.flush().unwrap();
}

fn main() {
    let listener = TcpListener::bind("127.0.0.1:7878").unwrap();

    for stream in listener.incoming() {
        let stream: TcpStream = stream.unwrap();

        handle_connection(stream);
    }
}

use std::net::{TcpListener, TcpStream};
use std::io::{Read, Write};
use std::sync::{mpsc, Arc, Mutex};
use std::thread;

pub struct ThreadPool {
    workers: Vec<Worker>,            // 每个 worker 对应一个线程
    sender: mpsc::Sender<Message>,   // 通道的发送端，提交任务就是往通道的发送端写入数据
}

type Job = Box<dyn FnOnce() + Send + 'static>;

enum Message {
    NewJob(Job),  // 表示新任务
    Terminate,    // 表示没有任务了，终止执行。暂时没用
}

impl ThreadPool {
    /// Create a new ThreadPool.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        // 创建一个通道
        // 通道“发送端”给 execute，通道“接收端”给每个 worker（一个 worker 对应一个线程）
        // 通道“接收端”被多个线程访问，所以需要使用使用 Mutex 和 Arc
        // Arc 使得多个 worker 拥有接收端，而 Mutex 可确保一次只有一个 worker 能从接收端得到任务。
        let (sender, receiver) = mpsc::channel();
        let receiver = Arc::new(Mutex::new(receiver));
        let mut workers = Vec::with_capacity(size);
        for id in 0..size {
            workers.push(Worker::new(id, Arc::clone(&receiver)));
        }
        ThreadPool { workers, sender }
    }

    /// 往通道中写入任务
    pub fn execute<F>(&self, f: F)
        where
            F: FnOnce() + Send + 'static,         // 参考 thread::spawn 的签名
    {
        let job = Box::new(f);
        self.sender.send(Message::NewJob(job)).unwrap();
    }
}

struct Worker {
    id: usize,
    thread: Option<thread::JoinHandle<()>>,
}

impl Worker {
    // 创建 worker，一个 worker 对应一个线程，线程在死循环中不停查看通道中是否有任务，有就执行
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Message>>>) -> Worker {
        let thread = thread::spawn(move ||
            loop {
                let message = receiver.lock().unwrap().recv().unwrap(); // lock: 访问通道前先拿锁
                                                                        // recv: 通道中有数据就接收，没数据就 block

                match message {
                    Message::NewJob(job) => {
                        println!("Worker {} got a job; executing.", id);

                        job();
                    }
                    Message::Terminate => {
                        println!("Worker {} was told to terminate.", id);

                        break;
                    }
                }
            });

        Worker {
            id,
            thread: Some(thread),
        }
    }
}

fn handle_connection(mut stream: TcpStream) {
    let mut buffer = [0; 512];
    stream.read(&mut buffer).unwrap();             // 从流中读取数据
    println!("Get request: [{}]", String::from_utf8_lossy(&buffer[..]));

    let response = "HTTP/1.1 200 OK\r\n\r\nThis is a Rust Web Server";
    stream.write(response.as_bytes()).unwrap();    // 往流中写入数据
    stream.flush().unwrap();
}

fn main() {
    let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
    let pool = ThreadPool::new(4);

    for stream in listener.incoming() {
        let stream = stream.unwrap();

        pool.execute(|| {               // 提交闭包到线程池
            handle_connection(stream);
        });
    }
}

use std::future::Future;
use std::pin::Pin;
use std::task::{Poll, Context};
use futures::executor::block_on;

struct IntFuture {
    x: i32,
}

impl Future for IntFuture {
    type Output = i32;

    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<i32> {
        Poll::Ready(self.x + 100)
    }
}

fn main() {
    let f = IntFuture{ x: 30 };

    let result = block_on(f);             // 阻塞当前线程直到 Future f 到达 Ready 状态

    assert_eq!(result, 130);
}

struct IntFuture {
    x: i32,
}

impl Future for IntFuture {
    type Output = i32;

    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<i32> {
        Poll::Ready(self.x + 100)
    }
}

let f = IntFuture{ x: 30 };

async fn int_future(x: i32) -> i32 {
    x + 100
}

let f = int_future(30);

use std::future::Future;
use futures::executor::block_on;

async fn int_future(x: i32) -> i32 {
    x + 100
}

fn main() {
    let f = int_future(30);

    let result = block_on(f);

    assert_eq!(result, 130);
}

use std::future::Future;
use futures::executor::block_on;

struct Song {}

async fn learn_song() -> Song { Song{} }
async fn sing_song(song: Song) { /* ... */ }
async fn dance() { /* ... */ }

fn main() {
    let song = block_on(learn_song());
    block_on(sing_song(song));
    block_on(dance());
}

use std::future::Future;
use futures::executor::block_on;

struct Song {}

async fn learn_song() -> Song { Song{} }
async fn sing_song(song: Song) { /* ... */ }
async fn dance() { /* ... */ }

async fn learn_and_sing() {
    // Wait until the song has been learned before singing it.
    // We use `.await` here rather than `block_on` to prevent blocking the
    // thread, which makes it possible to `dance` at the same time.
    let song = learn_song().await;
    sing_song(song).await;
}

async fn async_main() {
    let f1 = learn_and_sing();
    let f2 = dance();

    // `join!` is like `.await` but can wait for multiple futures concurrently.
    // If we're temporarily blocked in the `learn_and_sing` future, the `dance`
    // future will take over the current thread. If `dance` becomes blocked,
    // `learn_and_sing` can take back over. If both futures are blocked, then
    // `async_main` is blocked and will yield to the executor.
    futures::join!(f1, f2);
}

fn main() {
    block_on(async_main());
}

#[derive(Debug)]
struct Test {
    a: String,
    b: *const String,
}

impl Test {
    fn new(txt: &str) -> Self {
        Test {
            a: String::from(txt),
            b: std::ptr::null(),
        }
    }

    fn init(&mut self) {
        let self_ref: *const String = &self.a;
        self.b = self_ref;    // 这样，b 指向了 a，Test 是自引用结构
    }

    fn a(&self) -> &str {
        &self.a
    }

    fn b(&self) -> &String {
        assert!(!self.b.is_null(), "Test::b called without Test::init being called first");
        unsafe { &*(self.b) }
    }
}

fn main() {
    let mut test1 = Test::new("test1");
    test1.init();
    let mut test2 = Test::new("test2");
    test2.init();

    println!("a: {}, b: {}", test1.a(), test1.b());
    println!("a: {}, b: {}", test2.a(), test2.b());
    std::mem::swap(&mut test1, &mut test2);             // 交换 test1 和 test2
    println!("a: {}, b: {}", test1.a(), test1.b());
    println!("a: {}, b: {}", test2.a(), test2.b());
}

use futures::future;
use futures::select;
use futures::executor::block_on;

async fn async_main() {
    let mut a = future::ready(4);
    let mut b = future::pending::<()>();

    let res = select! {
    a_res = a => { println!("future a is ready, {:?}", a_res); a_res + 1 },
    b_res = b => { println!("future b is ready, {:?}", b_res); 0 },
    };

    assert_eq!(res, 5);
}

fn main() {
    block_on(async_main())
}