Add main_thread_only execmodel

CHANGELOG.rst

@@ -1,8 +1,23 @@
 2.1.0 (UNRELEASED)
+------------------
+
+* `#243 <https://github.com/pytest-dev/execnet/pull/243>`__: Added ``main_thread_only``
+ execmodel which is derived from the thread execmodel and only executes ``remote_exec``
+ calls in the main thread.
 
+ Callers of ``remote_exec`` must use the returned channel to wait for a task to complete
+ before they call remote_exec again, otherwise the ``remote_exec`` call will fail with a
+ ``concurrent remote_exec would cause deadlock`` error. The main_thread_only execmodel
+ provides solutions for `#96 <https://github.com/pytest-dev/execnet/issues/96>`__ and
+ `pytest-dev/pytest-xdist#620 <https://github.com/pytest-dev/pytest-xdist/issues/620>`__
+ (pending a new `pytest-xdist` release).
+
+ Also fixed ``init_popen_io`` to use ``closefd=False`` for shared stdin and stdout file
+ descriptors, preventing ``Bad file descriptor`` errors triggered by test_stdouterrin_setnull.
 * Removed support for Python 3.7.
 * Added official support for Python 3.12.
 
+
 2.0.2 (2023-07-09)
 ------------------
 

doc/basics.rst

@@ -138,14 +138,14 @@ processes then you often want to call ``group.terminate()``
 yourself and specify a larger or not timeout.
 
 
-threading models: gevent, eventlet, thread
-===========================================
+threading models: gevent, eventlet, thread, main_thread_only
+====================================================================
 
 .. versionadded:: 1.2 (status: experimental!)
 
-execnet supports "thread", "eventlet" and "gevent" as thread models
-on each of the two sides. You need to decide which model to use
-before you create any gateways::
+execnet supports "main_thread_only", "thread", "eventlet" and "gevent"
+as thread models on each of the two sides. You need to decide which
+model to use before you create any gateways::
 
  # content of threadmodel.py
  import execnet

src/execnet/gateway_base.py

@@ -61,7 +61,7 @@ def sleep(self, delay):
  raise NotImplementedError()
 
  @abc.abstractmethod
- def fdopen(self, fd, mode, bufsize=1):
+ def fdopen(self, fd, mode, bufsize=1, closefd=True):
  raise NotImplementedError()
 
  @abc.abstractmethod
@@ -113,10 +113,10 @@ def start(self, func, args=()):
 
  return _thread.start_new_thread(func, args)
 
- def fdopen(self, fd, mode, bufsize=1):
+ def fdopen(self, fd, mode, bufsize=1, closefd=True):
  import os
 
- return os.fdopen(fd, mode, bufsize, encoding="utf-8")
+ return os.fdopen(fd, mode, bufsize, encoding="utf-8", closefd=closefd)
 
  def Lock(self):
  import threading
@@ -134,6 +134,10 @@ def Event(self):
  return threading.Event()
 
 
+class MainThreadOnlyExecModel(ThreadExecModel):
+ backend = "main_thread_only"
+
+
 class EventletExecModel(ExecModel):
  backend = "eventlet"
 
@@ -170,10 +174,10 @@ def start(self, func, args=()):
 
  return eventlet.spawn_n(func, *args)
 
- def fdopen(self, fd, mode, bufsize=1):
+ def fdopen(self, fd, mode, bufsize=1, closefd=True):
  import eventlet.green.os
 
- return eventlet.green.os.fdopen(fd, mode, bufsize)
+ return eventlet.green.os.fdopen(fd, mode, bufsize, closefd=closefd)
 
  def Lock(self):
  import eventlet.green.threading
@@ -227,11 +231,11 @@ def start(self, func, args=()):
 
  return gevent.spawn(func, *args)
 
- def fdopen(self, fd, mode, bufsize=1):
+ def fdopen(self, fd, mode, bufsize=1, closefd=True):
  # XXX
  import gevent.fileobject
 
- return gevent.fileobject.FileObjectThread(fd, mode, bufsize)
+ return gevent.fileobject.FileObjectThread(fd, mode, bufsize, closefd=closefd)
 
  def Lock(self):
  import gevent.lock
@@ -254,6 +258,8 @@ def get_execmodel(backend):
  return backend
  if backend == "thread":
  return ThreadExecModel()
+ elif backend == "main_thread_only":
+ return MainThreadOnlyExecModel()
  elif backend == "eventlet":
  return EventletExecModel()
  elif backend == "gevent":
@@ -322,7 +328,7 @@ def __init__(self, execmodel, hasprimary=False):
  self._shuttingdown = False
  self._waitall_events = []
  if hasprimary:
- if self.execmodel.backend != "thread":
+ if self.execmodel.backend not in ("thread", "main_thread_only"):
  raise ValueError("hasprimary=True requires thread model")
  self._primary_thread_task_ready = self.execmodel.Event()
  else:
@@ -332,7 +338,7 @@ def integrate_as_primary_thread(self):
  """integrate the thread with which we are called as a primary
  thread for executing functions triggered with spawn().
  """
- assert self.execmodel.backend == "thread", self.execmodel
+ assert self.execmodel.backend in ("thread", "main_thread_only"), self.execmodel
  primary_thread_task_ready = self._primary_thread_task_ready
  # interacts with code at REF1
  while 1:
@@ -345,7 +351,11 @@ def integrate_as_primary_thread(self):
  with self._running_lock:
  if self._shuttingdown:
  break
- primary_thread_task_ready.clear()
+ # 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):
  with self._running_lock:
@@ -376,6 +386,19 @@ def _try_send_to_primary_thread(self, 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):
@@ -857,6 +880,9 @@ def reconfigure(self, py2str_as_py3str=True, py3str_as_py2str=False):
 
 ENDMARKER = object()
 INTERRUPT_TEXT = "keyboard-interrupted"
+MAIN_THREAD_ONLY_DEADLOCK_TEXT = (
+ "concurrent remote_exec would cause deadlock for main_thread_only execmodel"
+)
 
 
 class ChannelFactory:
@@ -1105,6 +1131,20 @@ def join(self, timeout=None):
 
 class WorkerGateway(BaseGateway):
  def _local_schedulexec(self, channel, sourcetask):
+ if self._execpool.execmodel.backend == "main_thread_only":
+ # 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))
 
@@ -1132,8 +1172,14 @@ def serve(self):
  def trace(msg):
  self._trace("[serve] " + msg)
 
- hasprimary = self.execmodel.backend == "thread"
+ 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:
@@ -1176,6 +1222,11 @@ def executetask(self, item):
  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()
 
 
 #
@@ -1631,8 +1682,10 @@ def init_popen_io(execmodel):
  os.dup2(fd, 2)
  os.close(fd)
  io = Popen2IO(stdout, stdin, execmodel)
- sys.stdin = execmodel.fdopen(0, "r", 1)
- sys.stdout = execmodel.fdopen(1, "w", 1)
+ # 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
 
 

src/execnet/multi.py

@@ -107,7 +107,7 @@ def makegateway(self, spec=None):
 
  id=<string> specifies the gateway id
  python=<path> specifies which python interpreter to execute
- execmodel=model 'thread', 'eventlet', 'gevent' model for execution
+ execmodel=model 'thread', 'main_thread_only', 'eventlet', 'gevent' model for execution
  chdir=<path> specifies to which directory to change
  nice=<path> specifies process priority of new process
  env:NAME=value specifies a remote environment variable setting.

testing/conftest.py

@@ -124,7 +124,7 @@ def anypython(request):
  pytest.skip(f"no {name} found")
  if "execmodel" in request.fixturenames and name != "sys.executable":
  backend = request.getfixturevalue("execmodel").backend
- if backend != "thread":
+ if backend not in ("thread", "main_thread_only"):
  pytest.xfail(f"cannot run {backend!r} execmodel with bare {name}")
  return executable
 
@@ -173,9 +173,11 @@ def gw(request, execmodel, group):
  return gw
 
 
-@pytest.fixture(params=["thread", "eventlet", "gevent"], scope="session")
+@pytest.fixture(
+ params=["thread", "main_thread_only", "eventlet", "gevent"], scope="session"
+)
 def execmodel(request):
- if request.param != "thread":
+ if request.param not in ("thread", "main_thread_only"):
  pytest.importorskip(request.param)
  if request.param in ("eventlet", "gevent") and sys.platform == "win32":
  pytest.xfail(request.param + " does not work on win32")

testing/test_basics.py

@@ -249,12 +249,25 @@ class Arg:
 
 @pytest.mark.skipif("not hasattr(os, 'dup')")
 def test_stdouterrin_setnull(execmodel, capfd):
- gateway_base.init_popen_io(execmodel)
- os.write(1, b"hello")
- os.read(0, 1)
- out, err = capfd.readouterr()
- assert not out
- assert not err
+ # Backup and restore stdin state, and rely on capfd to handle
+ # this for stdout and stderr.
+ orig_stdin = sys.stdin
+ orig_stdin_fd = os.dup(0)
+ try:
+ # The returned Popen2IO instance can be garbage collected
+ # prematurely since we don't hold a reference here, but we
+ # tolerate this because it is intended to leave behind a
+ # sane state afterwards.
+ gateway_base.init_popen_io(execmodel)
+ os.write(1, b"hello")
+ os.read(0, 1)
+ out, err = capfd.readouterr()
+ assert not out
+ assert not err
+ finally:
+ sys.stdin = orig_stdin
+ os.dup2(orig_stdin_fd, 0)
+ os.close(orig_stdin_fd)
 
 
 class PseudoChannel:

testing/test_gateway.py

@@ -525,3 +525,82 @@ def sendback(channel):
  if interleave_getstatus:
  print(gw.remote_status())
  assert ch.receive(timeout=0.5) == 1234
+
+
+def test_assert_main_thread_only(execmodel, makegateway):
+ if execmodel.backend != "main_thread_only":
+ pytest.skip("can only run with main_thread_only")
+
+ gw = makegateway(spec=f"execmodel={execmodel.backend}//popen")
+
+ try:
+ # Submit multiple remote_exec requests in quick succession and
+ # assert that all tasks execute in the main thread. It is
+ # necessary to call receive on each channel before the next
+ # remote_exec call, since the channel will raise an error if
+ # concurrent remote_exec requests are submitted as in
+ # test_main_thread_only_concurrent_remote_exec_deadlock.
+ for i in range(10):
+ ch = gw.remote_exec(
+ """
+ import time, threading
+ time.sleep(0.02)
+ channel.send(threading.current_thread() is threading.main_thread())
+ """
+ )
+
+ try:
+ res = ch.receive()
+ finally:
+ ch.close()
+ # This doesn't actually block because we closed
+ # the channel already, but it does check for remote
+ # errors and raise them.
+ ch.waitclose()
+ if res is not True:
+ pytest.fail("remote raised\n%s" % res)
+ finally:
+ gw.exit()
+ gw.join()
+
+
+def test_main_thread_only_concurrent_remote_exec_deadlock(execmodel, makegateway):
+ if execmodel.backend != "main_thread_only":
+ pytest.skip("can only run with main_thread_only")
+
+ gw = makegateway(spec=f"execmodel={execmodel.backend}//popen")
+ channels = []
+ try:
+ # Submit multiple remote_exec requests in quick succession and
+ # assert that MAIN_THREAD_ONLY_DEADLOCK_TEXT is raised if
+ # concurrent remote_exec requests are submitted for the
+ # main_thread_only execmodel (as compensation for the lack of
+ # back pressure in remote_exec calls which do not attempt to
+ # block until the remote main thread is idle).
+ for i in range(2):
+ channels.append(
+ gw.remote_exec(
+ """
+ import threading
+ channel.send(threading.current_thread() is threading.main_thread())
+ # Wait forever, ensuring that the deadlock case triggers.
+ channel.gateway.execmodel.Event().wait()
+ """
+ )
+ )
+
+ expected_results = (
+ True,
+ execnet.gateway_base.MAIN_THREAD_ONLY_DEADLOCK_TEXT,
+ )
+ for expected, ch in zip(expected_results, channels):
+ try:
+ res = ch.receive()
+ except execnet.RemoteError as e:
+ res = e.formatted
+ assert res == expected
+ finally:
+ for ch in channels:
+ ch.close()
+ gw.exit()
+ gw.join()