并发编程

// 进程控制块（PCB）简化结构：用于保存进程的运行上下文以及相关管理信息
pub struct ProcessControlBlock {
    pid: u16,
    state: ProcessState,

    // CPU 上下文（Context）：用于在进程切换（context switch）时
    // 保存和恢复当前进程的执行现场，确保恢复后能够从中断位置继续执行
    program_counter: u16,
    general_purpose_registers: [u8; 4],
    instruction_register: u8,
    flags: [u1, 3],
    stack_pointer: u16,
    index_registers: [u16; 2],
    ...

    // 内存管理信息：描述进程的虚拟地址空间边界，
    // 用于实现进程间的内存隔离，保证各进程地址空间相互独立
    memory_limits: [u16; 2],
    
    ...
    io_devices: Vec<IO_Device>, 
    open_files: Vec<File>, 
    
    parent: Arc<ProcessControlBlock>, 
    children: Vec<Arc<ProcessControlBlock>>
}

pub enum ProcessState {
	New, 
	Running, 
	Ready,
	Waiting, 
	Terminated
}

pub struct ProcessControlBlock {
	pid: u16, 
	memory_limits: [u16; 2], 
	io_devices: Vec<IO_Device>, 
	open_files: Vec<File>, 
	parent: Arc<ProcessControlBlock>, 
    children: Vec<Arc<ProcessControlBlock>>, 
    threads: Vec<Thread_t>, 
}

pub struct Thread_t {
	tid: u16, 
	state: ProcessState, 
	program_counter: u16, 
	general_purpose_registers: [u8; 4],
    instruction_register: u8,
    flags: [u1, 3],
    stack_pointer: u16,
    index_registers: [u16; 2],
    // process_ref: Arc<ProcessControlBlock>, 
}

use std::thread;

fn main() {
    let handler = thread::spawn(|| {
        for i in 1..10 {
            println!("worker: {i}");
        }
    });

    for i in 100..110 {
        println!("main: {i}");
    }
    
    handler.join().unwrap();
}

use futures::executor::LocalPool;
use futures::task::LocalSpawnExt;

async fn first_task() {
    for i in 1..5 {
        println!("first task: {i}");
        // 主动让出执行权，让其他任务有机会运行
        tokio::task::yield_now().await;
    }
}

async fn second_task() {
    for i in 100..105 {
        println!("second task: {i}");
        tokio::task::yield_now().await;
    }
}

#[tokio::main]
async fn main() {
    // 两个任务在同一个线程上交替执行
    tokio::join!(first_task(), second_task());
}

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

fn critical_section(v: Arc<Mutex<i32>>, change_v: i32) {
    // 第一段加锁
    let mut lock = v.lock().unwrap();
    *lock = change_v;
    println!("{}", *lock);

    // 显式释放锁
    drop(lock);

    // 在此期间，其他线程可以获得锁并修改 v

    // 第二段再次加锁
    let mut lock = v.lock().unwrap();
    *lock += 1;
    println!("{}", *lock);
}

fn main() {

    for round in 0..10 {
        println!("\n=== round {} ===", round);

        let v = Arc::new(Mutex::new(1));
        let v1 = Arc::clone(&v);
        let v2 = Arc::clone(&v);

        let t1 = thread::spawn(move || {
            critical_section(v1, 10);
        });

        let t2 = thread::spawn(move || {
            critical_section(v2, 100);
        });

        t1.join().unwrap();
        t2.join().unwrap();
    }
}

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

// 一个最简单的"主动让出一次"的 Future
struct YieldOnce {
    yielded: bool,
}

impl Future for YieldOnce {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        if self.yielded {
            Poll::Ready(())
        } else {
            self.yielded = true;
            cx.waker().wake_by_ref(); // 重新唤醒自己，等下一轮再执行
            Poll::Pending
        }
    }
}

fn yield_once() -> YieldOnce {
    YieldOnce { yielded: false }
}

async fn first_task() {
    for i in 1..10 {
        println!("hi number {} from the first task!", i);
        yield_once().await; // 主动让出执行权
    }
}

async fn second_task() {
    for i in 100..110 {
        println!("hi number {} from the second task!", i);
        yield_once().await; // 主动让出执行权
    }
}

fn main() {
    let mut pool = LocalPool::new();
    let spawner = pool.spawner();

    spawner.spawn_local(first_task()).unwrap();
    spawner.spawn_local(second_task()).unwrap();

    pool.run();
}

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

fn main() {
    // 标准库 mpsc：多生产者，单消费者
    let (tx, rx) = mpsc::channel::<i32>();

    // 因为 Receiver 不能 clone，所以用 Arc<Mutex<Receiver>>
    // 让多个消费者线程共享同一个接收端
    let rx = Arc::new(Mutex::new(rx));

    // 生产者
    let producer_tx = tx.clone();
    let producer = thread::spawn(move || {
        for i in 0..10 {
            // thread::sleep(Duration::from_millis(900));
            println!("producing {}", i);

            if producer_tx.send(i).is_err() {
                println!("receiver dropped, producer exits");
                break;
            }
        }
    });

    // 两个消费者
    let mut consumers = Vec::new();

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

        let consumer = thread::spawn(move || {
            loop {
                // 先拿锁，再 recv 一条消息
                let first_item = {
                    let guard = rx_clone.lock().unwrap();
                    guard.recv()
                };

                match first_item {
                    Ok(v) => {
                        // 模拟消费者处理较慢
                        // thread::sleep(Duration::from_millis(1000));
                        println!("consumer {} consuming {}", id, v);

                        // 尝试继续把当前队列里已经到达的消息一并取走
                        loop {
                            let next_item = {
                                let guard = rx_clone.lock().unwrap();
                                guard.try_recv()
                            };

                            match next_item {
                                Ok(x) => {
                                    println!("consumer {} consuming {}", id, x);
                                }
                                Err(mpsc::TryRecvError::Empty) => {
                                    break;
                                }
                                Err(mpsc::TryRecvError::Disconnected) => {
                                    println!("consumer {} exits", id);
                                    return;
                                }
                            }
                        }
                    }
                    Err(_) => {
                        println!("consumer {} exits", id);
                        break;
                    }
                }
            }
        });

        consumers.push(consumer);
    }

    producer.join().unwrap();
    for c in consumers {
        c.join().unwrap();
    }
}