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

#include <pthread.h>
#include <stdlib.h>
 
// =====================================================
// SCOPE 1: Global/Static Condition Variable
// =====================================================
// Suitable for program-wide coordination
// Lifetime: entire program execution
pthread_cond_t global_cond = PTHREAD_COND_INITIALIZER;
pthread_mutex_t global_mutex = PTHREAD_MUTEX_INITIALIZER;
 
// =====================================================
// SCOPE 2: Struct-Embedded Condition Variable
// =====================================================
// Condition variable as part of a data structure
// Lifetime: same as the containing structure
typedef struct {
    pthread_mutex_t mutex;
    pthread_cond_t not_empty;    // Signaled when buffer becomes non-empty
    pthread_cond_t not_full;     // Signaled when buffer becomes non-full
    int *buffer;
    int capacity;
    int count;
    int head;
    int tail;
} bounded_buffer_t;
 
// =====================================================
// SCOPE 3: Heap-Allocated (via containing struct)
// =====================================================
bounded_buffer_t *buffer_create(int capacity) {
    bounded_buffer_t *buf = malloc(sizeof(bounded_buffer_t));
    if (!buf) return NULL;
    
    buf->buffer = malloc(sizeof(int) * capacity);
    if (!buf->buffer) {
        free(buf);
        return NULL;
    }
    
    // Must use pthread_*_init for heap-allocated objects
    pthread_mutex_init(&buf->mutex, NULL);
    pthread_cond_init(&buf->not_empty, NULL);
    pthread_cond_init(&buf->not_full, NULL);
    
    buf->capacity = capacity;
    buf->count = 0;
    buf->head = 0;
    buf->tail = 0;
    
    return buf;
}

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>
 
// =====================================================
// Static Initialization: The Simplest Approach
// =====================================================
 
// Global condition variable - initialized at compile time
static pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t g_data_ready = PTHREAD_COND_INITIALIZER;
static int g_data_available = 0;
 
void producer(void) {
    pthread_mutex_lock(&g_mutex);
    
    // Produce data...
    g_data_available = 1;
    
    pthread_cond_signal(&g_data_ready);
    pthread_mutex_unlock(&g_mutex);
}
 
void *consumer(void *arg) {
    pthread_mutex_lock(&g_mutex);
    
    while (!g_data_available) {
        // Wait releases mutex, blocks, reacquires mutex on wakeup
        pthread_cond_wait(&g_data_ready, &g_mutex);
    }
    
    // Consume data...
    g_data_available = 0;
    
    pthread_mutex_unlock(&g_mutex);
    return NULL;
}
 
// Note: Static initialization means no pthread_cond_destroy() 
// is strictly required for global variables (though it's good practice
// in library code that might be unloaded).

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

#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <errno.h>
 
// =====================================================
// Dynamic Initialization: For Runtime-Created Objects
// =====================================================
 
typedef struct work_queue {
    pthread_mutex_t mutex;
    pthread_cond_t  not_empty;
    // ... queue data
} work_queue_t;
 
// Proper initialization with error handling
work_queue_t *work_queue_create(void) {
    work_queue_t *wq = malloc(sizeof(work_queue_t));
    if (!wq) {
        return NULL;
    }
    
    // Initialize mutex first
    int rc = pthread_mutex_init(&wq->mutex, NULL);
    if (rc != 0) {
        fprintf(stderr, "mutex init failed: %d
", rc);
        free(wq);
        return NULL;
    }
    
    // Then initialize condition variable
    rc = pthread_cond_init(&wq->not_empty, NULL);
    if (rc != 0) {
        fprintf(stderr, "cond init failed: %d
", rc);
        // Must clean up already-initialized mutex!
        pthread_mutex_destroy(&wq->mutex);
        free(wq);
        return NULL;
    }
    
    // Initialize other fields...
    
    return wq;
}
 
// Corresponding destruction
void work_queue_destroy(work_queue_t *wq) {
    if (!wq) return;
    
    // Destroy in reverse order of initialization
    int rc = pthread_cond_destroy(&wq->not_empty);
    if (rc != 0) {
        fprintf(stderr, "Warning: cond destroy failed: %d
", rc);
    }
    
    rc = pthread_mutex_destroy(&wq->mutex);
    if (rc != 0) {
        fprintf(stderr, "Warning: mutex destroy failed: %d
", rc);
    }
    
    free(wq);
}

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 <time.h>
#include <stdio.h>
 
// =====================================================
// Configuring Condition Variable Attributes
// =====================================================
 
// Example 1: Process-Shared Condition Variable
// (For use in shared memory between processes)
int create_shared_condvar(pthread_cond_t *cond) {
    pthread_condattr_t attr;
    int rc;
    
    rc = pthread_condattr_init(&attr);
    if (rc != 0) return rc;
    
    // Enable process sharing
    rc = pthread_condattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
    if (rc != 0) {
        pthread_condattr_destroy(&attr);
        return rc;
    }
    
    rc = pthread_cond_init(cond, &attr);
    pthread_condattr_destroy(&attr);  // Always clean up
    
    return rc;
}
 
// Example 2: Monotonic Clock for Robust Timeouts
// CRITICAL: Use CLOCK_MONOTONIC to prevent issues with system time changes
int create_monotonic_condvar(pthread_cond_t *cond) {
    pthread_condattr_t attr;
    int rc;
    
    rc = pthread_condattr_init(&attr);
    if (rc != 0) return rc;
    
    // Use monotonic clock for timedwait
    // This prevents spurious timeouts/waits when system time is adjusted
    rc = pthread_condattr_setclock(&attr, CLOCK_MONOTONIC);
    if (rc != 0) {
        pthread_condattr_destroy(&attr);
        return rc;
    }
    
    rc = pthread_cond_init(cond, &attr);
    pthread_condattr_destroy(&attr);
    
    return rc;
}
 
// Example 3: Query Current Attributes
void print_condvar_attributes(pthread_condattr_t *attr) {
    int pshared;
    clockid_t clock_id;
    
    pthread_condattr_getpshared(attr, &pshared);
    pthread_condattr_getclock(attr, &clock_id);
    
    printf("Process-shared: %s
", 
           pshared == PTHREAD_PROCESS_SHARED ? "yes" : "no");
    printf("Clock: %s
",
           clock_id == CLOCK_MONOTONIC ? "MONOTONIC" : "REALTIME");
}

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

#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdbool.h>
 
// =====================================================
// Safe Destruction Patterns
// =====================================================
 
typedef struct {
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    bool shutdown;
    int waiting_count;  // Track waiters for safe shutdown
    // ... other fields
} worker_pool_t;
 
// Pattern 1: Graceful Shutdown with Coordination
void worker_pool_shutdown(worker_pool_t *pool) {
    pthread_mutex_lock(&pool->mutex);
    
    // Signal shutdown to all workers
    pool->shutdown = true;
    
    // Wake ALL waiting threads so they can exit
    pthread_cond_broadcast(&pool->cond);
    
    pthread_mutex_unlock(&pool->mutex);
    
    // In a real implementation, you would join worker threads here
    // to ensure they've actually exited before destroying
}
 
void worker_pool_destroy(worker_pool_t *pool) {
    if (!pool) return;
    
    // Ensure shutdown was called first
    pthread_mutex_lock(&pool->mutex);
    if (!pool->shutdown) {
        pthread_mutex_unlock(&pool->mutex);
        fprintf(stderr, "Error: must call shutdown before destroy
");
        return;
    }
    
    // Check for remaining waiters
    if (pool->waiting_count > 0) {
        pthread_mutex_unlock(&pool->mutex);
        fprintf(stderr, "Error: %d threads still waiting
", 
                pool->waiting_count);
        return;
    }
    pthread_mutex_unlock(&pool->mutex);
    
    // Safe to destroy now
    int rc = pthread_cond_destroy(&pool->cond);
    if (rc == EBUSY) {
        fprintf(stderr, "Warning: condition variable still in use
");
        // Handle error - this shouldn't happen if shutdown was correct
    }
    
    pthread_mutex_destroy(&pool->mutex);
    free(pool);
}
 
// Pattern 2: Worker Thread that Tracks Waiting State
void *worker_thread(void *arg) {
    worker_pool_t *pool = (worker_pool_t *)arg;
    
    pthread_mutex_lock(&pool->mutex);
    
    while (!pool->shutdown) {
        // Track that we're about to wait
        pool->waiting_count++;
        
        pthread_cond_wait(&pool->cond, &pool->mutex);
        
        // No longer waiting
        pool->waiting_count--;
    }
    
    pthread_mutex_unlock(&pool->mutex);
    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

// =====================================================
// Conceptual Internal Structure (Simplified)
// Note: Actual implementations are more complex
// =====================================================
 
struct __pthread_cond_s {
    // Sequence number - incremented on each signal/broadcast
    // Used to detect missed wakeups after spurious kernel returns
    unsigned int __wseq;         // Wait sequence
    unsigned int __g1_start;     // Generation 1 start
    
    // Reference counts for waiting groups
    unsigned int __g_refs[2];    // Group reference counts
    unsigned int __g_size[2];    // Group sizes
    
    // Internal state (clock type, etc.)
    unsigned int __g1_orig_size;
    unsigned int __wrefs;
    unsigned int __g_signals[2];
};
 
// Key behavioral implications:
//
// 1. SIGNAL WHEN NO WAITERS:
//    - Nearly free (just atomic increment of sequence)
//    - No system call needed
//
// 2. WAIT WITH IMMEDIATE SIGNAL AVAILABLE:
//    - May return immediately without blocking
//    - Sequence comparison detects pending signal
//
// 3. BLOCKING WAIT:
//    - Uses futex to sleep efficiently in kernel
//    - Thread is descheduled, consuming no CPU
//
// 4. WAKEUP:
//    - futex_wake notifies kernel to reschedule waiter
//    - Waking thread doesn't context-switch to waiter

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

#include <pthread.h>
#include <stdio.h>
 
// =====================================================
// MISTAKE 1: Wrong Mutex Association
// =====================================================
 
pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mutex2 = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
 
void *thread_a(void *arg) {
    pthread_mutex_lock(&mutex1);
    pthread_cond_wait(&cond, &mutex1);  // Uses mutex1
    pthread_mutex_unlock(&mutex1);
    return NULL;
}
 
void *thread_b(void *arg) {
    pthread_mutex_lock(&mutex2);
    pthread_cond_wait(&cond, &mutex2);  // WRONG: Same cond, different mutex!
    pthread_mutex_unlock(&mutex2);       // Undefined behavior
    return NULL;
}
 
// CORRECT: Use the same mutex with a given condition variable
 
// =====================================================
// MISTAKE 2: Signaling Without Mutex
// =====================================================
 
// While technically allowed, signaling without the mutex is DANGEROUS:
 
int data_ready = 0;
 
void producer_wrong(void) {
    data_ready = 1;                    // Modify shared state
    pthread_cond_signal(&cond);        // Signal without lock
    // Race: consumer might test data_ready==0, miss signal,
    // then block forever after signal is lost
}
 
void producer_correct(void) {
    pthread_mutex_lock(&mutex1);
    data_ready = 1;                    // Modify shared state
    pthread_cond_signal(&cond);        // Signal WITH lock held
    pthread_mutex_unlock(&mutex1);
    // Safe: consumer either sees data_ready==1 before waiting,
    // or is already waiting and will be woken
}
 
// =====================================================
// MISTAKE 3: Using IF Instead of WHILE
// =====================================================
 
void consumer_wrong(void) {
    pthread_mutex_lock(&mutex1);
    
    if (!data_ready) {                 // WRONG: if
        pthread_cond_wait(&cond, &mutex1);
    }
    // Spurious wakeup can leave data_ready==0 here!
    process_data();  // BUG: may process nonexistent data
    
    pthread_mutex_unlock(&mutex1);
}
 
void consumer_correct(void) {
    pthread_mutex_lock(&mutex1);
    
    while (!data_ready) {              // CORRECT: while loop
        pthread_cond_wait(&cond, &mutex1);
    }
    // Guaranteed: data_ready==1 here
    process_data();
    
    pthread_mutex_unlock(&mutex1);
}