Recover from broken region connections: purge dead clients, release futures, raise the retry bound

pybase/client.py

@@ -35,15 +35,15 @@
     ZookeeperException
 )
 from .filters import _to_filter
+from .helpers import region_cache_key
 from .region.region import region_from_cell
 from .request import request
 
 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
+# Stops a permanently failing row spinning; not a memory limit, since the loop
+# frees each attempt. Must exceed a region reconnect's attempts: 10 did not.
+MAX_REQUEST_ATTEMPTS = 100
 
 
 class MainClient(object):
@@ -167,27 +167,16 @@ def get(self, table, key, families={}, filters=None):
         """
         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.
+                # Bound before the call so the except block can read it even when
+                # _find_hosting_region is what threw.
                 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.
+                # Reestablishes the region client so the next attempt can succeed,
+                # or re-raises here if the error is unrecoverable.
                 e._handle_exception(self, dest_region=dest_region)
                 if attempt == MAX_REQUEST_ATTEMPTS - 1:
                     raise
@@ -258,8 +247,7 @@ def increment(self, table, key, values):
         return self._mutate(table, key, values, request.increment_request)
 
     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.
+        # Shared by put/delete/append/increment; same retry shape as 'get'.
         for attempt in range(MAX_REQUEST_ATTEMPTS):
             try:
                 dest_region = None
@@ -469,8 +457,8 @@ def _create_new_region(self, response, table):
             elif cell.qualifier == b"server":
                 # Grab the host, port of the Region Server that this region is
                 # hosted on.
-                server_loc = cell.value
                 host, port = cell.value.split(b':')
+                server_loc = region_cache_key(host, port)
             else:
                 continue
         # Do we have an existing client for this region server already?
@@ -526,10 +514,15 @@ def _purge_client(self, region_client):
         # our cache, delete the reverse lookup entry and close the client
         # clearing up any file descriptors.
         with self._cache_lock:
+            key = region_cache_key(region_client.host, region_client.port)
+            if self.reverse_client_cache.get(key) is not region_client:
+                # A healthy client already took this address, so a late failure
+                # from the old one must retire it without evicting its successor.
+                region_client.close()
+                return
             for reg in region_client.regions:
                 self._delete_from_region_cache(reg.table, reg.start_key)
-            self.reverse_client_cache.pop(
-                region_client.host + b":" + region_client.port, None)
+            self.reverse_client_cache.pop(key, None)
             region_client.close()
 
     def _purge_region(self, reg):

pybase/exceptions.py

@@ -23,6 +23,8 @@
 
 from builtins import str
 
+from .helpers import region_cache_key
+
 logger = logging.getLogger(__name__)
 
 
@@ -75,9 +77,8 @@ def _handle_exception(self, main_client, **kwargs):
         # region_client not set? Then host/port must have been. Fetch the
         # client given the host, port
         if self.region_client is None:
-            concat = self.host + b":" + self.port
             self.region_client = main_client.reverse_client_cache.get(
-                concat, None)
+                region_cache_key(self.host, self.port), None)
 
         # we don't care about secondaries, move on
         if (self.region_client and self.region_client.secondary) or self.secondary:
@@ -90,7 +91,8 @@ def _handle_exception(self, main_client, **kwargs):
                 if self.region_client is not None:
                     # We need to make sure that a different thread hasn't already
                     # reestablished to this region.
-                    loc = self.region_client.host + ":" + self.region_client.port
+                    loc = region_cache_key(self.region_client.host,
+                                           self.region_client.port)
                     if loc in main_client.reverse_client_cache:
                         # We're the first in and it's our job to kill the client.
                         # Purge it.
@@ -102,8 +104,7 @@ def _handle_exception(self, main_client, **kwargs):
                         main_client._purge_client(self.region_client)
                         # Sleep for an arbitrary amount of time. If this returns
                         # False then we've hit our max retry threshold. Die.
-                        key = self.region_client.host + ":" + self.region_client.port
-                        if not _dynamic_sleep(self, key):
+                        if not _dynamic_sleep(self, loc):
                             raise self
             finally:
                 # Notify all the other threads to wake up because we've handled the
@@ -270,13 +271,14 @@ class MalformedValues(PyBaseException):
 # _let_all_through.
 def _let_one_through(exception, data):
     my_tuple = (exception.__class__.__name__, data)
-    # Grab my relevant semaphore.
+    # Claim under the same lock that guards the map, or the bucket can be retired
+    # between lookup and acquisition and this caller ends up owning a semaphore
+    # nobody will ever release, stranding everyone waiting on it.
     _buckets_lock.acquire()
     my_sem = _buckets[my_tuple]
+    claimed = my_sem.acquire(blocking=False)
     _buckets_lock.release()
-    # Try to non-blocking acquire my semaphore. If I get it, woohoo! Otherwise
-    # get comfy because we're sitting on the semaphore.
-    if my_sem.acquire(blocking=False):
+    if claimed:
         # Look at me - I'm the captain now.
         return True
     else:
@@ -294,6 +296,9 @@ def _let_all_through(exception, data):
     _buckets_lock.acquire()
     my_sem = _buckets[my_tuple]
     my_sem.release()
+    # The key holds the region client, so a retained bucket outlives every
+    # retired client. Waiters already hold their own reference to the semaphore.
+    _buckets.pop(my_tuple, None)
     _buckets_lock.release()
 
 

pybase/helpers/__init__.py

@@ -1,3 +1,16 @@
 from __future__ import absolute_import, print_function
 
 from . import varint  # noqa
+
+
+def region_cache_key(host, port):
+    """Key for MainClient.reverse_client_cache, as a str.
+
+    Region servers arrive as bytes from protobuf but Client decodes them, so the
+    two spellings have to be normalised or a cached client is never found again.
+    """
+    if isinstance(host, bytes):
+        host = host.decode('utf8')
+    if isinstance(port, bytes):
+        port = port.decode('utf8')
+    return host + ':' + port

pybase/region/client.py

@@ -128,10 +128,24 @@ def _send_request(self, rq, lock_timeout=10, _async=False):
         to_send += serialized_header + rpc_length_bytes + serialized_rpc
 
         # send and receive the request
-        future = self.thread_pool.submit(self.send_and_receive_rpc, my_id, rq, to_send)
+        try:
+            future = self.thread_pool.submit(self.send_and_receive_rpc, my_id, rq, to_send)
+        except RuntimeError:
+            # Retired between the region lookup and this submit; surface it as a
+            # region failure so the caller retries instead of dying on a
+            # RuntimeError the retry handlers do not catch.
+            if self.shutting_down:
+                raise RegionServerException(region_client=self)
+            raise
         if _async:
             return future
-        return future.result(timeout=self.thread_pool_timeout)
+        try:
+            return future.result(timeout=self.thread_pool_timeout)
+        finally:
+            # A failed Future holds its exception, whose traceback references this
+            # frame, which references the Future. Dropping the local breaks that
+            # cycle so a retried request is freed without waiting for cyclic gc.
+            del future
 
     # Sending an RPC, listens for the response and builds the correct pbResponse object.
     #
@@ -231,6 +245,10 @@ def close(self):
         self.shutting_down = True
         for sock in self.sock_pool:
             sock.close()
+        if self.thread_pool is not None:
+            # Closing sockets leaves the pool's workers running, so a retired
+            # client would strand its threads.
+            self.thread_pool.shutdown(wait=False)
         # We could still have greenlets waiting in the bad_call_id pools! Wake
         # them up so they can fail to error handling as well.
 

tests/test_client_retirement.py

@@ -0,0 +1,114 @@
+from __future__ import absolute_import, print_function, unicode_literals
+
+import threading
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+
+from pybase import exceptions
+from pybase.client import MainClient
+from pybase.exceptions import RegionServerException
+from pybase.helpers import region_cache_key
+from pybase.region.client import Client
+
+
+def _region_client(host, port, with_pool=False):
+    client = Client.__new__(Client)
+    client.host = host
+    client.port = port
+    client.secondary = False
+    client.regions = []
+    client.shutting_down = False
+    client.sock_pool = []
+    client.thread_pool = ThreadPoolExecutor(max_workers=1) if with_pool else None
+    return client
+
+
+def _main_with(cached):
+    main = MainClient.__new__(MainClient)
+    main.reverse_client_cache = dict(cached)
+    main.region_cache = {}
+    main._cache_lock = threading.Lock()
+    return main
+
+
+class TestClientRetirement(unittest.TestCase):
+
+    def test_late_failure_does_not_evict_the_replacement(self):
+        """Normalised keys let a retired client's failure match its successor."""
+        key = region_cache_key('rs1', '16020')
+        retired = _region_client('rs1', '16020')
+        replacement = _region_client('rs1', '16020')
+        main = _main_with({key: replacement})
+
+        main._purge_client(retired)
+
+        self.assertIs(main.reverse_client_cache.get(key), replacement)
+
+    def test_purging_the_cached_client_still_evicts_it(self):
+        key = region_cache_key('rs1', '16020')
+        cached = _region_client('rs1', '16020')
+        main = _main_with({key: cached})
+
+        main._purge_client(cached)
+
+        self.assertEqual(main.reverse_client_cache, {})
+
+    def test_close_retires_the_executor(self):
+        """Closing sockets alone leaves the pool's worker threads running."""
+        client = _region_client('rs1', '16020', with_pool=True)
+        pool = client.thread_pool
+
+        client.close()
+
+        self.assertTrue(pool._shutdown)
+
+    def test_repeated_retirement_does_not_grow_the_bucket_map(self):
+        """The bucket key holds the region client, so entries must be released."""
+        exceptions._buckets.clear()
+        for _ in range(25):
+            client = _region_client('rs1', '16020')
+            error = RegionServerException(region_client=client)
+            exceptions._let_one_through(error, client)
+            exceptions._let_all_through(error, client)
+        self.assertEqual(len(exceptions._buckets), 0)
+
+    def test_bucket_round_trip_claims_and_releases_one_semaphore(self):
+        """Covers the claim/release round trip. The interleaving that motivated
+        making the claim atomic is not reproduced here; see the PR discussion."""
+        exceptions._buckets.clear()
+        client = _region_client('rs1', '16020')
+        first = RegionServerException(region_client=client)
+        second = RegionServerException(region_client=client)
+
+        self.assertTrue(exceptions._let_one_through(first, client))
+        # Owner finishes and retires the bucket.
+        exceptions._let_all_through(first, client)
+        # A later caller must claim a live bucket, not the retired semaphore.
+        self.assertTrue(exceptions._let_one_through(second, client))
+        claimed = exceptions._buckets[(second.__class__.__name__, client)]
+        exceptions._let_all_through(second, client)
+        self.assertEqual(claimed._value, 1, 'released a different semaphore')
+
+    def test_submit_after_retirement_is_a_region_failure(self):
+        """RuntimeError from a retired pool would bypass the retry handlers."""
+        client = _region_client('rs1', '16020', with_pool=True)
+        client.call_lock = threading.Lock()
+        client.call_id = 0
+        client.thread_pool_timeout = 5
+
+        class FakePb(object):
+            def SerializeToString(self):
+                return b'x'
+
+        class FakeRequest(object):
+            type = 'Get'
+            pb = FakePb()
+
+        client.close()
+
+        with self.assertRaises(RegionServerException):
+            client._send_request(FakeRequest())
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_region_cache_key.py

@@ -0,0 +1,63 @@
+from __future__ import absolute_import, print_function, unicode_literals
+
+import unittest
+
+from pybase.client import MainClient
+from pybase.exceptions import RegionServerException
+from pybase.helpers import region_cache_key
+from pybase.region.client import Client
+
+
+def _region_client(host, port):
+    """A Client as pybase builds it: __init__ decodes host and port to str."""
+    client = Client.__new__(Client)
+    client.host = host.decode('utf8') if isinstance(host, bytes) else host
+    client.port = port.decode('utf8') if isinstance(port, bytes) else port
+    client.secondary = False
+    client.regions = []
+    client.closed = False
+    client.close = lambda: setattr(client, 'closed', True)
+    return client
+
+
+class TestRegionCacheKey(unittest.TestCase):
+    """Region servers arrive as bytes but Client stores them decoded."""
+
+    def test_bytes_and_str_spellings_agree(self):
+        self.assertEqual(region_cache_key(b'rs1', b'16020'),
+                         region_cache_key('rs1', '16020'))
+
+    def test_handler_reaches_purge_for_a_cached_client(self):
+        """Covers the handler calling purge at all, not the key spelling."""
+        client = _region_client(b'rs1.example', b'16020')
+        main = MainClient.__new__(MainClient)
+        main.reverse_client_cache = {region_cache_key(b'rs1.example', b'16020'): client}
+
+        purged = []
+        main._purge_client = lambda c: purged.append(c)
+
+        error = RegionServerException(region_client=client)
+        try:
+            error._handle_exception(main)
+        except RegionServerException:
+            pass
+
+        self.assertEqual(purged, [client],
+                         'cached client was not purged, so a retry reuses it')
+
+    def test_purge_removes_the_entry_the_cache_was_stored_under(self):
+        """Mixing a str host with a bytes separator raised, so nothing was purged."""
+        client = _region_client(b'rs1.example', b'16020')
+        main = MainClient.__new__(MainClient)
+        main.reverse_client_cache = {region_cache_key(b'rs1.example', b'16020'): client}
+        main.region_cache = {}
+        main._cache_lock = __import__('threading').Lock()
+
+        main._purge_client(client)
+
+        self.assertEqual(main.reverse_client_cache, {})
+        self.assertTrue(client.closed)
+
+
+if __name__ == '__main__':
+    unittest.main()