Following system colour scheme - Python 增强提案 Selected dark colour scheme - Python 增强提案 Selected light colour scheme - Python 增强提案

Python 增强提案

PEP 550 – 执行上下文

作者:
Yury Selivanov <yury at edgedb.com>, Elvis Pranskevichus <elvis at edgedb.com>
状态:
已撤回
类型:
标准跟踪
创建日期:
2017年8月11日
Python 版本:
3.7
发布历史:
2017年8月11日, 2017年8月15日, 2017年8月18日, 2017年8月25日, 2017年9月1日
目录

摘要

此 PEP 添加了一个新的通用机制，用于在诸如 Python 生成器和协程等乱序执行的上下文中确保对非本地状态的一致访问。

线程局部存储（如 threading.local()）对于在同一操作系统线程中并发执行的程序来说是不够的。此 PEP 提出了一种解决此问题的方法。

PEP 状态

由于其广泛性以及在某些方面缺乏普遍共识，此 PEP 已被撤回，并被更简单的 PEP 567 所取代，该 PEP 已被接受并包含在 Python 3.7 中。

PEP 567 实现了一个核心思想，但将 ContextVar 支持限制在异步任务，同时保持生成器行为不变。后者可能会在将来的 PEP 中重新讨论。

基本原理

在 Python 中出现异步编程之前，程序使用操作系统线程来实现并发。线程特定状态的需求通过 threading.local() 及其 C API 等效项 PyThreadState_GetDict() 来解决。

线程局部存储 (TLS) 常见依赖的示例

  • 上下文管理器，如 decimal contexts、numpy.errstate 和 warnings.catch_warnings。
  • 与请求相关的数据，例如 Web 应用程序中的安全令牌和请求数据、gettext 的语言上下文等。
  • 大型代码库中的性能分析、跟踪和日志记录。

不幸的是，TLS 对于在单个线程中并发执行的程序效果不佳。Python 生成器是并发程序的简单示例。考虑以下代码：

def fractions(precision, x, y):
    with decimal.localcontext() as ctx:
        ctx.prec = precision
        yield Decimal(x) / Decimal(y)
        yield Decimal(x) / Decimal(y ** 2)

g1 = fractions(precision=2, x=1, y=3)
g2 = fractions(precision=6, x=2, y=3)

items = list(zip(g1, g2))

items 的预期值是：

[(Decimal('0.33'), Decimal('0.666667')),
 (Decimal('0.11'), Decimal('0.222222'))]

令人惊讶的是，实际结果是：

[(Decimal('0.33'), Decimal('0.666667')),
 (Decimal('0.111111'), Decimal('0.222222'))]

这是因为隐式的 Decimal 上下文存储为线程局部变量，因此 fractions() 生成器的并发迭代会损坏状态。特别是对于 Decimal，当前唯一的解决方法是为所有算术运算使用显式的上下文方法调用 [28]。可以说，这削弱了重载运算符的有用性，并使简单的公式也难以读写。

协程是 TLS 不可靠性是一个重要问题的另一类 Python 代码。

TLS 在异步代码中的不足导致了临时解决方案的泛滥，这些解决方案范围有限，并且不支持所有必需的用例。

目前的状况是，任何依赖 TLS 的库（包括标准库）在异步代码或生成器中使用时都可能出错（参见 [3] 作为示例问题）。

一些支持协程或生成器的语言建议将上下文手动作为参数传递给每个函数，请参阅 [1] 了解示例。然而，这种方法对 Python 的用途有限，因为 Python 拥有一个庞大的生态系统，该生态系统是为了与类似 TLS 的上下文协同工作而构建的。此外，像 decimal 或 numpy 这样的库在重载运算符实现中隐式依赖于上下文。

.NET 运行时支持 async/await，它为这个问题提供了一个通用解决方案，称为 ExecutionContext（参见 [2]）。

目标

此 PEP 的目标是提供一个更可靠的 threading.local() 替代方案，该方案：

  • 提供机制和 API 来解决协程和生成器中的非本地状态问题；
  • 为同步代码实现类似 TLS 的语义，以便像 decimal 和 numpy 这样的用户可以切换到新机制，而向后兼容性风险最小；
  • 对现有代码或将使用新机制的代码（包括 C 扩展）没有或只有可忽略的性能影响。

高层级规范

此 PEP 的完整规范分为三个部分：

  • 高层级规范（本节）：对整体解决方案的描述。我们展示了它如何应用于用户代码中的生成器和协程，而不深入探讨实现细节。
  • 详细规范：对新概念、API 和标准库相关更改的完整描述。
  • 实现细节：对用于实现此 PEP 的数据结构和算法的描述和分析，以及对 CPython 的必要更改。

在本节中，我们将执行上下文定义为非本地状态的不透明容器，它允许在并发执行环境中对其内容进行一致访问。

上下文变量是在执行上下文中表示值的对象。调用 contextvars.ContextVar(name) 来创建新的上下文变量对象。上下文变量对象有三个方法：

  • get()：返回当前执行上下文中变量的值；
  • set(value)：设置当前执行上下文中变量的值；
  • delete()：可用于恢复变量状态，其目的和语义在设置和恢复上下文变量中解释。

常规单线程代码

在不涉及生成器或协程的常规单线程代码中，上下文变量的行为类似于全局变量。

var = contextvars.ContextVar('var')

def sub():
    assert var.get() == 'main'
    var.set('sub')

def main():
    var.set('main')
    sub()
    assert var.get() == 'sub'

多线程代码

在多线程代码中，上下文变量的行为类似于线程局部变量。

var = contextvars.ContextVar('var')

def sub():
    assert var.get() is None  # The execution context is empty
                              # for each new thread.
    var.set('sub')

def main():
    var.set('main')

    thread = threading.Thread(target=sub)
    thread.start()
    thread.join()

    assert var.get() == 'main'

生成器

与常规函数调用不同，生成器可以协作地将执行控制权让给调用者。此外，生成器不控制它在让出后从哪里继续执行。它可能从任意代码位置恢复。

出于这些原因，生成器最不令人意外的行为如下：

  • 对上下文变量的更改始终是局部的，在外部上下文中不可见，但在生成器调用的代码中可见；
  • 一旦在生成器中设置，上下文变量保证在迭代之间不会改变；
  • 外部上下文（生成器正在被迭代的地方）中的上下文变量更改对生成器可见，除非这些变量在生成器内部也被修改过。

让我们回顾一下：

var1 = contextvars.ContextVar('var1')
var2 = contextvars.ContextVar('var2')

def gen():
    var1.set('gen')
    assert var1.get() == 'gen'
    assert var2.get() == 'main'
    yield 1

    # Modification to var1 in main() is shielded by
    # gen()'s local modification.
    assert var1.get() == 'gen'

    # But modifications to var2 are visible
    assert var2.get() == 'main modified'
    yield 2

def main():
    g = gen()

    var1.set('main')
    var2.set('main')
    next(g)

    # Modification of var1 in gen() is not visible.
    assert var1.get() == 'main'

    var1.set('main modified')
    var2.set('main modified')
    next(g)

现在，让我们回顾一下原理部分中的十进制精度示例，看看执行上下文如何改善这种情况：

import decimal

# create a new context var
decimal_ctx = contextvars.ContextVar('decimal context')

# Pre-PEP 550 Decimal relies on TLS for its context.
# For illustration purposes, we monkey-patch the decimal
# context functions to use the execution context.
# A real working fix would need to properly update the
# C implementation as well.
def patched_setcontext(context):
    decimal_ctx.set(context)

def patched_getcontext():
    ctx = decimal_ctx.get()
    if ctx is None:
        ctx = decimal.Context()
        decimal_ctx.set(ctx)
    return ctx

decimal.setcontext = patched_setcontext
decimal.getcontext = patched_getcontext

def fractions(precision, x, y):
    with decimal.localcontext() as ctx:
        ctx.prec = precision
        yield MyDecimal(x) / MyDecimal(y)
        yield MyDecimal(x) / MyDecimal(y ** 2)

g1 = fractions(precision=2, x=1, y=3)
g2 = fractions(precision=6, x=2, y=3)

items = list(zip(g1, g2))

items 的值为：

[(Decimal('0.33'), Decimal('0.666667')),
 (Decimal('0.11'), Decimal('0.222222'))]

这与预期结果相符。

协程和异步任务

与生成器一样，协程也可以让出并重新获得控制权。与生成器的主要区别在于，协程不会将控制权让给直接的调用者。相反，整个协程调用堆栈（通过 await 链式连接的协程）会切换到另一个协程调用堆栈。在这方面，await 一个协程在概念上类似于常规函数调用，而协程链（或“任务”，例如 asyncio.Task）在概念上类似于线程。

从这种相似性，我们得出结论，协程中的上下文变量应该表现得像“任务局部变量”：

  • 在协程中对上下文变量的更改对等待它的协程是可见的；
  • 在等待之前在调用者中对上下文变量所做的更改对被等待的协程是可见的；
  • 在一个任务中对上下文变量所做的更改在其他任务中不可见；
  • 由其他任务生成的任务继承父任务的执行上下文，但父任务中在子任务生成之后对上下文变量所做的任何更改都不可见。

最后一点显示的行为与操作系统线程不同。操作系统线程默认不继承执行上下文。原因有两个：常见使用意图和向后兼容性。

任务继承上下文而线程不继承的主要原因是常见使用意图。任务通常用于相对短暂的操作，这些操作在逻辑上与生成任务的代码相关（例如，在 asyncio 中运行带有超时的协程）。另一方面，操作系统线程通常用于长时间运行、逻辑上独立的 कोड。

在向后兼容性方面，我们希望执行上下文的行为类似于 threading.local()。这样，库就可以开始使用执行上下文来代替 TLS，而不会有破坏现有代码兼容性的风险。

让我们回顾几个示例来说明我们刚刚定义的语义。

单个任务中的上下文变量传播

import asyncio

var = contextvars.ContextVar('var')

async def main():
    var.set('main')
    await sub()
    # The effect of sub() is visible.
    assert var.get() == 'sub'

async def sub():
    assert var.get() == 'main'
    var.set('sub')
    assert var.get() == 'sub'

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

任务之间的上下文变量传播

import asyncio

var = contextvars.ContextVar('var')

async def main():
    var.set('main')
    loop.create_task(sub())  # schedules asynchronous execution
                             # of sub().
    assert var.get() == 'main'
    var.set('main changed')

async def sub():
    # Sleeping will make sub() run after
    # "var" is modified in main().
    await asyncio.sleep(1)

    # The value of "var" is inherited from main(), but any
    # changes to "var" made in main() after the task
    # was created are *not* visible.
    assert var.get() == 'main'

    # This change is local to sub() and will not be visible
    # to other tasks, including main().
    var.set('sub')

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

如上所示，对执行上下文的更改是任务局部的，并且任务获取创建时的执行上下文的快照。

有一个狭窄的边缘情况可能导致意外行为。请考虑以下示例，其中我们在嵌套协程中修改了上下文变量：

async def sub(var_value):
    await asyncio.sleep(1)
    var.set(var_value)

async def main():
    var.set('main')

    # waiting for sub() directly
    await sub('sub-1')

    # var change is visible
    assert var.get() == 'sub-1'

    # waiting for sub() with a timeout;
    await asyncio.wait_for(sub('sub-2'), timeout=2)

    # wait_for() creates an implicit task, which isolates
    # context changes, which means that the below assertion
    # will fail.
    assert var.get() == 'sub-2'  #  AssertionError!

然而，依赖上下文更改泄露给调用者最终是一种不良模式。因此，上述示例中所示的行为不被视为主要问题，可以通过适当的文档来解决。

详细规范

概念上，执行上下文 (EC) 是逻辑上下文的堆栈。每个 Python 线程始终只有一个活动 EC。

逻辑上下文 (LC) 是上下文变量与其在该特定 LC 中的值之间的映射。

上下文变量是表示执行上下文中值的对象。通过调用 contextvars.ContextVar(name: str) 来创建新的上下文变量对象。传递的 name 参数的值不被 EC 机制使用，但可用于调试和内省。

上下文变量对象具有以下方法和属性：

  • name：传递给 ContextVar() 的值。
  • get(*, topmost=False, default=None)：如果 *topmost* 为 False（默认值），则从上到下遍历执行上下文，直到找到变量值。如果 *topmost* 为 True，则返回最顶层逻辑上下文中的变量值。如果未找到变量值，则返回 *default* 的值。
  • set(value)：在最顶层逻辑上下文中设置变量的值。
  • delete()：从最顶层逻辑上下文中删除变量。当恢复逻辑上下文到 set() 调用之前的状态时很有用，例如在上下文管理器中，有关更多信息请参阅设置和恢复上下文变量。

生成器

每个生成器对象在创建时在其 __logical_context__ 属性中存储一个空的逻辑上下文对象。此逻辑上下文在每次生成器迭代开始时推送到执行上下文，并在结束时弹出。

var1 = contextvars.ContextVar('var1')
var2 = contextvars.ContextVar('var2')

def gen():
    var1.set('var1-gen')
    var2.set('var2-gen')

    # EC = [
    #     outer_LC(),
    #     gen_LC({var1: 'var1-gen', var2: 'var2-gen'})
    # ]
    n = nested_gen()  # nested_gen_LC is created
    next(n)
    # EC = [
    #     outer_LC(),
    #     gen_LC({var1: 'var1-gen', var2: 'var2-gen'})
    # ]

    var1.set('var1-gen-mod')
    var2.set('var2-gen-mod')
    # EC = [
    #     outer_LC(),
    #     gen_LC({var1: 'var1-gen-mod', var2: 'var2-gen-mod'})
    # ]
    next(n)

def nested_gen():
    # EC = [
    #     outer_LC(),
    #     gen_LC({var1: 'var1-gen', var2: 'var2-gen'}),
    #     nested_gen_LC()
    # ]
    assert var1.get() == 'var1-gen'
    assert var2.get() == 'var2-gen'

    var1.set('var1-nested-gen')
    # EC = [
    #     outer_LC(),
    #     gen_LC({var1: 'var1-gen', var2: 'var2-gen'}),
    #     nested_gen_LC({var1: 'var1-nested-gen'})
    # ]
    yield

    # EC = [
    #     outer_LC(),
    #     gen_LC({var1: 'var1-gen-mod', var2: 'var2-gen-mod'}),
    #     nested_gen_LC({var1: 'var1-nested-gen'})
    # ]
    assert var1.get() == 'var1-nested-gen'
    assert var2.get() == 'var2-gen-mod'

    yield

# EC = [outer_LC()]

g = gen()  # gen_LC is created for the generator object `g`
list(g)

# EC = [outer_LC()]

上面的代码片段显示了生成器生命周期中执行上下文堆栈的状态。

contextlib.contextmanager

contextlib.contextmanager() 装饰器可用于将生成器转换为上下文管理器。可以这样定义一个临时修改上下文变量值的上下文管理器：

var = contextvars.ContextVar('var')

@contextlib.contextmanager
def var_context(value):
    original_value = var.get()

    try:
        var.set(value)
        yield
    finally:
        var.set(original_value)

不幸的是，这不会立即奏效，因为对 var 变量的修改仅限于 var_context() 生成器，因此在 with 块内不可见。

def func():
    # EC = [{}, {}]

    with var_context(10):
        # EC becomes [{}, {}, {var: 10}] in the
        # *precision_context()* generator,
        # but here the EC is still [{}, {}]

        assert var.get() == 10  # AssertionError!

修复此问题的方法是将生成器的 __logical_context__ 属性设置为 None。这将导致生成器避免修改执行上下文堆栈。

我们修改 contextlib.contextmanager() 装饰器以设置 genobj.__logical_context__ 为 None，以生成行为良好的上下文管理器。

def func():
    # EC = [{}, {}]

    with var_context(10):
        # EC = [{}, {var: 10}]
        assert var.get() == 10

    # EC becomes [{}, {var: None}]

枚举上下文变量

ExecutionContext.vars() 方法返回一个 ContextVar 对象的列表，这些对象在执行上下文中具有值。此方法主要用于内省和日志记录。

协程

在 CPython 中，协程与生成器共享实现。区别在于，在协程中 __logical_context__ 默认为 None。这影响了 async def 协程和旧式基于生成器的协程（用 @types.coroutine 装饰的生成器）。

异步生成器

异步生成器中的执行上下文语义与常规生成器没有区别。

asyncio

asyncio 使用 Loop.call_soon、Loop.call_later 和 Loop.call_at 来调度函数的异步执行。asyncio.Task 使用 call_soon() 来运行包装的协程。

我们修改 Loop.call_{at,later,soon} 以接受新的可选 *execution_context* 关键字参数，该参数默认为当前执行上下文的副本。

def call_soon(self, callback, *args, execution_context=None):
    if execution_context is None:
        execution_context = contextvars.get_execution_context()

    # ... some time later

    contextvars.run_with_execution_context(
        execution_context, callback, args)

contextvars.get_execution_context() 函数返回当前执行上下文的浅拷贝。这里的浅拷贝意味着一个新的执行上下文，它：

  • 对副本的查找提供与原始执行上下文相同的结果，并且
  • 原始执行上下文中的任何更改都不会影响副本，并且
  • 对副本的任何更改都不会影响原始执行上下文。

以下任一选项都满足复制要求：

  • 一个带有逻辑上下文浅拷贝的新堆栈；
  • 一个带有单个压缩逻辑上下文的新堆栈。

contextvars.run_with_execution_context(ec, func, *args, **kwargs) 函数以 *ec* 作为执行上下文来运行 func(*args, **kwargs)。该函数执行以下步骤：

  1. 将 *ec* 设置为当前线程的当前执行上下文堆栈。
  2. 将一个空的逻辑上下文推送到堆栈。
  3. 运行 func(*args, **kwargs)。
  4. 将逻辑上下文从堆栈中弹出。
  5. 恢复原始执行上下文堆栈。
  6. 返回或引发 func() 的结果。

这些步骤确保 *ec* 不能被 *func* 修改，这使得 run_with_execution_context() 具有幂等性。

asyncio.Task 的修改如下：

class Task:
    def __init__(self, coro):
        ...
        # Get the current execution context snapshot.
        self._exec_context = contextvars.get_execution_context()

        # Create an empty Logical Context that will be
        # used by coroutines run in the task.
        coro.__logical_context__ = contextvars.LogicalContext()

        self._loop.call_soon(
            self._step,
            execution_context=self._exec_context)

    def _step(self, exc=None):
        ...
        self._loop.call_soon(
            self._step,
            execution_context=self._exec_context)
        ...

转换为迭代器的生成器

任何 Python 生成器都可以表示为等效的迭代器。Cython 等编译器依赖于此公理。就执行上下文而言，这样的迭代器应与它所表示的生成器具有相同的行为。

这意味着需要有一个 Python API 来创建新的逻辑上下文并使用给定的逻辑上下文运行代码。

contextvars.LogicalContext() 函数创建一个新的空逻辑上下文。

contextvars.run_with_logical_context(lc, func, *args, **kwargs) 函数可用于在指定的逻辑上下文中运行函数。调用可能会修改 *lc*。

contextvars.run_with_logical_context() 函数执行以下步骤：

  1. 将 *lc* 推送到当前执行上下文堆栈。
  2. 运行 func(*args, **kwargs)。
  3. 将 *lc* 从执行上下文堆栈中弹出。
  4. 返回或引发 func() 的结果。

通过使用 LogicalContext() 和 run_with_logical_context()，我们可以模拟生成器行为，如下所示：

class Generator:

    def __init__(self):
        self.logical_context = contextvars.LogicalContext()

    def __iter__(self):
        return self

    def __next__(self):
        return contextvars.run_with_logical_context(
            self.logical_context, self._next_impl)

    def _next_impl(self):
        # Actual __next__ implementation.
        ...

让我们看看如何将此模式应用于示例生成器：

# create a new context variable
var = contextvars.ContextVar('var')

def gen_series(n):
    var.set(10)

    for i in range(1, n):
        yield var.get() * i

# gen_series is equivalent to the following iterator:

class CompiledGenSeries:

    # This class is what the `gen_series()` generator can
    # be transformed to by a compiler like Cython.

    def __init__(self, n):
        # Create a new empty logical context,
        # like the generators do.
        self.logical_context = contextvars.LogicalContext()

        # Initialize the generator in its LC.
        # Otherwise `var.set(10)` in the `_init` method
        # would leak.
        contextvars.run_with_logical_context(
            self.logical_context, self._init, n)

    def _init(self, n):
        self.i = 1
        self.n = n
        var.set(10)

    def __iter__(self):
        return self

    def __next__(self):
        # Run the actual implementation of __next__ in our LC.
        return contextvars.run_with_logical_context(
            self.logical_context, self._next_impl)

    def _next_impl(self):
        if self.i == self.n:
            raise StopIteration

        result = var.get() * self.i
        self.i += 1
        return result

对于手动编写的迭代器，通常不需要这种上下文管理方法，并且更容易直接在 __next__ 中设置和恢复上下文变量。

class MyIterator:

    # ...

    def __next__(self):
        old_val = var.get()
        try:
            var.set(new_val)
            # ...
        finally:
            var.set(old_val)

实施

执行上下文实现为逻辑上下文的不可变链接列表，其中每个逻辑上下文是不可变的弱键映射。当前活动的执行上下文的指针存储在 OS 线程状态中。

                  +-----------------+
                  |                 |     ec
                  |  PyThreadState  +-------------+
                  |                 |             |
                  +-----------------+             |
                                                  |
ec_node             ec_node             ec_node   v
+------+------+     +------+------+     +------+------+
| NULL |  lc  |<----| prev |  lc  |<----| prev |  lc  |
+------+--+---+     +------+--+---+     +------+--+---+
          |                   |                   |
LC        v         LC        v         LC        v
+-------------+     +-------------+     +-------------+
| var1: obj1  |     |    EMPTY    |     | var1: obj4  |
| var2: obj2  |     +-------------+     +-------------+
| var3: obj3  |
+-------------+

选择不可变列表作为基本数据结构，是因为它能够高效地实现 contextvars.get_execution_context()，后者将由异步任务和回调频繁使用。当 EC 不可变时，get_execution_context() 可以简单地按引用复制当前执行上下文。

def get_execution_context(self):
    return PyThreadState_Get().ec

让我们回顾所有可能的上下文修改场景：

  • 调用 ContextVariable.set() 方法：
    def ContextVar_set(self, val):
    # See a more complete set() definition
    # in the `Context Variables` section.

    tstate = PyThreadState_Get()
    top_ec_node = tstate.ec
    top_lc = top_ec_node.lc
    new_top_lc = top_lc.set(self, val)
    tstate.ec = ec_node(
        prev=top_ec_node.prev,
        lc=new_top_lc)
  • 调用 contextvars.run_with_logical_context()，在这种情况下，传递的逻辑上下文对象会被附加到执行上下文中。
    def run_with_logical_context(lc, func, *args, **kwargs):
    tstate = PyThreadState_Get()

    old_top_ec_node = tstate.ec
    new_top_ec_node = ec_node(prev=old_top_ec_node, lc=lc)

    try:
        tstate.ec = new_top_ec_node
        return func(*args, **kwargs)
    finally:
        tstate.ec = old_top_ec_node
  • 调用 contextvars.run_with_execution_context()，在这种情况下，当前执行上下文被设置为传递的执行上下文，并在其末尾附加一个新的空逻辑上下文。
    def run_with_execution_context(ec, func, *args, **kwargs):
    tstate = PyThreadState_Get()

    old_top_ec_node = tstate.ec
    new_lc = contextvars.LogicalContext()
    new_top_ec_node = ec_node(prev=ec, lc=new_lc)

    try:
        tstate.ec = new_top_ec_node
        return func(*args, **kwargs)
    finally:
        tstate.ec = old_top_ec_node
  • 调用 genobj.send()、genobj.throw() 或 genobj.close() 在 genobj 生成器上，在这种情况下，genobj 中记录的逻辑上下文被推送到堆栈。
    PyGen_New(PyGenObject *gen):
    if (gen.gi_code.co_flags &
            (CO_COROUTINE | CO_ITERABLE_COROUTINE)):
        # gen is an 'async def' coroutine, or a generator
        # decorated with @types.coroutine.
        gen.__logical_context__ = None
    else:
        # Non-coroutine generator
        gen.__logical_context__ = contextvars.LogicalContext()

gen_send(PyGenObject *gen, ...):
    tstate = PyThreadState_Get()

    if gen.__logical_context__ is not None:
        old_top_ec_node = tstate.ec
        new_top_ec_node = ec_node(
            prev=old_top_ec_node,
            lc=gen.__logical_context__)

        try:
            tstate.ec = new_top_ec_node
            return _gen_send_impl(gen, ...)
        finally:
            gen.__logical_context__ = tstate.ec.lc
            tstate.ec = old_top_ec_node
    else:
        return _gen_send_impl(gen, ...)
  • 协程和异步生成器与生成器共享实现，并且上述更改也适用于它们。

在某些情况下，可能需要压缩 EC 以限制链的长度。例如，考虑以下极端情况：

async def repeat(coro, delay):
    await coro()
    await asyncio.sleep(delay)
    loop.create_task(repeat(coro, delay))

async def ping():
    print('ping')

loop = asyncio.get_event_loop()
loop.create_task(repeat(ping, 1))
loop.run_forever()

在上述代码中，EC 链会随着 repeat() 的调用而增长。每个新任务都会调用 contextvars.run_with_execution_context()，该调用会将一个新的逻辑上下文附加到链中。为防止无界增长，contextvars.get_execution_context() 会检查链是否超过预定最大值，如果超过，则将链压缩成单个 LC。

def get_execution_context():
    tstate = PyThreadState_Get()

    if tstate.ec_len > EC_LEN_MAX:
        squashed_lc = contextvars.LogicalContext()

        ec_node = tstate.ec
        while ec_node:
            # The LC.merge() method does not replace
            # existing keys.
            squashed_lc = squashed_lc.merge(ec_node.lc)
            ec_node = ec_node.prev

        return ec_node(prev=NULL, lc=squashed_lc)
    else:
        return tstate.ec

逻辑上下文

逻辑上下文是具有以下垃圾回收属性的不可变弱键映射：

  • ContextVar 对象仅从应用程序代码中强引用，而不是从执行上下文机制或它们指向的值中引用。这意味着没有可能延长其寿命的引用循环，或者阻止 GC 收集它们。
  • 只要在线程中存在引用它们的 ContextVar 键，就可以保证保存在执行上下文中的值保持活动状态。
  • 如果 ContextVar 被垃圾回收，它所有的值都将从所有上下文中删除，从而允许它们在需要时被 GC。
  • 如果 OS 线程结束执行，其线程状态将随其执行上下文一起清理，从而清理线程中绑定到所有上下文变量的所有值。

如前所述，我们需要 contextvars.get_execution_context() 始终快速，无论执行上下文的大小如何，因此逻辑上下文必然是一个不可变映射。

选择 dict 作为底层实现是次优的，因为 LC.set() 会导致 dict.copy()，这是一个 O(N) 操作，其中 *N* 是 LC 中的项目数。

当压缩 EC 时，get_execution_context() 是一个 O(M) 操作，其中 *M* 是 EC 中上下文变量值的总数。

因此，我们选择哈希数组映射树 (HAMT) 而不是 dict 作为逻辑上下文的底层实现。（Scala 和 Clojure 使用 HAMT 来实现高性能的不可变集合 [5]，[6]。）

使用 HAMT，.set() 成为 O(log N) 操作，并且由于 HAMT 的结构共享，get_execution_context() 压缩在平均情况下更有效。

有关 HAMT 与 dict 性能比较的更详细分析，请参阅附录：HAMT 性能分析。

上下文变量

（伪代码）如下实现了 ContextVar.get() 和 ContextVar.set() 方法：

class ContextVar:

    def get(self, *, default=None, topmost=False):
        tstate = PyThreadState_Get()

        ec_node = tstate.ec
        while ec_node:
            if self in ec_node.lc:
                return ec_node.lc[self]
            if topmost:
                break
            ec_node = ec_node.prev

        return default

    def set(self, value):
        tstate = PyThreadState_Get()
        top_ec_node = tstate.ec

        if top_ec_node is not None:
            top_lc = top_ec_node.lc
            new_top_lc = top_lc.set(self, value)
            tstate.ec = ec_node(
                prev=top_ec_node.prev,
                lc=new_top_lc)
        else:
            # First ContextVar.set() in this OS thread.
            top_lc = contextvars.LogicalContext()
            new_top_lc = top_lc.set(self, value)
            tstate.ec = ec_node(
                prev=NULL,
                lc=new_top_lc)

    def delete(self):
        tstate = PyThreadState_Get()
        top_ec_node = tstate.ec

        if top_ec_node is None:
            raise LookupError

        top_lc = top_ec_node.lc
        if self not in top_lc:
            raise LookupError

        new_top_lc = top_lc.delete(self)

        tstate.ec = ec_node(
            prev=top_ec_node.prev,
            lc=new_top_lc)

为了在性能敏感的代码路径（如 numpy 和 decimal）中实现高效访问，我们缓存了 ContextVar.get() 中的查找，使其在缓存命中时成为 O(1) 操作。缓存键由以下内容组成：

  • 新的 uint64_t PyThreadState->unique_id，这是一个全局唯一的线程状态标识符。它从新的 uint64_t PyInterpreterState->ts_counter 计算得出，该计数器在每次创建新的线程状态时递增。
  • 新的 uint64_t PyThreadState->stack_version，这是一个线程特定的计数器，在任何非空逻辑上下文被推送到堆栈或从堆栈弹出时递增。
  • uint64_t ContextVar->version 计数器，该计数器在任何操作系统线程的任何逻辑上下文中上下文变量值更改时递增。

然后，缓存实现如下：

class ContextVar:

    def set(self, value):
        ...  # implementation
        self.version += 1

    def get(self, *, default=None, topmost=False):
        if topmost:
            return self._get_uncached(
                default=default, topmost=topmost)

        tstate = PyThreadState_Get()
        if (self.last_tstate_id == tstate.unique_id and
                self.last_stack_ver == tstate.stack_version and
                self.last_version == self.version):
            return self.last_value

        value = self._get_uncached(default=default)

        self.last_value = value  # borrowed ref
        self.last_tstate_id = tstate.unique_id
        self.last_stack_version = tstate.stack_version
        self.last_version = self.version

        return value

请注意，last_value 是一个借用引用。我们假设如果版本检查正确，则值对象将保持活动状态。这使得上下文变量的值可以被正确地垃圾回收。

这种通用缓存方法类似于当前 decimal 的 C 实现缓存当前十进制上下文的方式，并且具有相似的性能特征。

性能考量

基于此 PEP 早期修订版本的参考实现的测试表明，在生成器微基准测试中速度减慢了 1-2%，在宏基准测试中没有明显差异。

非生成器和非异步代码的性能不受此 PEP 的影响。

新 API 总结

Python

此 PEP 引入了以下新的 Python API：

  1. 新的 contextvars.ContextVar(name: str='...') 类，其实例具有以下特性：
    • 只读的 .name 属性；
    • .get() 方法，返回当前执行上下文中变量的值；
    • .set() 方法，在当前逻辑上下文中设置变量的值；
    • .delete() 方法，从当前逻辑上下文中删除变量的值。
  2. 新的 contextvars.ExecutionContext() 类，表示一个执行上下文。
  3. 新的 contextvars.LogicalContext() 类，表示一个逻辑上下文。
  4. 新的 contextvars.get_execution_context() 函数，返回一个表示当前执行上下文副本的 ExecutionContext 实例。
  5. contextvars.run_with_execution_context(ec: ExecutionContext, func, *args, **kwargs) 函数，使用提供的执行上下文运行 *func*。
  6. contextvars.run_with_logical_context(lc: LogicalContext, func, *args, **kwargs) 函数，在当前执行上下文之上，使用提供的逻辑上下文运行 *func*。

C API

  1. PyContextVar * PyContext_NewVar(char *desc)：创建一个 PyContextVar 对象。
  2. PyObject * PyContext_GetValue(PyContextVar *, int topmost)：返回当前执行上下文中变量的值。
  3. int PyContext_SetValue(PyContextVar *, PyObject *)：在当前逻辑上下文中设置变量的值。
  4. int PyContext_DelValue(PyContextVar *)：从当前逻辑上下文中删除变量的值。
  5. PyLogicalContext * PyLogicalContext_New()：创建一个新的空 PyLogicalContext。
  6. PyExecutionContext * PyExecutionContext_New()：创建一个新的空 PyExecutionContext。
  7. PyExecutionContext * PyExecutionContext_Get()：返回当前的执行上下文。
  8. int PyContext_SetCurrent( PyExecutionContext *, PyLogicalContext *)：将传递的 EC 对象设置为活动线程状态的当前执行上下文，以及/或将传递的 LC 对象设置为当前逻辑上下文。

设计考虑

“yield from”是否应泄露上下文更改？

否。有人可能会认为 yield from 在语义上等同于调用函数，并且应该泄露上下文更改。然而，不可能同时满足以下条件：

  • next(gen)不泄露在 gen 中所做的上下文更改，并且
  • yield from gen泄露在 gen 中所做的上下文更改。

原因是 yield from 可以与部分迭代的生成器一起使用，该生成器已经具有本地上下文更改。

var = contextvars.ContextVar('var')

def gen():
    for i in range(10):
        var.set('gen')
        yield i

def outer_gen():
    var.set('outer_gen')
    g = gen()

    yield next(g)
    # Changes not visible during partial iteration,
    # the goal of this PEP:
    assert var.get() == 'outer_gen'

    yield from g
    assert var.get() == 'outer_gen'  # or 'gen'?

另一个例子是将显式 for..in yield 结构重构为 yield from 表达式。考虑以下代码：

def outer_gen():
    var.set('outer_gen')

    for i in gen():
        yield i
    assert var.get() == 'outer_gen'

我们希望使用 yield from 来重构它：

def outer_gen():
    var.set('outer_gen')

    yield from gen()
    assert var.get() == 'outer_gen'  # or 'gen'?

上述示例说明，当 yield from 可能泄露上下文更改时，重构生成器代码是不安全的。

因此，唯一明确且一致的行为是始终隔离生成器中的上下文更改，无论它们如何被迭代。

PyThreadState_GetDict() 是否应使用执行上下文？

否。PyThreadState_GetDict 基于 TLS，改变其语义将破坏向后兼容性。

PEP 521

PEP 521 提出了该问题的一种替代解决方案，它通过两个新方法扩展了上下文管理器协议：__suspend__() 和 __resume__()。类似地，异步上下文管理器协议也扩展了 __asuspend__() 和 __aresume__()。

这使得实现管理非本地状态的上下文管理器成为可能，这些上下文管理器在生成器和协程中行为正确。

例如，考虑以下使用执行状态的上下文管理器：

class Context:

    def __init__(self):
        self.var = contextvars.ContextVar('var')

    def __enter__(self):
        self.old_x = self.var.get()
        self.var.set('something')

    def __exit__(self, *err):
        self.var.set(self.old_x)

使用 PEP 521 的等效实现：

local = threading.local()

class Context:

    def __enter__(self):
        self.old_x = getattr(local, 'x', None)
        local.x = 'something'

    def __suspend__(self):
        local.x = self.old_x

    def __resume__(self):
        local.x = 'something'

    def __exit__(self, *err):
        local.x = self.old_x

这种方法的缺点是给上下文管理器协议和解释器实现增加了显著的复杂性。这种方法也可能对生成器和协程的性能产生负面影响。

此外，PEP 521 中的解决方案仅限于上下文管理器，并未提供任何机制来在异步任务和回调中传播状态。

能否在不修改 CPython 的情况下实现执行上下文？

否。

诚然，“任务局部变量”的概念可以在库中为协程实现（请参见，例如 [29] 和 [30]）。另一方面，生成器由 Python 解释器直接管理，因此它们的上下文也必须由解释器管理。

此外，执行上下文根本无法在第三方模块中实现，否则标准库（包括 decimal）将无法依赖它。

我们是否应更新 sys.displayhook 和其他 API 以使用 EC？

像通过覆盖 sys.stdout 来重定向 stdout，或者通过覆盖 sys.displayhook 函数来指定新的异常显示挂钩这样的 API，设计上会影响整个 Python 进程。其用户假设更改它们的效果将在操作系统线程之间可见。因此，我们不能仅仅让这些 API 使用新的执行上下文。

话虽如此，我们认为设计能够感知上下文的新 API 是可能的，但这超出了此 PEP 的范围。

Greenlets

Greenlet 是 Python 的一种替代协同调度实现。尽管 greenlet 包不是 CPython 的一部分，但 gevent 等流行框架依赖它，并且 greenlet 的修改能够支持执行上下文很重要。

概念上，greenlet 的行为与生成器非常相似，这意味着可以在 greenlet 的入口和出口周围进行类似的更改来添加对执行上下文的支持。此 PEP 提供了必要的 C API 来做到这一点。

上下文管理器作为修改接口

此 PEP 专注于底层机制和支持基本执行上下文操作的最小 API。

为了方便开发者使用，可以在 contextvars 模块中添加一个高层级上下文管理器接口。例如：

with contextvars.set_var(var, 'foo'):
    # ...

设置和恢复上下文变量

ContextVar.delete() 方法从最顶层逻辑上下文中删除上下文变量。

如果在最顶层逻辑上下文中找不到该变量，则会引发 LookupError，类似于 del var 在 var 不在作用域时引发 NameError。

当有（罕见的）需要正确恢复逻辑上下文状态时，例如当嵌套生成器想要临时修改逻辑上下文时，此方法很有用。

var = contextvars.ContextVar('var')

def gen():
    with some_var_context_manager('gen'):
        # EC = [{var: 'main'}, {var: 'gen'}]
        assert var.get() == 'gen'
        yield

    # EC = [{var: 'main modified'}, {}]
    assert var.get() == 'main modified'
    yield

def main():
    var.set('main')
    g = gen()
    next(g)
    var.set('main modified')
    next(g)

上述示例仅在存在一种方法可以在 gen() 中删除 var 时才能正确工作。在 __exit__() 中将其设置为“先前值”会掩盖在迭代之间在 main() 中所做的更改。

ContextVar API 的替代设计

堆叠值的逻辑上下文

根据此 PEP 中提出的设计，逻辑上下文是一个简单的 LC({ContextVar: value, ...}) 映射。另一种表示方法是为每个上下文变量存储一个值的堆栈：LC({ContextVar: [val1, val2, ...], ...})。

则 ContextVar 方法将是：

  • get(*, default=None) – 遍历逻辑上下文堆栈，并返回第一个非空逻辑上下文的顶部值；
  • push(val) – 将 *val* 推送到当前逻辑上下文的值堆栈上；
  • pop() – 从当前逻辑上下文的值堆栈中弹出顶部值。

与带有 set() 和 delete() 方法的单值设计相比，基于堆栈的方法允许更简单地实现 set/restore 模式。然而，这种方法的心理负担被认为更高，因为需要考虑两个堆栈：一个 LC 堆栈和一个每个 LC 中的值堆栈。

（此想法由 Nathaniel Smith 提出。）

ContextVar “set/reset”

还有一种方法是从 ContextVar.set() 返回一个特殊对象，该对象表示对当前逻辑上下文中上下文变量的修改。

var = contextvars.ContextVar('var')

def foo():
    mod = var.set('spam')

    # ... perform work

    mod.reset()  # Reset the value of var to the original value
                 # or remove it from the context.

这种方法的一个关键缺陷是，可以将上下文变量“修改对象”传递给在不同执行上下文中运行的代码，这会导致未定义的副作用。

向后兼容性

本提案保留 100% 的向后兼容性。

被拒绝的想法

threading.local() 接口的复制

考虑到并拒绝了为上下文变量选择类似 threading.local() 的接口，原因如下：

  • 对标准库和 Django 的调查表明，绝大多数 threading.local() 的用法涉及单个属性，这表明命名空间方法在实际应用中没有那么有用。
  • 使用 __getattr__() 而不是 .get() 进行值查找，无法提供指定查找深度的任何方法（即仅搜索顶层逻辑上下文）。
  • 单值 ContextVar 更容易从可见性的角度进行推理。假设 ContextVar() 是一个命名空间，并考虑以下代码：
    ns = contextvars.ContextVar('ns')

def gen():
    ns.a = 2
    yield
    assert ns.b == 'bar' # ??

def main():
    ns.a = 1
    ns.b = 'foo'
    g = gen()
    next(g)
    # should not see the ns.a modification in gen()
    assert ns.a == 1
    # but should gen() see the ns.b modification made here?
    ns.b = 'bar'
    yield

    上面的示例表明，推理同一上下文变量不同属性的可见性并非易事。

  • 单值 ContextVar 允许轻松实现查找缓存；
  • 单值 ContextVar 接口允许 C API 简单且与 Python API 基本相同。

另请参阅邮件列表讨论：[26]，[27]。

协程默认不泄露上下文更改

在此 PEP 的 V4 版本（版本历史）中，协程在执行上下文方面的行为被认为与生成器完全相同：等待的协程中的更改在外部协程中不可见。

这一想法被拒绝，因为它破坏了任务模型和线程模型的语义相似性，更具体地说，使得实现修改上下文变量的异步上下文管理器变得不可靠，因为 __aenter__ 是一个协程。

附录：HAMT 性能分析

../_images/pep-0550-hamt_vs_dict-v2.png

图 1。基准测试代码可以在这里找到：[9]。

上图表明：

  • HAMT 在所有基准测试的字典大小上都显示出接近 O(1) 的性能。
  • dict.copy() 在大约 100 个项目时变得非常慢。
../_images/pep-0550-lookup_hamt.png

图 2。基准测试代码可以在这里找到：[10]。

图 2 比较了 dict 与基于 HAMT 的不可变映射的查找成本。HAMT 的查找时间平均比 Python dict 查找慢 30-40%，考虑到后者已经非常优化，这是一个非常好的结果。

有研究 [8] 表明，HAMT 的性能还有进一步改进的空间。

CPython 的 HAMT 参考实现可以在这里找到：[7]。

致谢

感谢 Victor Petrovykh 就该主题进行的无数次讨论以及 PEP 的校对和编辑。

感谢 Nathaniel Smith 提出了 ContextVar 设计 [17] [18]，推动 PEP 向更完整的设计发展，并提出了在线程状态中拥有上下文堆栈的想法。

感谢 Alyssa (Nick) Coghlan 在邮件列表中提出的无数建议和想法，并提出了一个导致最初 PEP 版本完全重写的情况 [19]。

版本历史

  1. 初稿，发布于 2017 年 8 月 11 日 [20]。
  2. V2 发布于 2017 年 8 月 15 日 [21]。

    导致第一个版本完全重新设计的根本限制是，无法实现与生成器以相同方式与 EC 交互的迭代器（请参阅 [19]）。

    版本 2 是一个完整的重写，引入了新的术语（Local Context、Execution Context、Context Item）和新的 API。

  3. V3 发布于 2017 年 8 月 18 日 [22]。

    更新

    • Local Context 被重命名为 Logical Context。术语“local”有歧义，并与局部名称作用域冲突。
    • Context Item 被重命名为 Context Key，请参阅与 Alyssa Coghlan、Stefan Krah 和 Yury Selivanov 的讨论 [23] 了解详情。
    • 根据 Nathaniel Smith 的想法 [25]，调整了 Context Item get 缓存设计。
    • 协程在没有逻辑上下文的情况下创建；ceval 循环不再需要特殊处理 await 表达式（由 Alyssa Coghlan 在 [24] 中提出）。
  4. V4 发布于 2017 年 8 月 25 日 [31]。
    • 规范部分已完全重写。
    • 协程现在拥有自己的逻辑上下文。这意味着在与执行上下文交互方面，协程、生成器和异步生成器之间没有区别。
    • Context Key 被重命名为 Context Var。
    • 删除了生成器和协程在逻辑上下文隔离方面的区别。
  5. V5 发布于 2017 年 9 月 1 日：当前版本。

参考资料

来源：https://github.com/python/peps/blob/main/peps/pep-0550.rst

最后修改：2025-02-01 08:59:27 GMT