concurrent.interpreters --- 在同一进程中实现多解释器

在 3.14 版本加入。

源代码: Lib/concurrent/interpreters


concurrent.interpreters 模块在底层的 _interpreters 模块之上构建了更高级别的接口。

该模块主要旨在提供一个基本的 API,用于管理解释器(也称为“子解释器”)并在其中运行代码。运行操作主要涉及切换到某个解释器(在当前线程中),并调用该执行上下文中的函数。

对于并发性,解释器本身(以及本模块)除了隔离之外,并没有提供更多功能,而隔离本身作用不大。实际的并发性可通过 线程 单独获得。请参阅下文

参见

InterpreterPoolExecutor

在一个熟悉的接口中将线程与解释器结合起来。

隔离扩展模块

如何更新扩展模块以支持多解释器

PEP 554

PEP 734

PEP 684

可用性:非 WASI。

此模块在 WebAssembly 上不起作用或不可用。有关更多信息,请参阅 WebAssembly 平台

关键细节

在我们深入探讨之前,关于使用多解释器,有几个关键细节需要牢记:

  • 默认情况下是隔离的

  • 无隐式线程

  • 并非所有 PyPI 包都支持在多解释器中使用

引言

“解释器”实际上是 Python 运行时的执行上下文。它包含了运行时执行程序所需的所有状态。这包括导入状态和内置函数等。(每个线程,即使只有一个主线程,除了当前的解释器外,还有一些与当前异常和字节码求值循环相关的额外运行时状态。)

解释器的概念和功能自 Python 2.2 版以来就已存在,但该功能仅通过 C-API 提供且鲜为人知,并且其隔离性在 3.12 版之前相对不完整。

多解释器与隔离

一个 Python 实现可能支持在同一进程中使用多个解释器。CPython 就支持此功能。每个解释器实际上都与其他解释器隔离(除了一些经过精心管理的进程全局例外情况)。

这种隔离主要用于程序中不同逻辑组件之间的强分离,当您希望仔细控制这些组件如何交互时,这种隔离尤其有用。

备注

同一进程中的解释器在技术上永远无法彼此严格隔离,因为同一进程内的内存访问限制很少。Python 运行时会尽力实现隔离,但扩展模块可能会轻易破坏这种隔离。因此,在安全敏感的情况下,不应使用多解释器,因为它们不应该能够访问彼此的数据。

在解释器中运行

在不同的解释器中运行涉及在当前线程中切换到该解释器,然后调用某个函数。运行时将使用当前解释器的状态来执行该函数。concurrent.interpreters 模块提供了一个基本的 API,用于创建和管理解释器,以及执行切换并调用的操作。

该操作不会自动启动其他线程。不过,有一个辅助函数可以实现此功能。还有一个专门的辅助函数,用于在解释器中调用内置的 exec()

当在解释器中调用 exec()(或 eval())时,它们会使用该解释器的 __main__ 模块作为“全局”命名空间。对于不与任何模块关联的函数也是如此。这与从命令行调用的脚本在 __main__ 模块中运行的方式相同。

并发与并行

如前所述,解释器本身不提供任何并发性。它们严格代表运行时将*在当前线程中*使用的隔离执行上下文。这种隔离使它们类似于进程,但它们仍然享有进程内的高效性,就像线程一样。

尽管如此,解释器确实自然地支持某些类型的并发性。这种隔离有一个强大的副作用。它实现了一种与异步或线程不同的并发方法。这是一种类似于 CSP 或 actor 模型的并发模型,这种模型相对容易理解。

你可以在单个线程中利用这种并发模型,在解释器之间来回切换,类似于 Stackless 的风格。然而,当您将解释器与多个线程结合使用时,这种模型更为有用。这主要涉及启动一个新线程,在该线程中切换到另一个解释器并在那里运行您想要的代码。

Python 中的每个实际线程,即使您只在主线程中运行,都有其自己的*当前*执行上下文。多个线程可以使用同一个解释器,也可以使用不同的解释器。

从高层次来看,您可以将线程和解释器的组合视为具有可选共享功能的线程。

一个显著的好处是,解释器是充分隔离的,它们不共享GIL,这意味着将线程与多个解释器结合可以实现完全的多核并行。(自 Python 3.12 以来就是如此。)

解释器间通信

实际上,只有当我们有办法在多个解释器之间进行通信时,它们才有用。这通常涉及某种形式的消息传递,但甚至可以意味着以某种精心管理的方式共享数据。

考虑到这一点,concurrent.interpreters 模块提供了一个 queue.Queue 的实现,可通过 create_queue() 获得。

“共享”对象

任何在解释器之间实际共享的数据都会失去由 GIL 提供的线程安全性。在扩展模块中有多种方法可以处理这个问题。然而,从 Python 代码的角度来看,缺乏线程安全性意味着对象实际上无法被共享,只有少数例外。相反,必须创建一个副本,这意味着可变对象将不会保持同步。

默认情况下,大多数对象在传递给另一个解释器时会通过 pickle 进行复制。几乎所有的不可变内置对象要么是直接共享的,要么是高效复制的。例如:

有少数 Python 类型确实在解释器之间共享可变数据:

参考

该模块定义了以下函数:

concurrent.interpreters.list_all()

返回一个包含 Interpreter 对象的 list,每个对象对应一个现有的解释器。

concurrent.interpreters.get_current()

返回当前运行的解释器的 Interpreter 对象。

concurrent.interpreters.get_main()

返回主解释器的 Interpreter 对象。这是运行时创建的用于运行 REPL 或命令行中给定脚本的解释器。它通常是唯一的解释器。

concurrent.interpreters.create()

初始化一个新的(空闲的)Python 解释器,并为其返回一个 Interpreter 对象。

concurrent.interpreters.create_queue()

初始化一个新的跨解释器队列,并为其返回一个 Queue 对象。

解释器对象

class concurrent.interpreters.Interpreter(id)

当前进程中的单个解释器。

通常情况下,不应直接调用 Interpreter。而是应该使用 create() 或其他模块函数。

id

(只读)

底层解释器的 ID。

whence

(只读)

一个描述解释器来源的字符串。

is_running()

如果解释器当前正在其 __main__ 模块中执行代码,则返回 True,否则返回 False

close()

终结并销毁解释器。

prepare_main(ns=None, **kwargs)

在解释器的 __main__ 模块中绑定对象。

一些对象是实际共享的,一些是高效复制的,但大多数是通过 pickle 复制的。请参阅 “共享”对象

exec(code, /, dedent=True)

在解释器中运行给定的源代码(在当前线程中)。

call(callable, /, *args, **kwargs)

返回在解释器中运行给定函数的结果(在当前线程中)。

call_in_thread(callable, /, *args, **kwargs)

在解释器中运行给定的函数(在一个新线程中)。

异常

exception concurrent.interpreters.InterpreterError

这个异常是 Exception 的子类,在发生与解释器相关的错误时引发。

exception concurrent.interpreters.InterpreterNotFoundError

这个异常是 InterpreterError 的子类,在目标解释器不存在时引发。

exception concurrent.interpreters.ExecutionFailed

这个异常是 InterpreterError 的子类,在运行的代码引发未捕获的异常时引发。

excinfo

在另一个解释器中引发的异常的基本快照。

exception concurrent.interpreters.NotShareableError

这个异常是 TypeError 的子类,当一个对象无法发送到另一个解释器时引发。

解释器间通信

class concurrent.interpreters.Queue(id)

一个底层跨解释器队列的包装器,它实现了 queue.Queue 接口。底层的队列只能通过 create_queue() 创建。

一些对象是实际共享的,一些是高效复制的,但大多数是通过 pickle 复制的。请参阅 “共享”对象

id

(只读)

队列的 ID。

exception concurrent.interpreters.QueueEmptyError

这个异常是 queue.Empty 的子类,当队列为空时,会由 Queue.get()Queue.get_nowait() 引发。

exception concurrent.interpreters.QueueFullError

这个异常是 queue.Full 的子类,当队列已满时,会由 Queue.put()Queue.put_nowait() 引发。

基本用法

创建一个解释器并在其中运行代码

from concurrent import interpreters

interp = interpreters.create()

# Run in the current OS thread.

interp.exec('print("spam!")')

interp.exec("""if True:
    print('spam!')
    """)

from textwrap import dedent
interp.exec(dedent("""
    print('spam!')
    """))

def run(arg):
    return arg

res = interp.call(run, 'spam!')
print(res)

def run():
    print('spam!')

interp.call(run)

# Run in new OS thread.

t = interp.call_in_thread(run)
t.join()