Xv6代码阅读报告之进程调度

 
 

Xv6代码阅读报告-Topic3

@肖剑楠 20111013223

 

 

 

1. 序

Xv6为了实现CPU多进程化需要解决一系列问题。1. 如何在进程间切换?2. 如何让这一切换变得透明?3. 需要锁机制来避免竞争。4. 内存、资源的自动释放。
Xv6通过实现上下文切换(Context Switching),时间中断处理,锁,睡眠与唤醒等机制基本解决了上述问题。主要代码包括swtch.S, defs.h, proc.h, proc.c, mmu.h等文件。
下面按模块对上述文件逐一分析。

 

2. 上下文切换

 

2.1 defs.h

在一切的一切之前,我们先看一下defs.h的结构体的定义以及函数的声明。
该文件中集中声明了一系列函数以及结构体,对于本章节之后需要讨论的部分,需要关注struct context, struct proc等结构体。
struct context在proc.h(40-56)中定义。其结构实际上是五个寄存器的值。也就是在上下文切换时,主要做的事情就是保存并更新寄存器值。同时根据惯例,调用者会保存%eax,%ecx,%edx的值。

struct context {
  uint edi;
  uint esi;
  uint ebx;
  uint ebp;
  uint eip;
};

顺便将proc和pipe的结构也分析一下。
struct proc 在proc.h(60-75)中定义,通过一个结构体记录每个进程的状态。

struct proc {
  uint sz;                     // 进程的内存大小(以byte计)
  pde_t* pgdir;                // 进程页路径的线性地址。
  char *kstack;                // 进程的内核栈底
  enum procstate state;        // 进程状态
  volatile int pid;            // 进程ID
  struct proc *parent;         // 父进程
  struct trapframe *tf;        // 当前系统调用的中断帧
  struct context *context;     // 进程运行的入口
  int killed;                  // 当非0时,表示已结束
  struct file *ofile[NOFILE];  // 打开的文件列表
  struct inode *cwd;           // 进程当前路径
  char name[16];               // 进程名称
};

pipe依赖对结构体spinlock,cpu的定义,见spinlock.h及proc.h(11-24)。
spinlock的作用在于当进程请求得到一个正在被占用的锁时,将进程处于循环检查,等待锁被释放的状态。

struct spinlock {
  uint locked;       // 锁是否处于锁住状态

  // For debugging:
  char *name;        // 锁名称
  struct cpu *cpu;   // 占有该锁的CPU信息
  uint pcs[10];      // 占有该锁的指令栈
};

pipe的结构在pipe.h(12-19)中定义,

struct pipe {
  struct spinlock lock; 
  char data[PIPESIZE];  // 保存pipe的内容,PIPESIZE为512
  uint nread;     // 读取的byte长度
  uint nwrite;    // 写入的byte长度
  int readopen;   // 是否正在读取
  int writeopen;  // 是否正在写入
};
 

2.2 swtch.S

该文件的作用在于使用汇编代码实现了swtch函数,

.globl swtch
swtch:
  # 将需要保存的context地址读取到%esp中,新context地址读取到%edx中
  # 4(%esp)对应的是需要保存的context
  # 8(%esp)对应的是新的context
  movl 4(%esp), %eax
  movl 8(%esp), %edx

  # 将寄存器中过期的数值压栈
  pushl %ebp
  pushl %ebx
  pushl %esi
  pushl %edi

  # 交换栈
  movl %esp, (%eax) # 保存需要保存的context地址
  movl %edx, %esp   # 读取新的context信息

  # 加载新的context信息
  popl %edi
  popl %esi
  popl %ebx
  popl %ebp
  ret
 

3. 进程调度

进程调度的主要函数集中在proc.c中,就让我们从这个文件开始说起吧。
对于单个CPU来说,scheduler是最主要的函数。当CPU初始化之后,即调用scheduler(),循环从进程队列中选择一个进程执行;当进程结束时,将控制权通过swtch()移交给scheduler。

void
scheduler(void)
{
  struct proc *p;

  for(;;){
    // 在每次执行一个进程之前,需要调用sti()函数开启CPU的中断
    sti();

    // 遍历进程表找到一个进程执行
    acquire(&ptable.lock); // 获取进程表的锁,避免其他CPU更改进程表
    for(p = ptable.proc; p < &ptable.proc[NPROC]; p++){
      // 如果进程的状态为不可运行,则略过
      if(p->state != RUNNABLE)
        continue;

      // 切换到选择的进程,释放进程表锁,当进程结束时,再重新获取
      proc = p;
      switchuvm(p);
      p->state = RUNNING;
      swtch(&cpu->scheduler, proc->context);
      switchkvm();

      // Process is done running for now.
      // It should have changed its p->state before coming back.
      proc = 0;
    }
    release(&ptable.lock);
  }
}

在每次Loop之后,都要及时释放进程表锁,这样可以避免当进程表中暂时没有可以运行的程序时,进程表会一直被该CPU锁死,其他CPU便不能访问。其中一种情况是,当进程等待IO时,不是RUNNABLE的,而CPU处于idle状态,一直在占有进程表锁,IO信号无法到达。

sched()切换至CPU context,在切换context之前,进行一系列判断,以避免出现冲突。

void
sched(void)
{
  int intena;

  // 是否获取到了进程表锁
  if(!holding(&ptable.lock))
    panic("sched ptable.lock");
  // 是否执行过pushcli
  if(cpu->ncli != 1)
    panic("sched locks");
  // 执行的程序应该处于结束或者睡眠状态
  if(proc->state == RUNNING)
    panic("sched running");
  // 判断中断是否可以关闭
  if(readeflags()&FL_IF)
    panic("sched interruptible");

  intena = cpu->intena;
  // 上下文切换至scheduler
  swtch(&proc->context, cpu->scheduler);
  cpu->intena = intena;
}

yield()函数将CPU主动让出一个调度周期(scheduling round),这个函数在xv6的当前版本中,仅在trap()中调用,见trap.c(100)。实际应用在于当一个进程正在使用CPU,同时中断处于打开状态,需要查看nlock。

void
yield(void)
{
  // 获取进程表锁
  acquire(&ptable.lock);
  // 将进程状态设为可运行,以便下次遍历时可以被唤醒
  proc->state = RUNNABLE;
  // 执行sched函数,准备将CPU切换到scheduler context
  sched();
  // 释放进程表锁
  release(&ptable.lock);
}

sleep和wakeup是两个互补的函数,共同作用实现改变进程执行顺序,
sleep函数有两个参数 void *chanstruct spinlock *lk

void
sleep(void *chan, struct spinlock *lk)
{
  if(proc == 0)
    panic("sleep");

  if(lk == 0)
    panic("sleep without lk");

  // 释放锁lk
  if(lk != &ptable.lock){  //DOC: sleeplock0
    acquire(&ptable.lock);  //DOC: sleeplock1
    release(lk);
  }

  // 更改状态为SLEEPING,并切换至CPU context
  proc->chan = chan;
  proc->state = SLEEPING;
  sched();

  // Tidy up.
  proc->chan = 0;

  // 重新获得刚刚释放的lk锁
  if(lk != &ptable.lock){  //DOC: sleeplock2
    release(&ptable.lock);
    acquire(lk);
  }
}

值得注意的是,使进程进入睡眠需要两个锁,lk和ptable.lock,由于之前已经得到了ptable.lock,所以wakeup在此期间不会执行,直至进程完全进入睡眠状态,所以lk这个锁可以释放。

wakeup函数的主体部分位于wakeup1函数中。

void
wakeup(void *chan)
{
  // 先获取ptable.lock,确保sleep不会执行,避免出现missed wakeup
  acquire(&ptable.lock);
  wakeup1(chan);
  // 唤醒结束,释放ptable.lock
  release(&ptable.lock);
}

wakeup1函数完成了唤醒的主要工作。wakeup1之所以与wakeup作为两个独立的函数,是因为除了被wakeup调用之外,还在exit中调用,后面会详细讲到。

static void
wakeup1(void *chan)
{
  struct proc *p;
  // 遍历进程表,当发现有符合运行条件的程序时,将其标记为RUNNABLE
  for(p = ptable.proc; p < &ptable.proc[NPROC]; p++)
    if(p->state == SLEEPING && p->chan == chan)
      p->state = RUNNABLE;
}

wait函数用于父进程等待子进程结束,如果没有子进程,则返回-1,否则返回已经结束的子进程的pid。

int
wait(void)
{
  struct proc *p;
  int havekids, pid;
  // 获取进程表锁
  acquire(&ptable.lock);
  for(;;){
    // 遍历查找是否有处于zombie状态的子进程
    havekids = 0;
    for(p = ptable.proc; p < &ptable.proc[NPROC]; p++){
      if(p->parent != proc)
        continue;
      // 如果发现有子进程
      havekids = 1;
      // 如果进程状态为zombie,则将其释放并返回该子进程的pid
      if(p->state == ZOMBIE){
        // Found one.
        pid = p->pid;
        kfree(p->kstack);
        p->kstack = 0;
        freevm(p->pgdir);
        p->state = UNUSED;
        p->pid = 0;
        p->parent = 0;
        p->name[0] = 0;
        p->killed = 0;
        release(&ptable.lock);
        return pid;
      }
    }

    // 如果没有子进程则直接返回
    if(!havekids || proc->killed){
      release(&ptable.lock);
      return -1;
    }

    // 如果有子进程处于睡眠状态,则将父进程置于睡眠状态
    sleep(proc, &ptable.lock);  //DOC: wait-sleep
  }
}

其中,当仍有子进程睡眠时,并没有释放ptable.lock,是因为释放操作放在了sleep函数中,且满足了sleep函数的调用条件,事先获得ptable.lock。

exit()完成了进程结束时的资源释放以及子进程处理等工作。其中只进行了一次acquire操作,这样可以使进程结束的操作原子化;同时可能存在多次的wakeup1操作,这样减少了很多时间。结束后,没有主动调用release,是因为sched进行context switching的时候需要获得ptable.lock,释放在scheduler中进行。

void
exit(void)
{
  struct proc *p;
  int fd;

  if(proc == initproc)
    panic("init exiting");

  // 关闭之前打开的文件
  for(fd = 0; fd < NOFILE; fd++){
    if(proc->ofile[fd]){
      fileclose(proc->ofile[fd]);
      proc->ofile[fd] = 0;
    }
  }
  iput(proc->cwd);
  proc->cwd = 0;

  acquire(&ptable.lock);

  // 唤醒父进程,一边父进程将处于zombie状态的该进程回收
  wakeup1(proc->parent);

  // 将子进程移交给initproc
  for(p = ptable.proc; p < &ptable.proc[NPROC]; p++){
    if(p->parent == proc){
      p->parent = initproc;
      // 如果子进程处于zombie状态,则唤醒其新父亲initproc来料理后事
      if(p->state == ZOMBIE)
        wakeup1(initproc);
    }
  }

  // 移交给scheduler,等待父进程处理
  proc->state = ZOMBIE;
  sched();
  panic("zombie exit");
}
 

4. 管道

xv6中实现管道的结构体pipe已经在前面关于defs.h的分析中提及。此处直接分析pipe.c。

pipealloc实现了pipe的创建,并将pipe关联到两个文件上f0, f1。如果创建成功,返回0;否则返回-1。

int
pipealloc(struct file **f0, struct file **f1)
{
  struct pipe *p;

  p = 0;
  *f0 = *f1 = 0;
  // 如果f0,f1不存在则返回-1
  if((*f0 = filealloc()) == 0 || (*f1 = filealloc()) == 0)
    goto bad;
  if((p = (struct pipe*)kalloc()) == 0)
    goto bad;
  // 
  // 初始化pipe
  p->readopen = 1;
  p->writeopen = 1;
  p->nwrite = 0;
  p->nread = 0;
  initlock(&p->lock, "pipe");
  (*f0)->type = FD_PIPE;
  (*f0)->readable = 1;
  (*f0)->writable = 0;
  (*f0)->pipe = p;
  (*f1)->type = FD_PIPE;
  (*f1)->readable = 0;
  (*f1)->writable = 1;
  (*f1)->pipe = p;
  return 0;

 // 如果创建失败,则将进度回滚,释放占用的内存、解除对文件的占有
 bad:
  if(p)
    kfree((char*)p);
  if(*f0)
    fileclose(*f0);
  if(*f1)
    fileclose(*f1);
  return -1;
}

pipeclose实现了关闭pipe的处理。

void
pipeclose(struct pipe *p, int writable)
{
  // 获取管道锁,避免在关闭的同时进行读写操作
  acquire(&p->lock);
  // 判断是否有未被读取的数据
  if(writable){
    // 如果存在,则唤醒pipe的读进程;否则唤醒写进程
    p->writeopen = 0;
    wakeup(&p->nread);
  } else {
    p->readopen = 0;
    wakeup(&p->nwrite);
  }
  // 当pipe的读写都已结束时,释放资源;否则释放pipe锁
  if(p->readopen == 0 && p->writeopen == 0) {
    release(&p->lock);
    kfree((char*)p);
  } else
    release(&p->lock);
}

pipewrite实现了管道的写操作。

int
pipewrite(struct pipe *p, char *addr, int n)
{
  int i;

  acquire(&p->lock);
  // 逐字节写入
  for(i = 0; i < n; i++){
    // 如果pipe已经写满
    while(p->nwrite == p->nread + PIPESIZE) {  //DOC: pipewrite-full
      // 唤醒读进程,写进程进入睡眠,并返回-1
      if(p->readopen == 0 || proc->killed){
        release(&p->lock);
        return -1;
      }
      wakeup(&p->nread);
      sleep(&p->nwrite, &p->lock);  //DOC: pipewrite-sleep
    }
    p->data[p->nwrite++ % PIPESIZE] = addr[i];
  }
  // 写完之后唤醒读进程
  wakeup(&p->nread);  //DOC: pipewrite-wakeup1
  release(&p->lock);
  return n;
}

piperead实现了pipe的读操作。

int
piperead(struct pipe *p, char *addr, int n)
{
  int i;

  acquire(&p->lock);
  // 如果pipe已经读空,并且正在写入,则进入睡眠状态
  while(p->nread == p->nwrite && p->writeopen){  //DOC: pipe-empty
    if(proc->killed){
      release(&p->lock);
      return -1;
    }
    sleep(&p->nread, &p->lock); //DOC: piperead-sleep
  }
  for(i = 0; i < n; i++){  //DOC: piperead-copy
    if(p->nread == p->nwrite)
      break;
    addr[i] = p->data[p->nread++ % PIPESIZE];
  }
  // 读取完毕,唤醒写进程
  wakeup(&p->nwrite);  //DOC: piperead-wakeup
  release(&p->lock);
  // 返回读取的字节长度
  return i;
}
 

5. 进程调度流程

进程切换:当CPU启动之后,执行scheduler函数,无限循环。在每个周期里,从进程表中找到一个RUNNABLE的进程,切换为进程的上下文,此时开始执行函数。当函数运行结束时,调用return函数,此时切换为CPU的上下文,开始下一循环。
进程唤醒与睡眠:如果一个程序需要等待IO,则CPU会将其设置为睡眠状态,此时不能被执行。当IO信号到达时,执行的进程会将IO信号对应的进程设置为RUNNABLE,即唤醒。下一个scheduler周期的时候,该进程就可能会被执行,处理IO信号。
进程表锁:对于多处理器架构而言,需要用到进程表的时候都需要事先获得表的锁,当结束之后再释放,这样保证了对进程表操作的原子化,可以避免多处理器的竞争问题。

 

6. Pipe实现概述

Pipe的主要部分实际上是一小段规定长度的连续数据存储,读写操作将其视为无限循环长度的内存块。
初始化时,将给定的文件输入、输出流与该结构体关联;关闭时,释放内存,解除文件占用。
读写操作时,分别需要判断是否超出读写的范围,避免覆盖未读数据或者读取已读数据;如果写操作未执行完,则需通过睡眠唤醒的方式来完成大段数据的读取。

 

7. 阅读心得

由于这部分的代码主要由C代码实现,因为相对来说比第一次的阅读任务简单一些。有两个难点,一需要了解依赖的各结构体信息,并通过实际看代码认清每个属性的作用;二需要将多个函数结合着看,才能理解进程表锁的管理机制。xv6的实现机制并不复杂,主观脑洞大开结合着sched.pdf,就比较容易理解。

郑重声明:本站内容如果来自互联网及其他传播媒体,其版权均属原媒体及文章作者所有。转载目的在于传递更多信息及用于网络分享,并不代表本站赞同其观点和对其真实性负责,也不构成任何其他建议。