💻 CSC3150 Week 5 More on Concurrency: Monitor

Lithos

CSC3150 Lecture8 Monitors

在用 3 个 Semaphore 实现 Buffered Producer-Consumer Lock 中，我们发现仅仅交换两个 P() 就可能造成 Deadlock —— 我们希望有一种更清晰的抽象把 Mutex 和 Scheduling Constraints 分开表达，这就是 Monitor 的设计思想。

Monitors 管程

Monitor 管程抽象 = 一个 Lock + 零个或多个 Condition Variables，用于协调对共享数据的并发访问。

其中锁负责维护 Mutex；条件变量负责维持 Scheduling 限制

Monitor 并不一定是某种独立的硬件结构或特殊对象，它首先是一种并发编程的组织方式 (Programming Paradigm)。Java 等语言原生提供 Monitor 机制；其他语言可以通过 Lock + Condition Variable 实现。


条件变量

Condition Variable 是一个等待队列，用于存放等待某个条件成立的线程 —— queue for threads in critical section to sleep

例如 Condition not_empty; 代表可能存在一些 Consumer 线程正在等待 Buffer 变成非空，它们并非 Busy Waiting，而是 BLOCKED Sleeping，当 Producer 添加数据后，可以通知它们。但是 ⚠️ Condition Variable 本身不存储条件是否成立。

对比 Semaphores：

  • Semaphore 内部有计数值；而 Condition Variable 没有类似的资源。

  • Semaphore 通常不使用普通 Mutex 而利用自身操作 (二进制信号量) 实现；Monitor 通常配合使用 Mutex + 条件变量。

  • Condition Variable 允许线程在临界区内 (持有 Mutex 情况下) 调用 wait()，因为它能在进入等待状态时 ⚠️ 原子性地释放锁，从而避免持锁睡眠造成的 Deadlock，也避免释放锁与进入等待之间发生的 Missed Wakeup。这也是为什么不能用普通的 release() + sleep() 替代 Condition Variable；

    Semaphore 不能在 critical section 中等待 (Deadlock)。

条件变量的操作 (必须持锁者才能执行)：

  • Wait(&lock)：原子地释放 Lock 并进入等待；返回前重新获取 Lock

    Thread A：acquire(mutex) → 检查条件 → xxx 条件不满足 cond_wait(&xxx) → 原子释放 mutex 并进入相应条件的等待队列 → BLOCK → 被 cond_signal 唤醒 → 等待重新获取 mutex → 获得锁后 cond_wait() 返回，继续执行

    ⚠️ 因此有一个非常重要的性质：cond_wait() 返回时，调用线程已经重新持有 Lock。不需要再获取锁，否则对普通的非递归 Mutex，可能造成自己等待自己释放锁的 Deadlock。

  • Signal()：唤醒一个等待线程（如果存在）

    cond_signal(&xxx) 表示如果有线程正在等待 xxx，唤醒其中一个，不保证出队顺序！唤醒谁由实现和调度策略决定。若没有等待线程通常不会留下任何通知或值。

  • Broadcast()：唤醒所有等待线程

    cond_broadcast(&xxx) 唤醒所有等待 xxx 的线程，但它们仍需重新竞争获得一个 Mutex。


Infinite Synchronized Buffer

通过 Monitor 实现无限容量缓冲，那么 Producer 永远不需要等待空位，Consumer 仍然需要等待数据。因此我们只需要一个 Condition Variable，它对应 Semaphore mutex。

Lock buf_lock;    // 保护 Queue 的锁
Condition buf_CV; // 等待 Queue 非空的条件
Queue queue;

Producer(item) {
    acquire(&buf_lock);    // 获取 Lock
    enqueue(&queue, item); // 添加数据
    cond_signal(&buf_CV);  // 通知正在等待数据的 Consumer
    release(&buf_lock);    // 释放 Lock
}
Consumer() {
    acquire(&buf_lock);
    while (is_empty(&queue)) cond_wait(&buf_CV, &buf_lock);
    item = dequeue(&queue);
    release(&buf_lock);
    return item;
}

⚠️ 为什么不用 if (is_empty(&queue))？

它们最关键的区别是执行 signal() 后，被唤醒线程能否立即获得 Monitor 的执行权——检查一次还是每次醒来都再检查一次？而这取决于调度的类型。

  • Hoare Monitor：当一个线程执行 signal()，等待线程立即获得 Monitor，实现需要复杂的执行权交接机制，但可以使用 if/while 逻辑

    if 相当于条件成立就等待一次，醒来后不再检查

  • Mesa Monitor：执行 signal() 的线程不会立即让出 Mutex，被唤醒的线程只是进入 Ready Queue，等待未来被调度 (期间其他线程也可竞争锁)，只能用 while 逻辑

    ⚠️ while **不是 Busy Waiting！**相当于每次条件不满足，线程都会通过 cond_wait() 释放锁并阻塞。每次醒来都重新检查，若条件仍不满足，就继续睡眠

    Mesa Monitor 的 Signal 只表示某个条件可能已经发生变化，不保证被唤醒线程真正运行时条件仍然成立。

    大多数操作系统采用 Mesa-style scheduling


Bounded Buffer with Monitor

这次不使用三个 Semaphore，而是三个 Monitor 来构建有限缓冲同步区

Lock buf_lock = <initially unlocked>;
Condition is_not_full  = <initially empty>;
Condition is_not_empty = <initially empty>;
// 等待队列初始为空，不是指条件值为 false
Producer(item) {
    acquire(&buf_lock);
    while (buffer_full) cond_wait(&is_not_full, &buf_lock);
    enqueue(item);
    cond_signal(&is_not_empty);
    release(&buf_lock);
}
Consumer() {
    acquire(&buf_lock);
    while (buffer_empty) cond_wait(&is_not_empty, &buf_lock);
    item = dequeue();
    cond_signal(&is_not_full);
    release(&buf_lock);
    return item;
}

这比 Semaphore 版本更加直观，条件判断部分直接表达：

P：当 Buffer 满时，等待 Buffer 不再满；C：当空时，等到不再空。

cond_wait() 让线程原子释放 Lock 并进入等待，所以其他线程可以进入临界区，这就避免了使用普通 Mutex 时的 Deadlock。


Mesa Monitor Conclusions

等待某个条件：

acquire(&lock);
while (need_to_wait) cond_wait(&CV, &lock);
// Access or modify shared state
release(&lock);

改变状态并通知其他线程：

acquire(&lock);
// Change shared state
cond_signal(&CV);
release(&lock);

最重要的三个规则：

  1. 检查共享条件时，必须持有对应的 Lock，否则检查条件与进入等待之间可能出现 Race Condition
  2. cond_wait() 会自动释放 Lock，并在返回前重新获取 Lock，可在临界区中等待其他线程改变共享状态
  3. 使用 while 重新检查条件，而不是假设被 Signal 后条件一定成立，Mesa 不保证被唤醒线程立即执行

⚠️ Semaphore 的 V() 和 Condition Variable 的 Signal 并不等价：因为信号量通过计数保存许可，而管程不保存通知。如果 signal() 执行时无等待者，通知就会丢弃；后来的 cond_wait() 不会因为之前执行过 signal() 而返回。因此条件变量必须与一个受 Lock 保护的共享状态条件 while 配合使用。



Readers-Writers Problem

相比上一部分的 Producer-Consumer，读者-写者问题的并发同步逻辑更复杂

对于一个共享数据库，存在两类线程：

  • Reader 读者：只读取数据，不修改
  • Writer 写者：可以读取并修改数据

只用 Mutex 可以解决这个问题……但这种方案有一个明显的缺点——无限制互斥：多个 Reader 都只读，但它们也无法同时访问，时间复用率 (multiplexing) 低。

三条 Correctness Constraints：

① Reader 可以在没有 Active Writer 时访问数据库 (读者间不互斥)

② Writer 只能在没有 Active Reader，也没有 Active Writer 时访问数据库 (写者全局互斥)

③ 同一时间只有一个线程可以修改同步状态变量

四个 state variables 状态变量 (受锁 lock 保护)：

int AR = 0;  // Active Reader
int WR = 0;  // Waiting Reader
int AW = 0;  // Active Writer
int WW = 0;  // Waiting Writer
Lock lock;  // 保护状态变量
Condition ok_to_read - NIL;  // 等待读的队列
Condition ok_to_write = NIL; // 等待写的队列

⚠️ 通常采用 Writer Priority (写者优先) 策略，即当已经有 Writer 正在等待，新 Reader 也必须等待。否则可能发生 Starvation：Reader 源源不断，Writer 被掐死无法执行的情况

因此，Reader 的等待条件不是 AW > 0 而是 AW + WW > 0 ——

Reader() {
    acquire(&lock);
    while ((AW + WW) > 0) {
        WR++;
        cond_wait(&ok_to_read, &lock);
        WR--;
    }
    AR++;
    release(&lock);
    // 读数据库
    access_database(read_only);
    acquire(&lock);
    AR--;
    if (AR == 0 && WW > 0)
        cond_signal(&ok_to_write);

不需要在整个数据库读取期间一直持有 Mutex，这样才能允许多个 Reader 同时读取数据库。但前提是 Reader 只读，Writer 也必须遵守同一套访问协议，否则仍然可能出现数据竞争。

Writer：AW 保证 Writer 之间的互斥，而 WW 用于实现 Writer Priority ——

Writer() {
    acquire(&lock);
    while ((AW + AR) > 0) { // Writer必须独占数据库
        WW++;
        cond_wait(&ok_to_write, &lock);
        WW--;
    }
    AW++;
    release(&lock);
    // 写数据库
    access_database(read_write);
    acquire(&lock);
    AW--;
    if (WW > 0)       // 优先写
        cond_signal(&ok_to_write);
    else if (WR > 0)  // 无等待写者才可以开放读
        cond_broadcast(&ok_to_read);
    release(&lock);
}

  • Readers 可能发生 Starve 吗？因为使用 Writer Priority，假设写者不断到达，就永远无法读——写优先并不等于 Fairness，只是避免 Writer Starve。

  • 如果删除 Reader 退出时的 AR == 0 判断？通常不会直接破坏互斥正确性，因为 Writer 有 while(AR+WR > 0) 保护。但造成不必要的唤醒、锁竞争和上下文切换。

  • 如果把给 write 的 signal() 改成 broadcast() ？正确性通常仍然可以维持，但会产生不必要的唤醒和锁竞争，这类现象称为 Thundering Herd (惊群)。

  • 如果只用一个条件变量 ok 而非分别处理读写？问题是 signal 不保证唤醒哪一种线程，可能导致系统没有线程继续推进，就必须用 broadcast 让所有等待线程重新检查自己的条件。代价是更多不必要的唤醒，不如两个条件变量高效清晰。

  • 为何没有 R/W Race Condition：读者的 AR++ 和写者的 AR==0 是在同把锁保护下执行的，不可能出现 Reader 认为已经获得读资格，同时 Writer 认为没有任何 Reader 正在读的可能。



Concurrency Programming Tips

C 语言

异常或提前返回时容易忘记释放 Lock：

acquire(&lock);
if (error) return -1;
release(&lock);
  • 发生错误时不能直接退出函数，必须确保每条退出路径都释放 Lock。
  • 必须 catch 关键区域 (Mutex 区域) 的所有 exception
acquire(&lock);
try {
// ...
Func()
} catch (…) {
release(&lock);
throw;
}
void Func() {if (exception) throw Exception; // ...}

C++

#include <mutex>
std::mutex global_mutex;
int global_i = 0;
void safe_increment() {
    std::lock_guard<std::mutex> lock(global_mutex);
    // 自动获取 Mutex
    global_i++;
}

当函数结束、离开作用域或发生异常导致栈展开时，lock_guard 的析构函数 destructor 会自动释放 Mutex

所以不需要手动写 global_mutex.unlock();。


Python

import threading
lock = threading.Lock()
with lock:
    shared_value += 1

with 关键词自动获取 lock，离开代码块时自动释放 release()；即使代码块中发生异常，也会执行相应的退出清理操作。


Java

class Account {
    private int balance;
    public synchronized void deposit(int amount) {
        balance += amount;
    }
}

synchronized 方法会获取对应对象的 Monitor Lock，同一个对象上的其他 synchronized 实例方法，需要竞争同一把锁。

Java 还提供 wait();、notify();、notifyAll(); 对应 Monitor wait、signal、broadcast。

⚠️ Java 的 wait() 也会释放对应对象的 Monitor Lock，并在返回前重新获取。

comments powered by Disqus