它来了！真正的 python 多线程

# http://devguide.python.org/getting-started/setup-building/#unix-compiling
git clone http://github.com/python/cpython.git
cd cpython

./configure --enable-optimizations --prefix=$(pwd)/python-3.12
make -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.

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

from textwrap import dedent
import os
# http://github.com/python/cpython/blob/
#   15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test__xxsubinterpreters.py#L75
def _captured_script(script):
    r, w = os.pipe()
    indented = script.replace('\n', '\n                ')
    wrapped = dedent(f"""
        import contextlib
        with 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()

from test.support import interpreters

main = 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)

# http://github.com/python/cpython/blob/
#   15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test__xxsubinterpreters.py#L236
code = dedent("""
            from test.support import interpreters
            cur = 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

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

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

def cleanup_interpreters():
    for i in interpreters.list_all():
        if i.id == 0:  # main
            continue
        try:
            print(f"Cleaning up interpreter: {i}")
            i.close()
        except RuntimeError:
            pass  # already destroyed

cleanup_interpreters()
# Cleaning up interpreter: Interpreter(id=1, isolated=None)
# Cleaning up interpreter: Interpreter(id=2, isolated=None)

import threading

def 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()

# <Thread(Thread-1 (create), initial)>
# <Thread(Thread-1 (create), started 139772371633728)>
# <Thread(Thread-1 (create), stopped 139772371633728)>
# <Thread(Thread-2 (create), initial)>
# <Thread(Thread-2 (create), started 139772371633728)>
# <Thread(Thread-2 (create), stopped 139772371633728)>

import time

def run_in_thread():
    interp = interpreters.create(isolated=True)
    t = threading.Thread(target=_run_output, args=(interp, dedent("""
            import _xxsubinterpreters as _interpreters
            cur = _interpreters.get_current()

            import time
            time.sleep(2)
            # Can't print from here, won't bubble-up to main interpreter

            assert isinstance(cur, _interpreters.InterpreterID)
            """)))
    print(f"Created Thread: {t}")
    t.start()
    return t


t1 = 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 complete
cleanup_interpreters()

# http://github.com/python/cpython/blob/
#   15665d896bae9c3d8b60bd7210ac1b7dc533b093/Lib/test/test_interpreters.py#L583
r, 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)

import test.support

def 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 interpreters
            cur = interpreters.get_current()
            print(cur)
            """)

run_in_thread(code)
# Interpreter(id=7, isolated=None)
run_in_thread(code)
# Interpreter(id=8, isolated=None)