Pthread Condition Variables

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

#include <pthread.h>
#include <stdio.h>
#include <stdbool.h>
 
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
bool shutdown = false;
int workers_waiting = 0;
 
// =====================================================
// Broadcast: The Shutdown Pattern
// =====================================================
 
// Worker threads wait for work or shutdown
void *worker_thread(void *arg) {
    int id = *(int *)arg;
    
    pthread_mutex_lock(&mutex);
    
    while (!shutdown) {
        workers_waiting++;
        printf("Worker %d: waiting...
", id);
        
        pthread_cond_wait(&cond, &mutex);
        
        workers_waiting--;
        
        if (shutdown) break;
        
        // Do work...
    }
    
    printf("Worker %d: exiting due to shutdown
", id);
    pthread_mutex_unlock(&mutex);
    return NULL;
}
 
// Master thread initiates shutdown
void initiate_shutdown(void) {
    pthread_mutex_lock(&mutex);
    
    printf("Shutdown: setting flag and broadcasting
");
    shutdown = true;
    
    // MUST use broadcast - ALL workers need to know
    pthread_cond_broadcast(&cond);
    
    pthread_mutex_unlock(&mutex);
}
 
// =====================================================
// What Happens After Broadcast
// =====================================================
 
/*
 * Timeline with 3 workers waiting on cond:
 *
 * Master: lock(mutex)
 * Master: shutdown = true
 * Master: broadcast(cond)
 *   -> Worker 1, 2, 3 all marked as runnable
 * Master: unlock(mutex)
 *
 * Now all workers compete for the mutex:
 *
 * Worker 1: acquires mutex, sees shutdown=true, exits
 * Worker 1: unlock(mutex)
 *
 * Worker 2: acquires mutex, sees shutdown=true, exits
 * Worker 2: unlock(mutex)
 *
 * Worker 3: acquires mutex, sees shutdown=true, exits
 *
 * All workers have exited cleanly.
 * Signal would have only woken ONE worker!
 */

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

#include <pthread.h>
#include <stdbool.h>
 
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
 
int count = 0;
 
// Thread A: waits for count to become EVEN
void *wait_for_even(void *arg) {
    pthread_mutex_lock(&mutex);
    while (count % 2 != 0) {  // Wait while ODD
        pthread_cond_wait(&cond, &mutex);
    }
    printf("Even condition met: count = %d
", count);
    pthread_mutex_unlock(&mutex);
    return NULL;
}
 
// Thread B: waits for count > 10
void *wait_for_gt_10(void *arg) {
    pthread_mutex_lock(&mutex);
    while (count <= 10) {
        pthread_cond_wait(&cond, &mutex);
    }
    printf("Greater-than-10 condition met: count = %d
", count);
    pthread_mutex_unlock(&mutex);
    return NULL;
}
 
// WRONG: Using signal
void increment_WRONG(void) {
    pthread_mutex_lock(&mutex);
    count++;
    pthread_cond_signal(&cond);  // May wake wrong thread!
    pthread_mutex_unlock(&mutex);
}
 
// Example: count goes from 10 to 11
// - Thread A wants even (11 is odd, will re-wait)
// - Thread B wants > 10 (11 > 10, should proceed)
// - Signal wakes Thread A
// - Thread A checks: 11 % 2 != 0, goes back to sleep
// - Thread B never woken! LOST WAKEUP!
 
// CORRECT: Using broadcast
void increment_CORRECT(void) {
    pthread_mutex_lock(&mutex);
    count++;
    pthread_cond_broadcast(&cond);  // Both threads check their predicates
    pthread_mutex_unlock(&mutex);
}

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

#include <pthread.h>
#include <stdbool.h>
 
// Readers-Writers: Multiple readers can proceed simultaneously
 
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t readers_ok = PTHREAD_COND_INITIALIZER;
 
int active_writers = 0;
int waiting_readers = 0;
int active_readers = 0;
 
void *reader_thread(void *arg) {
    pthread_mutex_lock(&mutex);
    
    waiting_readers++;
    while (active_writers > 0) {
        pthread_cond_wait(&readers_ok, &mutex);
    }
    waiting_readers--;
    active_readers++;
    
    pthread_mutex_unlock(&mutex);
    
    // ... read ...
    
    pthread_mutex_lock(&mutex);
    active_readers--;
    // Signal writers if last reader (handled elsewhere)
    pthread_mutex_unlock(&mutex);
    
    return NULL;
}
 
void writer_finish(void) {
    pthread_mutex_lock(&mutex);
    
    active_writers = 0;
    
    // Multiple readers may be waiting - ALL should proceed
    // Using signal would only wake ONE reader!
    pthread_cond_broadcast(&readers_ok);
    
    pthread_mutex_unlock(&mutex);
}

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

#include <pthread.h>
#include <stdio.h>
#include <stdbool.h>
 
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
 
int available_connections = 0;
int waiting_threads = 0;
int wakeup_count = 0;
 
// =====================================================
// Thundering Herd Demonstration
// =====================================================
 
void *worker(void *arg) {
    int id = *(int *)arg;
    
    pthread_mutex_lock(&mutex);
    
    while (available_connections == 0) {
        waiting_threads++;
        printf("Thread %d: waiting for connection (waiters=%d)
", 
               id, waiting_threads);
        
        pthread_cond_wait(&cond, &mutex);
        
        waiting_threads--;
        wakeup_count++;
        printf("Thread %d: woken up (total wakeups=%d)
", id, wakeup_count);
    }
    
    // Got a connection
    available_connections--;
    printf("Thread %d: acquired connection
", id);
    
    pthread_mutex_unlock(&mutex);
    
    // Use connection...
    
    return NULL;
}
 
// One connection becomes available
void release_one_connection(void) {
    pthread_mutex_lock(&mutex);
    
    available_connections = 1;
    printf("
Releasing 1 connection with %d waiting threads
", 
           waiting_threads);
    
    // BROADCAST: All waiting threads wake up
    pthread_cond_broadcast(&cond);
    
    pthread_mutex_unlock(&mutex);
    
    // Result with 10 waiting threads:
    // - 10 threads wake up
    // - 10 threads serialize on mutex
    // - 1 thread gets connection, proceeds
    // - 9 threads find available_connections==0, go back to sleep
    // - Wasted: 9 wakeups + 10 mutex acquisitions
}
 
// BETTER: Use signal when only releasing one
void release_one_connection_optimized(void) {
    pthread_mutex_lock(&mutex);
    
    available_connections = 1;
    
    // SIGNAL: Only one thread can use this anyway
    pthread_cond_signal(&cond);
    
    pthread_mutex_unlock(&mutex);
    
    // Result: 1 thread wakes, 1 thread proceeds
}

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

// =====================================================
// Conceptual Implementation of pthread_cond_broadcast
// (Simplified for understanding)
// =====================================================
 
int pthread_cond_broadcast(pthread_cond_t *cond) {
    struct __pthread_cond_t *c = (struct __pthread_cond_t *)cond;
    
    // Quick check: any waiters?
    if (__atomic_load_n(&c->__nwaiters, __ATOMIC_RELAXED) == 0) {
        return 0;  // No waiters, nothing to do
    }
    
    // Acquire internal lock
    __lock(&c->__internal_lock);
    
    // Wake ALL waiting threads
    unsigned int n = c->__nwaiters;
    if (n > 0) {
        // Increment wakeup sequence by number of waiters
        c->__wakeup_seq += n;
        
        // Use futex to wake all
        // FUTEX_WAKE with count = INT_MAX (or n)
        futex_wake(&c->__futex_word, INT_MAX);
    }
    
    __unlock(&c->__internal_lock);
    
    return 0;
}
 
// =====================================================
// Why Broadcast Is O(N)
// =====================================================
 
/*
 * The broadcast operation itself is relatively cheap:
 * - One system call (futex_wake with count=ALL)
 * - One atomic increment
 * 
 * The O(N) cost comes from what happens AFTER broadcast:
 * 
 * 1. Kernel marks all N threads as runnable
 *    - Kernel work: O(N) queue operations
 *    
 * 2. All N threads get scheduled (eventually)
 *    - Context switches: potentially N, across CPUs
 *    
 * 3. All N threads try to acquire mutex
 *    - Serialization: N-1 threads block
 *    - Cache bouncing: mutex line ping-pongs
 *    
 * 4. Each thread runs, checks condition, mostly re-sleeps
 *    - Wasted work: N-1 useless wakeups
 *    
 * The system call is O(1), but total impact is O(N).
 */

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

#include <pthread.h>
#include <stdio.h>
 
// =====================================================
// Thread Barrier: All Threads Wait for Each Other
// =====================================================
 
typedef struct {
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    int threshold;      // Number of threads to wait for
    int count;          // Current number arrived
    int generation;     // Barrier reuse tracking
} barrier_t;
 
void barrier_init(barrier_t *b, int n) {
    pthread_mutex_init(&b->mutex, NULL);
    pthread_cond_init(&b->cond, NULL);
    b->threshold = n;
    b->count = 0;
    b->generation = 0;
}
 
void barrier_wait(barrier_t *b) {
    pthread_mutex_lock(&b->mutex);
    
    int my_generation = b->generation;
    b->count++;
    
    if (b->count == b->threshold) {
        // Last thread to arrive
        b->count = 0;
        b->generation++;  // Prepare for reuse
        
        // BROADCAST: Release ALL waiting threads
        pthread_cond_broadcast(&b->cond);
    } else {
        // Wait for last thread
        // Use while + generation to handle spurious wakeups correctly
        while (my_generation == b->generation) {
            pthread_cond_wait(&b->cond, &b->mutex);
        }
    }
    
    pthread_mutex_unlock(&b->mutex);
}
 
// Usage: parallel computation with phases
void *parallel_worker(void *arg) {
    barrier_t *barrier = (barrier_t *)arg;
    
    do_phase1_work();
    barrier_wait(barrier);  // All threads sync here
    
    do_phase2_work();
    barrier_wait(barrier);  // All threads sync again
    
    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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

#include <pthread.h>
#include <stdbool.h>
 
// =====================================================
// One-Shot Event: Signal Once, All Waiters Released
// =====================================================
 
typedef struct {
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    bool triggered;
} latch_t;
 
void latch_init(latch_t *l) {
    pthread_mutex_init(&l->mutex, NULL);
    pthread_cond_init(&l->cond, NULL);
    l->triggered = false;
}
 
// Wait for event (blocks until triggered)
void latch_wait(latch_t *l) {
    pthread_mutex_lock(&l->mutex);
    
    while (!l->triggered) {
        pthread_cond_wait(&l->cond, &l->mutex);
    }
    
    pthread_mutex_unlock(&l->mutex);
}
 
// Trigger event (releases all current and future waiters)
void latch_trigger(latch_t *l) {
    pthread_mutex_lock(&l->mutex);
    
    if (!l->triggered) {
        l->triggered = true;
        
        // Wake ALL current waiters
        pthread_cond_broadcast(&l->cond);
    }
    
    pthread_mutex_unlock(&l->mutex);
}
 
// Usage: initialization completion
latch_t app_initialized;
 
void *component_thread(void *arg) {
    // Wait for app to be ready
    latch_wait(&app_initialized);
    
    // Now safe to proceed
    do_work();
    
    return NULL;
}
 
void main_init(void) {
    latch_init(&app_initialized);
    
    // Start components (they'll wait)
    start_all_components();
    
    // Do initialization
    load_config();
    connect_to_database();
    
    // Signal all components: "ready to go!"
    latch_trigger(&app_initialized);
}

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

#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
 
// =====================================================
// Graceful Shutdown: Wait for All Workers to Exit
// =====================================================
 
typedef struct {
    pthread_mutex_t mutex;
    pthread_cond_t work_cond;       // Workers wait here for work
    pthread_cond_t shutdown_cond;   // Shutdown waits here for workers
    
    bool shutdown_requested;
    int active_workers;
    int total_workers;
    
    // Work queue fields...
} worker_pool_t;
 
void *worker(void *arg) {
    worker_pool_t *pool = (worker_pool_t *)arg;
    
    pthread_mutex_lock(&pool->mutex);
    pool->active_workers++;
    
    while (!pool->shutdown_requested) {
        // Wait for work or shutdown
        pthread_cond_wait(&pool->work_cond, &pool->mutex);
        
        if (pool->shutdown_requested) break;
        
        // Process work (would unlock during processing)
    }
    
    pool->active_workers--;
    
    // If last worker, signal shutdown waiter
    if (pool->active_workers == 0) {
        pthread_cond_signal(&pool->shutdown_cond);
    }
    
    printf("Worker exiting, %d remaining
", pool->active_workers);
    pthread_mutex_unlock(&pool->mutex);
    
    return NULL;
}
 
void pool_shutdown_and_wait(worker_pool_t *pool) {
    pthread_mutex_lock(&pool->mutex);
    
    printf("Initiating shutdown of %d workers
", pool->active_workers);
    pool->shutdown_requested = true;
    
    // BROADCAST: All workers must see shutdown
    pthread_cond_broadcast(&pool->work_cond);
    
    // Wait for all workers to exit
    while (pool->active_workers > 0) {
        pthread_cond_wait(&pool->shutdown_cond, &pool->mutex);
    }
    
    printf("All workers have exited
");
    pthread_mutex_unlock(&pool->mutex);
}

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

#include <pthread.h>
 
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
 
int waiters = 0;  // Track waiter count explicitly
 
void *consumer(void *arg) {
    pthread_mutex_lock(&mutex);
    
    waiters++;
    while (!has_work()) {
        pthread_cond_wait(&cond, &mutex);
    }
    waiters--;
    
    do_work();
    
    pthread_mutex_unlock(&mutex);
    return NULL;
}
 
void producer_add_work(void) {
    pthread_mutex_lock(&mutex);
    
    add_work_item();
    
    // OPTIMIZATION: Only broadcast if there are waiters
    if (waiters > 0) {
        pthread_cond_broadcast(&cond);
    }
    
    pthread_mutex_unlock(&mutex);
}
 
// Analysis:
// - Without optimization: broadcast every add, even if no waiters
// - With optimization: skip broadcast when waiters == 0
// - Savings: ~5,000 cycles per no-op broadcast avoided

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

#include <pthread.h>
 
// BAD: All waiters on one cond var
typedef struct {
    pthread_mutex_t mutex;
    pthread_cond_t cond;  // Everyone waits here
    int connection_count;
    int work_count;
} pool_bad_t;
 
void add_connection_bad(pool_bad_t *p) {
    pthread_mutex_lock(&p->mutex);
    p->connection_count++;
    // Must broadcast - both connection waiters AND work waiters are on cond
    pthread_cond_broadcast(&p->cond);
    pthread_mutex_unlock(&p->mutex);
}
 
// GOOD: Separate cond vars for separate conditions
typedef struct {
    pthread_mutex_t mutex;
    pthread_cond_t connection_available;  // Only connection waiters
    pthread_cond_t work_available;        // Only work waiters
    int connection_count;
    int work_count;
} pool_good_t;
 
void add_connection_good(pool_good_t *p) {
    pthread_mutex_lock(&p->mutex);
    p->connection_count++;
    // Only wake connection waiters
    pthread_cond_signal(&p->connection_available);  // Can even use signal!
    pthread_mutex_unlock(&p->mutex);
}
 
void add_work_good(pool_good_t *p) {
    pthread_mutex_lock(&p->mutex);
    p->work_count++;
    // Only wake work waiters
    pthread_cond_signal(&p->work_available);
    pthread_mutex_unlock(&p->mutex);
}

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

#include <pthread.h>
 
// Release one resource - usually only one waiter needs it
void release_resource(pool_t *p) {
    pthread_mutex_lock(&p->mutex);
    
    p->available++;
    
    // Common case: signal one waiter
    pthread_cond_signal(&p->resource_cond);
    
    pthread_mutex_unlock(&p->mutex);
}
 
// Release multiple resources atomically
void release_resources(pool_t *p, int n) {
    pthread_mutex_lock(&p->mutex);
    
    p->available += n;
    
    // Multiple resources: broadcast or signal n times
    if (n > 1) {
        pthread_cond_broadcast(&p->resource_cond);
    } else {
        pthread_cond_signal(&p->resource_cond);
    }
    
    pthread_mutex_unlock(&p->mutex);
}

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
102

#include <pthread.h>
#include <stdbool.h>
 
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
bool resource_ready = false;
 
// =====================================================
// MISTAKE 1: Broadcast for Single-Consumer Events
// =====================================================
 
void release_single_resource_INEFFICIENT(void) {
    pthread_mutex_lock(&mutex);
    resource_ready = true;
    
    // WASTEFUL: Only one thread can use the resource
    // Using broadcast wakes all, N-1 go back to sleep
    pthread_cond_broadcast(&cond);
    
    pthread_mutex_unlock(&mutex);
}
 
void release_single_resource_OPTIMAL(void) {
    pthread_mutex_lock(&mutex);
    resource_ready = true;
    
    // BETTER: Signal wakes exactly one waiter
    pthread_cond_signal(&cond);
    
    pthread_mutex_unlock(&mutex);
}
 
// =====================================================
// MISTAKE 2: Broadcast Inside a Loop
// =====================================================
 
void add_items_WRONG(int *items, int n) {
    for (int i = 0; i < n; i++) {
        pthread_mutex_lock(&mutex);
        add_item(items[i]);
        pthread_cond_broadcast(&cond);  // WRONG: Broadcast N times!
        pthread_mutex_unlock(&mutex);
    }
    // Result: N broadcasts, each waking all waiters = N^2 wakeups
}
 
void add_items_RIGHT(int *items, int n) {
    pthread_mutex_lock(&mutex);
    for (int i = 0; i < n; i++) {
        add_item(items[i]);
    }
    pthread_cond_broadcast(&cond);  // ONE broadcast after all adds
    pthread_mutex_unlock(&mutex);
}
 
// =====================================================
// MISTAKE 3: Missing Broadcast on All Exit Paths
// =====================================================
 
void produce_WRONG(void) {
    pthread_mutex_lock(&mutex);
    
    if (error_condition) {
        pthread_mutex_unlock(&mutex);
        return;  // WRONG: No broadcast, waiters may be stuck!
    }
    
    add_work();
    pthread_cond_broadcast(&cond);
    
    pthread_mutex_unlock(&mutex);
}
 
void produce_CORRECT(void) {
    pthread_mutex_lock(&mutex);
    
    if (error_condition) {
        error_flag = true;  // Set error state
        pthread_cond_broadcast(&cond);  // Wake waiters to see error
        pthread_mutex_unlock(&mutex);
        return;
    }
    
    add_work();
    pthread_cond_broadcast(&cond);
    
    pthread_mutex_unlock(&mutex);
}
 
// =====================================================
// MISTAKE 4: Expecting Broadcast to Transfer the Mutex
// =====================================================
 
void wake_and_expect_immediate_WRONG(void) {
    pthread_mutex_lock(&mutex);
    ready = true;
    pthread_cond_broadcast(&cond);
    // WRONG assumption: waiters are now running
    // REALITY: waiters won't run until WE release mutex
    do_more_work_holding_mutex();  // This delays waiters
    pthread_mutex_unlock(&mutex);
}