concurrent: move request submission off event loop thread

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 now just 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.

Bug fixes included:
- Start submitter thread after initial batch (avoid race on _exec_count)
- Write _exec_count under _condition lock (avoid race with _current)
- Use exhausted flag to avoid repeated StopIteration on iterator

Author: mykaul

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
 import sys
 
 from cassandra.cluster import ResultSet, EXEC_PROFILE_DEFAULT
@@ -193,28 +193,126 @@ class ConcurrentExecutorListResults(_ConcurrentExecutor):
 
     def execute(self, concurrency, fail_fast):
         self._exception = None
-        return super(ConcurrentExecutorListResults, self).execute(concurrency, fail_fast)
+        self._submit_ready = deque()
+        self._submit_event = Event()
+        self._stop_event = Event()
+        # Submit the initial batch from the calling thread (no contention
+        # yet -- the submitter thread is not started until afterward).
+        result = super(ConcurrentExecutorListResults, self).execute(concurrency, fail_fast)
+        return result
+
+    def _results(self):
+        # Start the submitter thread *after* the initial batch has been
+        # fully dispatched so that _enum_statements and _exec_count are
+        # not accessed concurrently during the seeding phase.
+        self._submitter = Thread(target=self._submitter_loop,
+                                 daemon=True, name="concurrent-submitter")
+        self._submitter.start()
+
+        try:
+            with self._condition:
+                while self._current < self._exec_count:
+                    self._condition.wait()
+                    if self._exception and self._fail_fast:
+                        break
+        finally:
+            self._stop_event.set()
+            self._submit_event.set()  # wake submitter so it sees the stop
+            self._submitter.join()
+        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).  Increments
+        ``_current`` under the condition lock so the waiter in _results()
+        can detect completion.  Appends a signal to ``_submit_ready`` so
+        the submitter thread knows it can submit a replacement request.
+        """
         self._results_queue.append((idx, ExecutionResult(success, result)))
         with self._condition:
             self._current += 1
             if not success and self._fail_fast:
                 if not self._exception:
                     self._exception = result
                 self._condition.notify()
-            elif not self._execute_next() and self._current == self._exec_count:
+            elif 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 to send the next request instead of
+        # calling _execute_next() inline.  This keeps the event-loop thread
+        # (which fires the callback) free to process I/O rather than doing
+        # query-plan lookup, message serialisation, and connection borrowing.
+        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).
+
+        Completion accounting (_current) is handled entirely by
+        ``_put_result`` -- this thread only needs to know *how many*
+        new requests to submit (one per completion signal drained).
+        """
+        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
+        exhausted = False
+        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
+            if exhausted:
+                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).
+            for _ in range(count):
+                try:
+                    idx, (statement, params) = next(enum_stmts)
+                except StopIteration:
+                    exhausted = True
+                    break
+                # Bump _exec_count *before* dispatching so that _results()
+                # never sees _current >= _exec_count while requests are
+                # still in flight.
+                with condition:
+                    self._exec_count += 1
+                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)