# KLine Thread Pool — Concurrency Model > Part of the **tcpTDX** library (`kline_pool` module). --- ## Overview The `KLineThreadPool` implements a **bounded multi-stage pipeline** concurrency model designed for high-volume market data retrieval. The implementation uses: - Rust native threads - Bounded synchronous channels (`mpsc::sync_channel`) - A dispatcher thread for routing - Worker-side batching Rather than relying on an async runtime (Tokio, async-std), the system uses **deterministic thread scheduling** and **explicit backpressure control**. --- ## Key Characteristics - Thread-based concurrency - Bounded queues - Deterministic memory usage - Load-balanced worker execution - Graceful shutdown and draining --- ## System Architecture ```mermaid flowchart LR P[Producers] SQ[Global Submit Queue\nbounded] D[Dispatcher Thread] WQ1[Worker Queue 1] WQ2[Worker Queue 2] WQN[Worker Queue N] W1[Worker 1 Pipeline] W2[Worker 2 Pipeline] WN[Worker N Pipeline] R[Results Channel] P --> SQ SQ --> D D --> WQ1 D --> WQ2 D --> WQN WQ1 --> W1 WQ2 --> W2 WQN --> WN W1 --> R W2 --> R WN --> R ``` --- ## Pipeline Stages The system operates as a three-stage pipeline. ### Stage 1 — Job Submission Producers submit work into the thread pool: ```rust pool.submit(job)?; ``` Internally, each job is wrapped in an `Arc` and pushed into a bounded synchronous channel: ```rust mpsc::sync_channel::>(submit_queue_cap) ``` This queue acts as the **global intake buffer**. **Important property:** if the queue becomes full, `submit()` blocks — preventing unlimited job accumulation. --- ### Stage 2 — Dispatcher A dedicated dispatcher thread routes jobs from the submit queue to worker inboxes. **Dispatcher algorithm:** 1. Receive job from submit queue 2. Attempt `try_send` to each worker in round-robin order 3. If all workers are full → block on `send()` to the first full worker that is still alive Pseudo-logic: ``` receive job for worker in workers (round-robin): try_send(job) Ok → delivered, advance rotation Full → record as block_on candidate Disconnected → mark worker dead if not delivered and block_on exists: blocking send(job) to that worker ``` The dispatcher acts as a **load balancer**. It exits once the submit queue is closed and fully drained, at which point it drops all worker senders — closing every worker inbox and allowing workers to drain and exit. --- ### Stage 3 — Worker Execution Each worker thread performs the following loop: ```mermaid flowchart TD A[Block on first job\njob_rx.recv] B[Start Batch] C[Collect more jobs\nuntil batch_size\nor batch_wait timeout] D[Build KLine commands] E[pipeline.run] F[Send KLineResult per job\nto results channel] A --> B B --> C C --> D D --> E E --> F F --> A ``` **Batching stops when either:** - `batch_size` is reached, **or** - `batch_wait` timeout elapses Batching improves performance by reducing network round-trips and amortising connection overhead across many symbols. Workers exit when `job_rx.recv()` returns `Err` — meaning the dispatcher has dropped all worker senders and the inbox queue is fully drained. --- ## Data Types ### `KLineJob` One K-line request submitted by a producer: ```rust pub struct KLineJob { pub market: u16, pub code: String, pub category: u16, pub start: u16, pub count: u16, } ``` ### `KLineResult` One K-line result delivered to the results channel: ```rust pub struct KLineResult { pub market: u16, pub code: String, pub category: u16, pub ok: bool, pub data: Vec, pub error: Option, } ``` On pipeline success, `ok` is `true` and `data` contains the retrieved `KLineData` entries. On failure, `ok` is `false`, `data` is a placeholder, and `error` carries the error string. --- ## Backpressure Explained ### Definition Backpressure is a flow-control mechanism that **slows down producers when consumers cannot keep up**. In any pipeline system: ``` Producer speed > Consumer speed ``` If there is no backpressure: - Queues grow indefinitely - Memory usage increases - The system eventually crashes Backpressure ensures that when downstream components become saturated, upstream components are **forced to slow down**. --- ### Backpressure in This Architecture This system implements **two levels of backpressure**. #### Backpressure Layer 1 — Submit Queue The first control point is the global submission queue: ```rust mpsc::sync_channel::>(submit_queue_cap) ``` If producers push jobs faster than the dispatcher can handle them: ``` submit queue fills ↓ send() blocks ↓ producer pauses ``` ```mermaid flowchart LR Producer -->|send — blocks when full| SubmitQueue SubmitQueue --> Dispatcher ``` **Effect:** this prevents unlimited job generation by producers. --- #### Backpressure Layer 2 — Worker Queues Each worker has a bounded inbox queue: ```rust mpsc::sync_channel::>(worker_queue_cap) ``` Dispatcher behaviour: 1. Attempt `try_send` to each worker in round-robin order 2. If a worker queue is full → try the next worker 3. If **all** worker queues are full → block on `send()` to one worker ```mermaid flowchart LR Dispatcher -->|try_send| WQ1[Worker Queue 1] Dispatcher -->|try_send| WQ2[Worker Queue 2] Dispatcher -->|try_send| WQN[Worker Queue N] WQ1 --> Worker1 WQ2 --> Worker2 WQN --> WorkerN ``` --- ### Backpressure Propagation When workers become slow, pressure moves upstream through the entire pipeline: ```mermaid flowchart LR Producers --> SubmitQueue --> Dispatcher --> WorkerQueues --> Workers ``` **Propagation chain:** ``` workers slow down ↓ worker queues fill ↓ dispatcher blocks on send() ↓ submit queue fills ↓ producers block on submit() ``` This forms a **self-regulating feedback loop** — no component can fall indefinitely behind without eventually pausing the producers. --- ### Why Backpressure Is Important Backpressure provides three major guarantees. #### 1. Bounded Memory Usage The maximum number of in-flight jobs at any time is: ``` submit_queue_cap + worker_count × worker_queue_cap + worker_count × batch_size ← actively being processed ``` Memory usage is **fully predictable** and statically bounded by configuration. #### 2. System Stability Without backpressure: - Infinite job growth - Latency explosion - Process crash or OOM kill With backpressure: - The system stabilises automatically under any load #### 3. Natural Rate Limiting Throughput is bounded by worker capacity: ``` system throughput ≈ worker processing rate ``` Producers cannot exceed the processing capability of the pipeline. --- ## Worker Batching Model Workers accumulate jobs before executing the pipeline. **Benefits:** - Reduces network round-trips - Increases throughput under sustained load - Amortises connection overhead across many symbols **Batch formation logic:** ```rust // Block until at least one job arrives let first = job_rx.recv()?; jobs.push(first); // Collect more jobs until batch is full or timeout elapses let deadline = Instant::now() + batch_wait; while jobs.len() < batch_size { match job_rx.recv_timeout(deadline - Instant::now()) { Ok(j) => jobs.push(j), Err(_) => break, } } // Execute all jobs in a single pipeline call pipeline.run(&mut cmds)?; ``` --- ## Public API ```rust /// Construct the pool KLineThreadPool::new( worker_count: usize, submit_queue_cap: usize, worker_queue_cap: usize, batch_size: usize, batch_wait: Duration, ) /// Submit a single job (blocks if submit queue is full) pool.submit(job: KLineJob) -> Result<(), String> /// Submit multiple jobs (blocks if submit queue fills mid-batch) pool.submit_batch(jobs: impl IntoIterator) -> Result<(), String> /// Signal that no more jobs will be submitted pool.close_submit() /// Blocking results iterator — ends when all workers exit pool.iter_results() -> impl Iterator /// Access the raw results receiver directly pool.results() -> &Receiver /// Graceful shutdown — joins dispatcher and all workers /// Calls close_submit() internally if not already called pool.shutdown() ``` --- ## Graceful Shutdown Shutdown proceeds in several stages. 1. Stop accepting new jobs (`close_submit` or `shutdown`) 2. Dispatcher drains the submit queue and forwards remaining jobs 3. Dispatcher exits, dropping all worker senders → closes worker inboxes 4. Each worker drains its inbox and finishes in-progress batches 5. Workers exit, dropping their `res_tx` clones 6. Results channel closes → `iter_results()` terminates ```mermaid sequenceDiagram participant Producer participant Dispatcher participant Worker participant Results Producer->>Dispatcher: close_submit() Dispatcher->>Dispatcher: drain submit queue Dispatcher->>Worker: drop sender — closes inbox Worker->>Worker: drain inbox, finish batches Worker->>Results: send remaining KLineResults Worker-->>Dispatcher: exit Dispatcher-->>Producer: exit Results-->>Producer: channel closed — iter_results ends ``` --- ## Mental Model The system behaves like a **factory conveyor line** with bounded buffer stations at each stage. ``` Producers ↓ Submit Queue (bounded — blocks producers when full) ↓ Dispatcher ↓ Worker Queues (bounded — blocks dispatcher when full) ↓ Workers (batch → pipeline.run → results) ↓ Results Channel ``` Each stage has a bounded buffer, preventing overflow at any point in the pipeline. --- ## Summary The `KLineThreadPool` concurrency model provides: - **Deterministic memory usage** — all queues are capacity-bounded by configuration - **Automatic flow control** — two-level backpressure propagates naturally upstream - **Stable throughput** — worker batching reduces network overhead under sustained load - **Predictable shutdown** — full drain guarantees no in-flight work is silently dropped The two-level backpressure system ensures the pipeline never accumulates unbounded work, making it well-suited for large-scale data ingestion and market data retrieval workloads in the tcpTDX library.

Comments (0)

Leave a Comment