共享资源、消息传递与并发控制

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

fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = Vec::new();

    for _ in 0..4 {
        let counter = Arc::clone(&counter);

        let handle = thread::spawn(move || {
            for _ in 0..1000 {
                let mut value = counter.lock().unwrap();
                *value += 1;
            }
        });

        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("{}", *counter.lock().unwrap());
}

use std::collections::HashMap;
use std::sync::{Arc, RwLock};
use std::thread;

fn main() {
    let config = Arc::new(RwLock::new(HashMap::from([
        ("timeout".to_string(), "30s".to_string()),
        ("region".to_string(), "ap-east".to_string()),
    ])));

    let mut readers = Vec::new();

    for _ in 0..3 {
        let config = Arc::clone(&config);

        readers.push(thread::spawn(move || {
            let guard = config.read().unwrap();
            println!("region = {:?}", guard.get("region"));
        }));
    }

    for reader in readers {
        reader.join().unwrap();
    }

    let writer_config = Arc::clone(&config);
    let writer = thread::spawn(move || {
        let mut guard = writer_config.write().unwrap();
        guard.insert("timeout".to_string(), "60s".to_string());
    });

    writer.join().unwrap();
}

use std::sync::atomic::{AtomicUsize, Ordering};

static REQUESTS: AtomicUsize = AtomicUsize::new(0);

fn read_requests() -> usize {
    REQUESTS.load(Ordering::Relaxed)
}

fn record_request() {
    REQUESTS.fetch_add(1, Ordering::Relaxed);
}

#include <condition_variable>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>

std::mutex mutex;
std::condition_variable condition;
std::queue<int> tasks;
bool done = false;

void producer(int id) {
    for (int i = 0; i < 5; ++i) {
        {
            std::lock_guard<std::mutex> lock(mutex);
            tasks.push(id * 100 + i);
        }

        condition.notify_one();
    }
}

void consumer(int id) {
    while (true) {
        std::unique_lock<std::mutex> lock(mutex);

        while (tasks.empty() && !done) {
            condition.wait(lock);
        }

        if (tasks.empty() && done) {
            break;
        }

        int task = tasks.front();
        tasks.pop();

        lock.unlock();

        std::cout << "consumer " << id << " handles " << task << '\n';
    }
}

int main() {
    std::vector<std::thread> producers;
    std::vector<std::thread> consumers;

    for (int i = 0; i < 2; ++i) {
        producers.emplace_back(producer, i);
    }

    for (int i = 0; i < 2; ++i) {
        consumers.emplace_back(consumer, i);
    }

    for (auto& thread : producers) {
        thread.join();
    }

    {
        std::lock_guard<std::mutex> lock(mutex);
        done = true;
    }

    condition.notify_all();

    for (auto& thread : consumers) {
        thread.join();
    }
}

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

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

    for worker_id in 0..3 {
        let barrier = Arc::clone(&barrier);

        handles.push(thread::spawn(move || {
            println!("worker {worker_id} starts phase 1");

            thread::sleep(Duration::from_millis(100 * worker_id));

            println!("worker {worker_id} waits at barrier");
            barrier.wait();

            println!("worker {worker_id} starts phase 2");
        }));
    }

    for handle in handles {
        handle.join().unwrap();
    }
}

use std::sync::Arc;
use tokio::sync::Semaphore;

async fn fetch_with_limit(semaphore: Arc<Semaphore>) {
    let _permit = semaphore.acquire().await.unwrap();

    // 同一时间最多只有固定数量的任务能进入这里
    fetch_remote_data().await;
}

use std::collections::HashMap;
use std::sync::Mutex;
use tokio::sync::Semaphore;

struct JobStatus;

struct AppState {
    jobs: Mutex<HashMap<String, JobStatus>>,
    gpu_slots: Semaphore,
}

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

fn main() {
    let (tx, rx) = mpsc::channel();

    let worker = thread::spawn(move || {
        while let Ok(message) = rx.recv() {
            println!("worker received: {message}");
        }
    });

    tx.send("compile").unwrap();
    tx.send("test").unwrap();
    tx.send("deploy").unwrap();

    drop(tx);
    worker.join().unwrap();
}

use std::sync::{mpsc, Arc, Mutex};
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel::<i32>();
    let rx = Arc::new(Mutex::new(rx));

    let mut producers = Vec::new();

    for producer_id in 0..2 {
        let tx = tx.clone();

        producers.push(thread::spawn(move || {
            for task_id in 0..5 {
                let task = producer_id * 100 + task_id;
                tx.send(task).unwrap();
            }
        }));
    }

    drop(tx);

    let mut consumers = Vec::new();

    for consumer_id in 0..2 {
        let rx = Arc::clone(&rx);

        consumers.push(thread::spawn(move || {
            loop {
                let message = {
                    let receiver = rx.lock().unwrap();
                    receiver.recv()
                };

                match message {
                    Ok(task) => {
                        println!("consumer {consumer_id} handles {task}");
                    }
                    Err(_) => break,
                }
            }
        }));
    }

    for producer in producers {
        producer.join().unwrap();
    }

    for consumer in consumers {
        consumer.join().unwrap();
    }
}