.pre-commit-config.yaml

@@ -1,19 +1,19 @@
 repos:
 -   repo: https://github.com/PyCQA/autoflake
-    rev: v2.0.0
+    rev: v2.0.1
     hooks:
     -   id: autoflake
         args: ["--in-place", "--remove-unused-variables", "--remove-all-unused-imports"]
 -   repo: https://github.com/psf/black
-    rev: 22.10.0
+    rev: 23.1.0
     hooks:
     -   id: black
         args: [--safe, --quiet, --target-version, py35]
 -   repo: https://github.com/asottile/blacken-docs
-    rev: v1.12.1
+    rev: 1.13.0
     hooks:
     -   id: blacken-docs
-        additional_dependencies: [black==20.8b1]
+        additional_dependencies: [black==23.1.0]
 -   repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.4.0
     hooks:
@@ -26,7 +26,7 @@ repos:
     hooks:
     -   id: flake8
 -   repo: https://github.com/asottile/pyupgrade
-    rev: v3.2.2
+    rev: v3.3.1
     hooks:
     -   id: pyupgrade
         args: [--py3-plus]

CHANGELOG.rst

@@ -1,3 +1,27 @@
+pytest-xdist 3.2.0 (2023-02-07)
+===============================
+
+Improved Documentation
+----------------------
+
+- `#863 <https://github.com/pytest-dev/pytest-xdist/issues/863>`_: Document limitations for debugging due to standard I/O of workers not being forwarded. Also, mention remote debugging as a possible workaround.
+
+
+Features
+--------
+
+- `#855 <https://github.com/pytest-dev/pytest-xdist/issues/855>`_: Users can now configure ``load`` scheduling precision using ``--maxschedchunk`` command
+  line option.
+
+- `#858 <https://github.com/pytest-dev/pytest-xdist/issues/858>`_: New ``worksteal`` scheduler, based on the idea of `work stealing <https://en.wikipedia.org/wiki/Work_stealing>`_. It's similar to ``load`` scheduler, but it should handle tests with significantly differing duration better, and, at the same time, it should provide similar or better reuse of fixtures.
+
+
+Trivial Changes
+---------------
+
+- `#870 <https://github.com/pytest-dev/pytest-xdist/issues/870>`_: Make the tests pass even when ``$PYTEST_XDIST_AUTO_NUM_WORKERS`` is set.
+
+
 pytest-xdist 3.1.0 (2022-12-01)
 ===============================
 

docs/distribution.rst

@@ -82,4 +82,13 @@ The test distribution algorithm is configured with the ``--dist`` command-line o
   This will make sure ``test1`` and ``TestA::test2`` will run in the same worker.
   Tests without the ``xdist_group`` mark are distributed normally as in the ``--dist=load`` mode.
 
+* ``--dist worksteal``: Initially, tests are distributed evenly among all
+  available workers. When a worker completes most of its assigned tests and
+  doesn't have enough tests to continue (currently, every worker needs at least
+  two tests in its queue), an attempt is made to reassign ("steal") a portion
+  of tests from some other worker's queue. The results should be similar to
+  the ``load`` method, but ``worksteal`` should handle tests with significantly
+  differing duration better, and, at the same time, it should provide similar
+  or better reuse of fixtures.
+
 * ``--dist no``: The normal pytest execution mode, runs one test at a time (no distribution at all).

docs/how-to.rst

@@ -15,7 +15,7 @@ a test or fixture, you may use the ``worker_id`` fixture to do so:
 
     @pytest.fixture()
     def user_account(worker_id):
-        """ use a different account in each xdist worker """
+        """use a different account in each xdist worker"""
         return "account_%s" % worker_id
 
 When ``xdist`` is disabled (running with ``-n0`` for example), then
@@ -80,7 +80,7 @@ wanted to create a separate database for each test run:
 
     @pytest.fixture(scope="session", autouse=True)
     def create_unique_database(testrun_uid):
-        """ create a unique database for this particular test run """
+        """create a unique database for this particular test run"""
         database_url = f"psql://myapp-{testrun_uid}"
 
         with Semaphore(f"/{testrun_uid}-lock", flags=O_CREAT, initial_value=1):
@@ -90,7 +90,7 @@ wanted to create a separate database for each test run:
 
     @pytest.fixture()
     def db(testrun_uid):
-        """ retrieve unique database """
+        """retrieve unique database"""
         database_url = f"psql://myapp-{testrun_uid}"
         return database_get_instance(database_url)
 

docs/known-limitations.rst

@@ -6,7 +6,7 @@ pytest-xdist has some limitations that may be supported in pytest but can't be s
 Order and amount of test must be consistent
 -------------------------------------------
 
-Is is not possible to have tests that differ in order or their amount across workers.
+It is not possible to have tests that differ in order or their amount across workers.
 
 This is especially true with ``pytest.mark.parametrize``, when values are produced with sets or other unordered iterables/generators.
 
@@ -59,6 +59,13 @@ Output (stdout and stderr) from workers
 
 The ``-s``/``--capture=no`` option is meant to disable pytest capture, so users can then see stdout and stderr output in the terminal from tests and application code in real time.
 
-However this option does not work with ``pytest-xdist`` because `execnet <https://github.com/pytest-dev/execnet>`__ the underlying library used for communication between master and workers, does not support transferring stdout/stderr from workers.
+However, this option does not work with ``pytest-xdist`` because `execnet <https://github.com/pytest-dev/execnet>`__ the underlying library used for communication between master and workers, does not support transferring stdout/stderr from workers.
 
-Currenlty there are no plans ot support this in ``pytest-xdist``.
+Currently, there are no plans to support this in ``pytest-xdist``.
+
+Debugging
+~~~~~~~~~
+
+This also means that debugging using PDB (or any other debugger that wants to use standard I/O) will not work. The ``--pdb`` option is disabled when distributing tests with ``pytest-xdist`` for this reason.
+
+It is generally likely best to use ``pytest-xdist`` to find failing tests and then debug them without distribution; however, if you need to debug from within a worker process (for example, to address failures that only happen when running tests concurrently), remote debuggers (for example, `python-remote-pdb <https://github.com/ionelmc/python-remote-pdb>`__ or `python-web-pdb <https://github.com/romanvm/python-web-pdb>`__) have been reported to work for this purpose.

src/xdist/dsession.py

@@ -8,6 +8,7 @@
     LoadScopeScheduling,
     LoadFileScheduling,
     LoadGroupScheduling,
+    WorkStealingScheduling,
 )
 
 
@@ -100,6 +101,7 @@ def pytest_xdist_make_scheduler(self, config, log):
             "loadscope": LoadScopeScheduling,
             "loadfile": LoadFileScheduling,
             "loadgroup": LoadGroupScheduling,
+            "worksteal": WorkStealingScheduling,
         }
         return schedulers[dist](config, log)
 
@@ -282,6 +284,17 @@ def worker_runtest_protocol_complete(self, node, item_index, duration):
         """
         self.sched.mark_test_complete(node, item_index, duration)
 
+    def worker_unscheduled(self, node, indices):
+        """
+        Emitted when a node fires the 'unscheduled' event, signalling that
+        some tests have been removed from the worker's queue and should be
+        sent to some worker again.
+
+        This should happen only in response to 'steal' command, so schedulers
+        not using 'steal' command don't have to implement it.
+        """
+        self.sched.remove_pending_tests_from_node(node, indices)
+
     def worker_collectreport(self, node, rep):
         """Emitted when a node calls the pytest_collectreport hook.
 

src/xdist/looponfail.py

@@ -35,7 +35,6 @@ def pytest_addoption(parser):
 
 @pytest.hookimpl
 def pytest_cmdline_main(config):
-
     if config.getoption("looponfail"):
         usepdb = config.getoption("usepdb", False)  # a core option
         if usepdb:

src/xdist/plugin.py

@@ -94,7 +94,15 @@ def pytest_addoption(parser):
         "--dist",
         metavar="distmode",
         action="store",
-        choices=["each", "load", "loadscope", "loadfile", "loadgroup", "no"],
+        choices=[
+            "each",
+            "load",
+            "loadscope",
+            "loadfile",
+            "loadgroup",
+            "worksteal",
+            "no",
+        ],
         dest="dist",
         default="no",
         help=(
@@ -107,6 +115,8 @@ def pytest_addoption(parser):
             "loadfile: load balance by sending test grouped by file"
             " to any available environment.\n\n"
             "loadgroup: like load, but sends tests marked with 'xdist_group' to the same worker.\n\n"
+            "worksteal: split the test suite between available environments,"
+            " then rebalance when any worker runs out of tests.\n\n"
             "(default) no: run tests inprocess, don't distribute."
         ),
     )
@@ -153,6 +163,19 @@ def pytest_addoption(parser):
             "on every test run."
         ),
     )
+    group.addoption(
+        "--maxschedchunk",
+        action="store",
+        type=int,
+        help=(
+            "Maximum number of tests scheduled in one step for --dist=load. "
+            "Setting it to 1 will force pytest to send tests to workers one by "
+            "one - might be useful for a small number of slow tests. "
+            "Larger numbers will allow the scheduler to submit consecutive "
+            "chunks of tests to workers - allows reusing fixtures. "
+            "Unlimited if not set."
+        ),
+    )
 
     parser.addini(
         "rsyncdirs",

src/xdist/remote.py

@@ -6,6 +6,7 @@
     needs not to be installed in remote environments.
 """
 
+import contextlib
 import sys
 import os
 import time
@@ -56,14 +57,21 @@ def worker_title(title):
 
 
 class WorkerInteractor:
+    SHUTDOWN_MARK = object()
+
     def __init__(self, config, channel):
         self.config = config
         self.workerid = config.workerinput.get("workerid", "?")
         self.testrunuid = config.workerinput["testrunuid"]
         self.log = Producer(f"worker-{self.workerid}", enabled=config.option.debug)
         self.channel = channel
+        self.torun = self._make_queue()
+        self.nextitem_index = None
         config.pluginmanager.register(self)
 
+    def _make_queue(self):
+        return self.channel.gateway.execmodel.queue.Queue()
+
     def sendevent(self, name, **kwargs):
         self.log("sending", name, kwargs)
         self.channel.send((name, kwargs))
@@ -92,38 +100,60 @@ def pytest_sessionfinish(self, exitstatus):
     def pytest_collection(self, session):
         self.sendevent("collectionstart")
 
+    def handle_command(self, command):
+        if command is self.SHUTDOWN_MARK:
+            self.torun.put(self.SHUTDOWN_MARK)
+            return
+
+        name, kwargs = command
+
+        self.log("received command", name, kwargs)
+        if name == "runtests":
+            for i in kwargs["indices"]:
+                self.torun.put(i)
+        elif name == "runtests_all":
+            for i in range(len(self.session.items)):
+                self.torun.put(i)
+        elif name == "shutdown":
+            self.torun.put(self.SHUTDOWN_MARK)
+        elif name == "steal":
+            self.steal(kwargs["indices"])
+
+    def steal(self, indices):
+        indices = set(indices)
+        stolen = []
+
+        old_queue, self.torun = self.torun, self._make_queue()
+
+        def old_queue_get_nowait_noraise():
+            with contextlib.suppress(self.channel.gateway.execmodel.queue.Empty):
+                return old_queue.get_nowait()
+
+        for i in iter(old_queue_get_nowait_noraise, None):
+            if i in indices:
+                stolen.append(i)
+            else:
+                self.torun.put(i)
+
+        self.sendevent("unscheduled", indices=stolen)
+
     @pytest.hookimpl
     def pytest_runtestloop(self, session):
         self.log("entering main loop")
-        torun = []
-        while 1:
-            try:
-                name, kwargs = self.channel.receive()
-            except EOFError:
-                return True
-            self.log("received command", name, kwargs)
-            if name == "runtests":
-                torun.extend(kwargs["indices"])
-            elif name == "runtests_all":
-                torun.extend(range(len(session.items)))
-            self.log("items to run:", torun)
-            # only run if we have an item and a next item
-            while len(torun) >= 2:
-                self.run_one_test(torun)
-            if name == "shutdown":
-                if torun:
-                    self.run_one_test(torun)
-                break
+        self.channel.setcallback(self.handle_command, endmarker=self.SHUTDOWN_MARK)
+        self.nextitem_index = self.torun.get()
+        while self.nextitem_index is not self.SHUTDOWN_MARK:
+            self.run_one_test()
         return True
 
-    def run_one_test(self, torun):
+    def run_one_test(self):
         items = self.session.items
-        self.item_index = torun.pop(0)
+        self.item_index, self.nextitem_index = self.nextitem_index, self.torun.get()
         item = items[self.item_index]
-        if torun:
-            nextitem = items[torun[0]]
-        else:
+        if self.nextitem_index is self.SHUTDOWN_MARK:
             nextitem = None
+        else:
+            nextitem = items[self.nextitem_index]
 
         worker_title("[pytest-xdist running] %s" % item.nodeid)
 

src/xdist/scheduler/__init__.py

@@ -3,3 +3,4 @@
 from xdist.scheduler.loadfile import LoadFileScheduling  # noqa
 from xdist.scheduler.loadscope import LoadScopeScheduling  # noqa
 from xdist.scheduler.loadgroup import LoadGroupScheduling  # noqa
+from xdist.scheduler.worksteal import WorkStealingScheduling  # noqa

src/xdist/scheduler/load.py

@@ -64,6 +64,7 @@ def __init__(self, config, log=None):
         else:
             self.log = log.loadsched
         self.config = config
+        self.maxschedchunk = self.config.getoption("maxschedchunk")
 
     @property
     def nodes(self):
@@ -185,7 +186,9 @@ def check_schedule(self, node, duration=0):
                     # so let's rather wait with sending new items
                     return
                 num_send = items_per_node_max - len(node_pending)
-                self._send_tests(node, num_send)
+                # keep at least 2 tests pending even if --maxschedchunk=1
+                maxschedchunk = max(2 - len(node_pending), self.maxschedchunk)
+                self._send_tests(node, min(num_send, maxschedchunk))
         else:
             node.shutdown()
 
@@ -245,6 +248,9 @@ def schedule(self):
         if not self.collection:
             return
 
+        if self.maxschedchunk is None:
+            self.maxschedchunk = len(self.collection)
+
         # Send a batch of tests to run. If we don't have at least two
         # tests per node, we have to send them all so that we can send
         # shutdown signals and get all nodes working.
@@ -265,7 +271,8 @@ def schedule(self):
             # how many items per node do we have about?
             items_per_node = len(self.collection) // len(self.node2pending)
             # take a fraction of tests for initial distribution
-            node_chunksize = max(items_per_node // 4, 2)
+            node_chunksize = min(items_per_node // 4, self.maxschedchunk)
+            node_chunksize = max(node_chunksize, 2)
             # and initialize each node with a chunk of tests
             for node in self.nodes:
                 self._send_tests(node, node_chunksize)

src/xdist/scheduler/loadscope.py

@@ -213,13 +213,11 @@ def add_node_collection(self, node, collection):
 
         # A new node has been added later, perhaps an original one died.
         if self.collection_is_completed:
-
             # Assert that .schedule() should have been called by now
             assert self.collection
 
             # Check that the new collection matches the official collection
             if collection != self.collection:
-
                 other_node = next(iter(self.registered_collections.keys()))
 
                 msg = report_collection_diff(