从错误处理到结构化并发
原文由 Nelson Elhage 于 发布，订阅该博客
我们应该如何思考并发程序中的错误处理？
在单线程程序中，我们很大程度上已经收敛到了一种标准模式，尽管具体实现和细节形态五花八门。当发生错误时，它会沿着调用栈向上传播，直到找到一个准备好处理它的栈帧。在此过程中，我们会按顺序展开栈帧，让每一帧都有机会酌情清理或销毁资源。
这一模式清晰地描述了许多现代语言中显式的异常处理机制（C++、Python、Java），它们也都有在栈展开时进行清理的机制（RAII、finally 块、Python 上下文管理器）。但它同样描述了 Rust 中的标准模式（返回 Result、? 运算符以及在展开时调用 drop），也描述了 Go（经典的 if err != nil { return err } 模式和用于清理的 defer），甚至大多数现代 C 代码，例如通过 goto error 模式（参见Linux 内核中的例子）。
以今天的眼光看，这种描述或许会显得过于宽泛而空洞，但情况并非一直如此。其他一些（大多已被废弃的）错误处理方式包括 Lisp 的“restarts”机制、C 语言中的 longjmp1、类似 UNIX 信号的“trap”机制，以及 Visual Basic 中臭名昭著的on error 子句。“展开”本身也依赖于拥有结构化的调用栈，而这一概念本身也曾需要被发明和推广。
今天我想探讨的是：对于并发程序——那里并不存在单一的调用栈——我们应该如何更新这一模式？在存在多个并发任务的情况下2，我们又该如何组织代码来处理错误状况？
未处理的错误
或许错误处理最简单的情况，就是出现了错误，却没有任何代码显式地处理它。在单线程程序中，我们预期错误会“冒泡”到入口点，并终止程序，最好还能附带一条有用的错误信息和/或栈回溯。
在包含多个任务的并发程序中，会发生什么就没那么明显了。我们可以向上冒泡并终止抛出错误的那个任务，但接下来呢？为了更具体地说明，我们来看一个玩具程序3：
import threading
import time
def background_thread():
# This was supposed to be running some background work, but it
# encountered an error!
raise ValueError("oops")
def main():
threading.Thread(target=background_thread).start()
# do the main work
time.sleep(5)
print("All done, exiting!")
main()这个程序试图运行两个并发线程，一个执行某种“主”工作函数，另一个在“后台”做一些工作。然而，后台线程会立即抛出一个异常。此时应该发生什么？
在各种语言实现中运行这个程序的不同版本，我们会发现大致存在两种常见的做法，我认为这也基本就是两种显而易见的选择：
- 打印错误，终止该线程，然后继续运行，直到所有线程（或某种变体：也许只是主线程，也许是所有“非守护”线程）都已退出。（Java、Python）
- 打印错误，并立即终止整个程序（Go、Rust、C++）
在 Python 中，程序会立即记录异常，但随后继续运行：它会等待 time.sleep 结束，然后退出，甚至是以成功的退出码退出！
$ time python exception_thread.py
Exception in thread Thread-1 (background_thread):
Traceback (most recent call last):
File "/Users/nelhage/.pyenv/versions/3.11.0/lib/python3.11/threading.py", line 1038, in _bootstrap_inner
self.run()
File "/Users/nelhage/.pyenv/versions/3.11.0/lib/python3.11/threading.py", line 975, in run
self._target(*self._args, **self._kwargs)
File "/Users/nelhage/Sync/code/structured-concurrency/exception_thread.py", line 7, in background_thread
raise ValueError("oops")
ValueError: oops
All done, exiting!
python exception_thread.py 0.03s user 0.02s system 0% cpu 5.124 total
$ echo $?
0两种选择都不太令人满意。
直接杀掉整个程序是一记重锤。我曾亲自参与处理过高严重级别的故障，其直接原因就是后台监控 goroutine 中未处理的 panic，导致关键守护进程崩溃。在那种特定情况下，我们宁愿让“真正”的工作不受干扰地继续下去。
反过来说，让程序继续运行则意味着我们处于一种几乎肯定从未被测试或预料过的状态，如果其他任务期望已死的任务继续推进或执行某些操作，我们就会面临很高的死锁甚至更糟的风险。
让程序继续运行也会在开发阶段带来痛苦。在开发过程中，许多错误都是“低级”的开发者失误——拼写错误、简单的逻辑错误、其他细小的、局部的、基本无关紧要的疏漏。当我遇到这类错误时，通常希望能尽快修复并重启程序。如果程序仍在运行，我就得手动重启，而且如果其他任务正在产生输出，异常信息很可能会被淹没在滚动的日志中。
错误该投递到哪里？
在某种意义上，我们希望能有一个更好的地方来“转发”错误。在单线程程序中，那个地方就是“调用者”。而在并发场景下，任务并没有一个最终会返回的调用者，那么我们该怎么做呢？
我们可以从 Python 的asyncio 框架中获得一些启发，它采取了一种不同于上述两种的“第三条道路”。在 asyncio 中，任务由一个Task 对象来表示，它是一个可以被等待的对象，就像事件、锁或套接字等一样。等待一个 Task 会阻塞直到它完成；如果该任务抛出了异常，这个异常会被重新投递给所有等待该 Task 的对象。
这种做法不预设立场；它避免对谁应该处理从任务中逃逸的异常做出规定，而是把做决定的工具交给程序员自己。
然而，它也有一个严重的缺点。如果没有人等待某个 Task，而该任务又抛出了异常，那么这个异常实际上会被吞掉，直到程序退出时才会带着一条警告被打印出来。如果有人正等待该任务比如通过某个队列产生输出，那么程序就会永远挂起，悄无声息、难以捉摸而又神秘莫测。
事实上，情况糟糕到我有时会把它总结为“默认情况下，asyncio 会完全吞掉主任务之外的异常”。严格来说这并不准确，但在我的经验中，这是一个不错的初始心智模型，也恰当地描述了许多开发者使用 asyncio 时的体验。
如果我们总是等待任务会怎样？
asyncio 的做法有很多可取之处，只要你记得等待每一个 Task。我们如何才能将这一不变量作为规则强制执行呢？最简单的规则可能是：如果你创建了一个任务，你就有责任等待它。我们可以通过把 asyncio.create_task(coro) 做成一个异步上下文管理器来编码这一模式，它会在退出前等待任务完成。
# (n.b. this is not real API in any version of Python)
async with asyncio.create_task(background_task()) as task:
# …
# `task` will be waited for on exit from this block, and any exception raised当然，我们常常想创建许多任务——或者数量动态决定的任务——因此我们可以借鉴contextlib.ExitStack 的惯用法，让一个上下文管理器对象来允许创建任意数量的任务；类似于：
async with TaskLauncher() as tasks:
tasks.create_task(background_task())
# do the main work via a second task
tasks.create_task(asyncio.sleep(5))
# All tasks will be waited for on exit from the region只要我们只通过 TaskLauncher 来创建新任务，我们就有了一个清晰的父子关系，每个任务都“归属于”一个明确的父任务，而父任务负责等待从子任务中逃逸的异常。如果任何任务抛出未处理的异常，它就会沿着这个层级结构向上冒泡；如果没有人捕获它，它最终会在根任务和 asyncio.run 调用处触底。我们已经在很大程度上将并发异常处理问题转化为了熟悉的单线程版本！
两个问题
遗憾的是，我们远未完成。上面的草图存在两个严重的、相关联的挑战，没有一个有简单的解法。
死锁
首先是死锁。我们上面说过，任务的父任务会“最终”等待它。只有当父任务确实退出了上下文管理器时，这一点才成立。但如果父任务正在等待永远不会完成的工作，而本应处理该工作的某个子任务却遇到了错误，那会怎样呢？
下面是一个简短的例子，展示了这类问题的一种形态：
from task_launcher import TaskLauncher
import asyncio
async def do_work(job_id, done_event):
if job_id == 1:
raise ValueError("Oops, job 1 failed!")
done_event.set()
async def main():
async with TaskLauncher() as tasks:
events = []
for i in range(4):
done = asyncio.Event()
tasks.create_task(do_work(i, done))
events.append(done)
for ev in events:
await ev.wait()
print("All done!")
if __name__ == '__main__':
asyncio.run(main())这个例子是常见模式的一个程式化版本：许多“扇出”或“扇出/扇入”并发模式的基本形态就是“启动一些任务；这些任务执行某些工作；父任务等待工作完成”。
对于这个具体的 bug，有很多简单的解法4。然而，我们希望找到一种能“自动”或以某种通用方式修复它的模式，或者至少避免让这个浅显的陷阱就摆在那里等着坑我们。依我的经验，遇到这类死锁是极其容易的，尤其是在初次开发并发系统时。
资源泄漏
在单线程程序中，错误恢复的挑战不仅仅是展开控制流，还要确保“撤销”或“清理”任何正在进行且与失败操作相关的工作。我们可能需要释放内存、关闭文件句柄，或将某个数据结构恢复到一致状态。
在并发程序中，与失败操作相关联的资源可能包括多个不同的执行任务！如果与我们的 TaskLauncher 相关联的某个任务失败了，我们就需要确保所有已创建的任务都能以某种方式被停止，或至少有机会清理被遗弃的状态。
在某种意义上，最简单的修复办法是让 TaskLauncher 在退出前等待所有已创建的任务。然而在实践中，单单做出这一改动会让我们的死锁问题急剧恶化。
我讨厌取消，但……
如果我们想同时：
- 在返回前等待所有子任务，以确保我们能得知任何额外的错误，并清理相关资源，同时
- 又能对任何子任务中的错误做出及时响应，而无需等待一段可能无界的时间，
那么我认为我们基本上需要一种方式，能够响应发生在另一个任务中的事件，请求任意任务尽早并及时地退出。换句话说，我们需要一种取消机制。
我们是在当前正在探讨的特定范式下得出这一结论的，但我认为它具有更广泛的适用性，细想之下也相当直观。在任何并发范式中，你都会有某种形式的“多个协作的并发任务”，这就意味着你需要回答“如果其中一个意外死亡会发生什么”。而反过来，除了“我们请求其他任务取消并尽早终止”之外，我很难想象还有什么完全通用的答案。
当然，通过一些临时机制的组合以及谨慎的推理和构造，完全有可能在没有通用取消机制的情况下实现特定的并发程序或模式。但我很难设想一个通用的、可组合的并发范式会不需要它。
我觉得这个结论令人不快，因为实现和支持取消是很难的。它给几乎每一处代码都引入了一条额外的错误路径，而且这条路径本质上就是异步的，难以推理和测试。历史上对取消机制的尝试，如 C 语言的pthread_cancel、Java 的Thread.stop 以及 Ruby 的Thread.terminate，往好了说是极其微妙且容易出错，往坏了说则是根本无法使用。
至少在并发的背景下，我们确实有一些优势。如果我们已经在编写并发代码，取消机制只是多增加了一件可能异步发生的事情，但至少我们已经在面对某种形式的这类问题了。在像 asyncio 这样的协作式并发系统中，我们还可以将取消限制为仅在 await 点发生，这缩小了潜在混乱的范围。与此同时，Go 则将取消编码到Context 对象中，并要求代码显式地检查取消，这带来了另一组权衡。
更普遍地说，在过去几十年里，作为一个领域我们已经学到了很多，而其中一些较新的系统似乎确实拥有了在实践中或多或少可行的通用取消机制。不过，本文并不打算深入探讨取消机制的挑战和设计空间，所以暂时我就假设我们确实拥有某种取消 API5，然后继续往下说。
带取消的任务树
如果我们确实有能力取消任务，我们就可以将它与“任务树”的想法结合起来，产生一个对并发错误处理相当通用的解决方案：
- 如果由
TaskLauncher启动的任何任务抛出了异常（包括父任务本身——即运行上下文管理器的那个任务），我们就取消每一个其他任务（既包括子任务，也包括父任务本身）。 - 我们为子任务提供一种检测取消并清理自身资源的机制。通常这意味着以某种形式复用我们常规的错误处理机制；例如，取消可能会抛出一个
CancelledError异常，任务可以捕获并重新抛出它，或者它们可以使用finally块或上下文管理器。 - 在退出
TaskLauncher上下文时，我们会等待所有子任务退出，无论是成功退出、因未处理异常退出，还是响应取消而退出。 - 然后，如果任何子任务抛出了错误，我们就将其重新抛到父任务中。
有了这一机制，并发错误的行为现在就与单线程错误相当相似了。如果未被处理，它们会被捕获并向上传播。它们可以使用我们常规的错误处理机制来捕获和处理（包括源自子任务的错误）。只要我们按照常规单线程方式编写在出错后进行清理的代码，那么（至少在很大程度上）我们在多任务环境下也能获得恰当的清理。
作为交换，我们确实对并发代码施加了额外的结构要求：我们必须将任务嵌套到父子层级中，并确保它们的生命周期也相应地嵌套。
结构化并发
说到这里，我得承认，这些都不是什么新想法，也没有一个是我的发明（尽管我还没见过以这种形式来阐述该想法的其他文章）。这种具有嵌套生命周期、并在树上下自动取消的任务树范式，近年来正以“结构化并发”之名缓慢而稳定地获得关注和采用。
其中的许多想法和组件都有着悠久的渊源，但据我所知，这个概念在 2016 年被首次命名，并很可能由trio 框架以及 njs（Trio 的创建者和主要维护者）那篇深入探讨该思想的文章做了最好的普及。
从 Python 3.11 开始，Python 自带的 asyncio 就包含了一个TaskGroup 类，它本质上就是我上面勾勒的 TaskLauncher 的一个（可用于生产的）版本；trio 则把自己的版本称为“nursery”。在 Go 中，errgroup 包提供了基本相同的语义，它构建在 Go 提供了取消支持的context 包之上。
结构化并发有许多优点，我和许多其他人都发现，以这种风格编写程序会让编写正确、安全的并发代码容易得多（尽管其他挑战依然存在！）。我强烈推荐 njs 的那篇经典文章——我前面也链接过——以更深入地探索这一范式及其优势。
尾声：为何是错误处理？
最后，我想对错误处理做一点反思，以及当初为何会从这个视角开始思考并写下这篇文章。
我觉得，当程序员想到错误处理时，它常常被归为“健壮性”问题，或是“生产环境”或“严肃软件”才需要关心的问题——这是一个只有在“大规模”、需要“可靠”、或要在多种不同环境中运行并处理来自网络的意外输入等等情况下才需要关注的话题。
这些都没错，而且对于这类系统，仔细思考可能哪里会出错、会如何出错以及如何谨慎处理，确实非常重要。
话虽如此，我开始这一思考的出发点并非“成熟、健壮的程序”，而是完全相反的一端：思考从零开始编写新代码的开发体验。正如我前面简要提到的，在编写一个新程序时——无论是否并发——往往会有一个早期阶段充斥着大量“愚蠢的 bug”，你只需要尽快逐个解决它们。
我在结构化并发框架之外编写并发程序的经验是，想要仅仅跑通“运行程序、发现愚蠢 bug、修复愚蠢 bug”这一基本开发循环，往往会变得异常令人沮丧，恰恰是因为那些在单线程程序中会漂亮地打印出栈回溯并退出的愚蠢 bug，在并发程序中却常常会演变成死锁、被吞掉，或是出现更离奇的情况。而且，我发现临时去添加错误处理的尝试有时会让情况更糟！例如，我有时会发现“自然”的做法是通过某个管道来“转发”错误，以便在大型并发操作的末尾集中收集所有错误并统一记录。这种做法可行，但有时也意味着直到整个程序运行结束你才会发现任何错误，这在开发过程中实在令人抓狂！
因此，我发现采用结构化并发的方法，或至少将其作为一种基本的思维方式和范式——即便我的环境中可能并没有一个“真正”的结构化并发库——实际上会让并发程序从一开始就显著更容易编写和调试，即便是对于一次性的原型也是如此——它几乎能立刻带来回报，而不仅仅是“最终”或“在生产环境中”才见效。
longjmp可以作为一种原语来帮助实现异常和栈展开。但就其本身而言，它是一个更底层的原语，并允许多种多样的替代模式。↩我更关心的是逻辑上的并发而非硬件并行，因此我将使用“任务”一词来指代彼此独立的线性执行序列，它们可能在时间上与任意数量的其他此类序列交错执行。↩
本讨论中的大部分内容旨在广泛适用于多种语言和并发框架，但为了具体说明，我将使用 Python 示例。除其他好处外，Python 同时支持线程和协作式异步，这让我能够探索多种范式。↩
也许最简单的修复方法是完全去掉
asyncio.Event，转而自行对Task对象进行asyncio.gather。在这里这很简单，但在“工作单元”与“子任务”之间不一定存在一对一映射时，就不总是这么简单了。↩
评论
登录后参与讨论