1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

/* Work stealing deque structure and operations */
 
#define DEQUE_SIZE 4096  /* Power of 2 for efficient modulo */
#define DEQUE_MASK (DEQUE_SIZE - 1)
 
struct work_item {
    void (*function)(void *arg);
    void *argument;
    struct work_item *parent;  /* For join synchronization */
    atomic_int pending_children;
};
 
struct work_stealing_deque {
    /* Bottom: owned exclusively by local processor */
    /* Updated by push/pop operations */
    volatile size_t bottom;
    
    /* Top: contended between owner and thieves */
    /* Updated by steal operations, may be read by pop */
    volatile size_t top;
    
    /* Circular buffer of work items */
    struct work_item *buffer[DEQUE_SIZE];
    
    /* Owner processor ID */
    int owner_cpu;
    
    /* Statistics */
    unsigned long push_count;
    unsigned long pop_count;
    unsigned long steal_count;
    unsigned long steal_fail_count;
};
 
/* Deque operations - three core functions */
void deque_push(struct work_stealing_deque *deque, struct work_item *item);
struct work_item *deque_pop(struct work_stealing_deque *deque);
struct work_item *deque_steal(struct work_stealing_deque *deque);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

/* Chase-Lev Work Stealing Deque Implementation */
/* Based on "Dynamic Circular Work-Stealing Deque" by Chase & Lev */
 
#include <stdatomic.h>
 
/* Memory barriers for ordering */
#define LOAD_ACQUIRE(x) atomic_load_explicit(&(x), memory_order_acquire)
#define STORE_RELEASE(x, v) atomic_store_explicit(&(x), (v), memory_order_release)
 
/* PUSH: Owner pushes work onto bottom of deque */
/* Lock-free: no synchronization with other pushes/pops */
void deque_push(struct work_stealing_deque *deque, struct work_item *item) {
    size_t b = deque->bottom;
    size_t t = LOAD_ACQUIRE(deque->top);
    
    /* Check if deque is full */
    if (b - t >= DEQUE_SIZE - 1) {
        /* Deque overflow - could resize or block */
        handle_deque_overflow(deque);
        return;
    }
    
    /* Store item at bottom */
    deque->buffer[b & DEQUE_MASK] = item;
    
    /* Memory barrier: ensure item is visible before bottom update */
    atomic_thread_fence(memory_order_release);
    
    /* Increment bottom - no CAS needed, only owner writes */
    deque->bottom = b + 1;
    
    deque->push_count++;
}
 
/* POP: Owner pops work from bottom of deque */
/* May contend with steal at the boundary */
struct work_item *deque_pop(struct work_stealing_deque *deque) {
    size_t b = deque->bottom - 1;
    deque->bottom = b;  /* Tentatively decrement bottom */
    
    /* Memory barrier: ensure decrement visible before reading top */
    atomic_thread_fence(memory_order_seq_cst);
    
    size_t t = LOAD_ACQUIRE(deque->top);
    
    struct work_item *item = NULL;
    
    if (t <= b) {
        /* Deque non-empty (at least one item) */
        item = deque->buffer[b & DEQUE_MASK];
        
        if (t == b) {
            /* This is the LAST item - may race with steal */
            /* Use CAS to resolve contention */
            if (!atomic_compare_exchange_strong(
                    (atomic_size_t *)&deque->top, &t, t + 1)) {
                /* Lost race to thief - they got it */
                item = NULL;
            }
            /* Reset deque to empty state */
            deque->bottom = t + 1;
        }
        /* else: not last item, safe to take without CAS */
    } else {
        /* Deque was empty - restore bottom */
        deque->bottom = t;
    }
    
    if (item) deque->pop_count++;
    return item;
}
 
/* STEAL: Thief steals from top of deque */
/* May contend with owner's pop or other thieves */
struct work_item *deque_steal(struct work_stealing_deque *deque) {
    size_t t = LOAD_ACQUIRE(deque->top);
    
    /* Memory barrier: ensure top read before bottom */
    atomic_thread_fence(memory_order_seq_cst);
    
    size_t b = deque->bottom;
    
    struct work_item *item = NULL;
    
    if (t < b) {
        /* Deque non-empty - attempt to steal */
        item = deque->buffer[t & DEQUE_MASK];
        
        /* CAS to claim this item */
        if (!atomic_compare_exchange_strong(
                (atomic_size_t *)&deque->top, &t, t + 1)) {
            /* Lost race to another thief or to owner's pop */
            deque->steal_fail_count++;
            return NULL;
        }
        
        deque->steal_count++;
    }
    
    return item;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

/* Randomized victim selection with work stealing */
 
#include <stdlib.h>
 
/* Thread-local random state for fast, non-contended random numbers */
static __thread unsigned long random_state;
 
/* Fast random number generator (xorshift) */
static inline unsigned long quick_random(void) {
    unsigned long x = random_state;
    x ^= x << 13;
    x ^= x >> 17;
    x ^= x << 5;
    random_state = x;
    return x;
}
 
/* Initialize random state from unique seed */
void init_random_state(int cpu_id) {
    random_state = (unsigned long)cpu_id * 2654435761UL + 0xDEADBEEF;
}
 
/* Random victim selection - uniform distribution */
static int select_random_victim(int my_cpu, int num_cpus) {
    int victim;
    
    do {
        victim = quick_random() % num_cpus;
    } while (victim == my_cpu);  /* Don't steal from ourselves */
    
    return victim;
}
 
/* Main work stealing loop */
struct work_item *work_steal_loop(int my_cpu, 
                                   struct work_stealing_deque *deques,
                                   int num_cpus) {
    struct work_item *item = NULL;
    int attempts = 0;
    int max_attempts = num_cpus * 2;  /* Don't spin forever */
    
    while (!item && attempts < max_attempts) {
        /* Pick a random victim */
        int victim = select_random_victim(my_cpu, num_cpus);
        
        /* Attempt to steal */
        item = deque_steal(&deques[victim]);
        
        if (!item) {
            attempts++;
            
            /* Exponential backoff to reduce contention */
            if (attempts > num_cpus) {
                cpu_relax_n(attempts - num_cpus);
            }
        }
    }
    
    return item;
}
 
/* CPU relaxation with variable delay */
static inline void cpu_relax_n(int n) {
    for (int i = 0; i < n; i++) {
        __builtin_ia32_pause();  /* x86 PAUSE instruction */
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

/* Work stealing with adaptive backoff strategy */
struct work_item *adaptive_steal(int my_cpu,
                                  struct work_stealing_deque *deques,
                                  int num_cpus) {
    int failures = 0;
    int consecutive_empty = 0;
    
    while (true) {
        /* Try local deque first */
        struct work_item *item = deque_pop(&deques[my_cpu]);
        if (item) {
            return item;
        }
        
        /* Local empty - become a thief */
        int victim = select_random_victim(my_cpu, num_cpus);
        item = deque_steal(&deques[victim]);
        
        if (item) {
            /* Successful steal - reset backoff state */
            failures = 0;
            consecutive_empty = 0;
            return item;
        }
        
        /* Failed steal - apply backoff strategy */
        failures++;
        
        if (deques[victim].bottom == deques[victim].top) {
            /* Victim was truly empty (not contention loss) */
            consecutive_empty++;
        }
        
        /* Adaptive backoff based on failure pattern */
        if (consecutive_empty >= num_cpus) {
            /* Probed everyone, all empty - system likely idle */
            /* User-space: yield or sleep */
            /* Kernel: check for new arrivals, maybe idle */
            
            if (check_for_system_idle()) {
                return NULL;  /* Signal caller to block/idle */
            }
            
            /* Brief sleep to avoid busy-wait */
            usleep(10);  /* 10μs */
            consecutive_empty = 0;
        } else if (failures > num_cpus / 2) {
            /* Many contention failures - back off */
            cpu_relax_n(failures * 2);
        }
        
        /* Check for termination signal */
        if (should_terminate()) {
            return NULL;
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

/* Task spawn implementation */
 
/* Spawn a new task - child goes on local deque */
void task_spawn(void (*function)(void *), void *arg) {
    int my_cpu = get_cpu_id();
    struct work_stealing_deque *my_deque = &worker_deques[my_cpu];
    struct work_item *parent = current_task[my_cpu];
    
    /* Allocate a work item for the child */
    struct work_item *child = alloc_work_item();
    child->function = function;
    child->argument = arg;
    child->parent = parent;
    child->pending_children = 0;
    
    /* Update parent's child count */
    if (parent) {
        atomic_fetch_add(&parent->pending_children, 1);
    }
    
    /* Push child onto local deque */
    /* The current task continues; child may be stolen */
    deque_push(my_deque, child);
}
 
/* Example: Parallel fibonacci using spawn */
int fib(int n) {
    if (n < 2) return n;
    
    int x, y;
    
    /* Spawn child for fib(n-1) */
    task_spawn(fib_helper, &(struct fib_args){n-1, &x});
    
    /* Continue with fib(n-2) */
    y = fib(n - 2);
    
    /* Wait for spawned child */
    task_sync();
    
    return x + y;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

/* Task synchronization (join) implementation */
 
/* Wait for all spawned children to complete */
void task_sync(void) {
    int my_cpu = get_cpu_id();
    struct work_item *current = current_task[my_cpu];
    
    /* Keep working until all children are done */
    while (atomic_load(&current->pending_children) > 0) {
        struct work_item *work = NULL;
        
        /* Step 1: Try to process local work */
        work = deque_pop(&worker_deques[my_cpu]);
        
        /* Step 2: If no local work, become a thief */
        if (!work) {
            work = work_steal_loop(my_cpu, worker_deques, num_workers);
        }
        
        /* Step 3: If we found work, execute it */
        if (work) {
            execute_work_item(work);
        } else {
            /* No work found anywhere - brief yield */
            cpu_relax();
        }
    }
    
    /* All children complete */
}
 
/* Execute a work item and handle completion */
void execute_work_item(struct work_item *item) {
    int my_cpu = get_cpu_id();
    
    /* Set as current task */
    struct work_item *previous = current_task[my_cpu];
    current_task[my_cpu] = item;
    
    /* Execute the work function */
    item->function(item->argument);
    
    /* Handle completion - decrement parent's pending count */
    if (item->parent) {
        int remaining = atomic_fetch_sub(&item->parent->pending_children, 1) - 1;
        
        /* If we were the last child, parent may now continue */
        /* (parent's sync loop will observe pending_children == 0) */
    }
    
    /* Restore previous task context */
    current_task[my_cpu] = previous;
    
    /* Return work item to pool */
    free_work_item(item);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

/* Java Fork/Join Framework concepts (simplified) */
 
/* Each worker thread has a work-stealing deque */
public class WorkStealingPool {
    final ForkJoinWorkerThread[] workers;
    final WorkQueue[] workQueues;  // One per worker
    
    /* Tasks extend this abstract class */
    abstract class ForkJoinTask<V> {
        volatile int status;  // Pending, completed, exceptional
        
        /* Fork: push task onto current thread's deque */
        public final ForkJoinTask<V> fork() {
            Thread t = Thread.currentThread();
            ForkJoinWorkerThread wt = (ForkJoinWorkerThread) t;
            wt.workQueue.push(this);
            return this;
        }
        
        /* Join: wait for task completion, helping if needed */
        public final V join() {
            if (status >= DONE) {
                return getRawResult();
            }
            // Help by processing other tasks while waiting
            return awaitDone(true);
        }
    }
    
    /* RecursiveTask: returns a value */
    abstract class RecursiveTask<V> extends ForkJoinTask<V> {
        V result;
        protected abstract V compute();
        
        protected void exec() {
            result = compute();
        }
    }
}
 
/* Example: Parallel sum using Fork/Join */
class SumTask extends RecursiveTask<Long> {
    static final int THRESHOLD = 1000;
    final long[] array;
    final int start, end;
    
    protected Long compute() {
        if (end - start <= THRESHOLD) {
            // Sequential sum for small ranges
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += array[i];
            }
            return sum;
        }
        
        // Split and fork
        int mid = (start + end) / 2;
        SumTask left = new SumTask(array, start, mid);
        SumTask right = new SumTask(array, mid, end);
        
        left.fork();  // Push left onto deque
        Long rightResult = right.compute();  // Execute right locally
        Long leftResult = left.join();  // Wait for left
        
        return leftResult + rightResult;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

/* Intel TBB work stealing concepts */
 
#include <tbb/parallel_for.h>
#include <tbb/task_group.h>
 
/* TBB handles task spawning implicitly via constructs like parallel_for */
void example_parallel_for() {
    std::vector<double> data(1000000);
    
    /* TBB automatically creates tasks and uses work stealing */
    tbb::parallel_for(
        tbb::blocked_range<size_t>(0, data.size()),
        [&](const tbb::blocked_range<size_t>& range) {
            for (size_t i = range.begin(); i < range.end(); ++i) {
                data[i] = expensive_computation(i);
            }
        }
    );
}
 
/* Explicit task-based parallelism with task_group */
void example_task_group() {
    tbb::task_group tg;
    
    // Spawn multiple tasks
    tg.run([]{ task_a(); });  // Goes onto local deque
    tg.run([]{ task_b(); });  // May be stolen
    tg.run([]{ task_c(); });  // May be stolen
    
    // Wait for all tasks
    tg.wait();  // Helps by processing work while waiting
}
 
/* TBB-specific: task priorities and affinities */
class PriorityTask : public tbb::task {
public:
    tbb::task* execute() override {
        // TBB's deque respects task priorities
        // Higher priority tasks are less likely to be stolen
        return nullptr;
    }
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

// Rust's Rayon work stealing library
 
use rayon::prelude::*;
 
// Rayon provides simple parallel iterators
fn parallel_sum(data: &[i64]) -> i64 {
    // par_iter() uses work stealing under the hood
    data.par_iter()
        .map(|&x| x * 2)
        .sum()
}
 
// join() is the primitive spawn-sync
fn parallel_quicksort<T: Ord + Send>(slice: &mut [T]) {
    if slice.len() <= 1 {
        return;
    }
    
    let pivot = partition(slice);
    let (left, right) = slice.split_at_mut(pivot);
    
    // rayon::join spawns tasks and uses work stealing
    rayon::join(
        || parallel_quicksort(left),
        || parallel_quicksort(&mut right[1..])  // Skip pivot
    );
}
 
// Under the hood: Rayon's registry manages worker threads
// Each worker has a Chase-Lev deque
// Work stealing handles load balancing automatically

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

/* Kernel scheduler concepts influenced by work stealing */
 
/* Per-CPU run queue with deque-like characteristics */
struct cfs_rq {
    /* Tasks organized in rb-tree by virtual runtime */
    /* Leftmost = highest priority = top of "deque" for stealing */
    struct rb_root_cached tasks_timeline;
    struct rb_node *rb_leftmost;
    
    /* Statistics for balance decisions */
    unsigned int nr_running;
    unsigned long load_weight;
};
 
/* Picking tasks to steal - similar to work stealing heuristics */
static struct task_struct *pick_task_to_steal(struct cfs_rq *src_cfs_rq,
                                               struct cfs_rq *dst_cfs_rq) {
    /* Linux picks from the "back" of the queue - tasks with
     * highest vruntime (lowest priority) are stolen first.
     * This is opposite from theoretical work stealing
     * (which steals the "oldest/largest" work) but reflects
     * OS scheduling priorities. */
    
    return __pick_last_entity(src_cfs_rq);
}
 
/* Kernel worker pools - explicit work stealing */
struct worker_pool {
    struct list_head worklist;      /* Pending work items */
    int nr_workers;                 /* Worker thread count */
    int nr_idle;                    /* Idle worker count */
    
    /* Cross-pool work migration when local pool is overwhelmed */
    struct worker_pool *steal_pool;  /* Pool to steal from */
};
 
/* Workqueue's work stealing between pools */
static struct work_struct *steal_work(struct worker_pool *pool) {
    struct worker_pool *src;
    
    /* Try to steal from sibling pools on same NUMA node */
    list_for_each_entry(src, &pool->steal_list, steal_list) {
        if (!list_empty(&src->worklist)) {
            struct work_struct *work;
            
            spin_lock(&src->lock);
            work = list_first_entry_or_null(&src->worklist,
                                            struct work_struct, entry);
            if (work) {
                list_del_init(&work->entry);
            }
            spin_unlock(&src->lock);
            
            if (work) return work;
        }
    }
    
    return NULL;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

/* Automatic granularity control techniques */
 
/* Technique 1: Static threshold (simple) */
#define SEQUENTIAL_THRESHOLD 1000
 
void parallel_for(int start, int end, void (*body)(int)) {
    if (end - start <= SEQUENTIAL_THRESHOLD) {
        // Sequential execution
        for (int i = start; i < end; i++) {
            body(i);
        }
    } else {
        // Parallel split
        int mid = (start + end) / 2;
        task_spawn(parallel_for_helper, start, mid, body);
        parallel_for(mid, end, body);  // Continue locally
        task_sync();
    }
}
 
/* Technique 2: Lazy task creation */
/* Only create stealable tasks when thieves are actively looking */
void lazy_parallel_for(int start, int end, void (*body)(int)) {
    while (start < end) {
        // Check if anyone wants to steal
        if (steal_request_pending()) {
            // Split: give half to thief
            int mid = (start + end) / 2;
            respond_to_steal_request(mid, end, body);
            end = mid;  // We continue with first half
        }
        
        // Process one iteration
        body(start++);
    }
}
 
/* Technique 3: Adaptive threshold */
/* Adjust threshold based on observed steal rate */
static __thread int adaptive_threshold = 1000;
 
void update_threshold_from_stats(int steals, int tasks_completed) {
    float steal_ratio = (float)steals / tasks_completed;
    
    if (steal_ratio > 0.1) {
        // Too many steals - tasks too coarse
        adaptive_threshold = adaptive_threshold * 3 / 4;
    } else if (steal_ratio < 0.01) {
        // Too few steals - tasks may be too fine
        adaptive_threshold = adaptive_threshold * 4 / 3;
    }
    
    // Clamp to reasonable range
    adaptive_threshold = clamp(adaptive_threshold, 100, 100000);
}