Bound HBase request retries instead of recursing

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):

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()