updating streaming and architecture docs

Author: millerjp

docs/architecture.md

@@ -35,29 +35,134 @@ sequenceDiagram
 
 ## Solution Architecture
 
-async-cassandra wraps the driver's async operations to provide true async/await support:
+async-cassandra bridges the gap between the DataStax driver's thread-based callbacks and Python's asyncio event loop. Here's how it actually works:
+
+### The Bridge Pattern
+
+The DataStax driver uses a ThreadPoolExecutor for I/O operations and callbacks. We don't replace this - instead, we bridge between the driver's threading model and asyncio:
 
 ```mermaid
 sequenceDiagram
     participant App as Async Application
     participant AsyncWrapper as async-cassandra
     participant Driver as Cassandra Driver
-    participant EventLoop as Event Loop
+    participant ThreadPool as Driver Thread Pool
+    participant EventLoop as Asyncio Event Loop
     participant Cassandra as Cassandra DB
 
     App->>AsyncWrapper: await execute(query)
     AsyncWrapper->>Driver: execute_async(query)
-    Note over AsyncWrapper: Create Future
-    Driver->>Cassandra: Send query (non-blocking)
-    AsyncWrapper-->>EventLoop: Register callback
-    EventLoop-->>App: Control returned
-    Note over App: Can handle other requests
-    Cassandra-->>Driver: Response
-    Driver-->>AsyncWrapper: Callback triggered
-    AsyncWrapper-->>EventLoop: Set Future result
-    EventLoop-->>App: Resume coroutine
+    Driver->>ThreadPool: Submit I/O task
+    ThreadPool->>Cassandra: Send query (in thread)
+    AsyncWrapper->>EventLoop: Create asyncio.Future
+    AsyncWrapper-->>App: Return control (await)
+
+    Note over App: Free to handle other requests
+
+    Cassandra-->>ThreadPool: Response data
+    ThreadPool->>AsyncWrapper: Callback fired (in driver thread)
+    AsyncWrapper->>EventLoop: call_soon_threadsafe(set_result)
+    EventLoop-->>App: Resume coroutine with result
+```
+
+### Key Implementation Details
+
+1. **We DON'T create our own thread pool** - we use the driver's existing ThreadPoolExecutor
+2. **Thread-safe communication** - Callbacks from driver threads use `call_soon_threadsafe()` to safely communicate with the asyncio event loop
+3. **Futures bridging** - We create asyncio Futures that are resolved by callbacks from driver threads
+
+## How the Bridge Works: Code Deep Dive
+
+### 1. Query Execution Flow
+
+When you call `await session.execute(query)`, here's what happens:
+
+```python
+# In AsyncCassandraSession.execute() - src/async_cassandra/session.py:156-168
+response_future = self._session.execute_async(
+    query, parameters, trace, custom_payload, timeout,
+    execution_profile, paging_state, host, execute_as
+)
+
+handler = AsyncResultHandler(response_future)
+result = await handler.get_result(timeout=query_timeout)
+```
+
+The DataStax driver's `execute_async()` returns a `ResponseFuture` that will be completed by a driver thread.
+
+### 2. The Bridge: AsyncResultHandler
+
+The magic happens in `AsyncResultHandler` ([src/async_cassandra/result.py](../src/async_cassandra/result.py)):
+
+```python
+class AsyncResultHandler:
+    def __init__(self, response_future: ResponseFuture):
+        self.response_future = response_future
+        self.rows: List[Any] = []
+        self._future: Optional[asyncio.Future[AsyncResultSet]] = None
+        self._lock = threading.Lock()  # Thread safety!
+
+        # Register callbacks with the driver
+        self.response_future.add_callbacks(
+            callback=self._handle_page,
+            errback=self._handle_error
+        )
+```
+
+### 3. Thread-Safe Callback Handling
+
+When the driver completes the query (in a driver thread), our callback is invoked:
+
+```python
+def _handle_page(self, rows: List[Any]) -> None:
+    """Called from driver thread - must be thread-safe!"""
+    with self._lock:
+        if rows is not None:
+            self.rows.extend(list(rows))  # Defensive copy
+
+        if self.response_future.has_more_pages:
+            self.response_future.start_fetching_next_page()
+        else:
+            # All done - notify the asyncio Future
+            final_result = AsyncResultSet(list(self.rows), self.response_future)
+
+            if self._future and not self._future.done():
+                loop = getattr(self, "_loop", None)
+                if loop:
+                    # CRITICAL: Use call_soon_threadsafe to bridge threads!
+                    loop.call_soon_threadsafe(self._future.set_result, final_result)
 ```
 
+### 4. The Asyncio Side
+
+Meanwhile, the asyncio coroutine is waiting:
+
+```python
+async def get_result(self, timeout: Optional[float] = None) -> "AsyncResultSet":
+    # Create asyncio Future in the current event loop
+    loop = asyncio.get_running_loop()
+    self._future = loop.create_future()
+    self._loop = loop  # Store for callbacks to use
+
+    # Wait for the driver thread to complete
+    if timeout is not None:
+        return await asyncio.wait_for(self._future, timeout=timeout)
+    else:
+        return await self._future
+```
+
+### 5. Driver Thread Pool Configuration
+
+The driver's thread pool size is configurable ([src/async_cassandra/cluster.py](../src/async_cassandra/cluster.py)):
+
+```python
+def __init__(self, ..., executor_threads: int = 2, ...):
+    # This is passed to the DataStax Cluster constructor
+    # The driver creates: ThreadPoolExecutor(max_workers=executor_threads)
+```
+
+**Important**: This thread pool is shared for ALL I/O operations, so under high concurrency, you may need to increase `executor_threads`.
+
 ## Key Components
 
 ### AsyncCluster
@@ -94,22 +199,28 @@ sequenceDiagram
     participant User as User Code
     participant Session as AsyncCassandraSession
     participant Handler as AsyncResultHandler
+    participant EventLoop as Event Loop
     participant Driver as Cassandra Driver
+    participant ThreadPool as Driver Thread Pool
     participant DB as Cassandra
 
     User->>Session: await execute(query)
     Session->>Driver: execute_async(query)
+    Driver->>ThreadPool: Submit I/O task
+    ThreadPool->>DB: Send CQL query
     Driver-->>Session: ResponseFuture
     Session->>Handler: new AsyncResultHandler(ResponseFuture)
-    Handler->>Handler: Register callbacks
-    Session-->>User: Return Future
+    Handler->>Driver: add_callbacks(callback, errback)
+    Handler->>EventLoop: create_future()
+    Session-->>User: await handler.get_result()
 
-    Note over User,DB: Async execution in progress
+    Note over User: Coroutine suspended, event loop free
 
-    DB-->>Driver: Query result
-    Driver-->>Handler: Trigger callback
-    Handler->>Handler: Process result/pages
-    Handler-->>User: Resolve Future with AsyncResultSet
+    DB-->>ThreadPool: Query result
+    ThreadPool->>Handler: _handle_page(rows) [in driver thread]
+    Handler->>Handler: Lock, process rows
+    Handler->>EventLoop: call_soon_threadsafe(future.set_result)
+    EventLoop-->>User: Resume coroutine with AsyncResultSet
 ```
 
 ### Streaming Execution
@@ -121,26 +232,34 @@ sequenceDiagram
     participant App as Application
     participant Session as AsyncCassandraSession
     participant Stream as AsyncStreamingResultSet
-    participant Handler as StreamingResultHandler
+    participant EventLoop as Event Loop
     participant Driver as Cassandra Driver
+    participant ThreadPool as Driver Thread Pool
     participant DB as Cassandra
 
     App->>Session: await execute_stream(query, config)
     Session->>Driver: execute_async(query)
+    Driver->>ThreadPool: Submit I/O task
+    ThreadPool->>DB: Send CQL query
     Driver-->>Session: ResponseFuture
-    Session->>Handler: StreamingResultHandler(future, config)
-    Handler->>Stream: Create AsyncStreamingResultSet
+    Session->>Stream: Create AsyncStreamingResultSet(future, config)
+    Stream->>Driver: add_callbacks(callback, errback)
     Session-->>App: Return Stream
 
-    loop For each page request
+    loop For each row iteration
         App->>Stream: async for row in stream
-        Stream->>Handler: Request next page
-        Handler->>Driver: Fetch page asynchronously
-        Driver->>DB: Get page (fetch_size rows)
-        DB-->>Driver: Page data
-        Driver-->>Handler: Page received
-        Handler-->>Stream: Yield rows
-        Stream-->>App: Return row
+        alt Current page has rows
+            Stream-->>App: Return next row
+        else Page exhausted, need next page
+            Stream->>EventLoop: Create page_ready Event
+            Note over App: Coroutine suspended
+            DB-->>ThreadPool: Next page data
+            ThreadPool->>Stream: _handle_page(rows) [in driver thread]
+            Stream->>Stream: Lock, replace current page
+            Stream->>EventLoop: call_soon_threadsafe(page_ready.set)
+            EventLoop->>Stream: Resume iteration
+            Stream-->>App: Return first row of new page
+        end
     end
 ```
 
@@ -171,12 +290,55 @@ sequenceDiagram
 - Metrics and monitoring built-in
 - Battle-tested retry policies
 
+## Thread Pool Architecture Summary
+
+### The Two Worlds
+
+async-cassandra bridges two distinct execution environments:
+
+1. **Asyncio World** (Your Application)
+   - Single-threaded event loop
+   - Coroutines and async/await
+   - Non-blocking I/O via event loop
+   - Can handle thousands of concurrent operations
+
+2. **Driver Thread Pool World** (DataStax Driver)
+   - ThreadPoolExecutor with `executor_threads` workers (default: 2)
+   - Blocking I/O operations
+   - Callback-based completion
+   - Limited by thread count
+
+### The Bridge Mechanism
+
+We don't "map" thread pools - we **bridge** between them:
+
+```
+Your Async Code → asyncio.Future → Callback Bridge → Driver ResponseFuture → Thread Pool
+                       ↑                                                          ↓
+                       └──────── call_soon_threadsafe() ←──────────────────────┘
+```
+
+Key bridging components:
+- **asyncio.Future**: Created in the event loop, awaited by your code
+- **ResponseFuture**: Created by the driver, completed by driver threads
+- **Callbacks**: Registered with ResponseFuture, called from driver threads
+- **call_soon_threadsafe()**: The critical method that safely notifies the event loop from driver threads
+
+### Performance Implications
+
+Since we use the driver's thread pool:
+- Maximum concurrent I/O operations = `executor_threads` (default: 2)
+- Under high load, increase `executor_threads` when creating the cluster
+- Each blocking operation ties up a thread until completion
+- The event loop stays free, but total throughput is limited by thread count
+
 ## Important Limitations
 
 While async-cassandra provides async/await syntax, it's important to understand:
 
 1. **The underlying I/O is still synchronous** - The DataStax driver uses blocking sockets in threads
-2. **Thread pool constraints apply** - Concurrency is limited by the driver's thread pool size
+2. **Thread pool constraints apply** - Concurrency is limited by the driver's thread pool size (default: 2 threads)
 3. **Not a true async driver** - This is a compatibility layer, not a ground-up async implementation
+4. **No thread pool multiplication** - We use the driver's thread pool as-is, we don't add additional threads
 
 For more details on these limitations and when to use this wrapper, see [Why an Async Wrapper is Necessary](why-async-wrapper.md).