diff --git a/pybase/client.py b/pybase/client.py index 7ca1f9e..9961ff6 100644 --- a/pybase/client.py +++ b/pybase/client.py @@ -40,6 +40,11 @@ logger = logging.getLogger(__name__) +# Retries are bounded because these requests used to retry by recursing: each +# failed attempt kept its frame, row values and built request alive, so a row +# that never succeeds exhausted memory long before Python's recursion limit. +MAX_REQUEST_ATTEMPTS = 10 + class MainClient(object): @@ -160,31 +165,32 @@ def get(self, table, key, families={}, filters=None): :param filters: (optional) column filters :return: response with cells """ - try: - # Step 0. Set dest_region to None so if an exception is - # thrown in _find_hosting_region, the exception handling - # doesn't break trying to reference dest_region. - dest_region = None - # Step 1. Figure out where to send it. - dest_region = self._find_hosting_region(table, key) - # Step 2. Build the appropriate pb message. - rq = request.get_request(dest_region, key, families, filters) - # Step 3. Send the message and twiddle our thumbs. - response = dest_region.region_client._send_request(rq) - # Step 4. Success. - return Result(response) - except PyBaseException as e: - # Step X. Houston, we have an error. The cool thing about how - # this is coded is that exceptions know how to handle themselves. - # All we need to do is call _handle_exception and everything should - # be happy! If it cannot handle itself (unrecoverable) then it will - # re-raise the exception in the handle method and we'll die too. - # - # We pass dest_region in because the handling code needs to know - # which region or region_client it needs to reestablish. - e._handle_exception(self, dest_region=dest_region) - # Everything should be dandy now. Repeat the request! - return self.get(table, key, families=families, filters=filters) + for attempt in range(MAX_REQUEST_ATTEMPTS): + try: + # Step 0. Set dest_region to None so if an exception is + # thrown in _find_hosting_region, the exception handling + # doesn't break trying to reference dest_region. + dest_region = None + # Step 1. Figure out where to send it. + dest_region = self._find_hosting_region(table, key) + # Step 2. Build the appropriate pb message. + rq = request.get_request(dest_region, key, families, filters) + # Step 3. Send the message and twiddle our thumbs. + response = dest_region.region_client._send_request(rq) + # Step 4. Success. + return Result(response) + except PyBaseException as e: + # Step X. Houston, we have an error. The cool thing about how + # this is coded is that exceptions know how to handle themselves. + # All we need to do is call _handle_exception and everything should + # be happy! If it cannot handle itself (unrecoverable) then it will + # re-raise the exception in the handle method and we'll die too. + # + # We pass dest_region in because the handling code needs to know + # which region or region_client it needs to reestablish. + e._handle_exception(self, dest_region=dest_region) + if attempt == MAX_REQUEST_ATTEMPTS - 1: + raise def get_many(self, table, keys, families=None): """ @@ -254,15 +260,17 @@ def increment(self, table, key, values): def _mutate(self, table, key, values, rq_type): # Same exact methodology as 'get'. Because all mutate requests have # equivalent code I've combined them into a single function. - try: - dest_region = None - dest_region = self._find_hosting_region(table, key) - rq = rq_type(dest_region, key, values) - response = dest_region.region_client._send_request(rq) - return Result(response) - except PyBaseException as e: - e._handle_exception(self, dest_region=dest_region) - return self._mutate(table, key, values, rq_type) + for attempt in range(MAX_REQUEST_ATTEMPTS): + try: + dest_region = None + dest_region = self._find_hosting_region(table, key) + rq = rq_type(dest_region, key, values) + response = dest_region.region_client._send_request(rq) + return Result(response) + except PyBaseException as e: + e._handle_exception(self, dest_region=dest_region) + if attempt == MAX_REQUEST_ATTEMPTS - 1: + raise # Scan can get a bit gnarly - be prepared. def scan(self, table, start_key='', stop_key=None, families={}, filters=None): diff --git a/tests/test_retry_bounds.py b/tests/test_retry_bounds.py new file mode 100644 index 0000000..07e2044 --- /dev/null +++ b/tests/test_retry_bounds.py @@ -0,0 +1,106 @@ +from __future__ import absolute_import, print_function, unicode_literals + +import gc +import unittest +import weakref + +from pybase.client import MAX_REQUEST_ATTEMPTS, MainClient +from pybase.exceptions import PyBaseException + + +class RecoverableException(PyBaseException): + """A PyBaseException that _handle_exception treats as retryable.""" + + def _handle_exception(self, main_client, **kwargs): + return + + +class TestRequestRetryBounds(unittest.TestCase): + """Requests must retry a bounded number of times rather than recursing. + + The recursive implementation kept every attempt's frame, row values and + built request alive, so a row that never succeeded exhausted container + memory before Python's recursion limit was reached. + """ + + def setUp(self): + self.client = MainClient.__new__(MainClient) + self.attempts = [] + + def always_fails(table, key): + self.attempts.append(key) + raise RecoverableException('region unavailable') + + self.client._find_hosting_region = always_fails + + def test_mutate_stops_after_max_attempts(self): + with self.assertRaises(RecoverableException): + self.client._mutate('table', 'key', {'cf': {'col': 'val'}}, None) + self.assertEqual(len(self.attempts), MAX_REQUEST_ATTEMPTS) + + def test_get_stops_after_max_attempts(self): + with self.assertRaises(RecoverableException): + self.client.get('table', 'key') + self.assertEqual(len(self.attempts), MAX_REQUEST_ATTEMPTS) + + def test_mutate_returns_once_a_retry_succeeds(self): + calls = [] + + class FakeRegionClient(object): + def _send_request(self, rq): + return None + + class FakeRegion(object): + region_client = FakeRegionClient() + + def fails_once(table, key): + calls.append(key) + if len(calls) == 1: + raise RecoverableException('region moved') + return FakeRegion() + + self.client._find_hosting_region = fails_once + self.client._mutate('table', 'key', {}, lambda region, key, values: None) + self.assertEqual(len(calls), 2) + + def test_successful_retry_does_not_retain_payload(self): + """A saved exception would keep the row values alive through its traceback. + + The traceback references the frame that holds the values, so storing the + exception in a local builds a cycle that only the cyclic collector can + break. Re-raising in place keeps the retry path free of that reference. + """ + class Sentinel(object): + pass + + class FakeRegionClient(object): + def _send_request(self, rq): + return None + + class FakeRegion(object): + region_client = FakeRegionClient() + + calls = [] + + def fails_once(table, key): + calls.append(key) + if len(calls) == 1: + raise RecoverableException('region moved') + return FakeRegion() + + self.client._find_hosting_region = fails_once + sentinel = Sentinel() + values = {'payload': sentinel} + payload_ref = weakref.ref(sentinel) + + gc.disable() + try: + self.client._mutate('table', 'key', values, lambda region, key, vals: None) + del values, sentinel + self.assertIsNone(payload_ref()) + finally: + gc.enable() + + +if __name__ == '__main__': + unittest.main()