本文为 Stjepang 大佬的博客翻译，来自于 bai 的投稿。

• 接口
  
• 将输出传递给JoinHandle
  
• 任务的剖析
  
• 执行器线程
  
• 任务执行
  
• 一点魔法
  
• 改进的JoinHandle
  
• 处理恐慌(panic)
  
• 执行器的效率
  
• 正确性
  
• 适用于所有人的执行器
  
• 转载说明
  
  

接口

fn spawn<F, R>(future: F) -> JoinHandle<R>where    F: Future<Output = R> + Send + 'static,    R: Send + 'static,{    todo!()}

fn main() {    futures::executor::block_on(async {        let handle = spawn(async { 1 + 2 });        assert_eq!(handle.await, 3);    });}
将输出传递给JoinHandle 

type JoinHandle<R> = Pin<Box<dyn Future<Output = R> + Send>>;
这个方法目前可行，但是不要担心，稍后我们会将它作为一个新的结构清晰地重写，并手动实现 Future。

use futures::channel::oneshot;
fn spawn<F, R>(future: F) -> JoinHandle<R>where    F: Future<Output = R> + Send + 'static,    R: Send + 'static,{    let (s, r) = oneshot::channel();    let future = async move {        let _ = s.send(future.await);    };
    todo!()
    Box::pin(async { r.await.unwrap() })}
下一步是在堆上分配future包装器，并将其推入某种全局任务队列，以便由执行程序处理。我们称这种分配的future为一项任务。

任务的剖析

struct Task {    state: AtomicUsize,    future: Mutex<Pin<Box<dyn Future<Output = ()> + Send>>>,}
我们还没有确定状态到底是什么，但它将是某种可以从任何线程更新的 AtomicUsize。我们以后再说吧。

fn spawn<F, R>(future: F) -> JoinHandle<R>where    F: Future<Output = R> + Send + 'static,    R: Send + 'static,{    let (s, r) = oneshot::channel();    let future = async move {        let _ = s.send(future.await);    };
    let task = Arc::new(Task {        state: AtomicUsize::new(0),        future: Mutex::new(Box::pin(future)),    });    QUEUE.send(task).unwrap();
    Box::pin(async { r.await.unwrap() })}
一旦任务被分配，我们将其推入 QUEUE，这是一个包含可运行任务的全局队列。Spawn ()函数现在已经完成，所以让我们接下来定义 QUEUE..。

执行器线程

use crossbeam::channel;use once_cell::sync::Lazy;
static QUEUE: Lazy<channel::Sender<Arc<Task>>> = Lazy::new(|| {    let (sender, receiver) = channel::unbounded::<Arc<Task>>();
    for _ in 0..num_cpus::get().max(1) {        let receiver = receiver.clone();        thread::spawn(move || receiver.iter().for_each(|task| task.run()));    }
    sender});
非常简单——执行器线程实际上是一行代码！任务队列是一个无界通道，而执行器线程则从这个通道接收任务并运行每个任务.

任务执行

impl Task {    fn run(self: Arc<Task>) {        let waker = todo!();
        let cx = &mut Context::from_waker(&waker);        self.future.try_lock().unwrap().as_mut().poll(cx);    }}
请注意，我们需要锁定future，以获得可变访问权并对其进行轮询。根据设计，没有其他线程会同时持有锁，因此try_lock()必须总是成功。

• 如果一个任务已经完成了，然后被唤醒了怎么办? Waker生命周期会超过他关联的Future,并且我们也不想在队列中包含已经完成的任务.
  
• 如果一个任务在运行之前,连续被唤醒两次会怎么样? 我们不希望在队列中同一个任务出现两次.
  
• 如果一个任务正在运行的时候被唤醒了怎么办? 如果这时候将其加入队列中,另一个执行线程可能试图运行它,这将导致一个任务同时在两个线程上运行.
  
  

1. 如果还没有被唤醒并且当前没有正在运行，唤醒函数会安排此任务
   
2. 如果一个任务正在运行时被唤醒,由当前执行器线程(当前正在运行这个future的那个线程)重新调度它.
   
   

impl Task {    fn run(self: Arc<Task>) {        let waker = async_task::waker_fn(|| {            todo!("schedule if the task is not woken already and is not running");        });
        let cx = &mut Context::from_waker(&waker);        self.future.try_lock().unwrap().as_mut().poll(cx);
        todo!("schedule if the task was woken while running");    }}
还记得我们在 Task 中定义的 AtomicUsize 类型的状态字段吗？现在是时候在其中存储一些有用的数据了。关于任务，有两条信息可以帮助我们实现唤醒:

1. 任务是否已经被唤醒
   
2. 任务是否正在运行
   
   

const WOKEN: usize = 0b01;const RUNNING: usize = 0b10;
唤醒函数设置“WOKEN”位。如果两个位先前都是0(即任务既没有被唤醒也没有运行) ，那么我们通过将引用推入队列来调度任务:

let task = self.clone();let waker = async_task::waker_fn(move || {    if task.state.fetch_or(WOKEN, Ordering::SeqCst) == 0 {        QUEUE.send(task.clone()).unwrap();    }});
在轮询future之前，我们取消了WOKEN位的设置，并设置了RUNNING位:

self.state.store(RUNNING, Ordering::SeqCst);let cx = &mut Context::from_waker(&waker);let poll = self.future.try_lock().unwrap().as_mut().poll(cx);

在轮询future之后，我们取消RUNNING位的设置，并检查先前的状态是否已经设置了WOKEN 和 RUNNING 位(即任务在运行时被唤醒)。如果是这样，我们重新安排任务:

一点魔法

type Task = async_task::Task<()>;
fn spawn<F, R>(future: F) -> JoinHandle<R>where    F: Future<Output = R> + Send + 'static,    R: Send + 'static,{    let (task, handle) = async_task::spawn(future, |t| QUEUE.send(t).unwrap(), ());    task.schedule();    Box::pin(async { handle.await.unwrap() })}
async_task::spawn()接受三个参数:

1. 待运行的future
   
2. 一个将任务放入队列的调度函数. 该函数可能被唤醒器执行,也可能被run()在轮询future后执行.
   
3. 一个包含任意信息的tag,这个tag信息会保存在task中. 这篇博客中我们不考虑只是简单的保存(),也就是忽略它.
   
   

1. async_task::Task<()>,其中()就是刚刚传入的tag.
   
2. async_task::JoinHandle<R, ()>,这里的()还是刚刚的tag. 这个JoinHandle是一个future,它完成的时候会返回一个Option<R>. 当返回None的时候表示任务发生了panic或者被取消了.
   
   

static QUEUE: Lazy<channel::Sender<Task>> = Lazy::new(|| {    let (sender, receiver) = channel::unbounded::<Task>();
    for _ in 0..num_cpus::get().max(1) {        let receiver = receiver.clone();        thread::spawn(move || receiver.iter().for_each(|task| task.run()));    }
    sender});
type Task = async_task::Task<()>;type JoinHandle<R> = Pin<Box<dyn Future<Output = R> + Send>>;
fn spawn<F, R>(future: F) -> JoinHandle<R>where    F: Future<Output = R> + Send + 'static,    R: Send + 'static,{    let (task, handle) = async_task::spawn(future, |t| QUEUE.send(t).unwrap(), ());    task.schedule();    Box::pin(async { handle.await.unwrap() })}
完整的代码可以在 v2.rs 中找到。

改进的JoinHandle

type JoinHandle<R> = async_task::JoinHandle<R, ()>;
我们只能将async_task::JoinHandle封装到一个新的结构体中，如果任务发生panic或者被取消，它也会panic:

这句话感觉说不通呢,需要看看async_task源码才行

struct JoinHandle<R>(async_task::JoinHandle<R, ()>);
impl<R> Future for JoinHandle<R> {    type Output = R;
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {        match Pin::new(&mut self.0).poll(cx) {            Poll::Pending => Poll::Pending,            Poll::Ready(output) => Poll::Ready(output.expect("task failed")),        }    }}
完整的执行器实现可以在v3.rs中找到。

处理恐慌(panic)

use std::panic::catch_unwind;
static QUEUE: Lazy<channel::Sender<Task>> = Lazy::new(|| {    let (sender, receiver) = channel::unbounded::<Task>();
    for _ in 0..num_cpus::get().max(1) {        let receiver = receiver.clone();        thread::spawn(move || {            receiver.iter().for_each(|task| {                let _ = catch_unwind(|| task.run());            })        });    }
    sender});
在v4.rs 中可以找到忽略恐慌的完整执行器代码。

• 忽略恐慌  -- 恐慌直接被忽略,当JoinHandle<R>在await时也会发生恐慌
  
• 传播恐慌   panic被重新放入在等待JoinHandle<R>结果的那个任务中.
• 输出恐慌  JoinHandle<R>输出std::thread::Result<R>.

执行器的效率

正确性

适用于所有人的执行器

每个“任务”需要一个分配，结果通常是每个连接需要一个分配。