diff --git a/changelog/1313.bugfix.rst b/changelog/1313.bugfix.rst new file mode 100644 index 00000000..c441bfef --- /dev/null +++ b/changelog/1313.bugfix.rst @@ -0,0 +1 @@ +Fixed a scheduler deadlock in ``--dist loadscope``, ``loadfile`` and ``loadgroup`` where a worker crash caused already-completed work units to be requeued, making the session hang near completion with idle workers (and any replacement workers being assigned empty work units). diff --git a/src/xdist/scheduler/loadscope.py b/src/xdist/scheduler/loadscope.py index 695a4d45..16b98feb 100644 --- a/src/xdist/scheduler/loadscope.py +++ b/src/xdist/scheduler/loadscope.py @@ -249,7 +249,15 @@ def mark_test_complete( nodeid = self.registered_collections[node][item_index] scope = self._split_scope(nodeid) - self.assigned_work[node][scope][nodeid] = True + work_unit = self.assigned_work[node].get(scope) + if work_unit is not None: + work_unit[nodeid] = True + # Drop the work unit once fully completed. Otherwise, if this + # node crashes later, remove_node() would requeue the completed + # unit and a replacement node would be assigned an empty unit, + # deadlocking the scheduler (#1313). + if all(work_unit.values()): + del self.assigned_work[node][scope] self._reschedule(node) def mark_test_pending(self, item: str) -> NoReturn: @@ -266,8 +274,19 @@ def _assign_work_unit(self, node: WorkerController) -> None: """Assign a work unit to a node.""" assert self.workqueue - # Grab a unit of work - scope, work_unit = self.workqueue.popitem(last=False) + # Grab a unit of work, discarding any unit with no pending tests + # (assigning one would deadlock the scheduler: the node would be + # sent an empty "runtests" command and the unit could never be + # marked complete, see #1313) + while True: + scope, work_unit = self.workqueue.popitem(last=False) + if not all(work_unit.values()): + break + if not self.workqueue: + # Only fully-completed units were queued: nothing to assign, + # and the workqueue is now empty + node.shutdown() + return # Keep track of the assigned work assigned_to_node = self.assigned_work.setdefault(node, {}) diff --git a/testing/test_dsession.py b/testing/test_dsession.py index 680b7ae0..28d6430e 100644 --- a/testing/test_dsession.py +++ b/testing/test_dsession.py @@ -15,6 +15,7 @@ from xdist.report import report_collection_diff from xdist.scheduler import EachScheduling from xdist.scheduler import LoadScheduling +from xdist.scheduler import LoadScopeScheduling from xdist.scheduler import WorkStealingScheduling from xdist.workermanage import WorkerController @@ -290,6 +291,85 @@ def pytest_collectreport(self, report: pytest.CollectReport) -> None: assert "Different tests were collected between" in rep.longrepr +class TestLoadScopeScheduling: + def test_replacement_node_after_crash_gets_pending_work( + self, pytester: pytest.Pytester + ) -> None: + """A node crashing after completing scopes must not wedge the scheduler. + + Completed work units used to be requeued by remove_node(), so a + replacement node was assigned an empty work unit ("runtests" with no + indices) that could never be marked complete, deadlocking the whole + session (#1313). + """ + config = pytester.parseconfig("--tx=2*popen", "--dist=loadscope") + sched = LoadScopeScheduling(config) + node1, node2 = MockNode(), MockNode() + sched.add_node(node1) + sched.add_node(node2) + collection = [f"{scope}.py::test_{i}" for scope in "abc" for i in range(4)] + sched.add_node_collection(node1, collection) + sched.add_node_collection(node2, collection) + sched.schedule() + assert sched.collection_is_completed + # Initial distribution: one scope per node, c.py still in the queue + assert node1.sent == [0, 1, 2, 3] + assert node2.sent == [4, 5, 6, 7] + assert list(sched.workqueue) == ["c.py"] + + # node1 completes all of a.py; c.py gets assigned to it along the way + for i in range(4): + sched.mark_test_complete(node1, i) + assert node1.sent == [0, 1, 2, 3, 8, 9, 10, 11] + assert not sched.workqueue + + # node2 completes all of b.py and is told to shut down + for i in range(4, 8): + sched.mark_test_complete(node2, i) + assert node2.shutting_down + + # node1 completes c.py partially, then crashes on index 10 + sched.mark_test_complete(node1, 8) + sched.mark_test_complete(node1, 9) + crashitem = sched.remove_node(node1) + assert crashitem == "c.py::test_2" + + # Only the scope with pending tests may be requeued: the completed + # a.py must not come back + assert list(sched.workqueue) == ["c.py"] + + # A replacement node must be assigned the pending tests, not an + # empty work unit + node3 = MockNode() + sched.add_node(node3) + sched.add_node_collection(node3, collection) + sched.schedule() + assert node3.sent == [10, 11] + sched.mark_test_complete(node3, 10) + sched.mark_test_complete(node3, 11) + assert sched.tests_finished + + def test_remove_node_with_only_completed_work( + self, pytester: pytest.Pytester + ) -> None: + """A node crashing right after finishing its work requeues nothing.""" + config = pytester.parseconfig("--tx=2*popen", "--dist=loadscope") + sched = LoadScopeScheduling(config) + node1, node2 = MockNode(), MockNode() + sched.add_node(node1) + sched.add_node(node2) + collection = [f"{scope}.py::test_{i}" for scope in "ab" for i in range(4)] + sched.add_node_collection(node1, collection) + sched.add_node_collection(node2, collection) + sched.schedule() + for i in range(4): + sched.mark_test_complete(node1, i) + assert node1.shutting_down + crashitem = sched.remove_node(node1) + assert crashitem is None + assert not sched.workqueue + + class TestWorkStealingScheduling: def test_ideal_case(self, pytester: pytest.Pytester) -> None: config = pytester.parseconfig("--tx=2*popen")