async_nursery

结构化并发的原始类型。

孵化器允许编写遵循结构化并发的并发程序。如果您对这个概念不太熟悉，结构化并发论坛有一些非常优秀的资源，您可以访问 结构化并发论坛。这个库的名字灵感来源于优秀的 Python Trio 库。

目录

• 描述
• 安装
  • 升级
  • 依赖项
  • 安全性
  • 性能
• 用法
  • 基本示例
  • 返回错误
  • 恢复其他返回类型
  • 恐慌
  • 与 FuturesUnordered 的区别
  • API
• 贡献
  • 行为准则
• 许可证

描述

async_nursery 为 Rust 带来了结构化并发的原始类型。结构化并发有三个主要目标

1. 并发程序的合理控制流。

结构化并发的笔记，或：Go 语句被认为是有害的，由 Nathaniel J. Smith 解释得非常精彩。概括来说，如果一个函数想要分叉并发做一些工作，确保在函数返回时所有子任务都已完成。这样它就像我们熟悉的同步代码中的黑盒一样工作。函数有输入和返回值，当它完成时，它创建的代码不再运行。

您可以通过将来自 async_executors 的 JoinHandle 放入 FuturesUnordered 来做到这一点，但正如我们下面将看到的，async_nursery 要灵活和方便得多。与来自 tokio 或 async-std 的直接 JoinHandle 不同，来自 async_executors 的 JoinHandle 默认不会在丢弃时分离。

2. 防止资源泄漏。

孤儿任务，没有 JoinHandle 的任务可能会永远存活，要么在循环中运行，要么死锁。结构化并发确保没有泄漏，将所有资源整齐地放入调用树中，这与调用栈非常相似。在调用树中，一个栈帧可以是一系列并排坐在一起的几个栈帧，它们正在并发执行，但当我们返回到上一个栈帧时，它们都已完成。

3. 传播错误

在Rust中，向上传递错误到调用栈是很常见的。如果你启动一个任务并让它运行在虚空中，你需要通过通道等通道进行错误处理。在结构化并发中，由于所有任务都在父任务返回之前被连接，你可以像同步代码一样返回错误。如果某个任务遇到错误，也可以取消所有兄弟任务。

async_nursery 的特性

• Nursery 作为启动器。
• NurseryStream 实现了所有它照料的 futures 的结果的 Stream<Out>。
• NurseryStream 如果只想等待一切完成，但不在乎返回值，则实现了 Future<Output=()>。
• NurseryStream 主要为您管理 JoinHandle。
• 可以由任何实现了 SpawnHandle 或 LocalSpawnHandle 的 executor 支持。
• 在丢弃 NurseryStream 时取消所有正在运行的未来。
• Nursery 实现 FutureObj 和/或 LocalFutureObj 以及 Nurse 和 NurseExt 的 Sink。
• Nursery 将包装的 executor 的 async_executor 特性转发。这对于 Timer、TokioIo、YieldNow 和 SpawnBlocking 都有效。请注意，当使用这种方式像 SpawnBlocking 时，nursery 不管理任务，它只是让您使用包装的 executor。

缺失的功能

• 没有提供用于 协作取消 的 API。由于 std::task::Context 中没有对该功能的支持，您必须基本上传递一些取消令牌到任务中，该任务需要执行清理且不支持在任何 await 点被丢弃。由于这需要启动的任务具有特定的支持，我将此留给用户。在 examples 目录 中包括了一个使用 AtomicBool 的示例。优势在于灵活性。您可以取消 nurseries 中的某些任务，而让其他任务继续运行，或者如果它们支持的话，让其他任务在丢弃时取消等。异步析构（Async drop）可能会有一天缓解这种痛苦，但还不是时候。

• 没有提供用于运行非 'static' futures 的 API。在安全的 Rust 中这是不可能的，因为可以使用 std::mem::forget 来泄露 nurseries 并欺骗它超越其父栈帧的生命周期，此时它将保留一个无效的引用。如果您真的想这么做，我建议您查看 async-scoped crate，它通过要求您使用 unsafe 来允许这样做。

dependencies:

   async_nursery: ^0.5

[dependencies]

   async_nursery = "0.5"

use
{
   async_nursery   :: { Nursery, NurseExt } ,
   async_executors :: { AsyncStd          } ,
};

pub type DynResult<T> = Result<T, Box< dyn std::error::Error + Send + Sync + 'static >>;

async fn self_contained() -> DynResult<()>
{
   let (nursery, output) = Nursery::new( AsyncStd );

   for _ in 0..5
   {
      nursery.nurse( async { /* do something useful */ } )?;
   }

   // This is necessary. Since we could keep spawning tasks even after starting to poll
   // the output, it can't know that we are done, unless we drop all senders or call
   // `close_nursery`. If we don't, the await below deadlocks.
   //
   drop(nursery);

   // Resolves when all spawned tasks are done.
   //
   output.await;

   Ok(())
}