Skip to content

Improve concurrent.interpreters documentation #142041

Description

@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

Linked PRs

Activity

  1. picnixz commented on Nov 28, 2025

    @picnixz
    Member

    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

  2. FabriceSalvaire commented on Nov 28, 2025

    @FabriceSalvaire
    Author

    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

  3. FabriceSalvaire commented on Nov 28, 2025

    @FabriceSalvaire
    Author

    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)
    
  4. FabriceSalvaire commented on Nov 28, 2025

    @FabriceSalvaire
    Author

    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).

  5. FabriceSalvaire commented on Nov 28, 2025

    @FabriceSalvaire
    Author

    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.

  6. picnixz commented on Nov 28, 2025

    @picnixz
    Member

    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.

  7. ZeroIntensity commented on Nov 28, 2025

    @ZeroIntensity
    Member

    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.

  8. added a commit that references this issue on Mar 9, 2026
  9. mvanhorn commented on Mar 9, 2026

    @mvanhorn
    Contributor

    I've submitted a PR to improve the documentation: #145657

    It addresses the specific pain points mentioned in this issue:

    • Queue methods are now fully documented with parameter descriptions
    • prepare_main has a clear explanation and usage example
    • The incorrect dedent parameter on exec has been removed (it doesn't exist in the implementation)
    • The Basic usage section has been rewritten with separate, focused examples for common patterns (exec/call, prepare_main, queues, parallel execution)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions