MIT 6.S081 Lecture Notes

目录

课程网址:https://pdos.csail.mit.edu/6.S081/2020/index.html

我的lab实现的github repo:https://github.com/tommyfan34/MIT_6S081

Lecture 1 Introduction

操作系统应该提供的功能:1. 多进程支持 2. 进程间隔离 3. 受控制的进程间通信

  • xv6:一种在本课程中使用的类UNIX的教学操作系统,运行在RISC-V指令集处理器上,本课程中将使用QEMU模拟器代替

  • kernel(内核):为运行的程序提供服务的一种特殊程序。每个运行着的程序叫做进程,每个进程的内存中存储指令、数据和堆栈。一个计算机可以拥有多个进程,但是只能有一个内核

    每当进程需要调用内核时,它会触发一个system call(系统调用),system call进入内核执行相应的服务然后返回。

  • shell:一个普通的程序,其功能是让用户输入命令并执行它们,shell不是内核的一部分

1.1 Processes and memory

每个进程拥有自己的用户空间内存以及内核空间状态,当进程不再执行时xv6将存储和这些进程相关的CPU寄存器直到下一次运行这些进程。kernel将每一个进程用一个PID(process identifier)指代。

常用syscall

  • fork:形式:int fork()。其作用是让一个进程生成另外一个和这个进程的内存内容相同的子进程。在父进程中,fork的返回值是这个子进程的PID,在子进程中,返回值是0

  • exit:形式:int exit(int status)。让调用它的进程停止执行并且将内存等占用的资源全部释放。需要一个整数形式的状态参数,0代表以正常状态退出,1代表以非正常状态退出

  • wait:形式:int wait(int *status)。等待子进程退出,返回子进程PID,子进程的退出状态存储到int *status这个地址中。如果调用者没有子进程,wait将返回-1

    int pid = fork();
    if (pid > 0) {
        printf("parent: child=%d\n", pid);
        pid = wait((int *) 0);
        printf("child %d is done\n", pid);
    } else if (pid == 0) {
        printf("child: exiting\n");
        exit(0);
    } else {
        printf("fork error\n");
    }
    

    前两行输出可能是

    parent: child=1234
    child: exiting
    

    也可能是

    child: exiting
    parent: child=1234
    

    这是因为在fork了之后,父进程和子进程将同时开始判断PID的值,在父进程中,PID为1234,而在子进程中,PID为0。看哪个进程先判断好PID的值,以上输出顺序才会被决定。

    最后一行输出为

    parent: child 1234 is done
    

    子进程在判断完pid == 0之后将exit,父进程发现子进程exit之后,wait执行完毕,打印输出

    尽管fork了之后子进程和父进程有相同的内存内容,但是内存地址和寄存器是不一样的,也就是说在一个进程中改变变量并不会影响另一个进程。

  • exec:形式:int exec(char *file, char *argv[])。加载一个文件,获取执行它的参数,执行。如果执行错误返回-1,执行成功则不会返回,而是开始从文件入口位置开始执行命令。文件必须是ELF格式。

    xv6 shell使用以上四个system call来为用户执行程序。在shell进程的main中主循环先通过getcmd来从用户获取命令,然后调用fork来运行一个和当前shell进程完全相同的子进程。父进程调用wait等待子进程exec执行完(在runcmd中调用exec

    /* sh.c */
    int
    main(void)
    {
      static char buf[100];
      int fd;
    
      // Ensure that three file descriptors are open.
      while((fd = open("console", O_RDWR)) >= 0){
        if(fd >= 3){
          close(fd);
          break;
        }
      }
    
      // Read and run input commands.
      while(getcmd(buf, sizeof(buf)) >= 0){
        if(buf[0] == 'c' && buf[1] == 'd' && buf[2] == ' '){
          // Chdir must be called by the parent, not the child.
          buf[strlen(buf)-1] = 0;  // chop \n
          if(chdir(buf+3) < 0)
            fprintf(2, "cannot cd %s\n", buf+3);
          continue;
        }
        if(fork1() == 0)
          runcmd(parsecmd(buf));
        wait(0);
      }
      exit(0);
    }
    

1.2 I/O and File descriptors

  • file descriptor:文件描述符,用来表示一个被内核管理的、可以被进程读/写的对象的一个整数,表现形式类似于字节流,通过打开文件、目录、设备等方式获得。一个文件被打开得越早,文件描述符就越小。

    每个进程都拥有自己独立的文件描述符列表,其中0是标准输入,1是标准输出,2是标准错误。shell将保证总是有3个文件描述符是可用的

    while((fd = open("console", O_RDWR)) >= 0){
        if(fd >= 3){
            close(fd);
            break;
        }
    }
    
  • readwrite:形式int write(int fd, char *buf, int n)int read(int fd, char *bf, int n)。从/向文件描述符fd读/写n字节bf的内容,返回值是成功读取/写入的字节数。每个文件描述符有一个offset,read会从这个offset开始读取内容,读完n个字节之后将这个offset后移n个字节,下一个read将从新的offset开始读取字节。write也有类似的offset

    /* essence of cat program */
    char buf[512];
    int n;
    
    for (;;) {
        n = read(0, buf, sizeof buf);
        if (n == 0)
            break;
        if (n < 0){
            fprintf(2, "read errot\n");
            exit(1);
        }
        if (write(1, buf, n) != n){
            fprintf(2, "write error\n");
            exit(1);
        }
    }
    
  • close。形式是int close(int fd),将打开的文件fd释放,使该文件描述符可以被后面的openpipe等其他system call使用。

    使用close来修改file descriptor table能够实现I/O重定向

    /* implementation of I/O redirection,
     * more specifically, cat < input.txt
     */
    char *argv[2];
    argv[0] = "cat";
    argv[1] = 0;
    
    if (fork() == 0) {
        // in the child process
        close(0);  // this step is to release the stdin file descriptor
        open("input.txt", O_RDONLY); // the newly allocated fd for input.txt is 0, since the previous fd 0 is released
        exec("cat", argv); // execute the cat program, by default takes in the fd 0 as input, which is input.txt
    }
    

    父进程的fd table将不会被子进程fd table的变化影响,但是文件中的offset将被共享。

  • dup。形式是int dup(int fd),复制一个新的fd指向的I/O对象,返回这个新fd值,两个I/O对象(文件)的offset相同

    e.g.

    fd = dup(1);
    write(1, "hello ", 6);
    write(fd, "world\n", 6);
    // outputs hello world
    

    除了dupfork之外,其他方式不能使两个I/O对象的offset相同,比如同时open相同的文件

1.3 Pipes

  • pipe:管道,暴露给进程的一对文件描述符,一个文件描述符用来读,另一个文件描述符用来写,将数据从管道的一端写入,将使其能够被从管道的另一端读出

    pipe是一个system call,形式为int pipe(int p[])p[0]为读取的文件描述符,p[1]为写入的文件描述符

    /* run the program wc with stdin connected to the read end of pipe, parent process able to communicate with child process */
    int p[2];
    char *argv[2];
    
    argv[0] = "wc";
    argv[1] = 0;
    
    pipe(p); // read fd put into p[0], write fd put into p[1]
    if (fork() == 0) {
        close(0);
        dup(p[0]); // make the fd 0 refer to the read end of pipe
        close(p[0]); // original read end of pipe is closed
        close(p[1]); // fd p[1] is closed in child process, but not closed in the parent process. 注意这里关闭p[1]非常重要,因为如果不关闭p[1],管道的读取端会一直等待读取,wc就永远也无法等到EOF
        exec("/bin/wc", argv); // by default wc will take fd 0 as the input, which is the read end of pipe in this case
    } else {
        close(p[0]); // close the read end of pipe in parent process will not affect child process
        write(p[1], "hello world\n", 12); 
        close(p[1]); // write end of pipe closed, the pipe shuts down
    }
    

    xv6中的实现和上述的类似

    case PIPE:
    pcmd = (struct pipecmd*)cmd;
    if(pipe(p) < 0)
        panic("pipe");
    if(fork1() == 0){
        // in child process
        close(1); // close stdout
        dup(p[1]); // make the fd 1 as the write end of pipe
        close(p[0]);
        close(p[1]);
        runcmd(pcmd->left); // run command in the left side of pipe |, output redirected to the write end of pipe
    }
    if(fork1() == 0){
        // in child process
        close(0); // close stdin
        dup(p[0]); // make the fd 0 as the read end of pipe
        close(p[0]);
        close(p[1]);
        runcmd(pcmd->right); //  run command in the right side of pipe |, input redirected to the read end of pipe
    }
    close(p[0]);
    close(p[1]);
    wait(0); // wait for child process to finish
    wait(0); // wait for child process to finish
    break;
    

1.4 File system

xv6文件系统包含了文件(byte arrays)和目录(对其他文件和目录的引用)。目录生成了一个树,树从根目录/开始。对于不以/开头的路径,认为是是相对路径

  • mknod:创建设备文件,一个设备文件有一个major device #和一个minor device #用来唯一确定这个设备。当一个进程打开了这个设备文件时,内核会将readwrite的system call重新定向到设备上。

  • 一个文件的名称和文件本身是不一样的,文件本身,也叫inode,可以有多个名字,也叫link,每个link包括了一个文件名和一个对inode的引用。一个inode存储了文件的元数据,包括该文件的类型(file, directory or device)、大小、文件在硬盘中的存储位置以及指向这个inode的link的个数

  • fstat。一个system call,形式为int fstat(int fd, struct stat *st),将inode中的相关信息存储到st中。

  • link。一个system call,将创建一个指向同一个inode的文件名。unlink则是将一个文件名从文件系统中移除,只有当指向这个inode的文件名的数量为0时这个inode以及其存储的文件内容才会被从硬盘上移除

注意:Unix提供了许多在用户层面的程序来执行文件系统相关的操作,比如mkdirlnrm等,而不是将其放在shell或kernel内,这样可以使用户比较方便地在这些程序上进行扩展。但是cd是一个例外,它是在shell程序内构建的,因为它必须要改变这个calling shell本身指向的路径位置,如果是一个和shell平行的程序,那么它必须要调用一个子进程,在子进程里起一个新的shell,再进行cd,这是不符合常理的。

1.5 xv6环境构建

按照https://pdos.csail.mit.edu/6.828/2020/tools.html编译

注意需要follow other linux distributions的教程,因为直接使用ubuntu/debian的教程在调试的时候使用linux自带的gdb会无法远程连接到xv6,必须使用riscv64-unknown-elf-gdb,因此需要自己编译工具链。

debugging步骤:

可参考https://www.cnblogs.com/KatyuMarisaBlog/p/13727565.html

  1. 在xv6文件夹下make qemu-gdb
  2. 另外开一个终端,在相同的xv6文件夹下riscv64-unknown-elf-gdb。进入gdb后输入file user/_[execname]加载可执行文件
  3. 通过b [linenum]设置断点 c继续
  4. make qemu-gdb的终端中执行该可执行文件,进入debugging

可以输入tui enable来边debug边显示source code

可以输入layout asm显示汇编程序,输入layout reg显示寄存器

info breakpoints来显示关于断点的信息

也可以使用VSCode来进行debug,参考https://www.cnblogs.com/KatyuMarisaBlog/p/13727565.html

1.6 Lab 1: Unix Utilities

  1. Sleep

    调用system call sleep来实现休眠一定时间,注意如果没有传入参数,程序需要打印错误信息

    #include "kernel/types.h"
    #include "user/user.h"
    
    int main(int argc, char *argv[]) {
    	if (argc == 1) {
    		fprintf(2, "ERROR: sleep time required\n");
    		exit(1);
    	}
    	sleep(atoi(argv[1])); // atoi把字符串转化为int
    	exit(0);
    }
    
  2. pingpong

    Write a program that uses UNIX system calls to ‘‘ping-pong’’ a byte between two processes over a pair of pipes, one for each direction. The parent should send a byte to the child; the child should print “: received ping”, where is its process ID, write the byte on the pipe to the parent, and exit; the parent should read the byte from the child, print “: received pong”, and exit. Your solution should be in the file user/pingpong.c.

    解决思路:开两个pipe,一个pipe负责子进程写父进程读,另一个pipe负责父进程写子进程读。注意最后要把所有的pipe fd关闭掉。

    #include "kernel/types.h"
    #include "user/user.h"
    
    #define READEND 0
    #define WRITEEND 1
    
    int
    main(int argc, char *argv[]){
    	int p1[2];
    	int p2[2];
    	int pid;
    	char buf[1];
    
    	pipe(p1);
    	pipe(p2);
    
    	pid = fork();
    
    	if (pid < 0)
    		exit(1);
    	else if (pid == 0) {
    	// child process
    		close(p1[WRITEEND]);
    		close(p2[READEND]);
    		read(p1[READEND], buf, 1);
    		printf("%d: received ping\n", getpid());
    		write(p2[WRITEEND], " ", 1);
    		close(p1[READEND]);
    		close(p2[WRITEEND]);
    		exit(0);
    	} else {
    	// parent process
    		close(p1[READEND]);
    		close(p2[WRITEEND]);
    		write(p1[WRITEEND], " ", 1);
    		read(p2[READEND], buf, 1);
    		printf("%d: received pong\n", getpid());
    		close(p1[WRITEEND]);
    		close(p2[READEND]);
    		exit(0);
    	}
    }
    
  3. primes

    素数筛法:将一组数feed到一个进程里,先print出最小的一个数,这是一个素数,然后用其他剩下的数依次尝试整除这个素数,如果可以整除,则将其drop,不能整除则将其feed到下一个进程中,直到最后打印出所有的素数。

    注意最开始的父进程要等待所有子进程exit才能exit

    解决思路:采用递归,每次先尝试从左pipe中读取一个数,如果读不到说明已经到达终点,exit,否则再创建一个右pipe并fork一个子进程,将筛选后的数feed进这个右pipe。

    #include "kernel/types.h"
    #include "user/user.h"
    #define PRIME_NUM 35
    #define READEND 0
    #define WRITEEND 1
    
    void child(int* pl);
    
    int main(int argc, char *argv[]) {
    	int p[2];
    	pipe(p);
    	if (fork() == 0) {
    		child(p);	
    	} else {
    		close(p[READEND]);
    		// feed the int array
    		for (int i=2; i<PRIME_NUM+1; i++) {
    			write(p[WRITEEND], &i, sizeof(int));
    		}
    		close(p[WRITEEND]);
    		wait((int *) 0);
    	}
    	exit(0);
    }
    
    void child(int* pl) {
    	int pr[2];
    	int n;
    
    	close(pl[WRITEEND]);
    	// tries to read the first number
    	int read_result = read(pl[READEND], &n, sizeof(int));
    	if (read_result == 0)
    		exit(0);
    	// right side pipe
    	pipe(pr);
    
    	if (fork() == 0) {
    		child(pr);
    	} else {
    		close(pr[READEND]);
    		printf("prime %d\n", n);
    		int prime = n;
    		while (read(pl[READEND], &n, sizeof(int)) != 0) {
    			if (n%prime != 0) {
    				write(pr[WRITEEND], &n, sizeof(int));
    			}
    		}
    		close(pr[WRITEEND]);
    		wait((int *) 0);
    		exit(0);
    	}
    }
    
  4. find

    要求用递归方式找到指定的文件夹下符合某个名字的文件,参考user/ls.c的实现方法

    #include "kernel/types.h"
    #include "user/user.h"
    #include "kernel/fs.h"
    #include "kernel/stat.h"
    #include "kernel/fcntl.h"
    
    void find(char *path, char *target_file);
    
    int main(int argc, char *argv[]) {
    	if (argc != 3) {
    		fprintf(2, "ERROR: You need pass in only 2 arguments\n");
    		exit(1);
    	}
    	char *target_path = argv[1];
    	char *target_file = argv[2];
    	find(target_path, target_file);
    	exit(0);
    }
    
    void find(char *path, char *target_file) {
    	int fd;
    	struct stat st;
    	struct dirent de;
    	char buf[512], *p;
    
    	if ((fd = open(path, 0)) < 0) {
    		fprintf(2, "ERROR: cannot open %s\n", path);
    		return;
    	}
    
    	if (fstat(fd, &st) < 0) {
    		fprintf(2, "ERROR: cannot stat %s\n", path);
    		close(fd);
    		return;
    	}
    
    	// read the name of each file/folder under the folder specified by fd, which is $path, name is de.name
    	while (read(fd, &de, sizeof(de)) == sizeof(de)) {
    		strcpy(buf, path);
    		p = buf+strlen(buf);
    		*p++ = '/';
    		if (de.inum == 0)
    			continue;
            // get the full path name of the current file/directory selected
    		memmove(p, de.name, DIRSIZ);
    		p[DIRSIZ] = 0;
    
    		if (stat(buf, &st) < 0) {
    			fprintf(2, "ERROR: cannot stat %s\n", buf);
    		}
    
    		switch (st.type) {
    		case T_FILE:
    			if (strcmp(target_file, de.name) == 0) {
    				printf("%s\n", buf);
    			}
    			break;
    		case T_DIR:
    			if ((strcmp(de.name, ".") != 0) && (strcmp(de.name, "..") != 0)) {
    				find(buf, target_file);
    }
    		}	
    	}
    	close(fd);
    	return;
    }
    
  5. xargs

    实现将标准输入作为参数一起输入到xargs后面跟的命令中,比如

    $ echo hello too | xargs echo bye
    bye hello too
    

    如果标准输入有多行,那么也要执行多次命令

    使用fork起一个子进程,在子进程中用exec执行相应的命令。父进程wait。对标准输入每次读一个char,若读到\n需要执行命令。注意在执行xargs这个命令行的时候,最后肯定要按一个回车,这时标准输入最后会有一个回车,所以在EOF前是会有一个回车的!!!

    
    #include "kernel/types.h"
    #include "user/user.h"
    #include "kernel/param.h"
    #define MAX_LEN 100
    
    int main(int argc, char *argv[]) {
    	char *command = argv[1];
    	char bf;
    	char paramv[MAXARG][MAX_LEN]; // arguments
    	char *m[MAXARG];
    
    	while (1) {
    		int count = argc-1;  // # of current arguments
    		memset(paramv, 0, MAXARG * MAX_LEN);
    		// copy the parameter of command
    		for (int i=1; i<argc; i++) {
    			strcpy(paramv[i-1], argv[i]);
    		}
    
    		int cursor = 0; // cursor pointing the char position in single_arg
    		int flag = 0; // flag indicating whether thers is argument preceding space
    		int read_result;
    
    		while (((read_result = read(0, &bf, 1))) > 0 && bf != '\n') {
    			if (bf == ' ' && flag == 1) {
    				count++;
    				// reset flag and p
    				cursor = 0;
    				flag = 0;
    			}
    			else if (bf != ' ') {
    				paramv[count][cursor++] = bf;
    				flag = 1;
    			}
    		}
    		// encounters EOF of input or \n
    		if (read_result <= 0) {
    			break;
    		}
    		for (int i=0; i<MAXARG-1; i++) {
    			m[i] = paramv[i];
    		}
    		m[MAXARG-1] = 0;
    		if (fork() == 0) {
    			exec(command, m);
    			exit(0);
    		} else {
    			wait((int *) 0);
    		}
    	}
    	exit(0);
    }
    

Lecture 2 Operating System Organization

2.0 C pointer

补充资料: K&R《C Programming》

指针:a group of cells (bytes, often two or four) that can hold an address

指针的声明: int *p, 意思是*p是一个int

double *dp, atof(char *)意为声明一个dp,是指向double类型的指针,声明一个函数atof,需要一个指向char变量的指针作为参数,返回值类型为double

*ip += 1是将ip指向的变量加1,相当于++*ip(*ip)++。注意:一元操作符从右向左结合,所以*ip++相当于*(ip++)

指针和数组名的区别:指针是变量,而数组名不是变量,因此pointer = arrayname是合法的,而arrayname = pointer或者arrayname++;是不合法的

指针的大小比较:只有在两个指针指向同一个数组的元素时比较才合法。指针只有相减是合法的

char*char []的区别:前者是指针,并不会管到字符串,后者是数组,需要一个足够大的容量来容纳整个字符串

char amessage[] = "now is the time"; /* an array */
char *message = "now is the time"; /* a pointer, points to the string constant, could not modify the string content */

C程序的内存分配:

  1. 堆(heap):由程序员通过malloc()free()使用和释放,如果忘记释放可能造成内存泄漏。地址从低到高增长
  2. 栈(stack):编译器自动分配释放,存放函数的参数值、局部变量的值等。在退出函数后自动释放销毁。地址从高到低增长
  3. 全局区(static):通过static声明,全局都可以访问,不会在函数退出后释放

2.1 User mode and supervisor mode

为了实现进程隔离,RISC-V CPU在硬件上提供3种执行命令的模式:machine mode, supervisor mode, user mode

  1. machine mode的权限最高,CPU以machine mode启动,machine mode的主要目的是为了配置电脑,之后立即切换到supervisor mode。

  2. supervisor mode运行CPU执行privileged instructions,比如中断管理、对存储页表地址的寄存器进行读写操作、执行system call。运行在supervisor mode也称为在kernel space中运行。

  3. 应用程序只能执行user mode指令,比如改变变量、执行util function。运行在user mode也称为在user space中运行。要想让CPU从user mode切换到supervisor mode,RISC-V提供了一个特殊的ecall指令,要想从supervisor mode切换到user mode,调用sret指令

2.2 Kernel organization

monolithic kernel:整个操作系统在kernel中,所有system call都在supervisor mode下运行。xv6是一个monolithic kernel

micro kernel:将需要运行在supervisor mode下的操作系统代码压到最小,保证kernel内系统的安全性,将大部分的操作系统代码执行在user mode下。

如2.1所示,文件系统是一个user-level的进程,为其他进程提供服务,因此也叫做server

xv6 kernel source file如下所示

2.3 Process

隔离的单元叫做进程,一个进程不能够破坏或者监听另外一个进程的内存、CPU、文件描述符,也不能破坏kernel本身。

为了实现进程隔离,xv6提供了一种机制让程序认为自己拥有一个独立的机器。一个进程为一个程序提供了一个私有的内存系统,或address space,其他的进程不能够读/写这个内存。xv6使用page table(页表)来给每个进程分配自己的address space,页表再将这些address space,也就是进程自己认为的虚拟地址(virtual address)映射到RISC-V实际操作的物理地址(physical address)

虚拟地址从0开始,往上依次是指令、全局变量、栈、堆。RISC-V上的指针是64位的,xv6使用低38位,因此最大的地址是$2^{38}-1$=0x3fffffffff=MAXVA

进程最重要的内核状态:1. 页表 p->pagetable 2. 内核堆栈p->kstack 3. 运行状态p->state,显示进程是否已经被分配、准备运行/正在运行/等待IO或退出

每个进程中都有线程(thread),是执行进程命令的最小单元,可以被暂停和继续

每个进程有两个堆栈:用户堆栈(user stack)和内核堆栈(kernel stack)。当进程在user space中进行时只使用用户堆栈,当进程进入了内核(比如进行了system call)使用内核堆栈

2.4 Starting the first process

RISC-V启动时,先运行一个存储于ROM中的bootloader程序kernel.ld来加载xv6 kernel到内存中,然后在machine模式下从_entry开始运行xv6。bootloader将xv6 kernel加载到0x80000000的物理地址中,因为前面的地址中有I/O设备

_entry中设置了一个初始stack,stack0来让xv6执行kernel/start.c。在start函数先在machine模式下做一些配置,然后通过RISC-V提供的mret指令切换到supervisor mode,使program counter切换到kernel/main.c

main先对一些设备和子系统进行初始化,然后调用kernel/proc.c中定义的userinit来创建第一个用户进程。这个进程执行了一个initcode.S的汇编程序,这个汇编程序调用了exec这个system call来执行/init,重新进入kernel。exec将当前进程的内存和寄存器替换为一个新的程序(/init),当kernel执行完毕exec指定的程序后,回到/init进程。/init(user/init.c)创建了一个新的console device以文件描述符0,1,2打开,然后在console device中开启了一个shell进程,至此整个系统启动了

2.5 Lab 2: System calls

  1. System call tracing

    要求:trace [tracing_mask] [command] 要求当调用了给定的tracing mask所对应的system call时,打印输出调用该system call的进程PID、system call的名称、system call的返回值。已经给出了user space下的user/trace.c,需要注册并实现trace这一system call

    user/user.h中先注册trace这一user function:int trace(int)

    user/usys.pl注册一个ecall到kernel的kernel态的trace的入口:entry("trace");

    kernel/sys_trace.c中实现一个sys_trace(),将system call的传入参数,也就是tracing mask保存到kernel/proc.h中的proc结构体中新注册的int变量中,tracing mask的获取方式为argint(int, int *)

    // trace the sys call from the user space
    uint64
    sys_trace(void)
    {
      int n; // n is the tracing mask
    
      if(argint(0, &n) < 0)  // get the tracing mask
        return -1;
      myproc()->syscallnum = n;  // put the tracing mask into structure
      return 0;  
    }
    

    修改kernel/proc.c中的fork()来把trace mask从父进程复制到子进程

    np->syscallnum = p->syscallnum;
    

    修改kernel/syscall.c中的syscall()来打印输出,注意需要注册一个extern uint 64 sys_trace(void)[SYS_ trace] sys_trace。注意sys call的返回值是保存在进程的trapframe中的a0寄存器的

    void
    syscall(void)
    {
      int num;
      struct proc *p = myproc();
      char* syscall_name[22] = {"fork", "exit", "wait", "pipe", "read", "kill", "exec", "fstat", "chdir", "dup", "getpid", "sbrk", "sleep", "uptime", "open", "write", "mknod", "unlink", "link", "mkdir", "close", "trace"};
    
      num = p->trapframe->a7;
      if(num > 0 && num < NELEM(syscalls) && syscalls[num]) {
        p->trapframe->a0 = syscalls[num]();
        if((1<<num)&(p->syscallnum))
          printf("%d: syscall %s -> %d\n", p->pid, syscall_name[num-1], p->trapframe->a0);
      } else {
        printf("%d %s: unknown sys call %d\n",
                p->pid, p->name, num);
        p->trapframe->a0 = -1;
      }
    }
    
  2. Sysinfo

    要求:写一个sysinfo这个system call,需要获取当前空闲的内存大小填入struct sysinfo.freemem中,获取当前所有不是UNUSED的进程数量填入struct sysinfo.nproc

    和前面一样先要在user/user.h中声明system callint sysinfo(struct sysinfo *)以及struct sysinfo,在user/usys.pl中注册entry,在kernel/syscall.h中注册一个syscall number,在kernel/sysproc.c中对uint64 sys_sysinfo(void)进行实现,注意要在sysproc.c中添加头文件sysinfo.h

    uint64
    sys_sysinfo(void)
    {
      uint64 sysinfop; // address of the sys info structure pointer
      struct sysinfo si;
    
      if(argaddr(0, &sysinfop) < 0)
        return -1;
      si.freemem = freememsize(); // collect the free memory size
      si.nproc = nproc_active(); // collect the number of processes not in use
      if(copyout(myproc()->pagetable, sysinfop, (char *)&si, sizeof(si)) < 0) // copy out the sysinfo struct si from kernel space to user space
        return -1;
      return 0;
    }
    

    kernel/kalloc.c中实现freememsize()

    // get the size of free memory size
    uint64
    freememsize(void)
    {
      uint64 size = 0;
      struct run *r = kmem.freelist;
      uint64 i = 0;
      while(r){
        ++i;
        r = r->next;
      }
      size = i*PGSIZE;
      return size;
    }
    

    kernel/proc.c中实现nproc_active()

    int
    nproc_active()
    {
      int i;
      int num = 0;
      for (i=0; i<NPROC; i++) {
        if (proc[i].state != UNUSED)
          num++;
      }
      return num;
    }
    

Lecture 3 Page Tables

页表让每个进程都拥有自己独立的虚拟内存地址,从而实现内存隔离。

3.1 Paging Hardware

xv6运行于Sv39 RISC-V,即在64位地址中只有最下面的39位被使用作为虚拟地址,其中底12位是页内偏移,高27位是页表索引,即4096字节($2^{12}$)作为一个page,一个进程的虚拟内存可以有 $2^{27}$个page,对应到页表中就是$2^{27}$个page table entry (PTE)。每个PTE有一个44位的physical page number (PPN)用来映射到物理地址上和10位flag,总共需要54位,也就是一个PTE需要8字节存储。即每个物理地址的高44位是页表中存储的PPN,低12位是页内偏移,一个物理地址总共由56位构成。

在实际中,页表并不是作为一个包含了$2^{27}$个PTE的大列表存储在物理内存中的,而是采用了三级树状的形式进行存储,这样可以让页表分散存储。每个页表就是一页。第一级页表是一个4096字节的页,包含了512个PTE(因为每个PTE需要8字节),每个PTE存储了下级页表的页物理地址,第二级列表由512个页构成,第三级列表由512*512个页构成。因为每个进程虚拟地址的高27位用来确定PTE,对应到3级页表就是最高的9位确定一级页表PTE的位置,中间9位确定二级页表PTE的位置,最低9位确定三级页表PTE的位置。如下图所示。第一级根页表的物理页地址存储在satp寄存器中,每个CPU拥有自己独立的satp

PTE flag可以告诉硬件这些相应的虚拟地址怎样被使用,比如PTE_V表明这个PTE是否存在,PTE_RPTE_WPTE_X控制这个页是否允许被读取、写入和执行,PTE_U控制user mode是否有权访问这个页,如果PTE_U=0,则只有supervisor mode有权访问这个页。

3.2 Kernel address space

每个进程有一个页表,用于描述进程的用户地址空间,还有一个内核地址空间(所有进程共享这一个描述内核地址空间的页表)。为了让内核使用物理内存和硬件资源,内核需要按照一定的规则排布内核地址空间,以能够确定哪个虚拟地址对应自己需要的硬件资源地址。用户地址空间不需要也不能够知道这个规则,因为用户空间不允许直接访问这些硬件资源。

QEMU会模拟一个从0x80000000开始的RAM,一直到0x86400000。QEMU会将设备接口以控制寄存器的形式暴露给内核,这些控制寄存器在0x80000000以下。kernel对这些设备接口控制寄存器的访问是直接和这些设备而不是RAM进行交互的。

左边和右边分别是kernel virtual address和physical address的映射关系。在虚拟地址和物理地址中,kernel都位于KERNBASE=0x80000000的位置,这叫做直接映射。

用户空间的地址分配在free memory中

有一些不是直接映射的内核虚拟地址:

  • trampoline page(和user pagetable在同一个虚拟地址,以便在user space和kernel space之间跳转时切换进程仍然能够使用相同的映射,真实的物理地址位于kernel text中的trampoline.S
  • kernel stack page:每个进程有一个自己的内核栈kstack,每个kstack下面有一个没有被映射的guard page,guard page的作用是防止kstack溢出影响其他kstack。当进程运行在内核态时使用内核栈,运行在用户态时使用用户栈。注意:还有一个内核线程,这个线程只运行在内核态,不会使用其他进程的kstack,内核线程没有独立的地址空间。

3.3 Code: creating an address space

xv6中和页表相关的代码在kernel/vm.c中。最主要的结构体是pagetable_t,这是一个指向页表的指针。kvm开头的函数都是和kernel virtual address相关的,uvm开头的函数都是和user virtual address相关的,其他的函数可以用于这两者

几个比较重要的函数:

  • walk:给定一个虚拟地址和一个页表,返回一个PTE指针

  • mappages:给定一个页表、一个虚拟地址和物理地址,创建一个PTE以实现相应的映射

  • kvminit用于创建kernel的页表,使用kvmmap来设置映射

  • kvminithart将kernel的页表的物理地址写入CPU的寄存器satp中,然后CPU就可以用这个kernel页表来翻译地址了

  • procinit(kernel/proc.c)为每一个进程分配(kalloc)kstack。KSTACK会为每个进程生成一个虚拟地址(同时也预留了guard pages),kvmmap将这些虚拟地址对应的PTE映射到物理地址中,然后调用kvminithart来重新把kernel页表加载到satp中去。

每个RISC-V CPU会把PTE缓存到*Translation Look-aside Buffer (TLB)*中,当xv6更改了页表时,必须通知CPU来取消掉当前的TLB,取消当前TLB的函数是sfence.vma(),在kvminithart中被调用

3.4 Physical memory allocation for kernel

xv6对kernel space和PHYSTOP之间的物理空间在运行时进行分配,分配以页(4096 bytes)为单位。分配和释放是通过对空闲页链表进行追踪完成的,分配空间就是将一个页从链表中移除,释放空间就是将一页增加到链表中

kernel的物理空间的分配函数在kernel/kalloc.c中,每个页在链表中的元素是struct run,每个run存储在空闲页本身中。这个空闲页的链表freelist由spin lock保护,包装在struct kmem中。

  • kinit():对分配函数进行初始化,将kernel结尾到PHYSTOP之间的所有空闲空间都添加到kmem链表中,这是通过调用freerange(end, PHYSTOP)实现的
  • freerange()对这个范围内所有页都调用一次kfree来将这个范围内的页添加到freelist链表中

3.5 User space memory

每个进程有自己的用户空间下的虚拟地址,这些虚拟地址由每个进程自己的页表维护,用户空间下的虚拟地址从0到MAXVA

当进程向xv6索要更多用户内存时,xv6先用kalloc来分配物理页,然后向这个进程的页表增加指向这个新的物理页的PTE,同时设置这些PTE的flag

图3.4是一个进程在刚刚被exec调用时的用户空间下的内存地址,stack只有一页,包含了exec调用的命令的参数从而使main(argc, argv)可以被执行。stack下方是一个guard page来检测stack溢出,一旦溢出将会产生一个page fault exception

sbrk是一个可以让进程增加或者缩小用户空间内存的system call。sbrk调用了growproc(kernel/proc.c)来改变p->sz从而改变heap中的program break,growproc调用了uvmallocuvmdealloc,前者调用了kalloc来分配物理内存并且通过mappages向用户页表添加PTE,后者调用了kfree来释放物理内存

3.6 Code: exec

exec是一个system call,为以ELF格式定义的文件系统中的可执行文件创建用户空间。

exec先检查头文件中是否有ELF_MAGIC来判断这个文件是否是一个ELF格式定义的二进制文件,用proc_pagetable来为当前进程创建一个还没有映射的页表,然后用uvmalloc来为每个ELF segment分配物理空间并在页表中建立映射,然后用loadseg来把ELF segment加载到物理空间当中。注意uvmalloc分配的物理内存空间可以比文件本身要大。

接下来exec分配user stack,它仅仅分配一页给stack,通过copyout将传入参数的string放在stack的顶端,在ustack的下方分配一个guard page

如果exec检测到错误,将跳转到bad标签,释放新创建的pagetable并返回-1。exec必须确定新的执行能够成功才会释放进程旧的页表(proc_freepagetable(oldpagetable, oldsz)),否则如果system call不成功,就无法向旧的页表返回-1

3.7 Real world

xv6将kernel加载到0x8000000这一RAM物理地址中,但是实际上很多RAM的物理地址都是随机的,并不一定存在0x8000000这个地址

实际的处理器并不一定以4096bytes为一页,而可能使用各种不同大小的页

3.8 Lab 3: pgtbl

  1. print a page table

    实验要求:在程序刚刚启动(pid==1)时打印输出当前进程的pagetable。在exec.c中添加

    if(p->pid==1) vmprint(p->pagetable);
    

    vmprint()函数放在kernel/vm.c

    void
    _helper_vmprint(pagetable_t pagetable, int level)
    {
      for(int i = 0; i < 512; i++){
        pte_t pte = pagetable[i];
        if ((pte & PTE_V) && (pte & (PTE_R | PTE_W | PTE_X)) == 0) {
          // this pte points to a valid lower level page table
          uint64 child = PTE2PA(pte);
          for (int j=0; j <= level; j++) {
            printf("..");
            if ((j+1) <= level) {
              printf(" ");
            }
          }
          printf("%d: pte %p pa %p\n", i, pte, child);
          _helper_vmprint((pagetable_t)child, level+1);
        }
        else if (pte & PTE_V) {
          uint64 child = PTE2PA(pte);
          // the lowest valid page table
          printf(".. .. ..%d: pte %p pa %p\n", i, pte, child);
        }
      }
    }
    
    // print the page table
    void vmprint(pagetable_t pagetable)
    {
      printf("page table %p\n", pagetable);
      _helper_vmprint(pagetable, 0);
    }
    
  2. A kernel page table per process

    实验要求:为每个进程添加一个kernel pagetable来取代之前的global page table

    kernel/proc.h中的struct proc添加一个每个进程的kernel pagetable

    pagetable_t kernelpt;        // per process kernel pagetable
    

    仿照kvminit,在vm.c中实现一个proc_kpt_init()函数来为每个进程初始化kernel page table,其中uvmmap类似于kvmmap,只不过kvmmap是直接对全局的kernel_pagetable进行mappage,而uvmmap并没有指定page table

    pagetable_t
    proc_kpt_init()
    {
      pagetable_t kpt;
      kpt = uvmcreate();
      if (kpt == 0) return 0;
      uvmmap(kpt, UART0, UART0, PGSIZE, PTE_R | PTE_W);
      uvmmap(kpt, VIRTIO0, VIRTIO0, PGSIZE, PTE_R | PTE_W);
      uvmmap(kpt, CLINT, CLINT, 0x10000, PTE_R | PTE_W);
      uvmmap(kpt, PLIC, PLIC, 0x400000, PTE_R | PTE_W);
      uvmmap(kpt, KERNBASE, KERNBASE, (uint64)etext-KERNBASE, PTE_R | PTE_X);
      uvmmap(kpt, (uint64)etext, (uint64)etext, PHYSTOP-(uint64)etext, PTE_R | PTE_W);
      uvmmap(kpt, TRAMPOLINE, (uint64)trampoline, PGSIZE, PTE_R | PTE_X);
      return kpt;
    }
    void 
    uvmmap(pagetable_t pagetable, uint64 va, uint64 pa, uint64 sz, int perm)
    {
      if(mappages(pagetable, va, sz, pa, perm) != 0)
        panic("kvmmap");
    }
    

    allocproc()中调用初始化进程内核页表的函数

    p->kernelpt = proc_kpt_init();
    

    注意要让每个procees kernel pagetable拥有这个进程的kernel stack,因此在allocproc中分配kstack并将其map到p->kernelpt

    // Allocate a page for the process's kernel stack.
    // Map it high in memory, followed by an invalid
    // guard page.
    char *pa = kalloc();
    if(pa == 0)
        panic("kalloc");
    uint64 va = KSTACK((int) (p - proc));
    uvmmap(p->kernelpt, va, (uint64)pa, PGSIZE, PTE_R | PTE_W);
    p->kstack = va;
    

    修改scheduler()以将进程的kernel page table加载到satp寄存器中,如果CPU空闲,就使用global kernel page table

            w_satp(MAKE_SATP(p->kernelpt));
            sfence_vma();
    
            swtch(&c->context, &p->context);
    
            // Process is done running for now.
            // It should have changed its p->state before coming back.
            c->proc = 0;
    
          	w_satp(MAKE_SATP(kernel_pagetable));
          	sfence_vma();
            found = 1;
          }
    

    freeproc中释放掉process kernel page table,注意对于KERNBASE以下的部分,这个process kernel page table只能unmap这些PTE,而不能真的释放这些物理资源,因为KERNBASE以下的部分其他进程是共享的,因此要先释放掉进程的kstack物理资源,然后用freewalk_kproc这个函数将pagetable unmap和释放的同时也保留叶节点指向的物理资源

    // free the kernel stack in the RAM
    if (p->kstack)
    {
        pte_t* pte = walk(p->kernelpt, p->kstack, 0);
        if (pte == 0)
            panic("freeproc: walk");
        kfree((void*)PTE2PA(*pte));
    }
    p->kstack = 0;
    
    // ...
    if (p->kernelpt)
        freewalk_kproc(p->kernelpt);
    // ...
    
    // Recursively free page-table pages
    // but retain leaf physical addresses
    void
    freewalk_kproc(pagetable_t pagetable) {
      for(int i = 0; i < 512; i++){
        pte_t pte = pagetable[i];
        if((pte & PTE_V)){
          pagetable[i] = 0;
          if ((pte & (PTE_R|PTE_W|PTE_X)) == 0)
          {
            uint64 child = PTE2PA(pte);
            freewalk_kproc((pagetable_t)child);
          }
        }
      }
      kfree((void*)pagetable);
    }
    
  3. Simplify copyin/copyinstr

    实验要求:将每个进程的user page table复制到进程kernel page table上,从而让每个进程在copyin的时候不需要再利用process user page table来翻译传入的参数指针,而可以直接利用process kernel page table来dereference

    先实现一个复制page table的函数u2kvmcopy来将user page table复制到process kernel page table,注意在复制的过程中需要清除原先PTE中的PTE_U标志位,否则kernel无法访问

    // copy the user page table to kernel page table
    void
    u2kvmcopy(pagetable_t pagetable, pagetable_t kpagetable, uint64 oldsz, uint64 newsz)
    {
      pte_t *pte_from, *pte_to;
      uint64 a, pa;
      uint flags;
    
      if (newsz < oldsz)
        return;
      oldsz = PGROUNDUP(oldsz);
      for (a = oldsz; a < newsz; a += PGSIZE)
      {
        if ((pte_from = walk(pagetable, a, 0)) == 0)
          panic("u2kvmcopy: pte should exist");
        if ((pte_to = walk(kpagetable, a, 1)) == 0)  // copy the pte to the same address at kernel page table
          panic("u2kvmcopy: walk fails");
        pa = PTE2PA(*pte_from);
        flags = (PTE_FLAGS(*pte_from) & (~PTE_U));
        *pte_to = PA2PTE(pa) | flags;
      }
    }
    

    在每个改变了user page table的地方要调用u2kvmcopy()使得process kernel page table跟上这个变化。这些地方出现在fork()exec()sbrk()需要调用的growproc()。注意要防止user process过大导致virtual address超过PLIC

    // fork
    // increment reference counts on open file descriptors.
    for(i = 0; i < NOFILE; i++)
        if(p->ofile[i])
            np->ofile[i] = filedup(p->ofile[i]);
    np->cwd = idup(p->cwd);
    
    u2kvmcopy(np->pagetable, np->kernelpt, 0, np->sz);
    
    safestrcpy(np->name, p->name, sizeof(p->name));
    
    // exec
    // call u2kvmvcopy after user stack is set up
    stackbase = sp - PGSIZE;
    
    u2kvmcopy(pagetable, p->kernelpt, 0, sz);
    // Push argument strings, prepare rest of stack in ustack.
    for(argc = 0; argv[argc]; argc++) {
    
    // growproc
    int
    growproc(int n)
    {
      uint sz;
      struct proc *p = myproc();
    
      sz = p->sz;
    
      if(n > 0){
        //   return -1;
        if((sz = uvmalloc(p->pagetable, sz, sz + n)) == 0) {
          return -1;
        }
        u2kvmcopy(p->pagetable, p->kernelpt, sz-n, sz);
      } else if(n < 0){
        sz = uvmdealloc(p->pagetable, sz, sz + n);
      }
      p->sz = sz;
      return 0;
    }
    
    // prevent virtual address exceeding PLIC
    // in exec.c when loading the program
    uint64 sz1;
    if((sz1 = uvmalloc(pagetable, sz, ph.vaddr + ph.memsz)) == 0)
        goto bad;
    if(sz1 >= PLIC)
        goto bad;
    sz = sz1;
    

    userinit中复制process kernel page

    uvminit(p->pagetable, initcode, sizeof(initcode));
    p->sz = PGSIZE;
    u2kvmcopy(p->pagetable, p->kernelpt, 0, p->sz);
    // prepare for the very first "return" from kernel to user.
    p->trapframe->epc = 0;      // user program counter
    

    最后将copyin()copyinstr()替换为copyin_new()copyinstr_new()

Lecture 4 RISC-V calling convention

4.1 ISA & Assembly Language

ISA: Instruction Set

C -> Assembly(.S/.asm) -> binary (object.o)

汇编语言没有明确的workflow,只是一行行指令

汇编语言是基于寄存器进行操作的,而不是基于内存操作

RISC-V vs x86:

  • RISC-V:精简指令集,指令更少,更加简单,唯一开源的ISA。ARM也是RISC(Reduced Instruction Set Chip)
  • x86:复杂指令集,指令很多并且可以实现复杂功能

RISC-V assembly常用指令:

https://web.eecs.utk.edu/~smarz1/courses/ece356/notes/assembly/

  • ld/lb/lw rd, 8(rs):将*(rs+8)的值写入到rd寄存器,lb=load byte, ld=load double word, lw=load word
  • sd/sb/sw rd, 8(rs):将*(rd)`的值写入到rs+8地址上
  • add rd, rs1, rs2:*(rd) = *(rs1) + *(rs2)
  • addi rd, rs1, int: *(rd) = *(rs1) + int

4.2 Calling convention

调用约定(calling convention)是规定子过程如何获取参数以及如何返回的方案,调用约定一般规定了

  • 参数、返回值、返回地址等放置的位置(寄存器、栈或存储器)

    RISC-V寄存器通过寄存器而非栈来传递函数参数,a0-a7是int参数,fa0-fa7是float参数

  • 如何将调用子过程的准备工作与恢复现场的工作划分到调用者(caller)与被调用者(callee)身上

小于一个指针字(RISCV64中是8字节,RISCV32是4字节)的参数传入时将参数放在寄存器的最低位,因为RISC-V是小端系统,当2个指针字的参数传入时,低位的1个指针字放在偶数寄存器,比如a0上,高位的1个指针字放在奇数寄存器,比如a1上。当高于2个指针字的参数传入时以引用的方式传入。struct参数没有传到寄存器的部分将以栈的方式传入,sp栈指针将指向第一个没有传入到寄存器的参数。

从函数返回的值,如果是整数将放在a0和a1中,如果是小数将放置在fa0和fa1寄存器中。对于更大的返回值,将放置在内存中,caller开辟这个内存,并且把指向这个内存的指针作为第一个参数传递给callee

由caller保存的寄存器不会在函数调用之间被保存,又名易失性寄存器,如果要在过程调用后恢复该值,则调用方有责任将这些寄存器压入堆栈或复制到其他位置,而callee保存的寄存器会被保存,称为非易失性寄存器,可以期望这些寄存器在被调用者返回后保持相同的值。比如函数A调用了函数B,所有函数A保存的寄存器在函数B被调用后可以被B重写覆盖

4.3 Stack

栈从高地址向低地址增长,每个大的box叫一个stack frame(栈帧),栈帧由函数调用来分配,每个栈帧大小不一定一样,但是栈帧的最高处一定是return address

sp是stack pointer,用于指向栈顶(低地址),保存在寄存器中

fp是frame pointer,用于指向当前帧底部(高地址),保存在寄存器中,同时每个函数栈帧中保存了调用当前函数的函数(父函数)的fp(保存在to prev frame那一栏中)

这些栈帧都是由编译器编译生成的汇编文件生成的

Lecture 5 Traps

5.0 trap

3种可能的情况使得CPU暂停对正常指令的执行:1. syscall,移交给kernel 2. exception,指令执行了非法操作 3. 设备中断。以上情况合并称为trap

trap应该对于被打断的指令是透明的,也就是说被打断的指令不应该知道这个地方产生了trap,产生trap之后现场应该得以恢复并继续执行被打断的指令。

xv6对trap的处理分为四个阶段:1. RISC-V CPU的硬件的一些动作 2. 汇编文件为了kernel C文件进行的一些准备 3. 用C实现的trap handler 4. system call / device-driver service routine

通常对于user space的trap、kernel space的trap和计时器中断会有不同的trap handler

5.1 RISC-V trap machinery

RISC-V CPU有一系列的控制寄存器可以通知kernel发生了trap,也可以由kernel写入来告诉CPU怎样处理trap

  • stvec:trap handler的地址,由kernel写入

  • sepc:保存trap发生时的现场program counter,因为接下来pc要被取代为stvecsret是从trap回到现场的指令,将sepc写回到pc

  • scause:一个trap产生的原因代码,由CPU写入

  • sscratch:放在trap handler的最开始处

  • sstatus:控制设备中断是否被开启,如果sstatus中的SIE位被清除,则RISC-V将推迟设备中断。SPP位指示这个trap是在user space中产生的还是在kernel space产生的,并将控制sret回到什么模式

    以上寄存器只在supervisor模式下发生的trap被使用

当发生除了计时器中断以外的其他类型的trap时,RISC-V将执行以下步骤:

  1. 如果trap是一个设备产生的中断,而SIE又被清除的情况下,不做下方的任何动作
  2. 清除SIE来disable一切中断
  3. pc复制到sepc
  4. 把当前的模式(user / supervisor)保存到SPP
  5. 设置scause寄存器来指示产生trap的原因
  6. 将当前的模式设置为supervisor
  7. stvec的值复制到pc
  8. 开始执行pc指向的trap handler的代码

注意CPU并没有切换到kernel页表,也没有切换到kernel栈

5.2 Traps from user space

当user space中发生trap时,会将stvec的值复制到pc,而此时stvec的值是trampoline.S中的uservec,因此跳转到uservec,先保存一些现场的寄存器,恢复kernel栈指针、kernel page table到satp寄存器,再跳转到usertrap(kernel/trap.c)trap handler,然后返回usertrapret(kernel/trap.c),跳回到kernel/trampoline.S,最后用userret(kernel/trampoline.S)通过sret跳回到user space

RISC-V在trap中不会改变页表,因此user page table必须有对uservec的mapping,uservecstvec指向的trap vector instruction。uservec要切换satp到kernel页表,同时kernel页表中也要有和user页表中对uservec相同的映射。RISC-V将uservec保存在trampoline页中,并将TRAMPOLINE放在kernel页表和user页表的相同位置处(MAXVA)

uservec开始时所有的32个寄存器都是trap前代码的值,但是uservec需要对某些寄存器进行修改来设置satp,可以用sscratcha0的值进行交换,交换之前的sscratch中是指向user process的trapframe的地址,trapframe中预留了保存所有32个寄存器的空间。p->trapframe保存了每个进程的TRAPFRAME的物理空间从而让kernel页表也可以访问该进程的trapframe

当交换完a0sscratch之后,uservec可以通过a0把所有当前寄存器的值保存到trapframe中。由于当前进程的trapframe已经保存了当前进程的kernel stack、当前CPU的hartid、usertrap的地址、kernel page table的地址等,uservec需要获取这些值,然后切换到kernel pagetable,调用usertrap

usertrap主要是判断trap产生的原因并进行处理,然后返回。因为当前已经在kernel里了,所以这时候如果再发生trap,应该交给kernelvec处理,因此要把stvec切换为kernelvec。如果trap是一个system call,那么syscall将被调用,如果是设备中断,调用devintr,否则就是一个exception,kernel将杀死这个出现错误的进程

回到user space的第一步是调用usertrapret(),这个函数将把stvec指向uservec,从而当回到user space再出现trap的时候可以跳转到uservec,同时设置p->trapframe的一些值为下一次trap作准备,比如设置p->trapframe->kernel_sp = p->kstack + PGSIZE。清除SPP为从而使得调用sret后能够回到user mode。设置回到user space后的program counter为p->trapframe->epc,最后调用跳转到TRAMPOLINE页上的userret回到trampoline.S,加载user page table。userretuserrapret调用返回时a0寄存器中保存了TRAPFRAME,因此可以通过这个TRAPFRAME地址来恢复之前所有寄存器的值(包括a0),最后把TRAPFRAME保存在sscratch中,用sret回到user space

5.3 Calling system calls

user调用exec执行system call的过程:把给exec使用的参数放到a0和a1寄存器中,把system call的代码(SYS_exec)放到a7寄存器中,ecall指令将陷入内核中(通过usys.pl中的entry)。ecall的效果有三个,包括将CPU从user mode切换到supervisor mode、将pc保存到epc以供后续恢复、将uservec设置为stvec,并执行uservecusertrap,然后执行syscall。kernel trap code将把寄存器的值保存在当前进程的trapframe中。syscall将把trapframe中的a7寄存器保存的值提取出来,索引到syscalls这个函数数列中查找对应的syscall种类,并进行调用,然后把返回值放置在p->trapframe->a0中,如果执行失败,就返回-1。

syscall的argument可以用argintargaddrargfd等函数从内存中取出

5.4 Traps from kernel space

当执行kernel code发生CPU trap的时候,stvec是指向kernelvec的汇编代码的。kernelvec将寄存器的值保存在被中断的kernel thread的栈里而不是trapframe里,这样当trap需要切换kernel thread时,再切回来之后还可以从原先的thread栈里找到之前的寄存器值。

保存完寄存器之后,跳转到kerneltrap这个trap handler。kerneltrap可以对设备中断和exception这两种trap进行处理。如果是设备中断,调用devintr进行处理,如果是exception就panic,如果是因为计时器中断,就调用yield让其他kernel thread运行

最后返回到kernelvec中,kernelvec将保存的寄存器值从堆栈中弹出,执行sret,将sepc复制到pc来执行之前被打断的kernel code

5.5 Lab 4: Traps

  1. Backtrace

    实验要求:当调用sys_sleep时打印出函数调用关系的backtrace,即递归地打印每一个函数以及调用这个函数及其父函数的return address,比如

    backtrace:
    0x0000000080002cda
    0x0000000080002bb6
    0x0000000080002898
    

    gcc将当前执行的函数的frame pointer存储在s0寄存器中,可以在kernel/riscv.h中声明以下函数来获取s0

    static inline uint64
    r_fp()
    {
      uint64 x;
      asm volatile("mv %0, s0" : "=r" (x) );
      return x;
    }
    

    当前函数的return address位于fp-8的位置,previous frame pointer位于fp-16的位置。

    每个kernel stack都是一页,因此可以通过计算PGROUNDDOWN(fp)PGROUNDUP(fp)来确定当前的fp地址是否还位于这一页内,从而可以是否已经完成了所有嵌套的函数调用的backtrace。

    在kernel/printf.c中

    void
    backtrace(void)
    {
      uint64 fp, ra, top, bottom;
      printf("backtrace:\n");
      fp = r_fp();  // frame pointer of currently executing function
      top = PGROUNDUP(fp);
      bottom = PGROUNDDOWN(fp);
      while (fp < top && fp > bottom)
      {
        ra = *(uint64 *)(fp-8);
        printf("%p\n", ra);
        fp = *(uint64 *)(fp-16); // saved previous frame pointer
      }
    }
    
  2. Alarm

    实验要求:添加一个新的system callsigalarm(interval, handler)interval是一个计时器的tick的间隔大小,handler是指向一个函数的指针,这个函数是当计时器tick到达interval时触发的函数。

    首先修改Makefile使得alarmtest.c能够被编译,在user/user.h中添加函数声明

    int sigalarm(int ticks, void (*handler)());
    int sigreturn(void);
    

    在user/user.pl、kernel/syscall.h、kernel/syscall.c等添加sys_sigalarmsys_sigreturn这两个syscall的注册

    在kernel/proc.h的proc结构体中添加几个成员。int interval是保存的定时器触发的周期,void (*handler)()是指向handler函数的指针,这两个成员变量在sys_sigalarm中被保存。ticks用来记录从上一次定时器被触发之后到目前的ticks,in_handler用来记录当前是否在handler函数中,用来防止在handler函数的过程中定时被触发再次进入handler函数。下面的所有寄存器都是用来保护和恢复现场的。

      int interval;                // interval of alarm
      void (*handler)();           // pointer to the handler function
      int ticks;                   // how many ticks have passed since last call
      int in_handler;              // to prevent from reentering into handler when in handler
      uint64 saved_epc;
      uint64 saved_ra;
      uint64 saved_sp;
      uint64 saved_gp;
      uint64 saved_tp;
      uint64 saved_t0;
      uint64 saved_t1; 
      uint64 saved_t2;
      uint64 saved_s0;
      uint64 saved_s1;
      uint64 saved_s2;
      uint64 saved_s3;
      uint64 saved_s4;
      uint64 saved_s5;
      uint64 saved_s6;
      uint64 saved_s7;
      uint64 saved_s8;
      uint64 saved_s9;
      uint64 saved_s10;
      uint64 saved_s11;
      uint64 saved_a0;
      uint64 saved_a1;
      uint64 saved_a2;
      uint64 saved_a3;
      uint64 saved_a4;
      uint64 saved_a5;
      uint64 saved_a6;
      uint64 saved_a7;
      uint64 saved_t3;
      uint64 saved_t4;
      uint64 saved_t5;
      uint64 saved_t6;
    

    在kernel/sysfile.c中分别添加uint64 sys_sigreturn(void)uint64 sys_sigalarm(void)的实现

    对于sys_sigalarm,需要将从userfunction传入的两个参数分别保存到p结构体相应的成员变量中

    uint64
    sys_sigalarm(void)
    {
      int ticks;
      uint64 funaddr; // pointer to function
      if ((argint(0, &ticks) < 0) || (argaddr(1, &funaddr) < 0)) {
        return -1;
      }
      struct proc *p = myproc();
      p->interval = ticks;
      p->handler = (void(*)())funaddr;
      return 0;
    }
    

    对于sys_sigreturn,这个函数会在handler函数完成之后被调用,它的主要工作是恢复现场的寄存器,并且将in_handler这个flag置0

    uint64
    sys_sigreturn(void)
    {
      struct proc *p = myproc();
      p->trapframe->epc = p->saved_epc;
      p->trapframe->ra = p->saved_ra;
      p->trapframe->sp = p->saved_sp;
      p->trapframe->gp = p->saved_gp;
      p->trapframe->tp = p->saved_tp;
      p->trapframe->t0 = p->saved_t0;
      p->trapframe->t1 = p->saved_t1;
      p->trapframe->t2 = p->saved_t2;
      p->trapframe->t3 = p->saved_t3;
      p->trapframe->t4 = p->saved_t4;
      p->trapframe->t5 = p->saved_t5;
      p->trapframe->t6 = p->saved_t6;
      p->trapframe->s0 = p->saved_s0;
      p->trapframe->s1 = p->saved_s1;
      p->trapframe->s2 = p->saved_s2;
      p->trapframe->s3 = p->saved_s3;
      p->trapframe->s4 = p->saved_s4;
      p->trapframe->s5 = p->saved_s5;
      p->trapframe->s6 = p->saved_s6;
      p->trapframe->s7 = p->saved_s7;
      p->trapframe->s8 = p->saved_s8;
      p->trapframe->s9 = p->saved_s9;
      p->trapframe->s10 = p->saved_s10;
      p->trapframe->s11 = p->saved_s11;
      p->trapframe->a0 = p->saved_a0;
      p->trapframe->a1 = p->saved_a1;
      p->trapframe->a2 = p->saved_a2;
      p->trapframe->a3 = p->saved_a3;
      p->trapframe->a4 = p->saved_a4;
      p->trapframe->a5 = p->saved_a5;
      p->trapframe->a6 = p->saved_a6;
      p->trapframe->a7 = p->saved_a7;
      p->in_handler = 0;
      return 0;
    }
    

    kernel/trap.c中的usertrap()是对陷入定时器中断的处理,需要判断which_dev==2才是定时器中断,当p->ticks到达预设值p->interval时,将调用p->handler,这是通过向p->trapframe->epc加载handler地址实现的,这样当从usertrap()中返回时,pc在恢复现场时会加载p->trapframe->epc,这样就会跳转到handler地址。在跳转到handler之前先要保存p->trapframe的寄存器,因为执行handler函数会导致这些寄存器被覆盖。

     syscall();
      } else if((which_dev = devintr()) != 0){
        // ok
        if (which_dev == 2 && p->in_handler == 0) {
          p->ticks += 1;
          if ((p->ticks == p->interval) && (p->interval != 0)) {
            p->in_handler = 1;
            p->ticks = 0;
            p->saved_epc = p->trapframe->epc;
            p->saved_ra = p->trapframe->ra;
            p->saved_sp = p->trapframe->sp;
            p->saved_gp = p->trapframe->gp;
            p->saved_tp = p->trapframe->tp;
            p->saved_t0 = p->trapframe->t0;
            p->saved_t1 = p->trapframe->t1;
            p->saved_t2 = p->trapframe->t2;
            p->saved_t3 = p->trapframe->t3;
            p->saved_t4 = p->trapframe->t4;
            p->saved_t5 = p->trapframe->t5;
            p->saved_t6 = p->trapframe->t6;
            p->saved_s0 = p->trapframe->s0;
            p->saved_s1 = p->trapframe->s1;
            p->saved_s2 = p->trapframe->s2;
            p->saved_s3 = p->trapframe->s3;
            p->saved_s4 = p->trapframe->s4;
            p->saved_s5 = p->trapframe->s5;
            p->saved_s6 = p->trapframe->s6;
            p->saved_s7 = p->trapframe->s7;
            p->saved_s8 = p->trapframe->s8;
            p->saved_s9 = p->trapframe->s9;
            p->saved_s10 = p->trapframe->s10;
            p->saved_s11 = p->trapframe->s11;
            p->saved_a0 = p->trapframe->a0;
            p->saved_a1 = p->trapframe->a1;
            p->saved_a2 = p->trapframe->a2;
            p->saved_a3 = p->trapframe->a3;
            p->saved_a4 = p->trapframe->a4;
            p->saved_a5 = p->trapframe->a5;
            p->saved_a6 = p->trapframe->a6;
            p->saved_a7 = p->trapframe->a7;
            p->trapframe->epc = (uint64)p->handler;
          }
        }
      } else {
    

Lecture 6 Page Faults

当试图访问PTE_V为0的虚拟地址或user访问PTE_U为0/kernel访问PTE_U为1以及其他违反PTE_W/PTE_R等flag的情况下会出现page faults。Page faults是一个exception,总共有3种page faults:

  1. load page faults:当load instruction无法翻译虚拟地址时发生
  2. store page faults:当store instruction无法翻译虚拟地址时发生
  3. instruction page faults:当一个instruction的地址无法翻译时发生

page faults种类的代码存放在scause寄存器中,无法翻译的地址存放在stval寄存器中。

在xv6中对于exception一律都会将这个进程kill掉,但是实际上可以结合page faults实现一些功能。

  1. 可以实现copy-on-write fork。在fork时,一般都是将父进程的所有user memory复制到子进程中,但是fork之后一般会直接进行exec,这就会导致复制过来的user memory又被放弃掉。因此改进的思路是:子进程和父进程共享一个物理内存,但是mapping时将PTE_W置零,只有当子进程或者父进程的其中一个进程需要向这个地址写入时产生page fault,此时才会进行copy
  2. 可以实现lazy allocation。旧的sbrk()申请分配内存,但是申请的这些内存进程很可能不会全部用到,因此改进方案为:当进程调用sbrk()时,将修改p->sz,但是并不实际分配内存,并且将PTE_V置0。当在试图访问这些新的地址时发生page fault再进行物理内存的分配
  3. paging from disk:当内存没有足够的物理空间时,可以先将数据存储在其他的存储介质(比如硬盘)上,,将该地址的PTE设置为invalid,使其成为一个evicted page。当需要读或者写这个PTE时,产生Page fault,然后在内存上分配一个物理地址,将这个硬盘上的evicted page的内容写入到该内存上,设置PTE为valid并且引用到这个内存物理地址

6.1 Lab 5: lazy allocation

  1. 实验要求:在sbrk()时只增长进程的myproc()->sz而不实际分配内存。在kernel/trap.c中修改从而在产生page fault时分配物理内存给相应的虚拟地址。相应的虚拟地址可以通过r_stval()获得。r_scause()为13或15表明trap的原因是page fault

    kernel/sysproc.c中修改sys_sbrk()

    uint64
    sys_sbrk(void)
    {
      int addr;
      int n;
      if(argint(0, &n) < 0)
        return -1;
      addr = myproc()->sz;
      myproc()->sz += n;
      // if(growproc(n) < 0)
      //   return -1;
      return addr;
    }
    

    kernel/trap.c中,在usertrap()函数中

     } else if((which_dev = devintr()) != 0){
        // ok
      } else {
        if (r_scause() == 13 || r_scause() == 15) {
          // page fault
          uint64 va = r_stval();
          char *mem;
          va = PGROUNDDOWN(va);
          if ((mem = kalloc()) == 0) {
    		panic("cannot allocate for lazy alloc\n");
            exit(-1);
          } 
          if (mappages(p->pagetable, va, PGSIZE, (uint64)mem, PTE_W|PTE_X|PTE_R|PTE_U) != 0) {
            kfree(mem);
    		panic("cannot map for lazy alloc\n");
    		exit(-1);
          }
        }
        else {
          printf("usertrap(): unexpected scause %p pid=%d\n", r_scause(), p->pid);
          printf("            sepc=%p stval=%p\n", r_sepc(), r_stval());
          p->killed = 1;
        }
      }
    

    修改kernel/vm.c中的uvmunmap,因为lazy allocation可能会造成myproc()-sz以下的内容没有被分配的情况,因此在unmap的过程中可能会出现panic,这是正常情况,需要continue

    if((pte = walk(pagetable, a, 0)) == 0)
        continue;
    // panic("uvmunmap: walk");
    if((*pte & PTE_V) == 0)
        continue;	    
    // panic("uvmunmap: not mapped");
    if(PTE_FLAGS(*pte) == PTE_V)
        panic("uvmunmap: not a leaf");
    if(do_free){
    
  2. 实验要求: 在第一部分的基础上,要求处理

  • sbrk()的参数为负的情况,deallocate即可

    if (n < 0) {
        // deallocate the memory
        if ((myproc()->sz + n) < 0) {
            return -1;
        } else {
            if (uvmdealloc(myproc()->pagetable, addr, addr+n) != (addr+n)) {
                return -1;
            }
        }
    }
    
  • fork()中将父进程的内存复制给子进程的过程中用到了uvmcopyuvmcopy原本在发现缺失相应的PTE等情况下会panic,这里也要continue掉。在kernel/proc.c的uvmcopy

    if((pte = walk(old, i, 0)) == 0)
        continue;
    // panic("uvmcopy: pte should exist");
    if((*pte & PTE_V) == 0)
        continue;
    // panic("uvmcopy: page not present");
    
  • 当造成的page fault在进程的user stack以下(栈底)或者在p->sz以上(堆顶)时,kill这个进程。在kernel/trap.c的usertrap中增加以下判断条件

    if ((va < p->sz) && (va > PGROUNDDOWN(p->trapframe->sp)))
    
  • exec中,loadseg调用了walkaddr,可能会找不到相应虚拟地址的PTE,此时需要分配物理地址。在kernel/vm.c中

    if((pte == 0) || ((*pte & PTE_V) == 0)) {
        if (va > myproc()->sz || va < PGROUNDDOWN(myproc()->trapframe->sp)) {
            return 0;
        } 
        if ((mem = (uint64)kalloc()) == 0) return 0;
        va = PGROUNDDOWN(va);
        if (mappages(myproc()->pagetable, va, PGSIZE, mem, PTE_W|PTE_X|PTE_R|PTE_U) != 0) {
            kfree((void*)mem);
            return 0;
        }
        return mem;
    }
    

Lecture 7 Interrupts

设备会产生中断,xv6处理设备中断的代码位于kernel/trap.c中的devintr

进程的内核态中执行top half,中断时间中执行bottom half。top half是通过readwrite这样的system call来进行调用的,从而能让这个设备执行I/O操作。当设备执行完I/O操作之后,将产生一个设备中断,这个设备驱动的interrupt handler作为bottom half执行相应的操作。interrupt handler中没有任何用户进程的上下文,因此无法进行copyincopyout,只有top half才能和用户进程进行交互。

  • PLIC: Platform-Level Interrupt Controller,负责对从外部设备产生的中断进行管理
  • CLINT: Core-Local Interrupter,负责定时器相关的中断

7.1 Console input

console driver(kernel/console.c)是一个设备驱动,通过UART串口接受输入的符号。用户进程通过read system call来从console中一行行读取输入

xv6中使用的UART是QEMU模拟的16550芯片。UART硬件对于进程来说是一组memory-mapped寄存器,即RISC-V上有一些物理地址是直接和UART设备相连的。UART的地址从0x10000000或UART0开始,每个UART控制寄存器的大小为1字节,其位置定义在kernel/uart.c中。

  • LSR寄存器:line status register,用来指示输入的字节是否准备好被用户进程读取

  • RHR寄存器:receive holding register,用来放置可以被用户进程读取的字节。当RHR中的一个字节被读取时,UART硬件将其从内部的FIFO硬盘中删除,当FIFO中为空时,LSR寄存器被置0

  • THR寄存器:transmit holding register,当用户进程向THR写入一个字节时,UART将传输这个字节

xv6的main函数将调用consoleinit来初始化UART硬件,使得UART硬件在接收到字节或传输完成一个字节时发出中断

xv6 shell程序通过user/init.c开启的文件描述符来从console读取字节(在while循环中调用getcmd,在其中调用gets,再调用readsystem call)。在kernel中调用consoleread,等待输入完毕之后的中断,然后将输入缓存在cons.buf中,将输入either_copyout到user space后返回用户进程。如果用户没有输入完整的一行,则读取进程将在sleepsystem call中等待。

当用户输入了一个字符后,UART硬件将产生一个中断,这个终端将触发xv6进入trap。trap handler将调用devintr来通过scause寄存器判断是外部设备触发了这个中断,然后硬件将调用PLIC来判断是哪个外部设备触发的这个中断,如果是UART触发的,devintr将调用uartintruartintr将读取从UART硬件中写入的字符然后将其传送给consoleintrconsoleintr将积累这些字符直到整行都已经被读取,然后将唤醒仍在sleepconsoleread。当consoleread被唤醒后,将这一行命令复制给user space然后返回。

RISC-V对中断的支持:

SIE(supervisor interrupt enable) 寄存器用来控制中断,其中有一位是控制外部设备的中断(SEIE),一位控制suffer interrupt(一个CPU向另外一个CPU发出中断)(SSIE),一位控制定时器中断(STIE)

SSTATUS(supervisor status)寄存器,对某一个特定的CPU核控制是否接收寄存器,在kernel/riscv.h中的intr_on被设置

SIP(supervisor interrupt pending)寄存器,可以观察这个寄存器来判断有哪些中断在pending

case study:

用户在键盘上输入了一个字符l,这个l通过键盘被发送到UART,然后通过PLIC发送到CPU的一个核,这个核产生中断,跑到devintrdevintr发现是来自UART的,调用uartintr,调用uartgetc()通过RHR寄存器来获取这个字符,然后调用consoleintr,判断这个字符是否是特殊字符(backspace等),如果不是则将这个字符通过consputc(c)echo回给user,然后将其存储在cons.buf中,当发现整行已经输入完成后(c=='\n' || c ==C('D'))),唤醒consoleread()

7.2 Console output

对console上的文件描述符进行writesystem call,最终到达kernel/uart.c的uartputc函数。输出的字节将缓存在uart_tx_buf中,这样写入进程就不需要等待UART硬件完成字节的发送,只要当这个缓存区满了的情况下uartputc才会等待。当UART完成了一个字符的发送之后,将产生一个中断,uartintr将调用uartstart来判断设备是否确实已经完成发送,然后将下一个需要发送的字符发送给UART。因此让UART传送多个字符时,第一个字符由uartputcuartstart的调用传送,后面的字符由uartintruartstart的调用进行传送

I/O concurrency:设备缓冲和中断的解耦,从而让设备能够在没有进程等待读入的时候也能让console driver处理输入,等后面有进程需要读入的时候可以不需要等待。同时进程也可以不需要等待设备而直接写入字符到缓冲区。

consolereadconsoleintr中调用了acquire来获取一个锁,从而保护当前的console driver,防止同时期其他进程对其的访问造成的干扰。

7.3 Timer interrupts

xv6用计时器中断来在线程间切换,usertrapkerneltrap中的yield也会导致这种进程切换。RISC-V要求定时器中断的handler放在machine mode而不是supervisor mode中,而machine mode下是没有paging的,同时有另外一套完全独立的控制寄存器,因此不能将计时器中断的handler放在trap机制中执行。

kernel/start.c(在main之前)运行在machine mode下,timerinit()start.c中被调用,用来配置CLINT(core-local interruptor)从而能够在一定延迟之后发送一个中断,并设置一个类似于trapframe的scratch area来帮助定时器中断handler将寄存器和CLINT寄存器的地址保存到里面,最终start设置timervecmtvec(machine-mode trap handler)中使得在machine mode下发生中断后跳转到timervec然后enable定时器中断。

由于定时器中断可能在任意时间点发生,包括kernel在执行关键的操作时,无法强制关闭定时器中断,因此定时器中断的发生不能够影响这些被中断的操作。解决这个问题的方法是定时器中断handler让RISC-V CPU产生一个"software interrupt"然后立即返回,software interrupt以正常的trap机制被送到kernel中,可以被kernel禁用。

timervec是一组汇编指令,将一些寄存器保存在scratch area中,告知CLINT产生下一次定时器中断的时间,让RISC-V产生一个software interrupt,恢复寄存器并返回到trap.c中,判断which_dev==2为定时器中断后调用yield()

7.4 Real world

计时器中断将会通过调用yield进行强制的线程切换从而使CPU能够在各个内核线程之间均匀分配时间。

UART是通过对UART控制寄存器一个字节一个字节读取来获取数据的,这种方式叫做programmed I/O,因为是软件控制了数据的I/O,缺点是速度比较慢。DMA(Directed Memory Access)直接向RAM写入和读取数据,速度很快。现代的硬盘和网卡驱动使用DMA。

由于中断非常耗时,因此可以用一些技巧来减少中断。1. 用一个中断来处理很多一段时间内的事件。 2. 彻底禁止设备的中断,让CPU定时去检查这些设备是否有任务需要处理,这种技巧叫做polling

Lecture 8 Locking

多核CPU同时对某个共享的数据结构进行读写操作可能会发生冲突,因此需要concurrency control,即。锁提供了一种互斥机制,一段时间内只有一个CPU才能拥有这个锁,如果一个锁和一个被共享的数据结构联系起来,那么这个数据结构一次只能被一个CPU使用

8.1 Race conditions

kernel allocator中有一个free链表用来指示当前空闲待分配的内存,kalloc()将一页内存从free中弹出,kfree()将一页内存压入free。这个free链表被两个CPU的两个不同进程共享,如下图所示

race condition:一个内存地址同时被至少一个写入操作访问,会造成bug

struct element *list = 0;
struct lock listlock;

void push (int data) {
    struct element *l;
    l = malloc(sizeof *l);
    l->data = data;
    acquire(&listlock);
    // critical section
    l->next = list;
    list = l;
    release(&listlock);
}

acquirerelease之间的代码叫做critical section

当两个进程同时要求一个相同的锁时,这两个进程发生冲突,xv6对进程锁冲突没有做预防举措,但是更复杂的其他的kernel对此有实现。

注意acquirerelease的位置很重要,不要包围不必要的代码,否则会降低程序运行效率。

8.2 Code: Locks

xv6有两种锁:spinlock和sleep-lock。spinlock的代码位于kernel/spinlock.h的struct spinlock

struct spinlock {
  uint locked;       // Is the lock held?

  // For debugging:
  char *name;        // Name of lock.
  struct cpu *cpu;   // The cpu holding the lock.
};

locked为0时说明这个锁是可以acquire的。

acquire中需要让类似于

if (lk->locked == 0)
    lk->locked = 1;

这样的逻辑原子化,否则当两个不同的进程同时执行到上面的判断条件时,可能会同时获取这个锁。RISC-V是通过amoswap r, a来实现的,它将a内存地址中的内容和r寄存器中的内容互换。在acquire中,通过一个对amoswap的包装函数__sync_lock_test_and_set(&lk->locked, 1)来实现这个原子操作,这个函数的返回值是lk->locked的旧的值(被换下来的值)

while(__sync_lock_test_and_set(&lk->locked, 1) != 0)
    ;

通过while不断尝试将1和&lk->locked互换(spinning),当原先的lk->locked是0时跳出循环,这个锁被取得,否则当原先的lk->locked是1时不会跳出循环,并且lk->locked和1互换还是1,不会改变它的状态。

releaseacquire的反向操作,先将lk->cpu清零。然后调用

__sync_lock_release(&lk->locked);
// 相当于
// s1 = &lk->locked
// amoswap.w zero, zero, (s1)

lk->locked置0,这也是一个原子操作。

由于编译器有时候为了性能优化会重新排列代码的执行顺序,对于顺序执行的代码来说,这种重新排列顺序并不会改变代码执行的结果,但是对于并发执行的代码,则可能改变结果,因此需要在acquirerelease中用__sync_synchronize()来保证CPU和编译器不进行重新排列顺序。__sync_synchronize()是一个barrier,任何在这一行代码之前的代码都不能reorder到这一行代码的后面。

8.3 Deadlocks and lock ordering

如果一块代码需要同时拥有多个锁,那么应该让其他需要相同锁的进程按照相同的顺序acquire这些锁,否则可能出现死锁。比如进程1和2都需要锁A和锁B,如果进程1先acquire了锁A,进程2acquire了锁B,那么接下来进程1需要acquire锁B,进程2需要acquire锁A,但是这两个都不能acquire到也无法release各自的锁,就会出现死锁。

由于sleep在xv6中的机制,xv6中有很多长度为2的lock-order。比如consoleintr中要求先获得cons.lock,当整行输入完毕之后再唤醒等待输入的进程,这需要获得睡眠进程的锁。xv6的文件系统中有一个很长的lock chain,如果要创建一个文件需要同时拥有文件夹的锁、新文件的inode的锁、磁盘块缓冲区的锁、磁盘驱动器的vdisk_lock的锁以及调用进程的p->lock的锁

除了lock ordering之外,锁和中断的交互也可能造成死锁。比如当sys_sleep拥有tickslock时,发生定时器中断,定时器中断的handler也需要acquiretickslock,就会等待sys_sleep释放,但是因为在中断里面,只要不从中断返回sys_sleep就永远无法释放,因此造成了死锁。对这种死锁的解决方法是:如果一个中断中需要获取某个特定的spinlock,那么当CPU获得了这个spinlock之后,该中断必须被禁用。xv6的机制则更加保守:当CPU获取了任意一个lock之后,将disable掉这个CPU上的所有中断(其他CPU的中断保持原样)。当CPU不再拥有spinlock时,将通过pop_off重新使能中断

8.4 Sleep locks

spinlock的两个缺点:1. 如果一个进程拥有一个锁很长时间,另外一个企图acquire的进程将一直等待。2. 当一个进程拥有锁的时候,不允许把当前使用的CPU资源切换给其他线程,否则可能导致第二个线程也acquire这个线程,然后一直无法切回到原来的线程,无法release锁,从而导致死锁。

xv6提供了一种sleep-locks,可以在试图acquire一个被拥有的锁时yield CPU。spin-lock适合短时间的关键步骤,sleep-lock适合长时间的锁。

8.5 RCU

RCU(Read-Copy Update)是一种能让多个读进程对链表进行同时读取,并让一个写进程同时对链表进行写入修改操作的机制,这种机制避免了进程进行读/写操作都需要获取锁而造成的锁竞争问题,适用于大量进程同时对链表结构进行读取的操作。

基本原理是:写进程在写入某一个链表中的节点时,比如

head->E1->E2->E3->nil

试图修改E2->content,则不直接修改E2->content,因为在修改E2->content的过程中可能会有别的进程在读,此时可能读入写了一半的内容,我们希望一个读进程读取的内容要么是修改之前的,要么是修改之后的,而不是修改一半的内容。读进程的操作是

  1. lock,防止其他写进程同时进行写入
  2. e = alloc(),新分配一个element
  3. e->next = E2->next,此时同时有2个element指向E3,但是其他读进程在读的时候还是读取的是旧的E2
  4. e->content = new_content
  5. E1->next = E,此时其他读进程在读的时候是新的E2,这是一个原子操作
  6. unlock

由于编译器有时候为了优化会将2 3 4 5等步骤打乱,因此需要在第5步之前设置memory barrier,即只有在2 3 4均完成的情况下才能执行第5步

同时需要释放原先的E2,但是由于可能很多读进程已经获取了对原先E2的指针,必须等待这些读进程读取完毕不再使用E2才能将原先的E2释放掉,这是通过以下规则实现的:

  1. 所有的读进程不能够在进行context switch时拥有着对RCU-protected data的指针,也就是说在读进程读完E2之前,不能yield CPU
  2. 写进程需要等到所有的CPU都进行了一次context switch才能释放掉原先的数据,也就是E2(通过synchronize_rcu()实现)
// list reader using RCU interface
rcu_read_lock(); // 设置flag防止context switch
e = head;
while (p) {
    e = rcu_dereference(e); // 获取对e的指针
    a = e->content;
    e = e->next;
}
rcu_read_unlock(); // 可以开始context switch

// list writer using RCU interface, replacing head
acquire(lock);  // normal spin lock
old = head;     
e = alloc();
e->content = new_content;
e->next = head->next;
rcu_assign_pointer(&head, e); // commit the writes
release(lock);

synchronize_rcu(); // wait untill all cpus have context switched, meaning that no reader can hold the pointer to old head
free(old);

8.5 Lab 6: Copy on write fork

实验要求: 实现copy-on-write fork

  1. 修改uvmcopy(),使得父进程在调用该函数时将父进程的物理页映射到子进程的页表,而不是直接给子进程的页表分配新的物理页。要设置PTE_COW(1L >> 8)来表明这是一个copy-on-write页,在陷入page fault时需要进行特殊处理。将PTE_W置零,将该物理页的refc设置为1.

    int
    uvmcopy(pagetable_t old, pagetable_t new, uint64 sz)
    {
      pte_t *pte;
      uint64 pa, i;
      uint flags;
    //  char *mem;
    
      for(i = 0; i < sz; i += PGSIZE){
        if((pte = walk(old, i, 0)) == 0)
          panic("uvmcopy: pte should exist");
        if((*pte & PTE_V) == 0)
          panic("uvmcopy: page not present");
        pa = PTE2PA(*pte);
        flags = PTE_FLAGS(*pte);
        if (flags & PTE_W) {
          flags = (flags | PTE_COW) & (~PTE_W);
          *pte = PA2PTE(pa) | flags;
        } 
        refcinc((void*)pa); // increase the reference count of the physical page to 1
    //    if((mem = kalloc()) == 0)
    //      goto err;
    //    memmove(mem, (char*)pa, PGSIZE);
        if(mappages(new, i, PGSIZE, pa, flags) != 0){
    //      kfree(mem);
          goto err;
        }
      }
      return 0;
    
  2. usertrap()中用scause() == 13 || scause() == 15来判断是否为page fault,当发现是page fault并且r_stval()的物理页是COW页时,说明需要分配物理页,并重新映射到这个页表相应的虚拟地址上,当无法分配时,需要kill这个进程。注意:需要判断虚拟地址是否是有效的,其中包括需要判断这个虚拟地址是不是处在stack的guard page上,通过va <= PGROUNDDOWN(p->trapframe->sp) && va >= PGROUNDDOWN(p->trapframe->sp) - PGSIZE进行判断。按道理guard page的PTE应该是设置了PTE_V为0的,但是我尝试判断了这个条件,发现无法通过stacktest,这个问题需要后续确认。

    // in trap.c usertrap()
    } else if((which_dev = devintr()) != 0){
        // ok
    } else if (r_scause() == 13 || r_scause() == 15) {
        uint64 va = r_stval();
    
        if (va >= MAXVA || (va <= PGROUNDDOWN(p->trapframe->sp) && va >= PGROUNDDOWN(p->trapframe->sp) - PGSIZE)) {
            p->killed = 1;
        } else if (cow_alloc(p->pagetable, va) != 0) {
            p->killed = 1;
        }
    } else {
        p->killed = 1;
    }
    

    将为COW物理页分配地址的步骤包装成了函数cow_alloc(),方便后面copyout的使用。在将新的物理地址映射给页表之前,需要注意设置PTE_W为1,PTE_COW为0,设置完成之后尝试kfree掉旧的物理页,从而保证当没有任何进程的页表引用这个物理页的情况下这个物理页被释放掉。

    // allocate a physical address for virtual address va in pagetable
    // for copy on write lab
    int cow_alloc(pagetable_t pagetable, uint64 va) {
      uint64 pa;
      pte_t *pte;
      uint flags;
    
      if (va >= MAXVA) return -1; 
      va = PGROUNDDOWN(va);
      pte = walk(pagetable, va, 0);
      if (pte == 0) return -1;
      if ((*pte & PTE_V) == 0) return -1;
      pa = PTE2PA(*pte);
      if (pa == 0) return -1;
      flags = PTE_FLAGS(*pte);
    
      if (flags & PTE_COW) {
        char *mem = kalloc();
        if (mem == 0) return -1;
        memmove(mem, (char*)pa, PGSIZE);
        flags = (flags & ~PTE_COW) | PTE_W;
        *pte = PA2PTE((uint64)mem) | flags;
        kfree((void*)pa);
        return 0;
      }
      return 0;
    }
    
  3. 为了记录每个物理页被多少进程的页表引用,需要在kalloc.c中定义一个结构体refc,其中有一个大小为PGROUNDUP(PHYSTOP)/PGSIZE的int array来存放每个物理页的引用数。由于这个结构体是被所有的进程共享的,因此需要用一个spinlock进行保护

    // struct to maintain the ref counts
    struct {
      struct spinlock lock;
      int count[PGROUNDUP(PHYSTOP) / PGSIZE];
    } refc;
    

    初始化这个结构体,初始化这个锁,以及将数组所有元素置0。定义一些增加/减少/获取refc.count的函数,最后将refc的初始化函数放在kinit

    void
    refcinit()
    {
      initlock(&refc.lock, "refc");
      for (int i = 0; i < PGROUNDUP(PHYSTOP) / PGSIZE; i++) {
        refc.count[i] = 0;
      }
    }
    
    void
    refcinc(void *pa)
    {
      acquire(&refc.lock);
      refc.count[PA2IDX(pa)]++;
      release(&refc.lock);
    }
    
    void
    refcdec(void *pa)
    {
      acquire(&refc.lock);
      refc.count[PA2IDX(pa)]--;
      release(&refc.lock);
    }
    
    int
    getrefc(void *pa)
    {
      return refc.count[PA2IDX(pa)];
    }
    
    void
    kinit()
    {
      refcinit();
      initlock(&kmem.lock, "kmem");
      freerange(end, (void*)PHYSTOP);
    }
    

    在进行kalloc时,将对该物理页的引用加一

    void *
    kalloc(void)
    {
      struct run *r;
    
      acquire(&kmem.lock);
      r = kmem.freelist;
      if(r)
        kmem.freelist = r->next;
      release(&kmem.lock);
    
      if(r) {
        memset((char*)r, 5, PGSIZE); // fill with junk
        refcinc((void*)r);
      }
      return (void*)r;
    }
    

    相应的,在进行kfree时,要对该物理页的引用减一,然后再判断对该物理页的引用是否已经为0,如果已经为0,则将该物理页push回freelist

    // Free the page of physical memory pointed at by v,
    // which normally should have been returned by a
    // call to kalloc().  (The exception is when
    // initializing the allocator; see kinit above.)
    void
    kfree(void *pa)
    {
      struct run *r;
    
      if(((uint64)pa % PGSIZE) != 0 || (char*)pa < end || (uint64)pa >= PHYSTOP)
        panic("kfree");
      refcdec(pa);
      if (getrefc(pa) > 0) return;
      // Fill with junk to catch dangling refs.
      memset(pa, 1, PGSIZE);
    
      r = (struct run*)pa;
    
      acquire(&kmem.lock);
      r->next = kmem.freelist;
      kmem.freelist = r;
      release(&kmem.lock);
    }
    

    最后要注意,一开始进行kinit的时候调用了freerange(end, (void*)PHYSTOP),里面对所有物理页都进行了一次kfree,由于之前没有进行过kalloc,所以会导致每一页的初始refc.count都变成-1,因此需要在kinit时再给每一个物理页的refc.count加1。

    void
    kinit()
    {
      refcinit();
      initlock(&kmem.lock, "kmem");
      freerange(end, (void*)PHYSTOP);
      char *p;
      p = (char*)PGROUNDUP((uint64)end);
      for(; p + PGSIZE <= (char*)PHYSTOP; p += PGSIZE) {
        refcinc((void*)p);
      }
    }
    
  4. 最后, 由于copyout函数直接将kernel中的物理地址的内容复制给了用户进程中的物理地址, 没有经过mmu, 也无法进入page fault, 因此当将内容复制到用户进程的虚拟地址时,会将原来的内容覆盖掉而不是进行cow,因此需要修改copyout,调用cow_alloc

    // in kernel/vm.c copyput()
    uint64 n, va0, pa0;
    
    while(len > 0){
        va0 = PGROUNDDOWN(dstva);
    
        if (cow_alloc(pagetable, va0) != 0) {
            return -1;
        }
    
        pa0 = walkaddr(pagetable, va0);
        if(pa0 == 0)
            return -1;
        n = PGSIZE - (dstva - va0);
        if(n > len)
            n = len;
        memmove((void *)(pa0 + (dstva - va0)), src, n);
    

Lecture 9 Scheduling

由于操作系统需要同时运行的进程的数量可能大于电脑CPU的数量,因此需要一种让进程time share CPU的机制,理想情况下这种机制对于用户进程应该是透明的,即让每个进程都认为自己拥有一个单独的虚拟CPU

线程:一个串行的指令执行

xv6的kernel thread支持共享内存,user process不支持

9.1 Multiplexing

xv6在2种情况下在进程之间切换从而实现multiplexing。1. sleep/wakeup机制:进程等待设备或I/O、等待子进程退出、在sleepsys call中等待 2. 周期性强迫一个进程进行切换,防止一个进程占用过长时间。

9.2 Context Switching

进程的上下文切换涉及到用户空间和内核空间之间的来回转换。当进程需要切换时,首先通过system call或中断陷入内核态,进入该进程的内核线程,然后将内核线程的上下文(注意不是用户进程的上下文,用户进程的上下文已经保存在了trapframe里面)切换到当前CPU的scheduler线程,再将上下文切换到需要运行的进程的内核线程,最后返回用户空间。

从一个内核线程切换到另一个线程需要保存旧线程的寄存器,恢复新线程之前保存的寄存器。sppc将在此过程中被保存和切换。swtch可以实现这种寄存器组状态(也叫上下文)的保存和切换。当进程需要yieldCPU时,这个进程的内核线程将调用swtch来保存上下文并切换到scheduler的上下文,所有的上下文都保存在struct context中。swtch的传入参数为struct context *oldstruct context *new

yield()函数切换了进程的状态为RUNNABLE,调用了sched()sched调用了swtch(&p->context, &mycpu()->context)来将上下文切换到cpu->scheduler

sched(void)
{
  int intena;
  struct proc *p = myproc();

  if(!holding(&p->lock))
    panic("sched p->lock");
  if(mycpu()->noff != 1)
    panic("sched locks");
  if(p->state == RUNNING)
    panic("sched running");
  if(intr_get())
    panic("sched interruptible");

  intena = mycpu()->intena;
  swtch(&p->context, &mycpu()->context);
  mycpu()->intena = intena;
}

sched()中,先要检查是否还获取着p->lock,防止其他CPU的scheduler看见p->state==RUNNABLE的情况下试图去运行这个进程。通过检查mycpu()->noff来检查是否还获取着除了p->lock之外的其他锁,否则当切换到其他进程之后其他进程可能会acquire这个锁,而原先的进程由于没有在运行,因此一直无法释放掉这个锁,造成死锁。

swtch只保存callee saved寄存器,caller saved寄存器在栈中被调用的代码保存。swtch并没有保存pc寄存器,而是保存了ra,当恢复了新的进程之前保存的ra寄存器后,将返回到ra寄存器指向的上一个进程调用swtch的代码。如果保存pc寄存器,将只能回到swtch本身。由于切换到的&mycpu()->context是被schedulerswtch的调用所保存的,因此当进行swtch时,我们将返回到scheduler,栈指针也将指向当前CPU的scheduler stack。

9.3 Scheduling

调度器(scheduler)是每个CPU中都会运行的一个特殊的线程,这个线程中不断运行scheduler函数,来选取下一个需要运行的进程。

// Per-CPU process scheduler.
// Each CPU calls scheduler() after setting itself up.
// Scheduler never returns.  It loops, doing:
//  - choose a process to run.
//  - swtch to start running that process.
//  - eventually that process transfers control
//    via swtch back to the scheduler.
void
scheduler(void)
{
  struct proc *p;
  struct cpu *c = mycpu();
  
  c->proc = 0;
  for(;;){
    // Avoid deadlock by ensuring that devices can interrupt.
    intr_on();
    
    int nproc = 0;
    for(p = proc; p < &proc[NPROC]; p++) {
      acquire(&p->lock);
      if(p->state != UNUSED) {
        nproc++;
      }
      if(p->state == RUNNABLE) {
        // Switch to chosen process.  It is the process's job
        // to release its lock and then reacquire it
        // before jumping back to us.
        p->state = RUNNING;
        c->proc = p;
        swtch(&c->context, &p->context);

        // Process is done running for now.
        // It should have changed its p->state before coming back.
        c->proc = 0;
      }
      release(&p->lock);
    }
    if(nproc <= 2) {   // only init and sh exist
      intr_on();
      asm volatile("wfi");
    }
  }
}

想要yieldCPU的进程首先要获取自己进程的锁p->lock(防止其他CPU获取这个进程),修改当前的状态到RUNNABLErelease掉自己获取的其他锁,加载cpu->scheduler的上下文,返回到scheduler()之后,release掉自己的进程锁。

scheduler调用swtch到新的进程之前,scheduler需要已经获取这个进程的锁,并且将对这个进程的锁传递给被切换到的这个新的进程中,让新进程来release这个锁。一般来说,一个锁应该由acquire它的进程来进行release,但是由于一个进程的p->state是在scheduler中被改变的,需要对其进行保护,因此需要在scheduler中就获取这个进程的锁

当一个新的进程是第一次被scheduler调度的时候,不返回到sched,而是返回到forkret(因为之前并没有从sched中调用过swtch)。forkretp->lock释放掉,然后回到usertrapret

9.4 mycpu and myproc

xv6为每一个CPU都有一个struct cpu,记录当前运行在这个CPU上的进程的指针struct proc *proc、保存的寄存器struct context contextpush_off的nesting的数量int noff等变量。

RISC-V将所有CPU进行编号,该编号称为hartid,确保每个CPU的hartid都保存在这个CPU的tp寄存器内,可以让mycpu通过这个hartid来索引到一个struct cpu数组cpus[]中,从而获取对当前CPU的struct cpu的引用。当获取struct cpu之后如果发生了中断导致CPU被切换了,那么获取的struct cpu将是不正确的,因此需要用push_off来保证当前的中断使能被关闭。

使用myproc()函数来返回一个指向当前CPU运行的进程c->proc的指针。

9.5 Lab 7: thread

  1. Uthread

    实验要求: 实现一个用户层面的线程系统,需要实现thread_create() thread_schedule()以及uthread_switch.S

    uthread_switch.S中,仿照swtch.S进行上下文寄存器的保存和切换

    thread_switch:
    	/* YOUR CODE HERE */
            sd ra, 0(a0)
            sd sp, 8(a0)
            sd s0, 16(a0)
            sd s1, 24(a0)
            sd s2, 32(a0)
            sd s3, 40(a0)
            sd s4, 48(a0)
            sd s5, 56(a0)
            sd s6, 64(a0)
            sd s7, 72(a0)
            sd s8, 80(a0)
            sd s9, 88(a0)
            sd s10, 96(a0)
            sd s11, 104(a0)
    
            ld ra, 0(a1)
            ld sp, 8(a1)
            ld s0, 16(a1)
            ld s1, 24(a1)
            ld s2, 32(a1)
            ld s3, 40(a1)
            ld s4, 48(a1)
            ld s5, 56(a1)
            ld s6, 64(a1)
            ld s7, 72(a1)
            ld s8, 80(a1)
            ld s9, 88(a1)
            ld s10, 96(a1)
            ld s11, 104(a1)
    		ret    /* return to ra */
    

    uthread.cthread_create()中,第一次创建进程时需要初始化rasp寄存器. ra寄存器需要存放传入的函数地址, sp寄存器传入当前线程的栈底(最开始的位置)

    void 
    thread_create(void (*func)())
    {
      struct thread *t;
    
      for (t = all_thread; t < all_thread + MAX_THREAD; t++) {
        if (t->state == FREE) break;
      t->state = RUNNABLE;
      // YOUR CODE HERE
      t->context.ra = (uint64)func;
      t->context.sp = (uint64)t->stack + STACK_SIZE;
      }
    }
    

    thread_schedule()中,直接调用thread_switch,仿照swtch的格式进行上下文切换

        /* YOUR CODE HERE
         * Invoke thread_switch to switch from t to next_thread:
         * thread_switch(??, ??);
         */
        thread_switch((uint64)&t->context, (uint64)&current_thread->context);
    
  2. Using threads

    实验要求: 这个实验是基于实际的UNIXpthread库进行的,而非基于xv6. 需要对notxv6/ph.c进行补充,以实现多线程情况下能够正确地向一个哈希表插入键值对,其核心思想就是给每个线程对哈希表的操作都加锁.

    定义一个全局变量pthread_mutex_t lock[NBUCKET]; 其中NBUCKET是同时可以进行操作的线程最大数量

    main中对这个锁数列进行初始化

    for (int i=0; i<NBUCKET; i++) pthread_mutex_init(&lock[i], NULL);
    

    put()get()函数中上锁和解锁以保护对哈希表键值对的读写操作.

    static 
    void put(int key, int value)
    {
      int i = key % NBUCKET;
      pthread_mutex_lock(&lock[i]);
      // is the key already present?
      struct entry *e = 0;
      for (e = table[i]; e != 0; e = e->next) {
        if (e->key == key)
          break;
      }
      if(e){
        // update the existing key.
        e->value = value;
      } else {
        // the new is new.
        insert(key, value, &table[i], table[i]);
      }
      pthread_mutex_unlock(&lock[i]);
    }
    
    static struct entry*
    get(int key)
    {
      int i = key % NBUCKET;
    
      pthread_mutex_lock(&lock[i]);
      struct entry *e = 0;
      for (e = table[i]; e != 0; e = e->next) {
        if (e->key == key) break;
      }
      pthread_mutex_unlock(&lock[i]);
      return e;
    }
    
  3. Barrier

    实验要求: 实现线程的同步,即每一个线程必须等待其他所有线程都到达barrier之后才能继续进行下面的操作,需要用到pthread_cond_wait(&cond, &mutex)pthread_cond_broadcast(&cond)来进行线程的sleep和唤醒其他所有&cond中睡眠的线程. 需要对barrier.c中的barrier()进行实现.

    当每个线程调用了barrier()之后,需要增加bstate.nthread以表明到达当前round的线程数量增加了1, 但是由于bstate这个数据结构是线程之间共享的, 因此需要用pthread_mutex_lock对这个数据结构进行保护. 当bstate.nthread的数量达到线程总数nthread之后, 将bstate.round加1. 注意, 一定要等到所有的线程都达到了这个round, 将bstate.nthread清零之后才能将所有正在睡眠的线程唤醒, 否则如果先唤醒线程的话其他线程如果跑得很快, 在之前的线程将bstate.nthread清零之前就调用了bstate.nthread++,会出现问题(round之间是共用bstate.nthread的)

    static void 
    barrier()
    {
      // YOUR CODE HERE
      //
      // Block until all threads have called barrier() and
      // then increment bstate.round.
      //
      pthread_mutex_lock(&bstate.barrier_mutex);
      bstate.nthread++;
      if (bstate.nthread == nthread) {
        bstate.round++;
        bstate.nthread = 0;
        pthread_cond_broadcast(&bstate.barrier_cond);
        pthread_mutex_unlock(&bstate.barrier_mutex);
      } else {
        pthread_cond_wait(&bstate.barrier_cond, &bstate.barrier_mutex);
        pthread_mutex_unlock(&bstate.barrier_mutex);
      }
    }
    

Lecture 10 Sleep & Wakeup

10.1 Sleep and wakeup

sleep是当一个进程在等待某一个事件时陷入休眠状态,当这个事件发生时另外一个进程唤醒它。陷入休眠状态可以让这个进程不在等待的时候占用CPU资源

sleep(chan)让这个进程睡眠在chan这个wait channel上,wakeup(chan)将所有睡眠在chan上的进程全部唤醒。

lost wake-up problem:当一个进程A即将睡眠时,另外一个进程B发现已经满足了唤醒它的条件进行了唤醒,但是这时还没有进程睡眠在chan上,当进程A开始进入睡眠后,进程B可能不会再对进程A进行唤醒,进程A永远进入睡眠状态

lost wake-up problem的解决方法:用condition locksleepwakeup前后进行保护, 比如

void
V(struct semaphore *s)
{
    acquire(&s->lock);
    s->count += 1;
	wakeup(s);
	release(&s->lock);
}

void
P(struct semaphore *s)
{
	acquire(&s->lock);
	while(s->count == 0)
		sleep(s);
	s->count -= 1;
	release(&s->lock);
}

但是由于sleep时这个进程还拿着s->lockV永远也无法将P唤醒,因此会导致死锁。所以需要修改sleep,让sleep获取&s->lock这个参数,在sleep中将p->state设置为asleep之后将这个锁release掉,在从sleep中唤醒时,重新获取s->lock

sleep(s, &s->lock);

10.2 Code: Sleep and wakeup

kernel/proc.c中的sleep()函数

void
sleep(void *chan, struct spinlock *lk)
{
  struct proc *p = myproc();
  
  // Must acquire p->lock in order to
  // change p->state and then call sched.
  // Once we hold p->lock, we can be
  // guaranteed that we won't miss any wakeup
  // (wakeup locks p->lock),
  // so it's okay to release lk.

  acquire(&p->lock);  //DOC: sleeplock1
  release(lk);

  // Go to sleep.
  p->chan = chan;
  p->state = SLEEPING;

  sched();

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

  // Reacquire original lock.
  release(&p->lock);
  acquire(lk);
}

sleep中,首先需要获取一个进程锁来修改这个进程的p->chanp->state。当获取了这个进程锁之后可以立刻释放掉lk,因为在wakeup中只有先获取了sleep进程的进程锁才能进行尝试唤醒(判断p->state==SLEEPING),因此可以保证在sleep中将p->state修改为SLEEPING之前是无法进行唤醒的.

注意:当lk本身就是p->lock时会出现死锁的问题。

kernel/proc.c中的wakeup()函数

// Wake up all processes sleeping on chan.
// Must be called without any p->lock.
void
wakeup(void *chan)
{
  struct proc *p;

  for(p = proc; p < &proc[NPROC]; p++) {
    if(p != myproc()){
      acquire(&p->lock);
      if(p->state == SLEEPING && p->chan == chan) {
        p->state = RUNNABLE;
      }
      release(&p->lock);
    }
  }
}

唤醒进程遍历所有的进程,先获取每个进程的进程锁,然后判断每个进程是否睡眠在chan上,如果是则将其状态改为RUNNABLE,最后释放进程锁。

10.3 Code: Pipes

每一个pipe都有一个struct pipe,包括了一个lock和一个data缓冲数组

kernel/pipe.c中的pipewrite()

int
pipewrite(struct pipe *pi, uint64 addr, int n)
{
  int i = 0;
  struct proc *pr = myproc();

  acquire(&pi->lock);
  while(i < n){
    if(pi->readopen == 0 || pr->killed){
      release(&pi->lock);
      return -1;
    }
    if(pi->nwrite == pi->nread + PIPESIZE){ //DOC: pipewrite-full
      wakeup(&pi->nread);
      sleep(&pi->nwrite, &pi->lock);
    } else {
      char ch;
      if(copyin(pr->pagetable, &ch, addr + i, 1) == -1)
        break;
      pi->data[pi->nwrite++ % PIPESIZE] = ch;
      i++;
    }
  }
  wakeup(&pi->nread);
  release(&pi->lock);

  return i;
}

先获取pipe的锁,因为要对pi结构体里的变量进行修改。通过pi->nwrite == pi->nread+PIPESIZE判断缓冲区是否已经满了,如果已经满了就唤醒睡在&pi->nread上的piperead进程对缓冲区进行读取,自己睡在&pi->nwrite等待唤醒,否则就从user space的addrcopyin到内核态中的pi缓冲区内,完成n字节的读取之后将piperead进程唤醒,释放&pi->lock

piperead的代码

int
piperead(struct pipe *pi, uint64 addr, int n)
{
  int i;
  struct proc *pr = myproc();
  char ch;

  acquire(&pi->lock);
  while(pi->nread == pi->nwrite && pi->writeopen){  //DOC: pipe-empty
    if(pr->killed){
      release(&pi->lock);
      return -1;
    }
    sleep(&pi->nread, &pi->lock); //DOC: piperead-sleep
  }
  for(i = 0; i < n; i++){  //DOC: piperead-copy
    if(pi->nread == pi->nwrite)
      break;
    ch = pi->data[pi->nread++ % PIPESIZE];
    if(copyout(pr->pagetable, addr + i, &ch, 1) == -1)
      break;
  }
  wakeup(&pi->nwrite);  //DOC: piperead-wakeup
  release(&pi->lock);
  return i;
}

也要先获取pi->lock,判断当前缓冲区内是不是空的,如果是空的就进入睡眠,等待pipewrite进行写入并唤醒,否则循环读取n字节缓冲区数据,将缓冲区的数据copyout到用户空间的addr地址中,待n字节数据全部读取完成之后将pipewrite唤醒。

10.4 Code: Wait, exit, and kill

kernel/proc.c中的wait()代码

// Wait for a child process to exit and return its pid.
// Return -1 if this process has no children.
int
wait(uint64 addr)
{
  struct proc *np;
  int havekids, pid;
  struct proc *p = myproc();

  acquire(&wait_lock);

  for(;;){
    // Scan through table looking for exited children.
    havekids = 0;
    for(np = proc; np < &proc[NPROC]; np++){
      if(np->parent == p){
        // make sure the child isn't still in exit() or swtch().
        acquire(&np->lock);

        havekids = 1;
        if(np->state == ZOMBIE){
          // Found one.
          pid = np->pid;
          if(addr != 0 && copyout(p->pagetable, addr, (char *)&np->xstate,
                                  sizeof(np->xstate)) < 0) {
            release(&np->lock);
            release(&wait_lock);
            return -1;
          }
          freeproc(np);
          release(&np->lock);
          release(&wait_lock);
          return pid;
        }
        release(&np->lock);
      }
    }

    // No point waiting if we don't have any children.
    if(!havekids || p->killed){
      release(&wait_lock);
      return -1;
    }
    
    // Wait for a child to exit.
    sleep(p, &wait_lock);  //DOC: wait-sleep
  }
}

wait中是一个无限循环,每个循环中先对所有的进程循环查找自己的子进程,当发现有子进程并且子进程的状态为ZOMBIE时,将子进程的退出状态np->xstate copyoutwait传入的用户空间的addr中,然后释放掉子进程占用的所有的内存空间,返回子进程的pid。如果没有发现任何ZOMBIE子进程,睡眠在p上以等待子进程exit时唤醒pwait_lockwaitexit的condition lock用来防止错过wakeup。wait_lock实际上就是wait()调用者的p->lock

注意wait()先要获取调用进程的p->lock作为sleep的condition lock,然后在发现ZOMBIE子进程后获取子进程的np->lock,因此xv6中必须遵守先获取父进程的锁才能获取子进程的锁这一个规则。因此在循环查找np->parent == p时,不能先获取np->lock,因为np很有可能是自己的父进程,这样就违背了先获取父进程锁再获取子进程锁这个规则,可能造成死锁。

exit()代码

// Exit the current process.  Does not return.
// An exited process remains in the zombie state
// until its parent calls wait().
void
exit(int status)
{
  struct proc *p = myproc();

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

  // Close all open files.
  for(int fd = 0; fd < NOFILE; fd++){
    if(p->ofile[fd]){
      struct file *f = p->ofile[fd];
      fileclose(f);
      p->ofile[fd] = 0;
    }
  }

  begin_op();
  iput(p->cwd);
  end_op();
  p->cwd = 0;

  acquire(&wait_lock);

  // Give any children to init.
  reparent(p);

  // Parent might be sleeping in wait().
  wakeup(p->parent);
  
  acquire(&p->lock);

  p->xstate = status;
  p->state = ZOMBIE;

  release(&wait_lock);

  // Jump into the scheduler, never to return.
  sched();
  panic("zombie exit");
}

exit关闭所有打开的文件,将自己的子进程reparent给init进程,因为init进程永远在调用wait,这样就可以让自己的子进程在exit后由init进行freeproc等后续的操作。然后获取进程锁,设置退出状态和当前状态为ZOMBIE,进入scheduler中并且不再返回。

注意:在将p->state设置为ZOMBIE之后才能释放掉wait_lock,否则wait()的进程被唤醒之后发现了ZOMBIE进程之后直接将其释放,此时ZOMBIE进程还没运行完毕。

exit是让自己的程序进行退出,kill是让一个程序强制要求另一个程序退出。kill不能立刻终结另一个进程,因为另一个进程可能在执行敏感命令,因此kill仅仅设置了p->killed为1,且如果该进程在睡眠状态则将其唤醒。当被kill的进程进入usertrap之后,将会查看p->killed是否为1,如果为1则将调用exit

// Kill the process with the given pid.
// The victim won't exit until it tries to return
// to user space (see usertrap() in trap.c).
int
kill(int pid)
{
  struct proc *p;

  for(p = proc; p < &proc[NPROC]; p++){
    acquire(&p->lock);
    if(p->pid == pid){
      p->killed = 1;
      if(p->state == SLEEPING){
        // Wake process from sleep().
        p->state = RUNNABLE;
      }
      release(&p->lock);
      return 0;
    }
    release(&p->lock);
  }
  return -1;
}

10.5 Real world

xv6采用的scheduling的方式为round robin,即每个进程轮流运行,实际的操作系统的scheduling可以让进程有优先级。当高优先级和低优先级共享一个锁而低优先级拿到这个锁的情况下,将产生priority inversion,将导致大量高优先级进程等候低优先级进程,从而形成convoy

对整个进程列表查找睡眠在chan上的进程是非常低效的,更好的解决方案是将chan替代为一个可以存储睡眠在此结构体上的进程列表的结构体,比如Linux的wait queue

xv6中的wakeup唤醒所有睡在chan上的进程,然后这些进程将竞争检查sleep conditoin,这种情况通常需要被避免。许多是采用signalbroadcast两种唤醒模式,前面一种只唤醒一个进程,后面一种唤醒所有进程。

10.6 Lab 8: Locks

  1. Memory allocator

    实验要求:实现一个per CPU freelist,以减小各个进程同时调用kallockfree造成的对kmem.lock锁的竞争。可以调用cpuid()函数来获取当前进程运行的CPU ID,但是要在调用前加上push_off以关闭中断。

    kernel/kalloc.c中,修改kmem结构体为数组形式

    struct {
      struct spinlock lock;
      struct run *freelist;
    } kmem[NCPU];
    

    kinit()要循环初始化每一个kmem的锁

void
kinit()
{
  for (int i = 0; i < NCPU; i++) {
    initlock(&kmem[i].lock, "kmem");
  }
  freerange(end, (void*)PHYSTOP);
}

kfree将释放出来的freelist节点返回给调用kfree的CPU

void
kfree(void *pa)
{
  struct run *r;

  if(((uint64)pa % PGSIZE) != 0 || (char*)pa < end || (uint64)pa >= PHYSTOP)
    panic("kfree");

  // Fill with junk to catch dangling refs.
  memset(pa, 1, PGSIZE);

  r = (struct run*)pa;

  push_off();
  int ncpu = cpuid();

  acquire(&kmem[ncpu].lock);
  r->next = kmem[ncpu].freelist;
  kmem[ncpu].freelist = r;
  release(&kmem[ncpu].lock);
  pop_off();
}

kalloc中,当发现freelist已经用完后,需要向其他CPU的freelist借用节点

void *
kalloc(void)
{
  struct run *r;

  push_off();
  int ncpu = cpuid();

  acquire(&kmem[ncpu].lock);
  r = kmem[ncpu].freelist;
  if(r) {
    kmem[ncpu].freelist = r->next;
  } 
  release(&kmem[ncpu].lock);
  if (!r) {
    // steal other cpu's freelist
    for (int i = 0; i < NCPU; i++) {
      if (i == ncpu) continue;
      acquire(&kmem[i].lock);
      r = kmem[i].freelist;     
      if (r) {
        kmem[i].freelist = r->next;
        release(&kmem[i].lock);
        break;
      }
      release(&kmem[i].lock);
    }
  }
  pop_off();
  
  if(r)
    memset((char*)r, 5, PGSIZE); // fill with junk
  return (void*)r;
}
  1. Buffer cache

实验要求:xv6文件系统的buffer cache采用了一个全局的锁bcache.lock来负责对buffer cache进行读写保护,当xv6执行读写文件强度较大的任务时会产生较大的锁竞争压力,因此需要一个哈希表,将buf entry以buf.blockno为键哈希映射到这个哈希表的不同的BUCKET中,给每个BUCKET一个锁,NBUCKET最好选择素数,这里选择13。注意:这个实验不能像上一个一样给每个CPU一个bcache,因为文件系统在多个CPU之间是真正实现共享的,否则将会造成一个CPU只能访问某些文件的问题。

这里实验让我们可以使用ticks作为时间戳,来代替原来的双向链表实现LRU的功能。

kernel/bio.c中,首先设置NBUCKET宏定义为13,声明外部变量ticks,修改bcache以为每个BUCKET设置一个链表头,并设置一个锁

#define NBUCKET 13

uint extern ticks;

struct {
  struct spinlock lock[NBUCKET];
  struct buf buf[NBUF];

  // Linked list of all buffers, through prev/next.
  // Sorted by how recently the buffer was used.
  // head.next is most recent, head.prev is least.
  struct buf head[NBUCKET];
} bcache;

修改kernel/buf.h中的buf结构体,删除struct buf *prev,即将双向链表变为单向链表,添加uint time这个成员变量作为时间戳。

实现一个简单的哈希函数

int hash (int n) {
  int result = n % NBUCKET;
  return result;
}

修改binit函数,为每个bcache.lock以及b->lock进行初始化,并将所有buf先添加到bucket 0哈希表中

void
binit(void)
{
  struct buf *b;

  for (int i = 0; i < NBUCKET; i++) {
    initlock(&bcache.lock[i], "bcache");
  }

  bcache.head[0].next = &bcache.buf[0];
  // for initialization, append all bufs to bucket 0
  for (b = bcache.buf; b < bcache.buf+NBUF-1; b++) {
    b->next = b+1;
    initsleeplock(&b->lock, "buffer");
  }
}

修改bget,先查找当前哈希表中有没有和传入参数devblockno相同的buf。先要将blockno哈希到一个id值,然后获得对应id值的bcache.lock[id]锁,然后在这个bucket id哈希链表中查找符合对应条件的buf,如果找到则返回,返回前释放掉bcache.lock[id],并对buf加sleeplock。

int id = hash(blockno);
acquire(&bcache.lock[id]);
b = bcache.head[id].next;
while (b) {
    if (b->dev == dev && b->blockno == blockno) {
        b->refcnt++;
        if (holding(&bcache.lock[id]))
            release(&bcache.lock[id]);
        acquiresleep(&b->lock);
        return b;
    }
    b = b->next;
}

如果没有找到对应的buf,需要在整个哈希表中查找LRU(least recently used)buf,将其替换掉。这里由于总共有NBUCKET个哈希表,而此时一定是持有bcache.lock[id]这个哈希表的锁的,因此当查找其他哈希表时,需要获取其他哈希表的锁,这时就会有产生死锁的风险。风险1:查找的哈希表正是自己本身这个哈希表,在已经持有自己哈希表锁的情况下,不能再尝试acquire一遍自己的锁。风险2:假设有2个进程同时要进行此步骤,进程1已经持有了哈希表A的锁,尝试获取哈希表B的锁,进程2已经持有了哈希表B的锁,尝试获取哈希表A的锁,同样会造成死锁,因此要规定一个规则,是的当持有哈希表A的情况下如果能够获取哈希表B的锁,则当持有哈希表B锁的情况下不能够持有哈希表A的锁。该规则在can_lock函数中实现。

int can_lock(int id, int j) {
  int num = NBUCKET/2;
  if (id <= num) {
    if (j > id && j <= (id+num))
      return 0;
  } else {
    if ((id < j && j < NBUCKET) || (j <= (id+num)%NBUCKET)) {
      return 0;
    }
  }
  return 1;
}

其中id是已经持有的锁,j是判断是否能获取该索引哈希表的锁。这个规则实际上规定了在持有某一个锁的情况下,只能再尝试获取NBCUKET/2个哈希表锁,另一半哈希表锁是不能获取的。

确定了这个规则之后,尝试遍历所有的哈希表,通过b->time查找LRUbuf。先判断当前的哈希表索引是否为id,如果是,则不获取这个锁(已经获取过它了),但是还是要遍历这个哈希表的;同时也要判断当前哈希表索引是否满足can_lock规则,如果不满足,则不遍历这个哈希表,直接continue。如果哈希表索引j既不是id,也满足can_lock,则获取这个锁,并进行遍历。当找到了一个当前情况下的b->time最小值时,如果这个最小值和上一个最小值不在同一个哈希表中,则释放上一个哈希表锁,一直持有拥有当前情况下LRUbuf这个哈希表的锁,直到找到新的LRUbuf且不是同一个哈希表为止。找到LRUbuf后,由于此时还拥有这个哈希表的锁,因此可以直接将这个buf从该哈希链表中剔除,并将其append到bucketid哈希表中,修改这个锁的devblocknovalidrefcnt等属性。最后释放所有的锁。

int index = -1;
uint smallest_tick = __UINT32_MAX__;
// find the LRU unused buffer
for (int j = 0; j < NBUCKET; ++j) {
    if (j!=id && can_lock(id, j)) {
        // if j == id, then lock is already acquired
        // can_lock is to maintain an invariant of lock acquisition order
        // to avoid dead lock
        acquire(&bcache.lock[j]);
    } else if (!can_lock(id, j)) {
        continue;
    }
    b = bcache.head[j].next;
    while (b) {
        if (b->refcnt == 0) {
            if (b->time < smallest_tick) {
                smallest_tick = b->time;
                if (index != -1 && index != j && holding(&bcache.lock[index])) release(&bcache.lock[index]);
                index = j;
            }   
        }
        b = b->next;
    }
    if (j!=id && j!=index && holding(&bcache.lock[j])) release(&bcache.lock[j]);
}
if (index == -1) panic("bget: no buffers");
b = &bcache.head[index];

while (b) {
    if ((b->next)->refcnt == 0 && (b->next)->time == smallest_tick) {
        selected = b->next;
        b->next = b->next->next;
        break;
    }
    b = b->next;
}
if (index != id && holding(&bcache.lock[index])) release(&bcache.lock[index]);
b = &bcache.head[id];
while (b->next) {
    b = b->next;
}
b->next = selected;
selected->next = 0;
selected->dev = dev;
selected->blockno = blockno;
selected->valid = 0;
selected->refcnt = 1;
if (holding(&bcache.lock[id]))
    release(&bcache.lock[id]);
acquiresleep(&selected->lock);
return selected;
}

修改brelse。当b->refcnt==0时,说明这个buf已经被使用完了,可以进行释放,为其加上时间戳

void
brelse(struct buf *b)
{
  if(!holdingsleep(&b->lock))
    panic("brelse");

  releasesleep(&b->lock);

  int id = hash(b->blockno);
  acquire(&bcache.lock[id]);
  b->refcnt--;
  if (b->refcnt == 0) {
    b->time = ticks;
  }
  
  release(&bcache.lock[id]);
}

修改bpinbunpin,将bcache.lock修改为bcache.lock[id]

最后在param.h中修改NBUF的值,由于can_lock规则,NBUF实际上变成了原来的一半,可能会出现buffer run out的panic,无法通过writebig测试,因此适当增大NBUF,我这里将其修改为了(MAXOPBLOCKS*12)

Lecture 11 File system

11.0 Overview

文件系统:组织并存储数据。

文件系统的特性:

  • 需要一个存储文件夹和文件的数据结构来记录存储文件内容的硬盘块的ID,并且记录磁盘的哪些部分是空闲的
  • 在不同的用户和应用程序之间共享数据
  • 数据在重启/意外崩溃之后依然保持原样
  • 由于访问硬盘速度远慢于访问内存,因此文件系统必须在内存里设置一个对经常访问的文件内容的缓存区。

xv6文件系统的组织架构

  • disk层:对virtio硬盘上的文件块进行读写操作
  • buffer cache层:对磁盘文件块进行缓存,并确保只有1个内核进程能在一段时间内修改文件块上存储的数据。
  • logging层:让更高的层级能够将对文件块的所有update打包到一个transaction中,从而能保证所有文件块能够在将要崩溃时原子地进行update
  • inode层:为每个文件提供一个独一无二的inode number
  • directory层:将每个文件夹作为一个特殊的inode,这个inode的内容是文件夹entry
  • pathname层:将文件夹组织为层级,并通过递归查找来解析路径
  • file descriptor层:将管道、设备等UNIX资源用文件系统进行抽象

11.1 Disk layer

文件系统将磁盘分为了几个部分,每个部分的最小单元是block,一个block的大小为1024字节,如下图所示

  • block 0: 启动区域,文件系统不会使用,包含了操作系统启动所需要的代码
  • blcok 1: superblock,存储了文件系统的元数据(block的大小、block的数目、inode的数目等),里面有一个mkfs的程序,用来构建初始的文件系统
  • block 2-31:log block
  • block 32-44: inode,一个inode的大小为64字节,一个block的大小为1024字节,因此block32为inode 1-16,block33为inode 17-32
  • block 45 bitmap block,用来跟踪哪些block是在使用
  • 最后从block 46开始是data block,要么是在bitmap中被标记为空闲状态,要么存储了文件/文件夹的内容

11.2 Buffer cache layer

buffer cache层有两个作用:

  1. 将对磁盘块的访问权限进行同步,保证内存中只保存一个该磁盘块的拷贝,且一次只有一个内核线程访问这个拷贝,但同时可以有多个对这个block的引用
  2. 将被频繁访问的块缓存到内存中

buffer cache(bcache结构体)中的buf数量是一定的(NBUF),因此当新的文件块需要加入缓冲区时,需要将最早使用的缓冲区中的文件块替换为新的文件块。缓冲区的使用早晚通过head来判断。

buffer cache本身是一个双向链接的链表,链表元素为buf结构体。binit()如下

void
binit(void)
{
  struct buf *b;

  initlock(&bcache.lock, "bcache");

  // Create linked list of buffers
  bcache.head.prev = &bcache.head;
  bcache.head.next = &bcache.head;
  for(b = bcache.buf; b < bcache.buf+NBUF; b++){
    b->next = bcache.head.next;
    b->prev = &bcache.head;
    initsleeplock(&b->lock, "buffer");
    bcache.head.next->prev = b;
    bcache.head.next = b;
  }
}

buffer cache层的接口函数有2个,分别是bread()bwrite()

// Return a locked buf with the contents of the indicated block.
struct buf*
bread(uint dev, uint blockno)
{
  struct buf *b;

  b = bget(dev, blockno);
  if(!b->valid) {
    virtio_disk_rw(b, 0);
    b->valid = 1;
  }
  return b;
}

bread通过bget获取一个指定了设备devblocknobuf *,这是从硬盘指定的块中获取的一个缓冲数据结构体,保存在内存中,可以进行修改

// Look through buffer cache for block on device dev.
// If not found, allocate a buffer.
// In either case, return locked buffer.
static struct buf*
bget(uint dev, uint blockno)
{
  struct buf *b;

  acquire(&bcache.lock);

  // Is the block already cached?
  for(b = bcache.head.next; b != &bcache.head; b = b->next){
    if(b->dev == dev && b->blockno == blockno){
      b->refcnt++;
      release(&bcache.lock);
      acquiresleep(&b->lock);
      return b;
    }
  }

  // Not cached.
  // Recycle the least recently used (LRU) unused buffer.
  for(b = bcache.head.prev; b != &bcache.head; b = b->prev){
    if(b->refcnt == 0) {
      b->dev = dev;
      b->blockno = blockno;
      b->valid = 0;
      b->refcnt = 1;
      release(&bcache.lock);
      acquiresleep(&b->lock);
      return b;
    }
  }
  panic("bget: no buffers");
}

bget()先查看需要buffer的文件块是否已经在bcache中,如果没有,就将LRU(least recently used) buf替换为这个新的文件块。b->valid=0说明这个buf是刚刚被替换掉的而不是本来就有的,因此要让bread()从硬盘中再加载一下相应block中的数据到这个buf中。bcache.lock负责保护哪些block被缓存的信息,而b->lock负责对这个缓存块的读写行为进行保护。acquiresleep是获取这个锁之后立即让这个进程进入睡眠,这是因为当获取着锁的时候会disable掉中断,这样就永远也无法听到来自硬盘的中断。

// Write b's contents to disk.  Must be locked.
void
bwrite(struct buf *b)
{
  if(!holdingsleep(&b->lock))
    panic("bwrite");
  virtio_disk_rw(b, 1);
}

向硬盘指定块中写入数据。struct buf中已经保存了devblockno等数据,因此可以直接调用virtio_disk_rw(b,1)进行写入。brelse负责释放bread中返回的buf的锁。当发现指向这个buf的reference变为0时,将其移动到双向链表的最开头。

// Release a locked buffer.
// Move to the head of the most-recently-used list.
void
brelse(struct buf *b)
{
  if(!holdingsleep(&b->lock))
    panic("brelse");

  releasesleep(&b->lock);

  acquire(&bcache.lock);
  b->refcnt--;
  if (b->refcnt == 0) {
    // no one is waiting for it.
    b->next->prev = b->prev;
    b->prev->next = b->next;
    b->next = bcache.head.next;
    b->prev = &bcache.head;
    bcache.head.next->prev = b;
    bcache.head.next = b;
  }
  
  release(&bcache.lock);
}

11.3 Block allocator

文件和文件夹都存储在磁盘块中,磁盘块必须从一个空闲池中进行分配。block allocator为磁盘的是否空闲的状态准备了一个bitmap,每一位对应一个磁盘块,0表示空闲1表示正在使用,mkfs负责设置这些位。

balloc负责分配新的磁盘块,bfree负责释放磁盘块

// Allocate a zeroed disk block.
static uint
balloc(uint dev)
{
  int b, bi, m;
  struct buf *bp;

  bp = 0;
  for(b = 0; b < sb.size; b += BPB){
    bp = bread(dev, BBLOCK(b, sb));
    for(bi = 0; bi < BPB && b + bi < sb.size; bi++){
      m = 1 << (bi % 8);
      if((bp->data[bi/8] & m) == 0){  // Is block free?
        bp->data[bi/8] |= m;  // Mark block in use.
        log_write(bp);
        brelse(bp);
        bzero(dev, b + bi);
        return b + bi;
      }
    }
    brelse(bp);
  }
  panic("balloc: out of blocks");
}

balloc最外层循环读取每个bitmap bit所代表的block(BPB是一个Block的bit数目,BBLOCK负责把bit转化为blockno),内循环负责检查所有BPB位,查看这个block是否空闲

11.4 Inode layer

inode可能指代2种数据结构:1. 储存在硬盘上的数据结构,包含了inode类型、inode指向的文件/文件夹大小、一个数据blockno的列表 2. 存储在内存中的数据结构,拥有on-disk inode的拷贝以及其他kernel需要的metadata

on-disk inode放在一个连续的区域内,这个区域有很多block,叫做inode block,每个inode大小相同,为64字节

on-disk inode以struct dinode的形式定义

struct dinode {
  short type;           // File type
  short major;          // Major device number (T_DEVICE only)
  short minor;          // Minor device number (T_DEVICE only)
  short nlink;          // Number of links to inode in file system
  uint size;            // Size of file (bytes)
  uint addrs[NDIRECT+1];   // Data block addresses
};

type指定了是文件、文件夹还是设备,type==0表示这个inode处于空闲状态。nlink表示连接到这个inode的directory entry的个数,用来判断这个inode应该何时被释放。size记录了这个文件/文件夹的大小,addrs记录了这个inode拥有的文件内容分布在的disk block的所有blockno

内存中的inode是active inodes,用struct inode定义

// in-memory copy of an inode
struct inode {
  uint dev;           // Device number
  uint inum;          // Inode number
  int ref;            // Reference count
  struct sleeplock lock; // protects everything below here
  int valid;          // inode has been read from disk?

  short type;         // copy of disk inode
  short major;
  short minor;
  short nlink;
  uint size;
  uint addrs[NDIRECT+1];
};

所谓active inodes,是指内存中有C指针指向了这个inode,ref是指向这个inode的指针数量,当ref==0时kernel将把这个inode从内存中剔除。iget函数和iput函数实现对inode pointer的获取和释放。

在inode层中总共有4种lock。1. icache.lock负责确保1个inode只在cache中出现最多1次 ,并且保证ref正确记录引用到这个inode的数量,因此对ref的修改都需要用icache.lock进行保护 2. 每个inode自己也有1个lock来保护对inode成员变量以及inode指向的文件或文件夹的内容的保护 3. ref是内存中指向这个inode的个数,当ref大于0时,不会将这个inode从icache中剔除 4. nlink这个成员变量在on-disk inode中也存在,统计指向这个文件的directory entry的个数,当为0时将释放掉这个inode

典型的调用顺序:

ip = iget(dev, inum);
ilock(ip);
...examine and modify ip->xxx
iunlock(ip);
iput(ip);

iget返回了一个直到调用iput都有效的inode,任何代码均可同时访问,因此可以有很多指针指向同一个inode。iget返回的inode可能没有任何有用的内容(valid==0),因此需要调用ilock来从硬盘中读取内容,并将获取inode的锁。iunlock释放inode的锁。将对inode的引用获取和对inode上锁分离开来,从而在查找路径等情况中避免死锁。

inode cache主要目的是实现不同进程对inode访问的同步,cache只是次要的,因为当inode被经常访问时,这个inode很可能会被保存在buffer cahce中。inode cache是write-through的,当cache中的inode被修改,将会立即通过iupdate将修改写入到磁盘中。

11.5 Code: inodes

ialloc负责从硬盘上的inode blocks中寻找空闲的inode,当找到之后将新的type写入到disk中然后通过调用iget返回一个内存中的inode(将这个inode写入到inode cache)中。

// Allocate an inode on device dev.
// Mark it as allocated by  giving it type type.
// Returns an unlocked but allocated and referenced inode.
struct inode*
ialloc(uint dev, short type)
{
  int inum;
  struct buf *bp;
  struct dinode *dip;

  for(inum = 1; inum < sb.ninodes; inum++){
    bp = bread(dev, IBLOCK(inum, sb));
    dip = (struct dinode*)bp->data + inum%IPB;
    if(dip->type == 0){  // a free inode
      memset(dip, 0, sizeof(*dip));
      dip->type = type;
      log_write(bp);   // mark it allocated on the disk
      brelse(bp);
      return iget(dev, inum);
    }
    brelse(bp);
  }
  panic("ialloc: no inodes");
}

ialloc必须保证其他进程不会同时对这个inode进行读写,这是通过bp已经通过bget上了sleeplock来保证的。

iget在inode cache中查找和传入的device、inode no相同的active entry,如果找到了这个entry就返回对这个inode的一个新的指针,否则找到一个空的entry将其dev、inum等设置为对应的数值,并设置valid为0待后续从block中读取数据。

// Find the inode with number inum on device dev
// and return the in-memory copy. Does not lock
// the inode and does not read it from disk.
static struct inode*
iget(uint dev, uint inum)
{
  struct inode *ip, *empty;

  acquire(&icache.lock);

  // Is the inode already cached?
  empty = 0;
  for(ip = &icache.inode[0]; ip < &icache.inode[NINODE]; ip++){
    if(ip->ref > 0 && ip->dev == dev && ip->inum == inum){
      ip->ref++;
      release(&icache.lock);
      return ip;
    }
    if(empty == 0 && ip->ref == 0)    // Remember empty slot.
      empty = ip;
  }

  // Recycle an inode cache entry.
  if(empty == 0)
    panic("iget: no inodes");

  ip = empty;
  ip->dev = dev;
  ip->inum = inum;
  ip->ref = 1;
  ip->valid = 0;
  release(&icache.lock);

  return ip;
}

iput将释放1个对inode的C指针,ref--,如果ref==0nlink==0(不在任何文件夹中出现),则释放掉这个inode(ip->type=0)和它对应的所有data block(通过itrunc)。

// Drop a reference to an in-memory inode.
// If that was the last reference, the inode cache entry can
// be recycled.
// If that was the last reference and the inode has no links
// to it, free the inode (and its content) on disk.
// All calls to iput() must be inside a transaction in
// case it has to free the inode.
void
iput(struct inode *ip)
{
  acquire(&icache.lock);

  if(ip->ref == 1 && ip->valid && ip->nlink == 0){
    // inode has no links and no other references: truncate and free.

    // ip->ref == 1 means no other process can have ip locked,
    // so this acquiresleep() won't block (or deadlock).
    acquiresleep(&ip->lock);

    release(&icache.lock);

    itrunc(ip);
    ip->type = 0;
    iupdate(ip);
    ip->valid = 0;

    releasesleep(&ip->lock);

    acquire(&icache.lock);
  }

  ip->ref--;
  release(&icache.lock);
}

on-disk inode结构dinode包括了一个size和NDIRECT+1个addr,前NDIRECT个addr叫做direct blocks,最后一个addr给出了indirect block的地址,因此一个文件的前12kB(NDIRECT x BSIZE)可以从inode中的direct block addr直接读取,后256kB(NINDIRECT x BSIZE)可以通过indirect block addr翻译得到。因此xv6支持的最大的文件大小为268kB

bmap()负责获取inode中的第n块data block的地址。当bn<NDIRECT时直接返回ip->addrs[bn],如果没有这个地址就调用balloc分配一个data block。当NDIRECT<bn<NINDIRECT时先bread(ip->dev, ip->addrs[NDIRECT]),然后获取bp->data[bn-NDIRECT]

// Return the disk block address of the nth block in inode ip.
// If there is no such block, bmap allocates one.
static uint
bmap(struct inode *ip, uint bn)
{
  uint addr, *a;
  struct buf *bp;

  if(bn < NDIRECT){
    if((addr = ip->addrs[bn]) == 0)
      ip->addrs[bn] = addr = balloc(ip->dev);
    return addr;
  }
  bn -= NDIRECT;

  if(bn < NINDIRECT){
    // Load indirect block, allocating if necessary.
    if((addr = ip->addrs[NDIRECT]) == 0)
      ip->addrs[NDIRECT] = addr = balloc(ip->dev);
    bp = bread(ip->dev, addr);
    a = (uint*)bp->data;
    if((addr = a[bn]) == 0){
      a[bn] = addr = balloc(ip->dev);
      log_write(bp);
    }
    brelse(bp);
    return addr;
  }

  panic("bmap: out of range");
}

11.6 Directory layer

文件夹和文件的实现非常类似,它的inode类型为T_DIR,数据是许多directory entry,每个entry的数据类型为struct dirent,包含一个名称和inode number

dirlookup是在directoy中查找名称为name的directoy entry

// Look for a directory entry in a directory.
// If found, set *poff to byte offset of entry.
struct inode*
dirlookup(struct inode *dp, char *name, uint *poff)
{
  uint off, inum;
  struct dirent de;

  if(dp->type != T_DIR)
    panic("dirlookup not DIR");

  for(off = 0; off < dp->size; off += sizeof(de)){
    if(readi(dp, 0, (uint64)&de, off, sizeof(de)) != sizeof(de))
      panic("dirlookup read");
    if(de.inum == 0)
      continue;
    if(namecmp(name, de.name) == 0){
      // entry matches path element
      if(poff)
        *poff = off;
      inum = de.inum;
      return iget(dp->dev, inum);
    }
  }

  return 0;
}

注意,dirlookup正是iget没有直接对inode上锁的原因,因为在调用dirlookup的时候,实际上已经对dp上锁了,如果查找的directory entry是.(当前文件夹/inode本身),那么会试图对dp再acquire一次锁,造成死锁。

dirlink将一个新的directory entry写入文件夹dp中,查找dp中尚未分配的entry(inum==0),如果找到了就将off设置为这个entry的offset,然后用writei写入,如果没有找到就将off设置为dp->size,将这个entry加在最后。

// Write a new directory entry (name, inum) into the directory dp.
int
dirlink(struct inode *dp, char *name, uint inum)
{
  int off;
  struct dirent de;
  struct inode *ip;

  // Check that name is not present.
  if((ip = dirlookup(dp, name, 0)) != 0){
    iput(ip);
    return -1;
  }

  // Look for an empty dirent.
  for(off = 0; off < dp->size; off += sizeof(de)){
    if(readi(dp, 0, (uint64)&de, off, sizeof(de)) != sizeof(de))
      panic("dirlink read");
    if(de.inum == 0)
      break;
  }

  strncpy(de.name, name, DIRSIZ);
  de.inum = inum;
  if(writei(dp, 0, (uint64)&de, off, sizeof(de)) != sizeof(de))
    panic("dirlink");

  return 0;
}

11.7 Pathname layer

namei函数负责对pathname进行解析,并返回inode。namei调用了namex函数,namex中传入了参数nameiparent,当为1时返回的inode是传入path的父文件夹,并将最后的element名称复制到name中。namex不断调用了skipelem,一步步将当前的inode变为下一级inode,直到得到最终需要的inode为止。注意:在每个循环中都要先进行ilock(ip),因为直到ilock才能保证ip->type从硬盘中读取

// Look up and return the inode for a path name.
// If parent != 0, return the inode for the parent and copy the final
// path element into name, which must have room for DIRSIZ bytes.
// Must be called inside a transaction since it calls iput().
static struct inode*
namex(char *path, int nameiparent, char *name)
{
  struct inode *ip, *next;

  if(*path == '/')
    ip = iget(ROOTDEV, ROOTINO);
  else
    ip = idup(myproc()->cwd);

  while((path = skipelem(path, name)) != 0){
    ilock(ip);
    if(ip->type != T_DIR){
      iunlockput(ip);
      return 0;
    }
    if(nameiparent && *path == '\0'){
      // Stop one level early.
      iunlock(ip);
      return ip;
    }
    if((next = dirlookup(ip, name, 0)) == 0){
      iunlockput(ip);
      return 0;
    }
    iunlockput(ip);
    ip = next;
  }
  if(nameiparent){
    iput(ip);
    return 0;
  }
  return ip;
}

由于namex中可能涉及到多个data block的I/O,因此可能是比较耗时的,当一个kernel thread执行namex被disk I/O阻塞时,其他thread再执行namex查找不同的文件夹path时不会被阻塞。

11.8 File descriptor layer

file descriptor layer可以让UNIX中的所有资源,包括设备(比如console)来统一表示为文件。每个打开的文件(file descriptor)都用struct file来表示,这是一个pipe/inode/device的包装结构体。

struct file {
  enum { FD_NONE, FD_PIPE, FD_INODE, FD_DEVICE } type;
  int ref; // reference count
  char readable;
  char writable;
  struct pipe *pipe; // FD_PIPE
  struct inode *ip;  // FD_INODE and FD_DEVICE
  uint off;          // FD_INODE
  short major;       // FD_DEVICE
};

system call调用一次open()将增加一个open file,同一个file可以被多个不同的进程open,出现在这些进程的file table中。struct file中的reference count记录了同一个file被引用的次数。

所有打开的文件都保存在global file table,即ftable中。

filealloc负责在file table中分配一个文件,在ftable中扫描ref==0的file,增加ref后返回这个file *。

filedup负责对这个file descriptor的ref++并返回这个文件的file *。

fileclose负责对file descriptor的ref–,当ref==0时根据这个file的类型释放掉pipe或者inode

// Close file f.  (Decrement ref count, close when reaches 0.)
void
fileclose(struct file *f)
{
  struct file ff;

  acquire(&ftable.lock);
  if(f->ref < 1)
    panic("fileclose");
  if(--f->ref > 0){
    release(&ftable.lock);
    return;
  }
  ff = *f;
  f->ref = 0;
  f->type = FD_NONE;
  release(&ftable.lock);

  if(ff.type == FD_PIPE){
    pipeclose(ff.pipe, ff.writable);
  } else if(ff.type == FD_INODE || ff.type == FD_DEVICE){
    begin_op();
    iput(ff.ip);
    end_op();
  }
}

filereadfilewrite分别确认f->readablef->writable是否允许当前的读写操作,然后再根据f->typeFD_PIPE/FD_DEVICE/FD_INODE进行对应的piperead/readi等操作。filereadfilewrite都使用了f->off来指示当前的读写到了哪一个位置。

11.9 Code: System calls

sys_linksys_unlink这两个syscall实现对inode增加或者删除引用

sys_link传入一个参数old和一个参数newnew是需要链接到old的路径。sys_link增加ip->nlink,然后调用nameiparent查找到new的父文件夹,调用dirlink在父文件夹下创建一个名称为new的directory entry,链接到old所代表的inode。如果中间出现了错误,需要跳转到bad来减去ip->nlink

// Create the path new as a link to the same inode as old.
uint64
sys_link(void)
{
  char name[DIRSIZ], new[MAXPATH], old[MAXPATH];
  struct inode *dp, *ip;

  if(argstr(0, old, MAXPATH) < 0 || argstr(1, new, MAXPATH) < 0)
    return -1;

  begin_op();
  if((ip = namei(old)) == 0){
    end_op();
    return -1;
  }

  ilock(ip);
  if(ip->type == T_DIR){
    iunlockput(ip);
    end_op();
    return -1;
  }

  ip->nlink++;
  iupdate(ip);
  iunlock(ip);

  if((dp = nameiparent(new, name)) == 0)
    goto bad;
  ilock(dp);
  if(dp->dev != ip->dev || dirlink(dp, name, ip->inum) < 0){
    iunlockput(dp);
    goto bad;
  }
  iunlockput(dp);
  iput(ip);

  end_op();

  return 0;

bad:
  ilock(ip);
  ip->nlink--;
  iupdate(ip);
  iunlockput(ip);
  end_op();
  return -1;
}

sys_link为已经存在的inode创建一个新的directoy entry,而create则是为一个新的inode创建一个新的directory entry。create首先调用nameiparent获取父文件夹,然后调用dirlookup来查看这个文件夹下是否已经存在同名的inode,如果存在且调用这个create的是open来创建一个文件的话,那么直接返回这个inode。如果这个名称不存在,则调用ialloc。如果是mkdir调用的create(即type==T_DIR),则要创建...作为对父级inode和当前inode的引用,最终将当前的name dirlink到当前inode。

static struct inode*
create(char *path, short type, short major, short minor)
{
  struct inode *ip, *dp;
  char name[DIRSIZ];

  if((dp = nameiparent(path, name)) == 0)
    return 0;

  ilock(dp);

  if((ip = dirlookup(dp, name, 0)) != 0){
    iunlockput(dp);
    ilock(ip);
    if(type == T_FILE && (ip->type == T_FILE || ip->type == T_DEVICE))
      return ip;
    iunlockput(ip);
    return 0;
  }

  if((ip = ialloc(dp->dev, type)) == 0)
    panic("create: ialloc");

  ilock(ip);
  ip->major = major;
  ip->minor = minor;
  ip->nlink = 1;
  iupdate(ip);

  if(type == T_DIR){  // Create . and .. entries.
    dp->nlink++;  // for ".."
    iupdate(dp);
    // No ip->nlink++ for ".": avoid cyclic ref count.
    if(dirlink(ip, ".", ip->inum) < 0 || dirlink(ip, "..", dp->inum) < 0)
      panic("create dots");
  }

  if(dirlink(dp, name, ip->inum) < 0)
    panic("create: dirlink");

  iunlockput(dp);

  return ip;
}

sys_open有两种模式,当O_CREATEflag置1时调用了create,当置0时调用了namei来找到这个inode,然后调用fileallocfdalloc来分配struct file和file descriptor。

11.10 Real world

xv6中的buffer cache采用了一个非常简单的链表来对LRU进行剔除,但是实际的操作系统中采用了hash表和heap来进行LRU剔除,且加入了虚拟内存系统来支持memory-mapped文件。

在目录树中采用了线性扫描disk block的方式进行查找,在disk block较多的情况下会消耗很多时间,因此Windows的NTFS等文件系统将文件夹用balanced tree进行表示,以确保对数事件的查找。

xv6要求文件系统只能有一个硬盘设备,但是现代操作系统可以采用RAID或者软件的方式来将很多硬盘组成一个逻辑盘

现代操作系统还应该具备的其他特性:snapshots、增量式备份。

Lecture 12 Crash recovery

12.1 Logging layer

由于很多对文件系统的操作都涉及了对硬盘的多次写入,当某次写入后发生崩溃将导致文件系统出现问题。xv6通过logging来解决这个问题,xv6的syscall不会直接对硬盘上的block进行写入,而是将所有想要进行的对硬盘的写入操作的描述放到log中,当syscall将所有的写入操作都放到log后向硬盘写入一个commit记录来表示这个log已经记录了所有的操作,然后syscall进行全部的写入操作,并将硬盘上的log全部清除。

当操作系统崩溃后进行重启,将在进行任何进程之前从崩溃中恢复。如果在对硬盘的所有写入操作commit之前发生了崩溃,那么这个log将被视为不完整的log,xv6将直接忽略这个log,如果崩溃发生在commit之后,说明这个log是完整的,则恢复系统将重复这些步骤,最后删除log。

12.2 Log design

log位于硬盘上的log block中,由一个header block和后面的一系列被log的block的copy组成。header block中记录了所有被log的block的blockno和log block的总数count。xv6只有在一个transaction commits时才向header block写入,并在将logged block copy写入到文件系统中的logged block后将count归零。

为了支持不同的进程对文件系统同时的操作,可以将多个syscall对硬盘的写入打包到一个transaction当中,因此commit必须保证当前没有syscall

group commit可以将多个不同进程的syscall的transaction放在一起进行commit。

由于log block有数量限制,因此一个syscall能够写入的block数量也同样有限制,比如sys_write将一个write分成了好几个transaction以fit log。

  • write ahead规则:只有所有被修改的buffer都被写入到了log block才能开始向文件系统中的home location写入block
  • freeing规则:直到所有的log block都被写入了home location,并且消除了header block,才能开始修改或者释放log block

12.3 Code: logging

在syscall中对log使用的典型流程:

begin_op();
...
bp = bread();
bp->data = ...;
log_write(bp);
...
end_op();

begin_op将等待直到当前logging系统不再committing且有足够的空余log block能够容纳这次syscall中所有的写入,log.outstanding是需要进行block写入但是还没有调用log_write的syscall的个数,因此也记录了目前预定了log space的syscall的个数,MAXOPBLOCKS是一个syscall最高可以使用的log block

// called at the start of each FS system call.
void
begin_op(void)
{
  acquire(&log.lock);
  while(1){
    if(log.committing){
      sleep(&log, &log.lock);
    } else if(log.lh.n + (log.outstanding+1)*MAXOPBLOCKS > LOGSIZE){
      // this op might exhaust log space; wait for commit.
      sleep(&log, &log.lock);
    } else {
      log.outstanding += 1;
      release(&log.lock);
      break;
    }
  }
}

log_write在完成对buf的修改之后,需要实现bwrite的功能,但是先要在log block中给传入的这个buf预留一个位子,并且将这个bufpin在buffer cache中(此时这个传入的buf还没有被写入log block中,因此它还是修改后的block的唯一拷贝,因此不能让其在buffer cache中被evict掉)。当一个transaction中对同一个block进行了多次读写操作时,不会另外预留log block,这叫做log absorption

// Caller has modified b->data and is done with the buffer.
// Record the block number and pin in the cache by increasing refcnt.
// commit()/write_log() will do the disk write.
//
// log_write() replaces bwrite(); a typical use is:
//   bp = bread(...)
//   modify bp->data[]
//   log_write(bp)
//   brelse(bp)
void
log_write(struct buf *b)
{
  int i;

  if (log.lh.n >= LOGSIZE || log.lh.n >= log.size - 1)
    panic("too big a transaction");
  if (log.outstanding < 1)
    panic("log_write outside of trans");

  acquire(&log.lock);
  for (i = 0; i < log.lh.n; i++) {
    if (log.lh.block[i] == b->blockno)   // log absorbtion
      break;
  }
  log.lh.block[i] = b->blockno;
  if (i == log.lh.n) {  // Add new block to log?
    bpin(b);
    log.lh.n++;
  }
  release(&log.lock);
}

end_op先要减少outstanding syscall的数量,如果该数量被减为了0(即所有的syscall都已经完成了log_write),则调用commit()来将当前的transaction commit掉。

// called at the end of each FS system call.
// commits if this was the last outstanding operation.
void
end_op(void)
{
  int do_commit = 0;

  acquire(&log.lock);
  log.outstanding -= 1;
  if(log.committing)
    panic("log.committing");
  if(log.outstanding == 0){
    do_commit = 1;
    log.committing = 1;
  } else {
    // begin_op() may be waiting for log space,
    // and decrementing log.outstanding has decreased
    // the amount of reserved space.
    wakeup(&log);
  }
  release(&log.lock);

  if(do_commit){
    // call commit w/o holding locks, since not allowed
    // to sleep with locks.
    commit();
    acquire(&log.lock);
    log.committing = 0;
    wakeup(&log);
    release(&log.lock);
  }
}

commit分成四个阶段

static void
commit()
{
  if (log.lh.n > 0) {
    write_log();     // Write modified blocks from cache to log
    write_head();    // Write header to disk -- the real commit
    install_trans(0); // Now install writes to home locations
    log.lh.n = 0;
    write_head();    // Erase the transaction from the log
  }
}
  1. write_log将所有被修改了的block(buf)写入到log block中
// Copy modified blocks from cache to log.
static void
write_log(void)
{
  int tail;

  for (tail = 0; tail < log.lh.n; tail++) {
    struct buf *to = bread(log.dev, log.start+tail+1); // log block
    struct buf *from = bread(log.dev, log.lh.block[tail]); // cache block
    memmove(to->data, from->data, BSIZE);
    bwrite(to);  // write the log
    brelse(from);
    brelse(to);
  }
}
  1. write_head将header block写入到硬盘中,其中包括log block的总数log.lh.n和每个log block接下来需要写入到的data block的blockno,这是一个commit真正开始的节点。
// Write in-memory log header to disk.
// This is the true point at which the
// current transaction commits.
static void
write_head(void)
{
  struct buf *buf = bread(log.dev, log.start);
  struct logheader *hb = (struct logheader *) (buf->data);
  int i;
  hb->n = log.lh.n;
  for (i = 0; i < log.lh.n; i++) {
    hb->block[i] = log.lh.block[i];
  }
  bwrite(buf);
  brelse(buf);
}
  1. install_trans将log block写入到需要写入的home data block中
// Copy committed blocks from log to their home location
static void
install_trans(int recovering)
{
  int tail;

  for (tail = 0; tail < log.lh.n; tail++) {
    struct buf *lbuf = bread(log.dev, log.start+tail+1); // read log block
    struct buf *dbuf = bread(log.dev, log.lh.block[tail]); // read dst
    memmove(dbuf->data, lbuf->data, BSIZE);  // copy block to dst
    bwrite(dbuf);  // write dst to disk
    if(recovering == 0)
      bunpin(dbuf);
    brelse(lbuf);
    brelse(dbuf);
  }
}
  1. 最后将log header中的count变为0

recover_from_loginitlog中被调用,每次操作系统重启之前都将调用一次。它调用install_trans(1),当log.ln.n不为0时将重新进行一次将log block写入到data block中。

12.4 Lab 9: file system

  1. Large files

实验要求:增大xv6文件的最大大小,由于原来的大小是12个DIRECT BLOCK+256个INDIRECT BLOCK,现在要将一个DIRECT BLOCK变为一个DOUBLE INDIRECT BLOCK,即像page table一样的双层结构,从而使得一个文件的大小最高位256*256+256+11=65803 bytes。

修改kernel/fs.h中的宏定义

#define NDIRECT 11
#define NINDIRECT (BSIZE / sizeof(uint))
#define NDBINDIRECT ((BSIZE / sizeof(uint)) * (BSIZE / sizeof(uint)))
#define MAXFILE (NDIRECT + NINDIRECT + NDBINDIRECT)

并修改fs.hfile.h中的struct inode中的addrs

uint addrs[NDIRECT+1+1];

修改kernel/fs.c中的bmap,增加bn<NDBINDIRECT情况下的两级映射。先要让bn减掉上一级的NINDIRECT,DOUBLE INDIRECT BLOCK映射的第一级index应为bn/NINDIRECT,第二级index为bn%NINDIRECT。依次读取每一级block中的addr (blockno),当不存在时进行balloc

bn -= NINDIRECT;

if (bn < NDBINDIRECT) {
    uint index_1 = bn/NINDIRECT;
    uint index_2 = bn%NINDIRECT;
    if ((addr = ip->addrs[NDIRECT+1]) == 0)
        ip->addrs[NDIRECT+1] = addr = balloc(ip->dev);
    bp = bread(ip->dev, addr);
    a = (uint*)bp->data;
    if ((addr = a[index_1]) == 0) {
        a[index_1] = addr = balloc(ip->dev);
        log_write(bp);
    }
    brelse(bp);
    bp = bread(ip->dev, addr);
    a = (uint*)bp->data;
    if ((addr = a[index_2]) == 0) {
        a[index_2] = addr = balloc(ip->dev);
        log_write(bp);
    }
    brelse(bp);
    return addr;
}

itrunc中,也要释放掉DOUBLE INDIRECT BLOCK中每一级的block

if (ip->addrs[NDIRECT+1]) {
    bp = bread(ip->dev, ip->addrs[NDIRECT+1]);
    a = (uint*)bp->data;
    for (j = 0; j < NINDIRECT; j++) {
        if (a[j]) {
            bp2 = bread(ip->dev, a[j]);
            b = (uint*)bp2->data;
            for (i = 0; i < NINDIRECT; i++) {
                if (b[i]) {
                    bfree(ip->dev, b[i]);
                }
            }
            brelse(bp2);
            bfree(ip->dev, a[j]);
        }
    }
    brelse(bp);
    bfree(ip->dev, ip->addrs[NDIRECT+1]);
    ip->addrs[NDIRECT+1] = 0;
}
  1. Symbolic links

实验要求:实现xv6的软链接,即新增加一种T_SYMLINK类型的文件,这个文件中存有需要链接到的文件的pathname,当使用open并指定O_NOFOLLOW为0时,可以打开这个软链接文件指向的文件,而非软链接文件本身。要求实现一个symlink(char *target, char *path)这个syscall,实现path所代表的文件软链接到target代表的文件,target不存在也可以。修改open,从而可以打开软链接文件。注意:软链接文件指向的文件可能也是一个软链接文件,open需要递归地找到最终的非软链接文件,但是可能出现软链接文件互相链接的情况,会产生死循环,因此规定递归查找软链接的深度不能超过10.

首先在Makefile中添加编译项$U/_symlinktest\,在kernel/fcntl.h中添加open的flagO_NOFOLLOW,该flag不能和其他flag的位重叠

#define O_NOFOLLOW 0x010

kernel/stat.h中添加inode类型T_SYMLINK

#define T_SYMLINK 4   // Symlink

注册sys_symlink

kernel/syscall.c

extern uint64 sys_symlink(void);
...
[SYS_symlink] sys_symlink,

kernel/syscall.h

#define SYS_symlink 22

user/user.h

int symlink(char*, char*);

user/usys.pl

entry("symlink");

kernel/sysfile.c中,添加symlink的实现

uint64
sys_symlink(void) {
  char target[MAXPATH];
  char path[MAXPATH];
  struct inode *ip;
  // char test[MAXPATH];

  if (argstr(0, target, MAXPATH) < 0 || argstr(1, path, MAXPATH) < 0) {
    return -1;
  }
  begin_op();
  if ((ip = namei(path)) == 0) {
    // the path inode does not exist
    ip = create(path, T_SYMLINK, 0, 0);
    iunlock(ip);
  } 
  ilock(ip);
  // write the target path name into the end of inode
  if (writei(ip, 0, (uint64)target, ip->size, MAXPATH) != MAXPATH) {
    panic("symlink");
  }
  iunlockput(ip);
  end_op();
  return 0;
} 

首先要判断是否存在path所代表的inode,如果不存在就用create添加一个T_SYMLINK类型的inode。在inode的最后添加需要软链接到的target的路径名称

修改open,添加对T_SYMLINK类型文件的处理方法

if (ip->type == T_SYMLINK) {
    if ((omode & O_NOFOLLOW) == 0) {
        // recursively follow symlink
        int count = 0;
        char sympath[MAXPATH];
        while (1) {
            if (count >= 10) {
                iunlockput(ip);
                end_op();
                return -1;
            }
            // read the path name from inode
            if (readi(ip, 0, (uint64)sympath, ip->size-MAXPATH, MAXPATH) != MAXPATH) {
                panic("open symlink");
            }
            iunlockput(ip);
            if ((ip = namei(sympath)) == 0) {
                // could not find this file
                end_op();
                return -1;
            }
            ilock(ip);
            if (ip->type != T_SYMLINK) {
                break;
            }
            count++;
        }
    }
}

要判断打开open的flag是否为NOFOLLOW,如果是的话,就不打开软链接所指向的文件,否则需要将inode递归地替换为软链接指向的文件,直到最终的inode类型不是T_SYMLINK为止。整个递归的深度不能超过10,否则报错。

Lecture 13 Linux ext3fs crash recovery

xv6 logging系统的问题:速度太慢,每个block需要被写入磁盘2次(一次到log中一次到fs中),每个syscall需要等待commit完成,磁盘的I/O是同步阻塞的。

13.1 ext3 journal design

和xv6不同的是ext3可以跟踪多个transaction的状态。

在ext3的内存中和xv6一样有write-back block cache,也为每个transaction维护了一个transaction info,这些transaction info包括:

  1. 每个transaction有一个sequence number
  2. 被这个transaction修改的block no
  3. handles:这个transaction中的syscall

在硬盘上有circular log,log包括:

  1. log super block(注意和整个硬盘的super block是不一样的):记录了seq#最低的有效的transaction的offset和seq #
  2. descriptor block:每一个transaction开头的block,记录了这个transaction的seq #和home block #,有MAGIC #(为了和data block进行区分)
  3. data block
  4. commit block:每个transaction结尾的block,有MAGIC #

当log空间全部满了或者等待时间足够长时,将从最小的seq# transaction开始将log block写入到home disk中,并释放掉这一个transaction所占据的log block,因此这实际上是一个circular log。

commit transaction的步骤

  1. 暂时禁用新的syscall
  2. 等待outstanding syscall结束,因为一个transaction是每隔一段时间就close的,这个时间节点上已经调用了start()的syscall很可能还没有调用end(),需要等待所有的这些syscall结束
  3. 开始一个新的transaction, unblock新的syscall
  4. 向这个transaction的descriptor block写入需要写入的block#
  5. 向log写入这些需要写入的data block
  6. 写入commit block,当commit block写入完成,这就是commit point,这之后如果崩溃,Log中的内容将得到恢复
  7. 向home location写入block
  8. 释放/重新使用这个transaction中的log block

恢复的步骤

SB T8 (freed)T5 T6 T7

SB指向T6,T5已经被释放了,而T8已经覆盖了T5的descriptor

  1. 重启后,先查看Super Block,寻找指向的最小的valid seq#的transaction,这里是T6
  2. 找到log的尾部,将循环查找到一个缺失的commit(通过magic #)或者错误的seq#(太小),若找到这个partial transaction就将其忽略

13.2 ext3 journal performance

ext3的journaling速度提高的原因:

  1. asynchronous disk update:syscall不用等待disk I/O,而是等待修改了buffer cache就可以进行返回,不同的syscall的log可以放在一起进行统一的commit。缺点是当syscall返回时这个syscall需要完成的对硬盘的写入可能实际上并没有完成。fsync(fd)可以强迫这个fd对磁盘的修改已经被写入到log block才会返回

  2. batching:将许多syscall的log放在一个transaction中。ext3每隔数秒钟进行一次commit,所以每个transaction会包含了很多syscall。

    batching的优点:1)将查找block等本来每个syscall都会有的固定开销平摊到多个syscall中。2)允许write absorption,可以让许多对同一个block的写入放在一起。3)disk scheduling:可以将一个batch中所有对block的写入以blockno排序依次写入,比一个一个查找随机位置的block效率更高。

  3. concurrency:1)同时进行多个fs syscall 2) log中可以有多个transaction,每个transaction可以在不同的阶段中,总共有4种阶段,包括:

    • open transaction:可以接受新的syscall的写入请求
    • committing transaction:将buf写入到log中
    • committed transaction:将log写入到home block中
    • old transaction:被释放

13.3 ext3 syscall code

sys_unlink() {
    h = start(); // 告知logging system有新的system call, h是handle,即这个syscall的一个identifier
    get(h, blockno); // 告知logging system需要修改cache block,将handle和blockno添加到当前的transaction中
    // modify the block in the cache
    stop(h); // 当当前所有的syscall的start()都被stop()之后才能将这个transaction commit
}

13.4 Collision between transactions

为了提高文件系统的性能,在committing 一个transaction的时候不会完全禁止文件系统的更新,而是会创建一个新的transaction。如果新的transaction需要写入旧的正在committing的transaction的某一个block,可能会造成冲突。当我们在提交旧的transaction时,我们必须保证此时的buffer cache不能被修改,因此我们需要一个这个buffer cache的拷贝来进行新transaction的修改

ext3 ordered data mode:在普通模式下,需要将metadata和file content都写入到log block中,然后再写入到file system中,需要写入2次,速度很慢,在ordered data mode模式下,只将metadata写入到log中,file content直接写入file system的Home location,需要注意一定要在file content写入完成之后才能将metadata commit,这样当file content写入完成后metadata commit之前发生崩溃的话,恢复系统不会恢复metadata,但是这个file content也会因为metadata中的inode没有被写入而不会暴露给用户。

Lecture 14 Virtual memory for user applications

kernel可以通过虚拟内存来获得很多特性,比如copy-on-write、lazy allocation等,user program也可以通过虚拟内存获得一些特性。

比如:

  1. Concurrent garbage collector
  2. Generation garbage collector
  3. Concurrent check-pointing
  4. Data-compression paging
  5. Persistent stores

user-level VM primitive

trap: 以user mode进入page fault trap并进行处理

prot: 降低页的访问权限

e.g. mprotect(addr, len, PROT_READ):让被映射的地址addr只能有读权限

unprot: 提高页的访问权限

dirty: 返回一个自上一个调用以来已经被使用了的page

map2: 将同一个physical page map到2个不同的virtual address,设置不同的访问权限

14.1 mmap

将一个文件或者其他对象映射到进程的地址空间,实现文件磁盘地址和进程虚拟地址空间中一段虚拟地址的一一对应关系。

为什么要建立文件磁盘地址和进程虚拟地址空间的映射?因为常规的文件系统操作是用户态发起readsyscall,然后在buffer cache中查找是否有相应的数据,如果没有就从磁盘中拷贝数据到buffer cache中,因为buffer cache处在内核态,因此需要将buffer cachecopyout到用户进程的虚拟内存中,这就需要2次拷贝操作,而在mmap中只需要直接将文件数据拷贝到对应的用户空间虚拟内存即可。

文件映射

使用文件内容初始化物理内存

addr = mmap(NULL, len, PROT_READ|PROT_WRITE, MAP_PRIVATE, fd, offset) NULL表示让kernel自己查找需要映射到的文件磁盘地址

匿名映射

不需要打开文件,初始化一个全为0的内存空间

私有匿名映射用于malloc给进程分配虚拟的地址空间,当各个进程读的时候共享物理内存,写的时候copy on write

mmap(NULL, len, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0)

共享匿名映射在进行fork时不会采用写时copy on write,父子进程完全共享同样的物理内存页,实现父子进程的IPC

14.2 Virtual memory implementation

VMA

VMA (Virtual Memory Area)由一连串的虚拟地址组成,每个VMA都有相同的权限,指向相同的对象(文件、匿名内存等)

User-level traps

比如,当PTE被标记为invalid时发生缺页异常,CPU从用户态陷入内核态

内核态保存trapframe

内核将查看产生错误的VMA,判断接下来的动作。比如一个用户进程设置了一个sig_action(handler for signal),则内核将产生一个signal,将这个signal传播给用户态,回到用户态调用sig_action,然后sig_action将调用mprotect来修改PTE

最后sig_action将返回kernel,kernel继续返回到被中断的进程。

14.3 Use case: Garbage collector

为了将程序分配的内存在程序退出之后能够自动清理掉,需要垃圾收集器,其作用是从一个根集合root出发,递归地寻找所有被引用的对象,剩下的所有没有被引用到的对象就是垃圾内存,需要被清除。

复制算法是将Heap内存空间分为2块区域,一块是from区域一块是to区域,将from区域中的所有能够被root查找到的对象全部复制到to区域中,然后将from区域的所有内存清除掉。这个算法的缺点在于gc线程运行的过程中需要暂停所有的用户线程。

Baker实时垃圾收集算法:to区域包括new区域、scanned区域和unscanned区域。在最开始先将root复制到to区域,当用户线程调用new来分配新的对象时,from区域中被root及其它对象指向的对象会被增量地复制到to区域中。几个规则:

  • 用户线程只能看到to-space的指针,也就是说每个用户线程取得的对象指向的对象只能在to-space中,如果检查发现在from-space中,就要将这个被指向的对象拷贝到to-space,并且更新指针,然后才能返回给用户线程。
  • 每次调用new,被分配的对象放在new区域中,并且new区域中所有的对象的指针都只能指向to区域
  • scanned区域中的对象只能拥有指向to区域的指针,unscanned区域中的对象可以拥有指向to或from区域的指针。

为了避免检查每个从内存中取回的对象的指针使其满足第一个规则,gc给每个unscanned区域中的对象都加了no access,使得每次试图访问这个unscanned对象时都会陷入缺页保护异常,然后gc扫描这个对象,将其指向的对象拷贝到to区域中,然后再恢复访问权限。这个机制也提供了concurrency,因为可以防止gc thread和用户线程同时修改unscanned page,因为unscanned page已经被加上了none access保护,用户线程是无法访问的。但是由于map2的存在,可以将相同的物理地址再映射一个给gc thread,其访问权限是R|W,因此用户线程不能访问的unscanned page gc线程是可以访问的。

14.4 Lab 10: mmap

实验要求:实现mmapmunmap来实现将文件直接映射到用户虚拟内存中。mmapmunmap的原型函数分别是

void *mmap(void *addr, int length, int prot, int flags, int fd, int off);
int munmap(void *addr, int length);

将上述声明添加到user/user.h中。本实验中addroff一直都是0,因此可以不在sys_mmap传入这两个参数。length是需要映射的字节数,可能和文件大小不相等,prot表示映射到的内存的权限为PROT_READ还是PROT_WRITEflag可以是MAP_SHAREDMAP_PRIVATE,如果是前者就需要将内存中修改的相应部分写回文件中。fd是需要映射的已打开的文件描述符。

本实验中,两个映射了同一个MAP_SHARED文件的进程可以不共享物理页(由于进程的内存隔离,要实现这个比较麻烦)

munmap(addr, length)需要将从addr开始的length长度的mmap全部取消映射,这里的假设是只能从VMA的头部或尾部取消映射,不能对中间部分取消映射。

实验步骤:

Makefile中添加$U_mmaptest\,向kernel/syscall.c中添加

extern uint64 sys_mmap(void);
extern uint64 sys_munmap(void);
...
[SYS_mmap]    sys_mmap,
[SYS_munmap]  sys_munmap,

kernel/syscall.h中添加

#define SYS_mmap   22
#define SYS_munmap 23

user/usys.pl添加

entry("mmap");
entry("munmap");

完成对mmapmumap的注册

kernel/proc.h中添加对vma结构体的定义,每个进程中最多可以有16个VMA。

#define MAXVMA 16
struct vma {
  int valid;                // whether the vma is valid
  uint64 addr;              // starting virtual address of vma
  int len;                  // length of vma, unit: bytes
  int prot;                 // permission
  int flags;                // flag
  struct file *f;           // pointer to mapped file
  int off;                  // offset of the valid mapped address
  int valid_len;            // length of the valid mapped address
};
struct vma procvma[MAXVMA]; // process vma

kernel/sysfile.c中添加注册sys_mmap,在mmap中先不要分配物理内存,只是完成对vma结构体的写入。找到新的空闲内存地址p->sz作为vma返回的地址并p->sz+=len,让后面需要使用这个返回的内存地址时陷入缺页异常,在trap中再分配物理内存完成lazy allocation

uint64
sys_mmap(void)
{
  int length, prot, flags, fd;
  struct file *f;
  if (argint(1, &length)<0 || argint(2, &prot)<0 || argint(3, &flags)<0 || argfd(4, &fd, &f)<0) {
    return -1;
  }
  if (!f->writable && (prot & PROT_WRITE) && (flags & MAP_SHARED)) return -1; // to assert that readonly file could not be opened with PROT_WRITE && MAP_SHARED flags
  struct proc *p = myproc();
  struct vma *pvma = p->procvma;
  for (int i = 0; i < MAXVMA; i++) {
    if(pvma[i].valid == 0) {
      pvma[i].addr = p->sz;
      pvma[i].f = filedup(f);  // increment the refcount for f
      pvma[i].flags = flags;
      pvma[i].len = PGROUNDUP(length);
      pvma[i].prot = prot;
      pvma[i].valid = 1;
      pvma[i].off = 0;
      pvma[i].valid_len = pvma[i].len;
      p->sz += pvma[i].len;
      return pvma[i].addr;
    }
  }
  return -1;
}

kernel/trap.c中添加lazy allocation

  } else if((which_dev = devintr()) != 0){
    // ok
  } else if (r_scause() == 13 || r_scause() == 15) {
    uint64 va = r_stval();
    if (va < p->trapframe->sp || va >= p->sz) {
      goto exception;
    }
    struct vma *pvma = p->procvma;
    va = PGROUNDDOWN(va);

    for (int i = 0; i < MAXVMA; i++) {
      if (pvma[i].valid && va >= pvma[i].addr + pvma[i].off && va < pvma[i].addr + pvma[i].off + pvma[i].valid_len) {  
        // allocate one page in the physical memory
        char *mem = kalloc();
        if (mem == 0) goto exception;
        memset(mem, 0, PGSIZE);
        int flag = (pvma[i].prot << 1) | PTE_U; // PTE_R == 2 and PROT_READ == 1
        if (mappages(p->pagetable, va, PGSIZE, (uint64)mem, flag) != 0) {
          kfree(mem);
          goto exception;
        }   
        int off = va - pvma[i].addr;
        ilock(pvma[i].f->ip);
        readi(pvma[i].f->ip, 1, va, off, PGSIZE);
        iunlock(pvma[i].f->ip);
        break;
      }
    }
  } else {
    exception:
    printf("usertrap(): unexpected scause %p pid=%d\n", r_scause(), p->pid);
    printf("            sepc=%p stval=%p\n", r_sepc(), r_stval());
    p->killed = 1;

需要判断page fault的地址是合理的,如果fault的地址低于了当前进程的栈底(p->trapframe->sp)或者高于等于当前进程的堆顶(p->sz)就说明是不合理的,需要进入exception。然后判断当前fault的地址是在哪一个VMA的合法范围中,找到这个VMA后分配一页物理页,并用mappages将这一页物理页映射到fault的用户内存中,然后用readi打开需要映射的文件,将对应的文件内容用readi放入这一页内存中去。

实现munmap

sys_munmap(void)
{
  uint64 addr;
  int length;
  if (argaddr(0, &addr) < 0 || argint(1, &length) < 0) {
    return -1;
  }
  struct proc *p = myproc();
  struct vma *pvma = p->procvma;
  int close = 0;
  // find the corresponding vma
  for (int i = 0; i < MAXVA; i++) {
    if (pvma[i].valid && addr >= pvma[i].addr && addr < pvma[i].addr + pvma[i].len) {
      addr = PGROUNDDOWN(addr);
      if (addr == pvma[i].addr + pvma[i].off) {
        // starting at begin of the valid address of vma
        if (length >= pvma[i].valid_len) {
          // whole vma is unmmaped
          pvma[i].valid = 0;
          length = pvma[i].valid_len;
          close = 1;
          p->sz -= pvma[i].len;
        } else {
          pvma[i].off += length;
          pvma[i].valid_len -= length;
        }
      } else {
        // starting at middle, should unmap until the end
        length = pvma[i].addr + pvma[i].off + pvma[i].valid_len - addr;
        pvma[i].valid_len -= length;
      }
      if (pvma[i].flags & MAP_SHARED) {
        // write the page back to the file
        if (_filewrite(pvma[i].f, addr, length, addr - pvma[i].addr) == -1) return -1; 
      }
      uvmunmap(p->pagetable, addr, PGROUNDUP(length)/PGSIZE, 0);
      if (close) fileclose(pvma[i].f);
      return 0;
    }
  }
  return -1;
} 

首先需要找到对应的vma,然后根据unmap的大小和起点的不同进行讨论。如果是从vma有效部分的起点开始,当整个vma都被unmap掉时,需要标记这个打开的文件被关闭(但是现在还不能关闭,因为后面可能需要写回硬盘中的文件),将当前的vma设置为invalid,减小p->sz(这一部分可能有点问题,因为如果是unmap中间的vma的话不需要减小p->sz,但是本实验中是可以通过测试的)。如果只是部分vma被unmap,则修改vma的offvalid_len,如果是从中间部分开始被unmap一直到结尾,则不需要修改off,只需要修改valid_len和需要被uvmunmaplength。然后判断是否是MAP_SHARED,如果是就用_filewrite写回原文件,这里是对filewrite函数进行了修改,使其能够从某个offset开始写

int
_filewrite(struct file *f, uint64 addr, int n, uint off) {
  int r, ret = 0;

  if(f->writable == 0)
    return -1;

  if(f->type == FD_PIPE){
    ret = pipewrite(f->pipe, addr, n);
  } else if(f->type == FD_DEVICE){
    if(f->major < 0 || f->major >= NDEV || !devsw[f->major].write)
      return -1;
    ret = devsw[f->major].write(1, addr, n);
  } else if(f->type == FD_INODE){
    // write a few blocks at a time to avoid exceeding
    // the maximum log transaction size, including
    // i-node, indirect block, allocation blocks,
    // and 2 blocks of slop for non-aligned writes.
    // this really belongs lower down, since writei()
    // might be writing a device like the console.
    int max = ((MAXOPBLOCKS-1-1-2) / 2) * BSIZE;
    int i = 0;
    while(i < n){
      int n1 = n - i;
      if(n1 > max)
        n1 = max;

      begin_op();
      ilock(f->ip);
      if ((r = writei(f->ip, 1, addr + i, off, n1)) > 0)
        off += r;
      iunlock(f->ip);
      end_op();

      if(r != n1){
        // error from writei
        break;
      }
      i += r;
    }
    // ret = (i == n ? n : -1);
  } else {
    panic("filewrite");
  }
  return ret;
}

最后uvmunmap掉这个vma中对应的虚拟内存(这里释不释放物理内存都可以通过测试),如果需要关闭文件就调用fileclose

最后,由于p->sz以内的内存不是都有对应的映射,因此可能会造成uvmunmapuvmcopy出现panic。直接将对应的panic注释掉并continue就可以

// uvmunmap
// if((*pte & PTE_V) == 0)
      // panic("uvmunmap: not mapped");

// uvmcopy
    if((*pte & PTE_V) == 0)
      panic("uvmcopy: page not present");
      continue;
    //  panic("uvmcopy: page not present");

测试过程中发现kfree中似乎会试图杀掉0这个物理内存,没有搞明白是在哪里杀的,因此直接修改kfree,当pa==0时直接return避免panic

kfree(void *pa)
{
  struct run *r;
  if (pa == 0) return;

  if(((uint64)pa % PGSIZE) != 0 || (char*)pa < end || (uint64)pa >= PHYSTOP)
    panic("kfree"); 

Lecture 15 OS Organization, microkernel

kernel应该做的工作到底是什么?

15.1 Monolithic kernel

传统的方法是monolithic kernel:比如Linux、xv6,mono kernel是一个很大的single program。这些kernel为user program提供了很高程度的抽象,比如将硬盘中的数据抽象为了文件描述符,因此提高了很高的移植性。monolithic kernel也提供了资源管理从而实现进程安全,并更有效率地分配资源。这些功能导致mono kernel体积会比较大。kernel是一个很大的程序,可以让kernel下面的子系统(比如内存管理、文件系统等)很容易互相访问,所有的kernel code有很高的权限,因此没有内部的安全限制。

mono kernel的缺点:太过大,太复杂。很多设计都是写死的,没有很强的扩展性,比如可能想要wait除了子进程之外的其他进程

15.2 Micro kernel

很小的kernel,只提供IPC(进程间通信)和thread/task,而文件系统、网络驱动、paging等功能都作为user process建立在这个kernel之上

micro kernel可能由于体积小而更好维护,更安全,有更高的扩展性

15.3 L4

一种非常有代表性的micro kernel,强调精简性,只有7个syscall。L4 kernel提供的支持包括:task (相当于process,每个task可以分为多个thread,也有自己的address space)、IPC

L4 syscalls

  • 新建一个task
  • 新建/销毁task中的thread
  • 通过IPC收发消息
  • 将内存物理页映射到另一个task中的address space,另一个task必须同意,这是通过IPC实现的
  • priviledged task可以将device register直接映射到自己的地址空间中从而实现对设备的直接访问
  • 设备通过IPC向设备驱动进程发送中断

L4切换task的方式和xv6基本相同,即选择一个RUNNABLE的thread,将原来thread的寄存器保存并恢复新的thread的寄存器,切换page table并跳到新的task

Pager

L4中有一个专门的pager task。当一个user task发生了page fault陷入了kernel中,kernel会将这个page fault以及产生page fault的thread #转化为IPC信号传给pager task,pager task会根据信号来进行相应的动作,比如lazy allocation等

15.4 L4 performance

micro kernel需要进行大量的IPC,在早期的设计中IPC需要消耗大量资源。早期IPC设计:

P1如果要向P2发送IPC,调用send(ID, MSG)来指定dst thread和message,然后将消息放到kernel的TX buffer中等待发送,然后yield CPU给P2。kernel读取TX buffer的msg,将消息复制给P2的RX buffer,P2调用recv(MSG)来将kernel中的消息复制到自己的内存中。在kernel和user space中之间来回穿越需要切换Page table,并且yield CPU切换task也需要切换寄存器上下文,速度比较慢。

改进的IPC设计:同步、无缓存。P1调用send(),并等待P2调用recv()。待同步完成之后直接跳转到P2,就像从recv()返回一样。如果MSG非常小,可以直接将其放在某一个指定的寄存器中,kernel将保证在跳转过程中这个寄存器的值不会被改变,因此跳转到P2后可以直接读取这个寄存器中的值,而不需要复制内存。如果MSG比较大,就将相应MSG的page mapping发送给P2,同样不需要复制内存。

15.5 L4/Linux

为了让L4可以和Linux兼容,采用了一种Linux kernel作为一个L4 user process运行于L4 kernel之上的架构,每个Linux process对应一个L4 task,每个process试图调用Linux syscall时,先通过L4 IPC链接到linux kernel,然后让Linux kernel调用syscall并进行返回。Linux kernel只是在最底层(比如对硬件的访问)被修改,它实际上变成了一个很大的server task。

L4/Linux fork()流程

  1. P1进程(其实是一个L4 task)调用fork()
  2. P1进程的libc库将fork()翻译为一个对Linux server的IPC syscall
  3. Linux server调用L4创建task和thread的syscall,新建了一个新的task P2
  4. Linux server为P2分配内存页
  5. Linux server用IPC告诉L4让其将分配好的内存页映射到P2的page table
  6. Linux server将P1的内存页数据复制到P2中
  7. Linux server发送特殊的IPC,其中包含SP和PC,从而让P2开始运行
  8. Linux server通过IPC回复P1(成功或失败代码)

因此Linux server实际上是所有L4 task的pager task

L4/Linux的性能大约是native Linux的50%(因为所有的syscall都要先通过L4再到Linux server,实际上相当于一个syscall变成2个)

L4微内核被广泛应用于嵌入式处理器,同时这个O/S server架构也被应用到了虚拟机中。

Lecture 16 Virtual Machine

16.1 Virtual Machine

对计算机的仿真,在VM上可以运行多个OS,就像qemu一样

VMM: virtual machine monitor

VMM可以是stand-alone,也可以运行在host-OS中,比如Linux系统。VM通常会提供比进程更强的隔离性。VM的目标:让guest OS无法区分现在是运行在VM上还是真正的机器上;guest OS无法从其运行的VM上逃离。但是实际上现在很多虚拟机并不能做到让guest OS无法区分

一种较慢的VM实现方式:VMM将每个guest OS的指令进行翻译再进行执行

可以将guest OS的指令直接运行在真正的CPU上,但是如果guest kernel运行了一些privileged指令,为了安全性考虑不能直接把supervisor寄存器的访问权限给guest kernel,因此需要在user mode下运行guest kernel,当guest kernel试图运行privileged指令时将陷入VMM的trap,VMM的trap handler将检查并模拟privileged instruction,比如将指令应用于virtual state或者真正的hardware上(比如修改satp)。VMM将给每个guestOS维护一个virtual state,里面包括了寄存器、satp、mode、RAM等信息,实际上都在VMM内存中。RISC-V还需要知道当前的guest OS是在guest user mode还是在guest kernel mode中。

Guest OS page table

当guest kernel希望写入satp来改变page table时,VMM不能让其直接写入因为VMM必须确认guest OS只能访问自己的内存

每个guest OS page table都维护了一个从guest va到guest pa的映射,但是guest pa并不是真正的物理内存,VMM还需要维护一个guest pa到host pa,也就是真正的物理内存的映射,再根据这两个映射的结合,形成一个从guest va到host pa的shadow page table,将这个shadow page table写入到satp中

Simulate devices

  1. 拦截对映射到内存的设备控制寄存器的读写行为,这是通过将相应的内存地址设置为invalid实现的。当guest OS访问这些寄存器时,将陷入page fault,VMM的trap handler再根据指令对真实的设备进行操作
  2. 也可以采用虚拟设备,guest OS需要知道它运行在VM中。e.g. xv6的virtio_disk.c提供了一系列的函数,qemu将其转化为对fs.img的读写。
  3. 也可以直接将物理设备的访问提供给guest OS,这需要设备硬件支持

16.2 HW support for Virtual Machine

x86提供了VT-x:硬件支持的虚拟化,能够让guest OS直接执行privileged指令而不用进入trap,因此速度更快

VT-x硬件拥有2套寄存器组,一套给VMM在root-mode下使用,另外一套给guest OS在non root mode下使用,可以让guest kernel对这一套寄存器组进行读写操作,但是VT-x限制了一些non root mode的操作从而防止其从VM中逃离。

VT-x page table

VMM为每个guest OS都维护了一个EPT(extended page table),当guest kernel向non-root下的satp写入PTE时(而不会陷入trap),VMM将再通过这个EPT再进行1次映射,从而保证guest OS不能修改真正的physical address

16.3 Dune

Dune是一个loadable kernel module for Linux,可以作为Linux kernel的一个模块,Dune必须在VT-x硬件设备上运行,一个普通的进程可以切换为Dune进程,这样Dune就像一个VMM,而这个普通的进程将拥有自己的寄存器组,也可以分为user mode和supervisor mode。Dune进程可以通过VMCALL来调用Linux syscall,每个进程拥有自己的page table,并通过EPT实现内存保护。

可以通过DUNE实现一个sandbox,因为DUNE进程中的supervisor部分是被保护的,因此可以让一个进程中的不受信任的部分(比如浏览器中的第三方插件)运行在USER MODE中,这就是一个sandbox,可以限制它对内存的访问

也可以通过DUNE来实现快速的garbage collector,因为在gc查找object的过程中可能object会被修改,这时需要通过PTE_D(dirty)来查找内存中的哪些地址被修改了,如果每次都陷入内核态就会造成速度很慢,但是在DUNE进程下不需要陷入内核态,因为每个进程的SUPERVISOR模式可以直接查看PTE

Lecture 17 Networking

17.1 Network Architecture and Protocols

Ethernet

局域网(Local Area Network)下各个主机都是通过Ethernet进行连接

Ethernet frame结构:start flag + 48位目标eth地址 + 48位源eth地址 + 16位ether type(payload中的数据协议类型)+ payload + end flag

// an Ethernet packet header (start of the packet).
struct eth {
  uint8  dhost[ETHADDR_LEN];
  uint8  shost[ETHADDR_LEN];
  uint16 type;
} __attribute__((packed));

eth地址由NIC的制造商决定

Internet下各个LAN通过router之间的TCP/IP协议进行连接

ARP

Ethernet协议足以将packet在LAN中传送,但是如果需要将packet传送给远程Host,需要IP地址。ARP协议是将IP地址翻译为eth地址的协议,嵌套在ethernet packet中(即ether header后面就是ARP header,ARP header是ether header的payload),ARP header请求一个IP地址对应的eth地址,这个ARP header会被所有LAN内的host接收,对应的Host将返回IP地址

// an ARP packet (comes after an Ethernet header).
struct arp {
  uint16 hrd; // format of hardware address
  uint16 pro; // format of protocol address
  uint8  hln; // length of hardware address
  uint8  pln; // length of protocol address
  uint16 op;  // operation

  char   sha[ETHADDR_LEN]; // sender hardware address
  uint32 sip;              // sender IP address
  char   tha[ETHADDR_LEN]; // target hardware address
  uint32 tip;              // target IP address
} __attribute__((packed));

#define ARP_HRD_ETHER 1 // Ethernet

enum {
  ARP_OP_REQUEST = 1, // requests hw addr given protocol addr
  ARP_OP_REPLY = 2,   // replies a hw addr given protocol addr
};

IP

IP packet的ethernet type=0x0800,IP header中包括了32位的IP地址,高位的IP地址是network number来帮助router将packet进行转发,protocol number让接收者知道这个IP packet的更高级协议是什么。在packet从一个router转发到另一个router的过程中,ether header会被除去,但是IP header一直是保留着的

// an IP packet header (comes after an Ethernet header).
struct ip {
  uint8  ip_vhl; // version << 4 | header length >> 2
  uint8  ip_tos; // type of service
  uint16 ip_len; // total length
  uint16 ip_id;  // identification
  uint16 ip_off; // fragment offset field
  uint8  ip_ttl; // time to live
  uint8  ip_p;   // protocol
  uint16 ip_sum; // checksum
  uint32 ip_src, ip_dst;
};

UDP

UDP是传输层协议,根据sockets API规定了当一个packet到达正确host时需要被发送到的端口号,操作系统将把这个端口号上获取的packets作为一个file descriptor以供进程读取。端口号53是DNS server指定的端口号

// a UDP packet header (comes after an IP header).
struct udp {
  uint16 sport; // source port
  uint16 dport; // destination port
  uint16 ulen;  // length, including udp header, not including IP header
  uint16 sum;   // checksum
};

17.2 Kernel network stack

在network stack中的packet存放在缓存中,比如mbuf,每一层将解析、校验header或添加header实现nesting

17.3 Lab 11: net

E1000网卡基本结构:

以太网网卡包括OSI模型的2个层:物理层和数据链路层。物理层由PHY芯片控制,定义了数据传送与接收所需要的光电信号、时钟基准等。数据链路层由MAC芯片控制,提供寻址机构、数据帧的构建、向网络层提供标准数据接口等功能。

从PCI总线到MAC芯片的分解图:

DMA(Direct Memory Access)是可以不通过CPU直接访问内存的机制,在进行DMA传输时DMA Engine控制PCI总线,将内存中的数据和FIFO data buffer (64KB)中的数据互传。

发送数据的流程:

CPU将IP数据包打包放入内存中,通知DMA Engine进行DMA传输,数据放入FIFO data buffer中。MAC将IP数据包拆分为最小64KB,最大1518KB的数据帧,每一帧包含了目标MAC地址、自己的MAC地址和数据包中的协议类型以及CRC校验码。目标MAC地址通过ARP协议获取。PHY接受MAC传送的数据,将并行数据转化为串行数据后进行编码,在转变为模拟信号将数据进行传输。

Ring结构

RAM中的tx/rx buffer是一个环形结构,有head和tail2个指针,其中head的位置由网卡控制,在进行发送时,每发送完成一个packet网卡就会自动将head向前移动一个mbuf,而需要将某个packet发送时,软件将tail向前移动一个mbuf;在进行接收时,每接收到一个packet网卡自动将head向前移动一个mbuf,软件读取tail所指向的mbuf,并向前移动。移动到最后一个mbuf后从头开始,形成一个wrap-up的结构。

descriptor结构是在网卡的寄存器中的,用于描述每一个RAM中的mbuf

实验按照hint一步步进行即可

// e1000.c
int
e1000_transmit(struct mbuf *m)
{
  //
  // Your code here.
  //
  // the mbuf contains an ethernet frame; program it into
  // the TX descriptor ring so that the e1000 sends it. Stash
  // a pointer so that it can be freed after sending.
  //
  acquire(&e1000_lock);
  uint32 tx_index = regs[E1000_TDT];  // tail pointer
  if ((tx_ring[tx_index].status & E1000_TXD_STAT_DD) == 0) {
    release(&e1000_lock);
    return -1;
  } else if (tx_mbufs[tx_index] != 0) {
    mbuffree(tx_mbufs[tx_index]);
  }
  // fill in the mbuf ring
  tx_mbufs[tx_index] = m;
  // fill in the descriptor
  tx_ring[tx_index].addr = (uint64)m->head;
  tx_ring[tx_index].length = m->len;
  tx_ring[tx_index].cmd = E1000_TXD_CMD_EOP | E1000_TXD_CMD_RS;
  regs[E1000_TDT] = (tx_index + 1) % TX_RING_SIZE; // foward the tail by 1
  if (&tx_ring[tx_index] == 0 || tx_mbufs[tx_index] == 0) {
    release(&e1000_lock);
    return -1;
  }
  release(&e1000_lock);
  return 0;
}

static void
e1000_recv(void)
{
  //
  // Your code here.
  //
  // Check for packets that have arrived from the e1000
  // Create and deliver an mbuf for each packet (using net_rx()).
  //
  while (1) {
    uint32 rx_index = (regs[E1000_RDT]+1)%RX_RING_SIZE;
    if ((rx_ring[rx_index].status & E1000_RXD_STAT_DD) == 0) {
      break;
    }
    rx_mbufs[rx_index]->len = rx_ring[rx_index].length;
    net_rx(rx_mbufs[rx_index]);
    rx_mbufs[rx_index] = mbufalloc(0);
    // fill in the descriptor
    rx_ring[rx_index].addr = (uint64)rx_mbufs[rx_index]->head;
    rx_ring[rx_index].status = 0;
    regs[E1000_RDT] = rx_index;
  }
}

注意recv()中不需要lock,因为recv()是在bottom half的interrupt handler中,只有一个进程在跑这个handler,因此不存在共享的数据结构。

0%