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
32
33
34
35

// Basic monitor structure with condition variables
monitor SharedResource {
    // Shared state protected by implicit monitor lock
    private data: ResourceState;
    private available: boolean = false;
    
    // Condition variable for synchronization
    condition resourceAvailable;
    
    // Producer procedure - makes resource available
    procedure produce(newData: ResourceState) {
        data = newData;
        available = true;
        
        // Signal a waiting consumer
        // QUESTION: What happens here exactly?
        signal(resourceAvailable);
        
        // After signal - is 'available' still true?
        // The answer depends on signal semantics!
    }
    
    // Consumer procedure - waits for resource
    procedure consume(): ResourceState {
        // Wait until resource is available
        while (!available) {
            wait(resourceAvailable);
        }
        
        // After wait returns - is 'available' guaranteed true?
        // Again, depends on signal semantics!
        available = false;
        return data;
    }
}

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

// Signal-and-Wait implementation within a monitor
// This shows the internal mechanics, not user-visible code
 
monitor implementation {
    // Monitor lock
    private lock: Lock;
    
    // Queue for threads trying to enter the monitor
    private entryQueue: Queue<Thread>;
    
    // URGENT queue for signaling threads (signal-and-wait specific)
    private urgentQueue: Queue<Thread>;
    private urgentCount: integer = 0;
    
    // Condition variable wait operation
    procedure wait(cv: ConditionVariable) {
        // Add current thread to condition's wait queue
        cv.waitQueue.enqueue(currentThread);
        
        // Release monitor - priority to urgent queue first
        if (urgentCount > 0) {
            // Signaling threads get priority
            wakeup(urgentQueue.dequeue());
            urgentCount--;
        } else if (!entryQueue.isEmpty()) {
            // Then threads waiting to enter
            wakeup(entryQueue.dequeue());
        } else {
            // Otherwise just release the lock
            lock.release();
        }
        
        // Block until signaled
        block();
        
        // When we wake, we already have the lock (transferred by signaler)
    }
    
    // Signal operation - Signal-and-Wait semantics
    procedure signal(cv: ConditionVariable) {
        if (!cv.waitQueue.isEmpty()) {
            Thread waiter = cv.waitQueue.dequeue();
            
            // KEY: Signaler joins urgent queue instead of continuing
            urgentQueue.enqueue(currentThread);
            urgentCount++;
            
            // Transfer lock directly to waiter
            // This is atomic - no gap where lock is free
            transferLock(waiter);
            
            // Signaler blocks here until waiter releases monitor
            block();
            
            // When we resume, we have the lock again
        }
        // If no waiters, signal has no effect - but signaler continues
    }
    
    // Monitor procedure exit
    procedure exitMonitor() {
        // Priority: urgent queue > entry queue
        if (urgentCount > 0) {
            wakeup(urgentQueue.dequeue());
            urgentCount--;
        } else 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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

monitor BoundedBuffer<T> {
    private buffer: T[N];
    private count: integer = 0;
    private in: integer = 0;
    private out: integer = 0;
    
    condition notEmpty;  // Signaled when buffer has items
    condition notFull;   // Signaled when buffer has space
    
    procedure insert(item: T) {
        // Wait while buffer is full
        if (count == N) {
            wait(notFull);
        }
        // INVARIANT: count < N is GUARANTEED here with signal-and-wait
        // No other thread could have filled the buffer between
        // the signal and our resumption
        
        buffer[in] = item;
        in = (in + 1) % N;
        count++;
        
        // Signal a waiting consumer
        signal(notEmpty);
        // We are now in urgent queue; consumer runs next
        
        // When we resume here, consumer has already taken an item
        // (or we weren't interrupted because no one was waiting)
    }
    
    procedure remove(): T {
        // Wait while buffer is empty
        if (count == 0) {
            wait(notEmpty);
        }
        // INVARIANT: count > 0 is GUARANTEED here with signal-and-wait
        // The producer that signaled us cannot have modified state
        // before we execute
        
        T item = buffer[out];
        out = (out + 1) % N;
        count--;
        
        // Signal a waiting producer
        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

// Hoare-style proof annotation for signal-and-wait monitor
 
monitor ProofExample {
    private resource: boolean = false;
    condition resourceAvailable;
    
    procedure acquire() {
        // {true}                      -- precondition
        if (!resource) {
            // {!resource}
            wait(resourceAvailable);
            // {resource}              -- postcondition of wait!
            //                         -- This is ONLY valid with signal-and-wait
            //                         -- Because: signal happens only when resource=true
            //                         --          and no interference before we resume
        }
        // {resource}                  -- either we never waited, or we got signaled
        resource = false;
        // {!resource}                 -- postcondition
    }
    
    procedure release() {
        // {true}                      -- precondition
        resource = true;
        // {resource}
        signal(resourceAvailable);     -- signal establishes {resource} for waiter
        // {resource}                  -- we still hold monitor, resource unchanged
        //                             -- (we're now in urgent queue, not in procedure)
    }
}

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

// Complexity: Nested signaling with urgent queue
 
// Initial state: T1 is in monitor, T2 waiting on cv1, T3 waiting on cv2
 
// T1 signals cv1
//   - T1 enters urgent queue
//   - T2 runs
 
// T2 signals cv2
//   - T2 enters urgent queue (now: [T1, T2] in urgent queue)
//   - T3 runs
 
// T3 completes
//   - T2 resumes (urgent queue priority)
//   - T2 completes
//   - T1 resumes (was first in urgent queue)
//   - T1 completes
 
// Total state in urgent queue peaks at: 2 threads
// This can grow arbitrarily with deep nesting!
 
// Stack visualization:
// Time →
// T1: [in monitor] → [signal] → [urgent:waiting...] → [...] → [resume] → [exit]
// T2:               [waiting] → [woken] → [signal] → [urgent:waiting] → [resume] → [exit]  
// T3:                          [waiting] → [woken] → [executing...] → [exit]

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

// Using signal-and-wait invariants for proof
 
monitor SafeStack<T> {
    private elements: T[MAX];
    private top: integer = 0;  // Stack size
    
    condition notEmpty;
    condition notFull;
    
    // Invariant: 0 <= top <= MAX
    
    procedure push(item: T) {
        // Precondition: true
        if (top == MAX) {
            // State: top == MAX
            wait(notFull);
            // GUARANTEE: top < MAX (from signal-and-wait invariant)
            // The signaler established top < MAX before signaling
            // and no interference occurred
        }
        // Assertion: top < MAX
        
        elements[top] = item;
        top++;
        
        // Assertion: top > 0 (we just pushed an element)
        
        signal(notEmpty);  // P = "top > 0" holds at signal
        // After signal: we're in urgent queue
        // When we resume: top may have decreased, but we're at end of procedure
    }
    
    procedure pop(): T {
        // Precondition: true
        if (top == 0) {
            // State: top == 0
            wait(notEmpty);
            // GUARANTEE: top > 0 (from signal-and-wait invariant)
        }
        // Assertion: top > 0
        
        top--;
        T item = elements[top];
        
        // Assertion: top < MAX (we just decremented)
        
        signal(notFull);  // P = "top < MAX" holds at signal
        
        return item;
    }
}
 
// Correctness proof sketch:
// 1. push() only signals notEmpty when top > 0 (after increment)
// 2. pop() receives this signal; signal-and-wait guarantees top > 0
// 3. pop() can safely decrement and access elements[top-1]
// 
// Similarly:
// 4. pop() only signals notFull when top < MAX (after decrement)
// 5. push() receives this signal; signal-and-wait guarantees top < MAX
// 6. push() can safely store at elements[top] and increment

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

monitor ReadersWriters {
    private readerCount: integer = 0;   // Number of active readers
    private writerActive: boolean = false;  // Is a writer active?
    private waitingWriters: integer = 0;    // Writers waiting (for fairness)
    
    condition canRead;   // Signaled when reading is allowed
    condition canWrite;  // Signaled when writing is allowed
    
    procedure startRead() {
        // Wait if writer is active OR writers are waiting (fairness)
        if (writerActive || waitingWriters > 0) {
            wait(canRead);
            // GUARANTEE (signal-and-wait): !writerActive && waitingWriters == 0
            // The signaler verified this before signaling us
        }
        
        readerCount++;
        
        // Allow other waiting readers to proceed
        // Signal-and-wait: we yield to them immediately
        signal(canRead);
        
        // When we resume, we continue with readerCount still valid
    }
    
    procedure endRead() {
        readerCount--;
        
        // If we're the last reader, let a waiting writer in
        if (readerCount == 0) {
            signal(canWrite);
            // INVARIANT passed to writer: readerCount == 0 && !writerActive
        }
        
        // If no writer waiting, exit quietly
    }
    
    procedure startWrite() {
        // Track that we're waiting (for reader fairness logic)
        waitingWriters++;
        
        // Wait if readers active or another writer active
        if (readerCount > 0 || writerActive) {
            wait(canWrite);
            // GUARANTEE (signal-and-wait): readerCount == 0 && !writerActive
        }
        
        waitingWriters--;
        writerActive = true;
    }
    
    procedure endWrite() {
        writerActive = false;
        
        // Prefer waiting readers (batch throughput)
        // Or prefer waiting writers (prevent writer starvation)
        // Choice here affects fairness properties
        
        if (waitingWriters > 0) {
            signal(canWrite);  // Let next writer in
        } else {
            signal(canRead);   // Let readers in (will cascade via startRead)
        }
    }
}
 
// Usage:
// 
// Reader thread:                Writer thread:
// readersWriters.startRead()    readersWriters.startWrite()
// ... read data ...             ... write data ...
// readersWriters.endRead()      readersWriters.endWrite()