Rust Tokio 完整工程：有界任务、保序与失败收拢 · 全部同源工程文件至来源
前置：任务所有权。项目位于 examples/rust/p2-async-runtime，读取文件的每一行，真实计算 u64 checked sum；不需要账号、服务或端口。本次示例未执行，未运行 Cargo/Rust 编译、测试或格式化，也未安装依赖。未来读者可按 README 自跑核验；静态解析不能证明 Rust 类型正确或运行行为通过。
全部同源工程文件
以下包含完整工程与 README，均从仓库源文件 include，不维护第二份代码。Rust、TOML、输入和 Markdown 显式标记语言，阅读当前页即可取得未来自跑所需内容。
Cargo.toml：
toml
[package]
name = "p2-async-runtime"
version = "0.1.0"
edition = "2021"
rust-version = "1.71"
[dependencies]
tokio = { version = "=1.48.0", features = ["rt-multi-thread", "macros", "sync"] }src/lib.rs：
rust
use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};
use tokio::task::JoinSet;
#[derive(Debug)]
pub enum Job {
Sum(Vec<u64>),
Reject,
Panic,
}
#[derive(Debug, Default)]
pub struct Metrics {
active: AtomicUsize,
peak: AtomicUsize,
}
impl Metrics {
pub fn active(&self) -> usize { self.active.load(Ordering::SeqCst) }
pub fn peak(&self) -> usize { self.peak.load(Ordering::SeqCst) }
}
struct Active(Arc<Metrics>);
impl Active {
fn enter(metrics: Arc<Metrics>) -> Self {
let n = metrics.active.fetch_add(1, Ordering::SeqCst) + 1;
metrics.peak.fetch_max(n, Ordering::SeqCst);
Self(metrics)
}
}
impl Drop for Active {
fn drop(&mut self) { self.0.active.fetch_sub(1, Ordering::SeqCst); }
}
// Every line is one owned job. ERR deliberately becomes a task-level error.
pub fn parse(text: &str) -> Result<Vec<Job>, String> {
text.lines().enumerate().map(|(i, line)| {
if line == "ERR" { return Ok(Job::Reject); }
let numbers = line.split_whitespace().map(|word| {
word.parse::<u64>().map_err(|_| format!("line {}: invalid u64", i + 1))
}).collect::<Result<Vec<_>, _>>()?;
Ok(Job::Sum(numbers))
}).collect()
}
async fn calculate(job: Job, metrics: Arc<Metrics>) -> Result<u64, String> {
let _active = Active::enter(metrics);
match job {
Job::Reject => Err("rejected job".into()),
Job::Panic => panic!("intentional task panic"),
Job::Sum(numbers) => {
let mut sum = 0_u64;
for chunk in numbers.chunks(64) {
for value in chunk {
sum = sum.checked_add(*value).ok_or("sum overflow")?;
}
// Bounded CPU work followed by a cooperative scheduling point.
tokio::task::yield_now().await;
}
Ok(sum)
}
}
}
// The caller must await this function to completion to observe drain-on-error.
// Dropping this future drops JoinSet and requests abort, but cannot await cleanup.
pub async fn analyze(jobs: Vec<Job>, limit: usize, metrics: Arc<Metrics>)
-> Result<Vec<u64>, String>
{
if limit == 0 { return Err("limit must be positive".into()); }
let mut results = vec![0; jobs.len()];
let mut pending = jobs.into_iter().enumerate();
let mut running = JoinSet::new();
loop {
while running.len() < limit {
let Some((index, job)) = pending.next() else { break; };
let metrics = Arc::clone(&metrics);
running.spawn(async move { (index, calculate(job, metrics).await) });
}
let Some(completion) = running.join_next().await else { return Ok(results); };
let error = match completion {
Ok((index, Ok(value))) => { results[index] = value; continue; }
Ok((_, Err(error))) => error,
Err(error) => format!("task join failed: {error}"),
};
running.abort_all();
// Abort requests do not establish that destructors have completed.
while running.join_next().await.is_some() {}
return Err(error);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn abort_is_observed_after_join() {
let metrics = Arc::new(Metrics::default());
let copy = Arc::clone(&metrics);
let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
let mut tasks = JoinSet::new();
tasks.spawn(async move {
let _active = Active::enter(copy);
ready_tx.send(()).unwrap();
std::future::pending::<()>().await;
});
ready_rx.await.unwrap();
assert_eq!(metrics.active(), 1);
tasks.abort_all();
let error = tasks.join_next().await.unwrap().unwrap_err();
assert!(error.is_cancelled());
assert_eq!(metrics.active(), 0);
assert!(tasks.is_empty());
}
}src/main.rs：
rust
use std::{env, fs, process::ExitCode, sync::Arc};
use p2_async_runtime::{analyze, parse, Metrics};
fn run() -> Result<(), String> {
let args: Vec<String> = env::args().skip(1).collect();
if args.len() != 2 { return Err("usage: p2-async-runtime INPUT LIMIT".into()); }
let limit = args[1].parse::<usize>().map_err(|_| "invalid limit")?;
if limit == 0 { return Err("limit must be positive".into()); }
// Synchronous file read is intentionally before entering the runtime.
let text = fs::read_to_string(&args[0]).map_err(|e| format!("read input: {e}"))?;
let jobs = parse(&text)?;
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2).build().map_err(|e| format!("runtime: {e}"))?;
let values = runtime.block_on(analyze(jobs, limit, Arc::new(Metrics::default())))?;
for value in values { println!("{value}"); }
Ok(())
}
fn main() -> ExitCode {
match run() {
Ok(()) => ExitCode::SUCCESS,
Err(error) => { eprintln!("{error}"); ExitCode::FAILURE }
}
}tests/behavior.rs：
rust
use std::sync::Arc;
use p2_async_runtime::{analyze, parse, Job, Metrics};
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn bounded_and_ordered() {
let metrics = Arc::new(Metrics::default());
let jobs = vec![Job::Sum(vec![1; 4096]), Job::Sum(vec![7]), Job::Sum(vec![])];
assert_eq!(analyze(jobs, 2, Arc::clone(&metrics)).await.unwrap(), vec![4096, 7, 0]);
assert!((1..=2).contains(&metrics.peak()));
assert_eq!(metrics.active(), 0);
}
#[tokio::test]
async fn rejects_and_drains() {
let metrics = Arc::new(Metrics::default());
let jobs = vec![Job::Sum(vec![1; 4096]), Job::Reject];
assert_eq!(analyze(jobs, 2, Arc::clone(&metrics)).await.unwrap_err(), "rejected job");
assert_eq!(metrics.active(), 0);
}
#[tokio::test]
async fn panic_and_overflow_are_errors() {
for job in [Job::Panic, Job::Sum(vec![u64::MAX, 1])] {
let metrics = Arc::new(Metrics::default());
assert!(analyze(vec![job, Job::Sum(vec![1; 4096])], 2, Arc::clone(&metrics)).await.is_err());
assert_eq!(metrics.active(), 0);
}
}
#[tokio::test]
async fn empty_and_zero_limit() {
assert_eq!(analyze(vec![], 1, Arc::new(Metrics::default())).await.unwrap(), Vec::<u64>::new());
assert!(analyze(vec![], 0, Arc::new(Metrics::default())).await.is_err());
}
#[test]
fn parser_contract() {
assert_eq!(parse("1 2\r\n\r\n3").unwrap().len(), 3);
assert!(parse("-1").is_err());
assert!(parse("18446744073709551616").is_err());
assert!(matches!(parse("ERR").unwrap()[0], Job::Reject));
}inputs/normal.txt：
text
1 2 3
7
0 4inputs/reject.txt：
text
1 2
ERR
9inputs/overflow.txt：
text
18446744073709551615 1inputs/invalid.txt：
text
-1DEPENDENCIES.md：
md
# 依赖与版本
本工程示例未执行，未安装依赖，未生成 Cargo.lock。Tokio 固定 =1.48.0（公开发布版本，不宣称最新版）；edition 2021，声明的 rust-version 1.71 对齐该版 Tokio MSRV，不代表本工程已做最低版本编译。
启用 rt-multi-thread 提供多线程运行时并包含 rt；macros 提供测试属性；sync 提供取消测试中的 oneshot。没有网络、时间或 fs 驱动，不调用 enable_all。标准库提供文件读取、数值检查、原子计数。不依赖 tokio-util。
核对：<https://docs.rs/tokio/1.48.0/tokio/index.html>、<https://docs.rs/tokio/1.48.0/tokio/task/struct.JoinSet.html>。真正自跑前由读者准备可信 Rust 工具链和缓存依赖；首次正常 Cargo 解析会生成锁文件，应用应审阅并提交它。offline 不会替你下载缺失依赖。
固定tag的 [Tokio Cargo.toml](https://raw.githubusercontent.com/tokio-rs/tokio/tokio-1.48.0/tokio/Cargo.toml) 明确声明 rust-version = "1.71"；传递依赖将来解析的版本还可能提高最低工具链要求，未生成锁文件时不能承诺整个依赖图的最低版本可复现。
未来首次取得依赖：有可信网络和已安装 Rust/Cargo 的读者可在此目录自行运行 `cargo fetch`，审阅解析到的 Cargo.lock 与依赖来源后，再按 README 的 `cargo test --offline`/`cargo run --offline` 核验。若已有完整缓存，可直接离线；若缓存不全，offline 报缺依赖，需补齐可信缓存或使用联网流程。这里提供的是未来步骤，本轮没有执行 fetch、安装、解析依赖图、编译或测试。README.md：
md
# Tokio 有界求和实验
示例未执行。这里只做源码人工审阅、TOML/Markdown/JSON 纯解析，不声称编译或测试通过。
每个 UTF-8 输入行是一个任务，行内用空白分隔 u64；空行和为0，空文件输出为空。CRLF 支持。单行 ERR 是故意的任务失败；负数、非数字、超范围在提交任务前拒绝。求和溢出在任务中失败。每64项 yield 一次，实际计算输入而非 sleep 模拟成功。
## 未来读者自跑
准备依赖后在本目录执行（本次未执行）：
```sh
cargo test --offline
cargo run --offline -- inputs/normal.txt 2
cargo run --offline -- inputs/reject.txt 2
cargo run --offline -- inputs/overflow.txt 2
cargo run --offline -- inputs/invalid.txt 2
cargo run --offline -- inputs/normal.txt 0
cargo run --offline -- missing.txt 2
```
手工预期：正常 stdout 依输入顺序为6、7、4，各占一行，退出0。完成顺序不可预测，输出顺序固定。reject 包含 rejected job，overflow 包含 sum overflow，invalid 包含 invalid u64，零上限包含 limit must be positive，缺文件包含 read input；失败 stderr 且退出1，没有部分 stdout。测试预期全部通过但尚未执行；panic 测试可能产生 panic hook 文本，这是被 JoinError 收拢的预期行为。
先读文件与解析，再进入运行时；填充不超过 limit 的 JoinSet；回收一个才补一个；失败停止补充、abort_all，再逐个 join 到空。已经完成的成功结果在整体失败时丢弃。Active 的 Drop 计数用于检查返回后无在途计算；oneshot 测试确认任务已进入后才取消，不依赖 sleep。测试不承诺两个任务必定同时活跃。
## 取消、内存与清理
仅正常 await analyze 到返回的调用满足等待清理协议；外部直接 drop analyze 只会触发 JoinSet Drop 请求取消，不能同步等析构完成。abort 在让出执行权后生效，不抢占当前同步计算；每个工作片段是有限的64次数值操作。已运行的 spawn_blocking 无法靠 abort 停止，本工程没有使用它。
无服务、端口、网络和输出文件。自跑生成 target 和 Cargo.lock；无需业务回滚，重跑时选择正常输入。不要删除输入源文件；可在确认目录后自行清理 target。输入与结果全部驻内存，任务数有界不等于内存恒定，也不是通用CPU线程池吞吐基准。取消不是撤销已经产生的外部副作用。
文件：Cargo.toml、DEPENDENCIES.md、src/lib.rs、src/main.rs、tests/behavior.rs、inputs 下四个输入和本 README。练习
① 改成最大值计算，空行怎样定义，哪些契约与测试要改？② 把结果直接在任务中 println，会破坏哪两条保证？③ 为使内存随 limit 而非文件大小变化，应同时改哪些层？
参考解法
① 空行可返回 Option 或明确错误，需同步改结果类型、CLI、空行测试，调度器只作必要类型修改。② 完成先后不等于输入顺序，而且后续失败前已输出部分结果。③ 输入必须增量读取，提交容量与保序缓冲都要限制；只加入 Semaphore 而预读整文件仍是 O(n)。
来源
2026-10-17核对 Tokio 1.48.0 特性、yield_now；其余语义链接见前两篇。源码静态审阅与输入手工推导不等同于实际 Cargo 执行。
继续阅读
- 排错
- 排错
- 排错