操作系统三：同步问题

原创文章，转载、引用请注明出处！
代码、文章及图片挂载地址：https://github.com/MoyangSensei/OS

题目

• 1 通过fork的方式，产生4个进程P1,P2,P3,P4，每个进程打印输出自己的名字，例如P1输出“I am the process P1”。要求P1最先执行，P2、P3互斥执行，P4最后执行。通过多次测试验证实现是否正确。

• 2 火车票余票数ticketCount 初始值为1000，有一个售票线程，一个退票线程，各循环执行多次。添加同步机制，使得结果始终正确。要求多次测试添加同步机制前后的实验效果。(说明：为了更容易产生并发错误，可以在适当的位置增加一些pthread_yield()，放弃CPU，并强制线程频繁切换，例如售票线程的关键代码：

//售票线程的关键代码：
temp=ticketCount;
pthread_yield();
temp=temp-1;
pthread_yield();
ticketCount=temp;
//退票线程的关键代码：
temp=ticketCount;
pthread_yield();
temp=temp+1;
pthread_yield();
ticketCount=temp;

• 3 一个生产者一个消费者线程同步。设置一个线程共享的缓冲区， char buf[10]。一个线程不断从键盘输入字符到buf,一个线程不断的把buf的内容输出到显示器。要求输出的和输入的字符和顺序完全一致。（在输出线程中，每次输出睡眠一秒钟，然后以不同的速度输入测试输出是否正确）。要求多次测试添加同步机制前后的实验效果。

• 4 （选做题2）进程通信问题。阅读并运行共享内存、管道、消息队列三种机制的代码
  消息队列：https://www.cnblogs.com/Jimmy1988/p/7699351.html
  管道：https://www.cnblogs.com/Jimmy1988/p/7553069.html
  共享内存：https://www.cnblogs.com/Jimmy1988/p/7706980.html
  a）通过实验测试，验证共享内存的代码中，receiver能否正确读出sender发送的字符串？如果把其中互斥的代码删除，观察实验结果有何不同？如果在发送和接收进程中打印输出共享内存地址，他们是否相同，为什么？
  b) 有名管道和无名管道通信系统调用是否已经实现了同步机制？通过实验验证，发送者和接收者如何同步的。比如，在什么情况下，发送者会阻塞，什么情况下，接收者会阻塞？
  c) 消息通信系统调用是否已经实现了同步机制？通过实验验证，发送者和接收者如何同步的。比如，在什么情况下，发送者会阻塞，什么情况下，接收者会阻塞？

• 5 阅读Pintos操作系统，找到并阅读进程上下文切换的代码，说明实现的保存和恢复的上下文内容以及进程切换的工作流程。

---

解答

1

• （1）根据题意，四个进程的输出顺序仅有两种：

1->2->3->4
1->3->2->4

p1最先输出信息，p4最后输出信息，由于p2和p3在整个过程中处于相同的输出顺序，即位于p1和p4中间输出，那么便可将它们看作一个进程，称作p23。

执行过程中，p1无条件最先执行，无需控制。p23需要等待来自p1的信号，当p1执行完毕后，p23才可以执行。那么此处需要一个信号量来控制：

sem1 = sem_open("1", O_CREAT, 0666, 0);

同理，p4要等待p23执行完之后再执行，此处需要另一个信号量来控制：

sem4 = sem_open("4", O_CREAT, 0666, 0);

最后，在p23内部，p2和p3执行过程中需要满足互斥的条件，此处还需要一个信号量来控制：

sem2 = sem_open("2", O_CREAT, 0666, 0);

代码如下：

#include <pthread.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <semaphore.h>
#include <sys/types.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>

sem_t* sem1 = NULL;
sem_t* sem4 = NULL;
sem_t* sem2 = NULL;

int main(){
    pid_t p,p_father;
    sem1 = sem_open("1", O_CREAT, 0666, 0);
    sem4 = sem_open("4", O_CREAT, 0666, 0);
    sem2 = sem_open("2", O_CREAT, 0666, 0);
    p_father = getpid();
    p = fork();
    p = fork();
    if (p>0){
        if(getpid() != p_father){
            p=fork();
            if(p==0){
                sem_wait(sem1);
                sem_wait(sem4);
                printf("I am the process 4\n");
            }
            else{
                printf("I am the process 1\n");
                sem_post(sem2);
            }
        }
    }
    else if (p==0){
        if(getppid()!=p_father){
            sem_wait(sem2);
            printf("I am the process 3\n");
            sem_post(sem2);
            sem_post(sem4);
        }
        else{
            sem_wait(sem2);
            //sleep(1);
            printf("I am the process 2\n");
            sem_post(sem2);
            sem_post(sem1);
        }
    }
    sleep(1);
    sem_close(sem1);
    sem_close(sem4);
    sem_close(sem2);
    sem_unlink("1");
    sem_unlink("4");
    sem_unlink("2");
    return 0;
}

• （2）通过多次实验，发现运行结果仅出现上述两种结果中的一种。即1->2->3->4。此现象的原因由于代码顺序的缘故，p2一定比p3先进入，且本程序仅仅只是控制台输出一行话，几乎不存在p3由于运行的速度快而“反超”p2。
  若想要出现第二种结果，则可以p2进程执行的代码中加入了sleep(1)，代码见上方注释。
  两种运行结果测试如下：
  

2

• （1）根据题意编写代码如下：

#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
pthread_mutex_t mutex ;
int ticket_counter;
int producer_counter;
int consumer_counter;
int all_pro;int all_con;

void *producer( int i ) {
    //printf("this is producer");
    while(--i>0){
        pthread_mutex_lock( &mutex );
        //sched_yield();
        ticket_counter++;
        sched_yield();
        sched_yield();
        pthread_mutex_unlock( &mutex );
    }
    pthread_exit( NULL );
}

void *consumer( int j ) {
    while(j-->0){
        pthread_mutex_lock( &mutex );
        //sched_yield();
        ticket_counter--;
        sched_yield();
        sched_yield();
        pthread_mutex_unlock( &mutex );
    }
    pthread_exit( NULL );
}

void create_and_join_Pthread(){
    all_con=0;all_pro=0;
    pthread_t *pro, *con;
    pro=(pthread_t*)malloc(sizeof(pthread_t)*producer_counter);
    con=(pthread_t*)malloc(sizeof(pthread_t)*consumer_counter);
    printf("[%d]个线程分别产生的票数为: ",producer_counter);
    srand((unsigned)time(NULL));
    for(int i=0;i<producer_counter;i++){
        int a = rand() % 51 + 1;
        all_pro+=a;
        printf("%d ",a);
        pthread_create( &pro[i], NULL, producer, a );
    }
    printf("\n");
    printf("[%d]个线程分别售出的票数为: ",consumer_counter);
    for(int i=0;i<consumer_counter;i++){
        int a = rand() % 51 + 1;
        all_con+=a;
        printf("%d ",a);
        pthread_create( &con[i], NULL, consumer, a );
    }
    printf("\n");
    for(int i=0;i<producer_counter;i++){
        pthread_join( pro[i], NULL );
    }
    for(int i=0;i<producer_counter;i++){
        pthread_join( con[i], NULL );
    }
}

int main() {
    printf("输入产生车票线程数/销售车票线程数/开始时总票数:");
    scanf("%d %d %d",&producer_counter,&consumer_counter,&ticket_counter);
//    ticket_counter=10000;
//    producer_counter=12;
//    consumer_counter=2;
    int temp=ticket_counter;
    pthread_mutex_init(&mutex, NULL);
    create_and_join_Pthread();
    printf("完成该过程后应有余票[%d],其中售出[%d],生产[%d].\n",temp-all_con+all_pro,all_con,all_pro);
    printf("\n实际还有余票为[%d]\n",ticket_counter+producer_counter);
    return 0;
}

上述代码中，运行需要输入三项信息：产生车票线程数/销售车票线程数/开始时总票数。使用唯一一个信号量mutex来控制所有线程。create_and_join_Pthread()函数用来根据输入的信息添加相应数量的消费线程和生产线程。每个生产线程所生产的数目由随机函数生成，为1-50之间的随机数。这样做的目的是方便模拟不同线程数量的情况。

• （2）当添加*sched_yield()*时，多次运行程序，得到结果如下。发现错误概率很小，即使是填入了非常大的数据，也不一定会出错，但是错误是一定存在的：

将上述代码注释之后，结果如下，它们都是完全正确的：

填入极大数据测试，发现结果也是正确的：

Linux虚拟机无法容纳如下所示的线程量，更换测试环境到mac os下的编译器Xcode。
对于unix系统来说，是由贝尔实验室开发的多用户、多任务操作系统。
Linux是一类unix操作系统的统称，严格来说，Linux系统只有内核叫“Linux”，而Linux也只是表示其内核，但因为习惯使然，人们习惯了用linux称呼这类系统。一般也可以认为，Linux是一套自由使用和自由传播的类unix系统。
mac os是苹果公司个人电脑的专用系统，是基于unix内核的图形化操作系统，它可以使用Linux系统下的绝大部分指令和操作。
该测试环境不会影响运行结果，之后有可能还会使用此测试环境。

3

生产者、消费者问题可以从线程同步和进程同步两个方面考虑。倘若使用进程同步的思想，使用sem_wait和sam_pose函数控制两个信号量使得生产和消费互斥即可，当然同样需要考虑队满和队空的阻塞、唤醒机制。

• （1）根据题意，本题目为一个消费者线程和一个生产者线程进行互斥的问题。可以使用条件变量。
  条件变量是利用线程间共享的全局变量进行同步的一种机制，主要包括两个动作：一个线程等待”条件变量的条件成立“而挂起；另一个线程使“条件成立”（给出条件成立信号）。
  为了防止竞争，条件变量的使用总是和一个互斥锁结合在一起，具体用法如下：

  pthread_mutex_lock(&mutex)
      while或if(线程执行的条件是否成立)
          pthread_cond_wait(&cond, &mutex);
      //线程执行
      pthread_mutex_unlock(&mutex);

• （2）根据上述分析，编写代码：

  #include <string.h>
  #include <pthread.h>
  #include <unistd.h>
  #include <stdio.h>
  
  char buf[10];
  char print;
  
  int in = 0, out = 0;
  int items = 0, spaces = 10;
  pthread_mutex_t mutex ;
  pthread_cond_t count1 = PTHREAD_COND_INITIALIZER;  // 缓冲区不满
  pthread_cond_t count2 = PTHREAD_COND_INITIALIZER;
  void *producer( void *arg ) {
          //printf("this is producer");
          while(1){       
              pthread_mutex_lock( &mutex );  
              while( items==10 ) {
                  pthread_cond_wait( &count1, &mutex );
              }
              //printf("this is real producer");
              char temp;
              scanf("%c",&temp);
              buf[in]=temp;
              in = ( in + 1 ) % 10;
              items++;spaces--;
              pthread_cond_signal( &count2 );
              pthread_mutex_unlock( &mutex );
         
      }
      pthread_exit( NULL );
  }
  
  void *consumer( void *arg ) {
      while(1){    
              //printf("this is consumer");
              pthread_mutex_lock( &mutex );
              while( spaces==10 ) {
                  pthread_cond_wait( &count2, &mutex );
              }
              //printf("this is real consumer");
              //sleep(3);
              print=buf[out];
              out = ( out + 1 ) % 10;
              items--;spaces++;
              printf( "consumer : %c \n", print);
              pthread_cond_signal( &count1 );
              pthread_mutex_unlock( &mutex );
      }
      pthread_exit( NULL );
  }
  
  int main() {
      pthread_mutex_init(&mutex, NULL);
      pthread_t pro, con;
      pthread_create( &pro, NULL, producer, NULL );
      pthread_create( &con, NULL, consumer, NULL );
      pthread_join( pro, NULL );
      pthread_join( con, NULL );
      return 0;
  }

• （3）运行该程序。可在输出途中输入，证明互斥变量是生效的。

4

• （1）分别在两个终端内运行Sender和Receiver，结果如下：

阅读代码，发现进行互斥操作的部分如下：

//6. Operation procedure
    struct sembuf sem_b;
    sem_b.sem_num = 0;      //first sem(index=0)
    sem_b.sem_flg = SEM_UNDO;
    sem_b.sem_op = 1;           //Increase 1,make sem=1

将两份代码中的此部分全部注释，再次运行，发现接收端无法正常接受到内容（不输出提示信息），发送端可以正常发送，并且可以正常结束掉进程（输入end），结果如下：

由于是在两个终端下运行，且是先运行了发送端，根据上面的结果可以推断，如果先运行的是接收端，那么发送端的提示信息同样无法显示。总的来说，注释掉互斥部分后，代码无法正常运行。

在原来的基础上添加代码，使得代码可以输出共享内存的标示号shm_id。再次运行程序，发现两程序的共享内存标示号是一致的，这意味着它们所使用的共享内存的地址空间是一致的。
那么，由此可以推断共享内存的解释为：允许两个或多个进程共享一个给定的存储区。

• （2）运行无名管道程序，如下所示：

无名管道通行作为Linux进程通信中的一种，是UNXI系统IPC中最古老的一种。对于无名管道来说，他有如下特点：

• 它是半双工的（即数据只能在一个方向上流动），具有固定的读端和写端。
• 它只能用于具有亲缘关系的进程之间的通信（也是父子进程或者兄弟进程之间），实现依赖父子进程文件共享。
• 它可以看成是一种特殊的文件，对于它的读写也可以使用普通的read、write等函数。但是它不是普通的文件，并不属于其他任何文件系统，并且只存在于内存中。
• 它是阻塞读写的。

对于Linux管道文件来说，可以使用两种读写方式：阻塞读写与非阻塞读写。所谓阻塞读写，是无论是先读还是先写都要等到另一个操作才能离开阻塞。也就是：如果先读，陷入阻塞，等待写操作；如果先写，陷入阻塞，等待读操作。对应的有非阻塞读写，它们无须等待另一个操作的，直接执行read()或者write()能读就读，能写就写，不能就返回-1。

分别在两个终端上运行有名管道程序的读和写端，如下所示：

对于该程序来说，运行过程中，先运行写端后运行读端，当在运行写端输入内容并输入回车键之后，读端读出管道文件中的数据，并在同一时刻与写端一起结束，这种运行方式下，可以完成给定的要求。但是，如果把两端的运行顺序更换，变成先运行读端后运行写端，那么可以清楚的发现，给定功能会失效，对于写端来说，甚至不会输出提示信息。由此我们可以推断，有名管道的也是阻塞读写的，当读端占据了管道后，写端无法进入，该流程就无法达到原有的效果。换言之，该程序正确运行顺序为先运行写端，再运行读端，最后在写端内写入数据并输入回车，否则程序会因为初始的管道中无数据且先进入的读端阻塞了写端导致失效。

• （3）分别在两个终端上运行消息队列程序的读和写端，如下所示：

对于消息队列而言，它的机制非常类似于第三题中的输入和输出程序。可以推断，消息队列也是阻塞读写的。而且它还必须拥有队空阻塞读和队满阻塞写的机制，才可以真正的实现完整的传输机制。针对这样的推断进行实验，发现是正确的。

5

查阅资料，发现规定进程上下文切换的代码位于/threads/switch.h和/threads/switch.S中。对于该门课程原本的实验来说，这一部分的内容包含在实验一：thread里面。实验原理是通过bochs加载pintos操作系统，该操作系统会根据pintos的实现打印运行结果，通过比较标准输出文档和实际输出，来判断pintos实现是否符合要求。

如果想要了解该系统的上下文切换，我们就不得不去了解timer_sleep函数（/devices/timer.c）。系统原本是使用busy wait实现的，即线程不停地循环，直到时间片耗尽。timer_sleep的实现方式如下：

/* Sleeps for approximately TICKS timer ticks.  Interrupts must
   be turned on. */
void
timer_sleep (int64_t ticks)
{
  int64_t start = timer_ticks ();
  ASSERT (intr_get_level () == INTR_ON);
  while (timer_elapsed (start) < ticks)
    thread_yield();
}

对于timer_ticks函数来说，有intr_level通过intr_disable返回了一内容。继续向下寻找，可以明显发现，intr_level代表能否被中断，而intr_disable做了两件事情：调用intr_get_level()、直接执行汇编代码，调用汇编指令来保证这个线程不能被中断。

/* Returns the current interrupt status. */
enum intr_level
intr_get_level (void) 
{
  uint32_t flags;

  /* Push the flags register on the processor stack, then pop the
     value off the stack into `flags'.  See [IA32-v2b] "PUSHF"
     and "POP" and [IA32-v3a] 5.8.1 "Masking Maskable Hardware
     Interrupts". */
  asm volatile ("pushfl; popl %0" : "=g" (flags));

  return flags & FLAG_IF ? INTR_ON : INTR_OFF;
}

这个函数一样是调用了汇编指令，把标志寄存器的东西放到处理器棧上，然后把值pop到flags（代表标志寄存器IF位）上，通过判断flags来返回当前终端状态(intr_level)。

至此，函数嵌套了这么多层，整理逻辑：

1. intr_get_level返回了intr_level的值

2. intr_disable获取了当前的中断状态， 然后将当前中断状态改为不能被中断， 然后返回执行之前的中断状态。

有以上结论我们可以知道： timer_ticks的intr_get_level做了如下的事情： 禁止当前行为被中断， 保存禁止被中断前的中断状态（用old_level储存）。

/* Returns the current interrupt status. */
/* Enables or disables interrupts as specified by LEVEL and
   returns the previous interrupt status. */
enum intr_level
intr_set_level (enum intr_level level) 
{
  return level == INTR_ON ? intr_enable () : intr_disable ();
}

对于timer_ticks剩下的内容来说，就是用t获取了一个全局变量ticks, 然后返回， 其中调用了set_level函数。set_level的实现机制，根据上述内容容易得出： 如果之前是允许中断的（INTR_ON）则enable否则就disable。intr_enable的实现和之前基本一致：

/* Enables interrupts and returns the previous interrupt status. */
enum intr_level
intr_enable (void) 
{
  enum intr_level old_level = intr_get_level ();
  ASSERT (!intr_context ());

  /* Enable interrupts by setting the interrupt flag.

     See [IA32-v2b] "STI" and [IA32-v3a] 5.8.1 "Masking Maskable
     Hardware Interrupts". */
  asm volatile ("sti");

  return old_level;
}

这里直接返回了是否外中断的标志in_external_intr，就是说ASSERT断言这个中断不是外中断（IO等， 也称为硬中断）而是操作系统正常线程切换流程里的内中断（也称为软中断）。至此，我们可以分析出Pintos系统保持原子性的语句，即使用如下的语句包裹：

/* Enables interrupts and returns the previous interrupt status. */
enum intr_level old_level = intr_disable ();
...
intr_set_level (old_level);

对于ticks来说，从pintos被启动开始，ticks就一直在计时，代表着操作系统执行单位时间的前进计量。现在回过来看timer_sleep这个函数，start获取了起始时间，然后断言必须可以被中断，不然会一直死循环下去。

然后就是如下循环：

/* Enables interrupts and returns the previous interrupt status. */
while (timer_elapsed (start) < ticks)
thread_yield();

注意这个ticks是函数的形参不是全局变量，然后看一下这两个函数：对于timer_elapsed来说，它返回了当前时间距离then的时间间隔，那么这个循环实质就是在ticks的时间内不断执行thread_yield。最后来看thread_yield：

/* Yields the CPU.  The current thread is not put to sleep and
   may be scheduled again immediately at the scheduler's whim. */
void
thread_yield (void)
{
  struct thread *cur = thread_current ();
  enum intr_level old_level;

  ASSERT (!intr_context ());

  old_level = intr_disable ();
  if (cur != idle_thread)
    list_push_back (&ready_list, &cur->elem);
  cur->status = THREAD_READY;
  schedule ();
  intr_set_level (old_level);
}

thread_current很明显是返回当前线程起始指针位置。第9行断言是个软中断，第11行至第16包裹起来的就是我们之前分析的线程机制保证的一个原子性操作。对于第12行到15行，如何当前线程不是空闲的线程就调用list_push_back把当前线程的元素扔到就绪队列里面， 并把线程改成THREAD_READY状态。

15行调用schedule:

/* Schedules a new process.  At entry, interrupts must be off and
   the running process's state must have been changed from
   running to some other state.  This function finds another
   thread to run and switches to it.

   It's not safe to call printf() until thread_schedule_tail()
   has completed. */
static void
schedule (void)
{
  struct thread *cur = running_thread ();
  struct thread *next = next_thread_to_run ();
  struct thread *prev = NULL;

  ASSERT (intr_get_level () == INTR_OFF);
  ASSERT (cur->status != THREAD_RUNNING);
  ASSERT (is_thread (next));

  if (cur != next)
    prev = switch_threads (cur, next);
  thread_schedule_tail (prev);
}

可以看出，首先获取当前线程cur和调用next_thread_to_run获取下一个要run的线程，如果就绪队列空闲直接返回一个空闲线程指针， 否则拿就绪队列第一个线程出来返回。如果当前线程和下一个要跑的线程不是同一个的话调用switch_threads返回给prev。

 /* Switches from CUR, which must be the running thread, to NEXT,
    which must also be running switch_threads(), returning CUR in
    NEXT's context. */
struct thread *switch_threads (struct thread *cur, struct thread *next);

这个函数实现是用汇编语言实现的在threads/switch.S里。分析一下这个汇编代码： 先4个寄存器压栈保存寄存器状态（保护作用）， 这4个寄存器是switch_threads_frame的成员，然后全局变量thread_stack_ofs记录线程和栈之间的间隙，我们都知道线程切换有个保存现场的过程，先把当前的线程指针放到eax中，并把线程指针保存在相对基地址偏移量为edx的地址中，切换到下一个线程的线程棧指针，保存在ecx中，再把这个线程相对基地址偏移量edx地址（上一次保存现场的时候存放的）放到esp当中继续执行。这里ecx，eax起容器的作用，edx指向当前现场保存的地址偏移量。简单来说就是保存当前线程状态，恢复新线程之前保存的线程状态。然后再把4个寄存器拿出来，这个是硬件设计要求的，必须保护switch_threads_frame里面的寄存器才可以destroy掉eax， edx， ecx。然后注意到现在eax(函数返回值是eax)就是被切换的线程栈指针。我们由此得到一个结论，schedule先把当前线程丢到就绪队列，然后把线程切换如果下一个线程和当前线程不一样的话。

至此，schedule的重点部分分析完成。逻辑继续向上回溯：

thread_schedule_tail其实就是获取当前线程，分配恢复之前执行的状态和现场，如果当前线程死了就清空资源。

schedule其实就是拿下一个线程切换过来继续run。

thread_yield其实就是把当前线程扔到就绪队列里，然后重新schedule，注意这里如果ready队列为空的话当前线程会继续在cpu执行。

最后回溯到我们最顶层的函数逻辑：timer_sleep就是在ticks时间内，如果线程处于running状态就不断把他扔到就绪队列不让他执行。

我们对原来的timer_sleep的实现方式有了十分清楚的理解了，我们也很清楚的看到了它的缺点：此时的线程不断在cpu就绪队列和running队列之间来回（包括整个线程的上下文切换也是其中的一个部分），占用了cpu资源。

---

参考链接

消息队列：https://www.cnblogs.com/Jimmy1988/p/7699351.html

管道：https://www.cnblogs.com/Jimmy1988/p/7553069.html

共享内存：https://www.cnblogs.com/Jimmy1988/p/7706980.html

pthread_cond_wait学习笔记：https://www.cnblogs.com/secondtonone1/p/5580203.html

Pintos-斯坦福大学操作系统Project详解-Project1：https://www.cnblogs.com/laiy/p/pintos_project1_thread.html