diff --git a/ldclient/config.py b/ldclient/config.py index a58cc70b..5916943e 100644 --- a/ldclient/config.py +++ b/ldclient/config.py @@ -321,6 +321,7 @@ def __init__( omit_anonymous_contexts: bool = False, payload_filter_key: Optional[str] = None, datasystem_config: Optional[DataSystemConfig] = None, + shutdown_timeout: Optional[float] = 5, ): """ :param sdk_key: The SDK key for your LaunchDarkly account. This is always required. @@ -392,6 +393,12 @@ def __init__( :param omit_anonymous_contexts: Sets whether anonymous contexts should be omitted from index and identify events. :param payload_filter_key: The payload filter is used to selectively limited the flags and segments delivered in the data source payload. :param datasystem_config: Configuration for the upcoming enhanced data system design. This is experimental and should not be set without direction from LaunchDarkly support. + :param shutdown_timeout: The maximum number of seconds that :func:`ldclient.client.LDClient.close()` + will wait for pending analytics events to be delivered before giving up. Delivery of an event + payload can block for an unbounded time when name resolution hangs, because DNS lookups are not + covered by the connect and read timeouts in :class:`HTTPConfig`; without this limit, ``close()`` + would never return. Any events still undelivered when the timeout expires are discarded. Set this + to ``None`` to wait indefinitely, but be aware that doing so can prevent your process from exiting. """ self.__sdk_key = validate_sdk_key_format(sdk_key, log) @@ -429,6 +436,7 @@ def __init__( self.__enable_event_compression = enable_event_compression self.__omit_anonymous_contexts = omit_anonymous_contexts self.__payload_filter_key = payload_filter_key + self.__shutdown_timeout = None if shutdown_timeout is None else max(shutdown_timeout, 0) self._data_source_update_sink: Optional[DataSourceUpdateSink] = None self._instance_id: Optional[str] = None self._datasystem_config = datasystem_config @@ -644,6 +652,14 @@ def omit_anonymous_contexts(self) -> bool: """ return self.__omit_anonymous_contexts + @property + def shutdown_timeout(self) -> Optional[float]: + """ + The maximum number of seconds that :func:`ldclient.client.LDClient.close()` will wait for + pending analytics events to be delivered before giving up, or ``None`` to wait indefinitely. + """ + return self.__shutdown_timeout + @property def payload_filter_key(self) -> Optional[str]: """ diff --git a/ldclient/impl/events/event_processor.py b/ldclient/impl/events/event_processor.py index 20cf03a4..293b23b9 100644 --- a/ldclient/impl/events/event_processor.py +++ b/ldclient/impl/events/event_processor.py @@ -10,6 +10,7 @@ from collections import namedtuple from random import Random from threading import Event, Lock, Thread +from typing import Optional import urllib3 @@ -40,6 +41,21 @@ EventProcessorMessage = namedtuple('EventProcessorMessage', ['type', 'param']) +class _Deadline: + """ + Tracks how much of a time budget is left, so that a sequence of blocking waits can share a + single overall limit. A timeout of None means there is no limit. + """ + + def __init__(self, timeout: Optional[float]): + self._end = None if timeout is None else time.monotonic() + timeout + + def remaining(self) -> Optional[float]: + if self._end is None: + return None + return max(0.0, self._end - time.monotonic()) + + class EventPayloadSendTask: def __init__(self, http, config, formatter, payload, response_fn): self._http = http @@ -161,12 +177,21 @@ def _send_and_reset_diagnostics(self): self._diagnostic_flush_workers.execute(task.run) def _do_shutdown(self): + # Delivery of an event payload can block for an unbounded time - notably when name + # resolution hangs, which the connect and read timeouts do not cover - so these waits are + # bounded. Worker threads are daemons, so any that are still stuck do not keep the process + # alive; the events they were carrying are simply lost. + deadline = _Deadline(self._config.shutdown_timeout) + self._flush_workers.stop() - self._flush_workers.wait() + drained = self._flush_workers.wait(deadline.remaining()) if self._diagnostic_flush_workers: self._diagnostic_flush_workers.stop() - self._diagnostic_flush_workers.wait() + drained = self._diagnostic_flush_workers.wait(deadline.remaining()) and drained + + if not drained: + log.warning("Timed out waiting for analytics events to be delivered while shutting down; some events were dropped") if self._close_http: self._http.clear() @@ -188,6 +213,7 @@ def __init__(self, config, http=None, dispatcher_class=None, diagnostic_accumula self._close_lock = Lock() self._closed = False + self._shutdown_timeout = config.shutdown_timeout (dispatcher_class or EventDispatcher)(self._inbox, config, http, diagnostic_accumulator) @@ -208,8 +234,10 @@ def stop(self): self._diagnostic_event_timer.stop() self.flush() # Note that here we are not calling _post_to_inbox, because we *do* want to wait if the inbox - # is full; an orderly shutdown can't happen unless these messages are received. - self._post_message_and_wait('stop') + # is full; an orderly shutdown can't happen unless these messages are received. The wait is + # bounded, though, so that a stalled event delivery cannot block the caller forever. + if not self._post_message_and_wait('stop', self._shutdown_timeout): + log.warning("Timed out waiting for the event processor to shut down after %s seconds; some analytics events may not have been delivered" % self._shutdown_timeout) def _post_to_inbox(self, message): try: @@ -230,10 +258,25 @@ def _send_diagnostic(self): def _wait_until_inactive(self): self._post_message_and_wait('test_sync') - def _post_message_and_wait(self, type): + def _post_message_and_wait(self, type, timeout: Optional[float] = None) -> bool: + """ + Posts a message to the dispatcher and waits for it to be handled, for at most the given + number of seconds (None means wait indefinitely). Returns True if it was handled, or False + if the timeout elapsed while posting the message or while waiting for the reply. + """ reply = Event() - self._inbox.put(EventProcessorMessage(type, reply)) - reply.wait() + deadline = _Deadline(timeout) + try: + remaining = deadline.remaining() + if remaining is None: + self._inbox.put(EventProcessorMessage(type, reply)) + else: + # A zero timeout means "don't block at all" to Queue.put, so treat it as such + # rather than passing a value it would reject. + self._inbox.put(EventProcessorMessage(type, reply), block=remaining > 0, timeout=remaining or None) + except queue.Full: + return False + return reply.wait(deadline.remaining()) # These magic methods allow use of the "with" block in tests def __enter__(self): diff --git a/ldclient/impl/fixed_thread_pool.py b/ldclient/impl/fixed_thread_pool.py index 98c2ad05..b7ab3fe6 100644 --- a/ldclient/impl/fixed_thread_pool.py +++ b/ldclient/impl/fixed_thread_pool.py @@ -1,5 +1,7 @@ import queue +import time from threading import Event, Lock, Thread +from typing import Optional from ldclient.impl.util import log @@ -35,16 +37,25 @@ def execute(self, jobFn): return True """ - Waits until all currently busy worker threads have completed their jobs. + Waits until all currently busy worker threads have completed their jobs, or until the + specified number of seconds has elapsed. A timeout of None means to wait indefinitely. + Returns True if all jobs completed, or False if the timeout elapsed first. """ - def wait(self): + def wait(self, timeout: Optional[float] = None) -> bool: + deadline = None if timeout is None else time.monotonic() + timeout while True: with self._lock: if self._busy_count == 0: - return + return True self._event.clear() - self._event.wait() + if deadline is None: + self._event.wait() + continue + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + self._event.wait(remaining) """ Tells all the worker threads to terminate once all active jobs have completed. diff --git a/ldclient/testing/impl/events/test_event_processor.py b/ldclient/testing/impl/events/test_event_processor.py index 2c2f462b..59b13e36 100644 --- a/ldclient/testing/impl/events/test_event_processor.py +++ b/ldclient/testing/impl/events/test_event_processor.py @@ -2,7 +2,7 @@ import time import uuid from datetime import timedelta -from threading import Thread +from threading import Event, Thread from typing import Dict, Set import pytest @@ -27,7 +27,7 @@ from ldclient.migrations.types import Operation, Origin, Stage from ldclient.testing.builders import * from ldclient.testing.proxy_test_util import do_proxy_tests -from ldclient.testing.stub_util import MockHttp +from ldclient.testing.stub_util import MockHttp, MockResponse default_config = Config("fake_sdk_key") context = Context.builder('userkey').name('Red').build() @@ -727,6 +727,54 @@ def start_consuming_events(): assert had_no_more +def test_stop_returns_even_if_event_delivery_never_completes(): + """ + Sending an event payload can block for an unbounded time - name resolution is not covered by + the connect and read timeouts, so a hung resolver stalls a flush worker indefinitely. stop() + must give up after shutdown_timeout rather than wait on that worker, because anything it + blocks (notably LDClient.close(), which is commonly called from an atexit or interpreter + shutdown hook) would otherwise never return and the process would never exit. + """ + delivery_started = Event() + release_delivery = Event() + + def never_completes(): + delivery_started.set() + release_delivery.wait() + return MockResponse(200, {}) + + mock_http._response_func = never_completes + ep = DefaultTestProcessor(shutdown_timeout=0.5) + try: + ep.send_event(EventInputIdentify(timestamp, context)) + ep.flush() + assert delivery_started.wait(5), "the flush worker never started delivering the payload" + + stopped = Event() + Thread(target=lambda: (ep.stop(), stopped.set()), name="ldclient.testing.events.stopper", daemon=True).start() + + assert stopped.wait(10), "stop() never returned; it is waiting on a delivery that never completes" + finally: + release_delivery.set() + + +def test_stop_still_delivers_buffered_events(): + """ + Control for test_stop_returns_even_if_event_delivery_never_completes: bounding the shutdown + wait must not turn stop() into a no-op. When delivery works normally, stop() still flushes + what is buffered before returning. + """ + ep = DefaultTestProcessor(shutdown_timeout=5) + e = EventInputIdentify(timestamp, context) + ep.send_event(e) + ep.stop() + + assert mock_http.request_data is not None, "stop() returned without delivering the buffered event" + output = json.loads(mock_http.request_data) + assert len(output) == 1 + check_identify_event(output[0], e) + + def test_http_proxy(monkeypatch): def _event_processor_proxy_test(server, config, secure): with DefaultEventProcessor(config) as ep: