diff --git a/changelog/784.bugfix.rst b/changelog/784.bugfix.rst new file mode 100644 index 00000000..b918c98b --- /dev/null +++ b/changelog/784.bugfix.rst @@ -0,0 +1 @@ +Fixed ``--dist=loadscope``, ``loadfile`` and ``loadgroup`` hanging indefinitely after a worker crash. The crashed worker's completed work units were returned to the work queue and could later be assigned as empty commands that a node can never report against, and a node re-scheduled a single test never started it, since a worker holds its last test in reserve until it is sent further work or told to shut down. The crashed test itself is also no longer re-executed by the replacement worker: it is reported as crashed, matching ``--dist=load``. diff --git a/src/xdist/scheduler/loadscope.py b/src/xdist/scheduler/loadscope.py index 73162dcd..8cf1de15 100644 --- a/src/xdist/scheduler/loadscope.py +++ b/src/xdist/scheduler/loadscope.py @@ -187,6 +187,11 @@ def remove_node(self, node: WorkerController) -> str | None: for nodeid, completed in work_unit.items(): if not completed: crashitem = nodeid + # The crashed test is reported as failed by + # handle_crashitem, so mark it complete lest it be + # rescheduled and crash the next node too. This matches + # LoadScheduling, which pops the crashed item. + work_unit[nodeid] = True break else: continue @@ -196,8 +201,16 @@ def remove_node(self, node: WorkerController) -> str | None: "Unable to identify crashitem on a workload with pending items" ) - # Made uncompleted work unit available again - self.workqueue.update(workload) + # Make uncompleted work units available again. Work units in which + # every test has completed are dropped rather than re-queued: assigning + # one to a node would send it an empty "runtests" command, after which + # the node would never produce the test report that drives scheduling + # onwards, hanging the run (#784). + self.workqueue.update( + (scope, work_unit) + for scope, work_unit in workload.items() + if not all(work_unit.values()) + ) for node in self.assigned_work: self._reschedule(node) @@ -320,20 +333,32 @@ def _reschedule(self, node: WorkerController) -> None: if node.shutting_down: return - # Check that more work is available - if not self.workqueue: - node.shutdown() - return + while True: + # Check that more work is available + if not self.workqueue: + node.shutdown() + return - self.log("Number of units waiting for node:", len(self.workqueue)) + self.log("Number of units waiting for node:", len(self.workqueue)) - # Check that the node is almost depleted of work - # 2: Heuristic of minimum tests to enqueue more work - if self._pending_of(self.assigned_work[node]) > 2: - return + # Check that the node is almost depleted of work + # 2: Heuristic of minimum tests to enqueue more work + if self._pending_of(self.assigned_work[node]) > 2: + return + + # Pop one unit of work and assign it + self._assign_work_unit(node) - # Pop one unit of work and assign it - self._assign_work_unit(node) + # A worker holds its final pending test in reserve, not starting + # it until further work or a shutdown notice arrives, so a node + # must hold at least two pending tests to make progress (see the + # note about #277 in .schedule()). Keep assigning until it does, + # since only a test report drives scheduling onwards and a node + # left below that threshold never produces one; if the workqueue + # runs dry first, the loop shuts the node down, which is what + # releases its reserved test (#784). + if self._pending_of(self.assigned_work[node]) >= 2: + return def schedule(self) -> None: """Initiate distribution of the test collection. diff --git a/testing/acceptance_test.py b/testing/acceptance_test.py index 814c8c09..d909627b 100644 --- a/testing/acceptance_test.py +++ b/testing/acceptance_test.py @@ -1032,6 +1032,37 @@ def test(i): os._exit(1) ) assert "INTERNALERROR" not in res.stdout.str() + def test_loadfile_crashed_worker(self, pytester: pytest.Pytester) -> None: + """A crashed worker's workload is rescheduled without hanging the + run (#784). + + The worker crashes after completing the whole of test_a.py. Its + re-queued workload must not include the completed test_a.py work + unit (an empty command the replacement worker could never report + against), the crashed test must not run a second time, and the + replacement worker must be sent enough tests to make progress, since + a worker holds its last test in reserve until told what follows. + """ + pytester.makepyfile( + test_a=""" + def test_pass_1(): pass + def test_pass_2(): pass + """, + test_b=""" + import os + def test_crash(): os._exit(1) + def test_after(): pass + """, + ) + res = pytester.runpytest_subprocess("-n1", "--dist=loadfile", "-v", timeout=120) + res.stdout.fnmatch_lines( + [ + "replacing crashed worker gw*", + "worker*crashed while running*test_crash*", + "*1 failed*3 passed*", + ] + ) + def test_max_worker_restart_die(self, pytester: pytest.Pytester) -> None: f = pytester.makepyfile( """ diff --git a/testing/test_dsession.py b/testing/test_dsession.py index 680b7ae0..ff84522d 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 @@ -632,3 +633,85 @@ 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 + + +class TestLoadScopeScheduling: + def test_remove_node_requeues_only_pending_work_units( + self, pytester: pytest.Pytester + ) -> None: + """A crashed node's completed work units must not return to the + workqueue: assigning one to a node sends an empty "runtests" command + against which the node can never report, hanging the run (#784). The + crashed test itself must not be re-queued either, since it is + reported as failed by handle_crashitem.""" + 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"test_{m}.py::test_{i}" for m in "abcdef" for i in (1, 2)] + sched.add_node_collection(node1, collection) + sched.add_node_collection(node2, collection) + sched.schedule() + # node1 was assigned test_a.py and test_c.py. + assert node1.sent == [0, 1, 4, 5] + + # node1 completes the whole of test_a.py, picking up test_e.py. + sched.mark_test_complete(node1, 0) + sched.mark_test_complete(node1, 1) + assert node1.sent == [0, 1, 4, 5, 8, 9] + + # node1 crashes in test_c.py::test_1. + crashitem = sched.remove_node(node1) + assert crashitem == "test_c.py::test_1" + + # The completed test_a.py unit is gone for good, the units with + # pending tests are re-queued, and the crashed test is marked + # completed so that it is not run a second time. + assert list(sched.workqueue.keys()) == ["test_f.py", "test_c.py", "test_e.py"] + for work_unit in sched.workqueue.values(): + assert not all(work_unit.values()) + assert sched.workqueue["test_c.py"] == { + "test_c.py::test_1": True, + "test_c.py::test_2": False, + } + + def test_node_is_topped_up_until_it_can_report( + self, pytester: pytest.Pytester + ) -> None: + """A node must be given at least two pending tests, because a worker + does not start its last test until it is sent further work or told to + shut down. A replacement node given a single-test work unit would + otherwise never report, and a report is the only thing that drives + scheduling onwards (#784).""" + config = pytester.parseconfig("--tx=2*popen", "--dist=loadscope") + sched = LoadScopeScheduling(config) + node1, node2 = MockNode(), MockNode() + sched.add_node(node1) + sched.add_node(node2) + collection = [ + "test_a.py::test_1", + "test_a.py::test_2", + "test_a.py::test_3", + "test_b.py::test_1", + "test_b.py::test_2", + "test_b.py::test_3", + "test_c.py::test_1", + "test_d.py::test_1", + "test_e.py::test_1", + ] + sched.add_node_collection(node1, collection) + sched.add_node_collection(node2, collection) + sched.schedule() + assert node1.sent == [0, 1, 2] + assert node2.sent == [3, 4, 5] + + # A replacement node arrives while single-test scopes are queued. + node3 = MockNode() + sched.add_node(node3) + sched.add_node_collection(node3, collection) + sched.schedule() + + # One test is not enough to make progress: it must receive two. + assert node3.sent == [6, 7] + assert not node3.shutting_down