From e34f76f26916a041e6a3ea46450e80aac20f9da4 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Thu, 23 Apr 2026 09:08:13 +0300 Subject: [PATCH] perf: move request submission off event loop thread in execute_concurrent ConcurrentExecutorListResults now uses a dedicated submitter thread instead of calling _execute_next inline from the event loop callback. This decouples I/O completion processing from new request serialization and enqueuing, yielding ~6-9% higher write throughput. The callback signals a threading.Event; the submitter thread drains a deque and calls session.execute_async in batches. This avoids blocking the libev event loop thread with request preparation work (query plan, serialization, tablet lookup) that takes ~27us per request. The event-loop callback path is lock-free: it appends to a deque and sets an Event, with no Condition/Lock acquisition in the hot path. Follow-up fixes from a concurrency-focused self-review of this change: - _submitter_loop no longer lets an unexpected exception (e.g. the caller's statements_and_parameters iterable raising something other than StopIteration out of next()) kill the submitter thread silently. Nothing else ever advances _current/_exec_count or wakes _results() after the initial batch, so an unhandled exception there previously hung execute_concurrent() forever instead of raising. The loop body is now wrapped so such failures are recorded and force the waiter to stop waiting instead of hanging; added a regression test (test_submitter_thread_survives_broken_iterable) that hangs against the old code and passes against the fix. - Corrected the docstring in _put_result: it previously claimed the lock-free hot path "assumes a GIL build", which is inconsistent with this driver's CI matrix (which exercises free-threaded Python 3.14t for both the libev and asyncio reactors). The actual cross-thread visibility guarantee comes from the happens-before relationship established by Event.set()/.wait() (both backed by a real lock), not from GIL atomicity of the individual list/deque appends -- documented accordingly. No changes to the Eventlet/gevent story: the submitter thread is a plain threading.Thread signaled with a threading.Event, not concurrent.futures.ThreadPoolExecutor, so it does not hit the Python 3.7+/Eventlet ThreadPoolExecutor hang that Cluster._create_thread_pool_executor() exists to work around (that hang is specific to ThreadPoolExecutor's internal use of queue.SimpleQueue, which is a non-monkey-patchable C implementation). EventletConnection/GeventConnection both require full monkey-patching already, under which threading.Thread/Event become the green equivalents automatically. Ordering is unaffected: the returned list is still built from sorted(self._results_queue) keyed on the original statement index, exactly as before this change. --- cassandra/concurrent.py | 280 +++++++++++++++++++++++++++++++--- pyproject.toml | 1 + tests/unit/test_concurrent.py | 127 +++++++++++++++ 3 files changed, 390 insertions(+), 18 deletions(-) diff --git a/cassandra/concurrent.py b/cassandra/concurrent.py index 0e7bf794e0..d9634c9d97 100644 --- a/cassandra/concurrent.py +++ b/cassandra/concurrent.py @@ -13,10 +13,10 @@ # limitations under the License. -from collections import namedtuple +from collections import deque, namedtuple from heapq import heappush, heappop from itertools import cycle -from threading import Condition +from threading import Condition, Event, Thread from cassandra.cluster import ResultSet, EXEC_PROFILE_DEFAULT @@ -189,31 +189,275 @@ def _results(self): class ConcurrentExecutorListResults(_ConcurrentExecutor): _exception = None + _fatal_exception = None def execute(self, concurrency, fail_fast): self._exception = None - return super(ConcurrentExecutorListResults, self).execute(concurrency, fail_fast) + self._fatal_exception = None + self._submit_ready = deque() + self._submit_event = Event() + self._stop_event = Event() + self._exhausted = False + # Submit the initial batch from the calling thread (no contention + # yet -- the submitter thread is not started until afterward). + # Track whether the initial batch consumed all statements. + self._fail_fast = fail_fast + self._results_queue = [] + self._current = 0 + self._exec_count = 0 + with self._condition: + for n in range(concurrency): + if not self._execute_next(): + self._exhausted = True + break + # A statement can fail synchronously (e.g. execute_async + # raising, or its errback firing inline) while we're still + # dispatching this initial batch -- see ``_execute``. + # ``_put_result``'s fail-fast path records ``_exception`` + # right here (reentrantly, same thread/lock), so stop + # dispatching the rest of the batch immediately instead of + # continuing up to ``concurrency`` more statements the + # caller doesn't want. This check is race-free: it runs on + # the same thread that just set ``_exception``, under the + # same lock, with no other thread involved yet (the + # submitter is not started until ``_results()``). + if self._fail_fast and self._exception: + break + return self._results() + + def _results(self): + # Always start the submitter thread: it owns ``_current`` accounting + # (incrementing from drained completion signals) so the event-loop + # callback path can stay lock-free in the success case. Even when + # the iterator was fully consumed by the initial batch, the + # submitter still needs to run to record completions. + self._submitter = Thread(target=self._submitter_loop, + daemon=True, name="concurrent-submitter") + self._submitter.start() + + try: + with self._condition: + while not self._exhausted or self._current < self._exec_count: + # Check for an already-recorded fail-fast exception + # *before* waiting, not only after. ``_put_result``'s + # cold path may have set ``_exception`` and called + # ``notify()`` while we were still inside ``execute()`` + # dispatching the initial batch (i.e. before this + # thread ever reached ``wait()``) -- ``notify()`` only + # wakes threads already parked in ``wait()``, so that + # notification is silently dropped. Without this + # pre-check we would then block in ``wait()`` with no + # guarantee of another wakeup any time soon (the + # submitter only notifies again once the whole + # iterable is exhausted), so fail-fast would degrade + # into consuming the entire -- possibly unbounded -- + # iterable. Reading ``_exception``/``_fail_fast`` here + # is safe: both are only ever written while holding + # this same ``_condition`` lock, which we hold + # continuously in this loop except while inside + # ``wait()`` itself. + if self._exception and self._fail_fast: + break + self._condition.wait() + finally: + self._stop_event.set() + self._submit_event.set() # wake submitter so it sees the stop + self._submitter.join() + # A fatal error in the submitter thread itself (as opposed to a + # per-statement failure) means we can no longer trust that the + # remaining statements were ever dispatched or accounted for. + # Raise unconditionally -- silently returning a truncated/incomplete + # results list would violate the "one result per input statement" + # contract, and hanging (the alternative if this were left + # unraised) is worse. See _submitter_loop for what can trigger this. + if self._fatal_exception: + raise self._fatal_exception + if self._exception and self._fail_fast: + raise self._exception + return [r[1] for r in sorted(self._results_queue)] def _put_result(self, result, idx, success): + """Record a completion and signal the submitter thread. + + Called from the event-loop callback thread (or from the submitter + thread when execute_async raises synchronously). + + Hot path (success, not fail-fast): NO lock acquisition. We rely on + the submitter thread to bump ``_current`` from the drained signal + count under the same lock acquisition that bumps ``_exec_count``. + This removes ~0.5-1us of lock cost from every callback on the + event-loop thread. + + Note: ``self._results_queue.append`` and ``self._submit_ready.append`` + are individually safe under both the GIL and free-threaded builds + (PEP 703) -- CPython's list/deque append is atomic either way (a + per-object critical section protects it in free-threaded builds). + What actually needs to hold across threads is that the submitter + thread, once it observes a drained ``_submit_ready`` entry, also + observes the matching ``_results_queue`` append that happened + before it in this method. That ordering comes from + ``self._submit_event.set()``/``.wait()``: both are backed by a real + lock, so the ``set()`` here happens-after every write above it in + this function, and the submitter's ``wait()`` happens-after that + ``set()``. This holds under free-threaded Python too, which the + driver's CI exercises (see the "3.14t" jobs). + """ self._results_queue.append((idx, ExecutionResult(success, result))) - with self._condition: - self._current += 1 - if not success and self._fail_fast: + if not success and self._fail_fast: + # Cold path: take the lock to record the exception and wake + # the main thread immediately so it can stop waiting. + with self._condition: if not self._exception: self._exception = result self._condition.notify() - elif not self._execute_next() and self._current == self._exec_count: - self._condition.notify() - - def _results(self): - with self._condition: - while self._current < self._exec_count: - self._condition.wait() - if self._exception and self._fail_fast: - raise self._exception - if self._exception and self._fail_fast: # raise the exception even if there was no wait - raise self._exception - return [r[1] for r in sorted(self._results_queue)] + # Signal the submitter thread. It will: + # 1) bump _current under the lock from the drained signal count, + # 2) submit a replacement request, + # 3) notify _results() if all completions have arrived. + self._submit_ready.append(1) + self._submit_event.set() + + def _submitter_loop(self): + """Drain completion signals and submit follow-up requests. + + Runs on a dedicated thread so that the libev event-loop thread + only needs to do the lightweight ``deque.append`` + ``Event.set`` + in ``_put_result`` rather than the full execute_async cycle + (query-plan, borrow connection, serialise, enqueue). + + Owns ``_current`` accounting: each drained completion signal + increments ``_current`` by one under the same lock acquisition + that bumps ``_exec_count`` for the new batch. This keeps the + event-loop callback path lock-free in the success case. + """ + ready = self._submit_ready + ready_event = self._submit_event + stop_event = self._stop_event + enum_stmts = self._enum_statements + session = self.session + profile = self._execution_profile + on_success = self._on_success + on_error = self._on_error + condition = self._condition + try: + while not stop_event.is_set(): + ready_event.wait() + ready_event.clear() + # Drain all pending completion signals. + count = 0 + while True: + try: + ready.popleft() + count += 1 + except IndexError: + break + if count == 0: + continue + # Treat an already-recorded fail-fast exception exactly + # like a stop request: keep doing the accounting for + # completions that already happened, but stop pulling + # further statements from the caller's iterable and stop + # dispatching new requests. This must not rely on the + # main thread reaching ``_results()``'s ``finally`` and + # setting ``stop_event`` -- that can be delayed (or, pre + # fix, missed) -- so this loop checks ``_exception`` + # itself. ``_exception`` is only ever written under + # ``condition`` (see ``_put_result``'s cold path), so we + # read it under the same lock here to avoid racing that + # write (this matters on free-threaded builds too). + with condition: + fail_fast_stop = self._fail_fast and self._exception is not None + if stop_event.is_set() or fail_fast_stop: + # Main thread is shutting down (e.g. fail-fast). Do the + # accounting for already-completed requests but skip + # dispatching new ones. + with condition: + self._current += count + if self._exhausted and self._current >= self._exec_count: + condition.notify() + continue + if self._exhausted: + # No more statements to dispatch -- just account for the + # completions we just drained and notify the waiter if + # everything has caught up. + with condition: + self._current += count + if self._current >= self._exec_count: + condition.notify() + continue + # Submit follow-up requests directly (fast path). + # The iterator is only consumed from this thread (the initial + # batch was fully dispatched before this thread started). + # + # Pull statements from the iterator first, then bump _current + # and _exec_count for the entire batch in one lock acquisition, + # then dispatch. This avoids per-request lock overhead while + # ensuring _results() never sees _current >= _exec_count + # prematurely. + batch = [] + iterator_done = False + for _ in range(count): + try: + batch.append(next(enum_stmts)) + except StopIteration: + iterator_done = True + break + # Single lock acquisition: bump both _current (from the + # drained completion count) and _exec_count (from the new + # batch size) atomically. Setting _exhausted in the same + # critical section ensures the main thread never sees + # _exhausted=True with a stale _exec_count. + with condition: + self._current += count + self._exec_count += len(batch) + if iterator_done: + self._exhausted = True + # Wake the waiter if all completions have caught up. + if self._exhausted and self._current >= self._exec_count: + condition.notify() + fail_fast_stop = self._fail_fast and self._exception is not None + # Re-check after the lock release: a stop request or a + # fail-fast exception may have arrived while we were + # holding the lock; avoid dispatching requests we know + # will be discarded (they were already accounted for + # above, so skipping them here does not lose the + # accounting -- see the docstring note on _results() + # about the pre-wait exception check for why the main + # thread does not need this batch to ever "complete"). + if stop_event.is_set() or fail_fast_stop: + continue + for idx, (statement, params) in batch: + try: + future = session.execute_async(statement, params, + timeout=None, + execution_profile=profile) + args = (future, idx) + future.add_callbacks( + callback=on_success, callback_args=args, + errback=on_error, errback_args=args) + except Exception as exc: + # Record the failure directly. _put_result handles + # _current accounting and will enqueue another signal + # to _submit_ready -- but that is fine because the + # next drain will attempt another next(enum_stmts). + self._put_result(exc, idx, False) + except Exception as exc: + # Anything escaping the loop above (most plausibly the caller's + # statements_and_parameters iterable raising something other + # than StopIteration out of next(enum_stmts)) would otherwise + # kill this thread silently -- and since this thread is the + # only thing that ever advances _current/_exec_count or wakes + # _results() after the initial batch, the caller would then + # block in _results() forever. Record the failure and force + # the waiter's predicate to become true so it stops waiting on + # state we can no longer maintain, instead of hanging. + log.exception("concurrent-submitter thread aborted execute_concurrent early") + with condition: + if self._fatal_exception is None: + self._fatal_exception = exc + self._exhausted = True + self._current = self._exec_count + condition.notify() diff --git a/pyproject.toml b/pyproject.toml index 698ff4c37b..917381a301 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,6 +43,7 @@ auth-kerberos = [ [dependency-groups] dev = [ "pytest~=8.0", + "pytest-timeout", "PyYAML", "pure-sasl", "twisted[tls]", diff --git a/tests/unit/test_concurrent.py b/tests/unit/test_concurrent.py index d3888aa9de..549afa024a 100644 --- a/tests/unit/test_concurrent.py +++ b/tests/unit/test_concurrent.py @@ -307,3 +307,130 @@ def clear_callbacks(self): for success, result in results: assert not success assert result is error + + @pytest.mark.timeout(10) + def test_submitter_thread_survives_broken_iterable(self): + """ + ConcurrentExecutorListResults dispatches follow-up requests from a + dedicated submitter thread that pulls from the caller's + statements_and_parameters iterable. If that iterable raises + something other than StopIteration (e.g. a generator that blows up + mid-stream), the submitter thread must not die silently: nothing + else ever advances `_current`/`_exec_count` or wakes `_results()` + after the initial batch, so an unhandled exception there would hang + execute_concurrent() forever instead of surfacing an error. + """ + class ImmediateFuture: + _query_trace = None + _col_names = None + _col_types = None + has_more_pages = False + + def add_callbacks(self, callback, errback, + callback_args=(), callback_kwargs=None, + errback_args=(), errback_kwargs=None): + # Fire the success callback synchronously and immediately, + # forcing the submitter thread to keep pulling from the + # iterable well past the initial batch. + callback("row", *callback_args, **(callback_kwargs or {})) + + def clear_callbacks(self): + pass + + def broken_statements(): + for i in range(5): + yield ("SELECT 1", (i,)) + raise ValueError("boom from user iterable") + + mock_session = Mock() + mock_session.execute_async.side_effect = lambda *a, **kw: ImmediateFuture() + + with pytest.raises(ValueError, match="boom from user iterable"): + execute_concurrent(mock_session, broken_statements(), concurrency=2, + raise_on_first_error=False) + + @pytest.mark.timeout(10) + def test_fail_fast_stops_submitter_promptly(self): + """ + Regression test for a lost-wakeup + missing-check bug that broke the + ``raise_on_first_error=True`` ("fail-fast") contract. + + Two bugs combined to break fail-fast: + + 1. ``_put_result``'s fail-fast path takes ``self._condition`` and + calls ``notify()`` to wake the main thread up immediately -- but + if the failure happens synchronously during the *initial* batch + dispatched by ``execute()`` (which itself holds + ``self._condition`` while dispatching), the main thread has not + reached ``_results()``'s ``wait()`` yet. ``Condition.notify()`` + only wakes threads already parked in ``wait()``, so that + notification is silently dropped, and nothing else was checking + for the already-recorded exception before blocking. + 2. ``_submitter_loop``'s dispatch path only checked ``stop_event`` + before pulling more items from the caller's iterable and + dispatching them -- never ``_fail_fast``/``_exception`` -- so + even once the main thread noticed the failure, it had to wait + for the submitter thread to notice ``stop_event`` on its own + schedule. + + Net effect: fail-fast degraded into consuming (and dispatching) + the entire iterable instead of stopping right after the first + failure -- unbounded for a generator input. + + This test fails the very first dispatched statement synchronously + (so the failure is recorded during the initial batch, guaranteeing + the dropped-notify scenario rather than racing for it) against a + large iterable, and asserts that only a small, bounded number of + statements were ever pulled from it. Pre-fix, this either hangs + (caught by the ``pytest.mark.timeout`` guard) or consumes/dispatches + the entire iterable; post-fix, dispatch stops right after the first + failure. + """ + consumed = [] + TOTAL_STATEMENTS = 20000 + + def many_statements(): + for i in range(TOTAL_STATEMENTS): + consumed.append(i) + yield ("SELECT 1", (i,)) + + class ImmediateFuture: + _query_trace = None + _col_names = None + _col_types = None + has_more_pages = False + + def __init__(self, idx): + self._idx = idx + + def add_callbacks(self, callback, errback, + callback_args=(), callback_kwargs=None, + errback_args=(), errback_kwargs=None): + # Fail only the very first statement (idx 0); every other + # statement succeeds synchronously and immediately, so + # nothing but the fail-fast logic itself would ever stop + # dispatch. + if self._idx == 0: + errback(ValueError("boom on first statement"), *errback_args, + **(errback_kwargs or {})) + else: + callback("row", *callback_args, **(callback_kwargs or {})) + + def clear_callbacks(self): + pass + + def fake_execute_async(statement, params, timeout=None, execution_profile=None): + return ImmediateFuture(params[0]) + + mock_session = Mock() + mock_session.execute_async.side_effect = fake_execute_async + + with pytest.raises(ValueError, match="boom on first statement"): + execute_concurrent(mock_session, many_statements(), concurrency=5, + raise_on_first_error=True) + + assert len(consumed) < 100, ( + "fail-fast did not stop dispatch promptly: consumed %d of %d " + "statements from the iterable after the first failure" + % (len(consumed), TOTAL_STATEMENTS) + )