第8章 进程间通信
From 《Orange'S：一个操作系统的实现》
我 们提到过，当一个进程需要操作系统的帮助，它可以通过系统调用让内核来替它完成一些工作。迄今为止，我们已经熟悉了系统调用的工作机制，并且已经实现了不 止一个系统调用。接下来你会发现，用户进程将会有更多事情依赖于内核。比如我们想实现一个文件系统，最起码读写硬盘的工作要求助于内核。这里我们可以逐渐 地增加系统调用，但也可以采用另一种方案，就是将这些工作剥离出来，交给一些系统进程来完成，让内核只负责它必须负责的工作，比如进程调度。这种将内核工 作简单化的思想，便是微内核的基本思想。而所有工作通过系统调用扔给内核态的做法，被称为宏内核。
在基于宏内核的操作系统中，完成具体任务 时，用户进程通过系统调用让内核来做事，直来直去，我们之前已经很熟悉了。在基于微内核的操作系统中，这个过程稍微复杂一些。在完成具体任务时，内核的角 色很像是个中介。就比如我们将要实现的文件系统吧，设想用户进程P读取一个文件，首先通过内核告诉进程FS，然后FS再通过内核告诉驱动程序（也是一个独 立的进程），驱动程序读取硬盘，返回结果。这样一来，一项工作的完成变得有些曲折，需要多个进程协同工作。于是，进程间通信也就变得至关重要了。
到 如今，我们的操作系统慢慢长大，接下来我们要用它来管理磁盘和磁盘上的文件并管理内存等，这些都要向应用程序提供接口，到了必须决定用微内核还是宏内核的 时候了。怎么办呢？当然不能抛个硬币了事。我们不妨先找两个具体的例子来看看它们分别是怎么回事，看完了，明白了，再做决定也不迟。
微内核和宏内核的例子都非常好找。我们一直拿在手边的Minix，以及每天在用的Linux，便是两者的典型例子。Minix是微内核的，Linux则是宏内核的。
说 起这两个例子，有一段轶事不能不提。那就是当年Tanenbaum和Linus一老一少的口舌之争。话说Linus写了个操作系统叫做Linux，使用的 是宏内核，他把这个消息发在了comp.os.minix新闻组上，这时Tanenbaum说话了，把Linux批评了一通，年轻气盛的Linus于是发 信回击，这样一来二去，为我们留下一段微内核与宏内核的经典争论。
争论的全部内容在这里我们就不全部转述了，读者感兴趣的话可以用搜索引擎很容易地搜到（或者在维基百科上看一下），我们把其中的重点说一下。在谈到微内核和宏内核时，Andy（Andrew S. Tanenbaum）是这样说的：
老一点的操作系统都是宏内核的，也就是说，整个操作系统是一个运行在核心态的单独的a.out文件，这个二进制文件包含进程管理、内存管理、文件系统以及其他。具体实例包括UNIX、MS-DOS、 VMS、MVS、OS/360、MULTICS等。
另 一种便是微内核，在这种系统中操作系统的大部分都运行在单独的进程，而且多数在内核之外。它们之间通过消息传递来通信。内核的任务是处理消息传递、中断处 理、底层的进程管理，以及可能的I/O。这种设计的实例有RC4000、Amoeba、Chorus、Mach，以及还没有发布的Windows/NT。
我 完全可以（但不必）再讲述一段关于两者之间相对优势的很长的故事，然而在实际设计操作系统的人中间说说就够了，争论实际上已经结束。微内核已经取得了胜 利。对于宏内核而言唯一的争论焦点在于效率，不过已经有足够的证据表明微内核可以像宏内核一样快（比如Rick Rashid已经发表了Mach 3.0和宏内核系统的比较报告）所以那不过是喊喊而已罢了。
Minix是微内核的，文件系统和内存管理是单独的进程，它们运行在内核之外。 I/O驱动也是单独的进程（在内核之内，但仅仅是因为Intel CPU的糟糕设计使得很难不这样做）。Linux是个宏内核的系统。这相当于向七十年代倒退了一大步。就好比将一个已存在的工作得很好的C程序用 Basic重写一遍。在我看来，在1991年写一个宏内核的系统真不是个好主意。”
以上前两段基本上可以被认为是宏内核和微内核的基本概念。从概念上我们不难猜到，宏内核看上去试图包办一切，而微内核恰恰相反，它的任务只是“处理消息传递、中断处理、底层的进程管理，以及可能的I/O”，而其他事情都交给内核之外单独的进程来完成。
在 这段文字中Andy不但阐述了宏内核和微内核的概念，摆明了对于这个问题鲜明的观点，而且他也毫不掩饰自己对宏内核的不屑。而且这种不屑让他认为 Linux简直是技术的倒退。在随后的文字中，对于Linux的可移植性Andy也做了不客气的批评。也难怪Linus对此非常恼火。从Linus的第一 个回复开始，这场争论开始变得精彩起来。
Linus的回复是这样开始的：
好吧，既然是这么一个主题，我恐怕不能不做回答了。向已经听了太多关于Linux的Minix使用者们道歉。我很乐意上钩（Andy说了这些话，好像在引诱Linus开始一场争论——笔者注），该是吵一架的时候了！
啊哈，看来Linus真的被激怒了，我仿佛看到了他挽起袖子的样子:)。是啊，看到自己辛辛苦苦的劳动成果被人冠以“过时了”的形容，谁还能平心静气呢？
针对微内核和宏内核之争，他是这样回应的：
是 的，Linux是宏内核，我同意微内核更好些。如果不是你使用了具有争论性的主题，我可能会同意你大部分的观点。从理论上（以及美学上）讲Linux是输 了。如果去年春天GNU内核已经做好，我可能不会这么麻烦地开始我的工作：问题是它没有做好而且到现在都没有。在已经实现这一方面Linux赢大了。
>> Minix是微内核系统……Linux则是宏内核的。
如果这是评价内核好坏的唯一标准，那么你是对的。你没有提到的是Minix的微内核实现得并不好，而且（内核内）多任务存在问题。如果我做一个多线程文件系统存在问题的操作系统，我可能不会这么快就声讨别人：事实上我会尽最大努力让别人忘掉我的失败。
这 一段我觉得非常重要，因为看得出来，Linus内心还是承认微内核的优势的，而且他提到了“美学”（aesthetical）这个词，因为的确，微内核的 思想更加优雅，这在我们下文中的分析中也可以看到。不过尽管如此，他还是批评了Minix本身，认为它的微内核实现的并不令人满意。
在后来谈到可移植性的时候，Linus的话也颇具初生牛犊不怕虎的劲头：
可移植性是为不能编写新程序的人设计的
——我，现在（使用傲慢的语气）
真 的很精彩不是么？我甚至感觉有点像在看武侠片，一老一少，出招拆招，虽是打架，但颇有章法。不难看出，刚刚这句“我，现在（使用傲慢的语气）”甚至带有一 丝挑衅意味，看这句话我甚至在想像着Linus敲出这行字的时候该是带着怎样傲慢的神色——不过谁在年轻的时候不是这样气盛的呢？呵呵。
看过热闹之后，让我们来实地勘查一下，两种内核看上去是什么样子的。我们就以系统调用作为突破口，看看它们的代码。
为 了简单起见，我们拿Linux 0.01作为Linux方代表──最新的Linux内核代码量太大了，不利于短时间内弄懂。其中的系统调用不止一个，我们以fork()为例来分析一下。 为了让读者一下子就理清这个系统调用的脉络，对于代码细节这里就不细言了，我们直接来看程序的流程图（如下图所示）。
相关的具体代码读者可以在Linux 0.01的以下文件中找到：
init/main.c
include/unistd.h
kernel/sched.c
include/linux/sys.h
kernel/system_call.s
从 图中我们可以看出，调用fork()实际上是调用了中断0x80，通过事先初始化好的IDT，程序转移到了_system_call，最终通过一个函数指 针数组sys_call_table[]转化成了调用sys_fork()。这跟我们实现过的系统调用是很相似的，此处不再赘述。
Linux的fork系统调用很容易理解，但Minix的就不这么简单了，它刚开始甚至可能让你感到迷惑。我们来打开Minix代码文件src/kernel/proc.c，在函数sys_call()的开头，你可以看到这样的代码：
PUBLIC int sys_call(function, src_dest, m_ptr)
int function; /* SEND, RECEIVE, or BOTH */
int src_dest; /* source to receive from or dest to send to */
message *m_ptr; /* pointer to message */
{
/* The only system calls that exist in MINIX are sending and receiving
* messages. These are done by trapping to the kernel with an INT instruction.
* The trap is caught and sys_call() is called to send or receive a message
* (or both). The caller is always given by proc_ptr.
*/
……
}
开头这段注释非常重要，一个“only”道破天机（或者将你搞晕）：在Minix中，不再像Linux那样有许许多多的系统调用 （sys_call_table[]中列出的有几十个），而是仅有发送和接收消息的系统调用。通过sys_call的参数function的注释我们可以 知道，系统调用的种类总共有三个，那就是SEND、RECEIVE，以及BOTH。
可是系统调用虽少，实现的功能却不能少，那么Minix是怎样通过仅仅三个系统调用就实现与以Linux为代表的宏内核OS一样多的功能呢？我们仍以fork()为例，来看一下Minix是怎么做的。
相对于Linux，Minix的机制显得有点复杂，我们还是直接来看图。
跟 Linux不同，这里多出来一个内存管理器（MM），fork()所要做的工作是由它来负责的（如果是另外的系统调用，那么具体工作可能就不是由MM来负 责，比如系统调用read()就是由FS来负责的，跟MM类似，FS是单独运行的另一个进程），那么MM是如何得到用户进程的通知的呢？正是消息机制在进 程之间起到了重要作用，它类似于邮政系统，在信封（或包裹单）上写明目的地，消息就送达了。
图中使用了三种箭头，实线
表示消息的发送过程，点线
表示消息的获取过程，虚线
表示发送和接收消息都会经历的过程。
进 程对fork()的调用将最终转化成调用src/kernel/proc.c中的sys_call()，消息（即图中的m）的地址这时已经作为参数被传递 进来，sys_call()可以据此得知m的内容，并在适当的时候将内容传递给MM。MM的工作其实说起来很简单，它不断地获取并处理消息，所以它能够得 到用户进程发送的m，并将其存放在mm_in中。当MM通过获得的mm_in得知了消息的内容是要进行fork操作，它就进一步调用其do_fork() 完成整个过程。
消息的一送一收之间，fork()的大致过程我们就已经基本了解了。其实我们也完全可以猜测出其他系统调用的情况。不外乎是通过调用_syscall()转化成发送消息，将来会有相应的进程取出消息进行处理。
说 到这里，有一个情况需要说明，就是我们拿Linux中的fork()和Minix中的fork()来比较是有些不公平的。因为你也一定已经看到，实现方式 真正与Linux的fork()相同的是Minix的_sendrec()和_receive()，它们都是通过中断进入内核，在内核中完成任务。而 Minix中的fork()是通过两个进程分别调用_sendrec()和_receive()这两个系统调用来实现的，从这个意义上来说，Minix中 的fork()其实不算系统调用（这也是函数sys_call()的开头注释中说Minix的系统调用只有三个的原因），它只是在完成一个紧密依赖系统调 用的工作罢了。不过从用户的角度，这种差别是看不到的，而且只要调用fork()时能实现需要的功能，这种差别就无关紧要。因此用户完全可以称 fork()为一个系统调用。
实现方法的差别源于设计思路的不同。在Minix中，真正的系统调用只有三个，这意味着内核不必事无巨细地处 理用户进程要求的所有工作，只需要做好其“邮局”的职能，将消息按照需求来回传送就够了。在Linux中内核所做的工作，在Minix中被交给专门的进程 来完成。你可能一下子就明白了，原来微内核的“微”字是让内核功能最简化的意思。
到 这里，微内核和宏内核各自的工作原理我想读者已经明白了。同时它们的优缺点也基本上清楚了。宏内核的优势在于其逻辑简单，直截了当，实现起来也容易，而且 也因为它的直接，避免了像微内核那样在消息传递时占用资源。而微内核的优势在于，它的逻辑虽相对复杂但非常严谨，结构上显得非常优雅和精致，而且程序更容 易模块化，从而更容易移植。
从编程的难易程度上来看，宏内核看上去具有一定优势，因为它很直接，不需要绕弯子，但从长期来看，当内核逐渐变 大，微内核的结构会更加清晰。虽然选择微内核意味着有调试起来有些困难的消息机制摆在面前，但从设计理念上来看，微内核更加“摩登”，更“酷”。而且，从 学习编程的角度看，搞一个微内核可以为将来架构其他东西作为很有益的参考。基于这些原因，我们选择微内核。
选择了微内核，首要的任务就比较明显了，那就是实现一个进程间通信（IPC）机制。其实这也没有想像中那么难，我们下面就来看一看。
Minix 的IPC机制我们已经明白了，它的核心乃在于“int SYSVEC”这个软中断以及与之对应的sys_call()这个函数。增加一个系统调用对我们来讲已是信手拈来的事，按照表7.6一步一步来就好了。我 们把这个新的系统调用起名为sendrec。sendrec和sys_sendrec的函数体分别见下面两段代码：
25 sendrec:
26 mov eax, _NR_sendrec
27 mov ebx, [esp + 4] ; function
28 mov ecx, [esp + 8] ; src_dest
29 mov edx, [esp + 12] ; p_msg
30 int INT_VECTOR_SYS_CALL
31 ret
53 /*****************************************************************************
54 * sys_sendrec
55 *****************************************************************************/
56 /**
57 * <Ring 0> The core routine of system call ‘sendrec()’.
58 *
59 * @param function SEND or RECEIVE
60 * @param src_dest To/From whom the message is transferred.
61 * @param m Ptr to the MESSAGE body.
62 * @param p The caller proc.
63 *
64 * @return Zero if success.
65 *****************************************************************************/
66 PUBLIC int sys_sendrec(int function, int src_dest, MESSAGE* m, struct proc* p)
67 {
68 assert(k_reenter == 0); /* make sure we are not in ring0 */
69 assert((src_dest >= 0 && src_dest < NR_TASKS + NR_PROCS) ||
70 src_dest == ANY ||
71 src_dest == INTERRUPT);
72
73 int ret = 0;
74 int caller = proc2pid(p);
75 MESSAGE* mla = (MESSAGE*)va2la(caller, m);
76 mla->source = caller;
77
78 assert(mla->source != src_dest);
79
80 /**
81 * Actually we have the third message type: BOTH. However, it is not
82 * allowed to be passed to the kernel directly. Kernel doesn’t know
83 * it at all. It is transformed into a SEND followed by a RECEIVE
84 * by ‘send_recv()’.
85 */
86 if (function == SEND) {
87 ret = msg_send(p, src_dest, m);
88 if (ret != 0)
89 return ret;
90 }
91 else if (function == RECEIVE) {
92 ret = msg_receive(p, src_dest, m);
93 if (ret != 0)
94 return ret;
95 }
96 else {
97 panic(”{sys_sendrec}␣invalid␣function:␣”
98 ”%d␣(SEND:%d,␣RECEIVE:%d).”, function, SEND, RECEIVE);
99 }
100
101 return 0;
102 }
表7.6的最后一步中提到，如果参数个数与以前的系统调用比有所增加，则需要修改kernel.asm中的sys_call。额外要注意，我们新加的参数是通过edx这个参数传递的，而save这个函数中也用到了寄存器dx，所以我们同时需要修改save，见如下代码：
311 ; =============================================================================
312 ; save
313 ; =============================================================================
314 save:
315 pushad ; ‘.
316 push ds ; |
317 push es ; | 保存原寄存器值
318 push fs ; |
319 push gs ; /
320
321 ;; 注意，从这里开始，一直到‘mov esp, StackTop’，中间坚决不能用push/pop 指令，
322 ;; 因为当前esp 指向proc_table 里的某个位置，push 会破坏掉进程表，导致灾难性后果！
323
=> 324 mov esi, edx ; 保存edx，因为edx 里保存了系统调用的参数
325 ;（没用栈，而是用了另一个寄存器esi）
326 mov dx, ss
327 mov ds, dx
328 mov es, dx
329 mov fs, dx
330
=> 331 mov edx, esi ; 恢复edx
332
333 mov esi, esp ;esi = 进程表起始地址
334
335 inc dword [k_reenter] ;k_reenter++;
336 cmp dword [k_reenter], 0 ;if(k_reenter ==0)
337 jne .1 ;{
338 mov esp, StackTop ; mov esp, StackTop <--切换到内核栈
339 push restart ; push restart
340 jmp [esi + RETADR - P_STACKBASE] ; return;
341 .1: ;} else { 已经在内核栈，不需要再切换
342 push restart_reenter ; push restart_reenter
343 jmp [esi + RETADR - P_STACKBASE] ; return;
344 ;}
345
346
347 ; =============================================================================
348 ; sys_call
349 ; =============================================================================
350 sys_call:
351 call save
352
353 sti
354 push esi
355
356 push dword [p_proc_ready]
=> 357 push edx
358 push ecx
359 push ebx
360 call [sys_call_table + eax * 4]
=> 361 add esp, 4 * 4
362
363 pop esi
364 mov [esi + EAXREG - P_STACKBASE], eax
365 cli
366
367 ret
sys_sendrec()这个函数被设计得相当简单，它可以描述为：把SEND消息交给msg_send()处理，把RECEIVE消息交给msg_receive()处理。
msg_send() 和msg_receive()这两个函数我们过一会儿细细分解，先来看看之前没出现过的assert()和panic()。这两个函数虽然起的是辅助作 用，但绝对不是可有可无，因为在我们接下来要处理的消息收发中，有一些编程细节还真容易让人迷糊，这时候assert()就大显神威了，它会在错误被放大 之前通知你。panic()的作用也类似，用于通知你发生了严重的错误。
先来看assert()。你或许早就开始使用这个函数，但之前你使用的都是现成的assert，只要包含一个头文件，就可以方便地使用。如今什么都得自力更生了，不过不用怕，写一个assert函数并非难事，见下面的代码：
12 #define ASSERT
13 #ifdef ASSERT
14 void assertion_failure(char *exp, char *file, char *base_file, int line);
15 #define assert(exp) if (exp) ; \
16 else assertion_failure(#exp, __FILE__, __BASE_FILE__, __LINE__)
17 #else
18 #define assert(exp)
19 #endif
注意其中的__FILE__、__BASE_FILE__和__LINE__这三个宏，它们的意义如下[]：
__FILE__: 将被展开成当前输入的文件。在这里，它告诉我们哪个文件中产生了异常。
__BASE_FILE__: 可被认为是传递给编译器的那个文件名。比如你在m.c中包含了n.h，而n.h中的某一个assert函数失败了，则__FILE__为n.h，__BASE_FILE__为m.c。
__LINE__: 将被展开成当前的行号。
明白了这几个宏的意义，剩下的assertion_failure()这个函数就显得容易了，它的作用就是将错误发生的位置打印出来：
42 PUBLIC void assertion_failure(char *exp, char *file, char *base_file, int line)
43 {
44 printl(”%c␣␣assert(%s)␣failed:␣file:␣%s,␣base_file:␣%s,␣ln%d”,
45 MAG_CH_ASSERT,
46 exp, file, base_file, line);
47
48 /**
49 * If assertion fails in a TASK, the system will halt before
50 * printl() returns. If it happens in a USER PROC, printl() will
51 * return like a common routine and arrive here.
52 * @see sys_printx()
53 *
54 * We use a forever loop to prevent the proc from going on:
55 */
56 spin(”assertion_failure()”);
57
58 /* should never arrive here */
59 __asm__ __volatile__(”ud2”);
60 }
注意这里使用了一点点小伎俩，那就是使用了一个改进后的打印函数，叫做printl()，它其实就是一个定义成printf的宏，不过 这里的printf跟上一章中的稍有不同，它将调用一个叫做printx的系统调用，并最终调用函数sys_printx()，它位于tty.c中：
181 PUBLIC int sys_printx(int _unused1, int _unused2, char* s, struct proc* p_proc)
182 {
183 const char * p;
184 char ch;
185
186 char reenter_err[] = ”?␣k_reenter␣is␣incorrect␣for␣unknown␣reason”;
187 reenter_err[0] = MAG_CH_PANIC;
188
189 /**
190 * @note Code in both Ring 0 and Ring 1~3 may invoke printx().
191 * If this happens in Ring 0, no linear-physical address mapping
192 * is needed.
193 *
194 * @attention The value of ‘k_reenter’ is tricky here. When
195 * -# printx() is called in Ring 0
196 * - k_reenter > 0. When code in Ring 0 calls printx(),
197 * an ‘interrupt re-enter’ will occur (printx() generates
198 * a software interrupt). Thus ‘k_reenter’ will be increased
199 * by ‘kernel.asm::save’ and be greater than 0.
200 * -# printx() is called in Ring 1~3
201 * - k_reenter == 0.
202 */
203 if (k_reenter == 0) /* printx() called in Ring<1~3> */
204 p = va2la(proc2pid(p_proc), s);
205 else if (k_reenter > 0) /* printx() called in Ring<0> */
206 p = s;
207 else /* this should NOT happen */
208 p = reenter_err;
209
210 /**
211 * @note if assertion fails in any TASK, the system will be halted;
212 * if it fails in a USER PROC, it’ll return like any normal syscall
213 * does.
214 */
215 if ((*p == MAG_CH_PANIC) ||
216 (*p == MAG_CH_ASSERT && p_proc_ready < &proc_table[NR_TASKS])) {
217 disable_int();
218 char * v = (char*)V_MEM_BASE;
219 const char * q = p + 1; /* +1: skip the magic char */
220
221 while (v < (char*)(V_MEM_BASE + V_MEM_SIZE)) {
222 *v++ = *q++;
223 *v++ = RED_CHAR;
224 if (!*q) {
225 while (((int)v - V_MEM_BASE) % (SCR_WIDTH * 16)) {
226 /* *v++ = ’ ’; */
227 v++;
228 *v++ = GRAY_CHAR;
229 }
230 q = p + 1;
231 }
232 }
233
234 __asm__ __volatile__(”hlt”);
235 }
236
237 while ((ch = *p++) != 0) {
238 if (ch == MAG_CH_PANIC || ch == MAG_CH_ASSERT)
239 continue; /* skip the magic char */
240
241 out_char(tty_table[p_proc->nr_tty].p_console, ch);
242 }
243
244 return 0;
245 }
容易看到，sys_printx()将首先判断首字符是否为预先设定的“Magic Char”，如果是的话，则做响应的特殊处理。我们的assertion_failure()就使用了MAG_CH_ASSERT作为“Magic Char”。当sys_printx()发现传入字符串的第一个字符是MAG_CH_ASSERT时，会同时判断调用系统调用的进程是系统进程 （TASK）还是用户进程（USER PROC），如果是系统进程，则停止整个系统的运转，并将要打印的字符串打印在显存的各处；如果是用户进程，则打印之后像一个普通的printx调用一样 返回，届时该用户进程会因为assertion_failure()中对函数spin()的调用而进入死循环。换言之，系统进程的assert失败会导致 系统停转，用户进程的失败仅仅使自己停转。
到这里读者应该很清楚assert()函数的实现方法了，我们不妨来试验一下，在系统进程TTY中添加一句“assert(0);”，运行，读者将看到如图8.3所示的画面。再在用户进程TestC中添加一句“assert(0);”，将看到如图8.4所示的画面。
panic()跟assert()类似，也用到了sys_printx()和“Magic Char”，不过它要更简单一些，见下面的代码：
159 PUBLIC void panic(const char *fmt, ...)
160 {
161 int i;
162 char buf[256];
163
164 /* 4 is the size of fmt in the stack */
165 va_list arg = (va_list)((char*)&fmt + 4);
166
167 i = vsprintf(buf, fmt, arg);
168
169 printl(”%c␣!!panic!!␣%s”, MAG_CH_PANIC, buf);
170
171 /* should never arrive here */
172 __asm__ __volatile__(”ud2”);
173 }
由于panic只会用在系统任务所处的Ring1或Ring0，所以sys_printx()遇到MAG_CH_PANIC就直接叫停整个系统，因为我们使用panic的时候，必是发生了严重错误的时候。
我们同样可以在TTY中试验一下panic的效果，比如添加这么一行：
panic("in TTY");
运行，会看到如图8.5所示的效果。
在 我们接下来的代码中，很多地方用到了assert()和panic()，其实有些地方完全可以不用这两个函数，而是以返回值的形式向上层函数传递的，但使 用assert()和panic()可以减少代码量，并在第一时间通知我们哪里出了问题，作为一个试验性的操作系统，笔者认为这样做比使用某种方法来“消 除”错误还要好。
话题岔开这么久，让我们回到代码8.3，既然关键的函数是msg_send()和msg_receive()，那我们就来看一下，它们是IPC的核心代码：
145 /*****************************************************************************
146 * ldt_seg_linear
147 *****************************************************************************/
148 /**
149 * <Ring 0~1> Calculate the linear address of a certain segment of a given
150 * proc.
151 *
152 * @param p Whose (the proc ptr).
153 * @param idx Which (one proc has more than one segments).
154 *
155 * @return The required linear address.
156 *****************************************************************************/
157 PUBLIC int ldt_seg_linear(struct proc* p, int idx)
158 {
159 struct descriptor * d = &p->ldts[idx];
160
161 return d->base_high << 24 | d->base_mid << 16 | d->base_low;
162 }
163
164 /*****************************************************************************
165 * va2la
166 *****************************************************************************/
167 /**
168 * <Ring 0~1> Virtual addr --> Linear addr.
169 *
170 * @param pid PID of the proc whose address is to be calculated.
171 * @param va Virtual address.
172 *
173 * @return The linear address for the given virtual address.
174 *****************************************************************************/
175 PUBLIC void* va2la(int pid, void* va)
176 {
177 struct proc* p = &proc_table[pid];
178
179 u32 seg_base = ldt_seg_linear(p, INDEX_LDT_RW);
180 u32 la = seg_base + (u32)va;
181
182 if (pid < NR_TASKS + NR_PROCS) {
183 assert(la == (u32)va);
184 }
185
186 return (void*)la;
187 }
188
189 /*****************************************************************************
190 * reset_msg
191 *****************************************************************************/
192 /**
193 * <Ring 0~3> Clear up a MESSAGE by setting each byte to 0.
194 *
195 * @param p The message to be cleared.
196 *****************************************************************************/
197 PUBLIC void reset_msg(MESSAGE* p)
198 {
199 memset(p, 0, sizeof(MESSAGE));
200 }
201
202 /*****************************************************************************
203 * block
204 *****************************************************************************/
205 /**
206 * <Ring 0> This routine is called after ‘p_flags’ has been set (!= 0), it
207 * calls ‘schedule()’ to choose another proc as the ‘proc_ready’.
208 *
209 * @attention This routine does not change ‘p_flags’. Make sure the ‘p_flags’
210 * of the proc to be blocked has been set properly.
211 *
212 * @param p The proc to be blocked.
213 *****************************************************************************/
214 PRIVATE void block(struct proc* p)
215 {
216 assert(p->p_flags);
217 schedule();
218 }
219
220 /*****************************************************************************
221 * unblock
222 *****************************************************************************/
223 /**
224 * <Ring 0> This is a dummy routine. It does nothing actually. When it is
225 * called, the ‘p_flags’ should have been cleared (== 0).
226 *
227 * @param p The unblocked proc.
228 *****************************************************************************/
229 PRIVATE void unblock(struct proc* p)
230 {
231 assert(p->p_flags == 0);
232 }
233
234 /*****************************************************************************
235 * deadlock
236 *****************************************************************************/
237 /**
238 * <Ring 0> Check whether it is safe to send a message from src to dest.
239 * The routine will detect if the messaging graph contains a cycle. For
240 * instance, if we have procs trying to send messages like this:
241 * A -> B -> C -> A, then a deadlock occurs, because all of them will
242 * wait forever. If no cycles detected, it is considered as safe.
243 *
244 * @param src Who wants to send message.
245 * @param dest To whom the message is sent.
246 *
247 * @return Zero if success.
248 *****************************************************************************/
249 PRIVATE int deadlock(int src, int dest)
250 {
251 struct proc* p = proc_table + dest;
252 while (1) {
253 if (p->p_flags & SENDING) {
254 if (p->p_sendto == src) {
255 /* print the chain */
256 p = proc_table + dest;
257 printl(”=_=%s”, p->name);
258 do {
259 assert(p->p_msg);
260 p = proc_table + p->p_sendto;
261 printl(”->%s”, p->name);
262 } while (p != proc_table + src);
263 printl(”=_=”);
264
265 return 1;
266 }
267 p = proc_table + p->p_sendto;
268 }
269 else {
270 break;
271 }
272 }
273 return 0;
274 }
275
276 /*****************************************************************************
277 * msg_send
278 *****************************************************************************/
279 /**
280 * <Ring 0> Send a message to the dest proc. If dest is blocked waiting for
281 * the message, copy the message to it and unblock dest. Otherwise the caller
282 * will be blocked and appended to the dest’s sending queue.
283 *
284 * @param current The caller, the sender.
285 * @param dest To whom the message is sent.
286 * @param m The message.
287 *
288 * @return Zero if success.
289 *****************************************************************************/
290 PRIVATE int msg_send(struct proc* current, int dest, MESSAGE* m)
291 {
292 struct proc* sender = current;
293 struct proc* p_dest = proc_table + dest; /* proc dest */
294
295 assert(proc2pid(sender) != dest);
296
297 /* check for deadlock here */
298 if (deadlock(proc2pid(sender), dest)) {
299 panic(”>>DEADLOCK<<␣%s->%s”, sender->name, p_dest->name);
300 }
301
302 if ((p_dest->p_flags & RECEIVING) && /* dest is waiting for the msg */
303 (p_dest->p_recvfrom == proc2pid(sender) ||
304 p_dest->p_recvfrom == ANY)) {
305 assert(p_dest->p_msg);
306 assert(m);
307
308 phys_copy(va2la(dest, p_dest->p_msg),
309 va2la(proc2pid(sender), m),
310 sizeof(MESSAGE));
311 p_dest->p_msg = 0;
312 p_dest->p_flags &= ~RECEIVING; /* dest has received the msg */
313 p_dest->p_recvfrom = NO_TASK;
314 unblock(p_dest);
315
316 assert(p_dest->p_flags == 0);
317 assert(p_dest->p_msg == 0);
318 assert(p_dest->p_recvfrom == NO_TASK);
319 assert(p_dest->p_sendto == NO_TASK);
320 assert(sender->p_flags == 0);
321 assert(sender->p_msg == 0);
322 assert(sender->p_recvfrom == NO_TASK);
323 assert(sender->p_sendto == NO_TASK);
324 }
325 else { /* dest is not waiting for the msg */
326 sender->p_flags |= SENDING;
327 assert(sender->p_flags == SENDING);
328 sender->p_sendto = dest;
329 sender->p_msg = m;
330
331 /* append to the sending queue */
332 struct proc * p;
333 if (p_dest->q_sending) {
334 p = p_dest->q_sending;
335 while (p->next_sending)
336 p = p->next_sending;
337 p->next_sending = sender;
338 }
339 else {
340 p_dest->q_sending = sender;
341 }
342 sender->next_sending = 0;
343
344 block(sender);
345
346 assert(sender->p_flags == SENDING);
347 assert(sender->p_msg != 0);
348 assert(sender->p_recvfrom == NO_TASK);
349 assert(sender->p_sendto == dest);
350 }
351
352 return 0;
353 }
354
355
356 /*****************************************************************************
357 * msg_receive
358 *****************************************************************************/
359 /**
360 * <Ring 0> Try to get a message from the src proc. If src is blocked sending
361 * the message, copy the message from it and unblock src. Otherwise the caller
362 * will be blocked.
363 *
364 * @param current The caller, the proc who wanna receive.
365 * @param src From whom the message will be received.
366 * @param m The message ptr to accept the message.
367 *
368 * @return Zero if success.
369 *****************************************************************************/
370 PRIVATE int msg_receive(struct proc* current, int src, MESSAGE* m)
371 {
372 struct proc* p_who_wanna_recv = current; /**
373 * This name is a little bit
374 * wierd, but it makes me
375 * think clearly, so I keep
376 * it.
377 */
378 struct proc* p_from = 0; /* from which the message will be fetched */
379 struct proc* prev = 0;
380 int copyok = 0;
381
382 assert(proc2pid(p_who_wanna_recv) != src);
383
384 if ((p_who_wanna_recv->has_int_msg) &&
385 ((src == ANY) || (src == INTERRUPT))) {
386 /* There is an interrupt needs p_who_wanna_recv’s handling and
387 * p_who_wanna_recv is ready to handle it.
388 */
389
390 MESSAGE msg;
391 reset_msg(&msg);
392 msg.source = INTERRUPT;
393 msg.type = HARD_INT;
394 assert(m);
395 phys_copy(va2la(proc2pid(p_who_wanna_recv), m), &msg,
396 sizeof(MESSAGE));
397
398 p_who_wanna_recv->has_int_msg = 0;
399
400 assert(p_who_wanna_recv->p_flags == 0);
401 assert(p_who_wanna_recv->p_msg == 0);
402 assert(p_who_wanna_recv->p_sendto == NO_TASK);
403 assert(p_who_wanna_recv->has_int_msg == 0);
404
405 return 0;
406 }
407
408
409 /* Arrives here if no interrupt for p_who_wanna_recv. */
410 if (src == ANY) {
411 /* p_who_wanna_recv is ready to receive messages from
412 * ANY proc, we’ll check the sending queue and pick the
413 * first proc in it.
414 */
415 if (p_who_wanna_recv->q_sending) {
416 p_from = p_who_wanna_recv->q_sending;
417 copyok = 1;
418
419 assert(p_who_wanna_recv->p_flags == 0);
420 assert(p_who_wanna_recv->p_msg == 0);
421 assert(p_who_wanna_recv->p_recvfrom == NO_TASK);
422 assert(p_who_wanna_recv->p_sendto == NO_TASK);
423 assert(p_who_wanna_recv->q_sending != 0);
424 assert(p_from->p_flags == SENDING);
425 assert(p_from->p_msg != 0);
426 assert(p_from->p_recvfrom == NO_TASK);
427 assert(p_from->p_sendto == proc2pid(p_who_wanna_recv));
428 }
429 }
430 else {
431 /* p_who_wanna_recv wants to receive a message from
432 * a certain proc: src.
433 */
434 p_from = &proc_table[src];
435
436 if ((p_from->p_flags & SENDING) &&
437 (p_from->p_sendto == proc2pid(p_who_wanna_recv))) {
438 /* Perfect, src is sending a message to
439 * p_who_wanna_recv.
440 */
441 copyok = 1;
442
443 struct proc* p = p_who_wanna_recv->q_sending;
444 assert(p); /* p_from must have been appended to the
445 * queue, so the queue must not be NULL
446 */
447 while (p) {
448 assert(p_from->p_flags & SENDING);
449 if (proc2pid(p) == src) { /* if p is the one */
450 p_from = p;
451 break;
452 }
453 prev = p;
454 p = p->next_sending;
455 }
456
457 assert(p_who_wanna_recv->p_flags == 0);
458 assert(p_who_wanna_recv->p_msg == 0);
459 assert(p_who_wanna_recv->p_recvfrom == NO_TASK);
460 assert(p_who_wanna_recv->p_sendto == NO_TASK);
461 assert(p_who_wanna_recv->q_sending != 0);
462 assert(p_from->p_flags == SENDING);
463 assert(p_from->p_msg != 0);
464 assert(p_from->p_recvfrom == NO_TASK);
465 assert(p_from->p_sendto == proc2pid(p_who_wanna_recv));
466 }
467 }
468
469 if (copyok) {
470 /* It’s determined from which proc the message will
471 * be copied. Note that this proc must have been
472 * waiting for this moment in the queue, so we should
473 * remove it from the queue.
474 */
475 if (p_from == p_who_wanna_recv->q_sending) { /* the 1st one */
476 assert(prev == 0);
477 p_who_wanna_recv->q_sending = p_from->next_sending;
478 p_from->next_sending = 0;
479 }
480 else {
481 assert(prev);
482 prev->next_sending = p_from->next_sending;
483 p_from->next_sending = 0;
484 }
485
486 assert(m);
487 assert(p_from->p_msg);
488 /* copy the message */
489 phys_copy(va2la(proc2pid(p_who_wanna_recv), m),
490 va2la(proc2pid(p_from), p_from->p_msg),
491 sizeof(MESSAGE));
492
493 p_from->p_msg = 0;
494 p_from->p_sendto = NO_TASK;
495 p_from->p_flags &= ~SENDING;
496 unblock(p_from);
497 }
498 else { /* nobody’s sending any msg */
499 /* Set p_flags so that p_who_wanna_recv will not
500 * be scheduled until it is unblocked.
501 */
502 p_who_wanna_recv->p_flags |= RECEIVING;
503
504 p_who_wanna_recv->p_msg = m;
505
506 if (src == ANY)
507 p_who_wanna_recv->p_recvfrom = ANY;
508 else
509 p_who_wanna_recv->p_recvfrom = proc2pid(p_from);
510
511 block(p_who_wanna_recv);
512
513 assert(p_who_wanna_recv->p_flags == RECEIVING);
514 assert(p_who_wanna_recv->p_msg != 0);
515 assert(p_who_wanna_recv->p_recvfrom != NO_TASK);
516 assert(p_who_wanna_recv->p_sendto == NO_TASK);
517 assert(p_who_wanna_recv->has_int_msg == 0);
518 }
519
520 return 0;
521 }
围绕msg_send()和msg_receive()，代码中还列出了其他几个必要的函数，它们是：
ldt_seg_linear(): 每个进程都有自己的LDT，位于进程表的中间，这个函数就是根据LDT中描述符的索引来求得描述符所指向的段的基地址。
va2la(): 用来由虚拟地址求线性地址，它用到了ldt_seg_linear()。
reset_msg(): 用于把一个消息的每个字节清零。
block(): 阻塞一个进程。
unblock(): 解除一个进程的阻塞。
deadlock(): 简单地判断是否发生死锁。方法是判断消息的发送是否构成一个环，如果构成环则意味着发生死锁，比如A试图发消息给B，同时B试图给C，C试图给A发消息，那么死锁就发生了，因为A、B和C三个进程都将无限等待下去（如图8.6所示）。
在block()、unblock()和deadlock()中，都出现了struct proc这个结构体的一个新成员：p_flag。其实增加的新成员还有几个：
31 struct proc {
32 struct stackframe regs; /* process registers saved in stack frame */
33
34 u16 ldt_sel; /* gdt selector giving ldt base and limit */
35 struct descriptor ldts[LDT_SIZE]; /* local descs for code and data */
36
37 int ticks; /* remained ticks */
38 int priority;
39
40 u32 pid; /* process id passed in from MM */
41 char name[16]; /* name of the process */
42
=> 43 int p_flags; /**
44 * process flags.
45 * A proc is runnable iff p_flags==0
46 */
47
=> 48 MESSAGE * p_msg;
=> 49 int p_recvfrom;
=> 50 int p_sendto;
51
=> 52 int has_int_msg; /**
53 * nonzero if an INTERRUPT occurred when
54 * the task is not ready to deal with it.
55 */
56
=> 57 struct proc * q_sending; /**
58 * queue of procs sending messages to
59 * this proc
60 */
=> 61 struct proc * next_sending;/**
62 * next proc in the sending
63 * queue (q_sending)
64 */
65
66 int nr_tty;
67 };
所有增加的这些成员都是跟消息机制有关的。
p_flags: 用于标明进程的状态。目前它的取值可以有三种：
p_msg: 指向消息体的指针。
p_recvfrom: 假设进程P想要接收消息，但目前没有进程发消息给它，本成员记录P想要从谁那里接收消息。
p_sendto: 假设进程P想要发送消息，但目前没有进程接收它，本成员记录P想要发送消息给谁。
has_int_msg: 如果有一个中断需要某进程来处理，或者换句话说，某进程正在等待一个中断发生──比如硬盘驱动可能会等待硬盘中断的发生，系统在得知中断发生后会将此位置为1。
q_sending: 如果有若干进程──比如A、B和C──都向同一个进程P发送消息，而P此时并未准备接收消息，那么A、B和C将会排成一个队列。进程P的q_sending指向第一个试图发送消息的进程。
next_sending: 试图发送消息的A、B和C（依时间顺序）三进程排成的队列的实现方式是这样的：目的进程P的进程表的q_sending指向A，进程A的进程表的 next_sending指向B，进程B的进程表的next_sending指向C，进程C的进程表的next_sending指向空。
假设有进程A想要向B发送消息M，那么过程将会是这样的：
假设有进程B想要接收消息（来自特定进程、中断或者任意进程），那么过程将会是这样的：
B准备一个空的消息结构体M，用于接收消息。
B通过系统调用sendrec，最终调用msg_receive。
判断B是否有个来自硬件的消息（通过has_int_msg），如果是，并且B准备接收来自中断的消息或准备接收任意消息，则马上准备一个消息给B，并返回。
如果B想接收来自任意进程的消息，则从自己的发送队列中选取第一个（如果队列非空的话），将其消息复制给M。
如果B是想接收来自特定进程A的消息，则先判断A是否正在等待向B发送消息，若是的话，将其消息复制给M。
如果此时没有任何进程发消息给B，B会被阻塞。
值 得说明的是，不管是接收方还是发送方，都各自维护一个消息结构体，只不过发送方的结构体是携带了消息内容的而接收方是空的。由于我们使用同步IPC，一方 的需求──发送或接收──只有被满足之后才会继续运行，所以操作系统不需要维护任何的消息缓冲，实现起来也就相对简单。
在 上一节中我们提到，如今的每个进程增加了两种可能的状态：SENDING和RECEIVING。相应的，我们需要在进程调度的时候区别对待了。凡是处于 SENDING或RECEIVING状态的进程，我们就不再让它们获得CPU了，也就是说，将它们“阻塞”了。这也解释了为什么block()和 unblock()两个函数本质上没做任何工作──一个进程是否阻塞，已经由进程表中的p_flags项决定，我们不需要额外做什么工作。不过我们还是应 该保留这两个函数，一方面将来可能要扩展它们，另一方面它们也有助于理清编程的思路。
下面是修改后的调度函数：
31 PUBLIC void schedule()
32 {
33 struct proc* p;
34 int greatest_ticks = 0;
35
36 while (!greatest_ticks) {
37 for (p = &FIRST_PROC; p <= &LAST_PROC; p++) {
=> 38 if (p->p_flags == 0) {
39 if (p->ticks > greatest_ticks) {
40 greatest_ticks = p->ticks;
41 p_proc_ready = p;
42 }
43 }
44 }
45
46 if (!greatest_ticks)
47 for (p = &FIRST_PROC; p <= &LAST_PROC; p++)
=> 48 if (p->p_flags == 0)
49 p->ticks = p->priority;
50 }
51 }
可以看到，当且仅当p_flags为零时，一个进程才可能获得运行的机会。
到这里我们的消息机制已经可以用了，如果读者亲自实践的话，别忘了一些细枝末节的东西，比如在初始化进程时给新增加的进程表成员赋值，再比如增加一些必要的函数声明以及修改Makefile等零碎工作。
为验证消息机制是否工作正常，我们还是从最简单的工作着手，删掉原先的系统调用get_ticks，用收发消息的方法重新实现之。
不过且慢，既然是收发消息，必然是有两方参与。想想便知，我们需要一个系统进程来接收用户进程的消息，并且返回ticks值。我们就来建立一个新的系统进程，就叫它“SYSTASK”。
添加一个任务的工作还是按照第6.4.6节中所述步骤进行。它的主循环如下所示：
22 /*****************************************************************************
23 * task_sys
24 *****************************************************************************/
25 /**
26 * <Ring 1> The main loop of TASK SYS.
27 *
28 *****************************************************************************/
29 PUBLIC void task_sys()
30 {
31 MESSAGE msg;
32 while (1) {
33 send_recv(RECEIVE, ANY, &msg);
34 int src = msg.source;
35
36 switch (msg.type) {
37 case GET_TICKS:
38 msg.RETVAL = ticks;
39 send_recv(SEND, src, &msg);
40 break;
41 default:
42 panic(”unknown␣msg␣type”);
43 break;
44 }
45 }
46 }
代码很简单，不过要留心一下其中用到的函数send_recv()，它其实就是把sendrec这个系统调用给封装了一下：
104 /*****************************************************************************
105 * send_recv
106 *****************************************************************************/
107 /**
108 * <Ring 1~3> IPC syscall.
109 *
110 * It is an encapsulation of ‘sendrec’,
111 * invoking ‘sendrec’ directly should be avoided
112 *
113 * @param function SEND, RECEIVE or BOTH
114 * @param src_dest The caller’s proc_nr
115 * @param msg Pointer to the MESSAGE struct
116 *
117 * @return always 0.
118 *****************************************************************************/
119 PUBLIC int send_recv(int function, int src_dest, MESSAGE* msg)
120 {
121 int ret = 0;
122
123 if (function == RECEIVE)
124 memset(msg, 0, sizeof(MESSAGE));
125
126 switch (function) {
127 case BOTH:
128 ret = sendrec(SEND, src_dest, msg);
129 if (ret == 0)
130 ret = sendrec(RECEIVE, src_dest, msg);
131 break;
132 case SEND:
133 case RECEIVE:
134 ret = sendrec(function, src_dest, msg);
135 break;
136 default:
137 assert((function == BOTH) ||
138 (function == SEND) || (function == RECEIVE));
139 break;
140 }
141
142 return ret;
143 }
我们知道，一个完整的系统调用需要一个来回，那就是用户进程向内核请求一个东西，然后内核返回给它。我们用消息机制来实现这个过程同样 需要一个来回，这意味着用户进程发送一个消息之后需要马上等待接收一个消息，以便收到内核（其实是某个系统任务）给它的返回值。这个发送然后马上接收的行 为被send_recv()这个函数包装了一下，并在SEND和RECEIVE之外又提供了一个叫做BOTH的消息类型。今后我们想要收发消息时，就直接 使用这个send_recv()，而不再直接使用系统调用sendrec。
好了，系统进程SYSTASK已经就绪，下面就来修改一下函数get_ticks：
112 PUBLIC int get_ticks()
113 {
114 MESSAGE msg;
115 reset_msg(&msg);
116 msg.type = GET_TICKS;
117 send_recv(BOTH, TASK_SYS, &msg);
118 return msg.RETVAL;
119 }
120 · · ·
125 void TestA()
126 {
127 while (1) {
128 printf(”<Ticks:%d>”, get_ticks());
129 milli_delay(200);
130 }
131 }
我们以GET_TICKS为消息类型，不夹带其他任何信息地传递给SYSTASK，SYSTASK收到这个消息之后，把当前的ticks值放入消息并发给用户进程，用户进程会接收到它，完成整个任务。
我们来运行一下，结果如图8.7所示。
成功了！进程TestA调用get_ticks之后，成功地打印出了它们的值，这表明我们的消息机制工作良好！
