advance/async/multi-futures-simultaneous

[giscus[bot]]

advance/async/multi-futures-simultaneous

https://course.rs/advance/async/multi-futures-simultaneous.html

[mumumusuc]

跟 Unpin 和 FusedFuture 进行交互这部分好像有点问题，准确的说来自变量名/路径的future要求其实现Unpin+FusedFuture，对于来自表达式的future可以放宽Unpin的限制。

// Ok! Unpin+FusedFuture
let t1 = task_one().fuse();
let t2 = task_two().fuse();
pin_mut!(t1, t2);
select! {
    () = t1 => println!("任务1率先完成"),
    () = t2 => println!("任务2率先完成"),
}

// Ok! only FusedFuture
select! {
    () = task_one().fuse() => println!("任务1率先完成"),
    () = task_two().fuse() => println!("任务2率先完成"),
}

[horsefaced]

应该是自动 from 进行变换了

[runzhi214]

说的没错，因为这是原文档Quote、Unquote

If an expression which yields a Future is passed to select! (e.g. an async fn call) instead of a Future by name the Unpin requirement is relaxed, since the macro will pin the resulting Future on the stack.

[siliangZ]

future 的set方法是哪个trait为其添加的啊？它的行为是会中断旧的future，来运行新的future吗？

[mumumusuc]

https://doc.rust-lang.org/src/core/pin.rs.html#659-664

[horsefaced]

这里其实可以理解为把函数指针又换了一个新的, 然后就可以继续执行新的了.

[Nanbert]

我有个疑惑，关于default和complete分支的，如果所有future都执行完了，不就是所有future都不是ready状态，此时不就是同时满足default和complete吗？

[xiaoyueinfo]

select!里面loop轮询完所有分支和join!是不是完全一样的效果？

[ruaruagerry]

future如果都跑完了，default和complete确实是一样的效果。例子里也说明了，把default写到上面，让它露露脸。

[ruaruagerry]

join!是全部完成了，才能集中处理。而select!你可以理解为把join!的过程交由你来管控，你可以在join!的途中去做事情。

[horsefaced]

这里的例子最好把

mut interval_timer: impl Stream<Item = ()> + FusedStream + Unpin

解释一下, 要不然很难理解为什么在 interval_timer 连续不断产生值的情况下, select! 还能运行下一个, 而不是一直在运行第一条.

[horsefaced]

我写了下面这个例子, 包括了流与交换 Fuse 部分

#[test]
    fn test_future_in_select() {
        let fut = Fuse::terminated();
        let mut async_tasks: FuturesUnordered<Pin<Box<dyn Future<Output = i32>>>> =
            FuturesUnordered::new();
        async_tasks.push(Box::pin(async { 1 }));
        pin_mut!(fut);
        let mut total = 0;
        loop {
            select! {
                num = async_tasks.select_next_some() => {
                    println!("first num is {num} and total is {total}");
                    total += num;
                    println!("total is {total}");
                    if total >= 10 { break; }
                    if fut.is_terminated() {
                        fut.set(async { 1 }.fuse());
                    }
                },
                num = fut => {
                    println!("second num is {num} and total is {total}");
                    total += num;
                    println!("now total is {total}");
                    async_tasks.push(Box::pin(async { 1 }));
                },
                complete => break,
                default => panic!(),
            };
        }

        println!("total finally is {total}");
    }

[Linuxdazhao]

tokio select! 好像不支持default and completed 呢

[Linuxdazhao]

FusedFuture的原因跟上面类似，当 Future 一旦完成后，那 select 就不能再对其进行轮询使用。Fuse意味着熔断，相当于 Future 一旦完成，再次调用poll会直接返回Poll::Pending 这里应该是直接放回Poll::Ready(None) 吧。不然pending的话，下次还是会被调用

[duanyuluo]

我认为作者是对的：https://docs.rs/futures/latest/futures/future/trait.FusedFuture.html#

A future which tracks whether or not the underlying future should no longer be polled.

is_terminated will return true if a future should no longer be polled. Usually, this state occurs after poll (or try_poll) returned Poll::Ready. However, is_terminated may also return true if a future has become inactive and can no longer make progress and should be ignored or dropped rather than being polled again.

rust建议在使用FusedFuture时，应该先用is_terminated判断是否完成，避免再次poll。也就是说，在FusedFuture根本没有定义完成后，再次调用poll的行为是什么。
我们在参考一个具体实现FusedFuture的Join，https://docs.rs/futures-util/0.3.28/src/futures_util/future/join.rs.html#76-88

        impl<$($Fut: FusedFuture),*> FusedFuture for $Join<$($Fut),*> {
            fn is_terminated(&self) -> bool {
                $(
                    self.$Fut.is_terminated()
                ) && *
            }
        }

我们发现，Join实现FusedFuture时，也没有考虑继续调用poll的后果。
因为async其实是无数据的状态机代码，所以，再次调用时，应该和首次调用是一样的，也就是返回Pending。这其实也是trait FusedFuture的目的，就是帮助不具备记录状态能力的Future实现记录Future的完成状态。

[IronPanda0]

即便如此，也不关Poll::Pending或Poll:Ready什么事啊，难道不是通过is_terminated()来知晓是否已经完成吗？https://rust-lang.github.io/async-book/06_multiple_futures/03_select.html

[WenqingZong]

已阵亡在rust的汪洋大海中

[sinopec]

这部分也太枯燥了。。。

[runzhi214]

join_all(book_fut, music_fut)会报错、并提示你不是一个类型的async。

futures包给出的样例代码是这样的：

use futures::future::join_all;

async fn foo(i: u32) -> u32 { i }

let futures = vec![foo(1), foo(2), foo(3)];

assert_eq!(join_all(futures).await, [1, 2, 3]);

[runzhi214]

如果fused看不懂，建议看futures包的原始文档

[runzhi214]

Futures包的这个select!宏绝对是参考了GO的select语句设计的

[tjw123hh]

err_info -> err_into

[HC-miss]

我大抵是学不会了，横竖睡不着，坐起来点了一支烟，没由来的悲伤，黯然看着select!的两个不理解的问题，一个是我的，另一个大抵也是我的

[chawson]

如何修改代码才能走让select!执行default分支

[YdayangY]

文中说了把default放到completed前面

[programmer-sugar9527]

这怎么写的出相应的代码来😐

[Alkaidd]

放在js里大概就这个感觉吧：
join! -> promise.all()
try_join! -> promise.all().catch()
select! -> promise.all((async func).then(res => handle(res)))

[Nie-Tianyi]

FusedFuture 的原因跟上面类似，当 Future 一旦完成后，那 select 就不能再对其进行轮询使用。Fuse 意味着熔断，相当于 Future 一旦完成，再次调用 poll 会直接返回 Poll::Pending。

这里写错了把，以下是chatGPT的回答：

FusedFuture 是 Rust 标准库中 Future 的一个扩展特性，它允许你检查一个 Future 是否已经完成。与普通的 Future 不同，FusedFuture 在完成后可以通过 is_terminated 方法来判断它是否已经终止（即已经完成且不会再返回 Poll::Pending）。

具体来说，FusedFuture 具有以下特点：

1. is_terminated 方法：FusedFuture 定义了一个额外的 is_terminated 方法。这个方法返回一个布尔值，用于指示 Future 是否已经完成并进入了终止状态。一旦 is_terminated 返回 true，意味着这个 Future 永远不会再产生 Poll::Pending，它要么返回 Poll::Ready，要么一直保持完成状态。

2. poll 行为：对于一个实现了 FusedFuture 的 Future，在它完成并进入终止状态后，再次调用 poll 将持续返回 Poll::Ready。与普通的 Future 类似，FusedFuture 在完成后同样不会再回到 Poll::Pending。

3. 使用场景：FusedFuture 通常用于一些需要反复检查 Future 状态的场景。通过 is_terminated 方法，程序可以避免不必要的 poll 调用，从而提高效率。

FusedFuture 通常与像 select! 宏这样的高级异步编程工具一起使用，在并发场景中尤为有用。通过这种扩展特性，你可以更精确地控制和监控 Future 的状态。

use futures::future::{FutureExt, FusedFuture};
use futures::pin_mut;
use futures::task::Poll;

async fn example() -> i32 {
    42
}

fn main() {
    let future = example().fuse();
    pin_mut!(future);

    assert!(!future.is_terminated());

    let poll_result = future.poll_unpin(&mut futures::task::noop_waker_ref());
    assert!(matches!(poll_result, Poll::Ready(42)));
    assert!(future.is_terminated());
}

在这个例子中，future.fuse() 将普通的 Future 转换为 FusedFuture，并在完成后检查它是否终止。

[tc19901016]

这句话不是很理解：【default 分支，若没有任何 Future 或 Stream 处于 Ready 状态， 则该分支会被立即执行】
我尝试将a_fut和b_fut使用pending方法挂起，但执行结果：1）并没有执行default分支；2）loop没有无限循环，main函数正常执行完成。代码如下：

use futures::future::Pending;
use futures::future;
use futures::select;

pub fn main(){
    let mut a_fut:Pending<u8> = future::pending();
    let mut b_fut:Pending<u8> = future::pending();

    loop{
        select!{
            a = a_fut =>{},
            b = b_fut =>{},
            complete => break;
            default => panic!("错误"),
        };
    }
    println!("结束");
}

有没有可能future::pending()函数在骗我？（文档写的是...never finishes）
futures版本：0.3.30

[kouhe3]

讲太快了，不懂

[iamxvbaba]

应该是参考了Golang的 for select.