Rust异步浅谈

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

pub enum Poll<T> {
    Ready(T),
    Pending,
}

pub trait Future {
    type Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

1
2
3
4
5
6
7
8

pub struct Context<'a> {
    waker: &'a Waker,
    // Ensure we future-proof against variance changes by forcing
    // the lifetime to be invariant (argument-position lifetimes
    // are contravariant while return-position lifetimes are
    // covariant).
    _marker: PhantomData<fn(&'a ()) -> &'a ()>,
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

pub struct Waker {
    waker: RawWaker,
}

impl Waker {
    /// 唤醒绑定在 Waker 上的数据，通常是 Future
    pub fn wake(self) {}
    pub fn wake_by_ref(&self) {}
    pub fn will_wake(&self, other: &Waker) -> bool {}

    pub unsafe fn from_raw(waker: RawWaker) -> Waker {}
}

pub struct RawWaker {
    /// A data pointer, which can be used to store arbitrary data as required
    /// by the executor. This could be e.g. a type-erased pointer to an `Arc`
    /// that is associated with the task.
    /// The value of this field gets passed to all functions that are part of
    /// the vtable as the first parameter.
    data: *const (),
    /// Virtual function pointer table that customizes the behavior of this waker.
    vtable: &'static RawWakerVTable,
}

/// RawWaker 行为的虚函数表
pub struct RawWakerVTable {
    clone: unsafe fn(*const ()) -> RawWaker,
    wake: unsafe fn(*const ()),
    wake_by_ref: unsafe fn(*const ()),
    drop: unsafe fn(*const ()),
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

sequenceDiagram
    participant Executor
    participant Reactor
    activate Executor
    Executor->>Reactor: Pending Future
    deactivate Executor
    Note left of Executor: Execute other Future
    activate Reactor
    Reactor->>Executor: Ready Future
    deactivate Reactor
    activate Executor
    deactivate Executor

1
2
3
4
5
6

async fn read_and_write(s: TcpStream) {
  let (mut r, mut w) = s.split();
  let mut buffer = r.read().await.unwrap();
  buffer.append("Hello,world");
  w.write_all(buffer.as_bytes()).await.unwrap();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

sequenceDiagram
    participant Executor
    participant Reactor
    activate Executor
    deactivate Executor
    Executor->>Reactor: Pending on r.read()
    Note left of Executor: Execute other Future
    activate Reactor
    Reactor->>Executor: r.read() is ready
    Note left of Executor: Execute current Future
    deactivate Reactor
    Executor->>Reactor: Pending on w.write_all()
    Note left of Executor: Execute other Future
    activate Reactor
    deactivate Reactor
    Reactor->>Executor: w.write_all() is ready

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109

/// The state of an executor.
struct Pool {
    /// 全局任务队列
    injector: Injector<Runnable>,

    /// 线程的本地队列，用来进行任务的偷取
    stealers: Vec<Stealer<Runnable>>,

    /// 存放空闲的线程，用来后续的唤醒并执行任务
    sleepers: Sleepers,
}

// 全局的线程池
static POOL: Lazy<Pool> = Lazy::new(|| {
    let num_threads = num_cpus::get().max(1);
    let mut stealers = Vec::new();

    // Spawn worker threads.
    for _ in 0..num_threads {
        let worker = Worker::new_fifo();
        stealers.push(worker.stealer());

        let proc = Processor {
            worker,
            slot: Cell::new(None),
            slot_runs: Cell::new(0),
        };

        thread::Builder::new()
            .name("async-std/executor".to_string())
            .spawn(|| {
                let _ = PROCESSOR.with(|p| p.set(proc));
                abort_on_panic(main_loop);
            })
            .expect("cannot start a thread driving tasks");
    }

    Pool {
        injector: Injector::new(),
        stealers,
        sleepers: Sleepers::new(),
    }
});

/// 工作线程的状态
struct Processor {
    /// 本地任务队列
    worker: Worker<Runnable>,

    /// 存放了比本地队列中任务优先级更高的任务，通常第一次spawn会放到这里，
    /// 执行一次poll来快速判断状态，对于无阻塞的任务更高效，不需要等待。
    slot: Cell<Option<Runnable>>,

    /// How many times in a row tasks have been taked from the slot rather than the queue.
    slot_runs: Cell<u32>,
}

fn main_loop() {
    loop {
        match find_runnable() {
            Some(task) => task.run();
            None => {
                // 实际上，这里根据空循环的次数，会陷入睡眠状态或出让CPU资源，直到新的task来唤醒。
            }
        }
    }
}

fn find_runnable() -> Option<Task> {
    // 优先从本地的队列中获取
    let task = get_local();
    if task.is_some() {
        return task;
    }
    // 其次从全局队列中获取
    let task = get_global();
    if task.is_some() {
        return task;
    }
    // 最后尝试从其他线程的本地队列中偷取
    steal_other()
}

/// 安排新的任务到Executor的执行队列中
pub(crate) fn schedule(task: Runnable) {
    PROCESSOR.with(|proc| {
        // If the current thread is a worker thread, store it into its task slot or push it into
        // its local task queue. Otherwise, push it into the global task queue.
        match proc.get() {
            // 如果当前线程为worker线程，插入到当前线程的第一优先级任务槽
            Some(proc) => {
                // Replace the task in the slot.
                if let Some(task) = proc.slot.replace(Some(task)) {
                    // 尝试把任务的优先级提升到最高，并把上一个优先级最高的任务放到当前线程任务队列
                    // If the slot already contained a task, push it into the local task queue.
                    proc.worker.push(task);
                    POOL.sleepers.notify_one();
                }
            }
            // 如果当前线程不是worker线程的话，放到全局队列
            None => {
                // 将任务放到全局队列中
                POOL.injector.push(task);
                // 尝试唤醒一个睡眠的worker线程
                POOL.sleepers.notify_one();
            }
        }
    })
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

loop {
    poll.poll(&events, timeout);
    for event in events.iter() {
        if (event.is_readable()) {
            for waker in event.readers.wakers {
                waker.wake();
            }
        }
        if (event.is_writeable()) {
            for waker in event.writers.wakers {
                waker.wake();
            }
        }
    }
}

1
2
3
4
5

pub trait Stream {
    type Item;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>;
}

1
2
3
4
5

pub trait Iterator {
    type Item;

    fn next(&mut self) -> Option<Self::Item>;
}

1
2
3
4
5
6
7
8

pub trait Sink<Item> {
    type Error;

    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
    fn start_send(self: Pin<&mut Self>, item: Item) -> Result<(), Self::Error>;
    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

pub fn get_next_day() -> Time {
    thread::sleep(Duration::hour * 24);
    current_time()
}

pub fn run_every_hour() {
    for thread::sleep(Duration::hour) {
        do_something();
    }
}

1
2
3
4
5
6
7

loop {
    for f in futures {
        if f.is_ready() {
            return f.output();
        }
    }
}

1
2
3
4
5
6

loop {
    match f::poll(cx) {
        Poll::Ready(x) => return x;
        Poll::Pending => {}
    }
}

1

let x = f.await;