---
type: reference
title: "GPU Async Pipelines — run_in_executor, Prefetch Queues, and TDR Recovery"
description: "Async patterns for GPU compute: thread pool offloading, concurrent prefetch workers, starvation prevention, TDR recovery, and auto-tuning."
tags: [async, asyncio, gpu, prefetch, tdr, pipelines, compute-optimization]
timestamp: 2026-07-25
---

# GPU Async Pipelines

Async architecture patterns for mixing synchronous GPU kernel dispatch with Python's asyncio event loop. Both [[hashfita]] and [[crustybike]] use `aioquic` for distributed networking, requiring the GPU compute to not block the event loop.

---

## Core Pattern: run_in_executor

Both projects use the same fundamental pattern to offload synchronous GPU calls:

```python
# hashfita: network/worker.py
# crustybike: distributed/worker_opencl.py, line 334
batch_res = await loop.run_in_executor(None, self._run_opencl_chunk, ...)
```

`run_in_executor(None, fn)` dispatches `fn` to the default `ThreadPoolExecutor`, preventing the synchronous OpenCL kernel dispatch from blocking the asyncio event loop. The `None` argument uses the default executor — no custom thread pool configuration.

### Why Not asyncio.to_thread?

Both projects predate or bypass `asyncio.to_thread()` (Python 3.9+) in favor of the lower-level `run_in_executor`. The pattern is functionally identical but `run_in_executor` allows explicit executor selection.

---

## Prefetch Architecture

### hashfita — Single Worker

hashfita uses a single prefetch worker with an `asyncio.Queue`. The training loop pulls precomputed examples from the queue while the GPU processes the current batch.

### crustybike — "Tropa de Choque" (10 Concurrent Workers)

```python
# task_provider.py, lines 74-84
workers = [asyncio.create_task(self._prefetch_worker()) for _ in range(10)]
await asyncio.gather(*workers)
```

Ten concurrent prefetch workers populate an `asyncio.Queue(maxsize=10)` with precomputed blueprints. Each prefetch worker:
1. Calls `evaluate_topology()` — MLX tensor computation on Metal GPU
2. Calls `mx.eval()` — JIT trigger
3. Calls `.item()` — GPU→CPU scalar extraction
4. Serializes to JSON blueprint
5. Puts into queue

The maxsize=10 acts as backpressure — if the mining workers are slow, prefetch workers block on `queue.put()`.

### O(1) Task Pull

```python
# crustybike/task_provider.py, line 123
async def pull_task(self):
    return await self._queue.get()  # O(1) deque operation
```

Designed for zero-starvation task distribution.

---

## Starvation Prevention

```python
# crustybike/task_provider.py, line 424
await asyncio.sleep(0)
```

Explicit yield to the event loop after MLX tensor computation. Without this, the Metal GPU's synchronous `mx.eval()` would block the QUIC server from processing incoming messages.

**Rule:** Always `await asyncio.sleep(0)` after any synchronous GPU operation in an async context.

---

## Dashboard Update Loop

```python
# crustybike/worker_opencl.py, lines 269-272
async def _update_loop(self):
    while True:
        self._update_dashboard()
        await asyncio.sleep(0.25)  # 4Hz refresh rate
```

The TUI dashboard runs as a separate asyncio task at 4Hz, independent of GPU compute timing.

---

## TDR Recovery (crustybike)

GPU Timeout Detection and Recovery — handles driver-level GPU resets on consumer hardware:

```python
# opencl_context.py, lines 651-659
def rebuild_opencl_context(self):
    """Tear down and rebuild entire OpenCL context after GPU timeout."""
    # 1. Release all buffers
    # 2. Destroy command queue
    # 3. Release context
    # 4. Re-create everything from scratch
```

When the OS detects the GPU has been unresponsive (typically 2 seconds on macOS, configurable on Linux), it resets the GPU. The OpenCL context becomes poisoned — all subsequent operations fail. The only recovery is to destroy and rebuild the entire context.

**hashfita does not have this protection.** For 24/7 mining operation, this is a critical gap.

---

## Auto-Tuning Batch Size

### Hardware-Adaptive (crustybike)

```python
max_wg = device.max_work_group_size           # Query hardware limit
batch_step = max(1_000_000, max_wg * 4096)   # Scale from hardware
```

### Empirical Sweep (hashfita)

```python
# benchmark_grokking.py
for batch in [64, 128, 256, ..., 32768]:
    # Test forward + backward pass, detect OOM boundary
```

hashfita benchmarks across batch sizes to find the OOM boundary on Apple Silicon. crustybike queries the device directly and computes the optimal batch.

---

## Distributed Protocol

Both projects use QUIC (aioquic) for worker communication:

| Aspect | hashfita | crustybike |
|--------|----------|------------|
| Serialization | msgpack | JSON blueprints |
| Auth | mTLS (auto-generated certs) | Certificate-based |
| Streams | Per-task QUIC streams | Per-task QUIC streams |
| Worker type | .NET/C# + Python IPC | Pure Python |

---

## Cross-References

- [[compute-optimization]] — Hub
- [[mlx-opencl-bridge]] — The synchronous bottleneck hidden by async
- [[opencl-kernel-design]] — What the async pipeline dispatches
- [[ring-transformer]] — The training loop that drives the pipeline