Improve concurrent.interpreters documentation

[FabriceSalvaire]

Documentation

We need a better documentation for https://docs.python.org/3/library/concurrent.interpreters.html

I tried this, but I don't understand

• how to use queue
• prepare_main purpose
• exec parameters like dedent
• I got funny crashes

####################################################################################################

from concurrent import interpreters as CI
from pprint import pprint
from textwrap import dedent
import queue

####################################################################################################

main_interpreter = CI.get_main()
print(f"Main is {main_interpreter}")

current_interpreter = CI.get_current()
print(f"Current is {current_interpreter}")

queue = CI.create_queue()

def task_runner(index):
    print("Task runner", index)
    print(f"  Current is {current_interpreter}")

    globals_ = globals()
    # pprint(globals_)
    print("  Globals:")
    for key, value in globals().items():
        if key == '__builtins__':
            print(f"    __builtins__: ...")
        else:
            print(f"    {key}: {value}")

    # Crash
    #   Main is Interpreter(0)
    #   Current is Interpreter(1)
    #   concurrent.interpreters.ExecutionFailed: NameError: name 'foo' is not defined
    #! print(foo)

    # Block
    #! print(f"  #{index} from queue {queue.get()}")

    return 'ok'

# if True:
if current_interpreter is main_interpreter:
    # queue = CI.create_queue()

    interpreters = []
    NUMBER_OF_INTERPRETERS = 4
    for i in range(NUMBER_OF_INTERPRETERS):
        _ = CI.create()
        # Bind objects in the interpreter’s __main__ module
        _.prepare_main(ns=None, foo=123)
        interpreters.append(_)

    for _ in CI.list_all():
        print(f"Interpreter id {_.id} running {_.is_running()} whence '{_.whence}'")

    print("Call...")

    print()
    # exec(code, /, dedent=True)
    interpreters[0].exec(dedent('''
    print(f"Exec hello {foo}")
    '''))

    print()
    rc = interpreters[0].call(task_runner, 0)
    print(rc)

    print()
    threads = []
    for i in range(NUMBER_OF_INTERPRETERS):
        thread = interpreters[i].call_in_thread(task_runner, i)
        print(f"Thread #{i} {thread}")
        threads.append(thread)

    NUMBER_OF_TASKS = 10
    for i in range(NUMBER_OF_TASKS):
        queue.put(i)

    for thread in threads:
        thread.join()

    # print()
    # print(queue)

    # Block main
    #! print(queue.get())
    # try:
    # except queue.ShutDown:
    #     pass
else:
    pass

[picnixz]

I think dedent is actually self-explanatory, that is, it should dedent the code you give so that you don't need to worry about indented script. The following would raise SyntaxError or something like that:

interpreter.exec("""\
    if 1:
        pass
""", dedent=False)

The API for queues is documented in queue.Queue (that's what the docs say). I don't think we should explain more. What matters is that, for subinterpreters, you cannot just use queue.Queue() to share your data. You need this special queue. Which objects could be shared is explained in https://docs.python.org/3/library/concurrent.interpreters.html#sharing-objects.

• prepare_main may indeed have a bit more explanations, like a "roughly equivalent to" example. I cannot say I know its purpose but AFAICT, it's essentially a way to populate the __main__ namespace with some objects (it's as if you were writing __main__.update(ns) for the __main__ module of the subinterpreter being considered).

• "funny crashes" isn't enough to understand your issues; please be more precise. We could have a FAQ for such issues though.

cc @ZeroIntensity @ericsnowcurrently

[FabriceSalvaire]

I guessed what is dedent, but I probably had an issue with my experimental code...

Funny crashes: For example, #! print(foo). It looks like the executed code is different due to this line.

Indeed, I managed to populate the globals, but it looks like it depends on my code. So I am confused...

But queue doesn't work on my code, so I missed something...

However, I converted this example in the doc and it runs 100% on all the cpu cores:

import concurrent.futures
import math

PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419]
for i in range(10):
    PRIMES += PRIMES

def is_prime(n):
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    # with concurrent.futures.ProcessPoolExecutor() as executor:
    #     for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
    #         print('%d is prime: %s' % (number, prime))

    print('Using InterpreterPoolExecutor')
    # concurrent.futures.InterpreterPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=())
    with concurrent.futures.InterpreterPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))

if __name__ == '__main__':
    main()

See code in /usr/lib64/python3.14/concurrent/future/interpreter.py

[FabriceSalvaire]

I annotated the code of concurrent/future/interpreter.py

####################################################################################################

from concurrent import interpreters
from concurrent.futures import thread as _thread
import math
import sys
import threading
import traceback

####################################################################################################

def do_call(results, func, args, kwargs):
    try:
        current_thread = threading.current_thread()
        current_interpreter = interpreters.get_current()
        print(f"{current_interpreter}/{current_thread}: Call {func} {args} {kwargs}")
        _ = func(*args, **kwargs)
        # print(f"{current_interpreter}:   -> {_}")
        return _
    except BaseException as exc:
        # Send the captured exception out on the results queue,
        # but still leave it unhandled for the interpreter to handle.
        try:
            results.put(exc)
        except interpreters.NotShareableError:
            # The exception is not shareable.
            print('exception is not shareable:', file=sys.stderr)
            traceback.print_exception(exc)
            results.put(None)
        raise  # re-raise

####################################################################################################

class WorkerContext(_thread.WorkerContext):

    @classmethod
    def prepare(cls, initializer, initargs):
        print(f"prepare_context {cls} {initializer} {initargs}")
        def resolve_task(fn, args, kwargs):
            if isinstance(fn, str):
                # XXX Circle back to this later.
                raise TypeError('scripts not supported')
            else:
                task = (fn, args, kwargs)
            return task

        if initializer is not None:
            try:
                initdata = resolve_task(initializer, initargs, {})
            except ValueError:
                if isinstance(initializer, str) and initargs:
                    raise ValueError(f'an initializer script does not take args, got {initargs!r}')
                raise  # re-raise
        else:
            initdata = None
        def create_context():
            return cls(initdata)
        return create_context, resolve_task

    def __init__(self, initdata):
        self.initdata = initdata
        self.interp = None
        self.results = None

    def __del__(self):
        if self.interp is not None:
            self.finalize()

    def initialize(self):
        assert self.interp is None, self.interp
        self.interp = interpreters.create()
        print(f"Created {self.interp}")
        try:
            maxsize = 0
            print(f"Created queue for {self.interp}")
            self.results = interpreters.create_queue(maxsize)

            if self.initdata:
                print(f"run {self.interp} {self.initdata}")
                self.run(self.initdata)
        except BaseException:
            self.finalize()
            raise  # re-raise

    def finalize(self):
        print(f"Finalize {self.interp}")
        interp = self.interp
        results = self.results
        self.results = None
        self.interp = None
        if results is not None:
            del results
        if interp is not None:
            interp.close()

    def run(self, task):
        current_thread = threading.current_thread()
        print(f"{self.interp}/{current_thread} run {task}")
        try:
            _ = self.interp.call(do_call, self.results, *task)
            print(f"{self.interp} returned {_}")
            return _
        except interpreters.ExecutionFailed as wrapper:
            # Wait for the exception data to show up.
            exc = self.results.get()
            if exc is None:
                # The exception must have been not shareable.
                raise  # re-raise
            raise exc from wrapper

####################################################################################################

class BrokenInterpreterPool(_thread.BrokenThreadPool):
    """
    Raised when a worker thread in an InterpreterPoolExecutor failed initializing.
    """

####################################################################################################

class InterpreterPoolExecutor(_thread.ThreadPoolExecutor):

    BROKEN = BrokenInterpreterPool

    @classmethod
    def prepare_context(cls, initializer, initargs):
        print(f"prepare_context {cls} {initializer} {initargs}")
        return WorkerContext.prepare(initializer, initargs)

    def __init__(self, max_workers=None, thread_name_prefix='', initializer=None, initargs=()):
        """Initializes a new InterpreterPoolExecutor instance.

        Args:
            max_workers: The maximum number of interpreters that can be used to
                execute the given calls.
            thread_name_prefix: An optional name prefix to give our threads.
            initializer: A callable or script used to initialize
                each worker interpreter.
            initargs: A tuple of arguments to pass to the initializer.
        """
        thread_name_prefix = (thread_name_prefix or
                              (f"InterpreterPoolExecutor-{self._counter()}"))
        super().__init__(max_workers, thread_name_prefix, initializer, initargs)

####################################################################################################

def is_prime(n):
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

# This is mandatory else this code will be executed by each sub-interpreter !!!
def main():
    PRIMES = [
        112272535095293,
        112582705942171,
        112272535095293,
        115280095190773,
        115797848077099,
        1099726899285419]
    # for i in range(2):
    #     PRIMES += PRIMES

    # concurrent.futures.InterpreterPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=())
    with InterpreterPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))

####################################################################################################

# This is mandatory !!!
if __name__ == '__main__':
    main()

output

prepare_context <class '__main__.InterpreterPoolExecutor'> None ()
prepare_context <class '__main__.WorkerContext'> None ()
Created Interpreter(3)
Created queue for Interpreter(3)
Interpreter(3)/<Thread(InterpreterPoolExecutor-0_2, started 139826249340608)> run (<function is_prime at 0x7f2bd72714e0>, (112272535095293,), {})
Created Interpreter(2)
Created queue for Interpreter(2)
Interpreter(2)/<Thread(InterpreterPoolExecutor-0_1, started 139826259809984)> run (<function is_prime at 0x7f2bd72714e0>, (112582705942171,), {})
Created Interpreter(4)
Created Interpreter(1)
Created queue for Interpreter(1)
Interpreter(1)/<Thread(InterpreterPoolExecutor-0_0, started 139826269517504)> run (<function is_prime at 0x7f2bd72714e0>, (112272535095293,), {})
Created Interpreter(5)
Created queue for Interpreter(5)
Interpreter(5)/<Thread(InterpreterPoolExecutor-0_4, started 139826230744768)> run (<function is_prime at 0x7f2bd72714e0>, (115280095190773,), {})
Created Interpreter(6)
Created queue for Interpreter(6)
Interpreter(6)/<Thread(InterpreterPoolExecutor-0_5, started 139825881544384)> run (<function is_prime at 0x7f2bd72714e0>, (115797848077099,), {})
Created queue for Interpreter(4)
Interpreter(4)/<Thread(InterpreterPoolExecutor-0_3, started 139826239137472)> run (<function is_prime at 0x7f2bd72714e0>, (1099726899285419,), {})
Interpreter(3)/<_DummyThread(Dummy-1, started 139826249340608)>: Call <function is_prime at 0x7f2bbe7a7e20> (112272535095293,) {}
Interpreter(5)/<_DummyThread(Dummy-1, started 139826230744768)>: Call <function is_prime at 0x7f2bbe6a7e20> (115280095190773,) {}
Interpreter(2)/<_DummyThread(Dummy-1, started 139826259809984)>: Call <function is_prime at 0x7f2bbe30fe20> (112582705942171,) {}
Interpreter(1)/<_DummyThread(Dummy-1, started 139826269517504)>: Call <function is_prime at 0x7f2bbe10fe20> (112272535095293,) {}
Interpreter(6)/<_DummyThread(Dummy-1, started 139825881544384)>: Call <function is_prime at 0x7f2bbe00be20> (115797848077099,) {}
Interpreter(4)/<_DummyThread(Dummy-1, started 139826239137472)>: Call <function is_prime at 0x7f2bbdf07e20> (1099726899285419,) {}
Interpreter(4) returned False
Interpreter(5) returned True
Interpreter(3) returned True
112272535095293 is prime: True
Interpreter(6) returned True
Interpreter(2) returned True
112582705942171 is prime: True
Interpreter(1) returned True
112272535095293 is prime: True
115280095190773 is prime: True
115797848077099 is prime: True
1099726899285419 is prime: False
Finalize Interpreter(4)
Finalize Interpreter(6)
Finalize Interpreter(2)
Finalize Interpreter(1)
Finalize Interpreter(3)
Finalize Interpreter(5)

[FabriceSalvaire]

I think we need a tutorial.
Things are always obvious to core dev.

Also, a graphic could help to quickly understand how it works (interpreters, threads).

[FabriceSalvaire]

So multicore is due to

Main C Python interpreter / thread - do a function call to another ->* C Python interpreter running in another thread
The code path is independent and can be scheduled in parallel to a CPU core.
It returns a value to the main interpreter.
Task scheduling is done by main interpreter.
This is more simple than a multi-processes spawning and communication.

[picnixz]

Things are always obvious to core dev.

Not really. I can mostly guess what prepare_main does because of how it's being phrased.

Funny crashes: For example, #! print(foo). It looks like the executed code is different due to this line.

This may be a legitimate bug. Do you have a self-contained example? please show us the traceback as well.

---

To be clear, I am not against improving the docs, but I would rather see a separate guide for subinterpreters rather than changing the current page (like we did for asyncio). We only have two core devs that are working actively on subinterpreters (those that I tagged) and we simply don't have enough people that can write this kind of guide so I guess this issue will likely stay open for some time. I will leave what to write and where to write to Eric.

[ZeroIntensity]

prepare_main just lets you inject variables into a subinterpreter's namespace. Feel free to send a PR clarifying things and tag me for a review. For a general subinterpreter tutorial, I think Eric is working on a concurrency HOWTO that deals with that.