0 前置对比表时空与特权级全景维度进程 (Process)线程 (Thread)协程 (Coroutine)一句话定义资源分配的最小单位CPU 调度的最小单位用户态调度的轻量级实体时空边界隔离完全隔离拥有独立页表/物理内存共享进程空间仅保留专属栈/寄存器共享线程空间在同一个线程栈中切换掌控生杀大权者操作系统内核 (Kernel)操作系统内核 (Kernel)用户态程序 (User Runtime)切换成本 (时间)极高毫秒级/微秒级较高微秒级极低纳秒级内存占用 (空间)较大至少需要几MB一般Linux 默认 8MB 虚拟栈极小通常 2KB ~ 几KB 初始栈通信方式进程间通信 (IPC)管道、共享内存等因为有页表屏障不能直连直接读写全局变量需要锁保护并发直接读写内存/ 通过 Channel 通信1.Uthread: switching between threads要求在用户态实现线程切换上下文的机制。uthread.c函数链路1.初始化阶段 main()→thread_init()[current_thread all_thread[0], 状态设为 RUNNING]2.线程创建阶段 main()→thread_create(thread_a)[分配 all_thread 空间伪装现场: ra thread_a, sp stack_a 8192, 状态为 RUNNABLE]3.首次调度激活 main()→thread_schedule()[在 all_thread 中挑出 thread_a]; thread_switch(main-context, thread_a-context)[保存 main 现场(ra指向schedule); 加载 a 寄存器(rathread_a, spstack_a)] ; CPU 执行 ret→CPU 瞬间飞入 thread_a() 的第一行代码开始执行4.线程协作让步Yield 循环thread_a() 运行到第 i 次循环→打印 thread_a i→thread_yield(); thread_schedule()[挑出 thread_b]; thread_switch(thread_a-context, thread_b-context)→CPU 跳转到 thread_b() 运行; thread_switch 再次返回 thread_a←thread_a 在刚才被挂起的 thread_yield() 后面睁眼醒来继续第 i1 次循环5.线程退出销毁 thread_a() 100次循环结束→current_thread-state FREE; thread_schedule()[调度器跳过已 FREE 的 a挑选他人a 的物理空间被安全复用]先定义线程上下文结构体struct tcontext{ uint64 ra; //返回地址寄存器 uint64 sp; //栈顶寄存器 //callee-saved被调用者寄存器 uint64 s0; uint64 s1; uint64 s2; uint64 s3; uint64 s4; uint64 s5; uint64 s6; uint64 s7; uint64 s8; uint64 s9; uint64 s10; uint64 s11; };在threa结构体加入contextstruct thread { struct tcontext context; //在thred结构体加入context char stack[STACK_SIZE]; /* the threads stack */ int state; /* FREE, RUNNING, RUNNABLE */ /* 空闲、运行、就绪 */ };模仿kernel/swtch.S在user/uthread_switch.S中写入下面代码.globl thread_switch thread_switch: /* YOUR CODE HERE */ //在调用thread_switch(uint64, uint64)汇编函数第一个参数是旧context结构体的首地址存入a0寄存器第二个参数是新contex结构体首地址在a1寄存器 //把当前寄存器状态保存到旧context结构体中然后从新context结构体中恢复寄存器状态 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 /* 把ra寄存器下一条指令地址写入pc指针 */在thread_scheduler()函数中调用thread_switch语句切换进程void thread_schedule(void) //调度线程 { struct thread *t, *next_thread; /* Find another runnable thread. */ next_thread 0; t current_thread 1; //t指向当前线程的下一个线程 for(int i 0; i MAX_THREAD; i){ //循环遍历线程数组寻找下一个可运行的线程 if(t all_thread MAX_THREAD) //如果t指针超过线程数组的末尾则重新指向线程数组的开头 t all_thread; if(t-state RUNNABLE) { //如果线程状态为RUNNABLE则将其设置为next_thread并跳出循环 next_thread t; break; } t t 1; //继续遍历下一个线程 } if (next_thread 0) { printf(thread_schedule: no runnable threads\n); exit(-1); } if (current_thread ! next_thread) { /* switch threads? */ next_thread-state RUNNING; //把下一个线程设置运行态 t current_thread; current_thread next_thread; /* YOUR CODE HERE * Invoke thread_switch to switch from t to next_thread: * thread_switch(??, ??); */ //调用switch汇编函数 thread_switch((uint64)t-context, (uint64)current_thread-context);//t是指向当前线程结构体current_thread指向下一个就绪态线程结构体 } else next_thread 0; }在thread_create()函数中添加线程上下文的返回地址并为线程分配栈空间void thread_create(void (*func)()) //创建线程函数 { struct thread *t; for (t all_thread; t all_thread MAX_THREAD; t) { //在线程数组中寻找一个FREE的线程 if (t-state FREE) break; } t-state RUNNABLE; //设置线程状态为RUNNABLE // YOUR CODE HERE //ra寄存器保存函数的返回地址sp寄存器保存栈顶指针 t-context.ra (uint64)func; // t-stack数组的地址在低处所以初始栈顶指针sp应该在t-stackSTACK_SIZE处 t-context.sp (uint64)(t-stackSTACK_SIZE); }2.Using thread要求用互斥锁保护哈希桶资源多线程并发插入会出现插入条目丢失数量不对。ph.c链路main 读取线程数生成10万随机key数组 批量创建写线程pthread_create→ 每个子线程执行put_thread()put_thread循环分片调用put(key, 线程号)put计算桶下标、遍历链表查重无重复则调用insert头插节点。pthread_join等待所有写线程全部结束 批量创建读线程pthread_create→ 每个子线程执行get_thread()get_thread遍历全部 key调用get(key)查找统计丢失 keypthread_join等待所有读线程结束。打印读写吞吐量程序退出。为什么会出现数据丢失当线程a执行insert头插节点时新节点next指向原链表头在更新哈希桶链表头指针前发生时间片线程调度另一个线程b更新完它的哈希链表头指针后调度回原线程更新桶链表头指针将覆盖b的链表节点导致b数据丢失。对每个哈希桶初始化一把锁pthread_mutex_t lock[NBUCKET] {PTHREAD_MUTEX_INITIALIZER};在插入节点前后加锁解锁static void put(int key, int value) //把key-value插入哈希表 { int i key % NBUCKET; //计算key对应的桶号 // is the key already present? struct entry *e 0; for (e table[i]; e ! 0; e e-next) { //遍历桶i的链表查找key是否存在 if (e-key key) break; } if(e){ // update the existing key. key存在更新value e-value value; } else { // the new is new. 不存在调用insert头插新节点 pthread_mutex_lock(lock[i]); // 对应的散列桶的上锁 insert(key, value, table[i], table[i]); pthread_mutex_unlock(lock[i]); //解锁 } }3.Barrier要求实现一个屏障要求所有线程到达后再进入下一轮循环。前置pthread_cond_wait (cond, mutex) 调用本函数线程阻塞释放mutexpthread_cond_broadcast(cond) 唤醒所有阻塞线程数据变量结构体struct barrier { pthread_mutex_t barrier_mutex; //互斥锁 pthread_cond_t barrier_cond; //条件变量线程在等待条件变量时会阻塞直到其他线程发出信号 int nthread; // 当前轮次已经到达屏障的线程数 int round; // Barrier round 当前是第几轮屏障 } bstate;线程函数static void * thread(void *xa) { long n (long) xa; long delay; int i; for (i 0; i 20000; i) { int t bstate.round; //获取当前轮次 assert (i t); //断言当前轮次和循环次数相等 barrier(); //调用屏障函数等待所有线程到达屏障 usleep(random() % 100); //随机延迟模拟线程执行时间不同 } return 0; }屏障实现static void barrier() { // 申请持有线程到达数锁 pthread_mutex_lock(bstate.barrier_mutex); bstate.nthread; //次数1 if(bstate.nthreadnthread){ //当线程到达数和线程数相等 // 所有线程都已经到达barrier bstate.round; //循环轮次加1 bstate.nthread 0; //重置 pthread_cond_broadcast(bstate.barrier_cond); //释放其他阻塞线程 }else{ // 释放线程到达数锁将当前线程阻塞 pthread_cond_wait(bstate.barrier_cond, bstate.barrier_mutex); } // 释放锁 pthread_mutex_unlock(bstate.barrier_mutex); }