Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py	0 B	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 dispatcher.py # Copyright 2017, Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import absolute_import from __future__ import division import functools import itertools import logging import math import time import threading import typing from typing import List, Optional, Sequence, Union import warnings from google.api_core.retry import exponential_sleep_generator from opentelemetry import trace from google.cloud.pubsub_v1.subscriber._protocol import helper_threads from google.cloud.pubsub_v1.subscriber._protocol import requests from google.cloud.pubsub_v1.subscriber.exceptions import ( AcknowledgeStatus, ) from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( start_ack_span, start_nack_span, ) if typing.TYPE_CHECKING: # pragma: NO COVER import queue from google.cloud.pubsub_v1.subscriber._protocol.streaming_pull_manager import ( StreamingPullManager, ) RequestItem = Union[ requests.AckRequest, requests.DropRequest, requests.LeaseRequest, requests.ModAckRequest, requests.NackRequest, ] _LOGGER = logging.getLogger(__name__) _CALLBACK_WORKER_NAME = "Thread-CallbackRequestDispatcher" _MAX_BATCH_SIZE = 100 """The maximum number of requests to process and dispatch at a time.""" _MAX_BATCH_LATENCY = 0.01 """The maximum amount of time in seconds to wait for additional request items before processing the next batch of requests.""" _ACK_IDS_BATCH_SIZE = 1000 """The maximum number of ACK IDs to send in a single StreamingPullRequest. """ _MIN_EXACTLY_ONCE_DELIVERY_ACK_MODACK_RETRY_DURATION_SECS = 1 """The time to wait for the first retry of failed acks and modacks when exactly-once delivery is enabled.""" _MAX_EXACTLY_ONCE_DELIVERY_ACK_MODACK_RETRY_DURATION_SECS = 10 * 60 """The maximum amount of time in seconds to retry failed acks and modacks when exactly-once delivery is enabled.""" class Dispatcher(object): def __init__(self, manager: "StreamingPullManager", queue: "queue.Queue"): self._manager = manager self._queue = queue self._thread: Optional[threading.Thread] = None self._operational_lock = threading.Lock() def start(self) -> None: """Start a thread to dispatch requests queued up by callbacks. Spawns a thread to run :meth:`dispatch_callback`. """ with self._operational_lock: if self._thread is not None: raise ValueError("Dispatcher is already running.") worker = helper_threads.QueueCallbackWorker( self._queue, self.dispatch_callback, max_items=_MAX_BATCH_SIZE, max_latency=_MAX_BATCH_LATENCY, ) # Create and start the helper thread. thread = threading.Thread(name=_CALLBACK_WORKER_NAME, target=worker) thread.daemon = True thread.start() _LOGGER.debug("Started helper thread %s", thread.name) self._thread = thread def stop(self) -> None: with self._operational_lock: if self._thread is not None: # Signal the worker to stop by queueing a "poison pill" self._queue.put(helper_threads.STOP) self._thread.join() self._thread = None def dispatch_callback(self, items: Sequence[RequestItem]) -> None: """Map the callback request to the appropriate gRPC request. Args: items: Queued requests to dispatch. """ lease_requests: List[requests.LeaseRequest] = [] modack_requests: List[requests.ModAckRequest] = [] ack_requests: List[requests.AckRequest] = [] nack_requests: List[requests.NackRequest] = [] drop_requests: List[requests.DropRequest] = [] lease_ids = set() modack_ids = set() ack_ids = set() nack_ids = set() drop_ids = set() exactly_once_delivery_enabled = self._manager._exactly_once_delivery_enabled() for item in items: if isinstance(item, requests.LeaseRequest): if ( item.ack_id not in lease_ids ): # LeaseRequests have no futures to handle. lease_ids.add(item.ack_id) lease_requests.append(item) elif isinstance(item, requests.ModAckRequest): if item.ack_id in modack_ids: self._handle_duplicate_request_future( exactly_once_delivery_enabled, item ) else: modack_ids.add(item.ack_id) modack_requests.append(item) elif isinstance(item, requests.AckRequest): if item.ack_id in ack_ids: self._handle_duplicate_request_future( exactly_once_delivery_enabled, item ) else: ack_ids.add(item.ack_id) ack_requests.append(item) elif isinstance(item, requests.NackRequest): if item.ack_id in nack_ids: self._handle_duplicate_request_future( exactly_once_delivery_enabled, item ) else: nack_ids.add(item.ack_id) nack_requests.append(item) elif isinstance(item, requests.DropRequest): if ( item.ack_id not in drop_ids ): # DropRequests have no futures to handle. drop_ids.add(item.ack_id) drop_requests.append(item) else: warnings.warn( f'Skipping unknown request item of type "{type(item)}"', category=RuntimeWarning, ) _LOGGER.debug("Handling %d batched requests", len(items)) if lease_requests: self.lease(lease_requests) if modack_requests: self.modify_ack_deadline(modack_requests) # Note: Drop and ack must be after lease. It's possible to get both # the lease and the ack/drop request in the same batch. if ack_requests: self.ack(ack_requests) if nack_requests: self.nack(nack_requests) if drop_requests: self.drop(drop_requests) def _handle_duplicate_request_future( self, exactly_once_delivery_enabled: bool, item: Union[requests.AckRequest, requests.ModAckRequest, requests.NackRequest], ) -> None: _LOGGER.debug( "This is a duplicate %s with the same ack_id: %s.", type(item), item.ack_id, ) if item.future: if exactly_once_delivery_enabled: item.future.set_exception( ValueError(f"Duplicate ack_id for {type(item)}") ) # Futures may be present even with exactly-once delivery # disabled, in transition periods after the setting is changed on # the subscription. else: # When exactly-once delivery is NOT enabled, acks/modacks are considered # best-effort, so the future should succeed even though this is a duplicate. item.future.set_result(AcknowledgeStatus.SUCCESS) def ack(self, items: Sequence[requests.AckRequest]) -> None: """Acknowledge the given messages. Args: items: The items to acknowledge. """ # If we got timing information, add it to the histogram. for item in items: time_to_ack = item.time_to_ack if time_to_ack is not None: self._manager.ack_histogram.add(time_to_ack) # We must potentially split the request into multiple smaller requests # to avoid the server-side max request size limit. items_gen = iter(items) ack_ids_gen = (item.ack_id for item in items) total_chunks = int(math.ceil(len(items) / _ACK_IDS_BATCH_SIZE)) subscription_id: Optional[str] = None project_id: Optional[str] = None for item in items: if item.opentelemetry_data: item.opentelemetry_data.add_subscribe_span_event("ack start") if subscription_id is None: subscription_id = item.opentelemetry_data.subscription_id if project_id is None: project_id = item.opentelemetry_data.project_id for _ in range(total_chunks): ack_reqs_dict = { req.ack_id: req for req in itertools.islice(items_gen, _ACK_IDS_BATCH_SIZE) } subscribe_links: List[trace.Link] = [] subscribe_spans: List[trace.Span] = [] for ack_req in ack_reqs_dict.values(): if ack_req.opentelemetry_data: subscribe_span: Optional[ trace.Span ] = ack_req.opentelemetry_data.subscribe_span if ( subscribe_span and subscribe_span.get_span_context().trace_flags.sampled ): subscribe_links.append( trace.Link(subscribe_span.get_span_context()) ) subscribe_spans.append(subscribe_span) ack_span: Optional[trace.Span] = None if subscription_id and project_id: ack_span = start_ack_span( subscription_id, len(ack_reqs_dict), project_id, subscribe_links, ) if ( ack_span and ack_span.get_span_context().trace_flags.sampled ): # pragma: NO COVER ack_span_context: trace.SpanContext = ack_span.get_span_context() for subscribe_span in subscribe_spans: subscribe_span.add_link( context=ack_span_context, attributes={ "messaging.operation.name": "ack", }, ) requests_completed, requests_to_retry = self._manager.send_unary_ack( ack_ids=list(itertools.islice(ack_ids_gen, _ACK_IDS_BATCH_SIZE)), ack_reqs_dict=ack_reqs_dict, ) if ack_span: ack_span.end() for completed_ack in requests_completed: if completed_ack.opentelemetry_data: completed_ack.opentelemetry_data.add_subscribe_span_event("ack end") completed_ack.opentelemetry_data.set_subscribe_span_result("acked") completed_ack.opentelemetry_data.end_subscribe_span() # Remove the completed messages from lease management. self.drop(requests_completed) # Retry on a separate thread so the dispatcher thread isn't blocked # by sleeps. if requests_to_retry: self._start_retry_thread( "Thread-RetryAcks", functools.partial(self._retry_acks, requests_to_retry), ) def _start_retry_thread(self, thread_name, thread_target): # note: if the thread is not a daemon, a memory leak exists due to a cpython issue. # https://github.com/googleapis/python-pubsub/issues/395#issuecomment-829910303 # https://github.com/googleapis/python-pubsub/issues/395#issuecomment-830092418 retry_thread = threading.Thread( name=thread_name, target=thread_target, daemon=True, ) # The thread finishes when the requests succeed or eventually fail with # a back-end timeout error or other permanent failure. retry_thread.start() def _retry_acks(self, requests_to_retry: List[requests.AckRequest]): retry_delay_gen = exponential_sleep_generator( initial=_MIN_EXACTLY_ONCE_DELIVERY_ACK_MODACK_RETRY_DURATION_SECS, maximum=_MAX_EXACTLY_ONCE_DELIVERY_ACK_MODACK_RETRY_DURATION_SECS, ) while requests_to_retry: time_to_wait = next(retry_delay_gen) _LOGGER.debug( "Retrying {len(requests_to_retry)} ack(s) after delay of " + str(time_to_wait) + " seconds" ) time.sleep(time_to_wait) ack_reqs_dict = {req.ack_id: req for req in requests_to_retry} subscription_id: Optional[str] = None project_id: Optional[str] = None subscribe_links: List[trace.Link] = [] subscribe_spans: List[trace.Span] = [] for req in requests_to_retry: if req.opentelemetry_data: req.opentelemetry_data.add_subscribe_span_event("ack start") if subscription_id is None: subscription_id = req.opentelemetry_data.subscription_id if project_id is None: project_id = req.opentelemetry_data.project_id subscribe_span: Optional[ trace.Span ] = req.opentelemetry_data.subscribe_span if ( subscribe_span and subscribe_span.get_span_context().trace_flags.sampled ): subscribe_links.append( trace.Link(subscribe_span.get_span_context()) ) subscribe_spans.append(subscribe_span) ack_span: Optional[trace.Span] = None if subscription_id and project_id: ack_span = start_ack_span( subscription_id, len(ack_reqs_dict), project_id, subscribe_links, ) if ( ack_span and ack_span.get_span_context().trace_flags.sampled ): # pragma: NO COVER ack_span_context: trace.SpanContext = ack_span.get_span_context() for subscribe_span in subscribe_spans: subscribe_span.add_link( context=ack_span_context, attributes={ "messaging.operation.name": "ack", }, ) requests_completed, requests_to_retry = self._manager.send_unary_ack( ack_ids=[req.ack_id for req in requests_to_retry], ack_reqs_dict=ack_reqs_dict, ) if ack_span: ack_span.end() for completed_ack in requests_completed: if completed_ack.opentelemetry_data: completed_ack.opentelemetry_data.add_subscribe_span_event("ack end") completed_ack.opentelemetry_data.set_subscribe_span_result("acked") completed_ack.opentelemetry_data.end_subscribe_span() assert ( len(requests_to_retry) <= _ACK_IDS_BATCH_SIZE ), "Too many requests to be retried." # Remove the completed messages from lease management. self.drop(requests_completed) def drop( self, items: Sequence[ Union[requests.AckRequest, requests.DropRequest, requests.NackRequest] ], ) -> None: """Remove the given messages from lease management. Args: items: The items to drop. """ assert self._manager.leaser is not None self._manager.leaser.remove(items) ordering_keys = (k.ordering_key for k in items if k.ordering_key) self._manager.activate_ordering_keys(ordering_keys) self._manager.maybe_resume_consumer() def lease(self, items: Sequence[requests.LeaseRequest]) -> None: """Add the given messages to lease management. Args: items: The items to lease. """ assert self._manager.leaser is not None self._manager.leaser.add(items) self._manager.maybe_pause_consumer() def modify_ack_deadline( self, items: Sequence[requests.ModAckRequest], default_deadline: Optional[float] = None, ) -> None: """Modify the ack deadline for the given messages. Args: items: The items to modify. """ # We must potentially split the request into multiple smaller requests # to avoid the server-side max request size limit. items_gen = iter(items) ack_ids_gen = (item.ack_id for item in items) deadline_seconds_gen = (item.seconds for item in items) total_chunks = int(math.ceil(len(items) / _ACK_IDS_BATCH_SIZE)) subscription_id: Optional[str] = None project_id: Optional[str] = None for item in items: if item.opentelemetry_data: if math.isclose(item.seconds, 0): item.opentelemetry_data.add_subscribe_span_event("nack start") if subscription_id is None: subscription_id = item.opentelemetry_data.subscription_id if project_id is None: project_id = item.opentelemetry_data.project_id else: item.opentelemetry_data.add_subscribe_span_event("modack start") for _ in range(total_chunks): ack_reqs_dict = { req.ack_id: req for req in itertools.islice(items_gen, _ACK_IDS_BATCH_SIZE) } subscribe_links: List[trace.Link] = [] subscribe_spans: List[trace.Span] = [] for ack_req in ack_reqs_dict.values(): if ack_req.opentelemetry_data and math.isclose(ack_req.seconds, 0): subscribe_span: Optional[ trace.Span ] = ack_req.opentelemetry_data.subscribe_span if ( subscribe_span and subscribe_span.get_span_context().trace_flags.sampled ): subscribe_links.append( trace.Link(subscribe_span.get_span_context()) ) subscribe_spans.append(subscribe_span) nack_span: Optional[trace.Span] = None if subscription_id and project_id: nack_span = start_nack_span( subscription_id, len(ack_reqs_dict), project_id, subscribe_links, ) if ( nack_span and nack_span.get_span_context().trace_flags.sampled ): # pragma: NO COVER nack_span_context: trace.SpanContext = nack_span.get_span_context() for subscribe_span in subscribe_spans: subscribe_span.add_link( context=nack_span_context, attributes={ "messaging.operation.name": "nack", }, ) requests_to_retry: List[requests.ModAckRequest] requests_completed: Optional[List[requests.ModAckRequest]] = None if default_deadline is None: # no further work needs to be done for `requests_to_retry` requests_completed, requests_to_retry = self._manager.send_unary_modack( modify_deadline_ack_ids=list( itertools.islice(ack_ids_gen, _ACK_IDS_BATCH_SIZE) ), modify_deadline_seconds=list( itertools.islice(deadline_seconds_gen, _ACK_IDS_BATCH_SIZE) ), ack_reqs_dict=ack_reqs_dict, default_deadline=None, ) else: requests_completed, requests_to_retry = self._manager.send_unary_modack( modify_deadline_ack_ids=itertools.islice( ack_ids_gen, _ACK_IDS_BATCH_SIZE ), modify_deadline_seconds=None, ack_reqs_dict=ack_reqs_dict, default_deadline=default_deadline, ) if nack_span: nack_span.end() assert ( len(requests_to_retry) <= _ACK_IDS_BATCH_SIZE ), "Too many requests to be retried." for completed_modack in requests_completed: if completed_modack.opentelemetry_data: # nack is a modack with 0 extension seconds. if math.isclose(completed_modack.seconds, 0): completed_modack.opentelemetry_data.set_subscribe_span_result( "nacked" ) completed_modack.opentelemetry_data.add_subscribe_span_event( "nack end" ) completed_modack.opentelemetry_data.end_subscribe_span() else: completed_modack.opentelemetry_data.add_subscribe_span_event( "modack end" ) # Retry on a separate thread so the dispatcher thread isn't blocked # by sleeps. if requests_to_retry: self._start_retry_thread( "Thread-RetryModAcks", functools.partial(self._retry_modacks, requests_to_retry), ) def _retry_modacks(self, requests_to_retry): retry_delay_gen = exponential_sleep_generator( initial=_MIN_EXACTLY_ONCE_DELIVERY_ACK_MODACK_RETRY_DURATION_SECS, maximum=_MAX_EXACTLY_ONCE_DELIVERY_ACK_MODACK_RETRY_DURATION_SECS, ) while requests_to_retry: time_to_wait = next(retry_delay_gen) _LOGGER.debug( "Retrying {len(requests_to_retry)} modack(s) after delay of " + str(time_to_wait) + " seconds" ) time.sleep(time_to_wait) ack_reqs_dict = {req.ack_id: req for req in requests_to_retry} subscription_id = None project_id = None subscribe_links = [] subscribe_spans = [] for ack_req in ack_reqs_dict.values(): if ack_req.opentelemetry_data and math.isclose(ack_req.seconds, 0): if subscription_id is None: subscription_id = ack_req.opentelemetry_data.subscription_id if project_id is None: project_id = ack_req.opentelemetry_data.project_id subscribe_span = ack_req.opentelemetry_data.subscribe_span if ( subscribe_span and subscribe_span.get_span_context().trace_flags.sampled ): subscribe_links.append( trace.Link(subscribe_span.get_span_context()) ) subscribe_spans.append(subscribe_span) nack_span = None if subscription_id and project_id: nack_span = start_nack_span( subscription_id, len(ack_reqs_dict), project_id, subscribe_links, ) if ( nack_span and nack_span.get_span_context().trace_flags.sampled ): # pragma: NO COVER nack_span_context: trace.SpanContext = nack_span.get_span_context() for subscribe_span in subscribe_spans: subscribe_span.add_link( context=nack_span_context, attributes={ "messaging.operation.name": "nack", }, ) requests_completed, requests_to_retry = self._manager.send_unary_modack( modify_deadline_ack_ids=[req.ack_id for req in requests_to_retry], modify_deadline_seconds=[req.seconds for req in requests_to_retry], ack_reqs_dict=ack_reqs_dict, ) if nack_span: nack_span.end() for completed_modack in requests_completed: if completed_modack.opentelemetry_data: # nack is a modack with 0 extension seconds. if math.isclose(completed_modack.seconds, 0): completed_modack.opentelemetry_data.set_subscribe_span_result( "nacked" ) completed_modack.opentelemetry_data.add_subscribe_span_event( "nack end" ) completed_modack.opentelemetry_data.end_subscribe_span() else: completed_modack.opentelemetry_data.add_subscribe_span_event( "modack end" ) def nack(self, items: Sequence[requests.NackRequest]) -> None: """Explicitly deny receipt of messages. Args: items: The items to deny. """ self.modify_ack_deadline( [ requests.ModAckRequest( ack_id=item.ack_id, seconds=0, future=item.future, opentelemetry_data=item.opentelemetry_data, ) for item in items ] ) self.drop( [ requests.DropRequest( ack_id=item.ack_id, byte_size=item.byte_size, ordering_key=item.ordering_key, ) for item in items ] )	26.46 KB	2025-05-25 12:26
📄 heartbeater.py # Copyright 2018, Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import absolute_import import logging import threading import typing from typing import Optional if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud.pubsub_v1.subscriber._protocol.streaming_pull_manager import ( StreamingPullManager, ) _LOGGER = logging.getLogger(__name__) _HEARTBEAT_WORKER_NAME = "Thread-Heartbeater" # How often to send heartbeats in seconds. Determined as half the period of # time where the Pub/Sub server will close the stream as inactive, which is # 60 seconds. _DEFAULT_PERIOD = 30 class Heartbeater(object): def __init__(self, manager: "StreamingPullManager", period: int = _DEFAULT_PERIOD): self._thread: Optional[threading.Thread] = None self._operational_lock = threading.Lock() self._manager = manager self._stop_event = threading.Event() self._period = period def heartbeat(self) -> None: """Periodically send streaming pull heartbeats.""" while not self._stop_event.is_set(): if self._manager.heartbeat(): _LOGGER.debug("Sent heartbeat.") self._stop_event.wait(timeout=self._period) _LOGGER.debug("%s exiting.", _HEARTBEAT_WORKER_NAME) def start(self) -> None: with self._operational_lock: if self._thread is not None: raise ValueError("Heartbeater is already running.") # Create and start the helper thread. self._stop_event.clear() thread = threading.Thread( name=_HEARTBEAT_WORKER_NAME, target=self.heartbeat ) thread.daemon = True thread.start() _LOGGER.debug("Started helper thread %s", thread.name) self._thread = thread def stop(self) -> None: with self._operational_lock: self._stop_event.set() if self._thread is not None: # The thread should automatically exit when the consumer is # inactive. self._thread.join() self._thread = None	2.61 KB	2025-05-25 12:26
📄 helper_threads.py # Copyright 2017, Google LLC All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import logging import queue import time from typing import Any, Callable, List, Sequence, Optional import uuid __all__ = ("QueueCallbackWorker", "STOP") _LOGGER = logging.getLogger(__name__) # Helper thread stop indicator. This could be a sentinel object or None, # but the sentinel object's ID can change if the process is forked, and # None has the possibility of a user accidentally killing the helper # thread. STOP = uuid.uuid4() def _get_many( queue_: queue.Queue, max_items: Optional[int] = None, max_latency: float = 0 ) -> List[Any]: """Get multiple items from a Queue. Gets at least one (blocking) and at most ``max_items`` items (non-blocking) from a given Queue. Does not mark the items as done. Args: queue_: The Queue to get items from. max_items: The maximum number of items to get. If ``None``, then all available items in the queue are returned. max_latency: The maximum number of seconds to wait for more than one item from a queue. This number includes the time required to retrieve the first item. Returns: A sequence of items retrieved from the queue. """ start = time.time() # Always return at least one item. items = [queue_.get()] while max_items is None or len(items) < max_items: try: elapsed = time.time() - start timeout = max(0, max_latency - elapsed) items.append(queue_.get(timeout=timeout)) except queue.Empty: break return items class QueueCallbackWorker(object): """A helper that executes a callback for items sent in a queue. Calls a blocking ``get()`` on the ``queue`` until it encounters :attr:`STOP`. Args: queue: A Queue instance, appropriate for crossing the concurrency boundary implemented by ``executor``. Items will be popped off (with a blocking ``get()``) until :attr:`STOP` is encountered. callback: A callback that can process items pulled off of the queue. Multiple items will be passed to the callback in batches. max_items: The maximum amount of items that will be passed to the callback at a time. max_latency: The maximum amount of time in seconds to wait for additional items before executing the callback. """ def __init__( self, queue: queue.Queue, callback: Callable[[Sequence[Any]], Any], max_items: int = 100, max_latency: float = 0, ): self.queue = queue self._callback = callback self.max_items = max_items self.max_latency = max_latency def __call__(self) -> None: continue_ = True while continue_: items = _get_many( self.queue, max_items=self.max_items, max_latency=self.max_latency ) # If stop is in the items, process all items up to STOP and then # exit. try: items = items[: items.index(STOP)] continue_ = False except ValueError: pass # Run the callback. If any exceptions occur, log them and # continue. try: self._callback(items) except Exception as exc: _LOGGER.exception("Error in queue callback worker: %s", exc) _LOGGER.debug("Exiting the QueueCallbackWorker.")	4.02 KB	2025-05-25 12:26
📄 histogram.py # Copyright 2017, Google LLC All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from typing import Dict, Optional, Union MIN_ACK_DEADLINE = 10 MAX_ACK_DEADLINE = 600 class Histogram(object): """Representation of a single histogram. The purpose of this class is to store actual ack timing information in order to predict how long to renew leases. The default implementation uses the 99th percentile of previous ack times to implicitly lease messages; however, custom :class:`~.pubsub_v1.subscriber._consumer.Consumer` subclasses are free to use a different formula. The precision of data stored is to the nearest integer. Additionally, values outside the range of ``MIN_ACK_DEADLINE <= x <= MAX_ACK_DEADLINE`` are stored as ``MIN_ACK_DEADLINE`` or ``MAX_ACK_DEADLINE``, since these are the boundaries of leases in the actual API. """ def __init__(self, data: Optional[Dict[int, int]] = None): """Instantiate the histogram. Args: data: The data strucure to be used to store the underlying data. The default is an empty dictionary. This can be set to a dictionary-like object if required (for example, if a special object is needed for concurrency reasons). """ # The data is stored as a dictionary, with the keys being the # value being added and the values being the number of times that # value was added to the dictionary. # # This is depending on the Python interpreter's implicit ordering # of dictionaries, which is a bitwise sort by the key's ``hash()`` # value. Because ``hash(int i) -> i`` and all of our keys are # positive integers (negatives would be a problem because the sort # is bitwise), we can rely on this. if data is None: data = {} self._data = data self._len = 0 def __len__(self) -> int: """Return the total number of data points in this histogram. This is cached on a separate counter (rather than computing it using ``sum([v for v in self._data.values()])``) to optimize lookup. Returns: The total number of data points in this histogram. """ return self._len def __contains__(self, needle: int) -> bool: """Return ``True`` if needle is present in the histogram, ``False`` otherwise.""" return needle in self._data def __repr__(self): return "<Histogram: {len} values between {min} and {max}>".format( len=len(self), max=self.max, min=self.min ) @property def max(self) -> int: """Return the maximum value in this histogram. If there are no values in the histogram at all, return ``MAX_ACK_DEADLINE``. Returns: The maximum value in the histogram. """ if len(self._data) == 0: return MAX_ACK_DEADLINE return next(iter(reversed(sorted(self._data.keys())))) @property def min(self) -> int: """Return the minimum value in this histogram. If there are no values in the histogram at all, return ``MIN_ACK_DEADLINE``. Returns: The minimum value in the histogram. """ if len(self._data) == 0: return MIN_ACK_DEADLINE return next(iter(sorted(self._data.keys()))) def add(self, value: Union[int, float]) -> None: """Add the value to this histogram. Args: value: The value. Values outside of ``MIN_ACK_DEADLINE <= x <= MAX_ACK_DEADLINE`` will be raised to ``MIN_ACK_DEADLINE`` or reduced to ``MAX_ACK_DEADLINE``. """ # If the value is out of bounds, bring it in bounds. value = int(value) if value < MIN_ACK_DEADLINE: value = MIN_ACK_DEADLINE elif value > MAX_ACK_DEADLINE: value = MAX_ACK_DEADLINE # Add the value to the histogram's data dictionary. self._data.setdefault(value, 0) self._data[value] += 1 self._len += 1 def percentile(self, percent: Union[int, float]) -> int: """Return the value that is the Nth precentile in the histogram. Args: percent: The precentile being sought. The default consumer implementations consistently use ``99``. Returns: The value corresponding to the requested percentile. """ # Sanity check: Any value over 100 should become 100. if percent >= 100: percent = 100 # Determine the actual target number. target = len(self) - len(self) * (percent / 100) # Iterate over the values in reverse, dropping the target by the # number of times each value has been seen. When the target passes # 0, return the value we are currently viewing. for k in reversed(sorted(self._data.keys())): target -= self._data[k] if target < 0: return k # The only way to get here is if there was no data. # In this case, just return the shortest possible deadline. return MIN_ACK_DEADLINE	5.68 KB	2025-05-25 12:26
📄 leaser.py # Copyright 2017, Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import absolute_import import copy import logging import random import threading import time import typing from typing import Dict, Iterable, Optional, Union from google.cloud.pubsub_v1.subscriber._protocol.dispatcher import _MAX_BATCH_LATENCY from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( SubscribeOpenTelemetry, ) try: from collections.abc import KeysView KeysView[None] # KeysView is only subscriptable in Python 3.9+ except TypeError: # Deprecated since Python 3.9, thus only use as a fallback in older Python versions from typing import KeysView from google.cloud.pubsub_v1.subscriber._protocol import requests if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud.pubsub_v1.subscriber._protocol.streaming_pull_manager import ( StreamingPullManager, ) _LOGGER = logging.getLogger(__name__) _LEASE_WORKER_NAME = "Thread-LeaseMaintainer" class _LeasedMessage(typing.NamedTuple): sent_time: float """The local time when ACK ID was initially leased in seconds since the epoch.""" size: int ordering_key: Optional[str] opentelemetry_data: Optional[SubscribeOpenTelemetry] class Leaser(object): def __init__(self, manager: "StreamingPullManager"): self._thread: Optional[threading.Thread] = None self._manager = manager # a lock used for start/stop operations, protecting the _thread attribute self._operational_lock = threading.Lock() # A lock ensuring that add/remove operations are atomic and cannot be # intertwined. Protects the _leased_messages and _bytes attributes. self._add_remove_lock = threading.Lock() # Dict of ack_id -> _LeasedMessage self._leased_messages: Dict[str, _LeasedMessage] = {} self._bytes = 0 """The total number of bytes consumed by leased messages.""" self._stop_event = threading.Event() @property def message_count(self) -> int: """The number of leased messages.""" return len(self._leased_messages) @property def ack_ids(self) -> KeysView[str]: """The ack IDs of all leased messages.""" return self._leased_messages.keys() @property def bytes(self) -> int: """The total size, in bytes, of all leased messages.""" return self._bytes def add(self, items: Iterable[requests.LeaseRequest]) -> None: """Add messages to be managed by the leaser.""" with self._add_remove_lock: for item in items: # Add the ack ID to the set of managed ack IDs, and increment # the size counter. if item.ack_id not in self._leased_messages: self._leased_messages[item.ack_id] = _LeasedMessage( sent_time=float("inf"), size=item.byte_size, ordering_key=item.ordering_key, opentelemetry_data=item.opentelemetry_data, ) self._bytes += item.byte_size else: _LOGGER.debug("Message %s is already lease managed", item.ack_id) def start_lease_expiry_timer(self, ack_ids: Iterable[str]) -> None: """Start the lease expiry timer for `items`. Args: items: Sequence of ack-ids for which to start lease expiry timers. """ with self._add_remove_lock: for ack_id in ack_ids: lease_info = self._leased_messages.get(ack_id) # Lease info might not exist for this ack_id because it has already # been removed by remove(). if lease_info: self._leased_messages[ack_id] = lease_info._replace( sent_time=time.time() ) def remove( self, items: Iterable[ Union[requests.AckRequest, requests.DropRequest, requests.NackRequest] ], ) -> None: """Remove messages from lease management.""" with self._add_remove_lock: # Remove the ack ID from lease management, and decrement the # byte counter. for item in items: if self._leased_messages.pop(item.ack_id, None) is not None: self._bytes -= item.byte_size else: _LOGGER.debug("Item %s was not managed.", item.ack_id) if self._bytes < 0: _LOGGER.debug("Bytes was unexpectedly negative: %d", self._bytes) self._bytes = 0 def maintain_leases(self) -> None: """Maintain all of the leases being managed. This method modifies the ack deadline for all of the managed ack IDs, then waits for most of that time (but with jitter), and repeats. """ while not self._stop_event.is_set(): # 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. # Also update the deadline currently used if enough new ACK data has been # gathered since the last deadline update. deadline = self._manager._obtain_ack_deadline(maybe_update=True) _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 # we're iterating over it. leased_messages = copy.copy(self._leased_messages) # Drop any leases that are beyond the max lease time. This ensures # that in the event of a badly behaving actor, we can drop messages # and allow the Pub/Sub server to resend them. cutoff = time.time() - self._manager.flow_control.max_lease_duration to_drop = [ requests.DropRequest(ack_id, item.size, item.ordering_key) for ack_id, item in leased_messages.items() if item.sent_time < cutoff ] if to_drop: _LOGGER.warning( "Dropping %s items because they were leased too long.", len(to_drop) ) assert self._manager.dispatcher is not None for drop_msg in to_drop: leased_message = leased_messages.get(drop_msg.ack_id) if leased_message and leased_message.opentelemetry_data: leased_message.opentelemetry_data.add_process_span_event( "expired" ) leased_message.opentelemetry_data.end_process_span() leased_message.opentelemetry_data.set_subscribe_span_result( "expired" ) leased_message.opentelemetry_data.end_subscribe_span() self._manager.dispatcher.drop(to_drop) # Remove dropped items from our copy of the leased messages (they # have already been removed from the real one by # self._manager.drop(), which calls self.remove()). for item in to_drop: leased_messages.pop(item.ack_id) # Create a modack request. # We do not actually call `modify_ack_deadline` over and over # because it is more efficient to make a single request. ack_ids = leased_messages.keys() expired_ack_ids = set() if ack_ids: _LOGGER.debug("Renewing lease for %d ack IDs.", len(ack_ids)) # NOTE: This may not work as expected if ``consumer.active`` # has changed since we checked it. An implementation # without any sort of race condition would require a # way for ``send_request`` to fail when the consumer # is inactive. assert self._manager.dispatcher is not None ack_id_gen = (ack_id for ack_id in ack_ids) opentelemetry_data = [ message.opentelemetry_data for message in list(leased_messages.values()) if message.opentelemetry_data ] expired_ack_ids = self._manager._send_lease_modacks( ack_id_gen, deadline, opentelemetry_data, ) start_time = time.time() # If exactly once delivery is enabled, we should drop all expired ack_ids from lease management. if self._manager._exactly_once_delivery_enabled() and len(expired_ack_ids): assert self._manager.dispatcher is not None for ack_id in expired_ack_ids: msg = leased_messages.get(ack_id) if msg and msg.opentelemetry_data: msg.opentelemetry_data.add_process_span_event("expired") msg.opentelemetry_data.end_process_span() msg.opentelemetry_data.set_subscribe_span_result("expired") msg.opentelemetry_data.end_subscribe_span() self._manager.dispatcher.drop( [ requests.DropRequest( ack_id, leased_messages.get(ack_id).size, # type: ignore leased_messages.get(ack_id).ordering_key, # type: ignore ) for ack_id in expired_ack_ids if ack_id in leased_messages ] ) # Now wait an appropriate period of time and do this again. # # We determine the appropriate period of time based on a random # period between: # minimum: MAX_BATCH_LATENCY (to prevent duplicate modacks being created in one batch) # maximum: 90% of the deadline # This maximum time attempts to prevent ack expiration before new lease modacks arrive at the server. # This use of jitter (http://bit.ly/2s2ekL7) helps decrease contention in cases # where there are many clients. # If we spent any time iterating over expired acks, we should subtract this from the deadline. snooze = random.uniform( _MAX_BATCH_LATENCY, (deadline * 0.9 - (time.time() - start_time)) ) _LOGGER.debug("Snoozing lease management for %f seconds.", snooze) self._stop_event.wait(timeout=snooze) _LOGGER.debug("%s exiting.", _LEASE_WORKER_NAME) def start(self) -> None: with self._operational_lock: if self._thread is not None: raise ValueError("Leaser is already running.") # Create and start the helper thread. self._stop_event.clear() thread = threading.Thread( name=_LEASE_WORKER_NAME, target=self.maintain_leases ) thread.daemon = True thread.start() _LOGGER.debug("Started helper thread %s", thread.name) self._thread = thread def stop(self) -> None: with self._operational_lock: self._stop_event.set() if self._thread is not None: # The thread should automatically exit when the consumer is # inactive. self._thread.join() self._thread = None	12.07 KB	2025-05-25 12:26
📄 messages_on_hold.py # Copyright 2020, Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import collections import logging import typing from typing import Any, Callable, Iterable, Optional if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud.pubsub_v1 import subscriber _LOGGER = logging.getLogger(__name__) class MessagesOnHold(object): """Tracks messages on hold by ordering key. Not thread-safe.""" def __init__(self): self._size = 0 # A FIFO queue for the messages that have been received from the server, # but not yet sent to the user callback. # Both ordered and unordered messages may be in this queue. Ordered # message state tracked in _pending_ordered_messages once ordered # messages are taken off this queue. # The tail of the queue is to the right side of the deque; the head is # to the left side. self._messages_on_hold = collections.deque() # Dict of ordering_key -> queue of ordered messages that have not been # delivered to the user. # All ordering keys in this collection have a message in flight. Once # that one is acked or nacked, the next message in the queue for that # ordering key will be sent. # If the queue is empty, it means there's a message for that key in # flight, but there are no pending messages. self._pending_ordered_messages = {} @property def size(self) -> int: """Return the number of messages on hold across ordered and unordered messages. Note that this object may still store information about ordered messages in flight even if size is zero. Returns: The size value. """ return self._size def get(self) -> Optional["subscriber.message.Message"]: """Gets a message from the on-hold queue. A message with an ordering key wont be returned if there's another message with the same key in flight. Returns: A message that hasn't been sent to the user yet or ``None`` if there are no messages available. """ while self._messages_on_hold: msg = self._messages_on_hold.popleft() if msg.ordering_key: pending_queue = self._pending_ordered_messages.get(msg.ordering_key) if pending_queue is None: # Create empty queue to indicate a message with the # ordering key is in flight. self._pending_ordered_messages[ msg.ordering_key ] = collections.deque() self._size = self._size - 1 return msg else: # Another message is in flight so add message to end of # queue for this ordering key. pending_queue.append(msg) else: # Unordered messages can be returned without any # restrictions. self._size = self._size - 1 return msg return None def put(self, message: "subscriber.message.Message") -> None: """Put a message on hold. Args: message: The message to put on hold. """ if message.opentelemetry_data: message.opentelemetry_data.start_subscribe_scheduler_span() self._messages_on_hold.append(message) self._size = self._size + 1 def activate_ordering_keys( self, ordering_keys: Iterable[str], schedule_message_callback: Callable[["subscriber.message.Message"], Any], ) -> None: """Send the next message in the queue for each of the passed-in ordering keys, if they exist. Clean up state for keys that no longer have any queued messages. See comment at streaming_pull_manager.activate_ordering_keys() for more detail about the impact of this method on load. Args: ordering_keys: The ordering keys to activate. May be empty, or contain duplicates. schedule_message_callback: The callback to call to schedule a message to be sent to the user. """ for key in ordering_keys: pending_ordered_messages = self._pending_ordered_messages.get(key) if pending_ordered_messages is None: _LOGGER.warning( "No message queue exists for message ordering key: %s.", key ) continue next_msg = self._get_next_for_ordering_key(key) if next_msg: # Schedule the next message because the previous was dropped. # Note that this may overload the user's `max_bytes` limit, but # not their `max_messages` limit. schedule_message_callback(next_msg) else: # No more messages for this ordering key, so do clean-up. self._clean_up_ordering_key(key) def _get_next_for_ordering_key( self, ordering_key: str ) -> Optional["subscriber.message.Message"]: """Get next message for ordering key. The client should call clean_up_ordering_key() if this method returns None. Args: ordering_key: Ordering key for which to get the next message. Returns: The next message for this ordering key or None if there aren't any. """ queue_for_key = self._pending_ordered_messages.get(ordering_key) if queue_for_key: self._size = self._size - 1 return queue_for_key.popleft() return None def _clean_up_ordering_key(self, ordering_key: str) -> None: """Clean up state for an ordering key with no pending messages. Args ordering_key: The ordering key to clean up. """ message_queue = self._pending_ordered_messages.get(ordering_key) if message_queue is None: _LOGGER.warning( "Tried to clean up ordering key that does not exist: %s", ordering_key ) return if len(message_queue) > 0: _LOGGER.warning( "Tried to clean up ordering key: %s with %d messages remaining.", ordering_key, len(message_queue), ) return del self._pending_ordered_messages[ordering_key]	6.84 KB	2025-05-25 12:26
📄 requests.py # Copyright 2017, Google LLC All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import typing from typing import NamedTuple, Optional from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( SubscribeOpenTelemetry, ) if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud.pubsub_v1.subscriber import futures # Namedtuples for management requests. Used by the Message class to communicate # items of work back to the policy. class AckRequest(NamedTuple): ack_id: str byte_size: int time_to_ack: float ordering_key: Optional[str] future: Optional["futures.Future"] opentelemetry_data: Optional[SubscribeOpenTelemetry] = None class DropRequest(NamedTuple): ack_id: str byte_size: int ordering_key: Optional[str] class LeaseRequest(NamedTuple): ack_id: str byte_size: int ordering_key: Optional[str] opentelemetry_data: Optional[SubscribeOpenTelemetry] = None class ModAckRequest(NamedTuple): ack_id: str seconds: float future: Optional["futures.Future"] opentelemetry_data: Optional[SubscribeOpenTelemetry] = None class NackRequest(NamedTuple): ack_id: str byte_size: int ordering_key: Optional[str] future: Optional["futures.Future"] opentelemetry_data: Optional[SubscribeOpenTelemetry] = None	1.8 KB	2025-05-25 12:26
📄 streaming_pull_manager.py # Copyright 2017, Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import division import collections import functools import itertools import logging import threading import typing from typing import Any, Dict, Callable, Iterable, List, Optional, Set, Tuple import uuid from opentelemetry import trace import grpc # type: ignore from google.api_core import bidi from google.api_core import exceptions from google.cloud.pubsub_v1 import types from google.cloud.pubsub_v1.subscriber._protocol import dispatcher from google.cloud.pubsub_v1.subscriber._protocol import heartbeater from google.cloud.pubsub_v1.subscriber._protocol import histogram from google.cloud.pubsub_v1.subscriber._protocol import leaser from google.cloud.pubsub_v1.subscriber._protocol import messages_on_hold from google.cloud.pubsub_v1.subscriber._protocol import requests from google.cloud.pubsub_v1.subscriber.exceptions import ( AcknowledgeError, AcknowledgeStatus, ) from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( SubscribeOpenTelemetry, ) import google.cloud.pubsub_v1.subscriber.message from google.cloud.pubsub_v1.subscriber import futures from google.cloud.pubsub_v1.subscriber.scheduler import ThreadScheduler from google.pubsub_v1 import types as gapic_types from grpc_status import rpc_status # type: ignore from google.rpc.error_details_pb2 import ErrorInfo # type: ignore from google.rpc import code_pb2 # type: ignore from google.rpc import status_pb2 from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( start_modack_span, ) if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud.pubsub_v1 import subscriber _LOGGER = logging.getLogger(__name__) _REGULAR_SHUTDOWN_THREAD_NAME = "Thread-RegularStreamShutdown" _RPC_ERROR_THREAD_NAME = "Thread-OnRpcTerminated" _RETRYABLE_STREAM_ERRORS = ( exceptions.DeadlineExceeded, exceptions.ServiceUnavailable, exceptions.InternalServerError, exceptions.Unknown, exceptions.GatewayTimeout, exceptions.Aborted, ) _TERMINATING_STREAM_ERRORS = (exceptions.Cancelled,) _MAX_LOAD = 1.0 """The load threshold above which to pause the incoming message stream.""" _RESUME_THRESHOLD = 0.8 """The load threshold below which to resume the incoming message stream.""" _MIN_ACK_DEADLINE_SECS_WHEN_EXACTLY_ONCE_ENABLED = 60 """The minimum ack_deadline, in seconds, for when exactly_once is enabled for a subscription. We do this to reduce premature ack expiration. """ _DEFAULT_STREAM_ACK_DEADLINE: float = 60 """The default stream ack deadline in seconds.""" _MAX_STREAM_ACK_DEADLINE: float = 600 """The maximum stream ack deadline in seconds.""" _MIN_STREAM_ACK_DEADLINE: float = 10 """The minimum stream ack deadline in seconds.""" _EXACTLY_ONCE_DELIVERY_TEMPORARY_RETRY_ERRORS = { code_pb2.DEADLINE_EXCEEDED, code_pb2.RESOURCE_EXHAUSTED, code_pb2.ABORTED, code_pb2.INTERNAL, code_pb2.UNAVAILABLE, } def _wrap_as_exception(maybe_exception: Any) -> BaseException: """Wrap an object as a Python exception, if needed. Args: maybe_exception: The object to wrap, usually a gRPC exception class. Returns: The argument itself if an instance of ``BaseException``, otherwise the argument represented as an instance of ``Exception`` (sub)class. """ if isinstance(maybe_exception, grpc.RpcError): return exceptions.from_grpc_error(maybe_exception) elif isinstance(maybe_exception, BaseException): return maybe_exception return Exception(maybe_exception) def _wrap_callback_errors( callback: Callable[["google.cloud.pubsub_v1.subscriber.message.Message"], Any], on_callback_error: Callable[[BaseException], Any], message: "google.cloud.pubsub_v1.subscriber.message.Message", ): """Wraps a user callback so that if an exception occurs the message is nacked. Args: callback: The user callback. message: The Pub/Sub message. """ try: if message.opentelemetry_data: message.opentelemetry_data.end_subscribe_concurrency_control_span() message.opentelemetry_data.start_process_span() callback(message) except BaseException as exc: # Note: the likelihood of this failing is extremely low. This just adds # a message to a queue, so if this doesn't work the world is in an # unrecoverable state and this thread should just bail. _LOGGER.exception( "Top-level exception occurred in callback while processing a message" ) message.nack() on_callback_error(exc) def _get_status( exc: exceptions.GoogleAPICallError, ) -> Optional["status_pb2.Status"]: if not exc.response: _LOGGER.debug("No response obj in errored RPC call.") return None try: return rpc_status.from_call(exc.response) # Possible "If the gRPC call’s code or details are inconsistent # with the status code and message inside of the # google.rpc.status.Status" except ValueError: _LOGGER.debug("ValueError when parsing ErrorInfo.", exc_info=True) return None def _get_ack_errors( exc: exceptions.GoogleAPICallError, ) -> Optional[Dict[str, str]]: status = _get_status(exc) if not status: _LOGGER.debug("Unable to get status of errored RPC.") return None for detail in status.details: info = ErrorInfo() if not (detail.Is(ErrorInfo.DESCRIPTOR) and detail.Unpack(info)): _LOGGER.debug("Unable to unpack ErrorInfo.") return None return info.metadata return None def _process_requests( error_status: Optional["status_pb2.Status"], ack_reqs_dict: Dict[str, requests.AckRequest], errors_dict: Optional[Dict[str, str]], ): """Process requests when exactly-once delivery is enabled by referring to error_status and errors_dict. The errors returned by the server in as `error_status` or in `errors_dict` are used to complete the request futures in `ack_reqs_dict` (with a success or exception) or to return requests for further retries. """ requests_completed = [] requests_to_retry = [] for ack_id in ack_reqs_dict: # Handle special errors returned for ack/modack RPCs via the ErrorInfo # sidecar metadata when exactly-once delivery is enabled. if errors_dict and ack_id in errors_dict: exactly_once_error = errors_dict[ack_id] if exactly_once_error.startswith("TRANSIENT_"): requests_to_retry.append(ack_reqs_dict[ack_id]) else: if exactly_once_error == "PERMANENT_FAILURE_INVALID_ACK_ID": exc = AcknowledgeError(AcknowledgeStatus.INVALID_ACK_ID, info=None) else: exc = AcknowledgeError(AcknowledgeStatus.OTHER, exactly_once_error) future = ack_reqs_dict[ack_id].future if future is not None: future.set_exception(exc) requests_completed.append(ack_reqs_dict[ack_id]) # Temporary GRPC errors are retried elif ( error_status and error_status.code in _EXACTLY_ONCE_DELIVERY_TEMPORARY_RETRY_ERRORS ): requests_to_retry.append(ack_reqs_dict[ack_id]) # Other GRPC errors are NOT retried elif error_status: if error_status.code == code_pb2.PERMISSION_DENIED: exc = AcknowledgeError(AcknowledgeStatus.PERMISSION_DENIED, info=None) elif error_status.code == code_pb2.FAILED_PRECONDITION: exc = AcknowledgeError(AcknowledgeStatus.FAILED_PRECONDITION, info=None) else: exc = AcknowledgeError(AcknowledgeStatus.OTHER, str(error_status)) future = ack_reqs_dict[ack_id].future if future is not None: future.set_exception(exc) requests_completed.append(ack_reqs_dict[ack_id]) # Since no error occurred, requests with futures are completed successfully. elif ack_reqs_dict[ack_id].future: future = ack_reqs_dict[ack_id].future # success assert future is not None future.set_result(AcknowledgeStatus.SUCCESS) requests_completed.append(ack_reqs_dict[ack_id]) # All other requests are considered completed. else: requests_completed.append(ack_reqs_dict[ack_id]) return requests_completed, requests_to_retry class StreamingPullManager(object): """The streaming pull manager coordinates pulling messages from Pub/Sub, leasing them, and scheduling them to be processed. Args: client: The subscriber client used to create this instance. subscription: The name of the subscription. The canonical format for this is ``projects/{project}/subscriptions/{subscription}``. flow_control: The flow control settings. scheduler: The scheduler to use to process messages. If not provided, a thread pool-based scheduler will be used. use_legacy_flow_control: If set to ``True``, flow control at the Cloud Pub/Sub server is disabled, though client-side flow control is still enabled. If set to ``False`` (default), both server-side and client-side flow control are enabled. await_callbacks_on_shutdown: If ``True``, the shutdown thread will wait until all scheduler threads terminate and only then proceed with shutting down the remaining running helper threads. If ``False`` (default), the shutdown thread will shut the scheduler down, but it will not wait for the currently executing scheduler threads to terminate. This setting affects when the on close callbacks get invoked, and consequently, when the StreamingPullFuture associated with the stream gets resolved. """ def __init__( self, client: "subscriber.Client", subscription: str, flow_control: types.FlowControl = types.FlowControl(), scheduler: Optional[ThreadScheduler] = None, use_legacy_flow_control: bool = False, await_callbacks_on_shutdown: bool = False, ): self._client = client self._subscription = subscription self._exactly_once_enabled = False self._flow_control = flow_control self._use_legacy_flow_control = use_legacy_flow_control self._await_callbacks_on_shutdown = await_callbacks_on_shutdown self._ack_histogram = histogram.Histogram() self._last_histogram_size = 0 self._stream_metadata = [ ["x-goog-request-params", "subscription=" + subscription] ] # If max_duration_per_lease_extension is the default # we set the stream_ack_deadline to the default of 60 if self._flow_control.max_duration_per_lease_extension == 0: self._stream_ack_deadline = _DEFAULT_STREAM_ACK_DEADLINE # We will not be able to extend more than the default minimum elif ( self._flow_control.max_duration_per_lease_extension < _MIN_STREAM_ACK_DEADLINE ): self._stream_ack_deadline = _MIN_STREAM_ACK_DEADLINE # Will not be able to extend past the max elif ( self._flow_control.max_duration_per_lease_extension > _MAX_STREAM_ACK_DEADLINE ): self._stream_ack_deadline = _MAX_STREAM_ACK_DEADLINE else: self._stream_ack_deadline = ( self._flow_control.max_duration_per_lease_extension ) self._ack_deadline = max( min( self._flow_control.min_duration_per_lease_extension, histogram.MAX_ACK_DEADLINE, ), histogram.MIN_ACK_DEADLINE, ) self._rpc: Optional[bidi.ResumableBidiRpc] = None self._callback: Optional[functools.partial] = None self._closing = threading.Lock() self._closed = False self._close_callbacks: List[Callable[["StreamingPullManager", Any], Any]] = [] # Guarded by self._exactly_once_enabled_lock self._send_new_ack_deadline = False # A shutdown thread is created on intentional shutdown. self._regular_shutdown_thread: Optional[threading.Thread] = None # Generate a random client id tied to this object. All streaming pull # connections (initial and re-connects) will then use the same client # id. Doing so lets the server establish affinity even across stream # disconncetions. self._client_id = str(uuid.uuid4()) if scheduler is None: self._scheduler: Optional[ThreadScheduler] = ThreadScheduler() else: self._scheduler = scheduler # A collection for the messages that have been received from the server, # but not yet sent to the user callback. self._messages_on_hold = messages_on_hold.MessagesOnHold() # The total number of bytes consumed by the messages currently on hold self._on_hold_bytes = 0 # A lock ensuring that pausing / resuming the consumer are both atomic # operations that cannot be executed concurrently. Needed for properly # syncing these operations with the current leaser load. Additionally, # the lock is used to protect modifications of internal data that # affects the load computation, i.e. the count and size of the messages # currently on hold. self._pause_resume_lock = threading.Lock() # A lock guarding the self._exactly_once_enabled variable. We may also # acquire the self._ack_deadline_lock while this lock is held, but not # the reverse. So, we maintain a simple ordering of these two locks to # prevent deadlocks. self._exactly_once_enabled_lock = threading.Lock() # A lock protecting the current ACK deadline used in the lease management. This # value can be potentially updated both by the leaser thread and by the message # consumer thread when invoking the internal _on_response() callback. self._ack_deadline_lock = threading.Lock() # The threads created in ``.open()``. self._dispatcher: Optional[dispatcher.Dispatcher] = None self._leaser: Optional[leaser.Leaser] = None self._consumer: Optional[bidi.BackgroundConsumer] = None self._heartbeater: Optional[heartbeater.Heartbeater] = None @property def is_active(self) -> bool: """``True`` if this manager is actively streaming. Note that ``False`` does not indicate this is complete shut down, just that it stopped getting new messages. """ return self._consumer is not None and self._consumer.is_active @property def flow_control(self) -> types.FlowControl: """The active flow control settings.""" return self._flow_control @property def dispatcher(self) -> Optional[dispatcher.Dispatcher]: """The dispatcher helper.""" return self._dispatcher @property def leaser(self) -> Optional[leaser.Leaser]: """The leaser helper.""" return self._leaser @property def ack_histogram(self) -> histogram.Histogram: """The histogram tracking time-to-acknowledge.""" return self._ack_histogram @property def ack_deadline(self) -> float: """Return the current ACK deadline based on historical data without updating it. Returns: The ack deadline. """ return self._obtain_ack_deadline(maybe_update=False) def _obtain_ack_deadline(self, maybe_update: bool) -> float: """The actual `ack_deadline` implementation. This method is "sticky". It will only perform the computations to check on the right ACK deadline if explicitly requested AND if the histogram with past time-to-ack data has gained a significant amount of new information. Args: maybe_update: If ``True``, also update the current ACK deadline before returning it if enough new ACK data has been gathered. Returns: The current ACK deadline in seconds to use. """ with self._ack_deadline_lock: if not maybe_update: return self._ack_deadline 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 self._ack_deadline = self.ack_histogram.percentile(percent=99) if self.flow_control.max_duration_per_lease_extension > 0: # The setting in flow control could be too low, adjust if needed. flow_control_setting = max( self.flow_control.max_duration_per_lease_extension, histogram.MIN_ACK_DEADLINE, ) self._ack_deadline = min(self._ack_deadline, flow_control_setting) # If the user explicitly sets a min ack_deadline, respect it. if self.flow_control.min_duration_per_lease_extension > 0: # The setting in flow control could be too high, adjust if needed. flow_control_setting = min( self.flow_control.min_duration_per_lease_extension, histogram.MAX_ACK_DEADLINE, ) self._ack_deadline = max(self._ack_deadline, flow_control_setting) elif self._exactly_once_enabled: # Higher minimum ack_deadline for subscriptions with # exactly-once delivery enabled. self._ack_deadline = max( self._ack_deadline, _MIN_ACK_DEADLINE_SECS_WHEN_EXACTLY_ONCE_ENABLED ) # If we have updated the ack_deadline and it is longer than the stream_ack_deadline # set the stream_ack_deadline to the new ack_deadline. if self._ack_deadline > self._stream_ack_deadline: self._stream_ack_deadline = self._ack_deadline return self._ack_deadline @property def load(self) -> float: """Return the current load. The load is represented as a float, where 1.0 represents having hit one of the flow control limits, and values between 0.0 and 1.0 represent how close we are to them. (0.5 means we have exactly half of what the flow control setting allows, for example.) There are (currently) two flow control settings; this property computes how close the manager is to each of them, and returns whichever value is higher. (It does not matter that we have lots of running room on setting A if setting B is over.) Returns: The load value. """ if self._leaser is None: return 0.0 # Messages that are temporarily put on hold are not being delivered to # user's callbacks, thus they should not contribute to the flow control # load calculation. # However, since these messages must still be lease-managed to avoid # unnecessary ACK deadline expirations, their count and total size must # be subtracted from the leaser's values. return max( [ (self._leaser.message_count - self._messages_on_hold.size) / self._flow_control.max_messages, (self._leaser.bytes - self._on_hold_bytes) / self._flow_control.max_bytes, ] ) def add_close_callback( self, callback: Callable[["StreamingPullManager", Any], Any] ) -> None: """Schedules a callable when the manager closes. Args: The method to call. """ self._close_callbacks.append(callback) def activate_ordering_keys(self, ordering_keys: Iterable[str]) -> None: """Send the next message in the queue for each of the passed-in ordering keys, if they exist. Clean up state for keys that no longer have any queued messages. Since the load went down by one message, it's probably safe to send the user another message for the same key. Since the released message may be bigger than the previous one, this may increase the load above the maximum. This decision is by design because it simplifies MessagesOnHold. Args: ordering_keys: A sequence of ordering keys to activate. May be empty. """ with self._pause_resume_lock: if self._scheduler is None: return # We are shutting down, don't try to dispatch any more messages. self._messages_on_hold.activate_ordering_keys( ordering_keys, self._schedule_message_on_hold ) def maybe_pause_consumer(self) -> None: """Check the current load and pause the consumer if needed.""" with self._pause_resume_lock: if self.load >= _MAX_LOAD: if self._consumer is not None and not self._consumer.is_paused: _LOGGER.debug( "Message backlog over load at %.2f, pausing.", self.load ) self._consumer.pause() def maybe_resume_consumer(self) -> None: """Check the load and held messages and resume the consumer if needed. If there are messages held internally, release those messages before resuming the consumer. That will avoid leaser overload. """ with self._pause_resume_lock: # If we have been paused by flow control, check and see if we are # back within our limits. # # In order to not thrash too much, require us to have passed below # the resume threshold (80% by default) of each flow control setting # before restarting. if self._consumer is None or not self._consumer.is_paused: return _LOGGER.debug("Current load: %.2f", self.load) # Before maybe resuming the background consumer, release any messages # currently on hold, if the current load allows for it. self._maybe_release_messages() if self.load < _RESUME_THRESHOLD: _LOGGER.debug("Current load is %.2f, resuming consumer.", self.load) self._consumer.resume() else: _LOGGER.debug("Did not resume, current load is %.2f.", self.load) def _maybe_release_messages(self) -> None: """Release (some of) the held messages if the current load allows for it. The method tries to release as many messages as the current leaser load would allow. Each released message is added to the lease management, and the user callback is scheduled for it. If there are currently no messages on hold, or if the leaser is already overloaded, this method is effectively a no-op. The method assumes the caller has acquired the ``_pause_resume_lock``. """ released_ack_ids = [] while self.load < _MAX_LOAD: msg = self._messages_on_hold.get() if not msg: break if msg.opentelemetry_data: msg.opentelemetry_data.end_subscribe_scheduler_span() self._schedule_message_on_hold(msg) released_ack_ids.append(msg.ack_id) assert self._leaser is not None self._leaser.start_lease_expiry_timer(released_ack_ids) def _schedule_message_on_hold( self, msg: "google.cloud.pubsub_v1.subscriber.message.Message" ): """Schedule a message on hold to be sent to the user and change on-hold-bytes. The method assumes the caller has acquired the ``_pause_resume_lock``. Args: msg: The message to schedule to be sent to the user. """ assert msg, "Message must not be None." # On-hold bytes goes down, increasing load. self._on_hold_bytes -= msg.size if self._on_hold_bytes < 0: _LOGGER.warning( "On hold bytes was unexpectedly negative: %s", self._on_hold_bytes ) self._on_hold_bytes = 0 _LOGGER.debug( "Released held message, scheduling callback for it, " "still on hold %s (bytes %s).", self._messages_on_hold.size, self._on_hold_bytes, ) assert self._scheduler is not None assert self._callback is not None if msg.opentelemetry_data: msg.opentelemetry_data.start_subscribe_concurrency_control_span() self._scheduler.schedule(self._callback, msg) def send_unary_ack( self, ack_ids, ack_reqs_dict ) -> Tuple[List[requests.AckRequest], List[requests.AckRequest]]: """Send a request using a separate unary request instead of over the stream. If a RetryError occurs, the manager shutdown is triggered, and the error is re-raised. """ assert ack_ids assert len(ack_ids) == len(ack_reqs_dict) error_status = None ack_errors_dict = None try: self._client.acknowledge(subscription=self._subscription, ack_ids=ack_ids) except exceptions.GoogleAPICallError as exc: _LOGGER.debug( "Exception while sending unary RPC. This is typically " "non-fatal as stream requests are best-effort.", exc_info=True, ) error_status = _get_status(exc) ack_errors_dict = _get_ack_errors(exc) except exceptions.RetryError as exc: exactly_once_delivery_enabled = self._exactly_once_delivery_enabled() # Makes sure to complete futures so they don't block forever. for req in ack_reqs_dict.values(): # Futures may be present even with exactly-once delivery # disabled, in transition periods after the setting is changed on # the subscription. if req.future: if exactly_once_delivery_enabled: e = AcknowledgeError( AcknowledgeStatus.OTHER, "RetryError while sending ack RPC." ) req.future.set_exception(e) else: req.future.set_result(AcknowledgeStatus.SUCCESS) _LOGGER.debug( "RetryError while sending ack RPC. Waiting on a transient " "error resolution for too long, will now trigger shutdown.", exc_info=False, ) # The underlying channel has been suffering from a retryable error # for too long, time to give up and shut the streaming pull down. self._on_rpc_done(exc) raise if self._exactly_once_delivery_enabled(): requests_completed, requests_to_retry = _process_requests( error_status, ack_reqs_dict, ack_errors_dict ) else: requests_completed = [] requests_to_retry = [] # When exactly-once delivery is NOT enabled, acks/modacks are considered # best-effort. So, they always succeed even if the RPC fails. for req in ack_reqs_dict.values(): # Futures may be present even with exactly-once delivery # disabled, in transition periods after the setting is changed on # the subscription. if req.future: req.future.set_result(AcknowledgeStatus.SUCCESS) requests_completed.append(req) return requests_completed, requests_to_retry def send_unary_modack( self, modify_deadline_ack_ids, modify_deadline_seconds, ack_reqs_dict, default_deadline=None, ) -> Tuple[List[requests.ModAckRequest], List[requests.ModAckRequest]]: """Send a request using a separate unary request instead of over the stream. If a RetryError occurs, the manager shutdown is triggered, and the error is re-raised. """ assert modify_deadline_ack_ids # Either we have a generator or a single deadline. assert modify_deadline_seconds is None or default_deadline is None error_status = None modack_errors_dict = None try: if default_deadline is None: # Send ack_ids with the same deadline seconds together. deadline_to_ack_ids = collections.defaultdict(list) for n, ack_id in enumerate(modify_deadline_ack_ids): deadline = modify_deadline_seconds[n] deadline_to_ack_ids[deadline].append(ack_id) for deadline, ack_ids in deadline_to_ack_ids.items(): self._client.modify_ack_deadline( subscription=self._subscription, ack_ids=ack_ids, ack_deadline_seconds=deadline, ) else: # We can send all requests with the default deadline. self._client.modify_ack_deadline( subscription=self._subscription, ack_ids=modify_deadline_ack_ids, ack_deadline_seconds=default_deadline, ) except exceptions.GoogleAPICallError as exc: _LOGGER.debug( "Exception while sending unary RPC. This is typically " "non-fatal as stream requests are best-effort.", exc_info=True, ) error_status = _get_status(exc) modack_errors_dict = _get_ack_errors(exc) except exceptions.RetryError as exc: exactly_once_delivery_enabled = self._exactly_once_delivery_enabled() # Makes sure to complete futures so they don't block forever. for req in ack_reqs_dict.values(): # Futures may be present even with exactly-once delivery # disabled, in transition periods after the setting is changed on # the subscription. if req.future: if exactly_once_delivery_enabled: e = AcknowledgeError( AcknowledgeStatus.OTHER, "RetryError while sending modack RPC.", ) req.future.set_exception(e) else: req.future.set_result(AcknowledgeStatus.SUCCESS) _LOGGER.debug( "RetryError while sending modack RPC. Waiting on a transient " "error resolution for too long, will now trigger shutdown.", exc_info=False, ) # The underlying channel has been suffering from a retryable error # for too long, time to give up and shut the streaming pull down. self._on_rpc_done(exc) raise if self._exactly_once_delivery_enabled(): requests_completed, requests_to_retry = _process_requests( error_status, ack_reqs_dict, modack_errors_dict ) else: requests_completed = [] requests_to_retry = [] # When exactly-once delivery is NOT enabled, acks/modacks are considered # best-effort. So, they always succeed even if the RPC fails. for req in ack_reqs_dict.values(): # Futures may be present even with exactly-once delivery # disabled, in transition periods after the setting is changed on # the subscription. if req.future: req.future.set_result(AcknowledgeStatus.SUCCESS) requests_completed.append(req) return requests_completed, requests_to_retry def heartbeat(self) -> bool: """Sends a heartbeat request over the streaming pull RPC. The request is empty by default, but may contain the current ack_deadline if the self._exactly_once_enabled flag has changed. Returns: If a heartbeat request has actually been sent. """ if self._rpc is not None and self._rpc.is_active: send_new_ack_deadline = False with self._exactly_once_enabled_lock: send_new_ack_deadline = self._send_new_ack_deadline self._send_new_ack_deadline = False if send_new_ack_deadline: request = gapic_types.StreamingPullRequest( stream_ack_deadline_seconds=self._stream_ack_deadline ) _LOGGER.debug( "Sending new ack_deadline of %d seconds.", self._stream_ack_deadline ) else: request = gapic_types.StreamingPullRequest() self._rpc.send(request) return True return False def open( self, callback: Callable[["google.cloud.pubsub_v1.subscriber.message.Message"], Any], on_callback_error: Callable[[Exception], Any], ) -> None: """Begin consuming messages. Args: callback: A callback that will be called for each message received on the stream. on_callback_error: A callable that will be called if an exception is raised in the provided `callback`. """ if self.is_active: raise ValueError("This manager is already open.") if self._closed: raise ValueError("This manager has been closed and can not be re-used.") self._callback = functools.partial( _wrap_callback_errors, callback, on_callback_error ) # Create the RPC stream_ack_deadline_seconds = self._stream_ack_deadline get_initial_request = functools.partial( self._get_initial_request, stream_ack_deadline_seconds ) self._rpc = bidi.ResumableBidiRpc( start_rpc=self._client.streaming_pull, initial_request=get_initial_request, should_recover=self._should_recover, should_terminate=self._should_terminate, metadata=self._stream_metadata, throttle_reopen=True, ) self._rpc.add_done_callback(self._on_rpc_done) _LOGGER.debug( "Creating a stream, default ACK deadline set to {} seconds.".format( self._stream_ack_deadline ) ) # Create references to threads assert self._scheduler is not None scheduler_queue = self._scheduler.queue self._dispatcher = dispatcher.Dispatcher(self, scheduler_queue) self._consumer = bidi.BackgroundConsumer(self._rpc, self._on_response) self._leaser = leaser.Leaser(self) self._heartbeater = heartbeater.Heartbeater(self) # Start the thread to pass the requests. self._dispatcher.start() # Start consuming messages. self._consumer.start() # Start the lease maintainer thread. self._leaser.start() # Start the stream heartbeater thread. self._heartbeater.start() def close(self, reason: Any = None) -> None: """Stop consuming messages and shutdown all helper threads. This method is idempotent. Additional calls will have no effect. The method does not block, it delegates the shutdown operations to a background thread. Args: reason: The reason to close this. If ``None``, this is considered an "intentional" shutdown. This is passed to the callbacks specified via :meth:`add_close_callback`. """ self._regular_shutdown_thread = threading.Thread( name=_REGULAR_SHUTDOWN_THREAD_NAME, daemon=True, target=self._shutdown, kwargs={"reason": reason}, ) self._regular_shutdown_thread.start() def _shutdown(self, reason: Any = None) -> None: """Run the actual shutdown sequence (stop the stream and all helper threads). Args: reason: The reason to close the stream. If ``None``, this is considered an "intentional" shutdown. """ with self._closing: if self._closed: return # Stop consuming messages. if self.is_active: _LOGGER.debug("Stopping consumer.") assert self._consumer is not None self._consumer.stop() self._consumer = None # Shutdown all helper threads _LOGGER.debug("Stopping scheduler.") assert self._scheduler is not None dropped_messages = self._scheduler.shutdown( await_msg_callbacks=self._await_callbacks_on_shutdown ) self._scheduler = None # Leaser and dispatcher reference each other through the shared # StreamingPullManager instance, i.e. "self", thus do not set their # references to None until both have been shut down. # # NOTE: Even if the dispatcher operates on an inactive leaser using # the latter's add() and remove() methods, these have no impact on # the stopped leaser (the leaser is never again re-started). Ditto # for the manager's maybe_resume_consumer() / maybe_pause_consumer(), # because the consumer gets shut down first. _LOGGER.debug("Stopping leaser.") assert self._leaser is not None self._leaser.stop() total = len(dropped_messages) + len( self._messages_on_hold._messages_on_hold ) _LOGGER.debug(f"NACK-ing all not-yet-dispatched messages (total: {total}).") messages_to_nack = itertools.chain( dropped_messages, self._messages_on_hold._messages_on_hold ) for msg in messages_to_nack: msg.nack() _LOGGER.debug("Stopping dispatcher.") assert self._dispatcher is not None self._dispatcher.stop() self._dispatcher = None # dispatcher terminated, OK to dispose the leaser reference now self._leaser = None _LOGGER.debug("Stopping heartbeater.") assert self._heartbeater is not None self._heartbeater.stop() self._heartbeater = None self._rpc = None self._closed = True _LOGGER.debug("Finished stopping manager.") for callback in self._close_callbacks: callback(self, reason) def _get_initial_request( self, stream_ack_deadline_seconds: int ) -> gapic_types.StreamingPullRequest: """Return the initial request for the RPC. This defines the initial request that must always be sent to Pub/Sub immediately upon opening the subscription. Args: stream_ack_deadline_seconds: The default message acknowledge deadline for the stream. Returns: A request suitable for being the first request on the stream (and not suitable for any other purpose). """ # Put the request together. # We need to set streaming ack deadline, but it's not useful since we'll modack to send receipt # anyway. Set to some big-ish value in case we modack late. request = gapic_types.StreamingPullRequest( stream_ack_deadline_seconds=stream_ack_deadline_seconds, modify_deadline_ack_ids=[], modify_deadline_seconds=[], subscription=self._subscription, client_id=self._client_id, max_outstanding_messages=( 0 if self._use_legacy_flow_control else self._flow_control.max_messages ), max_outstanding_bytes=( 0 if self._use_legacy_flow_control else self._flow_control.max_bytes ), ) # Return the initial request. return request def _send_lease_modacks( self, ack_ids: Iterable[str], ack_deadline: float, opentelemetry_data: List[SubscribeOpenTelemetry], warn_on_invalid=True, receipt_modack: bool = False, ) -> Set[str]: exactly_once_enabled = False modack_span: Optional[trace.Span] = None if self._client.open_telemetry_enabled: subscribe_span_links: List[trace.Link] = [] subscribe_spans: List[trace.Span] = [] subscription_split: List[str] = self._subscription.split("/") assert len(subscription_split) == 4 subscription_id: str = subscription_split[3] project_id: str = subscription_split[1] for data in opentelemetry_data: subscribe_span: Optional[trace.Span] = data.subscribe_span if ( subscribe_span and subscribe_span.get_span_context().trace_flags.sampled ): subscribe_span_links.append( trace.Link(subscribe_span.get_span_context()) ) subscribe_spans.append(subscribe_span) modack_span = start_modack_span( subscribe_span_links, subscription_id, len(opentelemetry_data), ack_deadline, project_id, "_send_lease_modacks", receipt_modack, ) if ( modack_span and modack_span.get_span_context().trace_flags.sampled ): # pragma: NO COVER modack_span_context: trace.SpanContext = modack_span.get_span_context() for subscribe_span in subscribe_spans: subscribe_span.add_link( context=modack_span_context, attributes={ "messaging.operation.name": "modack", }, ) with self._exactly_once_enabled_lock: exactly_once_enabled = self._exactly_once_enabled if exactly_once_enabled: eod_items: List[requests.ModAckRequest] = [] if self._client.open_telemetry_enabled: for ack_id, data in zip( ack_ids, opentelemetry_data ): # pragma: NO COVER # Identical code covered in the same function below assert data is not None eod_items.append( requests.ModAckRequest( ack_id, ack_deadline, futures.Future(), data, ) ) else: eod_items = [ requests.ModAckRequest(ack_id, ack_deadline, futures.Future()) for ack_id in ack_ids ] assert self._dispatcher is not None self._dispatcher.modify_ack_deadline(eod_items, ack_deadline) if ( modack_span ): # pragma: NO COVER # Identical code covered in the same function below modack_span.end() expired_ack_ids = set() for req in eod_items: try: assert req.future is not None req.future.result() except AcknowledgeError as ack_error: if ( ack_error.error_code != AcknowledgeStatus.INVALID_ACK_ID or warn_on_invalid ): _LOGGER.warning( "AcknowledgeError when lease-modacking a message.", exc_info=True, ) if ack_error.error_code == AcknowledgeStatus.INVALID_ACK_ID: expired_ack_ids.add(req.ack_id) return expired_ack_ids else: items: List[requests.ModAckRequest] = [] if self._client.open_telemetry_enabled: for ack_id, data in zip(ack_ids, opentelemetry_data): assert data is not None items.append( requests.ModAckRequest( ack_id, self.ack_deadline, None, data, ) ) else: items = [ requests.ModAckRequest(ack_id, self.ack_deadline, None) for ack_id in ack_ids ] assert self._dispatcher is not None self._dispatcher.modify_ack_deadline(items, ack_deadline) if modack_span: modack_span.end() return set() def _exactly_once_delivery_enabled(self) -> bool: """Whether exactly-once delivery is enabled for the subscription.""" with self._exactly_once_enabled_lock: return self._exactly_once_enabled def _on_response(self, response: gapic_types.StreamingPullResponse) -> None: """Process all received Pub/Sub messages. For each message, send a modified acknowledgment request to the server. This prevents expiration of the message due to buffering by gRPC or proxy/firewall. This makes the server and client expiration timer closer to each other thus preventing the message being redelivered multiple times. After the messages have all had their ack deadline updated, execute the callback for each message using the executor. """ if response is None: _LOGGER.debug( "Response callback invoked with None, likely due to a " "transport shutdown." ) return # IMPORTANT: Circumvent the wrapper class and operate on the raw underlying # protobuf message to significantly gain on attribute access performance. received_messages = response._pb.received_messages subscribe_opentelemetry: List[SubscribeOpenTelemetry] = [] if self._client.open_telemetry_enabled: for received_message in received_messages: opentelemetry_data = SubscribeOpenTelemetry(received_message.message) opentelemetry_data.start_subscribe_span( self._subscription, response.subscription_properties.exactly_once_delivery_enabled, received_message.ack_id, received_message.delivery_attempt, ) subscribe_opentelemetry.append(opentelemetry_data) _LOGGER.debug( "Processing %s received message(s), currently on hold %s (bytes %s).", len(received_messages), self._messages_on_hold.size, self._on_hold_bytes, ) with self._exactly_once_enabled_lock: if ( response.subscription_properties.exactly_once_delivery_enabled != self._exactly_once_enabled ): self._exactly_once_enabled = ( response.subscription_properties.exactly_once_delivery_enabled ) # Update ack_deadline, whose minimum depends on self._exactly_once_enabled # This method acquires the self._ack_deadline_lock lock. self._obtain_ack_deadline(maybe_update=True) self._send_new_ack_deadline = True # Immediately (i.e. without waiting for the auto lease management) # modack the messages we received, as this tells the server that we've # received them. ack_id_gen = (message.ack_id for message in received_messages) expired_ack_ids = self._send_lease_modacks( ack_id_gen, self.ack_deadline, subscribe_opentelemetry, warn_on_invalid=False, receipt_modack=True, ) with self._pause_resume_lock: if self._scheduler is None or self._leaser is None: _LOGGER.debug( f"self._scheduler={self._scheduler} or self._leaser={self._leaser} is None. Stopping further processing." ) return i: int = 0 for received_message in received_messages: if ( not self._exactly_once_delivery_enabled() or received_message.ack_id not in expired_ack_ids ): message = google.cloud.pubsub_v1.subscriber.message.Message( received_message.message, received_message.ack_id, received_message.delivery_attempt, self._scheduler.queue, self._exactly_once_delivery_enabled, ) if self._client.open_telemetry_enabled: message.opentelemetry_data = subscribe_opentelemetry[i] i = i + 1 self._messages_on_hold.put(message) self._on_hold_bytes += message.size req = requests.LeaseRequest( ack_id=message.ack_id, byte_size=message.size, ordering_key=message.ordering_key, opentelemetry_data=message.opentelemetry_data, ) self._leaser.add([req]) self._maybe_release_messages() self.maybe_pause_consumer() def _should_recover(self, exception: BaseException) -> bool: """Determine if an error on the RPC stream should be recovered. If the exception is one of the retryable exceptions, this will signal to the consumer thread that it should "recover" from the failure. This will cause the stream to exit when it returns :data:`False`. Returns: Indicates if the caller should recover or shut down. Will be :data:`True` if the ``exception`` is "acceptable", i.e. in a list of retryable / idempotent exceptions. """ exception = _wrap_as_exception(exception) # If this is in the list of idempotent exceptions, then we want to # recover. if isinstance(exception, _RETRYABLE_STREAM_ERRORS): _LOGGER.debug("Observed recoverable stream error %s", exception) return True _LOGGER.debug("Observed non-recoverable stream error %s", exception) return False def _should_terminate(self, exception: BaseException) -> bool: """Determine if an error on the RPC stream should be terminated. If the exception is one of the terminating exceptions, this will signal to the consumer thread that it should terminate. This will cause the stream to exit when it returns :data:`True`. Returns: Indicates if the caller should terminate or attempt recovery. Will be :data:`True` if the ``exception`` is "acceptable", i.e. in a list of terminating exceptions. """ exception = _wrap_as_exception(exception) if isinstance(exception, _TERMINATING_STREAM_ERRORS): _LOGGER.debug("Observed terminating stream error %s", exception) return True _LOGGER.debug("Observed non-terminating stream error %s", exception) return False def _on_rpc_done(self, future: Any) -> None: """Triggered whenever the underlying RPC terminates without recovery. This is typically triggered from one of two threads: the background consumer thread (when calling ``recv()`` produces a non-recoverable error) or the grpc management thread (when cancelling the RPC). This method is non-blocking. It will start another thread to deal with shutting everything down. This is to prevent blocking in the background consumer and preventing it from being ``joined()``. """ _LOGGER.debug("RPC termination has signaled streaming pull manager shutdown.") error = _wrap_as_exception(future) thread = threading.Thread( name=_RPC_ERROR_THREAD_NAME, target=self._shutdown, kwargs={"reason": error} ) thread.daemon = True thread.start()	52.75 KB	2025-05-25 12:26

Terminal