C++ 编程面试：如何用 C++26 std::execution 组合可取消的异步流水线？

题干与适用场景

服务要把读取、变换和汇总组合成异步流水线。调用方可能在超时、客户端断开或资源不足时取消；每个阶段可能失败，部分工作已经在执行。请使用 C++26 std::execution 的 sender/receiver 模型，说明调度、完成信号、生命周期和降级路径。

这道题考察并发抽象的边界，不要求候选人背诵某个库的语法。标准执行控制库把 sender 描述的工作图与 receiver 的完成处理分开，并用 operation state 持有连接后的异步状态。答案应把可取消性、错误语义和资源回收写成契约。

面试官考察点

• 能否区分惰性 sender、连接后的 operation state 和真正开始执行的 start。
• 能否使用 scheduler 表达执行资源，而不是在业务函数中随意创建线程。
• 能否分别处理 setvalue、seterror 和 set_stopped，避免把取消伪装成异常。
• 能否让取消信号穿过每个阶段，并保证停止后不再提交新的副作用。
• 能否说明并行阶段的背压、限额、异常安全和结果聚合。
• 能否提供标准库不可用时的兼容层、功能探测和一致性测试。

回答前需要澄清的问题

1. 读取是本地文件、网络请求还是数据库游标？每个阶段是否可重复或有外部副作用？
2. 取消是尽快停止、在当前不可中断调用后停止，还是必须回滚已经提交的结果？
3. 并行度、内存上限、单项超时和整体截止时间分别是多少？
4. 汇总是否要求输入顺序、稳定浮点结果或部分结果可见？
5. 目标编译器和标准库是否实现 C++26 execution，还是只能使用实验性实现？

30 秒回答框架

我会先定义流水线的完成契约：值、错误、停止三条通道，取消只阻止尚未开始的工作并让可中断阶段尽快响应。每个 sender 保持惰性，连接后得到 operation state，只有调用 start 才开始。使用明确的 scheduler 管理执行资源，以并行度和队列上限保护内存。聚合器规定顺序和部分结果规则；工具链通过能力探测选择标准实现、兼容库或同步标量路径，三条路径共享同一组取消、错误和结果测试。

分步骤深入解答

1. 先画出惰性工作图

把读取 sender 接到变换 sender，再接汇总 sender。then 适合把已产生的值传给下一步，letvalue 适合根据结果创建下一段异步工作，whenall 适合并行分支。组合只构造工作图，不应在构造阶段执行 I/O。

2. 明确连接与生命周期

sender 与 receiver 通过 connect 生成 operation state；调用 start 后才允许执行。operation state 的地址必须在异步操作结束前保持有效，因此不能把它放在即将返回的栈帧中。拥有者应把状态放在请求上下文或异步作用域里，并在完成、错误和停止三条路径都释放资源。

3. 把资源交给 scheduler

scheduler 是执行资源的轻量句柄。读取可放在 I/O 资源，CPU 变换放在受限的并行资源，on、startson 或 continueson 表达阶段边界。不要在每个元素里创建线程；用固定并行度、队列长度和批大小限制内存与上下文切换。

4. 传播停止、错误和值

值完成进入下一阶段，错误进入统一错误处理，停止进入取消处理。receiver 环境中的 stop token 是取消观察点；阻塞系统调用必须有可中断等待或有限超时，否则只能在调用返回后响应停止。取消不等于回滚，若阶段已写入外部系统，应使用幂等键、补偿动作或明确的不可撤销边界。

5. 一个最小的组合示意

下面代码展示工作图形状；实际读取和线程池 sender 需要由项目提供。

using namespace std::execution;

auto pipeline = read_sender()
  | let_value([](Batch batch) {
      return bulk_transform(batch, get_parallel_scheduler());
    })
  | then([](Transformed value) { return summarize(value); })
  | upon_error([](std::exception_ptr error) { record_failure(error); })
  | upon_stopped([] { record_cancellation(); });

auto state = connect(std::move(pipeline), receiver);
start(state);

receiver 必须由仍然存活的请求上下文拥有，并在自己的环境中提供停止令牌。生产代码还应记录阶段、批次、截止时间和取消原因，避免只看到一个泛化错误。

6. 并行汇总和副作用边界

并行变换应保持每个任务的局部状态，汇总阶段再按定义的顺序合并。若结果允许无序合并，要说明非结合浮点运算带来的差异；若要求稳定结果，保留索引或分区序号。写外部系统前先检查停止令牌，提交后记录幂等键，不能把 set_stopped 当作已经撤销提交。

7. 降级、测试与观测

用 feature-test 宏、编译器版本和标准库能力构建矩阵。标准 execution 不可用时，兼容实现可以提供相同的内部 sender 契约；再不行就使用受限线程池或同步路径，但保持值、错误、停止语义一致。测试空输入、部分批次、重复取消、错误与取消竞速、资源耗尽、operation state 提前销毁和多次启动。基准记录吞吐、尾延迟、队列长度、取消响应时间和未完成任务数。

高质量示范回答

我会把流水线设计为惰性 sender 图：读取、并行变换、汇总分别定义完成签名，再通过 connect 生成 operation state，只有 start 才执行。I/O 与 CPU 使用不同 scheduler，并用并行度、队列和批大小限制资源。receiver 处理值、错误和停止三条通道；停止令牌在每个可中断点检查，外部写入用幂等键和补偿边界，取消不承诺回滚。

operation state 由请求作用域持有到完成，错误和停止都走统一清理。工具链先探测 C++26 execution，选择标准实现、兼容实现或同步降级；三条路径共享行为测试，覆盖空批次、错误竞速、取消响应、资源耗尽和提前销毁。上线后观察尾延迟、队列长度、取消响应和泄漏，确认并行化确实改善目标指标。

常见错误

• 把构造 sender 当成已经启动异步任务。
• 让 operation state 在函数返回后失效。
• 只使用异常通道，把取消当成普通错误。
• 每个元素创建线程，忽略队列、并行度和内存上限。
• 看到停止信号就声称外部副作用已经回滚。
• 并行汇总没有定义顺序、浮点误差或部分结果规则。
• 只写一条标准库路径，未准备能力探测和兼容降级。

追问及应对

sender 什么时候真正执行？

组合 sender 只描述工作图；连接生成 operation state，调用 start 后才启动异步操作。测试应分别覆盖构造、连接和启动阶段。

停止信号能强制终止系统调用吗？

不能保证。系统调用必须提供可中断接口、超时或分块检查；否则只能在当前调用返回后响应停止，并记录最坏响应时间。

错误和停止同时发生怎么办？

定义优先级和一次性完成规则，保证 receiver 只收到一个最终完成信号。记录原始错误与停止原因，避免把竞速隐藏成成功。

when_all 中一个分支失败会怎样？

明确其他分支是继续、请求停止还是等待清理；共享资源要有作用域和取消传播，汇总器不能读取已失效的分支结果。

如何保证汇总结果可重复？

保留分区序号并按固定顺序合并，或明确允许无序结果及误差范围。浮点运算不满足结合律时不能只依赖并行归约。

标准库还不支持 C++26 怎么办？

通过编译能力矩阵选择兼容实现或同步路径，保持内部完成语义与测试不变。不要让公共接口暴露某个实验库的私有类型。