CGraph Thread Pool Architecture and Scheduling Strategy: A Deep Dive into the Hybrid Work-Stealing Implementation

CGraph implements a hybrid thread pool that combines fixed primary threads with lock-free work-stealing queues and dynamically-scaling secondary threads to handle priority and long-running tasks, enabling high-throughput task scheduling with configurable CPU affinity and batch processing.

The chunelfeng/cgraph repository provides a high-performance C++ task scheduling framework built around a sophisticated thread pool architecture. Understanding the CGraph thread pool architecture and scheduling strategy is essential for optimizing pipeline performance, configuring resource limits, and handling heterogeneous workloads ranging from micro-tasks to long-running computations.

Core Architecture Components

The thread pool is orchestrated by UThreadPool and composed of three distinct thread types, each optimized for specific workload characteristics.

UThreadPool — The Central Manager

Located in src/UtilsCtrl/ThreadPool/UThreadPool.h, the UThreadPool class serves as the entry point for task submission and lifecycle management. It maintains two lock-free queues:

• UAtomicQueue<UTask> task_queue_ — Global queue for normal tasks
• UAtomicPriorityQueue<UTask> priority_task_queue_ — Dedicated queue for long-running or priority tasks

The manager initializes primary threads via init() (lines 60-66 of UThreadPool.cpp), monitors workload through a background monitor thread, and lazily creates secondary threads via createSecondaryThread() (lines 24-38).

UThreadPrimary — Work-Stealing Workers

Primary threads, defined in src/UtilsCtrl/ThreadPool/Thread/UThreadPrimary.h, constitute the fixed-size core worker pool. Each primary maintains a private work-stealing queue (UWorkStealingQueue) to minimize contention.

When a primary thread exhausts its local queue, it attempts to steal tasks from neighboring primaries through stealTask() (lines 82-100). The steal range is determined by config_->calcStealRange(), derived from max_steal_batch_size_ and max_task_steal_range_. Primary threads enter sleep state after primary_thread_busy_epoch_ empty cycles, waking on primary_thread_empty_interval_ intervals.

UThreadSecondary — Dynamic Auxiliary Threads

Secondary threads (src/UtilsCtrl/ThreadPool/Thread/UThreadSecondary.h) provide elastic capacity for priority and long-running tasks. Unlike primaries, secondaries consume directly from the global task_queue_ or priority_task_queue_.

Secondary threads feature automatic lifecycle management through freeze() (lines 8-15), entering a TTL (time-to-live) countdown when idle. The monitor thread removes expired secondaries (lines 65-67 of UThreadPool.cpp) after secondary_thread_ttl_ seconds, ensuring resources are released during low-load periods.

UThreadBase — Shared Foundation

Both primary and secondary threads inherit from UThreadBase (src/UtilsCtrl/ThreadPool/Thread/UThreadBase.h), which provides common functionality including queue access abstractions, task execution wrappers, sleep/wakeup mechanisms, and CPU affinity handling via bind_cpu_enable_.

UThreadPoolConfig — Runtime Tuning

Configuration parameters are centralized in src/UtilsCtrl/ThreadPool/UThreadPoolConfig.h, allowing fine-grained control over thread counts, stealing behavior, scheduling policies, and batch processing.

Scheduling Strategy and Task Dispatch

CGraph employs a multi-tiered dispatch system that routes tasks based on execution characteristics and current system load.

Task Routing Logic

When submitting tasks via UThreadPool::commit(), the pool determines routing through dispatch() (lines 209-221 of UThreadPool.cpp):

CIndex UThreadPool::dispatch(const CIndex origIndex) {
    if (CGRAPH_DEFAULT_TASK_STRATEGY == origIndex) {
        realIndex = cur_index_++;               // round-robin across primaries
        if (cur_index_ >= config_.max_thread_size_ || cur_index_ < 0) {
            cur_index_ = 0;
        }
    } else {
        realIndex = origIndex;                 // fixed-pool, long-time, etc.
    }
    return realIndex;
}

The system defines three primary strategies:

• CGRAPH_DEFAULT_TASK_STRATEGY (-1) — Round-robin distribution across primary threads, maximizing cache locality and work-stealing opportunities.
• CGRAPH_POOL_TASK_STRATEGY (-2) — Forces tasks into the global pool queue, making them available to secondary threads.
• CGRAPH_LONG_TIME_TASK_STRATEGY (-101) — Routes tasks to the priority queue, ensuring only secondary threads handle long-running computations to prevent primary thread starvation.

Work-Stealing Mechanism

Primary threads implement work-stealing to balance load without central coordination. During initialization, each primary builds a neighbor list via buildStealTargets() (lines 42-48 of UThreadPrimary.h):

void buildStealTargets() {
    for (int i = 0; i < config_->max_task_steal_range_; i++) {
        int target = (index_ + i + 1) % config_->default_thread_size_;
        steal_targets_.push_back(target);
    }
}

When a primary exhausts its local queue, it iterates through steal_targets_ calling stealTask() until successful or all neighbors are checked. This decentralized approach minimizes contention on the global queue while ensuring high CPU utilization.

Batch Processing vs Single Execution

The pool supports both single-task and batch-task execution modes controlled by config_->batch_task_enable_. In UThreadBase::loopProcess, the system selects between:

• processTask() — Handles one task per iteration, suitable for low-latency requirements.
• processTasks() — Pulls multiple tasks (up to max_local_batch_size_ for local queues or max_pool_batch_size_ for global queues) and executes them in a tight loop, reducing synchronization overhead for high-throughput scenarios.

Thread Lifecycle and Auto-Scaling

The hybrid architecture separates fixed-capacity primary threads from elastic secondary threads, each with distinct lifecycle management.

Primary Thread Initialization

Primary threads are created during UThreadPool::init():

CStatus UThreadPool::init() {
    // Lines 60-66: Create primary threads
    for (int i = 0; i < config_.default_thread_size_; i++) {
        auto *pt = new UThreadPrimary(i, &config_);
        pt->setThreadPoolInfo(this, i);
        primary_threads_.emplace_back(pt);
    }
    // ... monitor initialization
}

Each primary registers with the pool via setThreadPoolInfo, establishing the back-reference required for work-stealing and status monitoring.

Secondary Thread Creation and TTL

Secondary threads provide elastic capacity without pre-allocating resources. The monitor thread (lines 42-68 of UThreadPool.cpp) evaluates workload every monitor_span_ milliseconds:

void UThreadPool::monitor() {
    while (monitor_enable_) {
        // Check if we need more threads
        if (priority_task_queue_.size() > 0 || allPrimariesBusy()) {
            createSecondaryThread();
        }
        
        // Cleanup expired secondaries
        for (auto iter = secondary_threads_.begin(); iter != secondary_threads_.end(); ) {
            if ((*iter)->isFrozen() && (*iter)->getTTL() <= 0) {
                iter = secondary_threads_.erase(iter);
            } else {
                ++iter;
            }
        }
    }
}

Secondary threads enter a frozen state via freeze() when idle, decrementing their TTL each monitor cycle. Once TTL reaches zero, the monitor removes them, releasing system resources.

Configuration and Customization

Fine-tuning the CGraph thread pool requires understanding the configuration parameters defined in src/UtilsCtrl/ThreadPool/UThreadPoolConfig.h:

Parameter	Default	Purpose
default_thread_size_	8	Number of primary (core) threads
secondary_thread_size_	0	Initial secondary thread count
max_thread_size_	16	Upper bound for total threads
max_task_steal_range_	2	Maximum neighbors a primary can steal from
max_steal_batch_size_	64	Tasks stolen per steal attempt
max_local_batch_size_	128	Tasks processed from local queue per batch
max_pool_batch_size_	128	Tasks processed from global queue per batch
primary_thread_busy_epoch_	1000	Empty cycles before primary sleeps
primary_thread_empty_interval_	100	Sleep duration (ms) for idle primary
secondary_thread_ttl_	60	Secondary thread lifetime (seconds)
monitor_enable_	true	Enable workload monitoring
monitor_span_	1000	Monitor check interval (ms)
batch_task_enable_	false	Enable batch task processing
bind_cpu_enable_	false	Pin threads to CPU cores (Linux)
primary_thread_policy_	SCHED_OTHER	POSIX scheduling policy for primaries
primary_thread_priority_	0	POSIX priority for primaries
secondary_thread_policy_	SCHED_OTHER	POSIX scheduling policy for secondaries
secondary_thread_priority_	0	POSIX priority for secondaries

These settings allow precise control over throughput, latency, and resource isolation. For CPU-bound workloads, enabling bind_cpu_enable_ and adjusting max_task_steal_range_ can significantly reduce cache misses.

Practical Implementation Examples

Creating a Pipeline with Custom Thread Pool

Configure a dedicated thread pool for a specific pipeline to isolate resources:

// Create a pipeline
auto *pipeline = graph->createPipeline("myPipe");

// Configure thread pool parameters
UThreadPoolConfig cfg;
cfg.default_thread_size_ = 4;          // Four primary workers
cfg.secondary_thread_size_ = 0;        // No secondaries initially
cfg.max_task_steal_range_ = 2;         // Steal from 2 neighbors
pipeline->setUniqueThreadPoolConfig(cfg);

// Add nodes and edges
pipeline->addNode(...);
pipeline->addEdge(...);

// Execute (tasks scheduled by the custom pool)
pipeline->run();

Source: UThreadPoolConfig.h (lines 15-20) and UThreadPool.h (setter at line 49).

Direct Task Submission to the Pool

Submit tasks directly using different scheduling strategies:

UThreadPool pool;                     // Auto-initialize with defaults
pool.init();                          // Explicit initialization

// Normal task: round-robin across primaries (default strategy)
pool.commit([](){ std::cout << "Hello from primary\n"; });

// Long-running task: routed to priority queue for secondary processing
pool.commit([](){ heavyComputation(); }, CGRAPH_LONG_TIME_TASK_STRATEGY);

Source: UThreadPool::commit template (lines 70-78) and dispatch logic (lines 209-221).

Sharing a Thread Pool Across Multiple Pipelines

Optimize resource usage by sharing one pool among multiple pipelines:

UThreadPool sharedPool(true, UThreadPoolConfig());   // Auto-init enabled
auto *pipe1 = graph->createPipeline("p1");
auto *pipe2 = graph->createPipeline("p2");

// Both pipelines share the same underlying thread pool
pipe1->setSharedThreadPool(&sharedPool);
pipe2->setSharedThreadPool(&sharedPool);

Source: GraphPipeline API (see tutorial T07-MultiPipeline.cpp).

Tuning Work-Stealing and Batch Processing

Optimize for high-throughput scenarios with batch processing:

UThreadPoolConfig cfg;
cfg.default_thread_size_ = 8;
cfg.max_task_steal_range_ = 3;       // Wider steal range for uneven loads
cfg.max_local_batch_size_ = 4;       // Process 4 local tasks per iteration
cfg.batch_task_enable_ = true;       // Enable batch mode for throughput
pipeline->setUniqueThreadPoolConfig(cfg);

Source: UThreadPoolDefine.h (constants for defaults) and UThreadPrimary::buildStealTargets (lines 42-48).

Key Source Files

File	Contribution
src/UtilsCtrl/ThreadPool/UThreadPool.h	Public API, pool constructor, configuration setters, commit/submit entry points.
src/UtilsCtrl/ThreadPool/UThreadPool.cpp	Core lifecycle (init, destroy, monitor), task dispatch, secondary-thread management.
src/UtilsCtrl/ThreadPool/UThreadPoolConfig.h	All tunable parameters, special strategy constants.
src/UtilsCtrl/ThreadPool/Thread/UThreadPrimary.h	Primary-worker implementation, work-stealing logic, idle-sleep handling.
src/UtilsCtrl/ThreadPool/Thread/UThreadSecondary.h	Secondary-worker for priority/long-running tasks, auto-TTL release.
src/UtilsCtrl/ThreadPool/Thread/UThreadBase.h	Shared utilities (queue access, task execution, scheduler & affinity helpers).
tutorial/T07-MultiPipeline.cpp	Example showing shared vs. unique pools in practice.
example/E05-HttpServer.cpp	Real-world usage where the pool drives concurrent request handling.

Summary

• Hybrid Architecture: CGraph combines fixed primary threads with elastic secondary threads, separating fast micro-tasks from long-running computations.
• Work-Stealing: Primary threads use private lock-free queues and steal from neighbors when idle, minimizing global contention.
• Adaptive Scaling: A monitor thread automatically creates secondary threads when all primaries are busy or priority tasks arrive, releasing them after a configurable TTL.
• Flexible Dispatch: Three scheduling strategies (round-robin, global pool, priority queue) allow fine-grained control over task placement.
• Batch Optimization: Optional batch processing reduces synchronization overhead for high-throughput scenarios.

Frequently Asked Questions

How does CGraph decide whether to create a secondary thread?

The monitor thread evaluates workload every monitor_span_ milliseconds. It creates a secondary thread when either the priority task queue contains tasks or all primary threads report busy status. Secondaries self-terminate after secondary_thread_ttl_ seconds of inactivity.

What is the difference between primary and secondary threads in CGraph?

Primary threads are fixed-count workers (default 8) that maintain private work-stealing queues and execute short, latency-sensitive tasks. Secondary threads are dynamically created to handle priority and long-running tasks from global queues, preventing primary thread starvation and ensuring responsive scheduling for mixed workloads.

How does work-stealing improve performance in the CGraph thread pool?

Work-stealing allows idle primary threads to pull tasks from busy neighbors' private queues rather than waiting on the global queue. This decentralized approach reduces lock contention on task_queue_, improves cache locality by keeping tasks on the same CPU core, and automatically balances load across the primary thread pool without central coordination.

Can I pin CGraph threads to specific CPU cores?

Yes, setting bind_cpu_enable_ to true in UThreadPoolConfig enables CPU affinity binding for primary threads on Linux systems. Additionally, you can configure POSIX scheduling policies and priorities via primary_thread_policy_, primary_thread_priority_, secondary_thread_policy_, and secondary_thread_priority_ to optimize for real-time or throughput requirements.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}