熟悉完操作系统的第九篇章，开始学习第十篇章，关于操作系统的信号量&管程。

0猜你喜欢

操作系统系列文章

操作系统学习笔记 1概述

操作系统学习笔记 2操作系统介绍

操作系统学习笔记 3内存管理

操作系统学习笔记 4非连续内存分配

操作系统学习笔记 5虚拟内存

操作系统学习笔记 6页面置换算法

操作系统学习笔记 7进程管理

操作系统学习笔记 8CPU调度

操作系统学习笔记 9同步&互斥

操作系统学习笔记 10信号量&管程

操作系统学习笔记 11死锁

操作系统学习笔记 12进程间通信

操作系统学习笔记 13文件系统

10信号量&管程

1. 背景
2. 信号量
3. 信号量使用
4. 信号量实现
5. 管程
6. 经典同步问题

10.1背景

• 并发问题：竞争条件（竞态条件）

  多程序并发存在大的问题

• 同步

  多线程共享公共数据的协调执行

  包括互斥与条件同步

  互斥：在同一时间只有一个线程可以执行临界区

• 确保同步正确很难

  需要高层次的编程抽象（锁）

  从底层硬件支持编译

10.2信号量

如果仅仅保证互斥，那么针对某些特殊情况依然无法解决，比如同步机制。临界区有多个线程来执行，进入临界区做读操作，没必要限制一个线程或者进程执行。

抽象数据类型

• 一个整形（sem），两个原子操作
• P():sem减1，如果sem<0，那么等待否则继续
• V():sem加1，如果sem<=0，那么唤醒一个等待的P

信号量类似铁路

• 初始化2个资源控制信号灯

1）初始化

2）两个列车分别对应两个资源控制，第3个列车在信号量处等待

3）直到其中一辆列车走后，这个时候V操作，会唤醒等待在信号量的第3个列车

4）列车3也进入，两个列车分别对应两个资源控制

补充：信号量的提出者

信号量的属性

• 信号量是整数（int）
• 信号量是被保护的变量
• 初始化完成后，唯一改变一个信号量的值的办法是通过P()和V()
• 操作必须是原子
• P()能够阻塞，V()不会阻塞

我们假定信号量是“公平的”

• 没有线程被阻塞在P()仍然阻塞。如果V()被无限频繁调用（在同一个信号量）
• 在实践中，FIFO经常被使用

两种类型信号量

• 二进制信号量：可以是0或1
• 一般/计数信号量：可以取任何非负值
• 两者相互表现（给定一个可以实现另一个）

信号量可以用在2个方面

• 互斥
• 条件同步（调度约束，一个线程等待另一个线程的事情发生）

10.3信号量的使用

P操作会让进程阻塞或者挂起，V操作会唤醒进程。

10.3.1用二进制信号量实现的互斥

1mutex = new Semaphore(1);
2
3mutex->P();
4//...
5Critical Section;
6//...
7mutex->V();

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

1condition = new Sempaphore(0);

Thread A

1...
2condition->P();    
3...

Thread B

1...
2condition->V();
3...

P处等待，V发出信号

10.3.3条件同步

一个线程等待另一个线程处理事情

• 比如生产东西或者消费东西
• 互斥（锁机制）是不够的

生产者-消费者问题

存在一块缓冲区，有一个生产者往这里面写数据，另外有一个消费者会在这里读取数据。双方需要用到同步或者互斥两种机制的操作过程。

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

正确性要求

• 在任何一个时间只有一个线程操作缓冲区（互斥）
• 当缓冲区为空，消费者必须等待生产者，可以多个生产者（调度/同步约束）
• 当缓冲区满，生产者必须等待消费者，可以多个消费者（调度/同步约束）

每个约束用一个单独的信号量

• 二进制信号量互斥
• 一般信号量（fullBuffers）
• 一般信号量（emptyBuffers）

同步互斥两种机制操作

1）初始三个变量

1class BoundedBuffer{
2    mutex = new Semaphore(1);
3    fullBuffers = new Semaphore(0);
4    emptyBuffers = new Semaphore(n);
5};

2）生产者线程

1BoundedBuffer::Deposit(c){
2    emptyBuffers->P();
3    mutex->P();
4    Add to the bufer
5    mutex->V();
6    fullBuffers->V();
7}

3）消费者线程

1BoundedBuffer::Remove(c){
2    fullBuffers->P();
3    mutex->P();
4    Remove c frome bufer
5    mutex->V();
6    emptyBuffers->V();
7}

10.4信号量的实现

10.4.1使用硬件原语

• 禁用中断
• 源自指令（test-and-set）

1class Semaphore{
2    int sem;
3    WaitQueue q;
4};

10.4.2类似锁

比如使用禁用中断，实际上q用一个负数到0的队列来记录，排队进入临界区的进程数量sem（负数）。

P操作

1Semaphore::P()
2{
3    sem--;
4    if(sem < 0)
5    {
6        Add this thread t to q;
7        block(t);
8    }
9}

V操作

1Semaphore::V(){
2    sem++;
3    if(sem <= 0)
4    {
5        Remove a thread t from q;
6        wakeup(t);
7    }
8}

10.4.3信号量的优缺点

1.信号量的双用途

• 互斥和条件同步
• 但等待条件是独立的互斥

2.读/开发代码比较困难

• 程序员必须非常精通信号量

3.容易出错

• 使用的信号量已经被另一个线程占用
• 忘记释放信号量

4.不能够处理死锁问题

10.5管程

10.5.1概念

目的

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

管程

• 一个锁

  指临界区

• 0或者多个条件变量

  等待/通知信号量用于管理并发访问共享数据

一般方法

• 收集在对象/模块中的相关共享数据
• 定义方法来访问共享数据

如下图10-1所示。首先，进入管程就需要一个队列，entry queue。进入管程是互斥的，需要获取lock，取不到lock就在队列中等待。

进入管程的空间之后，线程就可以执行管程维护的一系列的函数。这些函数可能对共享变量进行操作，针对某一项共享变量得不到满足就会等待。把自身挂起，把lock释放掉，这样可以让在lock等待的其他线程进入临界区。上面的x和y分别有两个队列，满足一个条件会唤醒对应的队列的线程。

图10-1 管程原理图

Lock

• Lock::Acquire()-等待直到锁可用，然后抢占锁
• Lock::Release()-释放锁，唤醒等待者（如果有）

Condition Variable

• 允许等待状态进入临界区

  允许处于等待（睡眠）的线程进入临界区

  某个时候原子释放锁进入睡眠

• Wait() operation

  释放锁，睡眠，重新获得锁返回

• Signal() operation（for broadcast() operation）

  唤醒等待者（或者所有等待者），如果有

10.5.2管程实现

需要维持每个条件队列

线程等待的条件等待signal()

1class Condition{
2    int numWaiting = 0;
3    WaitQueue q;
4};

Wait

1Condition::Wait(lock)
2{
3    numWaiting++;
4    Add this Thread t to q;
5    release(lock);//这个lock是管程的lock
6    schedule();//need mutex，阻塞的
7    require(lock);
8}

Signal

1Condition::Signal(){
2    if(numWaiting > 0)
3    {
4        Remove a thread t from q;
5        Wakeup(t);//need mutex
6        numWaiting--;
7    }
8}

信号量中的sem和管程中的numWaiting区别

sem代表信号量个数，numWaiting当前等待线程个数。

信号量中的P和V是一定会做加操作和减操作的，numWaiting在Wait会做家操作，但是numWaiting不一定需要做减操作。

10.5.3生产者-消费者模型（管程方式）

这里的buffer和count都是互斥的，所以需要在外部加锁和释放锁。

1class BoundedBuffer{
2    Lock lock;
3    int count = 0;
4    Condition notFull,notEmpty;
5};

Deposit

 1BoundedBuffer::Deposit(c)
 2{
 3    lock->Acquire();
 4    while(count == n)//判断buffer是否满了
 5    {
 6        //这里会释放锁，如果不释放remove线程就获取不了锁，就唤醒不了这个线程
 7        notFull.Wait(&lock);
 8    }
 9    Add c to the Buffer;
10    count++;
11    notEmpty.Signal();
12    lock->Release();
13}

Remove

 1BoundedBuffer::Remove(c)
 2{
 3    lock->Acquire();
 4    while(count == 0)//判断buffer是否空了
 5    {
 6        //这里会释放锁，如果不释放Deposit线程就获取不了锁，就唤醒不了这个线程
 7        notEmpty.Wait(&lock);
 8    }
 9    Remove c frome Buffer;
10    count--;
11    notFull.Signal();
12    lock->Release();
13}

管程中其中有一个Signal唤醒，这个时候唤醒的当前线程和被唤醒的线程都会执行操作。那么哪个线程优先执行，也是一个问题。

如果用这两种方式，来处理生产-消费者问题，那么表现有所不同。

hansen是唤醒后才交接所以会唤醒多个，Hoare是交接之后才唤醒，相当于内定了

10.5.4总结

信号量和管程，用来解决更广泛式的同步问题，比之前的lock更近一步，更容易编写同步式问题的应用。

开发/调试并行程序很难

• 非确定性的交叉指令

同步结构

• 锁：互斥
• 条件变量：有条件的同步
• 其他原语：信号量

怎么有效的使用这些结构

• 制定并遵循严格的程序设计风格/策略

10.6经典同步问题

10.6.1读者-写者问题

动机

• 共享数据的访问

两种类型使用者

• 读者：不需要修改数据
• 写者：读取和修改数据

问题的约束

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

多个并发进程的数据集共享

• 读者-只读数据集；它们不执行任何更新
• 写者-可以读取和写入

共享数据

• 数据集
• 信号量CountMutex初始化为1，保证不会多个读者进行Rcount操作
• 信号量WriteMutex初始化为1
• 整数Rcount初始化为0

10.6.1.1读者优先（信号量实现）

基于读者优先策略的方法，只要有一个读者处于活动状态，后来的读者都会被接纳。如果读者源源不断的出现的话，那么写者就始终处于阻塞状态。

Writer

1sem_wait(WriteMutex);//这个是之前信号量的P
2write;
3sem_post(WriteMutex);//这个是之前信号量的V

Reader

 1sem_wait(CountMutex);//避免多个读线程同时操作Rcount
 2if(Rcount == 0)
 3{
 4    sem_wait(WriteMutex);
 5}
 6Rcount++;
 7sem_post(CountMutex);
 8read;
 9sem_wait(CountMutex);
10Rcount--;
11if(Rcount == 0)
12{
13    sem_post(WriteMutex);
14}
15sem_post(CountMutex);

10.6.1.2写者优先（信号量实现）

基于写者优先策略的方法：一旦写者就绪，那么写者会尽可能快地执行写操作。如果写者源源不断地出现的话，那么读者始终处于阻塞状态。（如何实现？）

跟上面的读者优先类似

Writer

 1sem_wait(WriteMutex);//避免多个写线程同时操作Rcount
 2if(Rcount == 0)
 3{
 4    sem_wait(CountMutex);
 5}
 6Rcount++;
 7sem_post(WriteMutex);
 8write;
 9sem_wait(WriteMutex);
10Rcount--;
11if(Rcount == 0)
12{
13    sem_post(CountMutex);
14}
15sem_post(WriteMutex);

Reader

1sem_wait(CountMutex);//这个是之前信号量的P
2read;
3sem_post(CountMutex);//这个是之前信号量的V

10.6.1.3写者优先（管程实现）

用四个变量表示读者写者的状态个数，AR代表正在活跃的读者，AW代表正在活跃的写者，WR代表正在等待的读者，WW代表正在等待的写者。okToRead代表当前可以执行读操作，okToWrite代表当前可以执行写操作。

Write

 1public Database::Write()
 2{
 3    //Wait until no writers
 4    lock.Acquire();
 5    while((AW+AR) > 0)
 6    {
 7        WW++;
 8        okToWrite.wait(&lock);
 9        WW--;
10    }
11    AW++;
12    lock.Release();
13    write database;
14    lock.Acquire();
15    AW--;
16    if(WW > 0)//这个判断，是写者优先
17    {
18        okToWrite.signal(&lock);
19    }
20    else if(WR > 0)
21    {
22        okToRead.broadcast(&lock);//可以同时执行多个读操作
23    }
24    lock.Release();
25}

Read

 1public Database::Read()
 2{
 3    //Wait until no writers
 4    lock.Acquire();
 5    //如果存在写操作,那么读等待，直到可以读操作
 6    //这里体现了写者优先，WW主要是为了判断是否有写者等待，如果有的话，继续写者，WW代表了写者优先
 7    while((AW+WW) > 0)
 8    {
 9        WR++;
10        okToRead.wait(&lock);
11        WR--;
12    }
13    AR++;
14    lock.Release();
15    read database;
16    lock.Acquire();
17    AR--;
18    if((AR == 0 && WW > 0))
19    {
20        okToWrite.signal(&lock);
21    }
22    lock.Release();
23}

10.6.2哲学家就餐问题

10.6.2.1方案一

 1#define N 5					//哲学家个数
 2void philosopher(int i)		//哲学家编号0-4
 3{
 4    while(true)
 5    {
 6        think();			//哲学家在思考
 7        take_fork(i);		//去拿左边的叉子
 8        take_fork((i+1)%N);	//去拿右边的叉子
 9        eat();				//吃面条
10        put_fork(i);		//放下左边的叉子
11        put_fork((i+1)%N);	//放下右边的叉子
12    }
13}

结果

• 不正确，可能导致死锁

10.6.2.2方案二

针对方案一进行改进

 1#define N 5						//哲学家个数
 2void philosopher(int i)			//哲学家编号0-4
 3{
 4    while(true)
 5    {
 6        think();				//哲学家在思考
 7        if(take_fork(i))		//去拿左边的叉子
 8        {
 9            take_fork((i+1)%N);	//去拿右边的叉子       
10            eat();				//吃面条
11            put_fork(i);		//放下左边的叉子
12            put_fork((i+1)%N);	//放下右边的叉子
13            break;
14        }else
15        {
16            put_fork(i);		//放下左边的叉子
17            waitsome_time();
18        }
19    }
20}

结果

• 对拿叉子的过程进行了改进，但仍不正确

10.6.2.3方案三

针对方案二继续改进

 1#define N 5						//哲学家个数
 2void philosopher(int i)			//哲学家编号0-4
 3{
 4    while(true)
 5    {
 6        think();				//哲学家在思考
 7        if(take_fork(i))		//去拿左边的叉子
 8        {
 9            take_fork((i+1)%N);	//去拿右边的叉子       
10            eat();				//吃面条
11            put_fork(i);		//放下左边的叉子
12            put_fork((i+1)%N);	//放下右边的叉子
13            break;
14        }else
15        {
16            put_fork(i);		//放下左边的叉子
17            wait_random_time();	//等待随机长时间
18        }
19    }
20}

结果

• 等待时间随机变化。可行，但非万全之策。

10.6.2.4方案四

通过互斥访问

 1semaphore mutex				//互斥信号量，初始值为1
 2void philosopher(int i)		//哲学家编号0-4
 3{
 4    while(true)
 5    {
 6        think();			//哲学家在思考
 7        P(mutex);			//进入临界区
 8        take_fork(i);		//去拿左边的叉子
 9        take_fork((i+1)%N);	//去拿右边的叉子
10        eat();				//吃面条
11        put_fork(i);		//放下左边的叉子
12        put_fork((i+1)%N);	//放下右边的叉子
13        V(mutex);			//退出临界区
14    }
15}

• 结果

  互斥访问。正确，但每次只允许一人进餐，浪费资源。

• 方案四的缺点

  把就餐（而不是叉子）看成是必须互斥访问的临界资源，因此造成（叉子）资源浪费。

  从理论上说，如果有五把叉子，应允许两个不相邻的哲学家同时进餐。

10.6.2.5思路（1）哲学家自己怎么来解决这个问题？

指导原则：要么不拿，要不就拿两把叉子。

• S1 思考中
• S2 进入饥饿状态
• S3 如果左邻居或右邻居正在进餐，等待；否则转到S4
• S4 拿起两把叉子
• S5 吃面条。。
• S6 放下左边的叉子
• S7 放下右边的叉子
• S8 新的循环又开始了，转S1

10.6.2.6思考（2）计算机程序怎么来解决这个问题？

指导原则：不能浪费CPU时间：进程间相互通信

• S1 思考中
• S2 进入饥饿状态
• S3 如果左邻居或右邻居正在进餐，进程进入阻塞态；否则转到S4
• S4 拿起两把叉子
• S5 吃面条。。
• S6 放下左边的叉子，看着左邻居现在能否进餐（饥饿状态，两把叉子都在），若能则唤醒左邻居
• S7 放下右边的叉子，看着右邻居现在能否进餐（饥饿状态，两把叉子都在），若能则唤醒右邻居
• S8 新的循环又开始了，转S1

10.6.2.7思考（3）怎么样来编写程序？

1. 必须有数据结构，来描述每个哲学家的当前状态
2. 该状态是一个临界资源，各个哲学家对它的访问应该互斥地进行（进程互斥）
3. 一个哲学家吃饱饭后，可能要唤醒它的左邻右舍，两者之间存在这同步关系（进程同步）

10.6.2.8实例

1. 必须有一个数据结构，来描述每个哲学家的当前状态

   1#define N 			5			//哲学家个数
   2#define LEFT 		(i-1+N)%N	//第i个哲学家的左邻居
   3#define RIGHT 		(i+1)%N		//第i个哲学家的右邻居
   4#define THINKING 	0			//思考状态
   5#define HUNGRY 		1			//饥饿状态
   6#define EATING 		2			//进餐状态
   7int state[N];					//记录每个人的状态
   

2. 该状态是一个临界资源，对他的访问应该互斥地进行

   1semaphore mutex;		//互斥信号量，初值1
   

3. 一个哲学家吃饱后，可能要唤醒邻居，存在同步关系

   1semaphore S[N];			//同步信号量，初值0
   

4. 函数philosopher定义

    1void philosopher(int i)
    2{
    3    while(true)
    4    {
    5        think();		//s1,思考中
    6        take_forks(i);	//s2-s4，拿到两把叉子
    7        eat();			//s5，吃面条中
    8        put_forks(i);	//s6-s7，把两把叉子放回去
    9    }
   10}
   

5. 函数take_forks的定义

   1//功能：要么拿到两把叉子，要么被阻塞起来
   2void take_forks(int i)
   3{
   4    P(mutex);						//进入临界区
   5    state[i] = HUNGRY;				//设置状态为饿了
   6    test_take_left_right_forks(i);	//尝试去拿叉子
   7    V(mutex);						//退出临界区
   8    P(s[i]);						//没有叉子阻塞
   9}
   

6. 函数test_take_left_right_forks的定义

    1void test_take_left_right_forks(int i)
    2{
    3    if(state[i] == HUNGRY && 
    4      state[LEFT] != EATING &&
    5      state[RIGHT] != EATING)
    6    {
    7        state[i] = EATING;		//两把叉子到手
    8        V(s[i]);				//通知第i人可以吃饭了,通知自身吃饭
    9    }
   10}
   

7. 函数put_forks的定义

   功能：把两个叉子放回原位，并在需要的时候去唤醒左邻右舍

   1void put_forks(int i)
   2{
   3    P(mutex);							//进入临界区
   4    state[i] = THINKING;				//交出两把叉子
   5    test_take_left_right_forks(LEFT);	//看左邻居是否进餐
   6    test_take_left_right_forks(RIGHT);	//看右邻居是否进餐
   7    V(mutex);							//退出临界区
   8}