advance/async/future-excuting

[giscus[bot]]

advance/async/future-excuting

https://course.rs/advance/async/future-excuting.html

[Rustln]

目前 Rust 缺少 Quartz 一样的大杀器，结合 Future 和 执行器可以搞一个，coding 中

[Kish29]

不是有吗？tokio-cron- scheduler

[minstrel1977]

大佬帮忙过来看看，下面我对文中的这段代码的注释加了我的理解，请帮忙看看有没有问题：

impl Executor {
    fn run(&self) {
        while let Ok(task) = self.ready_queue.recv() { // 如果发送端还在，且ready_queue里没有任务，主线程就在此阻塞
            // 获取一个future，若它还没有完成(仍然是Some，不是None)，则对它进行一次poll并尝试完成它
            let mut future_slot = task.future.lock().unwrap();
            if let Some(mut future) = future_slot.take() {
                // 基于任务自身创建一个 `LocalWaker`
                let waker = waker_ref(&task);
                let context = &mut Context::from_waker(&*waker);
                // `BoxFuture<T>`是`Pin<Box<dyn Future<Output = T> + Send + 'static>>`的类型别名
                // 通过调用`as_mut`方法，可以将上面的类型转换成`Pin<&mut dyn Future + Send + 'static>`
                // future是spawner.spawn的整个async块。第一次poll推动执行了块内await前的同步代码，即打印出"任务开始"
                // 同步代码执行完TimerFuture::new后返回一个类型为TimerFuture的future,
                // 之后的.await驱动执行TimerFuture的poll方法，顺便通过TimerFuture的poll方法把下面future.as_mut().poll(context)传入的context(包含了waker)接续传入TimerFuture内部
                // 第一次对TimerFuture做poll返回了Poll::Pending，也就是在.await处返回Poll::Pending
                // 那次轮执行外层task就在.await处保存好整个外层task的栈现场（task内.await后的代码不再执行），然后让出主线程的控制权.
                // 主线程不可能被挂起,所以继续执行后续同步代码。
                // 因为此轮while循环外层future的第一次poll返回了Poll::Pending,所以继续执行if块内部代码
                if future.as_mut().poll(context).is_pending() {
                    // Future还没执行完，因此将它放回任务中，等待下次被poll
                    // 之所以再放回，是因为前面用了take()。但如果前面不用take()，这里也不用放回了，是否可行？
                    *future_slot = Some(future);
                }
            }
        }
    }
}

fn main() {
    let (executor,spawner) = new_executor_and_spawner();
    // 将外层task发送到 executor 和 spawner 所在的同步通道中
    spawner.spawn(async {
        println!("howdy!");
        //等待TimerFuture睡醒后执行外层task的wake_by_ref将此外层task再次发送到此同步通道中
        TimerFuture::new(Duration::from_secs(2)).await; 
        println!("done!");
    });
    drop(spawner); // 因为spawn里复制了一份发送端给外层task，所以spawner被删除后，上面建立的同步通道仍在，下面一句仍能正常执行
    executor.run();
}

[CtrlZ233]

我的理解，这里必须在判断pending的时候把future放回，因为此时waker通过Arc保存了一份Task引用，如果不放回的话，在调用wake的时候会找不到之前的task信息。

[h4cnull]

前面需要take()，获取到future才能poll，future没有完成，就要放回去。当future完成了，触发waker，wake的时候会将task再次发送到channel，这时候executor再次收到task，再去poll future，future不是pending了，就不再放回task。

[TsGanT]

大佬们 有个问题呀 不是很能理解，啥时候在第二次poll的时候把task又放回channel了？感觉没有这个体现并且 wake函数也没有看到往channel里发送的task的迹象呀？

[MuckToobu]

大佬们 有个问题呀 不是很能理解，啥时候在第二次poll的时候把task又放回channel了？感觉没有这个体现并且 wake函数也没有看到往channel里发送的task的迹象呀？

@TsGanT 通过实现ArcWake，然后内部调用wake函数的时候会调用ArcWake::wake_by_ref，这个时候发送的task

[googya]

我觉得作者的注释

                // Future还没执行完，因此将它放回任务中，等待下次被poll

放在那里有点问题

      if let Some(waker) = shared_state.waker.take() {
        waker.wake();
      }

在这里调用了 wake 后， 会触发

    fn wake_by_ref(arc_self: &Arc<Self>) {
        // 通过发送任务到任务管道的方式来实现`wake`，这样`wake`后，任务就能被执行器`poll`
        let cloned = arc_self.clone();
        arc_self
            .task_sender
            .send(cloned)
            .expect("任务队列已满");
    }

这样才能重新入对

[zhangdekui]

有一点不太明白,
Executor中, while let Ok(task) = self.ready_queue.recv(),
数据是怎么跑到ready_queue里面去的, 没看到发送代码

[zhangdekui]

let (task_sender, ready_queue) = sync_channel(MAX_QUEUED_TASKS);
我好像知道啦, 他俩是一对儿~

[wangzhipeng]

将

    let f = async {
        println!("howdy!");
        // 创建定时器Future，并等待它完成
        TimerFuture::new(Duration::new(2, 0)).await;
        println!("done!");
    };

改为

 let f = TimerFuture::new(Duration::new(2, 0));

是不是更好理解 async .await，不然让他们在包裹了一层就很困惑， 这个拆解完语法糖怎么还需要语法糖

[wangzhipeng]

然后在
···
fn poll(self: Pin<&mut Self>, cx: &mut Context<'>) -> PollSelf::Output {
···
后面加上
···
fn poll(self: Pin<&mut Self>, cx: &mut Context<'>) -> PollSelf::Output {
println!("i am running");
···
再次执行，感觉就比较明白了

[chiehw]

这里是两个 Future 组合，然后底层涉及 Waker 的注册，我还在看这一部分，但是做来组合 Future 的案例还不错。

相关的文章：https://zhuanlan.zhihu.com/p/66028983

[wasd96040501]

我有一个问题，那如果我们自己写的 async 函数就是纯计算函数（不涉及系统 IO，极端点，一个死循环的 for，什么都不干），那是不是这个函数除非运行完，否则完全不能被调度（暂停）？

[ruaruagerry]

async里面没有await，那你写的async没有任何意义。只不过async这个整体作为一个future，但是在async内部运行时是不会产生任何调度的。

[ruaruagerry]

梳理了下整个executor + TimerFuture的流程：
1.executor run推出async，poll执行async同步代码，输出howdy
2.executor run poll内继续执行，遇到TimerFuture，执行TimerFuture poll，返回pending状态
3.TimerFuture执行完毕，awake唤醒executor run，再次执行executor run poll，然后进入TimerFuture poll
4.TimerFuture poll直接返回ready状态，executor run poll得以继续async内await下面的代码，输出done
5.输出done后，此时executor run poll才算执行完毕，返回ready状态，完成

总结一下几个要点：

1. async作为一个future，每次poll遇到await停止，进入await方法里的poll
2. await方法内awake唤醒的是上层调度逻辑，而不是await方法本身
3. 唤醒调度逻辑后会继续在上层调度逻辑进行poll，然后再次进入await方法里的poll直到该await方法ready
4. 当检测到await方法ready，那么该await方法就相当于poll内的同步代码了，直接往下走

再次总结一下async内碰到await的逻辑：

1. async碰到await，会沉入到await方法内的poll
2. await poll pending，那么记录当前future，等待唤醒
3. await poll ready，则await执行完毕，继续同步代码直到再次await

这只是根据这个例子总结的机制，至于rust内部是否是这样的，还需再了解，持续学习。

[evanzp0]

想问下各位大佬，在 Executor.run( ) 中

while let Ok(task) = self.ready_queue.recv() {
...
}

这个 task 在 while 的第一次循环后，ready_queue 就应该是空的了，那 task 不是应该就 drop 了吗，这个 task 怎么活到了 waker.wake() ? 我特地还把 TimerFuture::new( ) 中 创建新线程：

    let thread_shared_state = shared_state.clone();
    thread::spawn(move || {
         ...
        if let Some(waker) = shared_state.waker.take() {
            waker.wake()
        }
    });

这段内容全删了，发现程序第二次运行到 Executor:: run ( ) 的 while let Ok(task) = self.ready_queue.recv() { ... } 的时候就 block 住了，而且发现 task 也没 drop ，这是为什么啊？ 难道还有别的线程保存了 task ?

[ghost]

简单来说就是Task被轮询为Pending的时候会变成自引用结构，防止被Drop。同时Task持有Channel的发送端，这样Executor的接收端也不会被Drop，上面的第二个例子中因为Task无法被再次wake，所以task handle end之后就会一直block住。

[evanzp0]

谢谢回复，但是我把自引用和 future_slot.take() 中 take 出 future 的情况都算进去了，最后结果是 task.strong_count = 1... 是不是什么地方多算了...

[ghost]

thread::spawn(move || {

            thread::sleep(duration);
            let mut shared_state = thread_shared_state.lock().unwrap();

            shared_state.completed = true;
            // 这里take出waker，并在离开作用域后被drop，所以当task再次被send到executor的时候引用计数是1
            if let Some(waker) = shared_state.waker.take() {
                waker.wake()
            }
        });

[evanzp0]

谢谢，终于弄懂了，我的确漏了 waker.wake() 这块没算进去，waker会被 wake 消费掉，导致task的count被-1了

[h4cnull]

Executor轮询时如果Task中的future是Pending的，那么Executor会把被take出来的future放回Task。

[PSS-PDPZ]

impl Executor {
fn run(&self) {
while let Ok(task) = self.ready_queue.recv() {
// 获取一个future，若它还没有完成(仍然是Some，不是None)，则对它进行一次poll并尝试完成它
let mut future_slot = task.future.lock().unwrap();
if let Some(mut future) = future_slot.take() {
// 基于任务自身创建一个 LocalWaker
let waker = waker_ref(&task);
let context = &mut Context::from_waker(&*waker);
// BoxFuture<T>是Pin<Box<dyn Future<Output = T> + Send + 'static>>的类型别名
// 通过调用as_mut方法，可以将上面的类型转换成Pin<&mut dyn Future + Send + 'static>，此时Pin
里面的就是传过来
//的future即那个async块，执行这个poll方法，那就会执行块中的代码，在遇到.await后这个外层future（async块）会阻塞，
//因为执行到第二句，里面的.await会对子future即TimeFuture也进行poll，发现不能完成，则会阻塞
//当线程睡醒之后，改变共享状态，并调用wake方法，那就会调用wake_by_ref方法将任务再次放回通道，因为run方法一直在执行着
//当再次进入到while循环，执行到这里，即再次对async块进行poll，会从当时阻塞的地方开始，因为不会再次打印println!("howdy!")，
//那么.await再次对子future即TimeFuture也进行poll，发现其状态已改变，因此子future完成，开始往下执行，那么执行println!("done!")，
//到此外部future（async块）也完成，整个流程结束
if future.as_mut().poll(context).is_pending() {
// Future还没执行完，因此将它放回任务中，等待下次被poll，因为共享的同一任务，这边不放回 ，Task将不能将该任务自身放回到任务通道中，缺少future
//因为前面使用了take方法 Task里的Option已经变为None
*future_slot = Some(future);
}
}
}
}
}

[PSS-PDPZ]

个人理解，不对的请见谅并请指正。

[PSS-PDPZ]

context在第一次async块poll的时候，应该是已经可以被块内访问了，那么子future在poll的时候也就能使用了

[sinopec]

PIN的理解还是有点困难，pin的目的说是可以让地址不发生变化，但是什么情况下一个对象的地址会发生变化呢？就像是vector那样重新分配了地址空间的那种？ c/c++程序员一时没回过神来...

[xiaohongxiedaima]

use std::ptr::null_mut;

struct P {
    val: usize,
    ptr: *mut usize,
}

impl P {
    fn new(val: usize) -> P {
        P {
            val,
            ptr: null_mut(),
        }
    }

    fn init_ptr(&mut self) {
        self.ptr = &mut self.val as *mut usize;
    }
}

#[cfg(test)]
mod tests {
    use std::mem::swap;
    use crate::pin_demo::pin_demo3::P;

    #[test]
    fn test() {
        let mut p1 = P::new(1);
        let mut p2 = P::new(2);

        // 初始化p1,p2内部裸指针
        p1.init_ptr();
        p2.init_ptr();

        // 交换 p1，p2 对应的栈内存
        swap(&mut p1, &mut p2);

        // 获取 p2 裸指针对应的实际值
        let p2_val = unsafe { *(p2.ptr) };

        // p1的val 和 p2裸指针对应的是一个值
        assert_eq!(p1.val, p2_val);
    }
}

使用PIN能够固定自引用的数据类型的值，如p1,p2 让值保持一个不变的内存地址

[cxw620]

首先需要指出，Pin 本身不是一个 trait，而是一个 struct，只能包裹一个（智能）指针 P。所以我们严格说要写成 Pin<P> 的样子，Pin，例如 Pin 是没有意义的。最常见的就是 Pin<&T> 或者 Pin<&mut T>。

其次，Pin 一定程度上只是提供了个编译器层面的约束，阻止暴露 T （都 tm 直接暴露数据本体了随随便便被 move 掉）或者 &mut T（std::mem::swap 之类的内存操作也能把 T 拿走）进而防止指针 P 指向的 T 被移动。如果 unsafe 满天飞也是可能形同虚设的，不过非高级开发者一般只考虑 safe Rust。

再有，谈谈 Pin 的用武之地，其中最主要的就是 async Rust。
本文的内容解释了 Future 为什么不能被 move 的，用 Rust 的语言表达就是实现了 !Unpin 这个 trait。需要提一下的是，Rust 的世界绝大多数类型都是自动实现 Unpin 的（auto trait），毕竟大多数时候我们不关心是否需要移动 T。但是，在一些情况下，如 tokio::select! 去同时执行多个 Future，有可能移动之，所以传入的 Future （后面以F 代称）必须是 Unpin 的。咋办？拿 Pin 钉住咯。结合前面提到的只能 Pin 一个指针，我们 Pin 一下 &mut F 不就可以咯！所以就有了 tokio::pin! 这个宏，如：

// ...
let mut f = async fn {
    // ...
};
tokio::pin!(f);
// ...

本质就是变量遮蔽一下, 不需要你写个 unsafe 而已：

// ...
let mut f = async fn {
    // ...
};
let mut f = unsafe {
    std::pin::Pin::new_uncheck(&mut f)
}
// ...

个人结合各类教程和源代码阅读产出见解，或者说总结，请批评指正。强烈推荐各位仔细研读 https://folyd.com/blog/rust-pin-advanced，对 Pin 的介绍总结是相当详尽，而且难能可贵，是中文的！

异步放在哪个语言都是相当麻烦的，cpp 本来就地狱难度，遇到异步老鸟也得踮着脚过河，一不小心就出问题。go 广受好评就是相当强大的开箱即用的 goroutine 协程，等于是语言帮你处理了各种细节。Rust 在尽最大努力确保内存安全的前提下提供最大的自由，算是折中方案？

总之这块算是深水区，还是得多看源码，多写项目，出问题去分析解决问题，进步会更快。

[KujouRinka]

这章写得太好了。不仅学到了关于 rust 的异步的实现原理，还通过这个大致推测出 C++ asio 库的实现原理，以及为何其中任何一个异步函数都需要一个 io_context 作为参数传入。之前只会照猫画虎，现在算是彻底明白了，知其然，知其所以然。

[1793523411]

有跟我一样看懵的吗

[MrZLeo]

有一个问题，既然前面说 Future 是惰性的，只在 poll 后才会运行，那么这里的 TimerFuture 也不应该在 new的时候就生成 sleep的工作线程，而应该在 poll 的时候生成线程。🤔

[CandySunPlus]

同感，应该在 poll 第一次被执行时触发

[FuuuOverclocking]

纠错:

self.task_sender.send(task).expect("任务队列已满");

仅当 Receiver has disconnected and is no longer able to receive information 时, send() 才返回 Err; 任务队列已满时只会 block.

[xiandaoyan]

thread::spawn(move || {
// 睡眠指定时间实现计时功能
thread::sleep(duration);
let mut shared_state = thread_shared_state.lock().unwrap();
// 通知执行器定时器已经完成，可以继续poll对应的Future了
shared_state.completed = true;
if let Some(waker) = shared_state.waker.take() {
waker.wake()
}
});

关于这个TimerFuture请教一个问题，这里为什么在new方法中，要去启动一个线程？ 我理解所有的异步任务是实现一个线程内调度多个Future来轮流执行，请问这么理解是否正确？如果这里启动一个线程，相当于一个线程内只有一个Future了，这个是不是反而性能更差了，这样的做法是不是还不如直接多线程？ 当然可能这里只是一个例子而已，麻烦楼主回复前边的理解是否是准确的，谢谢！

[GSlun]

启动一个新线程是为了创建一个 Waker ，通过使新线程睡眠实现定时，然后调用wake 方法唤醒一个挂起的 Task

[h4cnull]

这里是在模拟”异步“，你可以想象这个线程就像是正在io，过一段时间才会完成，然后触发waker去告诉executor可以poll我了（实际上wake的时候触发了task将自己发送给executor）。

[TsGanT]

但是我有个问题，那就是这个生成的线程为啥不是在new的时候就直接开始执行，而是到了poll的时候才开始执行这个线程中的内容呢？

[478619644h]

new 的时候定时器就开始计时了，执行到 await 时去执行poll方法去判断一次到时间没，并注册唤醒函数

[Dawson-Jones]

spawner.spawn 这个方法, 获取 self 的所有权, main 函数中就不需要 drop 了

[bxb100]

看过 JYY 老师的视频, 再来看 rust 的线程调度, 别有一番感悟

[TsGanT]

怎么感觉看完了之后，Poll的实现还是有一些黑盒？看起来 TimeFuture中 new函数中的那个中的函数

thread::spawn(move || {
            println!("Start now; expect before executor");
            // 执行休息时间
            thread::sleep(duration);
            let mut s = mv_shared_state.lock().unwrap();
            s.completed = true;
            if let Some(wake) = s.waker.take() {
                println!("Wake here");
                wake.wake()
            }
        });

是在 if f.as_mut().poll(context).is_pending() 后被执行的，但看这个poll函数的实现

impl Future for TimeFuture {
    type Output = ();
    fn poll(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Self::Output> {
        let mut state = self.share_state.lock().unwrap();
        println!("Poll time {}", state.poll_times);
        state.poll_times = state.poll_times + 1; 
        if state.completed {
            Poll::Ready(())
        } else {
            state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

似乎也看不出为啥thread会在poll被调用的时候才执行？

[cj495840252]

放回了就能一直运行，不放回let的时候会failed，只能运行一次

use std::sync::Mutex;

fn main() {
    let mut a = Mutex::new(Some(0));
    let mut b = a.lock().unwrap();

    while let Some(i) = b.take() {
        {
            println!("loop: {i}");
            // *b = Some(i+1);
        }
    }
}

[TsGanT]

我想问一个问题 所以以上例子是基于线程的异步实现吗？那是不是其实就和 thread::spawn 后 用join handle的效果是一样的？

[mxq-151]

上面完整的代码可以在哪里找到

[aq15951]

看不懂，没有例子来说明很难看懂啊

[aq15951]

pub struct SocketRead<'a> {
socket: &'a Socket,
};Socket 引用的是哪个包呢，第一步就进行不下去了

[bobbyz007]

例子一些要点：

1. 例子中包含了两个Future，一个是async代码块修饰，另一个是定义的TimerFuture
2. executor以及Task中操纵的Future都是指async代码块修饰，TimerFuture的执行第一次是通过await触发，后续就是通过waker
3. 除了thread::spawn明确创建了一个新的线程，其他的poll以及wake等操作都是在同一个主线程中完成的。
4. 在新创建的线程中调用waker的wake方法时，会触发实现的ArcWake trait方法wake_by_ref，通过往channel中重复添加task来唤醒 executor 取出task再次执行（再次触发async以及TimerFuture的poll方法）

[meteorzh]

我有个细节的问题没理解到：
调用wake重新入队时，外层异步块的future第二次被poll，这个时候的future如何知道上次被poll时执行到的位置？
我的猜测是：
这个指令位置被存到了context中，第二次poll时，cpu从指定位置开始执行，这也是为什么future需要使用Pin的原因。

麻烦哪位老师解答一下rust实现这个逻辑的位置呢。

[ADkun]

最外面这个async块应该是生成了一个状态机Future

    spawner.spawn(async {
        println!("开始！");
        // 创建定时器Future，并等待它完成
        TimerFuture::new(Duration::new(2, 0)).await;
        println!("完成！");
    });

这个Future会match当前的状态来决定执行到哪里

[RossQAQ]

用过 C++20 Coroutine 自己封装个 runtime 出来就能看懂这里了，原理都一模一样

[bxb100]

那么问题来了，没用过 c艹 的人咋办🌚

[restart126]

关于执行器的设计部分，是不是实际执行器中，当一个Task返回是Pengding时候，仍然会将Task移出队列，否则在之后Waker时候，又一次将Task发给队列，就重复发送了，而且执行器的关闭时机也不是判断队列是否为空，在示例代码中，当计时器计时完毕时候又wake进行发送，那么这里是不是将一个Task发给队列发了两遍，求大佬解答

[hamster0523]

看代码像是把队列里的future取出来，poll一次之后看还是不是pending(),如果是就把原本在队列的future覆盖掉

[hamster0523]

应该就是执行器一直检查队头

[ycchanau]

感覺要自己call wait() 一點都不方便。。。

[limuyao1996]

有个疑惑，关于 Executor 的实现，因为是使用消息通道机制，而之前消息那章说消息的 recv() 方法是会阻塞线程的，那这里 while let Ok(task) = self.ready_queue.recv() 的写法，当一个 future 第一次 poll 没有执行完，或者同时执行多个 future 的时候，不会因为当前线程被阻塞住而无法执行吗？

[yvrenhao]

在上一章中，我们提到过 Future 需要被执行器poll(轮询)后才能运行，诺，这里 poll 就来了，通过调用该方法，可以推进 Future 的进一步执行，直到被切走...

这里的poll由谁调用？主语是谁?是imple Future被切走了吗?被谁切走？是异步运行时吗，就是那个executor？

原文很多个逗号，其分隔语句的主语均不明确，且一直在变。

[pinerge]

Rust高度抽象的语法确实适合写异步调度

[holihy]

我好笨啊，写了一遍源码，然后配合ai，不停地问，花了整整一天才搞懂整个流程。