圖解 - Linux進(jìn)程通信 - 管道實(shí)現(xiàn)
處于安全的考慮，不同進(jìn)程之間的內(nèi)存空間是相互隔離的，也就是說 進(jìn)程A 是不能訪問 進(jìn)程B 的內(nèi)存空間，反之亦然。如果不同進(jìn)程間能夠相互訪問和修改對(duì)方的內(nèi)存，那么當(dāng)前進(jìn)程的內(nèi)存就有可能被其他進(jìn)程非法修改，從而導(dǎo)致安全隱患。
不同的進(jìn)程就像是大海上孤立的島嶼，它們之間不能直接相互通信，如下圖所示：
但某些場(chǎng)景下，不同進(jìn)程間需要相互通信，比如：進(jìn)程A 負(fù)責(zé)處理用戶的請(qǐng)求，而 進(jìn)程B 負(fù)責(zé)保存處理后的數(shù)據(jù)。那么當(dāng) 進(jìn)程A 處理完請(qǐng)求后，就需要把處理后的數(shù)據(jù)提交給 進(jìn)程B 進(jìn)行存儲(chǔ)。此時(shí)，進(jìn)程A 就需要與 進(jìn)程B 進(jìn)行通信。如下圖所示：
由于不同進(jìn)程間是相互隔離的，所以必須借助內(nèi)核來作為橋梁來進(jìn)行相互通信，內(nèi)核相當(dāng)于島嶼之間的輪船，如下圖所示：
內(nèi)核提供多種進(jìn)程間通信的方式，如：共享內(nèi)存，信號(hào)，消息隊(duì)列 和 管道（pipe） 等。本文主要介紹 管道 的原理與實(shí)現(xiàn)。
一、管道的使用
管道 一般用于父子進(jìn)程之間相互通信，一般的用法如下：
- 父進(jìn)程使用 pipe 系統(tǒng)調(diào)用創(chuàng)建一個(gè)管道。
- 然后父進(jìn)程使用 fork 系統(tǒng)調(diào)用創(chuàng)建一個(gè)子進(jìn)程。
- 由于子進(jìn)程會(huì)繼承父進(jìn)程打開的文件句柄，所以父子進(jìn)程可以通過新創(chuàng)建的管道進(jìn)行通信。
其原理如下圖所示：
由于管道分為讀端和寫端，所以需要兩個(gè)文件描述符來管理管道：fd[0] 為讀端，fd[1] 為寫端。
下面代碼介紹了怎么使用 pipe 系統(tǒng)調(diào)用來創(chuàng)建一個(gè)管道：
#include <stdio.h> #include <unistd.h> #include <sys/types.h> #include <stdlib.h> #include <string.h> int main() { int ret = -1; int fd[2]; // 用于管理管道的文件描述符 pid_t pid; char buf[512] = {0}; char *msg = "hello world"; // 創(chuàng)建一個(gè)管理 ret = pipe(fd); if (-1 == ret) { printf("failed to create pipe\n"); return-1; } pid = fork(); // 創(chuàng)建子進(jìn)程 if (0 == pid) { // 子進(jìn)程 close(fd[0]); // 關(guān)閉管道的讀端 ret = write(fd[1], msg, strlen(msg)); // 向管道寫端寫入數(shù)據(jù) exit(0); } else { // 父進(jìn)程 close(fd[1]); // 關(guān)閉管道的寫端 ret = read(fd[0], buf, sizeof(buf)); // 從管道的讀端讀取數(shù)據(jù) printf("parent read %d bytes data: %s\n", ret, buf); } return0; }
編譯代碼：
[root@localhost pipe]# gcc -g pipe.c -o pipe
運(yùn)行代碼，輸出結(jié)果如下：
[root@localhost pipe]# ./pipe parent read 11 bytes data: hello world
二、管道的實(shí)現(xiàn)
每個(gè)進(jìn)程的用戶空間都是獨(dú)立的，但內(nèi)核空間卻是共用的。所以，進(jìn)程間通信必須由內(nèi)核提供服務(wù)。前面介紹了 管道(pipe) 的使用，接下來將會(huì)介紹管道在內(nèi)核中的實(shí)現(xiàn)方式。
本文使用 Linux-2.6.23 內(nèi)核作為分析對(duì)象。
1. 環(huán)形緩沖區(qū)（Ring Buffer）
在內(nèi)核中，管道 使用了環(huán)形緩沖區(qū)來存儲(chǔ)數(shù)據(jù)。環(huán)形緩沖區(qū)的原理是：把一個(gè)緩沖區(qū)當(dāng)成是首尾相連的環(huán)，其中通過讀指針和寫指針來記錄讀操作和寫操作位置。如下圖所示：
在 Linux 內(nèi)核中，使用了 16 個(gè)內(nèi)存頁(yè)作為環(huán)形緩沖區(qū)，所以這個(gè)環(huán)形緩沖區(qū)的大小為 64KB（16 * 4KB）。
當(dāng)向管道寫數(shù)據(jù)時(shí)，從寫指針指向的位置開始寫入，并且將寫指針向前移動(dòng)。而從管道讀取數(shù)據(jù)時(shí)，從讀指針開始讀入，并且將讀指針向前移動(dòng)。當(dāng)對(duì)沒有數(shù)據(jù)可讀的管道進(jìn)行讀操作，將會(huì)阻塞當(dāng)前進(jìn)程。而對(duì)沒有空閑空間的管道進(jìn)行寫操作，也會(huì)阻塞當(dāng)前進(jìn)程。
注意：可以將管道文件描述符設(shè)置為非阻塞，這樣對(duì)管道進(jìn)行讀寫操作時(shí)，就不會(huì)阻塞當(dāng)前進(jìn)程。
2. 管道對(duì)象
在 Linux 內(nèi)核中，管道使用 pipe_inode_info 對(duì)象來進(jìn)行管理。我們先來看看 pipe_inode_info 對(duì)象的定義，如下所示：
struct pipe_inode_info { wait_queue_head_t wait; unsignedint nrbufs, unsignedint curbuf; ... unsignedint readers; unsignedint writers; unsignedint waiting_writers; ... struct inode *inode; struct pipe_buffer bufs[16]; };
下面介紹一下 pipe_inode_info 對(duì)象各個(gè)字段的作用：
- wait：等待隊(duì)列，用于存儲(chǔ)正在等待管道可讀或者可寫的進(jìn)程。
- bufs：環(huán)形緩沖區(qū)，由 16 個(gè) pipe_buffer 對(duì)象組成，每個(gè) pipe_buffer 對(duì)象擁有一個(gè)內(nèi)存頁(yè) ，后面會(huì)介紹。
- nrbufs：表示未讀數(shù)據(jù)已經(jīng)占用了環(huán)形緩沖區(qū)的多少個(gè)內(nèi)存頁(yè)。
- curbuf：表示當(dāng)前正在讀取環(huán)形緩沖區(qū)的哪個(gè)內(nèi)存頁(yè)中的數(shù)據(jù)。
- readers：表示正在讀取管道的進(jìn)程數(shù)。
- writers：表示正在寫入管道的進(jìn)程數(shù)。
- waiting_writers：表示等待管道可寫的進(jìn)程數(shù)。
- inode：與管道關(guān)聯(lián)的 inode 對(duì)象。
由于環(huán)形緩沖區(qū)是由 16 個(gè) pipe_buffer 對(duì)象組成，所以下面我們來看看 pipe_buffer 對(duì)象的定義：
struct pipe_buffer { struct page *page; unsignedint offset; unsignedint len; ... };
下面介紹一下 pipe_buffer 對(duì)象各個(gè)字段的作用：
- page：指向 pipe_buffer 對(duì)象占用的內(nèi)存頁(yè)。
- offset：如果進(jìn)程正在讀取當(dāng)前內(nèi)存頁(yè)的數(shù)據(jù)，那么 offset 指向正在讀取當(dāng)前內(nèi)存頁(yè)的偏移量。
- len：表示當(dāng)前內(nèi)存頁(yè)擁有未讀數(shù)據(jù)的長(zhǎng)度。
下圖展示了 pipe_inode_info 對(duì)象與 pipe_buffer 對(duì)象的關(guān)系：
管道的環(huán)形緩沖區(qū)實(shí)現(xiàn)方式與經(jīng)典的環(huán)形緩沖區(qū)實(shí)現(xiàn)方式有點(diǎn)區(qū)別，經(jīng)典的環(huán)形緩沖區(qū)一般先申請(qǐng)一塊地址連續(xù)的內(nèi)存塊，然后通過讀指針與寫指針來對(duì)讀操作與寫操作進(jìn)行定位。
但為了減少對(duì)內(nèi)存的使用，內(nèi)核不會(huì)在創(chuàng)建管道時(shí)就申請(qǐng) 64K 的內(nèi)存塊，而是在進(jìn)程向管道寫入數(shù)據(jù)時(shí)，按需來申請(qǐng)內(nèi)存。
那么當(dāng)進(jìn)程從管道讀取數(shù)據(jù)時(shí)，內(nèi)核怎么處理呢？下面我們來看看管道讀操作的實(shí)現(xiàn)方式。
3. 讀操作
從 經(jīng)典的環(huán)形緩沖區(qū) 中讀取數(shù)據(jù)時(shí)，首先通過讀指針來定位到讀取數(shù)據(jù)的起始地址，然后判斷環(huán)形緩沖區(qū)中是否有數(shù)據(jù)可讀，如果有就從環(huán)形緩沖區(qū)中讀取數(shù)據(jù)到用戶空間的緩沖區(qū)中。如下圖所示：
而 管道的環(huán)形緩沖區(qū) 與 經(jīng)典的環(huán)形緩沖區(qū) 實(shí)現(xiàn)稍有不同，管道的環(huán)形緩沖區(qū) 其讀指針是由 pipe_inode_info 對(duì)象的 curbuf 字段與 pipe_buffer 對(duì)象的 offset 字段組合而成：
- pipe_inode_info 對(duì)象的 curbuf 字段表示讀操作要從 bufs 數(shù)組的哪個(gè) pipe_buffer 中讀取數(shù)據(jù)。
- pipe_buffer 對(duì)象的 offset 字段表示讀操作要從內(nèi)存頁(yè)的哪個(gè)位置開始讀取數(shù)據(jù)。
讀取數(shù)據(jù)的過程如下圖所示：
從緩沖區(qū)中讀取到 n 個(gè)字節(jié)的數(shù)據(jù)后，會(huì)相應(yīng)移動(dòng)讀指針 n 個(gè)字節(jié)的位置（也就是增加 pipe_buffer 對(duì)象的 offset 字段），并且減少 n 個(gè)字節(jié)的可讀數(shù)據(jù)長(zhǎng)度（也就是減少 pipe_buffer 對(duì)象的 len 字段）。
當(dāng) pipe_buffer 對(duì)象的 len 字段變?yōu)?0 時(shí)，表示當(dāng)前 pipe_buffer 沒有可讀數(shù)據(jù)，那么將會(huì)對(duì) pipe_inode_info 對(duì)象的 curbuf 字段移動(dòng)一個(gè)位置，并且其 nrbufs 字段進(jìn)行減一操作。
我們來看看管道讀操作的代碼實(shí)現(xiàn)，讀操作由 pipe_read 函數(shù)完成。為了突出重點(diǎn)，我們只列出關(guān)鍵代碼，如下所示：
static ssize_t pipe_read(struct kiocb *iocb, const struct iovec *_iov, unsigned long nr_segs, loff_t pos) { ... struct pipe_inode_info *pipe; // 1. 獲取管道對(duì)象 pipe = inode->i_pipe; for (;;) { // 2. 獲取管道未讀數(shù)據(jù)占有多少個(gè)內(nèi)存頁(yè) int bufs = pipe->nrbufs; if (bufs) { // 3. 獲取讀操作應(yīng)該從環(huán)形緩沖區(qū)的哪個(gè)內(nèi)存頁(yè)處讀取數(shù)據(jù) int curbuf = pipe->curbuf; struct pipe_buffer *buf = pipe->bufs + curbuf; ... /* 4. 通過 pipe_buffer 的 offset 字段獲取真正的讀指針, * 并且從管道中讀取數(shù)據(jù)到用戶緩沖區(qū). */ error = pipe_iov_copy_to_user(iov, addr + buf->offset, chars, atomic); ... ret += chars; buf->offset += chars; // 增加 pipe_buffer 對(duì)象的 offset 字段的值 buf->len -= chars; // 減少 pipe_buffer 對(duì)象的 len 字段的值 /* 5. 如果當(dāng)前內(nèi)存頁(yè)的數(shù)據(jù)已經(jīng)被讀取完畢 */ if (!buf->len) { ... curbuf = (curbuf + 1) & (PIPE_BUFFERS - 1); pipe->curbuf = curbuf; // 移動(dòng) pipe_inode_info 對(duì)象的 curbuf 指針 pipe->nrbufs = --bufs; // 減少 pipe_inode_info 對(duì)象的 nrbufs 字段 do_wakeup = 1; } total_len -= chars; // 6. 如果讀取到用戶期望的數(shù)據(jù)長(zhǎng)度, 退出循環(huán) if (!total_len) break; } ... } ... return ret; }
上面代碼總結(jié)來說分為以下步驟：
- 通過文件 inode 對(duì)象來獲取到管道的 pipe_inode_info 對(duì)象。
- 通過 pipe_inode_info 對(duì)象的 nrbufs 字段獲取管道未讀數(shù)據(jù)占有多少個(gè)內(nèi)存頁(yè)。
- 通過 pipe_inode_info 對(duì)象的 curbuf 字段獲取讀操作應(yīng)該從環(huán)形緩沖區(qū)的哪個(gè)內(nèi)存頁(yè)處讀取數(shù)據(jù)。
- 通過 pipe_buffer 對(duì)象的 offset 字段獲取真正的讀指針， 并且從管道中讀取數(shù)據(jù)到用戶緩沖區(qū)。
- 如果當(dāng)前內(nèi)存頁(yè)的數(shù)據(jù)已經(jīng)被讀取完畢，那么移動(dòng) pipe_inode_info 對(duì)象的 curbuf 指針，并且減少其 nrbufs 字段的值。
- 如果讀取到用戶期望的數(shù)據(jù)長(zhǎng)度，退出循環(huán)。
4. 寫操作
分析完管道讀操作的實(shí)現(xiàn)后，接下來，我們分析一下管道寫操作的實(shí)現(xiàn)。
經(jīng)典的環(huán)形緩沖區(qū) 寫入數(shù)據(jù)時(shí)，首先通過寫指針進(jìn)行定位要寫入的內(nèi)存地址，然后判斷環(huán)形緩沖區(qū)的空間是否足夠，足夠就把數(shù)據(jù)寫入到環(huán)形緩沖區(qū)中。如下圖所示：
但 管道的環(huán)形緩沖區(qū) 并沒有保存 寫指針，而是通過 讀指針 計(jì)算出來。那么怎么通過讀指針計(jì)算出寫指針呢？
其實(shí)很簡(jiǎn)單，就是：
寫指針 = 讀指針 + 未讀數(shù)據(jù)長(zhǎng)度
下面我們來看看，向管道寫入 200 字節(jié)數(shù)據(jù)的過程示意圖，如下所示：
如上圖所示，向管道寫入數(shù)據(jù)時(shí)：
-
首先通過 pipe_inode_info 的 curbuf 字段和 nrbufs 字段來定位到，應(yīng)該向哪個(gè) pipe_buffer 寫入數(shù)據(jù)。
-
然后再通過 pipe_buffer 對(duì)象的 offset 字段和 len 字段來定位到，應(yīng)該寫入到內(nèi)存頁(yè)的哪個(gè)位置。
下面我們通過源碼來分析，寫操作是怎么實(shí)現(xiàn)的，代碼如下（為了特出重點(diǎn)，代碼有所刪減）：
static ssize_t pipe_write(struct kiocb *iocb, const struct iovec *_iov, unsigned long nr_segs, loff_t ppos) { ... struct pipe_inode_info *pipe; ... pipe = inode->i_pipe; ... chars = total_len & (PAGE_SIZE - 1); /* size of the last buffer */ // 1. 如果最后寫入的 pipe_buffer 還有空閑的空間 if (pipe->nrbufs && chars != 0) { // 獲取寫入數(shù)據(jù)的位置 int lastbuf = (pipe->curbuf + pipe->nrbufs - 1) & (PIPE_BUFFERS-1); struct pipe_buffer *buf = pipe->bufs + lastbuf; conststruct pipe_buf_operations *ops = buf->ops; int offset = buf->offset + buf->len; if (ops->can_merge && offset + chars <= PAGE_SIZE) { ... error = pipe_iov_copy_from_user(offset + addr, iov, chars, atomic); ... buf->len += chars; total_len -= chars; ret = chars; // 如果要寫入的數(shù)據(jù)已經(jīng)全部寫入成功, 退出循環(huán) if (!total_len) goto out; } } // 2. 如果最后寫入的 pipe_buffer 空閑空間不足, 那么申請(qǐng)一個(gè)新的內(nèi)存頁(yè)來存儲(chǔ)數(shù)據(jù) for (;;) { int bufs; ... bufs = pipe->nrbufs; if (bufs < PIPE_BUFFERS) { int newbuf = (pipe->curbuf + bufs) & (PIPE_BUFFERS-1); struct pipe_buffer *buf = pipe->bufs + newbuf; ... // 申請(qǐng)一個(gè)新的內(nèi)存頁(yè) if (!page) { page = alloc_page(GFP_HIGHUSER); ... } ... error = pipe_iov_copy_from_user(src, iov, chars, atomic); ... ret += chars; buf->page = page; buf->ops = &anon_pipe_buf_ops; buf->offset = 0; buf->len = chars; pipe->nrbufs = ++bufs; pipe->tmp_page = NULL; // 如果要寫入的數(shù)據(jù)已經(jīng)全部寫入成功, 退出循環(huán) total_len -= chars; if (!total_len) break; } ... } out: ... return ret; }
上面代碼有點(diǎn)長(zhǎng)，但是邏輯卻很簡(jiǎn)單，主要進(jìn)行如下操作：
- 如果上次寫操作寫入的 pipe_buffer 還有空閑的空間，那么就將數(shù)據(jù)寫入到此 pipe_buffer 中，并且增加其 len 字段的值。
- 如果上次寫操作寫入的 pipe_buffer 沒有足夠的空閑空間，那么就新申請(qǐng)一個(gè)內(nèi)存頁(yè)，并且把數(shù)據(jù)保存到新的內(nèi)存頁(yè)中，并且增加 pipe_inode_info 的 nrbufs 字段的值。
- 如果寫入的數(shù)據(jù)已經(jīng)全部寫入成功，那么就退出寫操作。
三、思考一下
管道讀寫操作的實(shí)現(xiàn)已經(jīng)分析完畢，現(xiàn)在我們來思考一下以下問題。
1. 為什么父子進(jìn)程可以通過管道來通信？
這是因?yàn)楦缸舆M(jìn)程通過 pipe 系統(tǒng)調(diào)用打開的管道，在內(nèi)核空間中指向同一個(gè)管道對(duì)象（pipe_inode_info）。所以父子進(jìn)程共享著同一個(gè)管道對(duì)象，那么就可以通過這個(gè)共享的管道對(duì)象進(jìn)行通信。
2. 為什么內(nèi)核要使用 16 個(gè)內(nèi)存頁(yè)進(jìn)行數(shù)據(jù)存儲(chǔ)？
這是為了減少內(nèi)存使用。
因?yàn)槭褂?nbsp;pipe 系統(tǒng)調(diào)用打開管道時(shí)，并沒有立刻申請(qǐng)內(nèi)存頁(yè)，而是當(dāng)有進(jìn)程向管道寫入數(shù)據(jù)時(shí)，才會(huì)按需申請(qǐng)內(nèi)存頁(yè)。當(dāng)內(nèi)存頁(yè)的數(shù)據(jù)被讀取完后，內(nèi)核會(huì)將此內(nèi)存頁(yè)回收，來減少管道對(duì)內(nèi)存的使用。