Signal and Continue: Deferred Activation Semantics

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

// Signal-and-Continue implementation within a monitor
// This is the model used by pthreads, Java, and modern systems
 
monitor implementation {
    // Monitor lock
    private lock: Lock;
    
    // Queue for threads trying to enter the monitor
    private entryQueue: Queue<Thread>;
    
    // NOTE: No urgent queue needed! (simpler than signal-and-wait)
    
    // Condition variable wait operation
    procedure wait(cv: ConditionVariable) {
        // Add current thread to condition's wait queue
        cv.waitQueue.enqueue(currentThread);
        
        // Release monitor lock
        if (!entryQueue.isEmpty()) {
            wakeup(entryQueue.dequeue());
        } else {
            lock.release();
        }
        
        // Block until signaled
        block();
        
        // CRITICAL DIFFERENCE: When we wake, we do NOT have the lock
        // We must reacquire it like any entering thread
        
        // Reacquire the lock
        while (!lock.tryAcquire()) {
            entryQueue.enqueue(currentThread);
            block();
        }
        
        // Now we have the lock, but condition may be FALSE!
        // Caller MUST recheck the condition
    }
    
    // Signal operation - Signal-and-Continue semantics
    procedure signal(cv: ConditionVariable) {
        if (!cv.waitQueue.isEmpty()) {
            Thread waiter = cv.waitQueue.dequeue();
            
            // Move waiter to entry queue (or ready queue)
            // They will compete for the lock when we release it
            entryQueue.enqueue(waiter);
            wakeup(waiter);
            
            // KEY DIFFERENCE: We continue executing here
            // We still hold the lock
            // We can modify state further before releasing
        }
        // If no waiters, signal has no effect
    }
    
    // Monitor procedure exit
    procedure exitMonitor() {
        // Simply release lock; next thread in entry queue gets it
        if (!entryQueue.isEmpty()) {
            wakeup(entryQueue.dequeue());
        } else {
            lock.release();
        }
    }
}

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

// Example: Signaler invalidates condition before waiter runs
 
monitor Problem {
    private available: boolean = false;
    condition becameAvailable;
    
    procedure produce() {
        available = true;
        signal(becameAvailable);  // Wake up waiter
        
        // But we continue executing here...
        doSomeMoreWork();
        
        // Bug: We might invalidate the condition
        available = false;  // Waiter's condition is now invalid!
        
        // When waiter eventually runs, available == false
    }
    
    procedure consume() {
        if (!available) {   // BUG: Using 'if' with signal-and-continue
            wait(becameAvailable);
        }
        // With signal-and-continue, available might be FALSE here!
        // The 'if' check doesn't protect us
        
        doConsume();  // May operate on invalid state
    }
}

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

// Example: Another thread invalidates condition
 
// Timeline:
// T1: Producer - calls produce()
// T2: Consumer - waiting on becameAvailable
// T3: Consumer - about to enter consume()
 
// Time 0: T2 is waiting, T1 and T3 are ready
 
// Time 1: T1 executes
//   - available = true
//   - signal(becameAvailable)  -> T2 moved to entry queue
//   - T1 continues, eventually exits
 
// Time 2: T3 happens to acquire lock before T2
//   (Scheduler gives T3 priority, or T3 was already in entry queue)
//   - T3 sees available == true
//   - T3 sets available = false
//   - T3 exits
 
// Time 3: T2 finally acquires lock
//   - T2 was signaled (believes available should be true)
//   - But available == false (T3 consumed it)
//   - If T2 used 'if', it proceeds incorrectly
//   - If T2 used 'while', it rechecks and waits again
 
monitor BoundedBuffer {
    private count: integer = 0;
    condition notEmpty;
    
    procedure insert(item: T) {
        // ... insert logic ...
        count++;
        signal(notEmpty);  // Wake a consumer
    }
    
    procedure remove(): T {
        if (count == 0) {  // BUG!
            wait(notEmpty);
        }
        // Another consumer might have taken the item
        // between signal and our lock acquisition
        
        count--;  // BUG: count might already be 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

// THE CORRECT PATTERN for signal-and-continue
 
monitor CorrectExample {
    private condition: boolean = false;
    condition conditionVar;
    
    procedure waitForCondition() {
        // CORRECT: while loop rechecks condition after every wakeup
        while (!condition) {
            wait(conditionVar);
            // After waking, loop back and recheck!
            // If condition is still false, wait again
        }
        // When we exit the loop, condition is DEFINITELY true
        // (We hold the lock and just verified it)
    }
    
    procedure makeConditionTrue() {
        condition = true;
        signal(conditionVar);
        // We can continue doing things here...
        // The waiter will recheck anyway
    }
}
 
// Why while works:
// 1. Initial check: if condition is true, skip wait entirely
// 2. Wait: block until signaled
// 3. Wake up: reacquire lock
// 4. Recheck: if condition became false, wait again
// 5. Only exit when condition verified true while holding lock

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

// Correct bounded buffer with signal-and-continue
 
monitor BoundedBuffer<T> {
    private buffer: T[N];
    private count: integer = 0;
    private in: integer = 0;
    private out: integer = 0;
    
    condition notEmpty;
    condition notFull;
    
    procedure insert(item: T) {
        // CORRECT: while loop for all waits
        while (count == N) {
            wait(notFull);
            // After waking: recheck! Another producer might have filled buffer
        }
        
        // GUARANTEED: count < N (verified while holding lock)
        buffer[in] = item;
        in = (in + 1) % N;
        count++;
        
        signal(notEmpty);
        // We continue here; consumer will recheck before consuming
    }
    
    procedure remove(): T {
        // CORRECT: while loop protects against invalidation
        while (count == 0) {
            wait(notEmpty);
            // After waking: recheck! Another consumer might have taken item
        }
        
        // GUARANTEED: count > 0 (verified while holding lock)
        T item = buffer[out];
        out = (out + 1) % N;
        count--;
        
        signal(notFull);
        
        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

// Real-world spurious wakeup handling in pthreads
 
#include <pthread.h>
#include <stdbool.h>
 
typedef struct {
    pthread_mutex_t mutex;
    pthread_cond_t condition;
    bool data_ready;
} shared_state_t;
 
void* consumer(void* arg) {
    shared_state_t* state = (shared_state_t*)arg;
    
    pthread_mutex_lock(&state->mutex);
    
    // MANDATORY: while loop handles spurious wakeups
    // POSIX spec explicitly states:
    //   "Spurious wakeups from pthread_cond_wait() may occur"
    while (!state->data_ready) {
        // pthread_cond_wait atomically:
        // 1. Releases mutex
        // 2. Blocks on condition
        // 3. (eventually) Reacquires mutex before returning
        pthread_cond_wait(&state->condition, &state->mutex);
        
        // We might be here due to:
        // - An explicit signal (data_ready might be true)
        // - A spurious wakeup (data_ready is still false)
        // The while loop handles both correctly!
    }
    
    // GUARANTEED: data_ready == true && we hold the lock
    process_data();
    
    pthread_mutex_unlock(&state->mutex);
    return NULL;
}
 
void* producer(void* arg) {
    shared_state_t* state = (shared_state_t*)arg;
    
    pthread_mutex_lock(&state->mutex);
    
    prepare_data();
    state->data_ready = true;
    
    // Signal waiting thread(s)
    pthread_cond_signal(&state->condition);
    
    // We continue executing here (signal-and-continue)
    pthread_mutex_unlock(&state->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

// Signal-and-continue enables efficient batching
 
monitor TaskQueue {
    private tasks: Queue<Task>;
    condition tasksAvailable;
    
    // Producer adds multiple tasks atomically
    procedure addTasks(newTasks: List<Task>) {
        // Add all tasks while holding lock
        for task in newTasks {
            tasks.enqueue(task);
            signal(tasksAvailable);  // Wake a worker for each
            // With signal-and-continue, we continue here
            // All signals are "queued up"
        }
        // When we release the lock, all awakened workers
        // compete fairly for tasks
        
        // With signal-and-wait, we would context switch
        // after EACH signal, serializing the additions
        // and drastically reducing throughput
    }
    
    procedure getTask(): Task {
        while (tasks.isEmpty()) {
            wait(tasksAvailable);
        }
        return tasks.dequeue();
    }
}
 
// Example: Adding 100 tasks
// Signal-and-wait: 200 context switches during addTasks()
// Signal-and-continue: ~0 context switches during addTasks()
//                      Workers wake when addTasks() completes

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

// pthreads implementation of condition variables
 
#include <pthread.h>
#include <stdio.h>
 
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
int shared_resource = 0;
 
void* waiter_thread(void* arg) {
    pthread_mutex_lock(&mutex);
    
    // Canon pattern: while loop + wait
    while (shared_resource == 0) {
        // pthread_cond_wait:
        // 1. Atomically unlocks mutex and waits on cond
        // 2. When signaled, reacquires mutex before returning
        // 3. May return spuriously (while loop handles this)
        pthread_cond_wait(&cond, &mutex);
    }
    
    printf("Got resource: %d
", shared_resource);
    shared_resource = 0;  // Consume
    
    pthread_mutex_unlock(&mutex);
    return NULL;
}
 
void* signaler_thread(void* arg) {
    pthread_mutex_lock(&mutex);
    
    shared_resource = 42;
    
    // Signal: move one waiter to ready queue
    // We continue holding the mutex
    pthread_cond_signal(&cond);
    
    // We can do more work here...
    printf("Signaled, continuing with more work
");
    do_more_work();
    
    pthread_mutex_unlock(&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
39
40
41
42
43
44
45
46
47
48

// Java implementation of condition variables
 
public class BoundedBuffer<T> {
    private final Object[] buffer;
    private int count = 0, in = 0, out = 0;
    
    public BoundedBuffer(int capacity) {
        buffer = new Object[capacity];
    }
    
    public synchronized void put(T item) throws InterruptedException {
        // Canonical while loop pattern
        while (count == buffer.length) {
            // Object.wait():
            // 1. Releases the monitor lock
            // 2. Waits until notify()/notifyAll()
            // 3. Reacquires lock before returning
            // 4. May return spuriously!
            wait();
        }
        
        buffer[in] = item;
        in = (in + 1) % buffer.length;
        count++;
        
        // Signal-and-continue: notify() doesn't yield
        notify();  // or notifyAll()
        
        // We continue here, still holding the lock
        // Awakened thread won't run until we exit synchronized block
    }
    
    @SuppressWarnings("unchecked")
    public synchronized T take() throws InterruptedException {
        // Must use while, not if!
        while (count == 0) {
            wait();
        }
        
        T item = (T) buffer[out];
        buffer[out] = null;
        out = (out + 1) % buffer.length;
        count--;
        
        notify();
        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

// C++ std::condition_variable example
 
#include <condition_variable>
#include <mutex>
#include <queue>
 
template<typename T>
class ThreadSafeQueue {
private:
    std::queue<T> queue_;
    mutable std::mutex mutex_;
    std::condition_variable cond_;
    
public:
    void push(T value) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.push(std::move(value));
        cond_.notify_one();  // Signal-and-continue
        // We continue, lock will be released when lock_guard destructs
    }
    
    T pop() {
        std::unique_lock<std::mutex> lock(mutex_);
        
        // wait() with predicate - handles while loop internally!
        // This is syntactic sugar provided by C++
        cond_.wait(lock, [this] { return !queue_.empty(); });
        // Equivalent to:
        // while (queue_.empty()) {
        //     cond_.wait(lock);
        // }
        
        T value = std::move(queue_.front());
        queue_.pop();
        return value;
    }
};

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

// THE CLASSIC BUG
 
monitor BuggyQueue {
    private queue: Queue<Item>;
    condition hasItems;
    
    procedure dequeue(): Item {
        // BUG: if instead of while
        if (queue.isEmpty()) {
            wait(hasItems);
        }
        
        // BUG: queue might be empty here!
        // Scenarios:
        //   1. Another consumer dequeued between signal and our lock acquisition
        //   2. Spurious wakeup occurred
        //   3. Signaler enqueued then dequeued before releasing lock
        
        return queue.remove();  // CRASH or unexpected behavior
    }
}
 
// FIXED:
procedure dequeue(): Item {
    while (queue.isEmpty()) {  // CORRECT
        wait(hasItems);
    }
    return queue.remove();  // Safe: we verified queue is non-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

// BUG: Signaling without holding the lock
 
void producer() {
    mutex.lock();
    shared_data = produce_item();
    mutex.unlock();  // Released lock
    
    // BUG: Signaling without lock held
    // Race condition: between unlock and signal,
    // a consumer might check the condition, find it true,
    // AND consume the data before we even signal
    cond.notify_one();  // Lost signal if consumer runs first
}
 
// CORRECT:
void producer() {
    mutex.lock();
    shared_data = produce_item();
    cond.notify_one();  // Signal while holding lock
    mutex.unlock();
}
 
// Some systems tolerate signaling without the lock,
// but it's a race condition and bad practice.
// Always signal while holding the associated lock.

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

// BUG: Forgotten signal leads to indefinite waiting
 
monitor LeakyBuffer {
    private buffer: Item[];
    private count: integer = 0;
    condition notEmpty;
    condition notFull;
    
    procedure insert(item: Item) {
        while (count == MAX) {
            wait(notFull);
        }
        buffer[count++] = item;
        
        // BUG: Forgot to signal notEmpty!
        // Consumers will wait forever even though buffer has items
    }
    
    procedure remove(): Item {
        while (count == 0) {
            wait(notEmpty);  // May wait forever
        }
        Item item = buffer[--count];
        
        signal(notFull);  // At least this one is present
        return item;
    }
}