真正的Python多執行緒來了！

【CSDN 編者按】IBM工程師Martin Heinz發文表示，Python即將迎來了真正的多執行緒時刻！

原文：https://martinheinz.dev/blog/97

未經授權，禁止轉載！

作者 | Martin Heinz 責編 | 夢依丹

翻譯工具 | ChatGPT

32歲的Python依然沒有真正的並行性/併發性。然而，這種情況即將發生改變，因為在即將發佈的Python 3.12中引入了名為”Per-Interpreter GIL”（全局直譯器鎖）的新特性。在距離發佈還有幾個月的時間（預計2023年10月發佈），但相關程式碼已經有了，因此，我們可以提前了解如何使用子直譯器API編寫真正的併發Python程式碼。

子直譯器

首先，讓我們來解釋一下如何通過「Per-Interpreter GIL」來解決Python缺乏適當的併發性問題。

在Python中，GIL是一個互斥鎖，它只允許一個執行緒控制Python直譯器。這意味著即使在Python中創建多個執行緒（例如使用執行緒模組），也只有一個執行緒會運行。

隨著「Per-Interpreter GIL」的引入，各個Python直譯器不再共享同一個GIL。這種隔離級別允許每個子直譯器可以同時運行。這意味著我們可以通過生成額外的子直譯器來繞過Python的併發限制，其中每個子直譯器都有自己的GIL（全局狀態）。

更詳細的說明請參見PEP 684，該文件描述了此功能/更改：https://peps.python.org/pep-0684/#per-interpreter-state

上手體驗

安裝

為使用這項最新功能，我們必須安裝最新版的Python，並且需要從原始碼上進行構建：

# https://devguide.python.org/getting-started/setup-building/#unix-compilinggit clone https://github.com/python/cpython.gitcd cpython./configure --enable-optimizations --prefix=$(pwd)/python-3.12make -s -j2./python# Python 3.12.0a7+ (heads/main:22f3425c3d, May 10 2023, 12:52:07) [GCC 11.3.0] on linux# Type "help", "copyright", "credits" or "license" for more information.

C-API在哪裡？

既然已經安裝了最新的版本，那我們該如何使用子直譯器呢？可以直接匯入嗎？不，正如PEP-684中所提到的：「這是一個高級功能，專為C-API的一小部分使用者設計。」

目前，Per-Interpreter GIL特性只能通過C-API使用，因此Python開發人員沒有直接的接口可以使用。這樣的接口預計將隨著PEP 554一起推出，如果被採納，則應該會在Python 3.13中實現，在那之前，我們必須自己想辦法實現子直譯器。

通過CPython程式碼庫中的一些零散記錄，我們可以採用下面兩種方法：

使用_xxsubinterpreters模組，該模組是用C實現的，因此名稱看起來有些奇怪。由於它是用C實現的，所以開發者無法輕易地檢查程式碼（至少不是在 Python 中）；

或者可以利用CPython的測試模組，該模組具有用於測試的示例 Interpreter（和 Channel）類。

# Choose one of these:import _xxsubinterpreters as interpretersfrom test.support import interpreters

在接下來的演示中，我們將主要採用第二種方法。我們已經找到了子直譯器，但還需要從Python的測試模組中借用一些輔助函數，以便將程式碼傳遞給子直譯器：

from textwrap import dedentimport os# https://github.com/python/cpython/blob/#   15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test__xxsubinterpreters.py#L75def _captured_script(script):r, w = os.pipe()indented = script.replace('\n', '\n                ')wrapped = dedent(f"""import contextlibwith open({w}, 'w', encoding="utf-8") as spipe:with contextlib.redirect_stdout(spipe):{indented}""")return wrapped, open(r, encoding="utf-8")def _run_output(interp, request, channels=None):script, rpipe = _captured_script(request)with rpipe:interp.run(script, channels=channels)return rpipe.read()

將interpreters模組與上述的輔助程序組合在一起，便可以生成第一個子直譯器：

from test.support import interpretersmain = interpreters.get_main()print(f"Main interpreter ID: {main}")# Main interpreter ID: Interpreter(id=0, isolated=None)interp = interpreters.create()print(f"Sub-interpreter: {interp}")# Sub-interpreter: Interpreter(id=1, isolated=True)# https://github.com/python/cpython/blob/#   15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test__xxsubinterpreters.py#L236code = dedent("""from test.support import interpreterscur = interpreters.get_current()print(cur.id)""")out = _run_output(interp, code)print(f"All Interpreters: {interpreters.list_all()}")# All Interpreters: [Interpreter(id=0, isolated=None), Interpreter(id=1, isolated=None)]print(f"Output: {out}")  # Result of 'print(cur.id)'# Output: 1

生成和運行新直譯器的一個方法是使用create函數，然後將直譯器與要執行的程式碼一起傳遞給_run_output輔助函數。

更簡單點的方法是：

interp = interpreters.create()interp.run(code)

使用直譯器中的run方法。可是，如果我們運行上述程式碼中的任意一個，都會得到如下錯誤：

Fatal Python error: PyInterpreterState_Delete: remaining subinterpretersPython runtime state: finalizing (tstate=0x000055b5926bf398)

為避免此類錯誤發生，還需要清理一些懸掛的直譯器：

def cleanup_interpreters():for i in interpreters.list_all():if i.id == 0:  # maincontinuetry:print(f"Cleaning up interpreter: {i}")i.close()except RuntimeError:pass  # already destroyedcleanup_interpreters()# Cleaning up interpreter: Interpreter(id=1, isolated=None)# Cleaning up interpreter: Interpreter(id=2, isolated=None)

執行緒

雖然使用上述輔助函數運行程式碼是可行的，但使用執行緒模組中熟悉的接口可能更加方便：

import threadingdef run_in_thread():t = threading.Thread(target=interpreters.create)print(t)t.start()print(t)t.join()print(t)run_in_thread()run_in_thread()# # # # # #

我們通過把interpreters.create函數傳遞給Thread,它會自動線上程內部生成新的子直譯器。

我們也可以結合這兩種方法，並將輔助函數傳遞給threading.Thread：

import timedef run_in_thread():interp = interpreters.create(isolated=True)t = threading.Thread(target=_run_output, args=(interp, dedent("""import _xxsubinterpreters as _interpreterscur = _interpreters.get_current()import timetime.sleep(2)# Can't print from here, won't bubble-up to main interpreterassert isinstance(cur, _interpreters.InterpreterID)""")))print(f"Created Thread: {t}")t.start()return tt1 = run_in_thread()print(f"First running Thread: {t1}")t2 = run_in_thread()print(f"Second running Thread: {t2}")time.sleep(4)  # Need to sleep to give Threads time to completecleanup_interpreters()

這裡，我們演示瞭如何使用_xxsubinterpreters模組而不是test.support中的模組。我們還在每個執行緒中睡眠2秒鐘來模擬一些「工作」。請注意，我們甚至不必調用join()函數等待執行緒完成，只需線上程完成時清理直譯器即可。

Channels

如果我們深入研究CPython測試模組，我們還會發現有 RecvChannel 和 SendChannel 類的實現，它們類似於 Golang 中的通道（Channel）。要使用它們：

# https://github.com/python/cpython/blob/#   15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test_interpreters.py#L583r, s = interpreters.create_channel()print(f"Channel: {r}, {s}")# Channel: RecvChannel(id=0), SendChannel(id=0)orig = b'spam's.send_nowait(orig)obj = r.recv()print(f"Received: {obj}")# Received: b'spam'cleanup_interpreters()# Need clean up, otherwise:# free(): invalid pointer# Aborted (core dumped)

這個例子展示瞭如何創建一個帶有receiver（r）和sender（s）端的通道。我們可以使用send_nowait將資料傳遞給發送方，並使用recv函數在另一側讀取它。這個通道實際上只是另一個子直譯器 – 所以與之前一樣 – 我們需要在完成後進行清理。

深入挖掘

最後，如果我們想要干擾或調整在C程式碼中設置的子直譯器選項，那麼可以使用test.support模組中的程式碼，具體來說就是run_in_subinterp_with_config：

import test.supportdef run_in_thread(script):test.support.run_in_subinterp_with_config(script,use_main_obmalloc=True,allow_fork=True,allow_exec=True,allow_threads=True,allow_daemon_threads=False,check_multi_interp_extensions=False,own_gil=True,)code = dedent(f"""from test.support import interpreterscur = interpreters.get_current()print(cur)""")run_in_thread(code)# Interpreter(id=7, isolated=None)run_in_thread(code)# Interpreter(id=8, isolated=None)