Release 3.5.0

Follow up to # 632.

---------

Co-authored-by: Jörg Kohlsdorf <joerg@cornershopapp.com>

.github/workflows/deploy.yml

@@ -0,0 +1,55 @@
+name: deploy
+
+on:
+  workflow_dispatch:
+    inputs:
+      version:
+        description: 'Release version'
+        required: true
+        default: '1.2.3'
+
+jobs:
+
+  package:
+    runs-on: ubuntu-latest
+    env:
+      SETUPTOOLS_SCM_PRETEND_VERSION: ${{ github.event.inputs.version }}
+
+    steps:
+    - uses: actions/checkout@v3
+
+    - name: Build and Check Package
+      uses: hynek/build-and-inspect-python-package@v1.5
+
+  deploy:
+    needs: package
+    runs-on: ubuntu-latest
+    environment: deploy
+    permissions:
+      id-token: write  # For PyPI trusted publishers.
+      contents: write  # For tag.
+
+    steps:
+    - uses: actions/checkout@v3
+
+    - name: Download Package
+      uses: actions/download-artifact@v3
+      with:
+        name: Packages
+        path: dist
+
+    - name: Publish package to PyPI
+      uses: pypa/gh-action-pypi-publish@v1.8.5
+
+    - name: Push tag
+      run: |
+        git config user.name "pytest bot"
+        git config user.email "pytestbot@gmail.com"
+        git tag --annotate --message=v${{ github.event.inputs.version }} v${{ github.event.inputs.version }} ${{ github.sha }}
+        git push origin v${{ github.event.inputs.version }}
+
+    - name: GitHub Release
+      uses: softprops/action-gh-release@v1
+      with:
+        files: dist/*
+        tag_name: v${{ github.event.inputs.version }}

.github/workflows/main.yml

@@ -1,87 +0,0 @@
-name: build
-
-on: [push, pull_request]
-
-jobs:
-  build:
-
-    runs-on: ${{ matrix.os }}
-
-    strategy:
-      fail-fast: false
-      matrix:
-        tox_env:
-          - "py36-pytestlatest"
-          - "py37-pytestlatest"
-          - "py38-pytestlatest"
-          - "py39-pytestlatest"
-          - "py310-pytestlatest"
-          - "py38-pytestmain"
-          - "py38-psutil"
-          - "py38-setproctitle"
-
-        os: [ubuntu-latest, windows-latest]
-        include:
-          - tox_env: "py36-pytestlatest"
-            python: "3.6"
-          - tox_env: "py37-pytestlatest"
-            python: "3.7"
-          - tox_env: "py38-pytestlatest"
-            python: "3.8"
-          - tox_env: "py39-pytestlatest"
-            python: "3.9"
-          - tox_env: "py310-pytestlatest"
-            python: "3.10-dev"
-          - tox_env: "py38-pytestmain"
-            python: "3.8"
-          - tox_env: "py38-psutil"
-            python: "3.8"
-          - tox_env: "py38-setproctitle"
-            python: "3.8"
-
-    steps:
-    - uses: actions/checkout@v3
-      with:
-        # Needed to fetch tags, which are required by setuptools-scm.
-        fetch-depth: 0
-    - name: Set up Python
-      uses: actions/setup-python@v2
-      with:
-        python-version: ${{ matrix.python }}
-    - name: Install tox
-      run: |
-        python -m pip install --upgrade pip
-        pip install tox
-    - name: Test
-      run: |
-        tox -e ${{ matrix.tox_env }}
-
-  deploy:
-
-    if: github.event_name == 'push' && startsWith(github.event.ref, 'refs/tags')
-
-    runs-on: ubuntu-latest
-
-    needs: build
-
-    steps:
-    - uses: actions/checkout@v3
-      with:
-        # Needed to fetch tags, which are required by setuptools-scm.
-        fetch-depth: 0
-    - name: Set up Python
-      uses: actions/setup-python@v2
-      with:
-        python-version: "3.7"
-    - name: Install wheel
-      run: |
-        python -m pip install --upgrade pip
-        pip install wheel
-    - name: Build package
-      run: |
-        python setup.py sdist bdist_wheel
-    - name: Publish package to PyPI
-      uses: pypa/gh-action-pypi-publish@master
-      with:
-        user: __token__
-        password: ${{ secrets.pypi_token }}

.github/workflows/test.yml

@@ -0,0 +1,93 @@
+name: test
+
+on:
+  push:
+    branches:
+      - master
+      - "test-me-*"
+
+  pull_request:
+    branches:
+      - "*"
+
+# Cancel running jobs for the same workflow and branch.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+
+  package:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+    - name: Build and Check Package
+      uses: hynek/build-and-inspect-python-package@v1.5
+
+  test:
+
+    needs: [package]
+
+    runs-on: ${{ matrix.os }}
+
+    strategy:
+      fail-fast: false
+      matrix:
+        tox_env:
+          - "py37-pytestlatest"
+          - "py38-pytestlatest"
+          - "py39-pytestlatest"
+          - "py310-pytestlatest"
+          - "py311-pytestlatest"
+          - "py311-pytestmain"
+          - "py312-pytestlatest"
+          - "py310-psutil"
+          - "py310-setproctitle"
+
+        os: [ubuntu-latest, windows-latest]
+        include:
+          - tox_env: "py37-pytestlatest"
+            python: "3.7"
+          - tox_env: "py38-pytestlatest"
+            python: "3.8"
+          - tox_env: "py39-pytestlatest"
+            python: "3.9"
+          - tox_env: "py310-pytestlatest"
+            python: "3.10"
+          - tox_env: "py311-pytestlatest"
+            python: "3.11"
+          - tox_env: "py311-pytestmain"
+            python: "3.11"
+          - tox_env: "py312-pytestlatest"
+            python: "3.12"
+          - tox_env: "py310-psutil"
+            python: "3.10"
+          - tox_env: "py310-setproctitle"
+            python: "3.10"
+
+    steps:
+    - uses: actions/checkout@v3
+      with:
+        # Needed to fetch tags, which are required by setuptools-scm.
+        fetch-depth: 0
+
+    - name: Download Package
+      uses: actions/download-artifact@v3
+      with:
+        name: Packages
+        path: dist
+
+    - name: Set up Python
+      uses: actions/setup-python@v2
+      with:
+        python-version: ${{ matrix.python }}
+
+    - name: Install tox
+      run: |
+        python -m pip install --upgrade pip
+        pip install tox
+
+    - name: Test
+      shell: bash
+      run: |
+        tox run -e ${{ matrix.tox_env }} --installpkg `find dist/*.tar.gz`

.pre-commit-config.yaml

@@ -1,32 +1,32 @@
 repos:
 -   repo: https://github.com/PyCQA/autoflake
-    rev: v1.7.6
+    rev: v2.2.1
     hooks:
     -   id: autoflake
         args: ["--in-place", "--remove-unused-variables", "--remove-all-unused-imports"]
 -   repo: https://github.com/psf/black
-    rev: 22.3.0
+    rev: 23.11.0
     hooks:
     -   id: black
         args: [--safe, --quiet, --target-version, py35]
 -   repo: https://github.com/asottile/blacken-docs
-    rev: v1.12.1
+    rev: 1.16.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.2.0
+    rev: v4.5.0
     hooks:
     -   id: trailing-whitespace
     -   id: end-of-file-fixer
     -   id: check-yaml
     -   id: debug-statements
 -   repo: https://github.com/PyCQA/flake8
-    rev: 4.0.1
+    rev: 6.1.0
     hooks:
     -   id: flake8
 -   repo: https://github.com/asottile/pyupgrade
-    rev: v2.32.1
+    rev: v3.15.0
     hooks:
     -   id: pyupgrade
         args: [--py3-plus]
@@ -39,7 +39,7 @@ repos:
         language: python
         additional_dependencies: [pygments, restructuredtext_lint]
 -   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v0.960
+    rev: v1.7.0
     hooks:
     -   id: mypy
         files: ^(src/|testing/)

.readthedocs.yaml

@@ -0,0 +1,14 @@
+version: 2
+
+build:
+  os: ubuntu-22.04
+  tools:
+    python: "3.11"
+
+sphinx:
+  configuration: docs/conf.py
+
+python:
+  install:
+    - path: .
+    - requirements: docs/requirements.txt

CHANGELOG.rst

@@ -1,13 +1,105 @@
-pytest-xdist 3.0.1 (2022-10-25)
+pytest-xdist 3.5.0 (2023-11-21)
 ===============================
 
+Features
+--------
+
+- `#632 <https://github.com/pytest-dev/pytest-xdist/issues/632>`_: ``--dist=loadscope`` now sorts scopes by number of tests to assign largest scopes early -- in many cases this should improve overall test session running time, as there is less chance of a large scope being left to be processed near the end of the session, leaving other workers idle.
+
+
+pytest-xdist 3.4.0 (2023-11-11)
+===============================
+
+Features
+--------
+
+- `#963 <https://github.com/pytest-dev/pytest-xdist/issues/963>`_: Wait for workers to finish reporting when test run stops early.
+
+  This makes sure that the results of in-progress tests are displayed.
+  Previously these reports were being discarded, losing information about the
+  test run.
+
+- `#965 <https://github.com/pytest-dev/pytest-xdist/issues/965>`_: Added support for Python 3.12.
+
+
+pytest-xdist 3.3.1 (2023-05-19)
+===============================
+
+Bug Fixes
+---------
+
+- `#907 <https://github.com/pytest-dev/pytest-xdist/issues/907>`_: Avoid remote calls during startup as ``execnet`` by default does not ensure remote affinity with the
+  main thread and might accidentally schedule the pytest worker into a non-main thread, which breaks numerous frameworks,
+  for example ``asyncio``, ``anyio``, ``PyQt/PySide``, etc.
+
+  A more safe correction will require thread affinity in ``execnet`` (`pytest-dev/execnet#96 <https://github.com/pytest-dev/execnet/issues/96>`__).
+
+
+pytest-xdist 3.3.0 (2023-05-12)
+===============================
+
+Features
+--------
+
+- `#555 <https://github.com/pytest-dev/pytest-xdist/issues/555>`_: Improved progress output when collecting nodes to be less verbose.
+
+
+pytest-xdist 3.2.1 (2023-03-12)
+===============================
+
+Bug Fixes
+---------
+
+- `#884 <https://github.com/pytest-dev/pytest-xdist/issues/884>`_: Fixed hang in ``worksteal`` scheduler.
+
+
+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
 ---------------
 
-- `#830 <https://github.com/pytest-dev/pytest-xdist/issues/830>`_: Fixed failed deploy of ``3.0.0`` release.
+- `#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.0.0 (2022-10-24)
+pytest-xdist 3.1.0 (2022-12-01)
+===============================
+
+Features
+--------
+
+- `#789 <https://github.com/pytest-dev/pytest-xdist/issues/789>`_: Users can now set a default distribution mode in their configuration file:
+
+  .. code-block:: ini
+
+      [pytest]
+      addopts = --dist loadscope
+
+- `#842 <https://github.com/pytest-dev/pytest-xdist/issues/842>`_: Python 3.11 is now officially supported.
+
+
+Removals
+--------
+
+- `#842 <https://github.com/pytest-dev/pytest-xdist/issues/842>`_: Python 3.6 is no longer supported.
+
+
+pytest-xdist 3.0.2 (2022-10-25)
 ===============================
 
 Bug Fixes

README.rst

@@ -13,7 +13,7 @@ pytest-xdist
     :alt: Python versions
     :target: https://pypi.python.org/pypi/pytest-xdist
 
-.. image:: https://github.com/pytest-dev/pytest-xdist/workflows/build/badge.svg
+.. image:: https://github.com/pytest-dev/pytest-xdist/workflows/test/badge.svg
     :target: https://github.com/pytest-dev/pytest-xdist/actions
 
 .. image:: https://img.shields.io/badge/code%20style-black-000000.svg

RELEASING.rst

@@ -32,14 +32,8 @@ To publish a new release ``X.Y.Z``, the steps are as follows:
 
     $ tox -e release -- X.Y.Z
 
-#. Commit and push the branch for review.
+#. Commit and push the branch to ``upstream`` and open a PR.
 
-#. Once PR is **green** and **approved**, create and push a tag::
+#. Once the PR is **green** and **approved**, start the ``deploy`` workflow manually from the branch ``release-VERSION``, passing ``VERSION`` as parameter.
 
-    $ export VERSION=X.Y.Z
-    $ git tag v$VERSION release-$VERSION
-    $ git push git@github.com:pytest-dev/pytest-xdist.git v$VERSION
-
-That will build the package and publish it on ``PyPI`` automatically.
-
-#. Merge the release PR to `master`.
+#. Merge the release PR to ``master``.

docs/distribution.rst

@@ -14,7 +14,7 @@ With ``-n auto``, pytest-xdist will use as many processes as your computer
 has CPU cores.
 
 Use ``-n logical`` to use the number of *logical* CPU cores rather than
-physical ones. This currently requires the ``psutils`` package to be installed;
+physical ones. This currently requires the ``psutil`` package to be installed;
 if it is not, pytest-xdist will fall back to ``-n auto`` behavior.
 
 Pass a number, e.g. ``-n 8``, to specify the number of processes explicitly.
@@ -50,7 +50,8 @@ The test distribution algorithm is configured with the ``--dist`` command-line o
 .. _distribution modes:
 
 * ``--dist load`` **(default)**: Sends pending tests to any worker that is
-  available, without any guaranteed order.
+  available, without any guaranteed order. Scheduling can be fine-tuned with
+  the `--maxschedchunk` option, see output of `pytest --help`.
 
 * ``--dist loadscope``: Tests are grouped by **module** for *test functions*
   and by **class** for *test methods*. Groups are distributed to available
@@ -82,4 +83,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-it-works.rst

@@ -29,8 +29,8 @@ The execution flow is:
    corresponds to the position of that test in the original collection
    list. This works because all nodes have the same collection list, and
    saves bandwidth because the **controller** can now tell one of the
-   workers to just *execute test index 3* index of passing the full test
-   id.
+   workers to just *execute test index 3* instead of passing the full
+   test id.
 
 4. If **dist-mode** is **each**: the **controller** just sends the full
    list of test indexes to each node at this moment.

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)
 
@@ -221,7 +221,6 @@ Example:
     def pytest_configure(config):
         worker_id = os.environ.get("PYTEST_XDIST_WORKER")
         if worker_id is not None:
-            log_file = config.getini("worker_log_file")
             logging.basicConfig(
                 format=config.getini("log_file_format"),
                 filename=f"tests_{worker_id}.log",

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.

docs/remote.rst

@@ -9,13 +9,16 @@ Sending tests to remote SSH accounts
 
 .. warning::
 
-    This feature is deprecated because the support for ``rsync`` is faulty
+    The ``rsync`` feature is deprecated because its implementation is faulty
     in terms of reproducing the development environment in the remote
     worker, and there is no clear solution moving forward.
 
-    This feature is scheduled to be removed in release 4.0, to let the team
+    For that reason, ``rsync`` is scheduled to be removed in release 4.0, to let the team
     focus on a smaller set of features.
 
+    Note that SSH and socket server are not planned for removal, as they are part
+    of the ``execnet`` feature set.
+
 Suppose you have a package ``mypkg`` which contains some
 tests that you can successfully run locally. And you
 have a ssh-reachable machine ``myhost``.  Then

docs/requirements.txt

@@ -0,0 +1,2 @@
+sphinx
+sphinx-rtd-theme

pyproject.toml

@@ -1,6 +1,5 @@
 [build-system]
 requires = [
-  # sync with setup.py until we discard non-pep-517/518
   "setuptools>=45.0",
   "setuptools-scm[toml]>=6.2.3",
   "wheel",

setup.cfg

@@ -1,7 +1,8 @@
 [metadata]
 name = pytest-xdist
-description = pytest xdist plugin for distributed testing and loop-on-failing modes
+description = pytest xdist plugin for distributed testing, most importantly across multiple CPUs
 long_description = file: README.rst
+long_description_content_type = text/x-rst
 license = MIT
 author = holger krekel and contributors
 author_email = pytest-dev@python.org,holger@merlinux.eu
@@ -24,11 +25,12 @@ classifiers =
     Programming Language :: Python
     Programming Language :: Python :: 3
     Programming Language :: Python :: 3 :: Only
-    Programming Language :: Python :: 3.6
     Programming Language :: Python :: 3.7
     Programming Language :: Python :: 3.8
     Programming Language :: Python :: 3.9
     Programming Language :: Python :: 3.10
+    Programming Language :: Python :: 3.11
+    Programming Language :: Python :: 3.12
 license_file = LICENSE
 project_urls =
     Documentation=https://pytest-xdist.readthedocs.io/en/latest
@@ -40,11 +42,10 @@ project_urls =
 packages = find:
 package_dir = =src
 zip_safe = False
-python_requires = >=3.6
+python_requires = >=3.7
 install_requires =
     execnet>=1.1
     pytest>=6.2.0
-setup_requires = # left empty, enforce using isolated build system
 
 [options.packages.find]
 where = src
@@ -61,6 +62,8 @@ psutil = psutil>=3.0
 setproctitle = setproctitle
 
 [flake8]
+# Ignore any errors related to formatting, let black worry/fix them.
+ignore = E501, W503, E203
 max-line-length = 100
 
 [mypy]

setup.py

@@ -1,4 +0,0 @@
-from setuptools import setup
-
-if __name__ == "__main__":
-    setup()

src/xdist/dsession.py

@@ -1,3 +1,8 @@
+from __future__ import annotations
+import sys
+from enum import Enum, auto
+from typing import Sequence
+
 import pytest
 
 from xdist.remote import Producer
@@ -8,6 +13,7 @@ from xdist.scheduler import (
     LoadScopeScheduling,
     LoadFileScheduling,
     LoadGroupScheduling,
+    WorkStealingScheduling,
 )
 
 
@@ -100,6 +106,7 @@ class DSession:
             "loadscope": LoadScopeScheduling,
             "loadfile": LoadFileScheduling,
             "loadgroup": LoadGroupScheduling,
+            "worksteal": WorkStealingScheduling,
         }
         return schedulers[dist](config, log)
 
@@ -111,11 +118,14 @@ class DSession:
         assert self.sched is not None
 
         self.shouldstop = False
+        pending_exception = None
         while not self.session_finished:
             self.loop_once()
             if self.shouldstop:
                 self.triggershutdown()
-                raise Interrupted(str(self.shouldstop))
+                pending_exception = Interrupted(str(self.shouldstop))
+        if pending_exception:
+            raise pending_exception
         return True
 
     def loop_once(self):
@@ -169,7 +179,7 @@ class DSession:
         """
         self.config.hook.pytest_testnodedown(node=node, error=None)
         if node.workeroutput["exitstatus"] == 2:  # keyboard-interrupt
-            self.shouldstop = "{} received keyboard-interrupt".format(node)
+            self.shouldstop = f"{node} received keyboard-interrupt"
             self.worker_errordown(node, "keyboard-interrupt")
             return
         if node in self.sched.nodes:
@@ -230,7 +240,7 @@ class DSession:
     @pytest.hookimpl
     def pytest_terminal_summary(self, terminalreporter):
         if self.config.option.verbose >= 0 and self._summary_report:
-            terminalreporter.write_sep("=", "xdist: {}".format(self._summary_report))
+            terminalreporter.write_sep("=", f"xdist: {self._summary_report}")
 
     def worker_collectionfinish(self, node, ids):
         """worker has finished test collection.
@@ -249,14 +259,16 @@ class DSession:
         self._session.testscollected = len(ids)
         self.sched.add_node_collection(node, ids)
         if self.terminal:
-            self.trdist.setstatus(node.gateway.spec, "[%d]" % (len(ids)))
+            self.trdist.setstatus(
+                node.gateway.spec, WorkerStatus.CollectionDone, tests_collected=len(ids)
+            )
         if self.sched.collection_is_completed:
             if self.terminal and not self.sched.has_pending:
                 self.trdist.ensure_show_status()
                 self.terminal.write_line("")
                 if self.config.option.verbose > 0:
                     self.terminal.write_line(
-                        "scheduling tests via %s" % (self.sched.__class__.__name__)
+                        f"scheduling tests via {self.sched.__class__.__name__}"
                     )
             self.sched.schedule()
 
@@ -282,6 +294,17 @@ class DSession:
         """
         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.
 
@@ -331,21 +354,26 @@ class DSession:
     def _handlefailures(self, rep):
         if rep.failed:
             self.countfailures += 1
-            if self.maxfail and self.countfailures >= self.maxfail:
-                self.shouldstop = "stopping after %d failures" % (self.countfailures)
+            if (
+                self.maxfail
+                and self.countfailures >= self.maxfail
+                and not self.shouldstop
+            ):
+                self.shouldstop = f"stopping after {self.countfailures} failures"
 
     def triggershutdown(self):
-        self.log("triggering shutdown")
-        self.shuttingdown = True
-        for node in self.sched.nodes:
-            node.shutdown()
+        if not self.shuttingdown:
+            self.log("triggering shutdown")
+            self.shuttingdown = True
+            for node in self.sched.nodes:
+                node.shutdown()
 
     def handle_crashitem(self, nodeid, worker):
         # XXX get more reporting info by recording pytest_runtest_logstart?
         # XXX count no of failures and retry N times
         runner = self.config.pluginmanager.getplugin("runner")
         fspath = nodeid.split("::")[0]
-        msg = "worker {!r} crashed while running {!r}".format(worker.gateway.id, nodeid)
+        msg = f"worker {worker.gateway.id!r} crashed while running {nodeid!r}"
         rep = runner.TestReport(
             nodeid, (fspath, None, fspath), (), "failed", msg, "???"
         )
@@ -359,34 +387,51 @@ class DSession:
         self.config.hook.pytest_runtest_logreport(report=rep)
 
 
+class WorkerStatus(Enum):
+    """Status of each worker during creation/collection."""
+
+    # Worker spec has just been created.
+    Created = auto()
+
+    # Worker has been initialized.
+    Initialized = auto()
+
+    # Worker is now ready for collection.
+    ReadyForCollection = auto()
+
+    # Worker has finished collection.
+    CollectionDone = auto()
+
+
 class TerminalDistReporter:
-    def __init__(self, config):
+    def __init__(self, config) -> None:
         self.config = config
         self.tr = config.pluginmanager.getplugin("terminalreporter")
-        self._status = {}
+        self._status: dict[str, tuple[WorkerStatus, int]] = {}
         self._lastlen = 0
         self._isatty = getattr(self.tr, "isatty", self.tr.hasmarkup)
 
-    def write_line(self, msg):
+    def write_line(self, msg: str) -> None:
         self.tr.write_line(msg)
 
-    def ensure_show_status(self):
+    def ensure_show_status(self) -> None:
         if not self._isatty:
             self.write_line(self.getstatus())
 
-    def setstatus(self, spec, status, show=True):
-        self._status[spec.id] = status
+    def setstatus(
+        self, spec, status: WorkerStatus, *, tests_collected: int, show: bool = True
+    ) -> None:
+        self._status[spec.id] = (status, tests_collected)
         if show and self._isatty:
             self.rewrite(self.getstatus())
 
-    def getstatus(self):
+    def getstatus(self) -> str:
         if self.config.option.verbose >= 0:
-            parts = [
-                "{} {}".format(spec.id, self._status[spec.id]) for spec in self._specs
-            ]
-            return " / ".join(parts)
-        else:
-            return "bringing up nodes..."
+            line = get_workers_status_line(list(self._status.values()))
+            if line:
+                return line
+
+        return "bringing up nodes..."
 
     def rewrite(self, line, newline=False):
         pline = line + " " * max(self._lastlen - len(line), 0)
@@ -398,40 +443,43 @@ class TerminalDistReporter:
         self.tr.rewrite(pline, bold=True)
 
     @pytest.hookimpl
-    def pytest_xdist_setupnodes(self, specs):
+    def pytest_xdist_setupnodes(self, specs) -> None:
         self._specs = specs
         for spec in specs:
-            self.setstatus(spec, "I", show=False)
-        self.setstatus(spec, "I", show=True)
+            self.setstatus(spec, WorkerStatus.Created, tests_collected=0, show=False)
+        self.setstatus(spec, WorkerStatus.Created, tests_collected=0, show=True)
         self.ensure_show_status()
 
     @pytest.hookimpl
-    def pytest_xdist_newgateway(self, gateway):
+    def pytest_xdist_newgateway(self, gateway) -> None:
         if self.config.option.verbose > 0:
             rinfo = gateway._rinfo()
-            version = "%s.%s.%s" % rinfo.version_info[:3]
-            self.rewrite(
-                "[%s] %s Python %s cwd: %s"
-                % (gateway.id, rinfo.platform, version, rinfo.cwd),
-                newline=True,
-            )
-        self.setstatus(gateway.spec, "C")
+            different_interpreter = rinfo.executable != sys.executable
+            if different_interpreter:
+                version = "%s.%s.%s" % rinfo.version_info[:3]
+                self.rewrite(
+                    f"[{gateway.id}] {rinfo.platform} Python {version} cwd: {rinfo.cwd}",
+                    newline=True,
+                )
+        self.setstatus(gateway.spec, WorkerStatus.Initialized, tests_collected=0)
 
     @pytest.hookimpl
-    def pytest_testnodeready(self, node):
+    def pytest_testnodeready(self, node) -> None:
         if self.config.option.verbose > 0:
             d = node.workerinfo
-            infoline = "[{}] Python {}".format(
-                d["id"], d["version"].replace("\n", " -- ")
-            )
-            self.rewrite(infoline, newline=True)
-        self.setstatus(node.gateway.spec, "ok")
+            different_interpreter = d.get("executable") != sys.executable
+            if different_interpreter:
+                version = d["version"].replace("\n", " -- ")
+                self.rewrite(f"[{d['id']}] Python {version}", newline=True)
+        self.setstatus(
+            node.gateway.spec, WorkerStatus.ReadyForCollection, tests_collected=0
+        )
 
     @pytest.hookimpl
-    def pytest_testnodedown(self, node, error):
+    def pytest_testnodedown(self, node, error) -> None:
         if not error:
             return
-        self.write_line("[{}] node down: {}".format(node.gateway.id, error))
+        self.write_line(f"[{node.gateway.id}] node down: {error}")
 
 
 def get_default_max_worker_restart(config):
@@ -446,3 +494,36 @@ def get_default_max_worker_restart(config):
         # if --max-worker-restart was not provided, use a reasonable default (#226)
         result = config.option.numprocesses * 4
     return result
+
+
+def get_workers_status_line(
+    status_and_items: Sequence[tuple[WorkerStatus, int]]
+) -> str:
+    """
+    Return the line to display during worker setup/collection based on the
+    status of the workers and number of tests collected for each.
+    """
+    statuses = [s for s, c in status_and_items]
+    total_workers = len(statuses)
+    workers_noun = "worker" if total_workers == 1 else "workers"
+    if status_and_items and all(s == WorkerStatus.CollectionDone for s in statuses):
+        # All workers collect the same number of items, so we grab
+        # the total number of items from the first worker.
+        first = status_and_items[0]
+        status, tests_collected = first
+        tests_noun = "item" if tests_collected == 1 else "items"
+        return f"{total_workers} {workers_noun} [{tests_collected} {tests_noun}]"
+    if WorkerStatus.CollectionDone in statuses:
+        done = sum(1 for s, c in status_and_items if c > 0)
+        return f"collecting: {done}/{total_workers} {workers_noun}"
+    if WorkerStatus.ReadyForCollection in statuses:
+        ready = statuses.count(WorkerStatus.ReadyForCollection)
+        return f"ready: {ready}/{total_workers} {workers_noun}"
+    if WorkerStatus.Initialized in statuses:
+        initialized = statuses.count(WorkerStatus.Initialized)
+        return f"initialized: {initialized}/{total_workers} {workers_noun}"
+    if WorkerStatus.Created in statuses:
+        created = statuses.count(WorkerStatus.Created)
+        return f"created: {created}/{total_workers} {workers_noun}"
+
+    return ""

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:
@@ -146,7 +145,7 @@ def repr_pytest_looponfailinfo(failreports, rootdirs):
                 tr.line(report, red=True)
     tr.sep("#", "waiting for changes", bold=True)
     for rootdir in rootdirs:
-        tr.line("### Watching:   {}".format(rootdir), bold=True)
+        tr.line(f"### Watching:   {rootdir}", bold=True)
 
 
 def init_worker_session(channel, args, option_dict):

src/xdist/newhooks.py

@@ -25,14 +25,18 @@ def pytest_xdist_newgateway(gateway):
 
 
 @pytest.hookspec(
-    warn_on_impl="rsync feature is deprecated and will be removed in pytest-xdist 4.0"
+    warn_on_impl=DeprecationWarning(
+        "rsync feature is deprecated and will be removed in pytest-xdist 4.0"
+    )
 )
 def pytest_xdist_rsyncstart(source, gateways):
     """called before rsyncing a directory to remote gateways takes place."""
 
 
 @pytest.hookspec(
-    warn_on_impl="rsync feature is deprecated and will be removed in pytest-xdist 4.0"
+    warn_on_impl=DeprecationWarning(
+        "rsync feature is deprecated and will be removed in pytest-xdist 4.0"
+    )
 )
 def pytest_xdist_rsyncfinish(source, gateways):
     """called after rsyncing a directory to remote gateways takes place."""

src/xdist/plugin.py

@@ -18,7 +18,9 @@ def pytest_xdist_auto_num_workers(config):
         try:
             return int(env_var)
         except ValueError:
-            warnings.warn("PYTEST_XDIST_AUTO_NUM_WORKERS is not a number: {env_var!r}. Ignoring it.")
+            warnings.warn(
+                "PYTEST_XDIST_AUTO_NUM_WORKERS is not a number: {env_var!r}. Ignoring it."
+            )
 
     try:
         import psutil
@@ -92,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=(
@@ -105,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."
         ),
     )
@@ -151,6 +163,21 @@ 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. "
+            "Due to implementation reasons, at least 2 tests are scheduled per "
+            "worker at the start. Only later tests can be scheduled one by one. "
+            "Unlimited if not set."
+        ),
+    )
 
     parser.addini(
         "rsyncdirs",
@@ -188,7 +215,19 @@ def pytest_addhooks(pluginmanager):
 
 @pytest.hookimpl(trylast=True)
 def pytest_configure(config):
-    if config.getoption("dist") != "no" and not config.getvalue("collectonly"):
+    config_line = (
+        "xdist_group: specify group for tests should run in same session."
+        "in relation to one another. Provided by pytest-xdist."
+    )
+    config.addinivalue_line("markers", config_line)
+
+    # Skip this plugin entirely when only doing collection.
+    if config.getvalue("collectonly"):
+        return
+
+    # Create the distributed session in case we have a valid distribution
+    # mode and test environments.
+    if config.getoption("dist") != "no" and config.getoption("tx"):
         from xdist.dsession import DSession
 
         session = DSession(config)
@@ -197,6 +236,7 @@ def pytest_configure(config):
         if tr:
             tr.showfspath = False
 
+    # Deprecation warnings for deprecated command-line/configuration options.
     if config.getoption("looponfail", None) or config.getini("looponfailroots"):
         warning = DeprecationWarning(
             "The --looponfail command line argument and looponfailroots config variable are deprecated.\n"
@@ -211,12 +251,6 @@ def pytest_configure(config):
         )
         config.issue_config_time_warning(warning, 2)
 
-    config_line = (
-        "xdist_group: specify group for tests should run in same session."
-        "in relation to one another. " + "Provided by pytest-xdist."
-    )
-    config.addinivalue_line("markers", config_line)
-
 
 @pytest.hookimpl(tryfirst=True)
 def pytest_cmdline_main(config):

src/xdist/remote.py

@@ -6,6 +6,7 @@
     needs not to be installed in remote environments.
 """
 
+import contextlib
 import sys
 import os
 import time
@@ -28,8 +29,8 @@ class Producer:
     """
     Simplified implementation of the same interface as py.log, for backward compatibility
     since we dropped the dependency on pylib.
-    Note: this is defined here because this module can't depend on xdist, so we need 
-    to have the other way around.    
+    Note: this is defined here because this module can't depend on xdist, so we need
+    to have the other way around.
     """
 
     def __init__(self, name: str, *, enabled: bool = True):
@@ -56,14 +57,31 @@ def worker_title(title):
 
 
 class WorkerInteractor:
+    SHUTDOWN_MARK = object()
+    QUEUE_REPLACED_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 _get_next_item_index(self):
+        """Gets the next item from test queue. Handles the case when the queue
+        is replaced concurrently in another thread.
+        """
+        result = self.torun.get()
+        while result is self.QUEUE_REPLACED_MARK:
+            result = self.torun.get()
+        return result
+
     def sendevent(self, name, **kwargs):
         self.log("sending", name, kwargs)
         self.channel.send((name, kwargs))
@@ -92,38 +110,63 @@ class WorkerInteractor:
     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)
+        old_queue.put(self.QUEUE_REPLACED_MARK)
+
     @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._get_next_item_index()
+        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):
+        self.item_index = self.nextitem_index
+        self.nextitem_index = self._get_next_item_index()
+
         items = self.session.items
-        self.item_index = torun.pop(0)
         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)
 
@@ -149,7 +192,7 @@ class WorkerInteractor:
                     if len(mark.args) > 0
                     else mark.kwargs.get("name", "default")
                 )
-                item._nodeid = "{}@{}".format(item.nodeid, gname)
+                item._nodeid = f"{item.nodeid}@{gname}"
 
     @pytest.hookimpl
     def pytest_collection_finish(self, session):

src/xdist/scheduler/__init__.py

@@ -3,3 +3,4 @@ from xdist.scheduler.load import LoadScheduling  # noqa
 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 @@ class LoadScheduling:
         else:
             self.log = log.loadsched
         self.config = config
+        self.maxschedchunk = self.config.getoption("maxschedchunk")
 
     @property
     def nodes(self):
@@ -185,7 +186,9 @@ class LoadScheduling:
                     # 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 @@ class LoadScheduling:
         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 @@ class LoadScheduling:
             # 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 @@ class LoadScopeScheduling:
 
         # 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(
@@ -352,21 +350,28 @@ class LoadScopeScheduling:
             return
 
         # Determine chunks of work (scopes)
+        unsorted_workqueue = OrderedDict()
         for nodeid in self.collection:
             scope = self._split_scope(nodeid)
-            work_unit = self.workqueue.setdefault(scope, default=OrderedDict())
+            work_unit = unsorted_workqueue.setdefault(scope, default=OrderedDict())
             work_unit[nodeid] = False
 
+        # Insert tests scopes into work queue ordered by number of tests.
+        for scope, nodeids in sorted(
+            unsorted_workqueue.items(), key=lambda item: -len(item[1])
+        ):
+            self.workqueue[scope] = nodeids
+
         # Avoid having more workers than work
         extra_nodes = len(self.nodes) - len(self.workqueue)
 
         if extra_nodes > 0:
-            self.log("Shutting down {} nodes".format(extra_nodes))
+            self.log(f"Shutting down {extra_nodes} nodes")
 
             for _ in range(extra_nodes):
                 unused_node, assigned = self.assigned_work.popitem(last=True)
 
-                self.log("Shutting down unused node {}".format(unused_node))
+                self.log(f"Shutting down unused node {unused_node}")
                 unused_node.shutdown()
 
         # Assign initial workload

src/xdist/scheduler/worksteal.py

@@ -0,0 +1,325 @@
+from collections import namedtuple
+
+from _pytest.runner import CollectReport
+
+from xdist.remote import Producer
+from xdist.workermanage import parse_spec_config
+from xdist.report import report_collection_diff
+
+
+NodePending = namedtuple("NodePending", ["node", "pending"])
+
+# Every worker needs at least 2 tests in queue - the current and the next one.
+MIN_PENDING = 2
+
+
+class WorkStealingScheduling:
+    """Implement work-stealing scheduling.
+
+    Initially, tests are distributed evenly among all nodes.
+
+    When some node completes most of its assigned tests (when only one pending
+    test remains), an attempt is made to reassign ("steal") some tests from
+    other nodes to this node.
+
+    Attributes:
+
+    :numnodes: The expected number of nodes taking part.  The actual
+       number of nodes will vary during the scheduler's lifetime as
+       nodes are added by the DSession as they are brought up and
+       removed either because of a dead node or normal shutdown.  This
+       number is primarily used to know when the initial collection is
+       completed.
+
+    :node2collection: Map of nodes and their test collection.  All
+       collections should always be identical.
+
+    :node2pending: Map of nodes and the indices of their pending
+       tests.  The indices are an index into ``.pending`` (which is
+       identical to their own collection stored in
+       ``.node2collection``).
+
+    :collection: The one collection once it is validated to be
+       identical between all the nodes.  It is initialised to None
+       until ``.schedule()`` is called.
+
+    :pending: List of indices of globally pending tests.  These are
+       tests which have not yet been allocated to a chunk for a node
+       to process.
+
+    :log: A py.log.Producer instance.
+
+    :config: Config object, used for handling hooks.
+
+    :steal_requested_from_node: The node to which the current "steal" request
+       was sent. ``None`` if there is no request in progress. Only one request
+       can be in progress at any time, the scheduler doesn't send multiple
+       simultaneous requests.
+    """
+
+    def __init__(self, config, log=None):
+        self.numnodes = len(parse_spec_config(config))
+        self.node2collection = {}
+        self.node2pending = {}
+        self.pending = []
+        self.collection = None
+        if log is None:
+            self.log = Producer("workstealsched")
+        else:
+            self.log = log.workstealsched
+        self.config = config
+        self.steal_requested_from_node = None
+
+    @property
+    def nodes(self):
+        """A list of all nodes in the scheduler."""
+        return list(self.node2pending.keys())
+
+    @property
+    def collection_is_completed(self):
+        """Boolean indication initial test collection is complete.
+
+        This is a boolean indicating all initial participating nodes
+        have finished collection.  The required number of initial
+        nodes is defined by ``.numnodes``.
+        """
+        return len(self.node2collection) >= self.numnodes
+
+    @property
+    def tests_finished(self):
+        """Return True if all tests have been executed by the nodes."""
+        if not self.collection_is_completed:
+            return False
+        if self.pending:
+            return False
+        if self.steal_requested_from_node is not None:
+            return False
+        for pending in self.node2pending.values():
+            if len(pending) >= MIN_PENDING:
+                return False
+        return True
+
+    @property
+    def has_pending(self):
+        """Return True if there are pending test items
+
+        This indicates that collection has finished and nodes are
+        still processing test items, so this can be thought of as
+        "the scheduler is active".
+        """
+        if self.pending:
+            return True
+        for pending in self.node2pending.values():
+            if pending:
+                return True
+        return False
+
+    def add_node(self, node):
+        """Add a new node to the scheduler.
+
+        From now on the node will be allocated chunks of tests to
+        execute.
+
+        Called by the ``DSession.worker_workerready`` hook when it
+        successfully bootstraps a new node.
+        """
+        assert node not in self.node2pending
+        self.node2pending[node] = []
+
+    def add_node_collection(self, node, collection):
+        """Add the collected test items from a node
+
+        The collection is stored in the ``.node2collection`` map.
+        Called by the ``DSession.worker_collectionfinish`` hook.
+        """
+        assert node in self.node2pending
+        if self.collection_is_completed:
+            # A new node has been added later, perhaps an original one died.
+            # .schedule() should have
+            # been called by now
+            assert self.collection
+            if collection != self.collection:
+                other_node = next(iter(self.node2collection.keys()))
+                msg = report_collection_diff(
+                    self.collection, collection, other_node.gateway.id, node.gateway.id
+                )
+                self.log(msg)
+                return
+        self.node2collection[node] = list(collection)
+
+    def mark_test_complete(self, node, item_index, duration=None):
+        """Mark test item as completed by node
+
+        This is called by the ``DSession.worker_testreport`` hook.
+        """
+        self.node2pending[node].remove(item_index)
+        self.check_schedule()
+
+    def mark_test_pending(self, item):
+        self.pending.insert(
+            0,
+            self.collection.index(item),
+        )
+        self.check_schedule()
+
+    def remove_pending_tests_from_node(self, node, indices):
+        """Node returned some test indices back in response to 'steal' command.
+
+        This is called by ``DSession.worker_unscheduled``.
+        """
+        assert node is self.steal_requested_from_node
+        self.steal_requested_from_node = None
+
+        indices_set = set(indices)
+        self.node2pending[node] = [
+            i for i in self.node2pending[node] if i not in indices_set
+        ]
+        self.pending.extend(indices)
+        self.check_schedule()
+
+    def check_schedule(self):
+        """Reschedule tests/perform load balancing."""
+        nodes_up = [
+            NodePending(node, pending)
+            for node, pending in self.node2pending.items()
+            if not node.shutting_down
+        ]
+
+        def get_idle_nodes():
+            return [node for node, pending in nodes_up if len(pending) < MIN_PENDING]
+
+        idle_nodes = get_idle_nodes()
+        if not idle_nodes:
+            return
+
+        if self.pending:
+            # Distribute pending tests evenly among idle nodes
+            for i, node in enumerate(idle_nodes):
+                nodes_remaining = len(idle_nodes) - i
+                num_send = len(self.pending) // nodes_remaining
+                self._send_tests(node, num_send)
+
+            idle_nodes = get_idle_nodes()
+            # No need to steal anything if all nodes have enough work to continue
+            if not idle_nodes:
+                return
+
+        # Only one active stealing request is allowed
+        if self.steal_requested_from_node is not None:
+            return
+
+        # Find the node that has the longest test queue
+        steal_from = max(
+            nodes_up, key=lambda node_pending: len(node_pending.pending), default=None
+        )
+
+        if steal_from is None:
+            num_steal = 0
+        else:
+            # Steal half of the test queue - but keep that node running too.
+            # If the node has 2 or less tests queued, stealing will fail
+            # anyway.
+            max_steal = max(0, len(steal_from.pending) - MIN_PENDING)
+            num_steal = min(len(steal_from.pending) // 2, max_steal)
+
+        if num_steal == 0:
+            # Can't get more work - shutdown idle nodes. This will force them
+            # to run the last test now instead of waiting for more tests.
+            for node in idle_nodes:
+                node.shutdown()
+            return
+
+        steal_from.node.send_steal(steal_from.pending[-num_steal:])
+        self.steal_requested_from_node = steal_from.node
+
+    def remove_node(self, node):
+        """Remove a node from the scheduler
+
+        This should be called either when the node crashed or at
+        shutdown time.  In the former case any pending items assigned
+        to the node will be re-scheduled.  Called by the
+        ``DSession.worker_workerfinished`` and
+        ``DSession.worker_errordown`` hooks.
+
+        Return the item which was being executing while the node
+        crashed or None if the node has no more pending items.
+
+        """
+        pending = self.node2pending.pop(node)
+
+        # If node was removed without completing its assigned tests - it crashed
+        if pending:
+            crashitem = self.collection[pending.pop(0)]
+        else:
+            crashitem = None
+
+        self.pending.extend(pending)
+
+        # Dead node won't respond to "steal" request
+        if self.steal_requested_from_node is node:
+            self.steal_requested_from_node = None
+
+        self.check_schedule()
+        return crashitem
+
+    def schedule(self):
+        """Initiate distribution of the test collection
+
+        Initiate scheduling of the items across the nodes.  If this
+        gets called again later it behaves the same as calling
+        ``.check_schedule()`` on all nodes so that newly added nodes
+        will start to be used.
+
+        This is called by the ``DSession.worker_collectionfinish`` hook
+        if ``.collection_is_completed`` is True.
+        """
+        assert self.collection_is_completed
+
+        # Initial distribution already happened, reschedule on all nodes
+        if self.collection is not None:
+            self.check_schedule()
+            return
+
+        if not self._check_nodes_have_same_collection():
+            self.log("**Different tests collected, aborting run**")
+            return
+
+        # Collections are identical, create the index of pending items.
+        self.collection = list(self.node2collection.values())[0]
+        self.pending[:] = range(len(self.collection))
+        if not self.collection:
+            return
+
+        self.check_schedule()
+
+    def _send_tests(self, node, num):
+        tests_per_node = self.pending[:num]
+        if tests_per_node:
+            del self.pending[:num]
+            self.node2pending[node].extend(tests_per_node)
+            node.send_runtest_some(tests_per_node)
+
+    def _check_nodes_have_same_collection(self):
+        """Return True if all nodes have collected the same items.
+
+        If collections differ, this method returns False while logging
+        the collection differences and posting collection errors to
+        pytest_collectreport hook.
+        """
+        node_collection_items = list(self.node2collection.items())
+        first_node, col = node_collection_items[0]
+        same_collection = True
+        for node, collection in node_collection_items[1:]:
+            msg = report_collection_diff(
+                col, collection, first_node.gateway.id, node.gateway.id
+            )
+            if msg:
+                same_collection = False
+                self.log(msg)
+                if self.config is not None:
+                    rep = CollectReport(
+                        node.gateway.id, "failed", longrepr=msg, result=[]
+                    )
+                    self.config.hook.pytest_collectreport(report=rep)
+
+        return same_collection

src/xdist/workermanage.py

@@ -112,7 +112,7 @@ class NodeManager:
         for root in candidates:
             root = Path(root).resolve()
             if not root.exists():
-                raise pytest.UsageError("rsyncdir doesn't exist: {!r}".format(root))
+                raise pytest.UsageError(f"rsyncdir doesn't exist: {root!r}")
             if root not in roots:
                 roots.append(root)
         return roots
@@ -192,7 +192,7 @@ class HostRSync(execnet.RSync):
         if self._verbose > 0:
             path = os.path.basename(self._sourcedir) + "/" + modified_rel_path
             remotepath = gateway.spec.chdir
-            print("{}:{} <= {}".format(gateway.spec, remotepath, path))
+            print(f"{gateway.spec}:{remotepath} <= {path}")
 
 
 def make_reltoroot(roots: Sequence[Path], args: List[str]) -> List[str]:
@@ -219,7 +219,7 @@ def make_reltoroot(roots: Sequence[Path], args: List[str]) -> List[str]:
                 parts[0] = root.name + "/" + str(x)
                 break
         else:
-            raise ValueError("arg {} not relative to an rsync root".format(arg))
+            raise ValueError(f"arg {arg} not relative to an rsync root")
         result.append(splitcode.join(parts))
     return result
 
@@ -249,7 +249,7 @@ class WorkerController:
         self.log = Producer(f"workerctl-{gateway.id}", enabled=config.option.debug)
 
     def __repr__(self):
-        return "<{} {}>".format(self.__class__.__name__, self.gateway.id)
+        return f"<{self.__class__.__name__} {self.gateway.id}>"
 
     @property
     def shutting_down(self):
@@ -300,6 +300,9 @@ class WorkerController:
     def send_runtest_all(self):
         self.sendcommand("runtests_all")
 
+    def send_steal(self, indices):
+        self.sendcommand("steal", indices=indices)
+
     def shutdown(self):
         if not self._down:
             try:
@@ -310,11 +313,11 @@ class WorkerController:
 
     def sendcommand(self, name, **kwargs):
         """send a named parametrized command to the other side."""
-        self.log("sending command {}(**{})".format(name, kwargs))
+        self.log(f"sending command {name}(**{kwargs})")
         self.channel.send((name, kwargs))
 
     def notify_inproc(self, eventname, **kwargs):
-        self.log("queuing {}(**{})".format(eventname, kwargs))
+        self.log(f"queuing {eventname}(**{kwargs})")
         self.putevent((eventname, kwargs))
 
     def process_from_remote(self, eventcall):  # noqa too complex
@@ -336,7 +339,7 @@ class WorkerController:
                 return
             eventname, kwargs = eventcall
             if eventname in ("collectionstart",):
-                self.log("ignoring {}({})".format(eventname, kwargs))
+                self.log(f"ignoring {eventname}({kwargs})")
             elif eventname == "workerready":
                 self.notify_inproc(eventname, node=self, **kwargs)
             elif eventname == "internal_error":
@@ -359,6 +362,8 @@ class WorkerController:
                 self.notify_inproc(eventname, node=self, ids=kwargs["ids"])
             elif eventname == "runtest_protocol_complete":
                 self.notify_inproc(eventname, node=self, **kwargs)
+            elif eventname == "unscheduled":
+                self.notify_inproc(eventname, node=self, **kwargs)
             elif eventname == "logwarning":
                 self.notify_inproc(
                     eventname,
@@ -389,7 +394,7 @@ class WorkerController:
                     location=kwargs["location"],
                 )
             else:
-                raise ValueError("unknown event: {}".format(eventname))
+                raise ValueError(f"unknown event: {eventname}")
         except KeyboardInterrupt:
             # should not land in receiver-thread
             raise

testing/acceptance_test.py

@@ -101,21 +101,52 @@ class TestDistribution:
             """
         )
         result = pytester.runpytest(p1, "-v", "-d", "--tx=popen", "--tx=popen")
-        result.stdout.fnmatch_lines(["*1*Python*", "*2 failed, 1 passed, 1 skipped*"])
+        result.stdout.fnmatch_lines(
+            [
+                "created: 2/2 workers",
+                "*2 failed, 1 passed, 1 skipped*",
+            ]
+        )
         assert result.ret == 1
 
-    def test_n1_fail_minus_x(self, pytester: pytest.Pytester) -> None:
+    def test_exitfail_waits_for_workers_to_finish(
+        self, pytester: pytest.Pytester
+    ) -> None:
+        """The DSession waits for workers before exiting early on failure.
+
+        When -x/--exitfail is set, the DSession wait for the workers to finish
+        before raising an Interrupt exception. This prevents reports from the
+        faiing test and other tests from being discarded.
+        """
         p1 = pytester.makepyfile(
             """
+            import time
+
             def test_fail1():
+                time.sleep(0.1)
                 assert 0
             def test_fail2():
+                time.sleep(0.2)
+            def test_fail3():
+                time.sleep(0.3)
                 assert 0
+            def test_fail4():
+                time.sleep(0.3)
+            def test_fail5():
+                time.sleep(0.3)
+            def test_fail6():
+                time.sleep(0.3)
         """
         )
-        result = pytester.runpytest(p1, "-x", "-v", "-n1")
+        result = pytester.runpytest(p1, "-x", "-rA", "-v", "-n2")
         assert result.ret == 2
-        result.stdout.fnmatch_lines(["*Interrupted: stopping*1*", "*1 failed*"])
+        result.stdout.re_match_lines([".*Interrupted: stopping.*[12].*"])
+        m = re.search(r"== (\d+) failed, (\d+) passed in ", str(result.stdout))
+        assert m
+        n_failed, n_passed = (int(s) for s in m.groups())
+        assert 1 <= n_failed <= 2
+        assert 1 <= n_passed <= 3
+        assert (n_passed + n_failed) < 6
 
     def test_basetemp_in_subprocesses(self, pytester: pytest.Pytester) -> None:
         p1 = pytester.makepyfile(
@@ -151,7 +182,12 @@ class TestDistribution:
         """
         )
         result = pytester.runpytest(p1, "-d", "-v")
-        result.stdout.fnmatch_lines(["*2*Python*", "*2 failed, 1 passed, 1 skipped*"])
+        result.stdout.fnmatch_lines(
+            [
+                "created: 3/3 workers",
+                "*2 failed, 1 passed, 1 skipped*",
+            ]
+        )
         assert result.ret == 1
 
     def test_dist_tests_with_crash(self, pytester: pytest.Pytester) -> None:
@@ -237,9 +273,6 @@ class TestDistribution:
         assert result.ret == 0
         result.stdout.fnmatch_lines(
             [
-                "*0* *cwd*",
-                # "RSyncStart: [G1]",
-                # "RSyncFinished: [G1]",
                 "*1 passed*",
             ]
         )
@@ -276,7 +309,11 @@ class TestDistribution:
         p1 = pytester.makepyfile("def test_func(): pass")
         result = pytester.runpytest("-v", p1, "-d", "--tx=popen")
         result.stdout.fnmatch_lines(
-            ["*0*Python*", "*calculated result is 49*", "*1 passed*"]
+            [
+                "created: 1/1 worker",
+                "*calculated result is 49*",
+                "*1 passed*",
+            ]
         )
         assert result.ret == 0
 
@@ -393,14 +430,14 @@ class TestTerminalReporting:
         out = result.stdout.str()
         if verbosity == "-v":
             assert "scheduling tests" in out
-            assert "gw" in out
+            assert "1 worker [1 item]" in out
         elif verbosity == "-q":
             assert "scheduling tests" not in out
             assert "gw" not in out
             assert "bringing up nodes..." in out
         else:
             assert "scheduling tests" not in out
-            assert "gw" in out
+            assert "1 worker [1 item]" in out
 
     def test_pass_skip_fail(self, pytester: pytest.Pytester) -> None:
         pytester.makepyfile(
@@ -1099,8 +1136,9 @@ def test_color_yes_collection_on_non_atty(pytester, request) -> None:
     result = pytester.runpytest(*args)
     assert "test session starts" in result.stdout.str()
     assert "\x1b[1m" in result.stdout.str()
-    assert "gw0 [10] / gw1 [10]" in result.stdout.str()
-    assert "gw0 C / gw1 C" not in result.stdout.str()
+    assert "created: 2/2 workers" in result.stdout.str()
+    assert "2 workers [10 items]" in result.stdout.str()
+    assert "collecting:" not in result.stdout.str()
 
 
 def test_without_terminal_plugin(pytester, request) -> None:
@@ -1138,7 +1176,7 @@ def test_internal_error_with_maxfail(pytester: pytest.Pytester) -> None:
     """
     )
     result = pytester.runpytest_subprocess("--maxfail=1", "-n1")
-    result.stdout.fnmatch_lines(["* 1 error in *"])
+    result.stdout.re_match_lines([".* [12] errors? in .*"])
     assert "INTERNALERROR" not in result.stderr.str()
 
 
@@ -1194,6 +1232,22 @@ class TestLoadScope:
             "test_a.py::TestB", result.outlines
         ) in ({"gw0": 10}, {"gw1": 10})
 
+    def test_workqueue_ordered_by_size(self, pytester: pytest.Pytester) -> None:
+        test_file = """
+            import pytest
+            @pytest.mark.parametrize('i', range({}))
+            def test(i):
+                pass
+        """
+        pytester.makepyfile(test_a=test_file.format(10), test_b=test_file.format(20))
+        result = pytester.runpytest("-n2", "--dist=loadscope", "-v")
+        assert get_workers_and_test_count_by_prefix(
+            "test_a.py::test", result.outlines
+        ) == {"gw1": 10}
+        assert get_workers_and_test_count_by_prefix(
+            "test_b.py::test", result.outlines
+        ) == {"gw0": 20}
+
     def test_module_single_start(self, pytester: pytest.Pytester) -> None:
         """Fix test suite never finishing in case all workers start with a single test (#277)."""
         test_file1 = """
@@ -1554,10 +1608,28 @@ def test_collection_crash(testdir):
     assert result.ret == 1
     result.stdout.fnmatch_lines(
         [
-            "gw0 I",
-            "gw0 [[]0[]]",
+            "created: 1/1 worker",
+            "1 worker [[]0 items[]]",
             "*_ ERROR collecting test_collection_crash.py _*",
             "E   assert 0",
             "*= 1 error in *",
         ]
     )
+
+
+def test_dist_in_addopts(testdir):
+    """Users can set a default distribution in the configuration file (#789)."""
+    testdir.makepyfile(
+        """
+        def test():
+            pass
+        """
+    )
+    testdir.makeini(
+        """
+        [pytest]
+        addopts = --dist loadscope
+        """
+    )
+    result = testdir.runpytest()
+    assert result.ret == 0

testing/test_dsession.py

@@ -1,7 +1,13 @@
-from xdist.dsession import DSession, get_default_max_worker_restart
+from __future__ import annotations
+from xdist.dsession import (
+    DSession,
+    get_default_max_worker_restart,
+    get_workers_status_line,
+    WorkerStatus,
+)
 from xdist.report import report_collection_diff
-from xdist.scheduler import EachScheduling, LoadScheduling
-from typing import Optional
+from xdist.scheduler import EachScheduling, LoadScheduling, WorkStealingScheduling
+from typing import Sequence
 
 import pytest
 import execnet
@@ -17,6 +23,7 @@ class MockGateway:
 class MockNode:
     def __init__(self) -> None:
         self.sent = []  # type: ignore[var-annotated]
+        self.stolen = []  # type: ignore[var-annotated]
         self.gateway = MockGateway()
         self._shutdown = False
 
@@ -26,6 +33,9 @@ class MockNode:
     def send_runtest_all(self) -> None:
         self.sent.append("ALL")
 
+    def send_steal(self, indices) -> None:
+        self.stolen.extend(indices)
+
     def shutdown(self) -> None:
         self._shutdown = True
 
@@ -129,30 +139,79 @@ class TestLoadScheduling:
         assert node1.sent == [0, 1, 4, 5]
         assert not sched.pending
 
-    def test_schedule_fewer_tests_than_nodes(self, pytester: pytest.Pytester) -> None:
+    def test_schedule_maxchunk_none(self, pytester: pytest.Pytester) -> None:
         config = pytester.parseconfig("--tx=2*popen")
         sched = LoadScheduling(config)
         sched.add_node(MockNode())
         sched.add_node(MockNode())
+        node1, node2 = sched.nodes
+        col = [f"test{i}" for i in range(16)]
+        sched.add_node_collection(node1, col)
+        sched.add_node_collection(node2, col)
+        sched.schedule()
+        assert node1.sent == [0, 1]
+        assert node2.sent == [2, 3]
+        assert sched.pending == list(range(4, 16))
+        assert sched.node2pending[node1] == node1.sent
+        assert sched.node2pending[node2] == node2.sent
+        sched.mark_test_complete(node1, 0)
+        assert node1.sent == [0, 1, 4, 5]
+        assert sched.pending == list(range(6, 16))
+        sched.mark_test_complete(node1, 1)
+        assert node1.sent == [0, 1, 4, 5]
+        assert sched.pending == list(range(6, 16))
+
+        for i in range(7, 16):
+            sched.mark_test_complete(node1, i - 3)
+            assert node1.sent == [0, 1] + list(range(4, i))
+            assert node2.sent == [2, 3]
+            assert sched.pending == list(range(i, 16))
+
+    def test_schedule_maxchunk_1(self, pytester: pytest.Pytester) -> None:
+        config = pytester.parseconfig("--tx=2*popen", "--maxschedchunk=1")
+        sched = LoadScheduling(config)
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
+        node1, node2 = sched.nodes
+        col = [f"test{i}" for i in range(16)]
+        sched.add_node_collection(node1, col)
+        sched.add_node_collection(node2, col)
+        sched.schedule()
+        assert node1.sent == [0, 1]
+        assert node2.sent == [2, 3]
+        assert sched.pending == list(range(4, 16))
+        assert sched.node2pending[node1] == node1.sent
+        assert sched.node2pending[node2] == node2.sent
+
+        for complete_index, first_pending in enumerate(range(5, 16)):
+            sched.mark_test_complete(node1, node1.sent[complete_index])
+            assert node1.sent == [0, 1] + list(range(4, first_pending))
+            assert node2.sent == [2, 3]
+            assert sched.pending == list(range(first_pending, 16))
+
+    def test_schedule_fewer_tests_than_nodes(self, pytester: pytest.Pytester) -> None:
+        config = pytester.parseconfig("--tx=3*popen")
+        sched = LoadScheduling(config)
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
         sched.add_node(MockNode())
         node1, node2, node3 = sched.nodes
         col = ["xyz"] * 2
         sched.add_node_collection(node1, col)
         sched.add_node_collection(node2, col)
+        sched.add_node_collection(node3, col)
+        assert sched.collection_is_completed
         sched.schedule()
         # assert not sched.tests_finished
-        sent1 = node1.sent
-        sent2 = node2.sent
-        sent3 = node3.sent
-        assert sent1 == [0]
-        assert sent2 == [1]
-        assert sent3 == []
+        assert node1.sent == [0]
+        assert node2.sent == [1]
+        assert node3.sent == []
         assert not sched.pending
 
     def test_schedule_fewer_than_two_tests_per_node(
         self, pytester: pytest.Pytester
     ) -> None:
-        config = pytester.parseconfig("--tx=2*popen")
+        config = pytester.parseconfig("--tx=3*popen")
         sched = LoadScheduling(config)
         sched.add_node(MockNode())
         sched.add_node(MockNode())
@@ -161,14 +220,13 @@ class TestLoadScheduling:
         col = ["xyz"] * 5
         sched.add_node_collection(node1, col)
         sched.add_node_collection(node2, col)
+        sched.add_node_collection(node3, col)
+        assert sched.collection_is_completed
         sched.schedule()
         # assert not sched.tests_finished
-        sent1 = node1.sent
-        sent2 = node2.sent
-        sent3 = node3.sent
-        assert sent1 == [0, 3]
-        assert sent2 == [1, 4]
-        assert sent3 == [2]
+        assert node1.sent == [0, 3]
+        assert node2.sent == [1, 4]
+        assert node3.sent == [2]
         assert not sched.pending
 
     def test_add_remove_node(self, pytester: pytest.Pytester) -> None:
@@ -217,6 +275,169 @@ class TestLoadScheduling:
         assert "Different tests were collected between" in rep.longrepr
 
 
+class TestWorkStealingScheduling:
+    def test_ideal_case(self, pytester: pytest.Pytester) -> None:
+        config = pytester.parseconfig("--tx=2*popen")
+        sched = WorkStealingScheduling(config)
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
+        node1, node2 = sched.nodes
+        collection = [f"test_workstealing.py::test_{i}" for i in range(16)]
+        assert not sched.collection_is_completed
+        sched.add_node_collection(node1, collection)
+        assert not sched.collection_is_completed
+        sched.add_node_collection(node2, collection)
+        assert sched.collection_is_completed
+        assert sched.node2collection[node1] == collection
+        assert sched.node2collection[node2] == collection
+        sched.schedule()
+        assert not sched.pending
+        assert not sched.tests_finished
+        assert node1.sent == list(range(0, 8))
+        assert node2.sent == list(range(8, 16))
+        for i in range(8):
+            sched.mark_test_complete(node1, node1.sent[i])
+            sched.mark_test_complete(node2, node2.sent[i])
+        assert sched.tests_finished
+        assert node1.stolen == []
+        assert node2.stolen == []
+
+    def test_stealing(self, pytester: pytest.Pytester) -> None:
+        config = pytester.parseconfig("--tx=2*popen")
+        sched = WorkStealingScheduling(config)
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
+        node1, node2 = sched.nodes
+        collection = [f"test_workstealing.py::test_{i}" for i in range(16)]
+        sched.add_node_collection(node1, collection)
+        sched.add_node_collection(node2, collection)
+        assert sched.collection_is_completed
+        sched.schedule()
+        assert node1.sent == list(range(0, 8))
+        assert node2.sent == list(range(8, 16))
+        for i in range(8):
+            sched.mark_test_complete(node1, node1.sent[i])
+        assert node2.stolen == list(range(12, 16))
+        sched.remove_pending_tests_from_node(node2, node2.stolen)
+        for i in range(4):
+            sched.mark_test_complete(node2, node2.sent[i])
+        assert node1.stolen == [14, 15]
+        sched.remove_pending_tests_from_node(node1, node1.stolen)
+        sched.mark_test_complete(node1, 12)
+        sched.mark_test_complete(node2, 14)
+        assert node2.stolen == list(range(12, 16))
+        assert node1.stolen == [14, 15]
+        assert sched.tests_finished
+
+    def test_steal_on_add_node(self, pytester: pytest.Pytester) -> None:
+        node = MockNode()
+        config = pytester.parseconfig("--tx=popen")
+        sched = WorkStealingScheduling(config)
+        sched.add_node(node)
+        collection = [f"test_workstealing.py::test_{i}" for i in range(5)]
+        sched.add_node_collection(node, collection)
+        assert sched.collection_is_completed
+        sched.schedule()
+        assert not sched.pending
+        sched.mark_test_complete(node, 0)
+        node2 = MockNode()
+        sched.add_node(node2)
+        sched.add_node_collection(node2, collection)
+        assert sched.collection_is_completed
+        sched.schedule()
+        assert node.stolen == [3, 4]
+        sched.remove_pending_tests_from_node(node, node.stolen)
+        sched.mark_test_complete(node, 1)
+        sched.mark_test_complete(node2, 3)
+        assert sched.tests_finished
+        assert node2.stolen == []
+
+    def test_schedule_fewer_tests_than_nodes(self, pytester: pytest.Pytester) -> None:
+        config = pytester.parseconfig("--tx=3*popen")
+        sched = WorkStealingScheduling(config)
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
+        node1, node2, node3 = sched.nodes
+        col = ["xyz"] * 2
+        sched.add_node_collection(node1, col)
+        sched.add_node_collection(node2, col)
+        sched.add_node_collection(node3, col)
+        sched.schedule()
+        assert node1.sent == []
+        assert node1.stolen == []
+        assert node2.sent == [0]
+        assert node2.stolen == []
+        assert node3.sent == [1]
+        assert node3.stolen == []
+        assert not sched.pending
+        assert sched.tests_finished
+
+    def test_schedule_fewer_than_two_tests_per_node(
+        self, pytester: pytest.Pytester
+    ) -> None:
+        config = pytester.parseconfig("--tx=3*popen")
+        sched = WorkStealingScheduling(config)
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
+        sched.add_node(MockNode())
+        node1, node2, node3 = sched.nodes
+        col = ["xyz"] * 5
+        sched.add_node_collection(node1, col)
+        sched.add_node_collection(node2, col)
+        sched.add_node_collection(node3, col)
+        sched.schedule()
+        assert node1.sent == [0]
+        assert node2.sent == [1, 2]
+        assert node3.sent == [3, 4]
+        assert not sched.pending
+        assert not sched.tests_finished
+        sched.mark_test_complete(node1, node1.sent[0])
+        sched.mark_test_complete(node2, node2.sent[0])
+        sched.mark_test_complete(node3, node3.sent[0])
+        sched.mark_test_complete(node3, node3.sent[1])
+        assert sched.tests_finished
+        assert node1.stolen == []
+        assert node2.stolen == []
+        assert node3.stolen == []
+
+    def test_add_remove_node(self, pytester: pytest.Pytester) -> None:
+        node = MockNode()
+        config = pytester.parseconfig("--tx=popen")
+        sched = WorkStealingScheduling(config)
+        sched.add_node(node)
+        collection = ["test_file.py::test_func"]
+        sched.add_node_collection(node, collection)
+        assert sched.collection_is_completed
+        sched.schedule()
+        assert not sched.pending
+        crashitem = sched.remove_node(node)
+        assert crashitem == collection[0]
+
+    def test_different_tests_collected(self, pytester: pytest.Pytester) -> None:
+        class CollectHook:
+            def __init__(self):
+                self.reports = []
+
+            def pytest_collectreport(self, report):
+                self.reports.append(report)
+
+        collect_hook = CollectHook()
+        config = pytester.parseconfig("--tx=2*popen")
+        config.pluginmanager.register(collect_hook, "collect_hook")
+        node1 = MockNode()
+        node2 = MockNode()
+        sched = WorkStealingScheduling(config)
+        sched.add_node(node1)
+        sched.add_node(node2)
+        sched.add_node_collection(node1, ["a.py::test_1"])
+        sched.add_node_collection(node2, ["a.py::test_2"])
+        sched.schedule()
+        assert len(collect_hook.reports) == 1
+        rep = collect_hook.reports[0]
+        assert "Different tests were collected between" in rep.longrepr
+
+
 class TestDistReporter:
     @pytest.mark.xfail
     def test_rsync_printing(self, pytester: pytest.Pytester, linecomp) -> None:
@@ -258,7 +479,7 @@ def test_report_collection_diff_equal() -> None:
 def test_default_max_worker_restart() -> None:
     class config:
         class option:
-            maxworkerrestart: Optional[str] = None
+            maxworkerrestart: str | None = None
             numprocesses: int = 0
 
     assert get_default_max_worker_restart(config) is None
@@ -312,3 +533,70 @@ def test_pytest_issue419(pytester: pytest.Pytester) -> None:
     reprec = pytester.inline_run("-n1")
     reprec.assertoutcome(passed=2)
     assert 0
+
+
+Created = WorkerStatus.Created
+Initialized = WorkerStatus.Initialized
+ReadyForCollection = WorkerStatus.ReadyForCollection
+CollectionDone = WorkerStatus.CollectionDone
+
+
+@pytest.mark.parametrize(
+    "status_and_items, expected",
+    [
+        (
+            [],
+            "",
+        ),
+        (
+            [(Created, 0)],
+            "created: 1/1 worker",
+        ),
+        (
+            [(Created, 0), (Created, 0)],
+            "created: 2/2 workers",
+        ),
+        (
+            [(Initialized, 0), (Created, 0)],
+            "initialized: 1/2 workers",
+        ),
+        (
+            [(Initialized, 0), (Initialized, 0)],
+            "initialized: 2/2 workers",
+        ),
+        (
+            [(ReadyForCollection, 0), (Created, 0)],
+            "ready: 1/2 workers",
+        ),
+        (
+            [(ReadyForCollection, 0), (ReadyForCollection, 0)],
+            "ready: 2/2 workers",
+        ),
+        (
+            [(CollectionDone, 12), (Created, 0)],
+            "collecting: 1/2 workers",
+        ),
+        (
+            [(CollectionDone, 12), (CollectionDone, 12)],
+            "2 workers [12 items]",
+        ),
+        (
+            [(CollectionDone, 1), (CollectionDone, 1)],
+            "2 workers [1 item]",
+        ),
+        (
+            [(CollectionDone, 1)],
+            "1 worker [1 item]",
+        ),
+        # Different number of tests collected will raise an error and should not happen in practice,
+        # but we test for it anyway.
+        (
+            [(CollectionDone, 1), (CollectionDone, 12)],
+            "2 workers [1 item]",
+        ),
+    ],
+)
+def test_get_workers_status_line(
+    status_and_items: Sequence[tuple[WorkerStatus, int]], expected: str
+) -> None:
+    assert get_workers_status_line(status_and_items) == expected

testing/test_plugin.py

@@ -12,7 +12,8 @@ import pytest
 @pytest.fixture
 def monkeypatch_3_cpus(monkeypatch: pytest.MonkeyPatch):
     """Make pytest-xdist believe the system has 3 CPUs"""
-    monkeypatch.setitem(sys.modules, "psutil", None)  # block import
+    # block import
+    monkeypatch.setitem(sys.modules, "psutil", None)  # type: ignore
     monkeypatch.delattr(os, "sched_getaffinity", raising=False)
     monkeypatch.setattr(os, "cpu_count", lambda: 3)
 
@@ -53,6 +54,8 @@ def test_auto_detect_cpus(
 ) -> None:
     from xdist.plugin import pytest_cmdline_main as check_options
 
+    monkeypatch.delenv("PYTEST_XDIST_AUTO_NUM_WORKERS", raising=False)
+
     with suppress(ImportError):
         import psutil
 
@@ -100,6 +103,7 @@ def test_auto_detect_cpus_psutil(
 
     psutil = pytest.importorskip("psutil")
 
+    monkeypatch.delenv("PYTEST_XDIST_AUTO_NUM_WORKERS", raising=False)
     monkeypatch.setattr(psutil, "cpu_count", lambda logical=True: 84 if logical else 42)
 
     config = pytester.parseconfigure("-nauto")
@@ -116,6 +120,8 @@ def test_auto_detect_cpus_os(
 ) -> None:
     from xdist.plugin import pytest_cmdline_main as check_options
 
+    monkeypatch.delenv("PYTEST_XDIST_AUTO_NUM_WORKERS", raising=False)
+
     config = pytester.parseconfigure("-nauto")
     check_options(config)
     assert config.getoption("numprocesses") == 3
@@ -177,6 +183,8 @@ def test_hook_auto_num_workers_none(
     # but we document it so let's test it.
     from xdist.plugin import pytest_cmdline_main as check_options
 
+    monkeypatch.delenv("PYTEST_XDIST_AUTO_NUM_WORKERS", raising=False)
+
     pytester.makeconftest(
         """
         def pytest_xdist_auto_num_workers():

testing/test_remote.py

@@ -25,15 +25,14 @@ class EventCall:
         self.name, self.kwargs = eventcall
 
     def __str__(self):
-        return "<EventCall {}(**{})>".format(self.name, self.kwargs)
+        return f"<EventCall {self.name}(**{self.kwargs})>"
 
 
 class WorkerSetup:
-    use_callback = False
-
     def __init__(self, request, pytester: pytest.Pytester) -> None:
         self.request = request
         self.pytester = pytester
+        self.use_callback = False
         self.events = Queue()  # type: ignore[var-annotated]
 
     def setup(self) -> None:
@@ -41,7 +40,7 @@ class WorkerSetup:
         # import os ; os.environ['EXECNET_DEBUG'] = "2"
         self.gateway = execnet.makegateway()
         self.config = config = self.pytester.parseconfigure()
-        putevent = self.use_callback and self.events.put or None
+        putevent = self.events.put if self.use_callback else None
 
         class DummyMananger:
             testrunuid = uuid.uuid4().hex
@@ -60,7 +59,7 @@ class WorkerSetup:
             ev = EventCall(data)
             if name is None or ev.name == name:
                 return ev
-            print("skipping {}".format(ev))
+            print(f"skipping {ev}")
 
     def sendcommand(self, name, **kwargs):
         self.slp.sendcommand(name, **kwargs)
@@ -221,6 +220,91 @@ class TestWorkerInteractor:
         ev = worker.popevent()
         assert ev.name == "errordown"
 
+    def test_steal_work(self, worker: WorkerSetup, unserialize_report) -> None:
+        worker.pytester.makepyfile(
+            """
+            import time
+            def test_func(): time.sleep(1)
+            def test_func2(): pass
+            def test_func3(): pass
+            def test_func4(): pass
+        """
+        )
+        worker.setup()
+        ev = worker.popevent("collectionfinish")
+        ids = ev.kwargs["ids"]
+        assert len(ids) == 4
+        worker.sendcommand("runtests_all")
+
+        # wait for test_func setup
+        ev = worker.popevent("testreport")
+        rep = unserialize_report(ev.kwargs["data"])
+        assert rep.nodeid.endswith("::test_func")
+        assert rep.when == "setup"
+
+        worker.sendcommand("steal", indices=[1, 2])
+        ev = worker.popevent("unscheduled")
+        assert ev.kwargs["indices"] == [2]
+
+        reports = [
+            ("test_func", "call"),
+            ("test_func", "teardown"),
+            ("test_func2", "setup"),
+            ("test_func2", "call"),
+            ("test_func2", "teardown"),
+        ]
+
+        for func, when in reports:
+            ev = worker.popevent("testreport")
+            rep = unserialize_report(ev.kwargs["data"])
+            assert rep.nodeid.endswith(f"::{func}")
+            assert rep.when == when
+
+        worker.sendcommand("shutdown")
+
+        for when in ["setup", "call", "teardown"]:
+            ev = worker.popevent("testreport")
+            rep = unserialize_report(ev.kwargs["data"])
+            assert rep.nodeid.endswith("::test_func4")
+            assert rep.when == when
+
+        ev = worker.popevent("workerfinished")
+        assert "workeroutput" in ev.kwargs
+
+    def test_steal_empty_queue(self, worker: WorkerSetup, unserialize_report) -> None:
+        worker.pytester.makepyfile(
+            """
+            def test_func(): pass
+            def test_func2(): pass
+        """
+        )
+        worker.setup()
+        ev = worker.popevent("collectionfinish")
+        ids = ev.kwargs["ids"]
+        assert len(ids) == 2
+        worker.sendcommand("runtests_all")
+
+        for when in ["setup", "call", "teardown"]:
+            ev = worker.popevent("testreport")
+            rep = unserialize_report(ev.kwargs["data"])
+            assert rep.nodeid.endswith("::test_func")
+            assert rep.when == when
+
+        worker.sendcommand("steal", indices=[0, 1])
+        ev = worker.popevent("unscheduled")
+        assert ev.kwargs["indices"] == []
+
+        worker.sendcommand("shutdown")
+
+        for when in ["setup", "call", "teardown"]:
+            ev = worker.popevent("testreport")
+            rep = unserialize_report(ev.kwargs["data"])
+            assert rep.nodeid.endswith("::test_func2")
+            assert rep.when == when
+
+        ev = worker.popevent("workerfinished")
+        assert "workeroutput" in ev.kwargs
+
 
 def test_remote_env_vars(pytester: pytest.Pytester) -> None:
     pytester.makepyfile(

testing/test_workermanage.py

@@ -374,7 +374,7 @@ def test_unserialize_warning_msg(w_cls):
         v1 = getattr(w_msg, k)
         v2 = getattr(w_msg2, k)
         if k == "message":
-            assert type(v1) == type(v2)
+            assert type(v1) is type(v2)
             assert v1.args == v2.args
         else:
             assert v1 == v2

tox.ini

@@ -1,10 +1,10 @@
 [tox]
 envlist=
   linting
-  py{36,37,38,39,310}-pytestlatest
-  py38-pytestmain
-  py38-psutil
-  py38-setproctitle
+  py{37,38,39,310,311,312}-pytestlatest
+  py310-pytestmain
+  py310-psutil
+  py310-setproctitle
 isolated_build = true
 [testenv]
 extras = testing
@@ -14,14 +14,14 @@ deps =
 commands=
   pytest {posargs}
 
-[testenv:py38-psutil]
+[testenv:py310-psutil]
 extras =
   testing
   psutil
 commands =
   pytest {posargs:-k psutil}
 
-[testenv:py38-setproctitle]
+[testenv:py310-setproctitle]
 extras =
   testing
   setproctitle
@@ -40,7 +40,7 @@ commands = pre-commit run --all-files --show-diff-on-failure
 [testenv:release]
 changedir=
 description = do a release, required posarg of the version number
-basepython = python3.7
+basepython = python3.10
 skipsdist = True
 usedevelop = True
 passenv = *
@@ -50,7 +50,7 @@ commands =
   towncrier build --version {posargs} --yes
 
 [testenv:docs]
-basepython = python3
+basepython = python3.10
 usedevelop = True
 deps =
     sphinx