线程管理器

线程状态

从调度器的角度来看，每个线程都有一个独一无二的 Tid 来区分它和其他线程。

1. // in process/mod.rs
2. pub type Tid = usize;
3. pub type ExitCode = usize;

同时，线程的状态有下面几种：

1. // src/process/struct.rs
2. #[derive(Clone)]
3. pub enum Status {
4. // 就绪：可以运行，但是要等到 CPU 的资源分配给它
5. Ready,
6. // 正在运行
7. Running(Tid),
8. // 睡眠：当前被阻塞，要满足某些条件才能继续运行
9. Sleeping,
10. // 退出：该线程执行完毕并退出
11. Exited(ExitCode),
12. }

调度算法接口设计

在一个线程运行的过程中，调度器需要定期查看当前线程的已运行时间，如果已经达到一个阈值，那么出于公平起见，应该将 CPU 资源交给其他线程，也即切换到其他线程。 查看的间隔不能太长，这样的话等于线程调度根本没起到作用；但是也不能过于频繁， CPU 的资源大量投资在调度器上更是得不偿失。 我们的线程调度算法基于时钟中断，我们会在时钟中断中进入调度器看看当前线程是否需要切换出去。 因此，调度算法的接口 Scheduler 如下：

1. // src/process/scheduler.rs
2. pub trait Scheduler {
3. // 如果 tid 不存在，表明将一个新线程加入线程调度
4. // 否则表明一个已有的线程要继续运行
5. fn push(&mut self, tid: Tid);
6. // 从若干可运行线程中选择一个运行
7. fn pop(&mut self) -> Option<Tid>;
8. // 时钟中断中，提醒调度算法当前线程又运行了一个 tick
9. // 返回的 bool 表示调度算法认为当前线程是否需要被切换出去
10. fn tick(&mut self) -> bool;
11. // 告诉调度算法一个线程已经结束
12. fn exit(&mut self, tid: Tid);
13. }

线程池接口设计

调度算法 Scheduler 只管理 Tid ，和线程并没有关系。因此，我们使用线程池 ThreadPool 来给线程和 Tid 建立联系，将 Scheduler 的 Tid 调度变成线程调度。 事实上，每个线程刚被创建时并没有一个 Tid ，这是线程池给线程分配的。

1. // src/process/thread_pool.rs
2. // 线程池每个位置的信息
3. struct ThreadInfo {
4. // 占据这个位置的线程当前运行状态
5. status: Status,
6. // 占据这个位置的线程
7. thread: Option<Box<Thread>>,
8. }

10. pub struct ThreadPool {
11. // 线程池
12. // 如果一个位置是 None 表示未被线程占据
13. threads: Vec<Option<ThreadInfo>>,
14. // 调度算法
15. // 这里的 dyn Scheduler 是 Trait object 语法
16. // 表明 Box 里面的类型实现了 Scheduler Trait
17. scheduler: Box<dyn Scheduler>,
18. }

线程池的方法

作为一个线程池，需要实现调度相关的一系列操作：

  • alloc_tid：为新线程分配一个新的 Tid
  • add：添加一个可立即开始运行的线程
  • acquire：从线程池中取一个线程开始运行
  • retrieve：让当前线程交出 CPU 资源
  • tick：时钟中断时查看当前所运行线程是否要切换出去
  • exit：退出线程

下面，我们依次来看看线程池的方法：

1. // src/process/thread_pool.rs

3. impl ThreadPool {
4. // 新建一个线程池，其最大可容纳 size 个线程，使用调度器 scheduler
5. pub fn new(size: usize, scheduler: Box<dyn Scheduler>) -> ThreadPool {
6. ThreadPool {
7. threads: {
8. let mut v = Vec::new();
9. v.resize_with(size, Default::default);
10. v
11. },
12. scheduler,
13. }
14. }
15. // 在线程池中找一个编号最小的空着的位置
16. // 将编号作为 Tid 返回
17. fn alloc_tid(&self) -> Tid {
18. for (i, info) in self.threads.iter().enumerate() {
19. if info.is_none() {
20. return i;
21. }
22. }
23. panic!("alloc tid failed!");
24. }

26. // 加入一个可立即开始运行的线程
27. // 线程状态 Uninitialized -> Ready
28. pub fn add(&mut self, _thread: Box<Thread>) {
29. // 分配 Tid
30. let tid = self.alloc_tid();
31. // 修改线程池对应位置的信息
32. self.threads[tid] = Some(
33. ThreadInfo {
34. // 状态：随时准备运行，等待 CPU 资源中
35. status: Status::Ready,
36. // 传入线程
37. thread: Some(_thread),
38. }
39. );
40. // 将线程的 Tid 加入调度器
41. // 提醒调度器给这个线程分配 CPU 资源
42. self.scheduler.push(tid);
43. }

45. // 从线程池中取一个线程开始运行
46. // 线程状态 Ready -> Running
47. pub fn acquire(&mut self) -> Option<(Tid, Box<Thread>)> {
48. // 调用 Scheduler::pop ，从调度算法中获取接下来要运行的 Tid
49. if let Some(tid) = self.scheduler.pop() {
50. // 获取并更新线程池对应位置的信息
51. let mut thread_info = self.threads[tid].as_mut().expect("thread not exist!");
52. // 将线程状态改为 Running
53. thread_info.status = Status::Running(tid);
54. return Some((tid, thread_info.thread.take().expect("thread not exist!")));
55. }
56. else {
57. return None;
58. }
59. }
60. // 这个线程已运行了太长时间或者已运行结束，需要交出CPU资源
61. // 但是要提醒线程池它仍需要分配 CPU 资源
62. pub fn retrieve(&mut self, tid: Tid, thread: Box<Thread>) {
63. // 线程池位置为空，表明这个线程刚刚通过 exit 退出
64. if self.threads[tid].is_none() {
65. // 不需要 CPU 资源了，退出
66. return;
67. }
68. // 获取并修改线程池对应位置的信息
69. let mut thread_info = self.threads[tid].as_mut().expect("thread not exist!");
70. thread_info.thread = Some(thread);
71. // 此时状态可能是 Status::Sleeping(线程可能会自动放弃 CPU 资源，进入睡眠状态),
72. // 直到被唤醒之前都不必给它分配。
73. // 而如果此时状态是Running,就说明只是单纯的耗尽了这次分配CPU资源,但还要占用CPU资源继续执行。
74. if let Status::Running(_) = thread_info.status {
75. // Running -> Ready
76. thread_info.status = Status::Ready;
77. // 通知线程池继续给此线程分配资源
78. self.scheduler.push(tid);
79. }
80. }
81. // Scheduler 的简单包装：时钟中断时查看当前所运行线程是否要切换出去
82. pub fn tick(&mut self) -> bool {
83. let ret = self.scheduler.tick();
84. ret
85. }
86. // 这个线程已经退出了，线程状态 Running -> Exited
87. pub fn exit(&mut self, tid: Tid) {
88. // 清空线程池对应位置
89. self.threads[tid] = None;
90. // 通知调度器
91. self.scheduler.exit(tid);
92. }
93. }

现在我们有了一个线程池 ThreadPool ，它内含调度器，是一个不错的线程管理器。下一节我们将介绍调度线程 idle 以及调度单元 Processor。