diff --git a/reboot/aio/state_managers.py b/reboot/aio/state_managers.py index 74fa62e1..f1856ff9 100644 --- a/reboot/aio/state_managers.py +++ b/reboot/aio/state_managers.py @@ -712,8 +712,21 @@ async def claim_ownership( transaction_ids = context.transaction_ids assert transaction_ids is not None + # Presume a deadlock only with an owner that has owned this + # state for a whole grace period while we waited; an owner + # that changed in between is a sign the state is moving + # between nested transactions, and we keep waiting. + previous_owner_ids = list(self.owner_ids) + def on_grace() -> None: - self._abort_if_presumed_sibling_deadlock(transaction_ids) + nonlocal previous_owner_ids + current_owner_ids = list(self.owner_ids) + if current_owner_ids == previous_owner_ids: + self._abort_if_presumed_sibling_deadlock( + transaction_ids, + current_owner_ids, + ) + previous_owner_ids = current_owner_ids await self.wait_ownership( lambda: self.is_claimable_by(transaction_ids), @@ -742,25 +755,28 @@ def on_grace() -> None: def _abort_if_presumed_sibling_deadlock( self, transaction_ids: list[uuid.UUID], + owner_ids: list[uuid.UUID], ) -> None: """Raises `SystemAborted(NestedTransactionShouldRetry(...))` - when the nested transaction owning this state is an older - sibling of a nested transaction among `transaction_ids`, so - that the younger sibling rolls back, ownership of what it - claimed returns to the transaction that started it, and the - older sibling proceeds: the younger of two siblings waiting - on each other's states is always the one to go. A state - owned by a younger sibling, or by a descendant of the - caller, keeps the caller waiting. + when `owner_ids`, the nested transactions that have owned + this state for a whole grace period while a call of + `transaction_ids` waited, end in an older sibling of a + nested transaction among `transaction_ids`, so that the + younger sibling rolls back, ownership of what it claimed + returns to the transaction that started it, and the older + sibling proceeds: the younger of two siblings waiting on + each other's states is always the one to go. A state owned + by a younger sibling, or by a descendant of the caller, + keeps the caller waiting. """ nested_transaction_id = presumed_deadlocked_nested_transaction( transaction_ids, - self.owner_ids, + owner_ids, ) if nested_transaction_id is None: return level = transaction_ids.index(nested_transaction_id) - owner_id = self.owner_ids[level] + owner_id = owner_ids[level] grace_ms = TRANSACTION_DEADLOCK_GRACE // timedelta(milliseconds=1) message = ( f"Nested transaction {nested_transaction_id} waited longer " @@ -1695,9 +1711,12 @@ class Lock: promoted to `exclusive`. A caller can upgrade and skip other `exclusive` waiters, preserving the upgrading transaction's shared-consistent view of state. However, at most one upgrade - may be pending per lock; a second `upgrade(...)` raises - `SystemAborted(Unavailable())` immediately to avoid the deadlock - where two shared holders both want to upgrade. + may be pending per lock; a second `upgrade(...)` aborts + immediately, since two shared holders that both want to upgrade + are each waiting for the other's shared hold to go, which is a + deadlock by construction: with `SystemAborted(TransactionShouldRetry)` + carrying the transaction's age when the upgrader passed its + participant, and `SystemAborted(Unavailable())` otherwise. - `downgrade()` is the inverse: a caller that holds `exclusive` is demoted to `shared`, granting any queued `shared` waiters that @@ -2072,15 +2091,38 @@ async def _wait( if upgrade: # If we've already got a upgrader that is waiting we fail # fast because otherwise each upgrader would sit on their - # shared hold and we'd deadlock. + # shared hold and we'd deadlock. A transaction is asked to + # retry the way a presumed deadlock asks it, carrying its + # age, so that its retry skips the backoff and is not the + # youngest again; a caller without a participant is asked + # to retry as `Unavailable`. if self._upgrader is not None: + if transaction is None: + raise SystemAborted( + Unavailable(), + message=( + "Cannot upgrade shared lock to exclusive: " + "another transaction is already upgrading " + "the same state; retry the transaction." + ), + ) + pending = self._upgrader.transaction + message = ( + f"Transaction {transaction.root_id} cannot upgrade its " + "shared hold to exclusive: " + ( + f"transaction {pending.root_id} (age {pending.age})" + if pending is not None else "another transaction" + ) + " is already upgrading the same state, and each " + "would wait for the other's shared hold to go; " + "aborting so that it proceeds. Retry required." + ) + logger.warning(message) raise SystemAborted( - Unavailable(), - message=( - "Cannot upgrade shared lock to exclusive: " - "another transaction is already upgrading " - "the same state; retry the transaction." + TransactionShouldRetry( + reason=TransactionShouldRetry.PRESUMED_DEADLOCK, + retry_age=str(transaction.age), ), + message=message, ) self._upgrader = waiter else: @@ -2850,23 +2892,67 @@ def __init__( def latest_timestamp_ms(self) -> Optional[int]: return self._latest_timestamp_ms + def _presume_deadlock_on_grace( + self, + state_type: StateTypeName, + state_ref: StateRef, + transaction: StateManager.Transaction, + ) -> Callable[[], None]: + """The `on_grace` hook for `transaction` waiting on the lock of + `(state_type, state_ref)`: each time a grace period elapses it + presumes a deadlock with, and aborts `transaction` for, an + older holder that has held the lock since the previous grace + period elapsed, or since the wait began. A holder that arrived + in between is not a deadlock but the lock changing hands, and + `transaction`, queued behind it, keeps waiting. + + TODO: when every state in a cycle lives on this server, the + deadlock can be proven rather than presumed, and at once: record + on each participant the state it is waiting on (here and in + `claim_ownership`), and when a wait begins walk from the lock's + holders through what they wait on, and so on, looking for a + path back to `transaction`. A cycle found that way aborts its + youngest transaction immediately, with no grace period and no + false positive; a wait that leaves this server ends the walk + and falls back to the grace check. Most cycles in tests and in + small deployments are local, so this would resolve them in + microseconds instead of a grace period. + """ + lock = self._locks[state_type][state_ref] + previous_holders = set(lock.holders) + + def on_grace() -> None: + nonlocal previous_holders + current_holders = set(lock.holders) + self._abort_if_presumed_deadlock( + state_type, + state_ref, + transaction, + current_holders & previous_holders, + ) + previous_holders = current_holders + + return on_grace + def _abort_if_presumed_deadlock( self, state_type: StateTypeName, state_ref: StateRef, transaction: StateManager.Transaction, + holders: set[StateManager.Transaction], ) -> None: """Raises `SystemAborted(TransactionShouldRetry(...))`, with - reason `PRESUMED_DEADLOCK`, when the lock on `(state_type, - state_ref)` is held by a transaction older than `transaction`, - so that `transaction` aborts and the older one proceeds: the - younger of two transactions waiting on each other's states is - always the one to go. A lock held only by younger transactions, - or by a plain reader or writer outside any transaction (which - never waits on anything and so does not identify itself to the - lock), keeps `transaction` waiting. + reason `PRESUMED_DEADLOCK`, when `holders`, the transactions + that have held the lock on `(state_type, state_ref)` for a + whole grace period while `transaction` waited, include one + older than `transaction`, so that `transaction` aborts and the + older one proceeds: the younger of two transactions waiting on + each other's states is always the one to go. A lock held only + by younger transactions, or by a plain reader or writer outside + any transaction (which never waits on anything and so does not + identify itself to the lock), keeps `transaction` waiting. """ - for holder in self._locks[state_type][state_ref].holders: + for holder in holders: # An upgrader is itself among the holders through the # shared hold it upgrades from. if holder is transaction: @@ -4272,22 +4358,21 @@ async def _upgrade_lock( `(state_type, state_ref)` to exclusive, aborting `transaction` with `TransactionShouldRetry` (presumed deadlock) instead if it has waited longer than the grace period on a shared holder that - is older. + is older, or at once if another transaction's upgrade is + already pending, since the two would each wait for the other's + shared hold to go. """ assert transaction.mode == Lock.Mode.SHARED - def on_grace() -> None: - self._abort_if_presumed_deadlock( - state_type, - state_ref, - transaction, - ) - await self._locks[state_type][state_ref].upgrade( deadline=LOCK_ACQUIRE_DEADLINE_DEFAULT, transaction=transaction, grace=TRANSACTION_DEADLOCK_GRACE, - on_grace=on_grace, + on_grace=self._presume_deadlock_on_grace( + state_type, + state_ref, + transaction, + ), ) transaction.mode = Lock.Mode.EXCLUSIVE @@ -5441,12 +5526,11 @@ async def _transaction_participant_start( # success / failure we resolve `transaction.acquired_lock` # so concurrent callers on the same transaction can # proceed (or propagate our failure). - def on_grace() -> None: - self._abort_if_presumed_deadlock( - state_type, - state_ref, - transaction, - ) + on_grace = self._presume_deadlock_on_grace( + state_type, + state_ref, + transaction, + ) try: if transaction.mode == Lock.Mode.SHARED: diff --git a/tests/reboot/state_manager_tests.py b/tests/reboot/state_manager_tests.py index b8db8840..c3cb1c0e 100644 --- a/tests/reboot/state_manager_tests.py +++ b/tests/reboot/state_manager_tests.py @@ -1112,7 +1112,8 @@ async def test_second_upgrade_fast_fails(self) -> None: await asyncio.sleep(0) self.assertFalse(first_upgrade_task.done()) - # The second upgrade attempt must fail immediately. + # The second upgrade attempt must fail immediately; without a + # participant it is asked to retry as `Unavailable`. with self.assertRaises(SystemAborted) as aborted: await lock.upgrade(deadline=None) self.assertEqual(type(aborted.exception.error), Unavailable) @@ -1123,6 +1124,47 @@ async def test_second_upgrade_fast_fails(self) -> None: await asyncio.wait_for(first_upgrade_task, timeout=1.0) lock.release_exclusive() + async def test_second_upgrade_by_a_transaction_retries_with_its_age( + self, + ) -> None: + """A transaction whose upgrade meets another pending upgrade is + asked to retry the way a presumed deadlock asks it, carrying its + age, since the two would each wait for the other's shared hold. + """ + lock = Lock() + # Stand-ins for the participants; the lock only reports them. + first = unittest.mock.Mock(spec=StateManager.Transaction) + first.root_id = uuid7(timestamp_ms=1000) + first.age = first.root_id + second = unittest.mock.Mock(spec=StateManager.Transaction) + second.root_id = uuid7(timestamp_ms=2000) + second.age = second.root_id + await lock.acquire_shared(deadline=None, transaction=first) + await lock.acquire_shared(deadline=None, transaction=second) + + first_upgrade_task = asyncio.create_task( + lock.upgrade(deadline=None, transaction=first) + ) + await asyncio.sleep(0) + self.assertFalse(first_upgrade_task.done()) + + with self.assertRaises(SystemAborted) as aborted: + await lock.upgrade(deadline=None, transaction=second) + error = aborted.exception.error + assert isinstance(error, TransactionShouldRetry) + self.assertEqual( + error.reason, TransactionShouldRetry.PRESUMED_DEADLOCK + ) + self.assertEqual(error.retry_age, str(second.age)) + assert aborted.exception.message is not None + self.assertIn(str(first.root_id), aborted.exception.message) + + # The second still holds shared; releasing it lets the first + # upgrade complete, and then the first releases exclusive. + lock.release_shared(transaction=second) + await asyncio.wait_for(first_upgrade_task, timeout=1.0) + lock.release_exclusive(transaction=first) + async def test_downgrade_when_sole_holder(self) -> None: lock = Lock() await lock.acquire_exclusive(deadline=None) diff --git a/tests/reboot/transaction_tests.py b/tests/reboot/transaction_tests.py index 10687555..c46d8a6b 100644 --- a/tests/reboot/transaction_tests.py +++ b/tests/reboot/transaction_tests.py @@ -30,7 +30,12 @@ ) from reboot.aio.internals.contextvars import Servicing, _servicing from reboot.aio.resolvers import NoResolver -from reboot.aio.state_managers import Lock, SidecarStateManager, StateManager +from reboot.aio.state_managers import ( + TRANSACTION_DEADLOCK_GRACE, + Lock, + SidecarStateManager, + StateManager, +) from reboot.aio.stubs import ( NestedTransactionUnaryRetriedCall, Stub, @@ -744,6 +749,130 @@ async def Transaction( (_, first_exited), (second_entered, _) = runs self.assertGreaterEqual(second_entered, first_exited) + async def test_waiting_behind_a_moving_lock_is_not_a_deadlock( + self, + ) -> None: + """A younger transaction queued behind a run of older + transactions that each hold a state's lock briefly waits far + longer than the grace period, yet is never presumed + deadlocked: at each grace check the holder is a different + transaction from the one at the previous check, which means the + lock is changing hands, not stuck. Only a holder that has held + the lock for a whole grace period is a presumed deadlock. + """ + hold = 4 * TRANSACTION_DEADLOCK_GRACE / 5 + + class TargetServicer(GeneralServicer): + + def authorizer(self): + return allow() + + async def ConstructorWriter( + self, + context: WriterContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + return GeneralResponse() + + # A root that calls the target's exclusive transaction. + async def ConstructorTransaction( + self, + context: TransactionContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + await General.ref(request.content["target"]).Transaction( + context, + content={"hold": request.content["hold"]}, + ) + return GeneralResponse() + + # The target's transaction holds its lock for `hold`. + async def Transaction( + self, + context: TransactionContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + await asyncio.sleep(float(request.content["hold"])) + return GeneralResponse() + + # Record every `TransactionShouldRetry` the client parses, to + # prove that nothing was presumed deadlocked. + transaction_should_retry = UnaryRetriedCall._transaction_should_retry + should_retries: list[errors_pb2.TransactionShouldRetry] = [] + + async def mock_transaction_should_retry(unary_retried_call): + should_retry = await transaction_should_retry(unary_retried_call) + if should_retry is not None: + should_retries.append(should_retry) + return should_retry + + with mock.patch( + 'reboot.aio.stubs.UnaryRetriedCall._transaction_should_retry', + mock_transaction_should_retry, + ): + await self.rbt.up( + Application(servicers=[TargetServicer]), + # Run each body once, so that the holders queue in the + # order the roots were started. + effect_validation=EffectValidation.DISABLED, + ) + context = self.rbt.create_external_context(name=self.id()) + + await General.ConstructorWriter(context, 'target') + + # Four older roots, started in order so that each queues + # behind the previous one, together hold the target's lock + # for several grace periods. + older = [] + for index in range(4): + older.append( + asyncio.create_task( + General.ConstructorTransaction( + context, + f'older-{index}', + content={ + "target": "target", + "hold": str(hold.total_seconds()), + }, + ) + ) + ) + # Let this root reach the target before the next starts, + # so that the target's queue is in root order. + await asyncio.sleep(0.05) + + # The youngest root queues last and holds the lock for no + # time at all once it gets it. + started = time.monotonic() + await General.ConstructorTransaction( + context, + 'youngest', + content={ + "target": "target", + "hold": "0" + }, + ) + waited = time.monotonic() - started + await asyncio.gather(*older) + + # The youngest waited out more than one grace period behind the + # older holders, and nothing was presumed deadlocked. A retry + # for `RESTART_DETECTED` can still happen right after the server + # starts, when a transaction's id predates the server's recovery + # timestamp; that is unrelated to the lock. + self.assertGreater(waited, TRANSACTION_DEADLOCK_GRACE.total_seconds()) + self.assertEqual( + [ + should_retry for should_retry in should_retries + if should_retry.reason == + errors_pb2.TransactionShouldRetry.PRESUMED_DEADLOCK + ], + [], + ) + async def test_shared_transactions_on_one_state_overlap(self) -> None: """Two transactions declared `shared`, each nested in its own root so that neither carries an idempotency key, take the same @@ -817,6 +946,131 @@ async def SharedTransaction( self.assertTrue(both_entered.is_set()) + async def test_second_upgrader_retries_carrying_its_age(self) -> None: + """Two `shared` transactions on one state, each nested in its own + root, both write the state and so both upgrade. The two would + each wait for the other's shared hold to go, so the second to + upgrade aborts at once with `TransactionShouldRetry`, reason + `PRESUMED_DEADLOCK`, carrying its age, and its retry succeeds. + """ + # Each body waits for the other to have entered too, so that + # both hold the state shared when they go to write it. + both_entered = asyncio.Event() + entered = 0 + + class TargetServicer(GeneralServicer): + + def authorizer(self): + return allow() + + async def ConstructorWriter( + self, + context: WriterContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + return GeneralResponse() + + # A root that calls the target's shared transaction. + async def ConstructorTransaction( + self, + context: TransactionContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + await General.ref(request.content["target"] + ).SharedTransaction(context) + return GeneralResponse() + + # Reads its state alongside the other, then writes it. + async def SharedTransaction( + self, + context: TransactionContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + nonlocal entered + entered += 1 + if entered == 2: + both_entered.set() + await asyncio.wait_for(both_entered.wait(), timeout=10) + state.content["writes"] = str( + int(state.content.get("writes", "0")) + 1 + ) + return GeneralResponse() + + async def Reader( + self, + context: ReaderContext, + state: General.State, + request: GeneralRequest, + ) -> GeneralResponse: + return GeneralResponse(content=state.content) + + # Record every `TransactionShouldRetry` the client parses. + transaction_should_retry = UnaryRetriedCall._transaction_should_retry + should_retries: list[tuple[UnaryRetriedCall, + errors_pb2.TransactionShouldRetry]] = [] + + async def mock_transaction_should_retry(unary_retried_call): + should_retry = await transaction_should_retry(unary_retried_call) + if should_retry is not None: + should_retries.append((unary_retried_call, should_retry)) + return should_retry + + with mock.patch( + 'reboot.aio.stubs.UnaryRetriedCall._transaction_should_retry', + mock_transaction_should_retry, + ): + await self.rbt.up( + Application(servicers=[TargetServicer]), + # Run each body once, so that exactly two bodies meet at + # the barrier and exactly two upgrades are attempted. + effect_validation=EffectValidation.DISABLED, + ) + context = self.rbt.create_external_context(name=self.id()) + + await General.ConstructorWriter(context, 'target') + # Retries after the server starts are unrelated to the lock. + should_retries.clear() + + started = time.monotonic() + await asyncio.gather( + General.ConstructorTransaction( + context, + 'root-1', + content={"target": "target"}, + ), + General.ConstructorTransaction( + context, + 'root-2', + content={"target": "target"}, + ), + ) + elapsed = time.monotonic() - started + + # Both writes landed, one of them on a retry. + target = await General.ref('target').Reader(context) + self.assertEqual(target.content["writes"], "2") + + # The conflict was resolved at once rather than by the lock + # deadline, by the second upgrader asking to retry for a + # presumed deadlock with its age carried back on the retry. + self.assertLess(elapsed, 15) + deaths = [ + (call, should_retry) + for call, should_retry in should_retries + if should_retry.reason == + errors_pb2.TransactionShouldRetry.PRESUMED_DEADLOCK + ] + self.assertGreater(len(deaths), 0) + for call, should_retry in deaths: + self.assertNotEqual(should_retry.retry_age, '') + self.assertIn( + (TRANSACTION_RETRY_AGE_HEADER, should_retry.retry_age), + call._metadata, + ) + async def test_retry_carries_the_age_of_the_first_attempt(self) -> None: """A call that aborts with a `TransactionShouldRetry` whose reason skips backoff learns the transaction's age from the