minix3进程间通信 proc.c

在minix3中，进程间的通信使用的是消息传递，而系统调用也是通过消息传递的方式实现的，有必要仔细研究下消息传递的机制。

那么下面就深入的研究下minix3中消息传递是如何实现的。

消息传递的高层代码位于  kernel/proc.c文件中。

首先当发生系统调用时，从调用者发出进程通信原语，是 send,receive,sendrec,notify 之中的哪一个。然后，通过mpx386.s文件对于调用者寄存器的保存后，将调用源/目的进程号，消息地址，和一个代表了调用flag和function的值作为参数传递给位于proc.c中的sys_call函数。sys_call在接收到mpx386所传递过来的参数后，从代表了调用flag和function的值中分别取出function和flags。然后开始判断该调用的合法性。

首先是

1 2 3 4 5 6 7

if (! (priv(caller_ptr)->s_trap_mask & (1 << function)) ||          (iskerneln(src_dst) && function != SENDREC          && function != RECEIVE)) {      kprintf("sys_call: trap %d not allowed, caller %d, src_dst %d\n",          function, proc_nr(caller_ptr), src_dst);     return(ECALLDENIED);      /* trap denied by mask or kernel */ }

判断发出通信原语的进程是否具有发出该原语的权限(priv(caller_ptr)->s_trap_mask & (1 << function))，另一方面，判断被调用的对象src_dst这个进程号是否是内核进程，如果是，那么其原语是否是 receive或是 sendrec。 因为内核进程只能够 receive和sendrec。其他的通信原语是不能调用内核进程的。  如果这个判断没有成功。则会拒绝对于该系统调用的请求。

随后则是对于src_dst的进程号进行合法性判断，看是否是一个在合法范围内的进程号。虽然通常情况下不会出现非法的情况，但是还有必要判断一下的

1 2 3 4 5

if (! (isokprocn(src_dst) || src_dst == ANY || function == ECHO)) {      kprintf("sys_call: invalid src_dst, src_dst %d, caller %d\n",          src_dst, proc_nr(caller_ptr));     return(EBADSRCDST);       /* invalid process number */ }

在判断的时候列出了2个例外条件，其中有src_dst如果是ANY，这是一个特殊的字段，表示任意进程。如果该判断成功。 则继续下一步的检测。

1 2 3 4 5 6 7 8 9 10 11

if (function & CHECK_PTR) {        vlo = (vir_bytes) m_ptr >> CLICK_SHIFT;          vhi = ((vir_bytes) m_ptr + MESS_SIZE - 1) >> CLICK_SHIFT;     if (vlo < caller_ptr->p_memmap[D].mem_vir || vlo > vhi ||             vhi >= caller_ptr->p_memmap[S].mem_vir +              caller_ptr->p_memmap[S].mem_len) {         kprintf("sys_call: invalid message pointer, trap %d, caller %d\n",           function, proc_nr(caller_ptr));         return(EFAULT);       /* invalid message pointer */     } }

这个判断是要判断一下消息的指针是否是合法的。如果function是notify的话，则不需要判断了，因为通知是不带有消息的。因为消息是在调用者的栈中保存的，栈和数据段位于同一个段中。所以这个检查包含了3部分，分别是对于数据段的检查，对于消息指针的地址和消息结尾的地址的大小检查，以及消息长度是否超过栈长度的检查。如果都通过了。下面则是对于src_dst合法性的检查。

1 2 3 4 5 6 7 8 9 10 11 12 13

if (function & CHECK_DST) {        if (! get_sys_bit(priv(caller_ptr)->s_ipc_to, nr_to_id(src_dst))) {         kprintf("sys_call: ipc mask denied %d sending to %d\n",           proc_nr(caller_ptr), src_dst);         return(ECALLDENIED);      /* call denied by ipc mask */     }                   if (isemptyn(src_dst)) {   if(!shutdown_started)    kprintf("sys_call: dead dst; %d->%d\n", proc_nr(caller_ptr), src_dst);         return(EDEADDST);         /* cannot send to the dead */     } }

如果这个调用使用的是send，sendrec，notify原语。那么就要进行这个检查。这个检查是为了检查dst进程是否运行当前进程对其发送send，sendrec，notify原语。并且dst进程是否还存在。因为要检查能否发送，因此只需要检查特权结构体中的ipc_to字段就可以了。

当这些检查都通过后，那么就进入了sys_call函数的主要部分。即switch结构中调用相应的通信函数来完成通信。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

switch(function) { case SENDREC:     /* A flag is set so that notifications cannot interrupt SENDREC. */     priv(caller_ptr)->s_flags |= SENDREC_BUSY;     /* fall through */ case SEND:                 result = mini_send(caller_ptr, src_dst, m_ptr, flags);     if (function == SEND || result != OK) {            break;                /* done, or SEND failed */     }                     /* fall through for SENDREC */ case RECEIVE:              if (function == RECEIVE)         priv(caller_ptr)->s_flags &= ~SENDREC_BUSY;     result = mini_receive(caller_ptr, src_dst, m_ptr, flags);     break; case NOTIFY:     result = mini_notify(caller_ptr, src_dst);     break; case ECHO:     CopyMess(caller_ptr->p_nr, caller_ptr, m_ptr, caller_ptr, m_ptr);     result = OK;     break; default:     result = EBADCALL;            /* illegal system call */ }

在该部分中，通过判断从 mpx386.s中传递过来的function字段，来找到要调用的高层通信原语。但是这个switch部分也是经过仔细设计的，因为sendrec涉及到了send和receive两个阶段，因此被放到了第一个位置，并在sendrec的case中将发起原语进程的特权标志置位用以表示该进程处于通信状态。随后就陷入到了send的case中。下面我们就一起来看一下  mini_send这个函数。

该函数接收了4个参数，分别是发起原语的进程的进程指针，src_dst，消息地址指针，和flags（通常这个标志被置为0，代表着调用了阻塞通信原语）。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47

PRIVATE int mini_send(caller_ptr, dst, m_ptr, flags) register struct proc *caller_ptr;   /* who is trying to send a message? */ int dst;                /* to whom is message being sent? */ message *m_ptr;             /* pointer to message buffer */ unsigned flags;             /* system call flags */ { /* Send a message from 'caller_ptr' to 'dst'. If 'dst' is blocked waiting  * for this message, copy the message to it and unblock 'dst'. If 'dst' is  * not waiting at all, or is waiting for another source, queue 'caller_ptr'.  */   register struct proc *dst_ptr = proc_addr(dst);   register struct proc **xpp;   register struct proc *xp;            /* Check for deadlock by 'caller_ptr' and 'dst' sending to each other. */   xp = dst_ptr;   while (xp->p_rts_flags & SENDING) {        /* check while sending */     xp = proc_addr(xp->p_sendto);        /* get xp's destination */     if (xp == caller_ptr) return(ELOCKED);  /* deadlock if cyclic */   }            /* Check if 'dst' is blocked waiting for this message. The destination's     * SENDING flag may be set when its SENDREC call blocked while sending.      */   if ( (dst_ptr->p_rts_flags & (RECEIVING | SENDING)) == RECEIVING &&        (dst_ptr->p_getfrom == ANY || dst_ptr->p_getfrom == caller_ptr->p_nr)) {     /* Destination is indeed waiting for this message. */     CopyMess(caller_ptr->p_nr, caller_ptr, m_ptr, dst_ptr,          dst_ptr->p_messbuf);     if ((dst_ptr->p_rts_flags &= ~RECEIVING) == 0) enqueue(dst_ptr);   } else if ( ! (flags & NON_BLOCKING)) {     /* Destination is not waiting.  Block and dequeue caller. */     caller_ptr->p_messbuf = m_ptr;     if (caller_ptr->p_rts_flags == 0) dequeue(caller_ptr);     caller_ptr->p_rts_flags |= SENDING;     caller_ptr->p_sendto = dst;              /* Process is now blocked.  Put in on the destination's queue. */     xpp = &dst_ptr->p_caller_q;      /* find end of list */     while (*xpp != NIL_PROC) xpp = &(*xpp)->p_q_link;         *xpp = caller_ptr;          /* add caller to end */     caller_ptr->p_q_link = NIL_PROC; /* mark new end of list */   } else {     return(ENOTREADY);   }   return(OK); }

mini_send这个函数要考虑的情况比mini_receive要多一些。我们具体看一下mini_send都做了哪些工作。

dst_ptr指向了dst变量所表示的进程的地址。

首先要判断的就是检查目的进程是否也处在发送状态，如果是的话，要遍历其所发送给的进程链，如果该链中有任一个进程正在给当前发起send，或sendrec原语的进程发送消息，那么则会导致死锁。即该进程链中的进程包括该发送进程都在等待另外的进程接收消息，但是没有一个进程能够接收，且它们都被移出就绪队列。无法再次被调度。如果检测到这种状态，则mini_send函数返回一个死锁的错误码。但是并没有将该进程移出调度队列。

在完成了死锁的检测以后，该函数开始判断目的进程是否被阻塞在receive状态上，如果是的话，它等待的是否是当前进程（即发起send或sendrec原语的进程）。当然，如果是ANY进程的话，也是可以的。如果这一步判断成功了，即目的进程的确是在等待当前进程（或是ANY）。那么就将消息从当前进程拷贝到目的进程中。使用的是cp_mess函数，该函数位于 kernel/klib386.s中。当消息拷贝完成后，判断目的进程撤销掉receive状态后，p_rts_flags标志是否是0，如果是0，那么就调用enqueue函数将其插入到就绪队列中，等待下次调度时运行。

如果目的进程没有位于receive状态，那么判断flag字段是否是非阻塞的，如果不是非阻塞的，那么就将消息指针保存在当前进程的进程表中的消息缓冲区指针中。然后如果判断如果该进程状态p_rts_flags标志是0，则执行dequeue函数将其出队列，然后将p_rts_flags标志添加sending状态，并设置其sendto字段为目的进程的进程号。

随后则是对给目的进程发送send或sendrec原语的所有进程进行排队，这里用到了进程表结构中的两个进程指针，分别是 p_caller_q 和 p_q_link。

要说明的是，目的进程中的 p_caller_q 指针是所有试图给目的进程发送send或sendrec原语而被阻塞的进程所形成的链表的链表头地址。而这个被排队的链表随后的节点存在于每一个被阻塞的进程的 p_q_link 字段中。

而目的进程的 p_q_link字段则是， 目的进程企图给其他进程发送send或sendrec原语被阻塞时，排在其他进程的p_caller_q链表中的节点。 这个要区分清楚的。也就是说，同一个进程的进程表中的 p_caller_q 和 p_q_link 字段分别属于两个链表，一个属于发送给它却被阻塞的进程。另一个属于它发送给其他进程却因为其他进程处于不接受消息等状态而被阻塞的链表。

下面再来看一下mini_receive函数

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73

PRIVATE int mini_receive(caller_ptr, src, m_ptr, flags) register struct proc *caller_ptr;   /* process trying to get message */ int src;                /* which message source is wanted */ message *m_ptr;             /* pointer to message buffer */ unsigned flags;             /* system call flags */ { /* A process or task wants to get a message.  If a message is already queued,  * acquire it and deblock the sender.  If no message from the desired source  * is available block the caller, unless the flags don't allow blocking.    */   register struct proc **xpp;   register struct notification **ntf_q_pp;   message m;   int bit_nr;   sys_map_t *map;   bitchunk_t *chunk;   int i, src_id, src_proc_nr;         /* Check to see if a message from desired source is already available.    * The caller's SENDING flag may be set if SENDREC couldn't send. If it is    * set, the process should be blocked.    */   if (!(caller_ptr->p_rts_flags & SENDING)) {           /* Check if there are pending notifications, except for SENDREC. */     if (! (priv(caller_ptr)->s_flags & SENDREC_BUSY)) {               map = &priv(caller_ptr)->s_notify_pending;         for (chunk=&map->chunk[0]; chunk<&map->chunk[NR_SYS_CHUNKS]; chunk++) {                   /* Find a pending notification from the requested source. */             if (! *chunk) continue;             /* no bits in chunk */             for (i=0; ! (*chunk & (1<<i)); ++i) {}    /* look up the bit */             src_id = (chunk - &map->chunk[0]) * BITCHUNK_BITS + i;             if (src_id >= NR_SYS_PROCS) break;       /* out of range */             src_proc_nr = id_to_nr(src_id);     /* get source proc */             if (src!=ANY && src!=src_proc_nr) continue; /* source not ok */             *chunk &= ~(1 << i);          /* no longer pending */                   /* Found a suitable source, deliver the notification message. */         BuildMess(&m, src_proc_nr, caller_ptr); /* assemble message */             CopyMess(src_proc_nr, proc_addr(HARDWARE), &m, caller_ptr, m_ptr);             return(OK);                 /* report success */         }     }           /* Check caller queue. Use pointer pointers to keep code simple. */     xpp = &caller_ptr->p_caller_q;     while (*xpp != NIL_PROC) {         if (src == ANY || src == proc_nr(*xpp)) {         /* Found acceptable message. Copy it and update status. */         CopyMess((*xpp)->p_nr, *xpp, (*xpp)->p_messbuf, caller_ptr, m_ptr);             if (((*xpp)->p_rts_flags &= ~SENDING) == 0) enqueue(*xpp);             *xpp = (*xpp)->p_q_link;     /* remove from queue */             return(OK);             /* report success */     }     xpp = &(*xpp)->p_q_link;     /* proceed to next */     }   }         /* No suitable message is available or the caller couldn't send in SENDREC.     * Block the process trying to receive, unless the flags tell otherwise.    */   if ( ! (flags & NON_BLOCKING)) {       caller_ptr->p_getfrom = src;              caller_ptr->p_messbuf = m_ptr;       if (caller_ptr->p_rts_flags == 0) dequeue(caller_ptr);       caller_ptr->p_rts_flags |= RECEIVING;             return(OK);   } else {       return(ENOTREADY);   } }

该函数首先判断当前进程是否阻塞在了sending状态上，如果是的话，则说明当前进程所要接收的消息来源的进程在之前的发送操作中已经被阻塞了，于是将会跳过下面的部分并执行随后的阻塞部分，即判断flags标志是否是调用的非阻塞函数，如果调用的是阻塞原语，那么就设置当前进程表中的p_getfrom字段为希望得到消息的进程号，并将消息地址保存在进程表的消息缓冲区指针中，然后判断当前进程的状态是否是可运行，如果是，则执行dequeue将其移出就绪调度队列。然后在该进程的p_rts_flags标志中添加receiving状态。并返回。

再来看一下另一种状况，即当前进程已经成功的把消息发送给了其他进程，即它的p_rts_flags标志不是sending。此时进入mini_receive的主要的判断中。

随后判断该进程是不是sendrec状态。如果不是，那么该进程在等待一个通知，函数会进入一个判断中重建通知并将其拷贝到当前进程空间中。如果该进程位于sendrec状态，那么它等待的是另一个进程对它回复的消息。则会对该进程的 p_caller_q字段进行遍历，如果该进程等待的是ANY，那么将找到的第一个不是NULL的进程取出等待链表，然后从等待的进程中将消息拷贝到本进程中（这一过程通过cp_mess完成）。然后执行enqueue将从等待链表中取出的进程插入就绪队列。然后返回。如果等待的进程不是ANY，那么会根据等待的进程号来遍历等待链表队列。直到找到该进程。如果没有找到，则执行下一个判断，在接收时阻塞该进程，并将其移出就绪队列。选择一个新的进程继续执行。然后返回。

至此，进程间通信所使用的几个重要的函数我们已经学习过了，还有一个notify暂时我还没有学习 。。。 先不管了吧 -。-