"""Base execnet gateway code send to the other side for bootstrapping. :copyright: 2004-2015 :authors: - Holger Krekel - Armin Rigo - Benjamin Peterson - Ronny Pfannschmidt - many others """ from __future__ import annotations import abc import os import struct import sys import traceback import weakref from _thread import interrupt_main from io import BytesIO from typing import Any from typing import Callable from typing import Iterator from typing import Literal from typing import MutableSet from typing import Protocol from typing import cast from typing import overload class WriteIO(Protocol): def write(self, data: bytes, /) -> None: ... class ReadIO(Protocol): def read(self, numbytes: int, /) -> bytes: ... class IO(Protocol): execmodel: ExecModel def read(self, numbytes: int, /) -> bytes: ... def write(self, data: bytes, /) -> None: ... def close_read(self) -> None: ... def close_write(self) -> None: ... def wait(self) -> int | None: ... def kill(self) -> None: ... class Event(Protocol): """Protocol for types which look like threading.Event.""" def is_set(self) -> bool: ... def set(self) -> None: ... def clear(self) -> None: ... def wait(self, timeout: float | None = None) -> bool: ... class ExecModel(metaclass=abc.ABCMeta): @property @abc.abstractmethod def backend(self) -> str: raise NotImplementedError() def __repr__(self) -> str: return "" % self.backend @property @abc.abstractmethod def queue(self): raise NotImplementedError() @property @abc.abstractmethod def subprocess(self): raise NotImplementedError() @property @abc.abstractmethod def socket(self): raise NotImplementedError() @abc.abstractmethod def start(self, func, args=()) -> None: raise NotImplementedError() @abc.abstractmethod def get_ident(self) -> int: raise NotImplementedError() @abc.abstractmethod def sleep(self, delay: float) -> None: raise NotImplementedError() @abc.abstractmethod def fdopen(self, fd, mode, bufsize=1, closefd=True): raise NotImplementedError() @abc.abstractmethod def Lock(self): raise NotImplementedError() @abc.abstractmethod def RLock(self): raise NotImplementedError() @abc.abstractmethod def Event(self) -> Event: raise NotImplementedError() class ThreadExecModel(ExecModel): backend = "thread" @property def queue(self): import queue return queue @property def subprocess(self): import subprocess return subprocess @property def socket(self): import socket return socket def get_ident(self) -> int: import _thread return _thread.get_ident() def sleep(self, delay: float) -> None: import time time.sleep(delay) def start(self, func, args=()) -> None: import _thread _thread.start_new_thread(func, args) def fdopen(self, fd, mode, bufsize=1, closefd=True): import os return os.fdopen(fd, mode, bufsize, encoding="utf-8", closefd=closefd) def Lock(self): import threading return threading.RLock() def RLock(self): import threading return threading.RLock() def Event(self): import threading return threading.Event() class MainThreadOnlyExecModel(ThreadExecModel): backend = "main_thread_only" class EventletExecModel(ExecModel): backend = "eventlet" @property def queue(self): import eventlet return eventlet.queue @property def subprocess(self): import eventlet.green.subprocess return eventlet.green.subprocess @property def socket(self): import eventlet.green.socket return eventlet.green.socket def get_ident(self) -> int: import eventlet.green.thread return eventlet.green.thread.get_ident() # type: ignore[no-any-return] def sleep(self, delay: float) -> None: import eventlet eventlet.sleep(delay) def start(self, func, args=()) -> None: import eventlet eventlet.spawn_n(func, *args) def fdopen(self, fd, mode, bufsize=1, closefd=True): import eventlet.green.os return eventlet.green.os.fdopen(fd, mode, bufsize, closefd=closefd) def Lock(self): import eventlet.green.threading return eventlet.green.threading.RLock() def RLock(self): import eventlet.green.threading return eventlet.green.threading.RLock() def Event(self): import eventlet.green.threading return eventlet.green.threading.Event() class GeventExecModel(ExecModel): backend = "gevent" @property def queue(self): import gevent.queue return gevent.queue @property def subprocess(self): import gevent.subprocess return gevent.subprocess @property def socket(self): import gevent return gevent.socket def get_ident(self) -> int: import gevent.thread return gevent.thread.get_ident() # type: ignore[no-any-return] def sleep(self, delay: float) -> None: import gevent gevent.sleep(delay) def start(self, func, args=()) -> None: import gevent gevent.spawn(func, *args) def fdopen(self, fd, mode, bufsize=1, closefd=True): # XXX import gevent.fileobject return gevent.fileobject.FileObjectThread(fd, mode, bufsize, closefd=closefd) def Lock(self): import gevent.lock return gevent.lock.RLock() def RLock(self): import gevent.lock return gevent.lock.RLock() def Event(self): import gevent.event return gevent.event.Event() def get_execmodel(backend: str | ExecModel) -> ExecModel: if isinstance(backend, ExecModel): return backend if backend == "thread": return ThreadExecModel() elif backend == "main_thread_only": return MainThreadOnlyExecModel() elif backend == "eventlet": return EventletExecModel() elif backend == "gevent": return GeventExecModel() else: raise ValueError(f"unknown execmodel {backend!r}") class Reply: """Provide access to the result of a function execution that got dispatched through WorkerPool.spawn().""" def __init__(self, task, threadmodel: ExecModel) -> None: self.task = task self._result_ready = threadmodel.Event() self.running = True def get(self, timeout: float | None = None): """get the result object from an asynchronous function execution. if the function execution raised an exception, then calling get() will reraise that exception including its traceback. """ self.waitfinish(timeout) try: return self._result except AttributeError: raise self._exc from None def waitfinish(self, timeout: float | None = None) -> None: if not self._result_ready.wait(timeout): raise OSError(f"timeout waiting for {self.task!r}") def run(self) -> None: func, args, kwargs = self.task try: try: self._result = func(*args, **kwargs) except BaseException as exc: self._exc = exc finally: self._result_ready.set() self.running = False class WorkerPool: """A WorkerPool allows to spawn function executions to threads, returning a reply object on which you can ask for the result (and get exceptions reraised). This implementation allows the main thread to integrate itself into performing function execution through calling integrate_as_primary_thread() which will return when the pool received a trigger_shutdown(). By default allows unlimited number of spawns. """ _primary_thread_task: Reply | None def __init__(self, execmodel: ExecModel, hasprimary: bool = False) -> None: self.execmodel = execmodel self._running_lock = self.execmodel.Lock() self._running: MutableSet[Reply] = set() self._shuttingdown = False self._waitall_events: list[Event] = [] if hasprimary: if self.execmodel.backend not in ("thread", "main_thread_only"): raise ValueError("hasprimary=True requires thread model") self._primary_thread_task_ready: Event | None = self.execmodel.Event() else: self._primary_thread_task_ready = None def integrate_as_primary_thread(self) -> None: """Integrate the thread with which we are called as a primary thread for executing functions triggered with spawn().""" assert self.execmodel.backend in ("thread", "main_thread_only"), self.execmodel primary_thread_task_ready = self._primary_thread_task_ready assert primary_thread_task_ready is not None # interacts with code at REF1 while 1: primary_thread_task_ready.wait() reply = self._primary_thread_task if reply is None: # trigger_shutdown() woke us up break self._perform_spawn(reply) # we are concurrent with trigger_shutdown and spawn with self._running_lock: if self._shuttingdown: break # Only clear if _try_send_to_primary_thread has not # yet set the next self._primary_thread_task reply # after waiting for this one to complete. if reply is self._primary_thread_task: primary_thread_task_ready.clear() def trigger_shutdown(self) -> None: with self._running_lock: self._shuttingdown = True if self._primary_thread_task_ready is not None: self._primary_thread_task = None self._primary_thread_task_ready.set() def active_count(self) -> int: return len(self._running) def _perform_spawn(self, reply: Reply) -> None: reply.run() with self._running_lock: self._running.remove(reply) if not self._running: while self._waitall_events: waitall_event = self._waitall_events.pop() waitall_event.set() def _try_send_to_primary_thread(self, reply: Reply) -> bool: # REF1 in 'thread' model we give priority to running in main thread # note that we should be called with _running_lock hold primary_thread_task_ready = self._primary_thread_task_ready if primary_thread_task_ready is not None: if not primary_thread_task_ready.is_set(): self._primary_thread_task = reply # wake up primary thread primary_thread_task_ready.set() return True elif ( self.execmodel.backend == "main_thread_only" and self._primary_thread_task is not None ): self._primary_thread_task.waitfinish() self._primary_thread_task = reply # wake up primary thread (it's okay if this is already set # because we waited for the previous task to finish above # and integrate_as_primary_thread will not clear it when # it enters self._running_lock if it detects that a new # task is available) primary_thread_task_ready.set() return True return False def spawn(self, func, *args, **kwargs) -> Reply: """Asynchronously dispatch func(*args, **kwargs) and return a Reply.""" reply = Reply((func, args, kwargs), self.execmodel) with self._running_lock: if self._shuttingdown: raise ValueError("pool is shutting down") self._running.add(reply) if not self._try_send_to_primary_thread(reply): self.execmodel.start(self._perform_spawn, (reply,)) return reply def terminate(self, timeout: float | None = None) -> bool: """Trigger shutdown and wait for completion of all executions.""" self.trigger_shutdown() return self.waitall(timeout=timeout) def waitall(self, timeout: float | None = None) -> bool: """Wait until all active spawns have finished executing.""" with self._running_lock: if not self._running: return True # if a Reply still runs, we let run_and_release # signal us -- note that we are still holding the # _running_lock to avoid race conditions my_waitall_event = self.execmodel.Event() self._waitall_events.append(my_waitall_event) return my_waitall_event.wait(timeout=timeout) sysex = (KeyboardInterrupt, SystemExit) DEBUG = os.environ.get("EXECNET_DEBUG") pid = os.getpid() if DEBUG == "2": def trace(*msg: object) -> None: try: line = " ".join(map(str, msg)) sys.stderr.write(f"[{pid}] {line}\n") sys.stderr.flush() except Exception: pass # nothing we can do, likely interpreter-shutdown elif DEBUG: import os import tempfile fn = os.path.join(tempfile.gettempdir(), "execnet-debug-%d" % pid) # sys.stderr.write("execnet-debug at %r" % (fn,)) debugfile = open(fn, "w") def trace(*msg: object) -> None: try: line = " ".join(map(str, msg)) debugfile.write(line + "\n") debugfile.flush() except Exception as exc: try: sys.stderr.write(f"[{pid}] exception during tracing: {exc!r}\n") except Exception: pass # nothing we can do, likely interpreter-shutdown else: notrace = trace = lambda *msg: None class Popen2IO: error = (IOError, OSError, EOFError) def __init__(self, outfile, infile, execmodel: ExecModel) -> None: # we need raw byte streams self.outfile, self.infile = outfile, infile if sys.platform == "win32": import msvcrt try: msvcrt.setmode(infile.fileno(), os.O_BINARY) msvcrt.setmode(outfile.fileno(), os.O_BINARY) except (AttributeError, OSError): pass self._read = getattr(infile, "buffer", infile).read self._write = getattr(outfile, "buffer", outfile).write self.execmodel = execmodel def read(self, numbytes: int) -> bytes: """Read exactly 'numbytes' bytes from the pipe.""" # a file in non-blocking mode may return less bytes, so we loop buf = b"" while numbytes > len(buf): data = self._read(numbytes - len(buf)) if not data: raise EOFError("expected %d bytes, got %d" % (numbytes, len(buf))) buf += data return buf def write(self, data: bytes) -> None: """Write out all data bytes.""" assert isinstance(data, bytes) self._write(data) self.outfile.flush() def close_read(self) -> None: self.infile.close() def close_write(self) -> None: self.outfile.close() class Message: """Encapsulates Messages and their wire protocol.""" # message code -> name, handler _types: dict[int, tuple[str, Callable[[Message, BaseGateway], None]]] = {} def __init__(self, msgcode: int, channelid: int = 0, data: bytes = b"") -> None: self.msgcode = msgcode self.channelid = channelid self.data = data @staticmethod def from_io(io: ReadIO) -> Message: try: header = io.read(9) # type 1, channel 4, payload 4 if not header: raise EOFError("empty read") except EOFError as e: raise EOFError("couldn't load message header, " + e.args[0]) from None msgtype, channel, payload = struct.unpack("!bii", header) return Message(msgtype, channel, io.read(payload)) def to_io(self, io: WriteIO) -> None: header = struct.pack("!bii", self.msgcode, self.channelid, len(self.data)) io.write(header + self.data) def received(self, gateway: BaseGateway) -> None: handler = self._types[self.msgcode][1] handler(self, gateway) def __repr__(self) -> str: name = self._types[self.msgcode][0] return f"" def _status(message: Message, gateway: BaseGateway) -> None: # we use the channelid to send back information # but don't instantiate a channel object d = { "numchannels": len(gateway._channelfactory._channels), # TODO(typing): Attribute `_execpool` is only on WorkerGateway. "numexecuting": gateway._execpool.active_count(), # type: ignore[attr-defined] "execmodel": gateway.execmodel.backend, } gateway._send(Message.CHANNEL_DATA, message.channelid, dumps_internal(d)) gateway._send(Message.CHANNEL_CLOSE, message.channelid) STATUS = 0 _types[STATUS] = ("STATUS", _status) def _reconfigure(message: Message, gateway: BaseGateway) -> None: data = loads_internal(message.data, gateway) assert isinstance(data, tuple) strconfig: tuple[bool, bool] = data if message.channelid == 0: gateway._strconfig = strconfig else: gateway._channelfactory.new(message.channelid)._strconfig = strconfig RECONFIGURE = 1 _types[RECONFIGURE] = ("RECONFIGURE", _reconfigure) def _gateway_terminate(message: Message, gateway: BaseGateway) -> None: raise GatewayReceivedTerminate(gateway) GATEWAY_TERMINATE = 2 _types[GATEWAY_TERMINATE] = ("GATEWAY_TERMINATE", _gateway_terminate) def _channel_exec(message: Message, gateway: BaseGateway) -> None: channel = gateway._channelfactory.new(message.channelid) gateway._local_schedulexec(channel=channel, sourcetask=message.data) CHANNEL_EXEC = 3 _types[CHANNEL_EXEC] = ("CHANNEL_EXEC", _channel_exec) def _channel_data(message: Message, gateway: BaseGateway) -> None: gateway._channelfactory._local_receive(message.channelid, message.data) CHANNEL_DATA = 4 _types[CHANNEL_DATA] = ("CHANNEL_DATA", _channel_data) def _channel_close(message: Message, gateway: BaseGateway) -> None: gateway._channelfactory._local_close(message.channelid) CHANNEL_CLOSE = 5 _types[CHANNEL_CLOSE] = ("CHANNEL_CLOSE", _channel_close) def _channel_close_error(message: Message, gateway: BaseGateway) -> None: error_message = loads_internal(message.data) assert isinstance(error_message, str) remote_error = RemoteError(error_message) gateway._channelfactory._local_close(message.channelid, remote_error) CHANNEL_CLOSE_ERROR = 6 _types[CHANNEL_CLOSE_ERROR] = ("CHANNEL_CLOSE_ERROR", _channel_close_error) def _channel_last_message(message: Message, gateway: BaseGateway) -> None: gateway._channelfactory._local_close(message.channelid, sendonly=True) CHANNEL_LAST_MESSAGE = 7 _types[CHANNEL_LAST_MESSAGE] = ("CHANNEL_LAST_MESSAGE", _channel_last_message) class GatewayReceivedTerminate(Exception): """Receiverthread got termination message.""" def geterrortext( exc: BaseException, format_exception=traceback.format_exception, sysex: tuple[type[BaseException], ...] = sysex, ) -> str: try: # In py310, can change this to: # l = format_exception(exc) l = format_exception(type(exc), exc, exc.__traceback__) errortext = "".join(l) except sysex: raise except BaseException: errortext = f"{type(exc).__name__}: {exc}" return errortext class RemoteError(Exception): """Exception containing a stringified error from the other side.""" def __init__(self, formatted: str) -> None: super().__init__() self.formatted = formatted def __str__(self) -> str: return self.formatted def __repr__(self) -> str: return f"{self.__class__.__name__}: {self.formatted}" def warn(self) -> None: if self.formatted != INTERRUPT_TEXT: # XXX do this better sys.stderr.write(f"[{os.getpid()}] Warning: unhandled {self!r}\n") class TimeoutError(IOError): """Exception indicating that a timeout was reached.""" NO_ENDMARKER_WANTED = object() class Channel: """Communication channel between two Python Interpreter execution points.""" RemoteError = RemoteError TimeoutError = TimeoutError _INTERNALWAKEUP = 1000 _executing = False def __init__(self, gateway: BaseGateway, id: int) -> None: """:private:""" assert isinstance(id, int) assert not isinstance(gateway, type) self.gateway = gateway # XXX: defaults copied from Unserializer self._strconfig = getattr(gateway, "_strconfig", (True, False)) self.id = id self._items = self.gateway.execmodel.queue.Queue() self._closed = False self._receiveclosed = self.gateway.execmodel.Event() self._remoteerrors: list[RemoteError] = [] def _trace(self, *msg: object) -> None: self.gateway._trace(self.id, *msg) def setcallback( self, callback: Callable[[Any], Any], endmarker: object = NO_ENDMARKER_WANTED, ) -> None: """Set a callback function for receiving items. All already-queued items will immediately trigger the callback. Afterwards the callback will execute in the receiver thread for each received data item and calls to ``receive()`` will raise an error. If an endmarker is specified the callback will eventually be called with the endmarker when the channel closes. """ _callbacks = self.gateway._channelfactory._callbacks with self.gateway._receivelock: if self._items is None: raise OSError(f"{self!r} has callback already registered") items = self._items self._items = None while 1: try: olditem = items.get(block=False) except self.gateway.execmodel.queue.Empty: if not (self._closed or self._receiveclosed.is_set()): _callbacks[self.id] = (callback, endmarker, self._strconfig) break else: if olditem is ENDMARKER: items.put(olditem) # for other receivers if endmarker is not NO_ENDMARKER_WANTED: callback(endmarker) break else: callback(olditem) def __repr__(self) -> str: flag = self.isclosed() and "closed" or "open" return "" % (self.id, flag) def __del__(self) -> None: if self.gateway is None: # can be None in tests return # type: ignore[unreachable] self._trace("channel.__del__") # no multithreading issues here, because we have the last ref to 'self' if self._closed: # state transition "closed" --> "deleted" for error in self._remoteerrors: error.warn() elif self._receiveclosed.is_set(): # state transition "sendonly" --> "deleted" # the remote channel is already in "deleted" state, nothing to do pass else: # state transition "opened" --> "deleted" # check if we are in the middle of interpreter shutdown # in which case the process will go away and we probably # don't need to try to send a closing or last message # (and often it won't work anymore to send things out) if Message is not None: if self._items is None: # has_callback msgcode = Message.CHANNEL_LAST_MESSAGE else: msgcode = Message.CHANNEL_CLOSE try: self.gateway._send(msgcode, self.id) except (OSError, ValueError): # ignore problems with sending pass def _getremoteerror(self): try: return self._remoteerrors.pop(0) except IndexError: try: return self.gateway._error except AttributeError: pass return None # # public API for channel objects # def isclosed(self) -> bool: """Return True if the channel is closed. A closed channel may still hold items. """ return self._closed @overload def makefile(self, mode: Literal["r"], proxyclose: bool = ...) -> ChannelFileRead: pass @overload def makefile( self, mode: Literal["w"] = ..., proxyclose: bool = ..., ) -> ChannelFileWrite: pass def makefile( self, mode: Literal["r", "w"] = "w", proxyclose: bool = False, ) -> ChannelFileWrite | ChannelFileRead: """Return a file-like object. mode can be 'w' or 'r' for writeable/readable files. If proxyclose is true, file.close() will also close the channel. """ if mode == "w": return ChannelFileWrite(channel=self, proxyclose=proxyclose) elif mode == "r": return ChannelFileRead(channel=self, proxyclose=proxyclose) raise ValueError(f"mode {mode!r} not available") def close(self, error=None) -> None: """Close down this channel with an optional error message. Note that closing of a channel tied to remote_exec happens automatically at the end of execution and cannot be done explicitly. """ if self._executing: raise OSError("cannot explicitly close channel within remote_exec") if self._closed: self.gateway._trace(self, "ignoring redundant call to close()") if not self._closed: # state transition "opened/sendonly" --> "closed" # threads warning: the channel might be closed under our feet, # but it's never damaging to send too many CHANNEL_CLOSE messages # however, if the other side triggered a close already, we # do not send back a closed message. if not self._receiveclosed.is_set(): put = self.gateway._send if error is not None: put(Message.CHANNEL_CLOSE_ERROR, self.id, dumps_internal(error)) else: put(Message.CHANNEL_CLOSE, self.id) self._trace("sent channel close message") if isinstance(error, RemoteError): self._remoteerrors.append(error) self._closed = True # --> "closed" self._receiveclosed.set() queue = self._items if queue is not None: queue.put(ENDMARKER) self.gateway._channelfactory._no_longer_opened(self.id) def waitclose(self, timeout: float | None = None) -> None: """Wait until this channel is closed (or the remote side otherwise signalled that no more data was being sent). The channel may still hold receiveable items, but not receive any more after waitclose() has returned. Exceptions from executing code on the other side are reraised as local channel.RemoteErrors. EOFError is raised if the reading-connection was prematurely closed, which often indicates a dying process. self.TimeoutError is raised after the specified number of seconds (default is None, i.e. wait indefinitely). """ # wait for non-"opened" state self._receiveclosed.wait(timeout=timeout) if not self._receiveclosed.is_set(): raise self.TimeoutError("Timeout after %r seconds" % timeout) error = self._getremoteerror() if error: raise error def send(self, item: object) -> None: """Sends the given item to the other side of the channel, possibly blocking if the sender queue is full. The item must be a simple Python type and will be copied to the other side by value. OSError is raised if the write pipe was prematurely closed. """ if self.isclosed(): raise OSError(f"cannot send to {self!r}") self.gateway._send(Message.CHANNEL_DATA, self.id, dumps_internal(item)) def receive(self, timeout: float | None = None) -> Any: """Receive a data item that was sent from the other side. timeout: None [default] blocked waiting. A positive number indicates the number of seconds after which a channel.TimeoutError exception will be raised if no item was received. Note that exceptions from the remotely executing code will be reraised as channel.RemoteError exceptions containing a textual representation of the remote traceback. """ itemqueue = self._items if itemqueue is None: raise OSError("cannot receive(), channel has receiver callback") try: x = itemqueue.get(timeout=timeout) except self.gateway.execmodel.queue.Empty: raise self.TimeoutError("no item after %r seconds" % timeout) from None if x is ENDMARKER: itemqueue.put(x) # for other receivers raise self._getremoteerror() or EOFError() else: return x def __iter__(self) -> Iterator[Any]: return self def next(self) -> Any: try: return self.receive() except EOFError: raise StopIteration from None __next__ = next def reconfigure( self, py2str_as_py3str: bool = True, py3str_as_py2str: bool = False ) -> None: """Set the string coercion for this channel. The default is to try to convert py2 str as py3 str, but not to try and convert py3 str to py2 str """ self._strconfig = (py2str_as_py3str, py3str_as_py2str) data = dumps_internal(self._strconfig) self.gateway._send(Message.RECONFIGURE, self.id, data=data) ENDMARKER = object() INTERRUPT_TEXT = "keyboard-interrupted" MAIN_THREAD_ONLY_DEADLOCK_TEXT = ( "concurrent remote_exec would cause deadlock for main_thread_only execmodel" ) class ChannelFactory: def __init__(self, gateway: BaseGateway, startcount: int = 1) -> None: self._channels: weakref.WeakValueDictionary[int, Channel] = ( weakref.WeakValueDictionary() ) # Channel ID => (callback, end marker, strconfig) self._callbacks: dict[ int, tuple[Callable[[Any], Any], object, tuple[bool, bool]] ] = {} self._writelock = gateway.execmodel.Lock() self.gateway = gateway self.count = startcount self.finished = False self._list = list # needed during interp-shutdown def new(self, id: int | None = None) -> Channel: """Create a new Channel with 'id' (or create new id if None).""" with self._writelock: if self.finished: raise OSError(f"connection already closed: {self.gateway}") if id is None: id = self.count self.count += 2 try: channel = self._channels[id] except KeyError: channel = self._channels[id] = Channel(self.gateway, id) return channel def channels(self) -> list[Channel]: return self._list(self._channels.values()) # # internal methods, called from the receiver thread # def _no_longer_opened(self, id: int) -> None: try: del self._channels[id] except KeyError: pass try: callback, endmarker, strconfig = self._callbacks.pop(id) except KeyError: pass else: if endmarker is not NO_ENDMARKER_WANTED: callback(endmarker) def _local_close(self, id: int, remoteerror=None, sendonly: bool = False) -> None: channel = self._channels.get(id) if channel is None: # channel already in "deleted" state if remoteerror: remoteerror.warn() self._no_longer_opened(id) else: # state transition to "closed" state if remoteerror: channel._remoteerrors.append(remoteerror) queue = channel._items if queue is not None: queue.put(ENDMARKER) self._no_longer_opened(id) if not sendonly: # otherwise #--> "sendonly" channel._closed = True # --> "closed" channel._receiveclosed.set() def _local_receive(self, id: int, data) -> None: # executes in receiver thread channel = self._channels.get(id) try: callback, endmarker, strconfig = self._callbacks[id] except KeyError: queue = channel._items if channel is not None else None if queue is None: pass # drop data else: item = loads_internal(data, channel) queue.put(item) else: try: data = loads_internal(data, channel, strconfig) callback(data) # even if channel may be already closed except Exception as exc: self.gateway._trace("exception during callback: %s" % exc) errortext = self.gateway._geterrortext(exc) self.gateway._send( Message.CHANNEL_CLOSE_ERROR, id, dumps_internal(errortext) ) self._local_close(id, errortext) def _finished_receiving(self) -> None: with self._writelock: self.finished = True for id in self._list(self._channels): self._local_close(id, sendonly=True) for id in self._list(self._callbacks): self._no_longer_opened(id) class ChannelFile: def __init__(self, channel: Channel, proxyclose: bool = True) -> None: self.channel = channel self._proxyclose = proxyclose def isatty(self) -> bool: return False def close(self) -> None: if self._proxyclose: self.channel.close() def __repr__(self) -> str: state = self.channel.isclosed() and "closed" or "open" return "" % (self.channel.id, state) class ChannelFileWrite(ChannelFile): def write(self, out: bytes) -> None: self.channel.send(out) def flush(self) -> None: pass class ChannelFileRead(ChannelFile): def __init__(self, channel: Channel, proxyclose: bool = True) -> None: super().__init__(channel, proxyclose) self._buffer: str | None = None def read(self, n: int) -> str: try: if self._buffer is None: self._buffer = cast(str, self.channel.receive()) while len(self._buffer) < n: self._buffer += cast(str, self.channel.receive()) except EOFError: self.close() if self._buffer is None: ret = "" else: ret = self._buffer[:n] self._buffer = self._buffer[n:] return ret def readline(self) -> str: if self._buffer is not None: i = self._buffer.find("\n") if i != -1: return self.read(i + 1) line = self.read(len(self._buffer) + 1) else: line = self.read(1) while line and line[-1] != "\n": c = self.read(1) if not c: break line += c return line class BaseGateway: _sysex = sysex id = "" def __init__(self, io: IO, id, _startcount: int = 2) -> None: self.execmodel = io.execmodel self._io = io self.id = id self._strconfig = (Unserializer.py2str_as_py3str, Unserializer.py3str_as_py2str) self._channelfactory = ChannelFactory(self, _startcount) self._receivelock = self.execmodel.RLock() # globals may be NONE at process-termination self.__trace = trace self._geterrortext = geterrortext self._receivepool = WorkerPool(self.execmodel) def _trace(self, *msg: object) -> None: self.__trace(self.id, *msg) def _initreceive(self) -> None: self._receivepool.spawn(self._thread_receiver) def _thread_receiver(self) -> None: def log(*msg: object) -> None: self._trace("[receiver-thread]", *msg) log("RECEIVERTHREAD: starting to run") io = self._io try: while 1: msg = Message.from_io(io) log("received", msg) with self._receivelock: msg.received(self) del msg except (KeyboardInterrupt, GatewayReceivedTerminate): pass except EOFError as exc: log("EOF without prior gateway termination message") self._error = exc except Exception as exc: log(self._geterrortext(exc)) log("finishing receiving thread") # wake up and terminate any execution waiting to receive self._channelfactory._finished_receiving() log("terminating execution") self._terminate_execution() log("closing read") self._io.close_read() log("closing write") self._io.close_write() log("terminating our receive pseudo pool") self._receivepool.trigger_shutdown() def _terminate_execution(self) -> None: pass def _send(self, msgcode: int, channelid: int = 0, data: bytes = b"") -> None: message = Message(msgcode, channelid, data) try: message.to_io(self._io) self._trace("sent", message) except (OSError, ValueError) as e: self._trace("failed to send", message, e) # ValueError might be because the IO is already closed raise OSError("cannot send (already closed?)") from e def _local_schedulexec(self, channel: Channel, sourcetask: bytes) -> None: channel.close("execution disallowed") # _____________________________________________________________________ # # High Level Interface # _____________________________________________________________________ # def newchannel(self) -> Channel: """Return a new independent channel.""" return self._channelfactory.new() def join(self, timeout: float | None = None) -> None: """Wait for receiverthread to terminate.""" self._trace("waiting for receiver thread to finish") self._receivepool.waitall(timeout) class WorkerGateway(BaseGateway): def _local_schedulexec(self, channel: Channel, sourcetask: bytes) -> None: if self._execpool.execmodel.backend == "main_thread_only": assert self._executetask_complete is not None # It's necessary to wait for a short time in order to ensure # that we do not report a false-positive deadlock error, since # channel close does not elicit a response that would provide # a guarantee to remote_exec callers that the previous task # has released the main thread. If the timeout expires then it # should be practically impossible to report a false-positive. if not self._executetask_complete.wait(timeout=1): channel.close(MAIN_THREAD_ONLY_DEADLOCK_TEXT) return # It's only safe to clear here because the above wait proves # that there is not a previous task about to set it again. self._executetask_complete.clear() sourcetask_ = loads_internal(sourcetask) self._execpool.spawn(self.executetask, (channel, sourcetask_)) def _terminate_execution(self) -> None: # called from receiverthread self._trace("shutting down execution pool") self._execpool.trigger_shutdown() if not self._execpool.waitall(5.0): self._trace("execution ongoing after 5 secs," " trying interrupt_main") # We try hard to terminate execution based on the assumption # that there is only one gateway object running per-process. if sys.platform != "win32": self._trace("sending ourselves a SIGINT") os.kill(os.getpid(), 2) # send ourselves a SIGINT elif interrupt_main is not None: self._trace("calling interrupt_main()") interrupt_main() if not self._execpool.waitall(10.0): self._trace( "execution did not finish in another 10 secs, " "calling os._exit()" ) os._exit(1) def serve(self) -> None: def trace(msg: str) -> None: self._trace("[serve] " + msg) hasprimary = self.execmodel.backend in ("thread", "main_thread_only") self._execpool = WorkerPool(self.execmodel, hasprimary=hasprimary) self._executetask_complete = None if self.execmodel.backend == "main_thread_only": self._executetask_complete = self.execmodel.Event() # Initialize state to indicate that there is no previous task # executing so that we don't need a separate flag to track this. self._executetask_complete.set() trace("spawning receiver thread") self._initreceive() try: if hasprimary: # this will return when we are in shutdown trace("integrating as primary thread") self._execpool.integrate_as_primary_thread() trace("joining receiver thread") self.join() except KeyboardInterrupt: # in the worker we can't really do anything sensible trace("swallowing keyboardinterrupt, serve finished") def executetask( self, item: tuple[Channel, tuple[str, str | None, str | None, dict[str, object]]], ) -> None: try: channel, (source, file_name, call_name, kwargs) = item loc: dict[str, Any] = {"channel": channel, "__name__": "__channelexec__"} self._trace(f"execution starts[{channel.id}]: {repr(source)[:50]}") channel._executing = True try: co = compile(source + "\n", file_name or "", "exec") exec(co, loc) if call_name: self._trace("calling %s(**%60r)" % (call_name, kwargs)) function = loc[call_name] function(channel, **kwargs) finally: channel._executing = False self._trace("execution finished") except KeyboardInterrupt: channel.close(INTERRUPT_TEXT) raise except BaseException as exc: if not isinstance(exc, EOFError): if not channel.gateway._channelfactory.finished: self._trace(f"got exception: {exc!r}") errortext = self._geterrortext(exc) channel.close(errortext) return self._trace("ignoring EOFError because receiving finished") channel.close() if self._executetask_complete is not None: # Indicate that this task has finished executing, meaning # that there is no possibility of it triggering a deadlock # for the next spawn call. self._executetask_complete.set() # # Cross-Python pickling code, tested from test_serializer.py # class DataFormatError(Exception): pass class DumpError(DataFormatError): """Error while serializing an object.""" class LoadError(DataFormatError): """Error while unserializing an object.""" def bchr(n: int) -> bytes: return bytes([n]) DUMPFORMAT_VERSION = bchr(2) FOUR_BYTE_INT_MAX = 2147483647 FLOAT_FORMAT = "!d" FLOAT_FORMAT_SIZE = struct.calcsize(FLOAT_FORMAT) COMPLEX_FORMAT = "!dd" COMPLEX_FORMAT_SIZE = struct.calcsize(COMPLEX_FORMAT) class _Stop(Exception): pass class opcode: """Container for name -> num mappings.""" BUILDTUPLE = b"@" BYTES = b"A" CHANNEL = b"B" FALSE = b"C" FLOAT = b"D" FROZENSET = b"E" INT = b"F" LONG = b"G" LONGINT = b"H" LONGLONG = b"I" NEWDICT = b"J" NEWLIST = b"K" NONE = b"L" PY2STRING = b"M" PY3STRING = b"N" SET = b"O" SETITEM = b"P" STOP = b"Q" TRUE = b"R" UNICODE = b"S" COMPLEX = b"T" class Unserializer: num2func: dict[bytes, Callable[[Unserializer], None]] = {} py2str_as_py3str = True # True py3str_as_py2str = False # false means py2 will get unicode def __init__( self, stream: ReadIO, channel_or_gateway: Channel | BaseGateway | None = None, strconfig: tuple[bool, bool] | None = None, ) -> None: if isinstance(channel_or_gateway, Channel): gw: BaseGateway | None = channel_or_gateway.gateway else: gw = channel_or_gateway if channel_or_gateway is not None: strconfig = channel_or_gateway._strconfig if strconfig: self.py2str_as_py3str, self.py3str_as_py2str = strconfig self.stream = stream if gw is None: self.channelfactory = None else: self.channelfactory = gw._channelfactory def load(self, versioned: bool = False) -> Any: if versioned: ver = self.stream.read(1) if ver != DUMPFORMAT_VERSION: raise LoadError("wrong dumpformat version %r" % ver) self.stack: list[object] = [] try: while True: opcode = self.stream.read(1) if not opcode: raise EOFError try: loader = self.num2func[opcode] except KeyError: raise LoadError( f"unknown opcode {opcode!r} - wire protocol corruption?" ) from None loader(self) except _Stop: if len(self.stack) != 1: raise LoadError("internal unserialization error") from None return self.stack.pop(0) else: raise LoadError("didn't get STOP") def load_none(self) -> None: self.stack.append(None) num2func[opcode.NONE] = load_none def load_true(self) -> None: self.stack.append(True) num2func[opcode.TRUE] = load_true def load_false(self) -> None: self.stack.append(False) num2func[opcode.FALSE] = load_false def load_int(self) -> None: i = self._read_int4() self.stack.append(i) num2func[opcode.INT] = load_int def load_longint(self) -> None: s = self._read_byte_string() self.stack.append(int(s)) num2func[opcode.LONGINT] = load_longint load_long = load_int num2func[opcode.LONG] = load_long load_longlong = load_longint num2func[opcode.LONGLONG] = load_longlong def load_float(self) -> None: binary = self.stream.read(FLOAT_FORMAT_SIZE) self.stack.append(struct.unpack(FLOAT_FORMAT, binary)[0]) num2func[opcode.FLOAT] = load_float def load_complex(self) -> None: binary = self.stream.read(COMPLEX_FORMAT_SIZE) self.stack.append(complex(*struct.unpack(COMPLEX_FORMAT, binary))) num2func[opcode.COMPLEX] = load_complex def _read_int4(self) -> int: value: int = struct.unpack("!i", self.stream.read(4))[0] return value def _read_byte_string(self) -> bytes: length = self._read_int4() as_bytes = self.stream.read(length) return as_bytes def load_py3string(self) -> None: as_bytes = self._read_byte_string() if self.py3str_as_py2str: # XXX Should we try to decode into latin-1? self.stack.append(as_bytes) else: self.stack.append(as_bytes.decode("utf-8")) num2func[opcode.PY3STRING] = load_py3string def load_py2string(self) -> None: as_bytes = self._read_byte_string() if self.py2str_as_py3str: s: bytes | str = as_bytes.decode("latin-1") else: s = as_bytes self.stack.append(s) num2func[opcode.PY2STRING] = load_py2string def load_bytes(self) -> None: s = self._read_byte_string() self.stack.append(s) num2func[opcode.BYTES] = load_bytes def load_unicode(self) -> None: self.stack.append(self._read_byte_string().decode("utf-8")) num2func[opcode.UNICODE] = load_unicode def load_newlist(self) -> None: length = self._read_int4() self.stack.append([None] * length) num2func[opcode.NEWLIST] = load_newlist def load_setitem(self) -> None: if len(self.stack) < 3: raise LoadError("not enough items for setitem") value = self.stack.pop() key = self.stack.pop() self.stack[-1][key] = value # type: ignore[index] num2func[opcode.SETITEM] = load_setitem def load_newdict(self) -> None: self.stack.append({}) num2func[opcode.NEWDICT] = load_newdict def _load_collection(self, type_: type) -> None: length = self._read_int4() if length: res = type_(self.stack[-length:]) del self.stack[-length:] self.stack.append(res) else: self.stack.append(type_()) def load_buildtuple(self) -> None: self._load_collection(tuple) num2func[opcode.BUILDTUPLE] = load_buildtuple def load_set(self) -> None: self._load_collection(set) num2func[opcode.SET] = load_set def load_frozenset(self) -> None: self._load_collection(frozenset) num2func[opcode.FROZENSET] = load_frozenset def load_stop(self) -> None: raise _Stop num2func[opcode.STOP] = load_stop def load_channel(self) -> None: id = self._read_int4() assert self.channelfactory is not None newchannel = self.channelfactory.new(id) self.stack.append(newchannel) num2func[opcode.CHANNEL] = load_channel def dumps(obj: object) -> bytes: """Serialize the given obj to a bytestring. The obj and all contained objects must be of a builtin Python type (so nested dicts, sets, etc. are all OK but not user-level instances). """ return _Serializer().save(obj, versioned=True) # type: ignore[return-value] def dump(byteio, obj: object) -> None: """write a serialized bytestring of the given obj to the given stream.""" _Serializer(write=byteio.write).save(obj, versioned=True) def loads( bytestring: bytes, py2str_as_py3str: bool = False, py3str_as_py2str: bool = False ) -> Any: """Deserialize the given bytestring to an object. py2str_as_py3str: If true then string (str) objects previously dumped on Python2 will be loaded as Python3 strings which really are text objects. py3str_as_py2str: If true then string (str) objects previously dumped on Python3 will be loaded as Python2 strings instead of unicode objects. If the bytestring was dumped with an incompatible protocol version or if the bytestring is corrupted, the ``execnet.DataFormatError`` will be raised. """ io = BytesIO(bytestring) return load( io, py2str_as_py3str=py2str_as_py3str, py3str_as_py2str=py3str_as_py2str ) def load( io: ReadIO, py2str_as_py3str: bool = False, py3str_as_py2str: bool = False ) -> Any: """Derserialize an object form the specified stream. Behaviour and parameters are otherwise the same as with ``loads`` """ strconfig = (py2str_as_py3str, py3str_as_py2str) return Unserializer(io, strconfig=strconfig).load(versioned=True) def loads_internal( bytestring: bytes, channelfactory=None, strconfig: tuple[bool, bool] | None = None, ) -> Any: io = BytesIO(bytestring) return Unserializer(io, channelfactory, strconfig).load() def dumps_internal(obj: object) -> bytes: return _Serializer().save(obj) # type: ignore[return-value] class _Serializer: _dispatch: dict[type, Callable[[_Serializer, object], None]] = {} def __init__(self, write: Callable[[bytes], None] | None = None) -> None: if write is None: self._streamlist: list[bytes] = [] write = self._streamlist.append self._write = write def save(self, obj: object, versioned: bool = False) -> bytes | None: # calling here is not re-entrant but multiple instances # may write to the same stream because of the common platform # atomic-write guarantee (concurrent writes each happen atomically) if versioned: self._write(DUMPFORMAT_VERSION) self._save(obj) self._write(opcode.STOP) try: streamlist = self._streamlist except AttributeError: return None return b"".join(streamlist) def _save(self, obj: object) -> None: tp = type(obj) try: dispatch = self._dispatch[tp] except KeyError: methodname = "save_" + tp.__name__ meth: Callable[[_Serializer, object], None] | None = getattr( self.__class__, methodname, None ) if meth is None: raise DumpError(f"can't serialize {tp}") from None dispatch = self._dispatch[tp] = meth dispatch(self, obj) def save_NoneType(self, non: None) -> None: self._write(opcode.NONE) def save_bool(self, boolean: bool) -> None: if boolean: self._write(opcode.TRUE) else: self._write(opcode.FALSE) def save_bytes(self, bytes_: bytes) -> None: self._write(opcode.BYTES) self._write_byte_sequence(bytes_) def save_str(self, s: str) -> None: self._write(opcode.PY3STRING) self._write_unicode_string(s) def _write_unicode_string(self, s: str) -> None: try: as_bytes = s.encode("utf-8") except UnicodeEncodeError as e: raise DumpError("strings must be utf-8 encodable") from e self._write_byte_sequence(as_bytes) def _write_byte_sequence(self, bytes_: bytes) -> None: self._write_int4(len(bytes_), "string is too long") self._write(bytes_) def _save_integral(self, i: int, short_op: bytes, long_op: bytes) -> None: if i <= FOUR_BYTE_INT_MAX: self._write(short_op) self._write_int4(i) else: self._write(long_op) self._write_byte_sequence(str(i).rstrip("L").encode("ascii")) def save_int(self, i: int) -> None: self._save_integral(i, opcode.INT, opcode.LONGINT) def save_long(self, l: int) -> None: self._save_integral(l, opcode.LONG, opcode.LONGLONG) def save_float(self, flt: float) -> None: self._write(opcode.FLOAT) self._write(struct.pack(FLOAT_FORMAT, flt)) def save_complex(self, cpx: complex) -> None: self._write(opcode.COMPLEX) self._write(struct.pack(COMPLEX_FORMAT, cpx.real, cpx.imag)) def _write_int4( self, i: int, error: str = "int must be less than %i" % (FOUR_BYTE_INT_MAX,) ) -> None: if i > FOUR_BYTE_INT_MAX: raise DumpError(error) self._write(struct.pack("!i", i)) def save_list(self, L: list[object]) -> None: self._write(opcode.NEWLIST) self._write_int4(len(L), "list is too long") for i, item in enumerate(L): self._write_setitem(i, item) def _write_setitem(self, key: object, value: object) -> None: self._save(key) self._save(value) self._write(opcode.SETITEM) def save_dict(self, d: dict[object, object]) -> None: self._write(opcode.NEWDICT) for key, value in d.items(): self._write_setitem(key, value) def save_tuple(self, tup: tuple[object, ...]) -> None: for item in tup: self._save(item) self._write(opcode.BUILDTUPLE) self._write_int4(len(tup), "tuple is too long") def _write_set(self, s: set[object] | frozenset[object], op: bytes) -> None: for item in s: self._save(item) self._write(op) self._write_int4(len(s), "set is too long") def save_set(self, s: set[object]) -> None: self._write_set(s, opcode.SET) def save_frozenset(self, s: frozenset[object]) -> None: self._write_set(s, opcode.FROZENSET) def save_Channel(self, channel: Channel) -> None: self._write(opcode.CHANNEL) self._write_int4(channel.id) def init_popen_io(execmodel: ExecModel) -> Popen2IO: if not hasattr(os, "dup"): # jython io = Popen2IO(sys.stdout, sys.stdin, execmodel) import tempfile sys.stdin = tempfile.TemporaryFile("r") sys.stdout = tempfile.TemporaryFile("w") else: try: devnull = os.devnull except AttributeError: if os.name == "nt": devnull = "NUL" else: devnull = "/dev/null" # stdin stdin = execmodel.fdopen(os.dup(0), "r", 1) fd = os.open(devnull, os.O_RDONLY) os.dup2(fd, 0) os.close(fd) # stdout stdout = execmodel.fdopen(os.dup(1), "w", 1) fd = os.open(devnull, os.O_WRONLY) os.dup2(fd, 1) # stderr for win32 if os.name == "nt": sys.stderr = execmodel.fdopen(os.dup(2), "w", 1) os.dup2(fd, 2) os.close(fd) io = Popen2IO(stdout, stdin, execmodel) # Use closefd=False since 0 and 1 are shared with # sys.__stdin__ and sys.__stdout__. sys.stdin = execmodel.fdopen(0, "r", 1, closefd=False) sys.stdout = execmodel.fdopen(1, "w", 1, closefd=False) return io def serve(io: IO, id) -> None: trace(f"creating workergateway on {io!r}") WorkerGateway(io=io, id=id, _startcount=2).serve()