perf: move request submission off event loop thread in execute_concurrent

[mykaul]

Summary

• Moves execute_async() calls in execute_concurrent from the event-loop callback thread to a dedicated submitter thread
• The event loop callback now only appends to a deque and signals an Event, reducing per-callback overhead from ~27μs to ~100ns
• The submitter thread drains the deque in batches and calls execute_async(), which includes serialization — keeping that CPU work off the event loop

v2: Reduce per-request lock overhead in ResponseFuture

Second commit (7cae6a14e) reduces lock/synchronization cost per request in the execute_concurrent hot path:

1. Lazy Event creation: ResponseFuture._event starts as None instead of Event(). The Event is only materialized in result() (the synchronous path). For execute_concurrent, which never calls result() on individual futures, this eliminates ~620ns per request (351ns Event construction + 267ns Event.set()).

2. Merged add_callbacks(): Registers both callback and errback under a single _callback_lock acquisition instead of two separate lock/unlock cycles. Saves ~80ns per request.

3. _set_final_result / _set_final_exception: Capture _event reference under _callback_lock before calling .set() outside the lock. Skip .set() when Event was never created. Null-check callback/errback lists before building to_call tuple.

4. _wait_for_result(): Checks result availability under _callback_lock before creating Event — avoids Event creation entirely when the result arrived before the caller waits.

5. _on_speculative_execute: Checks _final_result/_final_exception directly instead of Event.is_set(), since Event may be None with lazy creation.

All changes are safe under both GIL and free-threaded (PEP 703) Python. No GIL assumptions.

Benchmark Results

On our vector ingestion benchmark (100K rows, 768-dim float32 vectors, ScyllaDB 2026.1.1):

• Stock master + execute_concurrent: ~7,500 rows/s
• Enhanced driver + this change: +6-9% throughput improvement (additive with Cython serializer gains)
• The improvement is modest because serialization still dominates; with Cython serializers reducing serialization cost, this change becomes more impactful

How It Works

• _ConcurrentExecutorBase spawns a daemon submitter thread alongside the existing callback mechanism
• Callbacks do deque.append(1); event.set() — minimal work on the hot path
• Submitter thread wakes on the event, drains pending count, and calls _execute_next() in a batch
• Thread-safe via collections.deque (atomic append/popleft in CPython) + threading.Event
• Graceful shutdown: sentinel None in deque signals the thread to exit; join() in wait()

Testing

• 642 unit tests pass, 0 failures
• All 10 existing test_concurrent.py unit tests pass
• Tested with real ScyllaDB cluster under sustained load (100K+ inserts)

[mykaul]

v2 changes: reduce per-request lock overhead in ResponseFuture

New commit 7cae6a14e on top of the submitter thread change. Focuses on reducing lock/synchronization cost per request in the execute_concurrent hot path.

Changes

1. Lazy Event creation (cluster.py): ResponseFuture._event starts as None instead of Event(). The Event is only materialized in _wait_for_result() (the synchronous result() path). For execute_concurrent, which never calls result() on individual futures, this eliminates ~620ns per request (351ns Event construction + 267ns Event.set()).

2. Merged add_callbacks() (cluster.py): Registers both callback and errback under a single _callback_lock acquisition instead of two separate lock/unlock cycles. Saves ~80ns per request.

3. _set_final_result / _set_final_exception (cluster.py): Capture _event reference under _callback_lock before calling .set() outside the lock. Skips .set() when Event was never created. Null-checks callback/errback lists before building to_call tuple. All safe under free-threaded Python (PEP 703).

4. _wait_for_result() (cluster.py): New extracted method. Checks result availability under _callback_lock before creating Event — avoids Event creation entirely when the result arrived before the caller waits. Thread-safe under both GIL and no-GIL.

5. _on_speculative_execute (cluster.py): Checks _final_result/_final_exception directly instead of relying on Event.is_set(), since Event may be None with lazy creation.

6. Properties / paging (cluster.py): warnings, custom_payload properties and start_fetching_next_page handle _event is None.

Design notes

• All changes are safe under both GIL and free-threaded (PEP 703) Python. No GIL assumptions.
• _callback_lock is reused as the synchronization point for lazy Event creation (no new locks).
• pool.py is not modified — an earlier attempt to optimize _stream_available_condition.notify() was reverted due to lost-wakeup risk under free-threaded Python.

Test results

• 643 unit tests pass
• 6 failures are all pre-existing (confirmed failing on origin/master):
  • 4× test_response_future — tablet= parameter mismatch from PR perf: optimize Tablet memory layout and per-query lookup speed #812
  • 1× test_batch_message_with_keyspace
  • 1× test_eq in test_resultset

[mykaul]

Correction on test failures: The 6 "pre-existing failures" reported in the v2 comment were all caused by a stale Cython .so in the working tree taking precedence over the .py source. After rebuilding (uv sync --reinstall-package scylla-driver):

642 passed, 0 failures, 8 skipped

No pre-existing failures. Clean test run.

[coderabbitai]

Important

Review skipped

Draft detected.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: e0890ee9-30c5-4ad8-b5fe-f2670710a9b1

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

---

_{Comment @coderabbitai help to get the list of available commands.}

[mykaul]

Rebased onto current origin/master and did a concurrency-focused self-review of the submitter-thread change (same care as a locking change, per the checklist we use for this kind of PR). Summary:

Eventlet/gevent executor-factory issue (the #637 class of bug): not present here. The submitter thread is a plain threading.Thread signaled with a threading.Event/collections.deque — it never touches concurrent.futures.ThreadPoolExecutor, so it doesn't go anywhere near Cluster._create_thread_pool_executor() and can't reintroduce the Python 3.7+/Eventlet hang that factory exists for. That hang is specific to ThreadPoolExecutor's internal queue.SimpleQueue, a non-monkey-patchable C implementation; threading.Thread/Event are plain names in the threading module, which EventletConnection/GeventConnection already require to be fully monkey-patched (eventlet.monkey_patch()/gevent.monkey.patch_all()), so they become the green equivalents automatically under those reactors.

Thread-safety: mostly sound (single-writer-under-lock for _current/_exec_count/_exhausted, and the deque+Event completion-signal channel is a correct level-triggered handoff with no lost wakeups), but found and fixed one real hang: _submitter_loop had no guard around its body, so if the caller's statements_and_parameters iterable raised anything other than StopIteration (e.g. a generator that blows up mid-stream), the submitter thread died silently — and since that thread is the only thing that ever advances _current/_exec_count or wakes _results() after the initial batch, execute_concurrent() would then hang forever instead of raising. Reproduced it directly (confirmed hang via pytest-timeout), fixed it by wrapping the loop so such failures are recorded and force the waiter to stop waiting, and added a regression test (test_submitter_thread_survives_broken_iterable) that hangs against the old code and passes against the fix. Also corrected a docstring that claimed the lock-free hot path "assumes a GIL build" — inconsistent with this repo's CI, which exercises free-threaded Python 3.14t for both reactors; the actual guarantee comes from the happens-before relationship of Event.set()/.wait(), not GIL atomicity, so I documented that instead.

Ordering: unaffected. The returned list is still built from sorted(self._results_queue) keyed on the original statement index, same as before this change.

Tests: tests/unit/test_concurrent.py (11 incl. the new regression test) and the full tests/unit/ suite (721 passed, 88 skipped, no failures) all green, no hangs. Also ran a few hundred iterations of an ad hoc multi-threaded stress harness (real worker threads firing callbacks concurrently with the submitter, both fail-fast and non-fail-fast paths) with no hangs or incorrect results.

Amended the fixes into the existing commit and force-pushed (8c883813a). No unresolved review threads to address. Keeping this as a draft.

[Copilot]

Pull request overview

Note

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Moves execute_concurrent() follow-up request submission off the event-loop callback thread and reduces per-request synchronization overhead in the hot path.

Changes:

• Adds a dedicated submitter thread that drains completion signals and submits subsequent execute_async() calls in batches.
• Makes the event-loop callback path lighter by replacing per-callback locking/submission work with deque.append() + Event.set().
• Adds a regression test to ensure user iterables that raise mid-stream surface errors (instead of hanging) when the submitter thread is involved.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
cassandra/concurrent.py	Introduces submitter thread + signaling to move submission/serialization work off the event-loop callback thread.
tests/unit/test_concurrent.py	Adds a test to ensure exceptions from a broken statements_and_parameters iterable propagate instead of causing a hang.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[coderabbitai]

Actionable comments posted: 2

🧹 Nitpick comments (2)

tests/unit/test_concurrent.py (1)

329-334: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Match ResponseFuture.add_callbacks's parameter order in the test double.

The real signature is add_callbacks(self, callback, errback, callback_args=(), ...) (cassandra/cluster.py:5674). The double reorders errback after callback_args, so it only works because concurrent.py happens to pass keywords; a future positional call would mis-bind instead of failing loudly.

♻️ Align the stub signature

-            def add_callbacks(self, callback, callback_args=(), callback_kwargs=None,
-                              errback=None, errback_args=(), errback_kwargs=None):
+            def add_callbacks(self, callback, errback,
+                              callback_args=(), callback_kwargs=None,
+                              errback_args=(), errback_kwargs=None):

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/unit/test_concurrent.py` around lines 329 - 334, Update the test
double’s add_callbacks method in tests/unit/test_concurrent.py to match
ResponseFuture.add_callbacks positional parameter order, placing errback
immediately after callback and retaining the remaining argument defaults. Keep
the synchronous callback behavior unchanged.

cassandra/concurrent.py (1)

254-258: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low value

Hot path still takes a lock via Event.set().

Event.set() acquires the event's internal lock on every callback, so the "NO lock acquisition" claim in the docstring isn't quite accurate. Skipping the redundant set() is safe because the submitter clears the event before draining, so any append that observes is_set() == True is still picked up by the in-flight or next drain.

♻️ Optional signaling tweak

         self._submit_ready.append(1)
-        self._submit_event.set()
+        if not self._submit_event.is_set():
+            self._submit_event.set()

Also applies to: 286-287

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@cassandra/concurrent.py` around lines 254 - 258, Update the hot-path callback
logic described near the concurrent submission/draining flow to avoid calling
Event.set() on successful non-fail-fast callbacks, since it acquires an internal
lock; rely on the submitter’s pre-drain event clearing and existing drain
signaling instead. Apply the same change to the corresponding logic around the
additionally referenced location, while preserving signaling for fail-fast or
otherwise necessary paths.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@cassandra/concurrent.py`:
- Around line 226-233: Update the wait loop in the main result-consumption path
to check _exception and _fail_fast before calling _condition.wait(), while
preserving the existing post-wait check. In _submitter_loop, extend the
pre-dispatch condition to stop scheduling additional statements when _exception
is set under fail-fast mode, so remaining iterable items are not dispatched
after an early failure.

In `@tests/unit/test_concurrent.py`:
- Line 311: Add pytest-timeout to the project’s declared test or development
dependencies so the pytest.mark.timeout decorator in the concurrent test is
provided by an explicitly installed plugin. Preserve the existing timeout marker
usage.

---

Nitpick comments:
In `@cassandra/concurrent.py`:
- Around line 254-258: Update the hot-path callback logic described near the
concurrent submission/draining flow to avoid calling Event.set() on successful
non-fail-fast callbacks, since it acquires an internal lock; rely on the
submitter’s pre-drain event clearing and existing drain signaling instead. Apply
the same change to the corresponding logic around the additionally referenced
location, while preserving signaling for fail-fast or otherwise necessary paths.

In `@tests/unit/test_concurrent.py`:
- Around line 329-334: Update the test double’s add_callbacks method in
tests/unit/test_concurrent.py to match ResponseFuture.add_callbacks positional
parameter order, placing errback immediately after callback and retaining the
remaining argument defaults. Keep the synchronous callback behavior unchanged.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 445f3db9-d3e1-44b6-9d82-1197e5dc9569

📥 Commits

Reviewing files that changed from the base of the PR and between 9b5b037 and 8c88381.

📒 Files selected for processing (2)

• cassandra/concurrent.py
• tests/unit/test_concurrent.py

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (4)

cassandra/concurrent.py:274

• A user iterable can raise an exception whose __bool__ returns False. In that case _submitter_loop records it as _fatal_exception and terminates the work, but this truthiness check skips the raise and returns a truncated result list, contradicting the one-result-per-input contract described above. Check explicitly for None, as the submitter already does when assigning the field.

        if self._fatal_exception:
            raise self._fatal_exception

cassandra/concurrent.py:196

• This implementation only changes ConcurrentExecutorListResults. Calls with the public results_generator=True option still use ConcurrentExecutorGenResults._put_result, which invokes _execute_next() (and therefore execute_async()) directly from the completion callback thread. The PR description and title currently present this as an optimization to execute_concurrent generally, so either the generator path needs equivalent handling or the documented scope should explicitly exclude it.

    def execute(self, concurrency, fail_fast):
        self._exception = None
        self._fatal_exception = None

cassandra/concurrent.py:444

• The submitter is the sole thread that can advance the wait predicate, but this handler excludes BaseException subclasses. If a user-supplied iterator (or execute_async) raises SystemExit, KeyboardInterrupt, or another direct BaseException, the submitter terminates without setting _fatal_exception or notifying the condition, leaving execute_concurrent() blocked forever. Catch BaseException at this thread boundary so it can be relayed to the caller.

        except Exception as exc:

cassandra/concurrent.py:289

• This says the hot path acquires no lock, but _submit_event.set() below acquires the Event's internal condition lock (and lines 299-302 rely on that synchronization). The optimization avoids self._condition; it does not eliminate lock acquisition entirely, so the comment should make that distinction.

        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.