cpython/Lib/test/support/interpreters.py

"""Subinterpreters High Level Module."""

import _xxsubinterpreters as _interpreters

# aliases:
from _xxsubinterpreters import (
    ChannelError, ChannelNotFoundError, ChannelEmptyError,
    is_shareable,
)


__all__ = [
    'Interpreter', 'get_current', 'get_main', 'create', 'list_all',
    'SendChannel', 'RecvChannel',
    'create_channel', 'list_all_channels', 'is_shareable',
    'ChannelError', 'ChannelNotFoundError',
    'ChannelEmptyError',
    ]


def create(*, isolated=True):
    """
    Initialize a new (idle) Python interpreter.
    """
    id = _interpreters.create(isolated=isolated)
    return Interpreter(id, isolated=isolated)


def list_all():
    """
    Get all existing interpreters.
    """
    return [Interpreter(id) for id in
            _interpreters.list_all()]


def get_current():
    """
    Get the currently running interpreter.
    """
    id = _interpreters.get_current()
    return Interpreter(id)


def get_main():
    """
    Get the main interpreter.
    """
    id = _interpreters.get_main()
    return Interpreter(id)


class Interpreter:
    """
    The Interpreter object represents
    a single interpreter.
    """

    def __init__(self, id, *, isolated=None):
        self._id = id
        self._isolated = isolated

    @property
    def id(self):
        return self._id

    @property
    def isolated(self):
        if self._isolated is None:
            self._isolated = _interpreters.is_isolated(self._id)
        return self._isolated

    def is_running(self):
        """
        Return whether or not the identified
        interpreter is running.
        """
        return _interpreters.is_running(self._id)

    def close(self):
        """
        Finalize and destroy the interpreter.

        Attempting to destroy the current
        interpreter results in a RuntimeError.
        """
        return _interpreters.destroy(self._id)

    def run(self, src_str, /, *, channels=None):
        """
        Run the given source code in the interpreter.
        This blocks the current Python thread until done.
        """
        _interpreters.run_string(self._id, src_str)


def create_channel():
    """
    Create a new channel for passing data between
    interpreters.
    """

    cid = _interpreters.channel_create()
    return (RecvChannel(cid), SendChannel(cid))


def list_all_channels():
    """
    Get all open channels.
    """
    return [(RecvChannel(cid), SendChannel(cid))
            for cid in _interpreters.channel_list_all()]


_NOT_SET = object()


class RecvChannel:
    """
    The RecvChannel object represents
    a recieving channel.
    """

    def __init__(self, id):
        self._id = id

    def recv(self, *, _delay=10 / 1000):  # 10 milliseconds
        """
        Get the next object from the channel,
        and wait if none have been sent.
        Associate the interpreter with the channel.
        """
        import time
        sentinel = object()
        obj = _interpreters.channel_recv(self._id, sentinel)
        while obj is sentinel:
            time.sleep(_delay)
            obj = _interpreters.channel_recv(self._id, sentinel)
        return obj

    def recv_nowait(self, default=_NOT_SET):
        """
        Like recv(), but return the default
        instead of waiting.

        This function is blocked by a missing low-level
        implementation of channel_recv_wait().
        """
        if default is _NOT_SET:
            return _interpreters.channel_recv(self._id)
        else:
            return _interpreters.channel_recv(self._id, default)


class SendChannel:
    """
    The SendChannel object represents
    a sending channel.
    """

    def __init__(self, id):
        self._id = id

    def send(self, obj):
        """
        Send the object (i.e. its data) to the receiving
        end of the channel and wait. Associate the interpreter
        with the channel.
        """
        import time
        _interpreters.channel_send(self._id, obj)
        time.sleep(2)

    def send_nowait(self, obj):
        """
        Like send(), but return False if not received.

        This function is blocked by a missing low-level
        implementation of channel_send_wait().
        """

        _interpreters.channel_send(self._id, obj)
        return False