這篇文章描述了管道在 Unix 內核中的實現。 最近一篇題為“
我們在說啥啊?
管道“可能是 Unix 中最重要的發明”——這是 Unix 將小程序組合在一起的基本哲學的定義特徵,也是熟悉的命令行口號:
$ echo hello | wc -c
6
此功能取決於內核提供的系統調用 pipe
,在文檔頁面上進行了描述
管道為進程間通信提供了一種單向通道。 管道有一個輸入端(寫端)和一個輸出端(讀端)。 寫入管道輸入的數據可以在輸出處讀取。
管道是通過調用創建的
pipe(2)
,它返回兩個文件描述符:一個指向管道的輸入,第二個指向輸出。
上述命令的跟踪輸出顯示了管道的創建以及從一個進程到另一個進程的數據流:
$ strace -qf -e execve,pipe,dup2,read,write
sh -c 'echo hello | wc -c'
execve("/bin/sh", ["sh", "-c", "echo hello | wc -c"], …)
pipe([3, 4]) = 0
[pid 2604795] dup2(4, 1) = 1
[pid 2604795] write(1, "hellon", 6) = 6
[pid 2604796] dup2(3, 0) = 0
[pid 2604796] execve("/usr/bin/wc", ["wc", "-c"], …)
[pid 2604796] read(0, "hellon", 16384) = 6
[pid 2604796] write(1, "6n", 2) = 2
父進程調用 pipe()
獲取附加的文件描述符。 一個子進程寫入一個描述符,另一個進程從另一個描述符讀取相同的數據。 shell 使用 dup2“重命名”描述符 3 和 4 以匹配 stdin 和 stdout。
如果沒有管道,shell 將不得不將一個進程的輸出寫入文件並將其通過管道傳輸到另一個進程以從文件中讀取數據。 結果,我們會浪費更多的資源和磁盤空間。 然而,管道不僅僅適用於避免臨時文件:
如果一個進程試圖從一個空的管道中讀取,那麼
read(2)
將阻塞直到數據可用。 如果一個進程試圖寫入一個完整的管道,那麼write(2)
將阻塞,直到從管道中讀取了足夠的數據以完成寫入。
與 POSIX 要求一樣,這是一個重要的屬性:寫入管道最多 PIPE_BUF
字節(至少 512)必須是原子的,以便進程可以通過管道以普通文件(不提供此類保證)無法實現的方式相互通信。
使用常規文件,進程可以將其所有輸出寫入其中並將其傳遞給另一個進程。 或者進程可以在硬並行模式下運行,使用外部信號機制(如信號量)相互通知寫入或讀取的完成。 輸送機使我們免於所有這些麻煩。
我們在找什麼?
我將親手解釋,讓您更容易想像傳送帶的工作原理。 您將需要在內存中分配一個緩衝區和一些狀態。 您將需要函數來向緩衝區添加和刪除數據。 在對文件描述符進行讀寫操作期間,您將需要一些工具來調用函數。 並且需要鎖來實現上述特殊行為。
我們現在準備在明亮的燈光下審視內核的源代碼,以確認或反駁我們模糊的心智模型。 但始終要為意外做好準備。
我們在看哪裡?
我不知道我那本名著在哪兒。
漫步在 TUHS 檔案中就像參觀博物館一樣。 我們可以看看我們共同的歷史,我很尊重多年來從舊磁帶和打印輸出中一點一點地恢復所有這些材料的努力。 我敏銳地意識到那些仍然缺失的碎片。
滿足了我們對管道古代歷史的好奇之後,我們可以看看現代的核心來進行比較。
順便說一句, pipe
是表中的系統調用號42 sysent[]
. 巧合?
傳統的 Unix 內核 (1970–1974)
我沒有找到任何踪跡 pipe(2)
既不在
TUHS 聲稱
第三版 Unix 是最後一個使用彙編語言編寫內核的版本,也是第一個使用管道的版本。 1973 年,第三版的改進工作正在進行,內核用 C 重寫,第四版 Unix 誕生了。
一位讀者找到了一份文件的掃描件,道格·麥克羅伊在其中提出了“像花園軟管一樣連接程序”的想法。
在 Brian Kernighan 的書中
當 Unix 出現時,我對協程的熱情促使我向 OS 作者 Ken Thompson 提出要求,讓寫入某個進程的數據不僅可以到達設備,還可以到達另一個進程的出口。 肯認為這是可能的。 然而,作為一個極簡主義者,他希望每個系統功能都能發揮重要作用。 進程之間直接寫入真的比寫入中間文件有很大優勢嗎? 並且只有當我提出了一個具體的建議,並給出了一個朗朗上口的名稱“管道”,並描述了流程交互的語法時,Ken 才終於驚呼:“我會做的!”。
並且做到了。 一個決定性的夜晚,Ken 更改了內核和 shell,修復了幾個標準程序以標準化它們接受輸入(可能來自管道)的方式,並更改了文件名。 第二天,管道在應用中得到了非常廣泛的應用。 到週末,秘書們用它們將文件從文字處理器發送到打印機。 稍後,Ken 將用於包裝管道使用的原始 API 和語法替換為從那時起一直使用的更清晰的約定。
不幸的是,第三版 Unix 內核的源代碼已經丟失。 雖然我們有用 C 編寫的內核源代碼
我們有文檔文本 pipe(2)
來自兩個版本,因此您可以從搜索文檔開始 pipe(2)
用彙編語言編寫,只返回一個文件描述符,但已經提供了預期的核心功能:
系統調用 管 創建稱為管道的 I/O 機制。 返回的文件描述符可用於讀寫操作。 當有東西被寫入管道時,它最多緩衝 504 字節的數據,之後寫入過程被暫停。 從管道讀取時,會獲取緩衝的數據。
到次年,內核已用 C 重寫,並且 pipe(fildes)
»:
系統調用 管 創建稱為管道的 I/O 機制。 返回的文件描述符可用於讀寫操作。 當向管道寫入內容時,將使用 r1 (resp.fildes[1]) 中返回的描述符,緩衝最多 4096 字節的數據,之後暫停寫入過程。 從管道讀取時,返回到 r0 (resp.fildes[0]) 的描述符獲取數據。
假設一旦定義了管道,兩個(或更多)交互進程(由後續調用創建 叉) 將使用調用從管道傳遞數據 閱讀 и 寫.
shell 具有用於定義通過管道連接的線性進程數組的語法。
調用從只有一端(所有寫入文件描述符關閉)的空管道(不包含緩衝數據)讀取返回“文件結尾”。 忽略類似情況下的寫入調用。
最早的
Unix 第六版 (1975)
開始閱讀 Unix 源代碼
多年來書 獅子 是貝爾實驗室以外唯一可用的關於 Unix 內核的文檔。 雖然第六版許可允許教師使用其源代碼,但第七版許可排除了這種可能性,因此該書以非法打印副本分發。
今天,您可以購買這本書的重印本,封面描繪的是複印機旁的學生。 感謝 Warren Toomey(TUHS 項目的發起人),您可以下載
15 多年前,我輸入了一份提供的源代碼副本 獅子因為我不喜歡未知數量的其他副本中我的副本的質量。 TUHS 還不存在,我也無法訪問舊資源。 但在 1988 年,我發現了一盤有 9 條軌道的舊磁帶,它有一台 PDP11 計算機的備份。 很難知道它是否有效,但有一個完好無損的 /usr/src/ 樹,其中大部分文件都標記為 1979,即使在那時看起來也很古老。 我想那是第七版,或者是 PWB 的衍生版本。
我以查找為基礎,手動編輯源代碼到第六版的狀態。 部分代碼保持不變,部分代碼需要稍微編輯,將現代標記 += 更改為過時的 =+。 有些東西被簡單地刪除了,有些東西必須完全重寫,但不要太多。
今天我們可以在 TUHS 在線閱讀第六版的源代碼
順便說一句,乍一看,Kernighan 和 Ritchie 時期之前的 C 代碼的主要特徵是它的 簡潔. 我很少能夠在不進行大量編輯以適應網站上相對狹窄的顯示區域的情況下插入代碼片段。
早
/*
* Max allowable buffering per pipe.
* This is also the max size of the
* file created to implement the pipe.
* If this size is bigger than 4096,
* pipes will be implemented in LARG
* files, which is probably not good.
*/
#define PIPSIZ 4096
自第四版以來緩衝區大小沒有改變。 但是在這裡我們看到,在沒有任何公開文檔的情況下,管道曾經使用文件作為後備存儲!
至於 LARG 文件,它們對應於
這是真正的系統調用 pipe
:
/*
* The sys-pipe entry.
* Allocate an inode on the root device.
* Allocate 2 file structures.
* Put it all together with flags.
*/
pipe()
{
register *ip, *rf, *wf;
int r;
ip = ialloc(rootdev);
if(ip == NULL)
return;
rf = falloc();
if(rf == NULL) {
iput(ip);
return;
}
r = u.u_ar0[R0];
wf = falloc();
if(wf == NULL) {
rf->f_count = 0;
u.u_ofile[r] = NULL;
iput(ip);
return;
}
u.u_ar0[R1] = u.u_ar0[R0]; /* wf's fd */
u.u_ar0[R0] = r; /* rf's fd */
wf->f_flag = FWRITE|FPIPE;
wf->f_inode = ip;
rf->f_flag = FREAD|FPIPE;
rf->f_inode = ip;
ip->i_count = 2;
ip->i_flag = IACC|IUPD;
ip->i_mode = IALLOC;
}
評論清楚地描述了這裡發生的事情。 但是理解代碼並不那麼容易,部分原因是“R0
и R1
傳遞系統調用參數和返回值。
讓我們試試
pipe()
由於通過 R0
и R1
返回用於讀取和寫入的文件描述符編號。 falloc()
返回指向文件結構的指針,但也通過“返回” u.u_ar0[R0]
和一個文件描述符。 也就是說,代碼存儲在 r
用於讀取的文件描述符,並分配一個用於直接寫入的描述符 u.u_ar0[R0]
第二次通話後 falloc()
.
旗 FPIPE
,我們在創建管道時設置,控制函數的行為
/*
* common code for read and write calls:
* check permissions, set base, count, and offset,
* and switch out to readi, writei, or pipe code.
*/
rdwr(mode)
{
register *fp, m;
m = mode;
fp = getf(u.u_ar0[R0]);
/* … */
if(fp->f_flag&FPIPE) {
if(m==FREAD)
readp(fp); else
writep(fp);
}
/* … */
}
然後是函數 readp()
в pipe.c
從管道中讀取數據。 但最好從以下位置開始跟踪實施 writep()
. 同樣,由於參數傳遞約定的性質,代碼變得更加複雜,但可以省略一些細節。
writep(fp)
{
register *rp, *ip, c;
rp = fp;
ip = rp->f_inode;
c = u.u_count;
loop:
/* If all done, return. */
plock(ip);
if(c == 0) {
prele(ip);
u.u_count = 0;
return;
}
/*
* If there are not both read and write sides of the
* pipe active, return error and signal too.
*/
if(ip->i_count < 2) {
prele(ip);
u.u_error = EPIPE;
psignal(u.u_procp, SIGPIPE);
return;
}
/*
* If the pipe is full, wait for reads to deplete
* and truncate it.
*/
if(ip->i_size1 == PIPSIZ) {
ip->i_mode =| IWRITE;
prele(ip);
sleep(ip+1, PPIPE);
goto loop;
}
/* Write what is possible and loop back. */
u.u_offset[0] = 0;
u.u_offset[1] = ip->i_size1;
u.u_count = min(c, PIPSIZ-u.u_offset[1]);
c =- u.u_count;
writei(ip);
prele(ip);
if(ip->i_mode&IREAD) {
ip->i_mode =& ~IREAD;
wakeup(ip+2);
}
goto loop;
}
我們想將字節寫入管道輸入 u.u_count
. 首先我們需要鎖定索引節點(見下文 plock
/prele
).
然後我們檢查 inode 引用計數。 只要管道的兩端保持打開狀態,計數器就應該為 2。我們保留一個鏈接(來自 rp->f_inode
),所以如果計數器小於 2,那麼這應該意味著讀取進程已經關閉了它的管道末端。 換句話說,我們試圖寫入一個封閉的管道,這是一個錯誤。 第一個錯誤代碼 EPIPE
和信號 SIGPIPE
出現在 Unix 第六版中。
但即使傳送帶打開,它也可能已滿。 在這種情況下,我們釋放鎖並進入睡眠狀態,希望另一個進程將從管道中讀取並釋放足夠的空間。 當我們醒來時,我們回到起點,再次掛掉鎖,開始新的寫週期。
如果管道中有足夠的可用空間,那麼我們使用 i_size1
inode'a(空管道可以等於 0)指向它已經包含的數據的末尾。 如果有足夠的空間可以寫入,我們可以從 i_size1
對 PIPESIZ
. 然後我們釋放鎖並嘗試喚醒任何正在等待從管道讀取的進程。 我們回到開頭,看看我們是否設法寫入了我們需要的字節數。 如果失敗,則我們開始新的記錄週期。
通常參數 i_mode
inode用來存放權限 r
, w
и x
. 但是在管道的情況下,我們用位表示某個進程正在等待寫入或讀取 IREAD
и IWRITE
分別。 該進程設置標誌並調用 sleep()
, 預計將來其他進程會調用 wakeup()
.
真正的魔法發生在 sleep()
и wakeup()
. 它們是在
/*
* Give up the processor till a wakeup occurs
* on chan, at which time the process
* enters the scheduling queue at priority pri.
* The most important effect of pri is that when
* pri<0 a signal cannot disturb the sleep;
* if pri>=0 signals will be processed.
* Callers of this routine must be prepared for
* premature return, and check that the reason for
* sleeping has gone away.
*/
sleep(chan, pri) /* … */
/*
* Wake up all processes sleeping on chan.
*/
wakeup(chan) /* … */
調用的過程 sleep()
對於特定通道,稍後可能會被另一個進程喚醒,該進程將調用 wakeup()
對於同一個頻道。 writep()
и readp()
通過這樣的配對呼叫協調他們的行動。 注意 pipe.c
總是優先考慮 PPIPE
當被調用時 sleep()
, 所以所有 sleep()
可以被信號打斷。
現在我們已經了解了該功能的所有內容 readp()
:
readp(fp)
int *fp;
{
register *rp, *ip;
rp = fp;
ip = rp->f_inode;
loop:
/* Very conservative locking. */
plock(ip);
/*
* If the head (read) has caught up with
* the tail (write), reset both to 0.
*/
if(rp->f_offset[1] == ip->i_size1) {
if(rp->f_offset[1] != 0) {
rp->f_offset[1] = 0;
ip->i_size1 = 0;
if(ip->i_mode&IWRITE) {
ip->i_mode =& ~IWRITE;
wakeup(ip+1);
}
}
/*
* If there are not both reader and
* writer active, return without
* satisfying read.
*/
prele(ip);
if(ip->i_count < 2)
return;
ip->i_mode =| IREAD;
sleep(ip+2, PPIPE);
goto loop;
}
/* Read and return */
u.u_offset[0] = 0;
u.u_offset[1] = rp->f_offset[1];
readi(ip);
rp->f_offset[1] = u.u_offset[1];
prele(ip);
}
您可能會發現從下到上閱讀此函數更容易。 “讀取並返回”分支通常在管道中有一些數據時使用。 在這種情況下,我們使用 f_offset
讀取,然後更新相應偏移量的值。
在後續讀取中,如果讀取偏移量已達到,管道將為空 i_size1
在索引節點。 我們將位置重置為 0 並嘗試喚醒任何想要寫入管道的進程。 我們知道,當傳送帶裝滿時, writep()
睡著了 ip+1
. 現在管道是空的,我們可以喚醒它以恢復其寫入周期。
如果沒有什麼可讀的,那麼 readp()
可以設置一個標誌 IREAD
睡著了 ip+2
. 我們知道什麼會喚醒他 writep()
當它向管道寫入一些數據時。
評論於 u
» 我們可以將它們視為常規 I/O 函數,這些函數獲取文件、位置、內存中的緩衝區,併計算要讀取或寫入的字節數。
/*
* Read the file corresponding to
* the inode pointed at by the argument.
* The actual read arguments are found
* in the variables:
* u_base core address for destination
* u_offset byte offset in file
* u_count number of bytes to read
* u_segflg read to kernel/user
*/
readi(aip)
struct inode *aip;
/* … */
/*
* Write the file corresponding to
* the inode pointed at by the argument.
* The actual write arguments are found
* in the variables:
* u_base core address for source
* u_offset byte offset in file
* u_count number of bytes to write
* u_segflg write to kernel/user
*/
writei(aip)
struct inode *aip;
/* … */
至於“保守”封鎖,那麼 readp()
и writep()
鎖定 inode 直到它們完成或得到結果(即調用 wakeup
). plock()
и prele()
工作簡單:使用一組不同的調用 sleep
и wakeup
允許我們喚醒任何需要我們剛剛釋放的鎖的進程:
/*
* Lock a pipe.
* If its already locked, set the WANT bit and sleep.
*/
plock(ip)
int *ip;
{
register *rp;
rp = ip;
while(rp->i_flag&ILOCK) {
rp->i_flag =| IWANT;
sleep(rp, PPIPE);
}
rp->i_flag =| ILOCK;
}
/*
* Unlock a pipe.
* If WANT bit is on, wakeup.
* This routine is also used to unlock inodes in general.
*/
prele(ip)
int *ip;
{
register *rp;
rp = ip;
rp->i_flag =& ~ILOCK;
if(rp->i_flag&IWANT) {
rp->i_flag =& ~IWANT;
wakeup(rp);
}
}
一開始我不明白為什麼 readp()
不會導致 prele(ip)
通話前 wakeup(ip+1)
. 第一件事 writep()
在它的循環中調用,這個 plock(ip)
,這會導致死鎖,如果 readp()
尚未刪除其塊,因此代碼必須以某種方式正常工作。 如果你看 wakeup()
, 很明顯它只將休眠進程標記為準備執行,以便將來 sched()
真的推出了。 所以 readp()
原因 wakeup()
, 解鎖, 設置 IREAD
和電話 sleep(ip+2)
- 這一切之前 writep()
重新開始循環。
這完成了第六版中管道的描述。 簡單的代碼,深遠的影響。
Xv6,一個簡單的類 Unix 內核
創建一個核
該代碼包含清晰而周到的實現 pipealloc()
:
#define PIPESIZE 512
struct pipe {
struct spinlock lock;
char data[PIPESIZE];
uint nread; // number of bytes read
uint nwrite; // number of bytes written
int readopen; // read fd is still open
int writeopen; // write fd is still open
};
int
pipealloc(struct file **f0, struct file **f1)
{
struct pipe *p;
p = 0;
*f0 = *f1 = 0;
if((*f0 = filealloc()) == 0 || (*f1 = filealloc()) == 0)
goto bad;
if((p = (struct pipe*)kalloc()) == 0)
goto bad;
p->readopen = 1;
p->writeopen = 1;
p->nwrite = 0;
p->nread = 0;
initlock(&p->lock, "pipe");
(*f0)->type = FD_PIPE;
(*f0)->readable = 1;
(*f0)->writable = 0;
(*f0)->pipe = p;
(*f1)->type = FD_PIPE;
(*f1)->readable = 0;
(*f1)->writable = 1;
(*f1)->pipe = p;
return 0;
bad:
if(p)
kfree((char*)p);
if(*f0)
fileclose(*f0);
if(*f1)
fileclose(*f1);
return -1;
}
pipealloc()
設置所有其餘實現的狀態,其中包括函數 piperead()
, pipewrite()
и pipeclose()
. 實際的系統調用 sys_pipe
是一個包裝器實現
Linux 0.01
您可以找到 Linux 0.01 的源代碼。 在他的研究中研究管道的實現將是有益的 fs
/pipe.c
. 在這裡,一個 inode 用於表示管道,但管道本身是用現代 C 語言編寫的。如果您已經通過第六版代碼進行了破解,那麼您在這裡就不會有任何問題。 這就是函數的樣子 write_pipe()
:
int write_pipe(struct m_inode * inode, char * buf, int count)
{
char * b=buf;
wake_up(&inode->i_wait);
if (inode->i_count != 2) { /* no readers */
current->signal |= (1<<(SIGPIPE-1));
return -1;
}
while (count-->0) {
while (PIPE_FULL(*inode)) {
wake_up(&inode->i_wait);
if (inode->i_count != 2) {
current->signal |= (1<<(SIGPIPE-1));
return b-buf;
}
sleep_on(&inode->i_wait);
}
((char *)inode->i_size)[PIPE_HEAD(*inode)] =
get_fs_byte(b++);
INC_PIPE( PIPE_HEAD(*inode) );
wake_up(&inode->i_wait);
}
wake_up(&inode->i_wait);
return b-buf;
}
即使不查看結構定義,您也可以弄清楚索引節點引用計數是如何用於檢查寫入操作是否導致 SIGPIPE
. 除了逐字節的工作,這個函數很容易和上面的思路進行比較。 偶邏輯 sleep_on
/wake_up
看起來不那麼陌生。
現代 Linux 內核、FreeBSD、NetBSD、OpenBSD
我很快瀏覽了一些現代內核。 它們都沒有基於磁盤的實現(不足為奇)。 Linux 有自己的實現。 儘管三個現代 BSD 內核都包含基於 John Dyson 編寫的代碼的實現,但多年來它們彼此之間的差異太大了。
讀書 fs
/pipe.c
(在 Linux 上)或 sys
/kern
/sys_pipe.c
(在 *BSD 上),它需要真正的奉獻精神。 性能和對矢量和異步 I/O 等功能的支持在當今的代碼中非常重要。 並且內存分配、鎖和內核配置的細節都大不相同。 這不是大學操作系統入門課程所需要的。
無論如何,發掘一些舊模式對我來說很有趣(例如,生成 SIGPIPE
並返回 EPIPE
在所有這些如此不同的現代內核中寫入封閉管道時。 我可能永遠不會現場看到 PDP-11 計算機,但是從我出生前幾年編寫的代碼中仍然可以學到很多東西。
迪維·卡普爾 (Divi Kapoor) 於 2011 年撰寫的文章“
來源: www.habr.com