操作系统（五）：锁、信号量与管程

竞态条件/Race condition：系统缺陷——结果依赖于并发执行或事件的顺序，时间（不确定性，不可重现）

原子操作/Atomic Operation：一次不存在任何中断或失败的执行

要么执行成功，要么根本没有执行，并且不应该发现任何部分执行的状态

临界区/Critical Section：值进程中的一段代码区域，这段区域需要访问共享资源，并且当另一个进程处于相应代码区域时便不会执行

互斥/Mutual Exclusive：当一个进程处于临界区并访问共享资源时，没有其他进程会处于临界区并访问任何相同的资源

死锁/Dead Lock：两个或以上的进程，在互相等待完成特点的任务，而最终没法将自身任务进行下去

饥饿/Starvation：一个可执行的进程，被调度器持续忽略

锁/Lock：

lock.acquire()：在锁释放前一直等待，直到获得锁
lock.release()：解锁并唤醒任何等待中的进程

在临界区执行的属性：

互斥：同一临界区最多存在一个线程
progress：若一个线程想进入临界区，那么它最终会成功
有限等待：若一个线程处于入口区，在其请求被接受之前，其他线程进入临界区的时间是有限的
无忙等待(可选)：若一个进程在等待进入临界区，那它可以在进入前被挂起

1.实现锁的方法

方法一：禁用硬件中断

仅限单处理器
没有中断，没有上下文切换，因此并发不存在
进入临界区：禁用中断
离开临界区：开启中断

缺点：

一旦禁用中断，线程就无法停止，整个系统都要为它停下来。可能会导致其他进程处于饥饿状态
无法限制响应中断所需时间（临界区任意长？）

方法二：基于软件

Peterson算法

使用两个共享项

int turn;//指示该谁进入临界区
bool flag[]; //进程是否准备好进入临界区

进程 P_i 的算法：

do
{
flag[i] = true;
turn = i;
while (flag[i] && trun == i)
CRITICAL SECTION;
flag[i] = false;
REMINDER SECTION;
} while (true);

Bakery算法——N个进程的临界区

进入临界区，进程接收一个数字
得到数字最小的进入临界区
若进程 P_i 和 P_j 得到的数字相同，则比较 i 和 j 的大小，小的进入临界区
编号方案总是按照枚举的增加顺序生成数字

方案三：更高级的抽象

特殊的原子操作指令，通过特殊内存访问电路

原子指令：Test-and-set/测试与置位

从内存中读取值
测试该值是否为一（返回真假）
内存值置为1

bool TestAndSet(bool *target)
{
bool rv = *target;
*target = true;
return(rv);
}

忙等待

class Lock {int value = 0;};

Lock::Acquire()
{
while(TestAndSet(value)
};

Lock::Release()
{
value = 0;
}

无忙等待

class Lock {int value = 0; waitQueue q;};

Lock::Acquire()
{
while(TestAndSet(value)
{
add this TCB to waitQueue q;
schedule();
}
};

Lock::Release()
{
value = 0;
remove a thread t from q;
wakeup(t);
}

原子指令：Exchange/交换

交换内存中的两个值

void exchange(bool* a, bool* b)
{
bool temp = *a;
*a = *b;
*b = temp;
}

共享数据：int lock = 0;

int key;

do
{
key = 1;

while (key == 1)
Exchange(lock, key);
CRITICAL SECTION;

lock = 0;
REMINDER SECTION;
}

基于原子操作的机器指令

优点：

适用于单处理器或则共享多处理器的多处理的任意数量的进程
简单，并易证明
可以用于支持多临界区

缺点：

忙等待消耗处理器时间
当进程离开临界区并且多个进程在等待，可能会导致饥饿
死锁（实时系统中），例如：一个低优先级的进程拥有临界区并且一个高优先级进程也需求，那么高优先级的进程会获得处理器并等待临界区

信号量

一个整型（sem），两个原子操作：

P()：sem减一。若 sem < 0, 等待；否则继续
V()：sem加一。若sem <= 0, 唤醒一个等待的进程

整数（sem）是被保护的变量。只能通过 P() 和 V() 访问，必须是原子操作

P() 能够阻塞；V()不会阻塞

假设信号量是公平的（FIFO队列经常被使用）

两种类型的信号量：

二进制：0 or 1
一般/计数信号量：可以取任意非负值

可以用在两个方面：

互斥
条件同步（调度约束——一个进程等待另一个进程的发生）

1.信号量的使用

二进制信号量实现互斥

mutex = new semaphore(1);
mutex->P();
CRITICAL SECTION;
mutex->V();

二进制信号量实现调度约束

2.有界缓冲的生产者——消费者问题

一个线程等待另一个线程处理事情。比如生产东西或消费东西，互斥（锁机制）是不够的。

一个或者多个生产者产出数据，并将数据放在一个缓冲区
单个消费者每次从缓冲区取出数据
在任何一个时间只有一个生产者或者消费者可以访问缓冲区

正确性要求：

互斥：任何时间只能由一个线程操作缓冲区
调度/同步约束： · buffer为空：消费者必须等待生产者 · buffer满了：生产者必须等待消费者

每个约束用一个信号量：

二进制信号量互斥
一般信号量 fullbuffer
一般信号量 emptybuffer

class BoundedBuffer
{
mutex = new samephore(1);
fullBuffer = new samephore(0);
emptyBuffer = new samephore(n);
};

BoundedBuffer::Deposit()
{
emptyBuffer->P();
mutex->P();
ADD C TO THE BUFFER;
mutex->V();
fullBuffer->V();
}

BoundedBuffer::Remove()
{
fullBuffer->P();
mutex->P();
REMOVE C TO THE BUFFER;
mutex->V();
emptyBuffer->V();
}

3.信号量的实现

class semaphore
{
int sem;
waitQueue q;
};

semaphore::P()
{
sem--;
if(sem < 0)
{
Add the thread t to q;
block(t);
}
}

semaphore::V()
{
sem++;
if(sem >= 0)
{
remove a thread t to q;
weakup(t);
}
}

信号量不能解决死锁问题

例子：交替打印零与奇偶数（010203040506）

#include <semaphore.h>

class ZeroEvenOdd {
private:
int n;
bool oddPrint;
sem_t zeroDone;
sem_t oddDone;
sem_t evenDone;

public:
ZeroEvenOdd(int n) {
this->n = n;
sem_init(&zeroDone,0,1);
sem_init(&oddDone,0,0);
sem_init(&evenDone,0,0);
oddPrint = true;
}

// printNumber(x) outputs "x", where x is an integer.
void zero(function<void(int)> printNumber) {
for(int i=0; i<n; i++)
{
sem_wait(&zeroDone);
printNumber(0);
if(oddPrint)
{
oddPrint = false;
sem_post(&oddDone);
}
else
{
oddPrint = true;
sem_post(&evenDone);
}
}

}

void even(function<void(int)> printNumber) {
for(int i=1; 2*i <= n; i++)
{
sem_wait(&evenDone);
printNumber(2*i);
sem_post(&zeroDone);
}

}

void odd(function<void(int)> printNumber) {
for(int i=0; 2*i+1 <= n; i++)
{
sem_wait(&oddDone);
printNumber(2*i+1);
sem_post(&zeroDone);
}
}
};

管程

目的：分离互斥和条件同步的关注

定义：

一个锁：指定临界区
0个或多个条件变量：等待/通知信号量用于管程并发访问共享数据

一个锁——Lock：

lock.acquire()：在锁释放前一直等待，直到获得锁
lock.release()：解锁并唤醒任何等待中的进程

条件变量/Conditional Variables：

允许等待状态进入临界区允许处于睡眠状态的线程进入临界区某个原子时刻释放锁进入睡眠
wait()：释放锁，随眠（重新获得锁后返回）
signal()：operation or boardcast() operation 唤醒等待者或所有等待者

class condition
{
int numWaiting = 0;
waitQueue q = 0;
};

condition::wait(lock)
{
numWaiting++;
ADD THIS THREAD t TO q;
release(lock);
schedule();
require(lock);
}

condition::signal()
{
if (numWaiting > 0)
{
REMOVE A THREAD t FROM q;
weakup(t);
numWaiting--;
}
}

生产者——消费者问题

class BoundedBuffer
{
Lock lock;
int count = 0;
Condition notFull, notEmpty;
};

BoundedBuffer::Deposit()
{
lock->Acquire();
while (count == n)
notFull.wait(&lock);
ADD C TO THE BUFFER;
count++;
notEmpty.signal();
lock->Release();
}

BoundedBuffer::Remove()
{
lock->Acquire();
while (count == 0)
notEmpty.wait();
REMOVE C FROM THE BUFFER;
count--;
notFull.signal();
lock->Release();
}

Hansen-style：

x.signal() 之后先运行完，再运行被唤醒的线程（不能保证唤醒的线程抢到锁）

Hoare-style：

x.signal() 之后立马执行被唤醒线程

例子：交替打印零与奇偶数（010203040506）

#include<iostream>
#include<thread>
#include<mutex>
#include<condition_variable>

using namespace std;

condition_variable cond;
mutex print_mutex;
bool zero = true;
bool odd = true;

//thread::id tid = this_thread::get_id();

void printZero()
{
for (int i = 0; i < 10; i++)
{
unique_lock<mutex> lock(print_mutex);
cond.wait(lock, [&]() {return zero; });
cout << 0;
zero = false;
cond.notify_all();
}
}

void printOdd()
{
for (int i = 1; i <= 10; i+=2)
{
unique_lock<mutex> lock(print_mutex);
cond.wait(lock, [&]() {return ((!zero) && odd); });
cout << i;
odd = false;
zero = true;
cond.notify_all();
}
}

void printEven()
{
for (int i = 2; i <= 10; i+=2)
{
unique_lock<mutex> lock(print_mutex);
cond.wait(lock, [&]() {return ((!zero) && (!odd)); });
cout << i;
odd = true;
zero = true;
cond.notify_all();
}
}

int main()
{
thread t1(printZero);
thread t2(printOdd);
thread t3(printEven);
t1.join();
t2.join();
t3.join();
}

经典同步问题

1.读者——写者问题

动机：共享数据访问（并发进程的数据集共享）

约束：

允许同一时间有多个读者，但任何时候只有一个写者
没有写者时，读者才能访问数据
没有写者和读者时，写者才能访问数据
任何时候只有一个线程可以操作共享变量

有读者优先和写者优先两种情况

读者优先

共享数据：

数据集
信号量 CountMutex 初始化为 1
信号量WriteMutex初始化为1
整数Rcount初始化为0

//writer
sem_wait(WriteMutex);
WRITE DATA;
sem_post(WriteMutex);

//reader
sem_wait(CountMutex) //保护Rcount，开始
if(Rcount == 0)
sem_wait(WriteMutex)
Rcount++;
sem_post(CountMutex) //保护Rcount，结束

READ DATA;

sem_wait(CountMutex) //保护Rcount，开始
Rcount--;
if(Rcount == 0)
sem_post(WriteMutex)
sem_post(CountMutex) //保护Rcount，结束

2.哲学家就餐

信号量

管 程

经典同步问题

其 他

管程

其他