lab7 — 同步互斥

发表于 分类于
本文字数： 6.3k 阅读时长 ≈ 6 分钟

前置知识

临界区、信号量、条件变量、管程。

改动点

相比于 lab6 源代码，lab7 主要做了如下改动：

  • sched.[ch] 增加定时器机制，用以实现 do_sleep() 功能
  • wait.[ch] 实现基于链表形式的等待队列
  • sem.[ch] 实现信号量机制
  • monitor.[ch] 实现基于管程的条件变量机制

练习零

该练习用于了解定时器机制的实现流程。

为实现此机制，首先需要使用相关数据结构以表征定时任务：

1
2
3
4
5
6
7
8
9
// 表征定时任务
typedef struct {
    unsigned int expires;       // 任务的到期时间 (实际实现中，各定时任务按到期时间，由近至远串接至 timer_list。为简化定时更新操作，该字段含义变更为：当前定时任务距离前一个定时任务的时间间隔)
    struct proc_struct *proc;   // 所涉的进程
    list_entry_t timer_link;    // 链接至 timer_list
} timer_t;

// 定时任务链表首部，用以串接各定时任务
static list_entry_t timer_list;

各字段含义已经明确，add_timer()/del_timer() 的实现就比较简单，唯一需要注意的是：正确更新相关 timer_t 的 expires 字段。

接下来，便是如何动态感知定时任务是否到期，从而唤醒相关进程？

对于操作系统而言，它借助于时钟中断以感知时间变化，因此当时钟中断发生时，它会调用特定函数 (ucore 中的 run_time_list() ) 以动态感知定时任务，并且可能执行相关唤醒操作，最后，动态更新当前进程的时间片信息，从而判断是否需要切换调度。

简单提一下，用户进程如何使用定时器？

简单流程：用户调用 sleep(time) –> 中断触发 sys_sleep() –> 间接调用 do_sleep(time) –> add_timer()。

练习一

该练习用于了解信号量机制的实现流程。

为实现互斥方法，总共存在三种方案：基于软件设计、基于硬件中断、基于硬件提供的原子操作。

在 ucore 中，使用最简单的 “基于硬件中断” 实现信号量机制。

先行给出信号量机制实现的形式化描述 (信号量实现基本与此相同)：

对于信号量而言，它使用如下数据结构进行表征：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 信号量
typedef struct {
    int value; 					// 共享资源的数目
    wait_queue_t wait_queue; 	// 欲共享该资源的等待队列
} semaphore_t;

// 等待队列
typedef struct {
    list_entry_t wait_head; 	// 链首
} wait_queue_t;

// 等待队列中的元素
typedef struct {
    struct proc_struct *proc; 	// 所涉的进程
    uint32_t wakeup_flags;  	// 该进程放置于等待队列中的原因 (例如：信号量、定时)
    wait_queue_t *wait_queue;	// 所在的等待队列
    list_entry_t wait_link;		// 联结至 wait_head
} wait_t;

信号量对应的 P()/V() 操作，具体见源代码：

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
static __noinline void __up(semaphore_t *sem, uint32_t wait_state) {
    bool intr_flag;
    // local_intr_save 表示关中断，local_intr_restore 表示开中断。
    // 中断关闭，保证只有当前进程可以运行，从而保证这部分操作的原子性。
    local_intr_save(intr_flag);
    {
        wait_t *wait;
        // 如果等待队列内部不存在等待进程，则资源数量加一，否则选择一个等待进程调度即可。
        if ((wait = wait_queue_first(&(sem->wait_queue))) == NULL) {
            sem->value ++;
        }
        else {
            assert(wait->proc->wait_state == wait_state);
            wakeup_wait(&(sem->wait_queue), wait, wait_state, 1);
        }
    }
    local_intr_restore(intr_flag);
}

static __noinline uint32_t __down(semaphore_t *sem, uint32_t wait_state) {
    bool intr_flag;
    local_intr_save(intr_flag);
    // 如果仍存在资源可访问，则直接访问即可。
    if (sem->value > 0) {
        sem->value --;
        local_intr_restore(intr_flag);
        return 0;
    }
    // 表明已不存在资源可访问，则将当前进程放置于等待队列内部。
    wait_t __wait, *wait = &__wait;
    wait_current_set(&(sem->wait_queue), wait, wait_state);
    local_intr_restore(intr_flag);

    // 调度其他进程运行。
    schedule();

    // 当前进程再次运行，则表明其已获取资源。
	...
    return 0;
}

虽然形式化与具体实现并不相同，但是两者是等价的。

两者的主要差别在于：在形式化中，sem 表征正在使用、预定使用共享资源后，共享资源的剩余数目；在具体实现中，value 表征正在使用共享资源后，共享资源的剩余数目 (此时，共享资源的剩余数目只可能大于等于零，而不可能小于零)。

如何为用户态进程提供信号量机制？

肯定需要使用系统调用。

当用户态进程使用创建信号量的系统调用时，OS 内部创建 semaphore_t 结构体，但是返回给用户态进程的标识则是另一个 (可能的情况，在 PCB 内部维护信号量数组，返回的是信号量在此数组的下标)。

当用户态进程使用 up/down 的系统调用时，OS 从当前进程的 PCB 找到相应的 semaphore_t 结构体，然后执行相关操作即可。

练习二

该练习用于了解基于管程的条件变量机制的实现流程。

信号量可以实现互斥访问，也可实现进程间同步。因为基于信号量的进程间同步比较麻烦，而且容易出错误，因此出现了 条件变量 (注意：条件变量仍是以信号量为基础)。

信号量和条件变量均是偏底层、用于互斥访问和进程间同步的方法，使用起来也是比较麻烦的。为进一步简化使用，更高层级的抽象形式 管程 便出现了。

简而言之，管程 是一个黑盒，程序员往里扔的函数，它可确保在同一时刻，只有一个函数在执行 (亦因如此，确保其内部共享数据的互斥访问)。

管程的实现方式分为多种，其主要区别在于：假定线程 A 因等待某条件而处于等待队列，线程 B 满足该条件后，线程 B 具有哪种行为？

  • Mesa Semantics：线程 B 执行 cond_signal ，因而线程 A 从等待队列移除，并放置于就绪队列，然后线程 B 继续执行。
  • Hanson Semantics：线程 B 执行完成并退出的同时，执行 cond_signal，因而线程 A 从等待队列移除，并放置于就绪队列。
  • Hoare Semantics：线程 B 执行 cond_signal，因而线程 A 从等待队列移除，并放置于就绪队列，然后立即阻塞线程 B，并等待线程 A 被执行。

对于这三种方式，实际实现基于前两种 (因为可以减少一次上下文切换)，书本介绍基于后一种。

在 ucore 中，管程即是基于后一种实现的，由于需要保证 cond_signal 执行的同时，阻塞当前线程，因此其实现有些麻烦。

另外，管程属于更高层级的抽象形式，往往适用于 Java 等高级语言实现，这里实现一种基于 C 语言的、简化版的管程。

管程基于信号量和条件变量而实现，信号量的实现前面已经谈及，在此给出条件变量实现的形式化描述 (条件变量实现基本与此大不相同)：

根据此图，简单说明 Mesa Semantics 实现所存在的小缺陷。

线程 B 执行 cond_signal 后，它只是将线程 A 放置于就绪队列，此时即便调度至线程 A，由于其需要再次获取 lock，而此时 lock 仍归线程 B 所有，因此线程 A 仍无法运行，只能等待线程 B 执行完成并释放 lock，然后才能执行线程 A (如果调度线程 A 前调度了其他进程进入管程，有可能使得线程 A 的条件再次无法满足。因此，当线程 A 执行时，其需要循环判断条件是否满足)。

对于管程而言，它使用如下数据结构进行表征：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 条件变量
typedef struct condvar{
    semaphore_t sem;        // 借助于信号量，间接使用等待队列 (最初设置信号量为 0)。
    int count;              // 在此条件变量上等待的进程数量
    monitor_t * owner;      // 所属的管程
} condvar_t;

// 管程
typedef struct monitor{
    semaphore_t mutex;      // 互斥访问管程内部的数据结构，初始化为 1。
    // next_count 指代由于发出 cond_signal 而睡眠的进程数量，next 只是用于构建一个发出 cond_signal 而睡眠的进程的等待队列。此二者是实现 Hoare Semantics 的关键。
    semaphore_t next;      
    int next_count;         
    condvar_t *cv;          // 管程内部的条件变量列表
} monitor_t;

首先给出初始化过程：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
void monitor_init (monitor_t * mtp, size_t num_cv) {
    int i;
    // 初始值 next_count 自然为 0。
    mtp->next_count = 0;
    mtp->cv = NULL;
    // mutex 置 1，表示当前管程尚无进程访问，next 只是用于构建等待队列，因此置 0。
    sem_init(&(mtp->mutex), 1); 
    sem_init(&(mtp->next), 0);
    // 为条件变量分配空间，并依据字段含义进行初始化。
    mtp->cv =(condvar_t *) kmalloc(sizeof(condvar_t)*num_cv);
    for(i=0; i<num_cv; i++){
        mtp->cv[i].count=0;
        sem_init(&(mtp->cv[i].sem),0);
        mtp->cv[i].owner=mtp;
    }
}

接下来是条件变量的 cond_wait()/cond_signal()，具体见源代码：

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
void cond_signal (condvar_t *cvp) {
    // 如果当前条件变量上并不存在等待进程，则直接返回，否则执行如下操作。
   if (cvp->count > 0) {
       monitor_t* mon = cvp->owner;
       mon->next_count++;
       // 唤醒当前条件变量所示等待队列上的一个。
       up(&(cvp->sem));
       // 将当前进程添加至 next 所示的等待队列上、使 next_count 加一，并选择调度其他进程。
       down(&(mon->next));
       // 再次执行时，当前进程唤醒，因而需要使next_count 减一。 
       mon->next_count--;
   }
}

void cond_wait (condvar_t *cvp) {
    // 当前条件变量上等待进程数加一。
    cvp->count++;
    monitor_t* mon = cvp->owner;
    // 如果 next 所示的等待队列上存在进程，则唤醒其中的一个，否则唤醒管程等待队列上的一个。
    if (mon->next_count > 0) {
        up(&(mon->next));
    } else {
        up(&(mon->mutex));
    }
    // 将当前进程添加至等待队列，并选择调度其他进程。
    down(&(cvp->sem));
    // 再次执行时，表明条件已经满足，当前条件变量上等待进程数减一。
    cvp->count--;
}

另外，当编写管程内部函数时，其格式如下：

1
2
3
4
5
6
7
8
9
10
void xxx() {
    down(&(mtp->mutex));

    //--------the real body of function--------------

    if(mtp->next_count>0)
        up(&(mtp->next));
    else
        up(&(mtp->mutex));
}

似乎，高级语言内部实现管程是比较简单的。

对于一个普通的类而言，隐式添加成员变量 mutex、在各方法前后隐式添加获取和释放 mutex 的代码、向 cond_wait() 传递 mutex 即可(条件变量的设置和使用需要由用户编写，因为这部分涉及具体逻辑)。

鉴于条件变量的两个操作难于理解，在此以一个例子进行说明。

所涉线程：

1
2
3
4
线程 A:							线程 B:
	...								...
	cond_wait();					cond_signal();
	...								...

执行流程：

  1. 线程 A 获取 mutex，从而开始执行。后续因等待某条件发生，因而执行 cond_wait()。
  2. cond_wait() 执行后，将线程 A 添加至该条件变量对应的等待队列中，并调度其他线程执行 (因为目前 next 所示的等待队列为空，因而执行 up(&(mon->mutex))，从而释放 mutex)。
  3. 线程 B 获取 mutex，从而开始执行。后续满足某条件，因而执行 cond_signal()。
  4. cond_signal() 执行后，唤醒该条件变量对应的等待队列上的一个线程，将当前线程添加至 next 所示的等待队列上，并调度其他线程执行。
  5. 线程 A 继续执行其函数体，后续因为 next_count > 0 唤醒 next 所示等待队列上的一个线程，然后完成执行。
  6. 线程 B 继续执行其函数体，后续因为 next_count = 0，释放 mutex，然后完成执行。

针对上述流程，需要留意几点：

  1. 线程 A 再次获取执行权限时，其并没有获取锁 mutex，而是继续使用线程 B 所获取的锁 mutex。因为线程 B 已经睡眠，因此这里不会发生互斥访问 。
  2. 线程 A 退出时，它所做的操作是唤醒线程 B，而非释放锁。释放锁的操作由最初获取锁的线程 B 自己释放。

综合说明 Hoare Semantics 实现中保证互斥访问的机制：

  1. 线程间如果不存在条件变量的羁绊，则其依靠 mutex 实现互斥访问。
  2. 线程间如果存在条件变量的羁绊，执行 cond_signal() 的线程与执行 cond_wait() 的线程共享前者的锁，但是由于 cond_signal() 执行会阻塞一个、唤醒一个，故仍然保证互斥访问。