diff --git a/newsfragments/3534.bugfix.rst b/newsfragments/3534.bugfix.rst new file mode 100644 index 0000000000..f28c7949be --- /dev/null +++ b/newsfragments/3534.bugfix.rst @@ -0,0 +1,3 @@ +Tasks moved into an already broken `trio.lowlevel.ParkingLot` by +`~trio.lowlevel.ParkingLot.repark` or `~trio.lowlevel.ParkingLot.repark_all` +now raise `trio.BrokenResourceError` instead of remaining blocked. diff --git a/src/trio/_core/_parking_lot.py b/src/trio/_core/_parking_lot.py index ddf6276117..127bca280e 100644 --- a/src/trio/_core/_parking_lot.py +++ b/src/trio/_core/_parking_lot.py @@ -256,6 +256,9 @@ async def main(): If there are fewer than ``count`` tasks parked, then reparks as many tasks as are available and then returns successfully. + If ``new_lot`` is broken, the moved tasks' calls to :meth:`park` + raise :exc:`trio.BrokenResourceError` instead. + Args: new_lot (ParkingLot): the parking lot to move tasks to. count (int|math.inf): the number of tasks to move. @@ -264,8 +267,18 @@ async def main(): if not isinstance(new_lot, ParkingLot): raise TypeError("new_lot must be a ParkingLot") for task in self._pop_several(count): - new_lot._parked[task] = None - task.custom_sleep_data = new_lot + if new_lot.broken_by: + _core.reschedule( + task, + outcome.Error( + _core.BrokenResourceError( + f"Attempted to repark in parking lot broken by {new_lot.broken_by}", + ), + ), + ) + else: + new_lot._parked[task] = None + task.custom_sleep_data = new_lot def repark_all(self, new_lot: ParkingLot) -> None: """Move all parked tasks from one :class:`ParkingLot` object to diff --git a/src/trio/_core/_tests/test_parking_lot.py b/src/trio/_core/_tests/test_parking_lot.py index 55c5144f91..7d8f684754 100644 --- a/src/trio/_core/_tests/test_parking_lot.py +++ b/src/trio/_core/_tests/test_parking_lot.py @@ -1,5 +1,6 @@ from __future__ import annotations +import math import re from typing import TypeVar @@ -225,6 +226,50 @@ async def test_parking_lot_repark_with_count() -> None: lot1.unpark_all() +@pytest.mark.parametrize( + ("count", "expected_broken"), + [(0, 0), (1, 1), (2, 2), (5, 3), (math.inf, 3), (None, 3)], +) +async def test_parking_lot_repark_to_broken_lot( + count: float | None, + expected_broken: int, +) -> None: + """Reparked tasks fail instead of waiting forever in a broken destination.""" + source = ParkingLot() + target = ParkingLot() + target.break_lot() + broken: set[int] = set() + woken: set[int] = set() + + async def waiter(i: int) -> None: + try: + await source.park() + except _core.BrokenResourceError: + broken.add(i) + else: + woken.add(i) + + async with _core.open_nursery() as nursery: + for i in range(3): + nursery.start_soon(waiter, i) + await wait_all_tasks_blocked() + + if count is None: + source.repark_all(target) + else: + source.repark(target, count=count) + await wait_all_tasks_blocked() + + assert broken == set(range(expected_broken)) + assert not woken + assert len(target) == 0 + assert len(source) == 3 - expected_broken + + source.unpark_all() + + assert woken == set(range(expected_broken, 3)) + + async def dummy_task( task_status: _core.TaskStatus[_core.Task] = trio.TASK_STATUS_IGNORED, ) -> None: