Nachos实习——Lab3同步机制实习报告

文章目录

• Nachos实习——Lab3同步机制实习报告
• • 内容一：总体概述
  • 内容二：任务完成情况
  • 内容三：具体完成Exercise情况
  • • Exercise 1 调研
    • Exercise 2 源代码阅读
    • Exercise 3 实现锁和条件变量
    • • 1、Lock实现
      • 2、条件变量Condition
    • Exercise 4 实现同步互斥实例
    • • 1、信号量实现生产者消费者问题
      • 2、条件变量实现生产者消费者问题
    • Challenge 1 实现barrier（至少选做一个Challenge）
    • Challenge 2 实现read/write lock
    • Challenge 3 研究Linux的kfifo机制是否可以移植到Nachos上作为一个新的同步模块。
  • 内容四：遇到的困难以及解决方法
  • 内容五：收获及感想
  • 内容六：对课程的意见和建议
  • 内容七：参考文献

内容一：总体概述

本实习希望通过修改Nachos系统平台的底层源代码，达到“扩展同步机制，实现同步互斥实例”的目标。

内容二：任务完成情况

	Exercise 1	Exercise 2	Exercise 3	Exercise 4	Challenge
完成情况	Y	Y	Y	Y	Y

内容三：具体完成Exercise情况

Exercise 1 调研

​ 调研Linux中实现的同步机制。具体内容见课堂要求。

调研情况：

我是在Linux-4.14版本下做调查，Linux的同步机制的相关变量函数都定义在/include/asm-generic目录下。该目录下定义了：原子操作、自旋锁、读写自旋锁、信号量、读写信号量、互斥量、完成变量、顺序锁、禁止抢占、顺序和屏障等同步机制。

Exercise 2 源代码阅读

​ 阅读下列源代码，理解Nachos现有的同步机制。

• code/threads/synch.h和code/threads/synch.cc
• code/threads/synchlist.h和code/threads/synchlist.cc

1、threads/synch.h(cc)

上述文件中声明了三个类Semaphore、Lock、Condition。其中条件变量和锁只是声明，并未定义；而信号量既声明也定义了。下面简单介绍下每个类的关键函数成员。

• Semaphore

该类定义了一个信号量值和一个等待队列，对信号量有两个操作：P操作和V操作，这两个操作都是原子操作(Nachos因为是一个运行在单一处理器上的操作系统，所以原子操作只需要进行开挂中断即可实现)。其中P操作：当value等于0时，将当前运行线程放入线程等待队列，当前进程进入睡眠状态，并切换到其他线程运行；当value大于0时，value–。V操作：如果线程等待队列中有等待该信号量的线程，取出其中一个将其设置成就绪态，准备运行，value++。

• Lock

在现有的 Nachos 中，没有给出锁机制的实现，锁有两个操作 Acquire 和 Release，它们都是原子操作。 Acquire: 申请锁：当锁处于 BUSY 态，进入睡眠状态。 当锁处于 FREE 态，当前线程获得该锁，继续运行。Release: 释放锁（注意：只有拥有锁的线程才能释放锁）将锁的状态设置成 FREE 态，如果有其它线程等待该锁，将其中的一个唤醒，进入就绪态。

• Condition

在现有的 Nachos 中，没有给出条件变量的实现，条件变量有三个操作 Wait、Signal 以及 BroadCast， 所有的这些操作必须在当前线程获得一个锁的前提下，而且所有对一个条件变量进行的操作必须建立在同一个锁的前提下。其中void Wait (Lock *conditionLock)把线程放入条件变量的等待队列上。void Signal (Lock *conditionLock) 唤醒一个等待该条件变量的线程（如果存在的话）。void BroadCast (Lock *conditionLock)） 唤醒所有等待该条件变量的线程（如果存在的话）

2、code/threads/synchlist.h(cc)

利用锁、条件变量实现的一个消息队列。提供了对List的Append()，Remove()和Mapcar()操作。每个操作都要先获得该锁，然后才能对List进行相应的操作，确保在多线程同时访问该List时，不会发生同步错误。

Exercise 3 实现锁和条件变量

​ 可以使用sleep和wakeup两个原语操作（注意屏蔽系统中断），也可以使用Semaphore作为唯一同步原语（不必自己编写开关中断的代码）。

1、Lock实现

a、基本思想

我选择使用值为1的信号量来实现锁功能。首先在Lock类中添加成员变量semaphore和holder(表示持有锁的进程)，同时请求锁和释放锁都必须关中断。

b、synch.h

class Lock {
  private:
    Thread* holderThread;
    Semaphore* semaphore; 
};

c、synch.cc

下面代码的思想都很简单，就不做解释了！

Lock::Lock(char* debugName){
    name = debugName;
    semaphore = new Semaphore("Lock",1);//设置值为1的信号量实现锁
}
Lock::~Lock(){
    delete semaphore;
}
void Lock::Acquire(){
    IntStatus oldLevel = interrupt->SetLevel(IntOff);//关中断
    DEBUG('s', "锁\"%s\"被线程\"%s\"请求\n", name, currentThread->getName());
    semaphore->P();
    holder = currentThread;
    (void)interrupt->SetLevel(oldLevel);//恢复中断
}
bool Lock::isHeldByCurrentThread()
{
    return currentThread == holder;
}
void Lock::Release(){
    IntStatus oldLevel = interrupt->SetLevel(IntOff);
    DEBUG('s', "锁\"%s\"被线程\"%s\"释放\n", name, currentThread->getName());
    ASSERT(this->isHeldByCurrentThread());//判断当前进程是否持有锁
    holder=NULL;
    semaphore->V();
    (void) interrupt->SetLevel(oldLevel);
}

d、测试

结合生产者-消费者问题进行测试

2、条件变量Condition

a、设计思想

条件变量的所有操作都是在当前线程获得一个锁的前提下进行的，而且所有对一个条件变量进行的操作 必须建立在同一个锁的前提下。因此需要基于锁机制实现，同时为条件变量添加一个成员变量queue，用于存放所有等待在该条件变量上的线程。

b、synch.h

因为条件变量必须建立在同一个锁的前提下，所以当进行条件变量操作的时候我需要判断该条件变量是被锁，为此我们在类里面添加一个函数isLocked来进行判断

class Condition {
  bool isLocked() { return !semaphore->getValue(); }
  private:
    List* waitQueue;
};
class Semaphore {
  public:
    int getValue() { return value; }
};

c、synch.cc

下面的代码也很简单，只要理解条件变量机制就能实现，就不做过多解释！

Condition::Condition(char* debugName)
{
    name = debugName;
    waitQueue = new List();
}
Condition::~Condition()
{
    delete waitQueue;
}
void Condition::Wait(Lock* conditionLock) { 
    IntStatus oldLevel = interrupt->SetLevel(IntOff);//关中断
    ASSERT(conditionLock->isHeldByCurrentThread());//
    ASSERT(conditionLock->isLocked());//条件变量操作必须建立在锁的前提下
    conditionLock->Release();//释放锁，进入睡眠状态
    waitQueue->Append(currentThread);
    currentThread->Sleep();
    conditionLock->Acquire();//被signal唤醒，重新申请锁
    (void)interrupt->SetLevel(oldLevel);

}
void Condition::Signal(Lock* conditionLock){
    IntStatus oldLevel = interrupt->SetLevel(IntOff);
    ASSERT(conditionLock->isHeldByCurrentThread());
    if(!waitQueue->IsEmpty()){
        Thread* thread = (Thread*) waitQueue->Remove();
        scheduler->ReadyToRun(thread);
    }
    (void)interrupt->SetLevel(oldLevel);
}
void Condition::Broadcast(Lock* conditionLock){

    IntStatus oldLevel = interrupt->SetLevel(IntOff);
    ASSERT(conditionLock->isHeldByCurrentThread());
    DEBUG('b', "Condition \"%s\" Broadcasting: ", name);
    while(!(waitQueue->IsEmpty()))
    {
        Thread* thread = (Thread*) waitQueue->Remove();
        DEBUG('b', "Thread \"%s\", ", thread->getName());
        scheduler->ReadyToRun(thread);
    }
    (void)interrupt->SetLevel(oldLevel);
}

d、测试

结合生产者-消费者问题进行测试

Exercise 4 实现同步互斥实例

​ 基于Nachos中的信号量、锁和条件变量，采用两种方式实现同步和互斥机制应用（其中使用条件变量实现同步互斥机制为必选题目）。具体可选择“生产者-消费者问题”、“读者-写者问题”、“哲学家就餐问题”、“睡眠理发师问题”等。（也可选择其他经典的同步互斥问题）

1、信号量实现生产者消费者问题

a、设计思想

这部分的所有代码全都写在了ThreadTest中，因为需要用到synch.h中的类，所以需要在ThreadTest的头文件中导入synch.h。核心方法包括生产一个产品produceItem、消费一个产品consumeItem、生产者进程ProducerThread、消费者进程ConsumerThread、将产品方法缓冲区putItem、将产品从缓冲区中取走removeItem。具体实现流程：首先定一个buffer类，类中主要包含成员putItem、removeItem；然后实现生产者进程ProducerThread，ProducerThread每次调用produceItem生产出一个产品，然后调用putItem把产品放入缓冲区；消费者进程ConsumerThread每次调用consumeItem消费一个产品，然后调用removeItem把相应产品从缓冲区取走。

b、具体实现

#define BUFFER_SIZE 10//缓冲区大小
typedef struct PRODUCT {
    int value;
} product;//产品编号
class buffer{
  private:
    int count1;
    Lock* buffer_mutex;//互斥访问缓冲区
    Semaphore* fillCount;//满缓冲区个数
    Semaphore* emptyCount;//空缓冲区个数
    product list[BUFFER_SIZE];//缓冲区数组
  public:
    buffer(){
        fillCount = new Semaphore("FillCount",0);
        emptyCount = new Semaphore("EmptyCount",BUFFER_SIZE);
        buffer_mutex = new Lock("Buffer_mutex");
        count1 = 0;
    }
    void putItem(product* item){
        emptyCount->P();//判断是否还有空缓冲区
        buffer_mutex->Acquire();//获得对临界资源的互斥访问权

        list[count1++]=*item;//将产品放入数组

        buffer_mutex->Release();//释放对临界资源的访问权
        fillCount->V();//唤醒在等待的消费者进程
    }
    product* removeItem(){
        fillCount->P();//判断是否有可以消费的产品
        buffer_mutex->Acquire();// 获取对临界资源的互斥访问权

        product* item = &list[count1-- -1];//从临界资源中获取产品

        buffer_mutex->Release();//释放临界资源的访问权
        emptyCount->V(); // 唤醒在等待的生产线程

        return item;
    }
}*shared_buffer;
product* produceItem(int value)//生产一个产品
{
    printf("Producing item with value %d!!\n", value);
    product item;
    item.value = value;
    return &item;
}

void consumeItem(product* item)/消费一个产品
{
    printf("Consuming item with value %d!!\n", item->value);
}

void ProducerThread(int iterNum)//生产者线程
{
    for (int i = 0; i < iterNum; i++) {
        printf("## %s ##: ", currentThread->getName());
        product* item = produceItem(i);
        shared_buffer->putItem(item);
        interrupt->OneTick();//让时钟前进，从而使得使用-rs调用TimerInterruptHandler
    }
}

void ConsumerThread(int iterNum)
{
    for (int i = 0; i < iterNum; i++) {
        printf("$$ %s $$: ", currentThread->getName());
        product* item = shared_buffer->removeItem();
        consumeItem(item);
        interrupt->OneTick();
    }
}

c、测试

我创建了两个生产者进程和两个消费者进程，具体信息如下：

Producer1	9
Producer1	6
Consumer1	7
Consumer2	8

void
Lab3ProducerConsumer()
{
    DEBUG('t', "Entering Lab3ProducerConsumer");

    shared_buffer = new buffer();

    Thread *producer1 = new Thread("Producer 1");
    Thread *producer2 = new Thread("Producer 2");
    Thread *consumer1 = new Thread("Consumer 1");
    Thread *consumer2 = new Thread("Consumer 2");

    producer1->Fork(ProducerThread, (void*)9);
    consumer1->Fork(ConsumerThread, (void*)7);
    consumer2->Fork(ConsumerThread, (void*)8);
    producer2->Fork(ProducerThread, (void*)6);

    currentThread->Yield(); // Yield the main thread
}

测试截图：

2、条件变量实现生产者消费者问题

a、设计思想

设计思想与信号量几乎一样，只是操作上的区别

b、具体实现

在信号量的基础上修改private定义、构造函数、putItem和removeItem函数

class buffer{
  private:
    int count1;
    Lock* buffer_mutex;//互斥访问缓冲区
    Condition* condc;
		Condition* condp;
    product list[BUFFER_SIZE];//缓冲区数组
  public:
     buffer(){
        condc = new Condition("ConsumerCondition");
        condp = new Condition("ProducerCondition");
        buffer_mutex = new Lock("Buffer_mutex");
        count1 = 0;
    }
    void putItem(product* item){
        buffer_mutex->Acquire();//获得对临界资源的互斥访问权
        while(count1>=BUFFER_SIZE){
            printf("Product alread full:[%d]\n",count1);
            condp->Wait(buffer_mutex);
        }
        list[count1++]=*item;//将产品放入数组
        condc->Signal(buffer_mutex);//唤醒消费者进程
        buffer_mutex->Release();//释放对临界资源的访问权
    }
    product* removeItem(){
        buffer_mutex->Acquire();// 获取对临界资源的互斥访问权
        while(count1<=0)
        {
            printf("-->Product alread empty:[%d],\n",count1);
            condc->Wait(buffer_mutex);
        }
        product* item = &list[count1--];//从临界资源中获取产品
        condp->Signal(buffer_mutex);//唤醒生产者进程
        buffer_mutex->Release();//释放临界资源的访问权
        return item;
    }
}*shared_buffer;

c、测试

你可以尝试设置一个更小的buffer_size进行测试，能帮助你更好理解条件变量！

Challenge 1 实现barrier（至少选做一个Challenge）

​ 可以使用Nachos 提供的同步互斥机制（如条件变量）来实现barrier，使得当且仅当若干个线程同时到达某一点时方可继续执行。

TODO

Challenge 2 实现read/write lock

​ 基于Nachos提供的lock(synch.h和synch.cc)，实现read/write lock。使得若干线程可以同时读取某共享数据区内的数据，但是在某一特定的时刻，只有一个线程可以向该共享数据区写入数据。

1、设计思想

读者-写者问题的核心思想：使得若干线程可以同时读取某共享数据区内的数据，但是在某一特定的时刻，只有一个线程可以向该共享数据区写入数据。

伪代码：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-QnVieTEz-1605675571326)(/Users/jaggerqian/Library/Application Support/typora-user-images/image-20201118112727901.png)]

2、synch.h

class ReaderWriterLock {
  public:
    ReaderWriterLock(char* debugName); 
    ~ReaderWriterLock(); 
    char* getName() { return (name); } 

    void ReaderAcquire();
    void ReaderRelease();
    void WriterAcquire();
    void WriterRelease();
  	int getBlockingReader(){return blockingReader;};
  private:
    char* name;            
    int blockingReader;     //记录读者访问数量
    Semaphore* binary_semaphore_writer;//读者写着互斥访问临界区
    Lock* mutex_reader;//多个读者互斥访问blockingReader
};

3、synch.cc

ReaderWriterLock::ReaderWriterLock(char* debugName)
{
    name = debugName;
    blockingReader = 0;
    binary_semaphore_writer = new Semaphore("ReaderWriterLock Binary Semaphore for Writer", 1);
    mutex_reader = new Lock("ReaderWriterLock Mutex for Readers");
}
ReaderWriterLock::~ReaderWriterLock()
{
    delete binary_semaphore_writer;
    delete mutex_reader;
}

void
ReaderWriterLock::ReaderAcquire()//完全仿照伪代码实现
{
    IntStatus oldLevel = interrupt->SetLevel(IntOff); 
    mutex_reader->Acquire(); //
    blockingReader++;
    if (blockingReader == 1) {
        binary_semaphore_writer->P(); 
    }
    mutex_reader->Release(); 
    (void) interrupt->SetLevel(oldLevel);  
}
void ReaderWriterLock::ReaderRelease()
{
    IntStatus oldLevel = interrupt->SetLevel(IntOff);  
    mutex_reader->Acquire(); 
    blockingReader--;
    if (blockingReader == 0) {
        binary_semaphore_writer->V(); 
    }
    mutex_reader->Release(); 
    (void) interrupt->SetLevel(oldLevel);  
}

void ReaderWriterLock::WriterAcquire()
{
    binary_semaphore_writer->P();
}
void ReaderWriterLock::WriterRelease()
{
    binary_semaphore_writer->V();
}

4、测试

我设置了一个数组wandr，里面存放的是，写进程需要写的内容；设置了一个quote，里面为写进程已经写好的内容；我最开始设置两个读进程，后来发现读进程已经运行结束了，写进程还没写完（因为随机时钟中断的原因）；因此我又设置了3个进程。

const char wandr[9][100] = {
    "自我介绍:", "我", "的", "英", "文", "名", "叫","Jagger","!"
};
int words;//用来记录已经写了多少个单词
char quote[9][100];
ReaderWriterLock* RWLock;
void
WriterThread(int dummy)
{
    while (words < 9) {
        RWLock->WriterAcquire();

        strcpy(quote[words], wandr[words]); 
        words++;
        printf("Writer is writting: ");
        for (int i = 0; i < words; i++) {
            printf("%s ", quote[i]);
        }
        printf("\n");
        RWLock->WriterRelease();
    }
}
void
ReaderThread(int reader_id)
{
    do {
        RWLock->ReaderAcquire();

        printf("Reader %d is reading: ", reader_id);
        for (int i = 0; i < words; i++) {
            printf("%s ", quote[i]);
        }
        printf("\t(blockingReader=%d)\n",RWLock->getBlockingReader());

        RWLock->ReaderRelease();
    } while (words < 9);
}
void
Lab3ReaderWriter()
{
    const int total_reader = 2;
    Thread* writer = new Thread("Writer");
    writer->Fork(WriterThread, (void*)0);

    for (int i = 0; i < total_reader; i++) {
        char readerName[10];
        sprintf(readerName, "Reader %d", i+1); 
        Thread* reader = new Thread(strdup(readerName)); 
        reader->Fork(ReaderThread, (void*)i+1);
    }
    RWLock = new ReaderWriterLock("RWLock");
    currentThread->Yield(); 
}

第一次测试截图：

第二次测试截图：

Challenge 3 研究Linux的kfifo机制是否可以移植到Nachos上作为一个新的同步模块。

TODO

内容四：遇到的困难以及解决方法

无

内容五：收获及感想

这次的实验比较简单，只要掌握了同步互斥机制，就能顺利实现。

内容六：对课程的意见和建议

这是一门非常值得推荐的课。

内容七：参考文献

• [1] Nachos中文教程

• [2] 同步机制实习报告

• [3] 同步机制实习报告