diff --git a/pybase/client.py b/pybase/client.py index 9961ff6..40a36f1 100644 --- a/pybase/client.py +++ b/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): diff --git a/pybase/exceptions.py b/pybase/exceptions.py index 43d0188..de2e675 100644 --- a/pybase/exceptions.py +++ b/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() diff --git a/pybase/helpers/__init__.py b/pybase/helpers/__init__.py index c36125f..3246372 100644 --- a/pybase/helpers/__init__.py +++ b/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 diff --git a/pybase/region/client.py b/pybase/region/client.py index 608a96f..f59cfd0 100644 --- a/pybase/region/client.py +++ b/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. diff --git a/tests/test_client_retirement.py b/tests/test_client_retirement.py new file mode 100644 index 0000000..8e9b744 --- /dev/null +++ b/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() diff --git a/tests/test_region_cache_key.py b/tests/test_region_cache_key.py new file mode 100644 index 0000000..2de16af --- /dev/null +++ b/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() diff --git a/tests/test_retry_bounds.py b/tests/test_retry_bounds.py index 07e2044..0bb063a 100644 --- a/tests/test_retry_bounds.py +++ b/tests/test_retry_bounds.py @@ -3,6 +3,8 @@ import gc import unittest import weakref +from concurrent.futures import ThreadPoolExecutor +from threading import Lock from pybase.client import MAX_REQUEST_ATTEMPTS, MainClient from pybase.exceptions import PyBaseException @@ -16,12 +18,7 @@ def _handle_exception(self, main_client, **kwargs): 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. - """ + """Retries must stay bounded and must not accumulate memory per attempt.""" def setUp(self): self.client = MainClient.__new__(MainClient) @@ -64,12 +61,7 @@ def fails_once(table, key): 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. - """ + """A saved exception retains the values via its traceback's frame.""" class Sentinel(object): pass @@ -101,6 +93,98 @@ def fails_once(table, key): finally: gc.enable() + def test_retrying_does_not_accumulate_memory(self): + """At most one attempt is live at a time; recursion held all of them.""" + built = [] + live_at_build = [] + + class FakeRequest(object): + def __init__(self): + self.payload = bytearray(1000000) + + def rq_type(region, key, values): + # Sampled here, not after unwinding: recursion holds every attempt at + # peak and frees them all on the way out, which a post-hoc check misses. + live_at_build.append(len([r for r in built if r() is not None])) + request = FakeRequest() + built.append(weakref.ref(request)) + return request + + class FakeRegionClient(object): + def _send_request(self, rq): + raise RecoverableException('broken pipe') + + class FakeRegion(object): + region_client = FakeRegionClient() + + self.client._find_hosting_region = lambda table, key: FakeRegion() + + gc.disable() + try: + with self.assertRaises(RecoverableException): + self.client._mutate('table', 'key', {'cf': {}}, rq_type) + self.assertEqual(len(built), MAX_REQUEST_ATTEMPTS) + # One previous request may still be bound while the next is built. + self.assertLessEqual(max(live_at_build), 1) + self.assertEqual([r for r in built if r() is not None], []) + finally: + gc.enable() + + def test_bound_leaves_room_for_a_region_reconnect(self): + """10 sat below a reconnect's attempts and failed live writes.""" + self.assertGreaterEqual(MAX_REQUEST_ATTEMPTS, 50) + + +class TestSendRequestReleasesFuture(unittest.TestCase): + """A failed Future holds its request, so retries must not wait on cyclic gc.""" + + def test_failed_request_is_released_without_cyclic_gc(self): + from pybase.region.client import Client + + client = Client.__new__(Client) + client.call_lock = Lock() + client.call_id = 0 + client.thread_pool = ThreadPoolExecutor(max_workers=2) + client.thread_pool_timeout = 5 + client.host, client.port = b'host', b'1' + + def always_fails(call_id, rq, to_send): + raise RecoverableException('broken pipe') + + client.send_and_receive_rpc = always_fails + + class FakePb(object): + def SerializeToString(self): + return b'x' * 100000 + + class FakeRequest(object): + type = 'Get' + pb = FakePb() + + refs = [] + + def one_attempt(): + # The request must stay local to a frame that only the traceback can + # reach, as it is in production; deleting it here would hide the leak. + request = FakeRequest() + refs.append(weakref.ref(request)) + try: + client._send_request(request) + except RecoverableException: + pass + + gc.disable() + try: + for _ in range(10): + one_attempt() + # result() can return before the worker drops its work item, which also + # holds the request, so join the pool to measure persistent retention. + client.thread_pool.shutdown(wait=True) + retained = len([r for r in refs if r() is not None]) + self.assertEqual(retained, 0, '%s of %s requests retained' % (retained, len(refs))) + finally: + gc.enable() + if __name__ == '__main__': unittest.main()