性能 – 在将来-rs中封装阻塞I / O的最佳方法是什么？

我读了 tokio documentation,我想知道将来封装昂贵的同步I / O的最佳方法是什么.

使用reactor框架,我们可以获得绿色线程模型的优势：一些OS线程通过执行程序处理大量并发任务.

未来的tokio模型是需求驱动的,这意味着未来本身将轮询其内部状态以提供有关其完成的信息;允许背压和取消功能.据我了解,未来的投票阶段必须是非阻塞才能运作良好.

我想要封装的I / O可以看作是一个长期的原子和昂贵的操作.理想情况下,独立任务将执行I / O,并且相关联的未来将轮询I / O线程以获得完成状态.

我看到的两个唯一选择是：

>将阻塞I / O包含在将来的轮询功能中.
>生成OS线程以执行I / O并使用未来机制轮询其状态,如shown in the documentation

据我所知,这两种解决方案都不是最优的,并且没有充分利用绿色线程模型(首先不在文档中建议,其次不通过reactor框架提供的执行程序).还有其他解决方案吗？

Ideally, an independent task would perform the I/O and the associated future would poll the I/O thread for the completion status.

是的,这就是Tokio推荐的东西,以及为futures-cpupool和tokio-threadpool创建的包装箱.请注意,这不仅限于I / O,但对于任何长时间运行的同步任务都有效！

在这种情况下,您计划在池中运行闭包.池本身执行工作以检查阻塞闭包是否已完成并满足Future特征.

extern crate futures;
extern crate futures_cpupool;

use futures::{future, Future};
use futures_cpupool::CpuPool;
use std::thread;
use std::time::Duration;

fn main() {
    let pool = CpuPool::new(8);

    let a = pool.spawn_fn(|| {
        thread::sleep(Duration::from_secs(3));
        future::ok::<_, ()>(3)
    });
    let b = pool.spawn_fn(|| {
        thread::sleep(Duration::from_secs(1));
        future::ok::<_, ()>(1)
    });

    let c = a.join(b).map(|(a, b)| a + b);

    let result = c.wait();
    println!("{:?}", result);
}

请注意,这不是一种有效的睡眠方式,它只是一个阻塞操作的占位符.如果您确实需要睡觉,请使用类似futures-timer或tokio-timer的内容.

你可以看到总时间只有3秒：

$time ./target/debug/example
Ok(4)

real    0m3.021s
user    0m0.007s
sys     0m0.010s

同样,您可以使用tokio-threadpool获得相同的结果：

extern crate tokio; // 0.1.7
extern crate tokio_threadpool; // 0.1.2

use std::{thread, time::Duration};
use tokio::{prelude::*, runtime::Runtime};

fn delay_for(seconds: u64) -> impl Future<Item = u64, Error = tokio_threadpool::BlockingError> {
    future::poll_fn(move || {
        tokio_threadpool::blocking(|| {
            thread::sleep(Duration::from_secs(seconds));
            seconds
        })
    })
}

fn main() {
    let a = delay_for(3);
    let b = delay_for(1);
    let sum = a.join(b).map(|(a, b)| a + b);

    let mut runtime = Runtime::new().expect("Unable to start the runtime");
    let result = runtime.block_on(sum);
    println!("{:?}", result);
}

neither solution is optimal and don’t get the full advantage of the green-threading model

这是正确的 – 因为你没有异步的东西！您正在尝试将两种不同的方法结合起来,并且必须在某处进行翻译才能在它们之间进行转换.

second don’t pass through the executor provided by reactor framework

我不确定你的意思.上面的例子中只有一个执行者;等待隐含创建的那个.线程池有一些内部逻辑,用于检查线程是否完成,但只应在用户的执行程序轮询它时触发.