ToB企服应用市场:ToB评测及商务社交产业平台

标题: 它来了！真正的 python 多线程 [打印本页]
作者: 石小疯    时间: 2023-5-25 15:26
标题: 它来了！真正的 python 多线程
哈喽大家好，我是咸鱼
几天前，IBM 工程师 Martin Heinz 发文表示 python 3.12 版本回引入"Per-Interpreter GIL”，有了这个 Per-Interpreter 全局解释器锁，python 就能实现真正意义上的并行/并发
我们知道，python 的多线程/进程并不是真正意义上的多线程/进程，这是因为 python GIL （Global Interpreter Lock）导致的
而即将发布的 Python 3.12 中引入了名为 "Per-Interpreter GIL" 的新特性，能够实现真正的并发
接下来我们来看下这篇文章，原文链接如下：
https://martinheinz.dev/blog/97
译文

Python 到现在已经 32 岁了，但它到现在还没有实现适当的、真正的并发/并行
由于将在 Python 3.12 （预计 2023 年 10 月发布）中引入 “Per-Interpreter GIL”(全局解释器锁)，这种情况将会被改变
虽然距离 python 3.12 的发布还有几个月的时间，但是相关代码已经实现了。所以让我们提前来了解一下如何使用子解释器 API（ub-interpreters API） 来编写出真正的并发Python代码
子解释器（Sub-Interpreters）

我们首先来看下这个 “Per-Interpreter GIL” 是如何解决 Python 缺失适当并发性这个问题的
简单来讲，GIL(全局解释器锁)是一个互斥锁，它只允许一个线程控制 Python 解释器（某个线程想要执行，必须要先拿到 GIL ，在一个 python 解释器里面，GIL 只有一个，拿不到 GIL 的就不允许执行）
这就意味着即使你在 Python 中创建多个线程，也只会有一个线程在运行
随着 “Per-Interpreter GIL” 的引用，单个 python 解释器不再共享同一个 GIL。这种隔离级别允许每个子 python 解释器真正地并发运行
这意味着我们可以通过生成额外的子解释器来绕过 Python 的并发限制，其中每个子解释器都有自己的GIL（拿到一个 GIL 锁）
更详细的说明请参见 PEP 684，该文档描述了此功能/更改：https://peps.python.org/pep-0684/#per-interpreter-state
如何安装

想要使用这个新功能，我们需要安装最新的 python 版本，这需要源码编译安装
  1. # https://devguide.python.org/getting-started/setup-building/#unix-compiling
  2. git clone https://github.com/python/cpython.git
  3. cd cpython
  5. ./configure --enable-optimizations --prefix=$(pwd)/python-3.12
  6. make -s -j2
  7. ./python
  8. # Python 3.12.0a7+ (heads/main:22f3425c3d, May 10 2023, 12:52:07) [GCC 11.3.0] on linux
  9. # Type "help", "copyright", "credits" or "license" for more information.
复制代码
C-API 在哪里

现在我们已经安装好了最新版本，那么我们该如何使用子解释器呢？我们可以直接通过 import 来导入吗？不幸的是，还不能
正如 PEP-684 中指出的:  ...this is an advanced feature meant for a narrow set of users of the C-API.
Per-Interpreter GIL 的特性目前只能通过 C-API 使用，还没有直接的接口供开发人员使用
接口预计会在 PEP 554中出现，如果大家能够接受，它应该会在 Python 3.13 中出现，在这个版本出现之前，我们必须自己想办法来实现子解释器
虽然还没有相关文档，也没有相关模块可以导入，但 CPython 代码库中有一些代码段向我们展示了如何使用它：
  1. # Choose one of these:
  2. import _xxsubinterpreters as interpreters
  3. from test.support import interpreters
复制代码
通常情况下我们一般用上面的第二种方法来实现
我们已经找到了子解释器，但我们还需要通过 test 模块去借用一些辅助函数，以便将代码传递给子解释器，辅助函数如下
  1. from textwrap import dedent
  2. import os
  3. # https://github.com/python/cpython/blob/
  4. #   15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test__xxsubinterpreters.py#L75
  5. def _captured_script(script):
  6.     r, w = os.pipe()
  7.     indented = script.replace('\n', '\n                ')
  8.     wrapped = dedent(f"""
  9.         import contextlib
  10.         with open({w}, 'w', encoding="utf-8") as spipe:
  11.             with contextlib.redirect_stdout(spipe):
  12.                 {indented}
  13.         """)
  14.     return wrapped, open(r, encoding="utf-8")
  17. def _run_output(interp, request, channels=None):
  18.     script, rpipe = _captured_script(request)
  19.     with rpipe:
  20.         interp.run(script, channels=channels)
  21.         return rpipe.read()
复制代码
将 interpreters 模块与上面的辅助函数组合在一起，便可以生成第一个子解释器：
  1. from test.support import interpreters
  3. main = interpreters.get_main()
  4. print(f"Main interpreter ID: {main}")
  5. # Main interpreter ID: Interpreter(id=0, isolated=None)
  7. interp = interpreters.create()
  9. print(f"Sub-interpreter: {interp}")
  10. # Sub-interpreter: Interpreter(id=1, isolated=True)
  12. # https://github.com/python/cpython/blob/
  13. #   15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test__xxsubinterpreters.py#L236
  14. code = dedent("""
  15.             from test.support import interpreters
  16.             cur = interpreters.get_current()
  17.             print(cur.id)
  18.             """)
  20. out = _run_output(interp, code)
  22. print(f"All Interpreters: {interpreters.list_all()}")
  23. # All Interpreters: [Interpreter(id=0, isolated=None), Interpreter(id=1, isolated=None)]
  24. print(f"Output: {out}")  # Result of 'print(cur.id)'
  25. # Output: 1
复制代码
生成和运行新解释器的一种方法是使用 create() 函数，然后将解释器与我们想要执行的代码一起传递给 _run_output() 辅助函数
还有一种更简单的方法，如下所示
  1. interp = interpreters.create()
  2. interp.run(code)
复制代码
直接使用 interpreters  模块的 run 方法。
但如果我们运行上面这两段代码时，会收到以下报错
  1. Fatal Python error: PyInterpreterState_Delete: remaining subinterpreters
  2. Python runtime state: finalizing (tstate=0x000055b5926bf398)
复制代码
为了避免这个报错，我们还需要清理一些悬挂的解释器：
  1. def cleanup_interpreters():
  2.     for i in interpreters.list_all():
  3.         if i.id == 0:  # main
  4.             continue
  5.         try:
  6.             print(f"Cleaning up interpreter: {i}")
  7.             i.close()
  8.         except RuntimeError:
  9.             pass  # already destroyed
  11. cleanup_interpreters()
  12. # Cleaning up interpreter: Interpreter(id=1, isolated=None)
  13. # Cleaning up interpreter: Interpreter(id=2, isolated=None)
复制代码
线程

虽然使用上面的辅助函数运行代码是可行的，但在 threading 模块中使用熟悉的接口可能会更方便
  1. import threading
  3. def run_in_thread():
  4.     t = threading.Thread(target=interpreters.create)
  5.     print(t)
  6.     t.start()
  7.     print(t)
  8.     t.join()
  9.     print(t)
  11. run_in_thread()
  12. run_in_thread()
  14. # <Thread(Thread-1 (create), initial)>
  15. # <Thread(Thread-1 (create), started 139772371633728)>
  16. # <Thread(Thread-1 (create), stopped 139772371633728)>
  17. # <Thread(Thread-2 (create), initial)>
  18. # <Thread(Thread-2 (create), started 139772371633728)>
  19. # <Thread(Thread-2 (create), stopped 139772371633728)>
复制代码
我们通过把 interpreters.create 函数传递给Thread，它会自动在线程内部生成新的子解释器
我们也可以结合这两种方法，并将辅助函数传递给 threading.Thread：
  1. import time
  3. def run_in_thread():
  4.     interp = interpreters.create(isolated=True)
  5.     t = threading.Thread(target=_run_output, args=(interp, dedent("""
  6.             import _xxsubinterpreters as _interpreters
  7.             cur = _interpreters.get_current()
  9.             import time
  10.             time.sleep(2)
  11.             # Can't print from here, won't bubble-up to main interpreter
  13.             assert isinstance(cur, _interpreters.InterpreterID)
  14.             """)))
  15.     print(f"Created Thread: {t}")
  16.     t.start()
  17.     return t
  20. t1 = run_in_thread()
  21. print(f"First running Thread: {t1}")
  22. t2 = run_in_thread()
  23. print(f"Second running Thread: {t2}")
  24. time.sleep(4)  # Need to sleep to give Threads time to complete
  25. cleanup_interpreters()
复制代码
上面的代码中演示了如何使用 _xxsubinterpreters 模块来实现 （方法一）
我们还在每个线程中休眠 2 秒来模拟“工作”状态
请注意，我们甚至不必调用 join() 函数等待线程完成，只需在线程完成时清理解释器即可
Channels

如果我们进一步挖掘 CPython  test 模块，我们还会发现 RecvChannel 和 SendChannel 类的实现类似于 Golang 中已知的通道
  1. # https://github.com/python/cpython/blob/
  2. #   15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test_interpreters.py#L583
  3. r, s = interpreters.create_channel()
  5. print(f"Channel: {r}, {s}")
  6. # Channel: RecvChannel(id=0), SendChannel(id=0)
  8. orig = b'spam'
  9. s.send_nowait(orig)
  10. obj = r.recv()
  11. print(f"Received: {obj}")
  12. # Received: b'spam'
  14. cleanup_interpreters()
  15. # Need clean up, otherwise:
  17. # free(): invalid pointer
  18. # Aborted (core dumped)
复制代码
上面的例子介绍了如何创建一个接收端通道（r）和发送端通道（s），然后我们使用 send_nowait 方法将数据发送，通过 recv 方法来接收数据
这个通道实际上只是另一个解释器，和以前一样，我们需要在处理完它之后进行清理
Digging Deeper

如果我们想要修改或者调整子解释器的选项（这些选项通常在 C 代码中设置），我们可以使用
test.support 模块中的代码，具体来说是run_in_subinterp_with_config
  1. import test.support
  3. def run_in_thread(script):
  4.     test.support.run_in_subinterp_with_config(
  5.         script,
  6.         use_main_obmalloc=True,
  7.         allow_fork=True,
  8.         allow_exec=True,
  9.         allow_threads=True,
  10.         allow_daemon_threads=False,
  11.         check_multi_interp_extensions=False,
  12.         own_gil=True,
  13.     )
  15. code = dedent(f"""
  16.             from test.support import interpreters
  17.             cur = interpreters.get_current()
  18.             print(cur)
  19.             """)
  21. run_in_thread(code)
  22. # Interpreter(id=7, isolated=None)
  23. run_in_thread(code)
  24. # Interpreter(id=8, isolated=None)
复制代码
上面这个run_in_subinterp_with_config函数是 C 函数的 Python API。它提供了一些子解释器选项，如 own_gil，指定子解释器是否应该拥有自己的 GIL

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4