os_lab7:同步互斥

课程主页：https://www.xuetangx.com/course/THU08091000267/5883104?channel=i.area.learn_title
实验指导书：https://objectkuan.gitbooks.io/ucore-docs/content/
github：https://github.com/chyyuu/ucore_os_lab (master 分支)

前置知识

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

改动点

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

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

练习零

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

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

// 表示定时器结构
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 中，使用最简单的 “基于硬件中断” 实现信号量机制。

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

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

// 信号量
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;  	// 该进程放置于等待队列中的原因 (例如：信号量的P操作、定时)
    wait_queue_t *wait_queue;	// 所在的等待队列
    list_entry_t wait_link;		// 链接至等待队列 wait_queue
} wait_t;

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

// P 操作
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();

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

  	// 将当前进程从等待队列中移除
		local_intr_save(intr_flag);
    wait_current_del(&(sem->wait_queue), wait);
    local_intr_restore(intr_flag);
		// 判断 唤醒该进程的原因 和 当时让该进程等待的原因是否一致， 若不一致则返回 wait->wakeup_flags （造成等待的原因）
    if (wait->wakeup_flags != wait_state) {
        return wait->wakeup_flags;
    }
    return 0;
}
// V 操作
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);
}

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

两者的主要差别在于：在形式化中，sem 大于等于零时，表示共享资源的剩余数目，小于零时，表示正在等待资源的进程的数量；在具体实现中，value 仅表示共享资源的剩余数目，等待进程用等待队列来管理 (此时，因为共享资源的剩余数目只可能大于等于零，而不可能小于零，所以 value大于等于零)。

问答

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

肯定需要使用系统调用。

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

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

练习二

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

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

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

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

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

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

对于这三种方式，实际实现基于前两种 (因为可以减少一次上下文切换)，书本介绍基于后一种。（其中Mesa Semantics和Hanson Semantics类似，但前者是B线程先通知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 执行时，其需要循环判断条件是否满足)。

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

// 条件变量
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;

首先给出初始化过程：

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()（Hoare Semantics），具体见源代码：

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--;
}

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

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 即可(条件变量的设置和使用需要由用户编写，因为这部分涉及具体逻辑)。

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

所涉线程：

线程 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() 执行会阻塞一个、唤醒一个，故仍然保证互斥访问。

用管程解决哲学家就餐问题：

check_sync.c:
// 测试编号为i的哲学家是否能获得刀叉 如果能获得 则将状态改为正在吃 并且 尝试唤醒 因为wait操作睡眠的进程
// cond_signal 还会阻塞自己 等被唤醒的进程唤醒自己
void phi_test_condvar (i) { 
    if(state_condvar[i]==HUNGRY&&state_condvar[LEFT]!=EATING
            &&state_condvar[RIGHT]!=EATING) {
        cprintf("phi_test_condvar: state_condvar[%d] will eating\n",i);
        state_condvar[i] = EATING ;
        cprintf("phi_test_condvar: signal self_cv[%d] \n",i);
        cond_signal(&mtp->cv[i]) ;
    }
}
// 拿刀叉
void phi_take_forks_condvar(int i) {
    down(&(mtp->mutex));    // P操作进入临界区
    state_condvar[i] = HUNGRY; // 饥饿状态 准备进食
    phi_test_condvar(i); // 测试当前是否能获得刀叉 
    while (state_condvar[i] != EATING) {
        cond_wait(&mtp->cv[i]); // 若不能拿 则阻塞自己 等其它进程唤醒
    }
    if(mtp->next_count>0)
        up(&(mtp->next));
    else
        up(&(mtp->mutex));
}

// 放刀叉
void phi_put_forks_condvar(int i) {
    down(&(mtp->mutex)); // P操作进入临界区
    state_condvar[i] = THINKING; // 思考状态
    phi_test_condvar(LEFT); // 试试左右两边能否获得刀叉
    phi_test_condvar(RIGHT);
    if(mtp->next_count>0) // 有哲学家睡在 signal操作 则将其唤醒
        up(&(mtp->next));
    else
        up(&(mtp->mutex)); // 离开临界区
}
// 主体函数，创建若干线程模拟哲学家就餐问题
int philosopher_using_condvar(void * arg) { /* arg is the No. of philosopher 0~N-1*/
  
    int i, iter=0;
    i=(int)arg;
    cprintf("I am No.%d philosopher_condvar\n",i);
    while(iter++<TIMES)
    { /* iterate*/
        cprintf("Iter %d, No.%d philosopher_condvar is thinking\n",iter,i); /* thinking*/
        do_sleep(SLEEP_TIME);
        phi_take_forks_condvar(i); 
        /* need two forks, maybe blocked */
        cprintf("Iter %d, No.%d philosopher_condvar is eating\n",iter,i); /* eating*/
        do_sleep(SLEEP_TIME);
        phi_put_forks_condvar(i); 
        /* return two forks back*/
    }
    cprintf("No.%d philosopher_condvar quit\n",i);
    return 0;    
}