From e235b024c84e9880e7ca7fb669e4983ace0f57be Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Tue, 25 Feb 2020 17:42:57 +0300 Subject: [PATCH 1/6] feat(pubsub): implement max_extension_period --- .../pubsub_v1/subscriber/_protocol/leaser.py | 8 ++++---- .../_protocol/streaming_pull_manager.py | 17 ++++++++++++++--- google/cloud/pubsub_v1/types.py | 8 +++++++- tests/unit/pubsub_v1/subscriber/test_leaser.py | 1 + .../subscriber/test_streaming_pull_manager.py | 9 +++++++++ 5 files changed, 35 insertions(+), 8 deletions(-) diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/leaser.py b/google/cloud/pubsub_v1/subscriber/_protocol/leaser.py index b60379444..5830680da 100644 --- a/google/cloud/pubsub_v1/subscriber/_protocol/leaser.py +++ b/google/cloud/pubsub_v1/subscriber/_protocol/leaser.py @@ -130,8 +130,8 @@ def maintain_leases(self): # Determine the appropriate duration for the lease. This is # based off of how long previous messages have taken to ack, with # a sensible default and within the ranges allowed by Pub/Sub. - p99 = self._manager.ack_histogram.percentile(99) - _LOGGER.debug("The current p99 value is %d seconds.", p99) + deadline = self._manager.ack_deadline + _LOGGER.debug("The current deadline value is %d seconds.", deadline) # Make a copy of the leased messages. This is needed because it's # possible for another thread to modify the dictionary while @@ -173,7 +173,7 @@ def maintain_leases(self): # way for ``send_request`` to fail when the consumer # is inactive. self._manager.dispatcher.modify_ack_deadline( - [requests.ModAckRequest(ack_id, p99) for ack_id in ack_ids] + [requests.ModAckRequest(ack_id, deadline) for ack_id in ack_ids] ) # Now wait an appropriate period of time and do this again. @@ -182,7 +182,7 @@ def maintain_leases(self): # period between 0 seconds and 90% of the lease. This use of # jitter (http://bit.ly/2s2ekL7) helps decrease contention in cases # where there are many clients. - snooze = random.uniform(0.0, p99 * 0.9) + snooze = random.uniform(0.0, deadline * 0.9) _LOGGER.debug("Snoozing lease management for %f seconds.", snooze) self._stop_event.wait(timeout=snooze) diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py index 9b3f8d5fe..e10d17269 100644 --- a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py +++ b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py @@ -191,9 +191,20 @@ def ack_deadline(self): Returns: int: The ack deadline. """ - target = min([self._last_histogram_size * 2, self._last_histogram_size + 100]) - if len(self.ack_histogram) > target: - self._ack_deadline = self.ack_histogram.percentile(percent=99) + target_size = min( + self._last_histogram_size * 2, self._last_histogram_size + 100 + ) + hist_size = len(self.ack_histogram) + + if hist_size > target_size: + self._last_histogram_size = hist_size + p99 = self.ack_histogram.percentile(percent=99) + + if self.flow_control.max_extension_period > 0: + self._ack_deadline = min(p99, self.flow_control.max_extension_period) + else: + self._ack_deadline = p99 + return self._ack_deadline @property diff --git a/google/cloud/pubsub_v1/types.py b/google/cloud/pubsub_v1/types.py index 28019f478..6231ac2af 100644 --- a/google/cloud/pubsub_v1/types.py +++ b/google/cloud/pubsub_v1/types.py @@ -87,12 +87,14 @@ # these settings can be altered to tweak Pub/Sub behavior. # The defaults should be fine for most use cases. FlowControl = collections.namedtuple( - "FlowControl", ["max_bytes", "max_messages", "max_lease_duration"] + "FlowControl", + ["max_bytes", "max_messages", "max_lease_duration", "max_extension_period"], ) FlowControl.__new__.__defaults__ = ( 100 * 1024 * 1024, # max_bytes: 100mb 1000, # max_messages: 1000 1 * 60 * 60, # max_lease_duration: 1 hour. + 0, # max_extension_period: disabled ) if sys.version_info >= (3, 5): @@ -112,6 +114,10 @@ "The maximum amount of time in seconds to hold a lease on a message " "before dropping it from the lease management." ) + FlowControl.max_extension_period.__doc__ = ( + "Bounds the maximum amount of time before a message redelivery in the " + "event the subscriber fails to extend the deadline." + ) _shared_modules = [ diff --git a/tests/unit/pubsub_v1/subscriber/test_leaser.py b/tests/unit/pubsub_v1/subscriber/test_leaser.py index ec954b89d..17409cb3f 100644 --- a/tests/unit/pubsub_v1/subscriber/test_leaser.py +++ b/tests/unit/pubsub_v1/subscriber/test_leaser.py @@ -84,6 +84,7 @@ def create_manager(flow_control=types.FlowControl()): manager.is_active = True manager.flow_control = flow_control manager.ack_histogram = histogram.Histogram() + manager.ack_deadline = 10 return manager diff --git a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py index 0886a4508..78ebbe357 100644 --- a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py +++ b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py @@ -144,6 +144,15 @@ def test_ack_deadline(): assert manager.ack_deadline == 20 +def test_ack_deadline_w_max_extension_period(): + manager = make_manager() + manager._flow_control = types.FlowControl(max_extension_period=5) + for _ in range(5): + manager.ack_histogram.add(20) + + assert manager.ack_deadline == 5 + + def test_maybe_pause_consumer_wo_consumer_set(): manager = make_manager( flow_control=types.FlowControl(max_messages=10, max_bytes=1000) From c82838e8528ae7ccf2bbba065c59485ddb973759 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Wed, 26 Feb 2020 13:11:24 +0300 Subject: [PATCH 2/6] comment change --- google/cloud/pubsub_v1/types.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/google/cloud/pubsub_v1/types.py b/google/cloud/pubsub_v1/types.py index 6231ac2af..50ef75414 100644 --- a/google/cloud/pubsub_v1/types.py +++ b/google/cloud/pubsub_v1/types.py @@ -115,8 +115,8 @@ "before dropping it from the lease management." ) FlowControl.max_extension_period.__doc__ = ( - "Bounds the maximum amount of time before a message redelivery in the " - "event the subscriber fails to extend the deadline." + "Bounds the maximum amount of time in seconds before a message redelivery " + "in the event the subscriber fails to extend the deadline." ) From 45b36dacbc146de192f50346eb0db89893e2e8d4 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Thu, 27 Feb 2020 14:37:46 +0300 Subject: [PATCH 3/6] simplify ack_deadline() --- .../subscriber/_protocol/streaming_pull_manager.py | 11 +++++------ .../subscriber/test_streaming_pull_manager.py | 2 ++ 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py index e10d17269..efb43e80d 100644 --- a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py +++ b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py @@ -198,13 +198,12 @@ def ack_deadline(self): if hist_size > target_size: self._last_histogram_size = hist_size - p99 = self.ack_histogram.percentile(percent=99) - - if self.flow_control.max_extension_period > 0: - self._ack_deadline = min(p99, self.flow_control.max_extension_period) - else: - self._ack_deadline = p99 + self._ack_deadline = self.ack_histogram.percentile(percent=99) + if self.flow_control.max_extension_period > 0: + self._ack_deadline = min( + self._ack_deadline, self.flow_control.max_extension_period + ) return self._ack_deadline @property diff --git a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py index 78ebbe357..75727a326 100644 --- a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py +++ b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py @@ -147,6 +147,8 @@ def test_ack_deadline(): def test_ack_deadline_w_max_extension_period(): manager = make_manager() manager._flow_control = types.FlowControl(max_extension_period=5) + + assert manager.ack_deadline == 5 for _ in range(5): manager.ack_histogram.add(20) From a23e8527a600cb3e38d11314b45749404782b9ed Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 2 Mar 2020 12:45:07 +0300 Subject: [PATCH 4/6] rename the attribute --- .../_protocol/streaming_pull_manager.py | 4 ++-- google/cloud/pubsub_v1/types.py | 16 +++++++++++----- 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py index efb43e80d..0a25d4625 100644 --- a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py +++ b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py @@ -200,9 +200,9 @@ def ack_deadline(self): self._last_histogram_size = hist_size self._ack_deadline = self.ack_histogram.percentile(percent=99) - if self.flow_control.max_extension_period > 0: + if self.flow_control.max_duration_per_lease_extension > 0: self._ack_deadline = min( - self._ack_deadline, self.flow_control.max_extension_period + self._ack_deadline, self.flow_control.max_duration_per_lease_extension ) return self._ack_deadline diff --git a/google/cloud/pubsub_v1/types.py b/google/cloud/pubsub_v1/types.py index 50ef75414..95f1f78ba 100644 --- a/google/cloud/pubsub_v1/types.py +++ b/google/cloud/pubsub_v1/types.py @@ -88,13 +88,18 @@ # The defaults should be fine for most use cases. FlowControl = collections.namedtuple( "FlowControl", - ["max_bytes", "max_messages", "max_lease_duration", "max_extension_period"], + [ + "max_bytes", + "max_messages", + "max_lease_duration", + "max_duration_per_lease_extension", + ], ) FlowControl.__new__.__defaults__ = ( 100 * 1024 * 1024, # max_bytes: 100mb 1000, # max_messages: 1000 1 * 60 * 60, # max_lease_duration: 1 hour. - 0, # max_extension_period: disabled + 0, # max_duration_per_lease_extension: disabled ) if sys.version_info >= (3, 5): @@ -114,9 +119,10 @@ "The maximum amount of time in seconds to hold a lease on a message " "before dropping it from the lease management." ) - FlowControl.max_extension_period.__doc__ = ( - "Bounds the maximum amount of time in seconds before a message redelivery " - "in the event the subscriber fails to extend the deadline." + FlowControl.max_duration_per_lease_extension.__doc__ = ( + "The max amount of time in seconds for a single lease extension attempt. " + "Bounds the delay before a message redelivery in the event the " + "subscriber fails to extend the deadline." ) From ebde0c0771805d12ab84e5fdd108ddc11d5483e0 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Mon, 2 Mar 2020 12:49:24 +0300 Subject: [PATCH 5/6] change arg name in test --- .../unit/pubsub_v1/subscriber/test_streaming_pull_manager.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py index 75727a326..802802d04 100644 --- a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py +++ b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py @@ -144,9 +144,9 @@ def test_ack_deadline(): assert manager.ack_deadline == 20 -def test_ack_deadline_w_max_extension_period(): +def test_ack_deadline_w_max_duration_per_lease_extension(): manager = make_manager() - manager._flow_control = types.FlowControl(max_extension_period=5) + manager._flow_control = types.FlowControl(max_duration_per_lease_extension=5) assert manager.ack_deadline == 5 for _ in range(5): From 8eb7e8135218835e0893763bd1bf5baf8b60b054 Mon Sep 17 00:00:00 2001 From: IlyaFaer Date: Tue, 3 Mar 2020 11:07:05 +0300 Subject: [PATCH 6/6] cosmetic changes --- google/cloud/pubsub_v1/types.py | 4 ++-- .../unit/pubsub_v1/subscriber/test_streaming_pull_manager.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/google/cloud/pubsub_v1/types.py b/google/cloud/pubsub_v1/types.py index 95f1f78ba..eb4f00681 100644 --- a/google/cloud/pubsub_v1/types.py +++ b/google/cloud/pubsub_v1/types.py @@ -121,8 +121,8 @@ ) FlowControl.max_duration_per_lease_extension.__doc__ = ( "The max amount of time in seconds for a single lease extension attempt. " - "Bounds the delay before a message redelivery in the event the " - "subscriber fails to extend the deadline." + "Bounds the delay before a message redelivery if the subscriber " + "fails to extend the deadline." ) diff --git a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py index 802802d04..70f320fcc 100644 --- a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py +++ b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py @@ -144,7 +144,7 @@ def test_ack_deadline(): assert manager.ack_deadline == 20 -def test_ack_deadline_w_max_duration_per_lease_extension(): +def test_ack_deadline_with_max_duration_per_lease_extension(): manager = make_manager() manager._flow_control = types.FlowControl(max_duration_per_lease_extension=5)