Signal Semantics in 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

// The Thundering Herd Problem
 
monitor ConnectionPool {
    private available: List<Connection>;
    condition hasConnection;
    
    procedure getConnection(): Connection {
        while (available.isEmpty()) {
            wait(hasConnection);
        }
        return available.removeFirst();
    }
    
    procedure returnConnection(conn: Connection) {
        available.add(conn);
        
        // BUG PATTERN: broadcast wakes all waiters
        broadcast(hasConnection);  // Thundering herd!
        
        // Alternative: signal wakes only one
        signal(hasConnection);  // Better, but not always correct
    }
}
 
// Thundering herd scenario:
// - 100 threads waiting for connections
// - 1 connection returned, broadcast() called
// - All 100 threads wake up
// - 99 threads: recheck condition, find no connection, wait again
// - Result: 100 context switches for 1 useful wakeup
//           Plus lock contention from 100 simultaneous acquisitions

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

// Coarse-grained: One CV for all conditions
// Simple but may cause spurious processing
 
monitor CoarseGrained {
    condition changed;  // Single CV for everything
    
    // Must broadcast to ensure relevant waiter wakes
    procedure addItem() {
        items++;
        broadcast(changed);  // Wake all - some may not care
    }
    
    procedure setShutdown() {
        shutdown = true;
        broadcast(changed);  // Wake all - same CV
    }
}
 
// Fine-grained: Separate CVs for each condition
// More complex but targeted wakeups
 
monitor FineGrained {
    condition hasItems;     // CV for item availability
    condition hasSpace;     // CV for space availability
    condition shutdownCV;   // CV for shutdown
    
    procedure addItem() {
        items++;
        signal(hasItems);   // Wake only item waiters
    }
    
    procedure removeItem() {
        items--;
        signal(hasSpace);   // Wake only space waiters
    }
    
    procedure setShutdown() {
        shutdown = true;
        broadcast(shutdownCV);  // Wake shutdown waiters
    }
}

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

// Signal timing affects performance
 
// Pattern 1: Signal inside lock (standard)
void addItemStandard(Item item) {
    std::lock_guard<std::mutex> lock(mutex_);
    queue_.push(item);
    cv_.notify_one();  // Signal while holding lock
}  // Lock released here; waiter can run
 
// Pattern 2: Signal outside lock (potentially faster)
void addItemOptimized(Item item) {
    {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(item);
    }  // Lock released here
    cv_.notify_one();  // Signal after releasing lock
}
 
// Analysis:
// - Pattern 1: Waiter wakes, immediately blocks on lock (contention)
// - Pattern 2: Waiter wakes, lock is free, can proceed immediately
//
// However:
// - Pattern 2 can cause lost wakeup in some edge cases
// - Pattern 2 is technically race-prone (waiter might check before signal)
// - For safety, prefer Pattern 1 unless profiling shows it's a bottleneck
//
// Note: pthreads and C++ explicitly allow signaling outside the lock
// but the semantics are subtle. Modern implementations optimize this.

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

// Simplified view of pthread_cond_wait on Linux
 
int pthread_cond_wait_internal(pthread_cond_t *cond, 
                               pthread_mutex_t *mutex) {
    // Increment waiter count (atomic)
    atomic_increment(&cond->waiters);
    
    // Save current sequence number
    uint32_t seq = atomic_load(&cond->sequence);
    
    // Release mutex
    pthread_mutex_unlock(mutex);
    
    // Block in kernel until sequence changes (signal/broadcast)
    // This is the futex call
    futex_wait(&cond->sequence, seq);
    
    // Reacquire mutex before returning
    pthread_mutex_lock(mutex);
    
    // Decrement waiter count
    atomic_decrement(&cond->waiters);
    
    return 0;
}
 
int pthread_cond_signal_internal(pthread_cond_t *cond) {
    // Increment sequence (atomic)
    atomic_increment(&cond->sequence);
    
    // If there are waiters, wake one
    if (atomic_load(&cond->waiters) > 0) {
        futex_wake(&cond->sequence, 1);
    }
    
    return 0;
}
 
// Key insight: The sequence number is the futex word
// - Wait: blocks until sequence != current value
// - Signal: increments sequence, wakes waiter
// - Spurious wakeup: sequence matched, but condition false

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

// Windows condition variable usage
 
#include <windows.h>
 
SRWLOCK lock = SRWLOCK_INIT;
CONDITION_VARIABLE cv = CONDITION_VARIABLE_INIT;
BOOL dataReady = FALSE;
 
// Consumer
DWORD WINAPI Consumer(LPVOID arg) {
    AcquireSRWLockExclusive(&lock);
    
    while (!dataReady) {
        // SleepConditionVariableSRW atomically:
        // 1. Releases the lock
        // 2. Waits on the CV
        // 3. Reacquires the lock before returning
        SleepConditionVariableSRW(&cv, &lock, INFINITE, 0);
    }
    
    // Process data...
    
    ReleaseSRWLockExclusive(&lock);
    return 0;
}
 
// Producer
DWORD WINAPI Producer(LPVOID arg) {
    AcquireSRWLockExclusive(&lock);
    
    dataReady = TRUE;
    
    // Wake one waiter
    WakeConditionVariable(&cv);
    // Or wake all: WakeAllConditionVariable(&cv);
    
    ReleaseSRWLockExclusive(&lock);
    return 0;
}

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

// Java monitor conceptual structure
 
class Object {
    // Implicit fields (conceptual, not actual Java)
    transient Monitor _monitor;
    
    class Monitor {
        Thread owner;           // Thread holding the lock
        int recursionCount;     // Reentrant lock count
        Queue<Thread> entrySet; // Threads waiting to enter
        Queue<Thread> waitSet;  // Threads that called wait()
    }
}
 
// synchronized(obj) does:
// 1. Acquire obj._monitor.lock (or add to entrySet)
// 2. Set owner = currentThread
// 3. Execute synchronized block
// 4. Release lock, wake someone from entrySet
 
// obj.wait() does:
// 1. Verify we hold the lock
// 2. Add currentThread to waitSet
// 3. Release lock (fully, even if reentrant)
// 4. Block until signaled
// 5. Reacquire lock
// 6. Return
 
// obj.notify() does:
// 1. Verify we hold the lock
// 2. Move ONE thread from waitSet to entrySet
// 3. Continue executing (Mesa semantics)
 
// obj.notifyAll() does:
// 1. Verify we hold the lock  
// 2. Move ALL threads from waitSet to entrySet
// 3. Continue executing

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# Diagnosing stuck threads
 
# Linux: Get thread stack traces
gdb -p <pid>
(gdb) thread apply all bt
 
# Or use pstack
pstack <pid>
 
# Java: Thread dump
jstack <pid>
# Or send SIGQUIT: kill -3 <pid>
 
# Look for:
# - Multiple threads in pthread_cond_wait or Object.wait()
# - No threads holding the associated lock
# - No threads making progress toward signal
 
# Common causes:
# 1. Signal was never sent (forgot to signal in some code path)
# 2. Signal was sent before wait (missed signal / lost wakeup)
# 3. Wrong condition variable was signaled
# 4. Deadlock preventing signaler from running

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

// Diagnosing intermittent issues
 
// 1. Add logging with timestamps and thread IDs
void addItem(Item item) {
    std::lock_guard<std::mutex> lock(mutex_);
    LOG("[{}] Thread {} adding item", timestamp(), thread_id());
    
    queue_.push(item);
    LOG("[{}] Thread {} signaling, queue size = {}", 
        timestamp(), thread_id(), queue_.size());
    cv_.notify_one();
}
 
Item getItem() {
    std::unique_lock<std::mutex> lock(mutex_);
    LOG("[{}] Thread {} waiting for item", timestamp(), thread_id());
    
    while (queue_.empty()) {
        cv_.wait(lock);
        LOG("[{}] Thread {} woke up, queue size = {}", 
            timestamp(), thread_id(), queue_.size());
    }
    
    Item item = queue_.front();
    queue_.pop();
    LOG("[{}] Thread {} got item, queue size = {}", 
        timestamp(), thread_id(), queue_.size());
    return item;
}
 
// 2. Use thread sanitizer
// Compile with: g++ -fsanitize=thread ...
// This catches data races that signal/if bugs can cause
 
// 3. Add assertions for invariants
void getItem() {
    std::unique_lock<std::mutex> lock(mutex_);
    while (queue_.empty()) {
        cv_.wait(lock);
    }
    assert(!queue_.empty() && "If-instead-of-while bug detected!");
    // ...
}

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

// PREFER: Higher-level primitives that hide CV complexity
 
// Bad: Manual condition variable management
class ManualBlockingQueue<T> {
    std::mutex mutex_;
    std::condition_variable cv_;
    std::queue<T> queue_;
    
    void push(T item) { /* manual CV logic */ }
    T pop() { /* manual CV logic with while loop */ }
};
 
// Good: Use library-provided concurrent containers
#include <folly/MPMCQueue.h>  // Facebook's multi-producer multi-consumer queue
folly::MPMCQueue<T> queue;
 
// Or use standard library futures/promises
#include <future>
std::promise<T> promise;
std::future<T> future = promise.get_future();
// Producer: promise.set_value(result);
// Consumer: T result = future.get();
 
// Or use higher-level frameworks
// - Go: channels
// - Rust: crossbeam channels
// - Java: BlockingQueue, CompletableFuture

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

// Encapsulate condition variable patterns
 
// Reusable: A gate that blocks until opened
class Gate {
    std::mutex mutex_;
    std::condition_variable cv_;
    bool open_ = false;
    
public:
    void wait() {
        std::unique_lock<std::mutex> lock(mutex_);
        cv_.wait(lock, [this]{ return open_; });
    }
    
    template<typename Rep, typename Period>
    bool waitFor(std::chrono::duration<Rep, Period> timeout) {
        std::unique_lock<std::mutex> lock(mutex_);
        return cv_.wait_for(lock, timeout, [this]{ return open_; });
    }
    
    void open() {
        std::lock_guard<std::mutex> lock(mutex_);
        open_ = true;
        cv_.notify_all();
    }
    
    void close() {
        std::lock_guard<std::mutex> lock(mutex_);
        open_ = false;
    }
};
 
// Reusable: A one-shot latch
class Latch {
    std::mutex mutex_;
    std::condition_variable cv_;
    std::atomic<int> count_;
    
public:
    explicit Latch(int count) : count_(count) {}
    
    void countDown() {
        if (count_.fetch_sub(1) == 1) {
            std::lock_guard<std::mutex> lock(mutex_);
            cv_.notify_all();
        }
    }
    
    void await() {
        std::unique_lock<std::mutex> lock(mutex_);
        cv_.wait(lock, [this]{ return count_ == 0; });
    }
};

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

// Always use timeouts in production
 
Item getItemWithTimeout(std::chrono::milliseconds timeout) {
    std::unique_lock<std::mutex> lock(mutex_);
    
    auto deadline = std::chrono::steady_clock::now() + timeout;
    
    while (queue_.empty()) {
        auto remaining = deadline - std::chrono::steady_clock::now();
        if (remaining <= std::chrono::milliseconds::zero()) {
            throw TimeoutException("Timed out waiting for item");
        }
        
        auto status = cv_.wait_for(lock, remaining);
        // Even with wait_for returning timeout, the loop rechecks
    }
    
    return std::move(queue_.front());
    queue_.pop();
}
 
// Benefits of timeouts:
// 1. Prevents hung threads in production
// 2. Enables detection of subtle deadlocks
// 3. Allows periodic health checks
// 4. Enables graceful degradation under load

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

// C++ best practices for condition variables
 
// GOOD: Use wait with predicate
cv_.wait(lock, [this]{ return condition(); });
 
// GOOD: Use wait_for with predicate
if (!cv_.wait_for(lock, 100ms, [this]{ return condition(); })) {
    // Timeout handling
}
 
// AVOID: Manual while loops when predicate works
// They're more error-prone
 
// GOOD: Use std::unique_lock (not lock_guard) for wait
std::unique_lock<std::mutex> lock(mutex_);  // Can be unlocked
cv_.wait(lock, pred);
 
// WRONG: lock_guard doesn't support unlock needed by wait
// std::lock_guard<std::mutex> lock(mutex_);
// cv_.wait(lock);  // Won't compile

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

// Java best practices for synchronization
 
// AVOID: Low-level synchronized/wait/notify
synchronized (monitor) {
    while (!condition) {
        monitor.wait();  // Checked exception, single CV
    }
}
 
// GOOD: Use Lock and Condition from java.util.concurrent
private final Lock lock = new ReentrantLock();
private final Condition hasData = lock.newCondition();
private final Condition hasSpace = lock.newCondition();
 
public void put(T item) throws InterruptedException {
    lock.lock();
    try {
        while (isFull()) {
            hasSpace.await();  // Specific condition
        }
        buffer.add(item);
        hasData.signal();  // Wake only data waiters
    } finally {
        lock.unlock();
    }
}
 
// BEST: Use provided concurrent collections
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
 
BlockingQueue<T> queue = new LinkedBlockingQueue<>(100);
queue.put(item);     // Blocks if full
T item = queue.take(); // Blocks if empty

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

# Python best practices for condition variables
 
import threading
from queue import Queue  # Prefer this!
 
# AVOID: Manual condition variable management
condition = threading.Condition()
data = None
 
def consumer():
    with condition:  # Acquires underlying lock
        while data is None:  # while loop required!
            condition.wait()
        process(data)
 
def producer(item):
    with condition:
        global data
        data = item
        condition.notify()  # or notify_all()
 
# GOOD: Use wait_for (Python 3.2+)
def consumer_better():
    with condition:
        condition.wait_for(lambda: data is not None)
        process(data)
 
# BEST: Use queue.Queue - handles synchronization internally
from queue import Queue
 
q = Queue(maxsize=100)
q.put(item)    # Blocks if full
item = q.get()  # Blocks if empty

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

// Go best practices
 
// AVOID: sync.Cond - low-level, easy to misuse
var mu sync.Mutex
var cond = sync.NewCond(&mu)
var ready bool
 
func waiter() {
    mu.Lock()
    for !ready {  // Must use for, not if!
        cond.Wait()
    }
    // process
    mu.Unlock()
}
 
func signaler() {
    mu.Lock()
    ready = true
    cond.Signal()
    mu.Unlock()
}
 
// GOOD: Use channels instead
ch := make(chan Item, 100)
 
// Producer
ch <- item
 
// Consumer
item := <-ch
 
// Channels are Go's primary synchronization mechanism
// They're safer and more idiomatic than 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
80
81
82
83
84
85
86

// Production-quality thread pool
 
class ThreadPool {
public:
    explicit ThreadPool(size_t numThreads) : stop_(false) {
        for (size_t i = 0; i < numThreads; ++i) {
            workers_.emplace_back([this] { workerLoop(); });
        }
    }
    
    ~ThreadPool() {
        shutdown();
        for (auto& worker : workers_) {
            worker.join();
        }
    }
    
    template<typename F>
    std::future<typename std::result_of<F()>::type> submit(F&& task) {
        using ReturnType = typename std::result_of<F()>::type;
        
        auto packaged = std::make_shared<std::packaged_task<ReturnType()>>(
            std::forward<F>(task)
        );
        auto future = packaged->get_future();
        
        {
            std::lock_guard<std::mutex> lock(mutex_);
            if (stop_) {
                throw std::runtime_error("ThreadPool is stopped");
            }
            tasks_.emplace([packaged]{ (*packaged)(); });
        }
        cv_.notify_one();  // Wake one worker
        
        return future;
    }
    
    void shutdown() {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            stop_ = true;
        }
        cv_.notify_all();  // Wake all workers to see shutdown
    }
    
private:
    void workerLoop() {
        while (true) {
            std::function<void()> task;
            
            {
                std::unique_lock<std::mutex> lock(mutex_);
                
                // CRITICAL: while loop with correct predicate
                // Wait for: task available OR shutdown
                cv_.wait(lock, [this] {
                    return stop_ || !tasks_.empty();
                });
                
                // Check shutdown first (allows queue draining)
                if (stop_ && tasks_.empty()) {
                    return;  // Exit the worker loop
                }
                
                // Get task (we know queue is non-empty)
                task = std::move(tasks_.front());
                tasks_.pop();
            }
            
            // Execute outside lock
            task();
        }
    }
    
    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    std::mutex mutex_;
    std::condition_variable cv_;
    bool stop_;
};
 
// Usage:
// ThreadPool pool(4);
// auto future = pool.submit([]{ return compute(); });
// auto result = future.get();