subinterpreters high level module import time import xxsubinterpreters as interpreters import xxinterpchannels as channels aliases from xxsubinterpreters import isshareable from xxinterpchannels import channelerror channelnotfounderror channelclosederror channelemptyerror channelnotemptyerror all interpreter getcurrent getmain create listall runfailederror sendchannel recvchannel createchannel listallchannels isshareable channelerror channelnotfounderror channelemptyerror class runfailederrorruntimeerror def initself excinfo msg excinfo formatted if not msg if excinfo type and snapshot msg msg f snapshot type name snapshot msg else msg snapshot type name or snapshot msg super initmsg self snapshot excinfo def create isolatedtrue return all existing interpreters return interpreterid for id in interpreters listall def getcurrent return the main interpreter id interpreters getmain return interpreterid class interpreter xxx the lowlevel function has not been added yet see bpo return whether or not the identified interpreter is running return interpreters isrunningself id def closeself return interpreters destroyself id xxx rename run to exec def runself srcstr channelsnone excinfo interpreters execself id srcstr channels if excinfo is not none raise runfailederrorexcinfo def createchannel cid channels create recv send recvchannelcid sendchannelcid return recv send def listallchannels the base class for recvchannel and sendchannel end none def initself cid if self end send cid channels channelidcid sendtrue forcetrue elif self end recv cid channels channelidcid recvtrue forcetrue else raise notimplementederrorself end self id cid def reprself return f typeself nameidintself id def hashself return hashself id def eqself other if isinstanceself recvchannel if not isinstanceother recvchannel return notimplemented elif not isinstanceother sendchannel return notimplemented return other id self id property def idself return self id property def infoself return channels getinfoself id property def isclosedself return self info closed notset object class recvchannelchannelend return the next object from the channel this blocks until an object has been sent if none have been sent already return the next object from the channel if none have been sent then return the default if one is provided or fail with channelemptyerror otherwise this is the same as recv the sending end of a crossinterpreter channel end send property def isclosedself info self info return info closed or info closing def sendself obj timeoutnone channels sendself id obj timeouttimeout blockingtrue def sendnowaitself obj xxx note that at the moment channelsend only ever returns none this should be fixed when channelsendwait is added see bpo32604 and gh19829 return channels sendself id obj blockingfalse def sendbufferself obj timeoutnone channels sendbufferself id obj timeouttimeout blockingtrue def sendbuffernowaitself obj return channels sendbufferself id obj blockingfalse def closeself channels closeself id sendtrue xxx this is causing leaks gh110318 channels registerendtypessendchannel recvchannel aliases return a new idle python interpreter return all existing interpreters return the currently running interpreter return the main interpreter a single python interpreter xxx the low level function has not been added yet see bpo return whether or not the identified interpreter is running finalize and destroy the interpreter attempting to destroy the current interpreter results in a runtimeerror xxx rename run to exec run the given source code in the interpreter this is essentially the same as calling the builtin exec with this interpreter using the __dict__ of its __main__ module as both globals and locals there is no return value if the code raises an unhandled exception then a runfailederror is raised which summarizes the unhandled exception the actual exception is discarded because objects cannot be shared between interpreters this blocks the current python thread until done during that time the previous interpreter is allowed to run in other threads return recv send for a new cross interpreter channel the channel may be used to pass data safely between interpreters return a list of recv send for all open channels the base class for recvchannel and sendchannel the receiving end of a cross interpreter channel 10 milliseconds return the next object from the channel this blocks until an object has been sent if none have been sent already return the next object from the channel if none have been sent then return the default if one is provided or fail with channelemptyerror otherwise this is the same as recv the sending end of a cross interpreter channel send the object i e its data to the channel s receiving end this blocks until the object is received send the object to the channel s receiving end if the object is immediately received then return true else false otherwise this is the same as send xxx note that at the moment channel_send only ever returns none this should be fixed when channel_send_wait is added see bpo 32604 and gh 19829 send the object s buffer to the channel s receiving end this blocks until the object is received send the object s buffer to the channel s receiving end if the object is immediately received then return true else false otherwise this is the same as send xxx this is causing leaks gh 110318
import time
import _xxsubinterpreters as _interpreters
import _xxinterpchannels as _channels


from _xxsubinterpreters import is_shareable
from _xxinterpchannels import (
    ChannelError, ChannelNotFoundError, ChannelClosedError,
    ChannelEmptyError, ChannelNotEmptyError,
)


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


class RunFailedError(RuntimeError):

    def __init__(self, excinfo):
        msg = excinfo.formatted
        if not msg:
            if excinfo.type and snapshot.msg:
                msg = f'{snapshot.type.__name__}: {snapshot.msg}'
            else:
                msg = snapshot.type.__name__ or snapshot.msg
        super().__init__(msg)
        self.snapshot = excinfo


def create(*, isolated=True):
    
    id = _interpreters.create(isolated=isolated)
    return Interpreter(id, isolated=isolated)


def list_all():
    
    return [Interpreter(id) for id in _interpreters.list_all()]


def get_current():
    
    id = _interpreters.get_current()
    return Interpreter(id)


def get_main():
    
    id = _interpreters.get_main()
    return Interpreter(id)


class Interpreter:
    

    def __init__(self, id, *, isolated=None):
        if not isinstance(id, (int, _interpreters.InterpreterID)):
            raise TypeError(f'id must be an int, got {id!r}')
        self._id = id
        self._isolated = isolated

    def __repr__(self):
        data = dict(id=int(self._id), isolated=self._isolated)
        kwargs = (f'{k}={v!r}' for k, v in data.items())
        return f'{type(self).__name__}({", ".join(kwargs)})'

    def __hash__(self):
        return hash(self._id)

    def __eq__(self, other):
        if not isinstance(other, Interpreter):
            return NotImplemented
        else:
            return other._id == self._id

    @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 _interpreters.is_running(self._id)

    def close(self):
        
        return _interpreters.destroy(self._id)

    
    def run(self, src_str, /, channels=None):
        
        excinfo = _interpreters.exec(self._id, src_str, channels)
        if excinfo is not None:
            raise RunFailedError(excinfo)


def create_channel():
    
    cid = _channels.create()
    recv, send = RecvChannel(cid), SendChannel(cid)
    return recv, send


def list_all_channels():
    
    return [(RecvChannel(cid), SendChannel(cid))
            for cid in _channels.list_all()]


class _ChannelEnd:
    

    _end = None

    def __init__(self, cid):
        if self._end == 'send':
            cid = _channels._channel_id(cid, send=True, force=True)
        elif self._end == 'recv':
            cid = _channels._channel_id(cid, recv=True, force=True)
        else:
            raise NotImplementedError(self._end)
        self._id = cid

    def __repr__(self):
        return f'{type(self).__name__}(id={int(self._id)})'

    def __hash__(self):
        return hash(self._id)

    def __eq__(self, other):
        if isinstance(self, RecvChannel):
            if not isinstance(other, RecvChannel):
                return NotImplemented
        elif not isinstance(other, SendChannel):
            return NotImplemented
        return other._id == self._id

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

    @property
    def _info(self):
        return _channels.get_info(self._id)

    @property
    def is_closed(self):
        return self._info.closed


_NOT_SET = object()


class RecvChannel(_ChannelEnd):
    

    _end = 'recv'

    def recv(self, timeout=None, *,
             _sentinel=object(),
             _delay=10 / 1000,  
             ):
        
        if timeout is not None:
            timeout = int(timeout)
            if timeout < 0:
                raise ValueError(f'timeout value must be non-negative')
            end = time.time() + timeout
        obj = _channels.recv(self._id, _sentinel)
        while obj is _sentinel:
            time.sleep(_delay)
            if timeout is not None and time.time() >= end:
                raise TimeoutError
            obj = _channels.recv(self._id, _sentinel)
        return obj

    def recv_nowait(self, default=_NOT_SET):
        
        if default is _NOT_SET:
            return _channels.recv(self._id)
        else:
            return _channels.recv(self._id, default)

    def close(self):
        _channels.close(self._id, recv=True)


class SendChannel(_ChannelEnd):
    

    _end = 'send'

    @property
    def is_closed(self):
        info = self._info
        return info.closed or info.closing

    def send(self, obj, timeout=None):
        
        _channels.send(self._id, obj, timeout=timeout, blocking=True)

    def send_nowait(self, obj):
        
        
        
        
        return _channels.send(self._id, obj, blocking=False)

    def send_buffer(self, obj, timeout=None):
        
        _channels.send_buffer(self._id, obj, timeout=timeout, blocking=True)

    def send_buffer_nowait(self, obj):
        
        return _channels.send_buffer(self._id, obj, blocking=False)

    def close(self):
        _channels.close(self._id, send=True)



_channels._register_end_types(SendChannel, RecvChannel)