管道概念

? ?管道是Unix中最古老的進程間通信的形式,我們把從一個進程連接到另一個進程的一個數據流稱為一個“管道”,?管道的本質是固定大小的內核緩沖區;

? ?如:ps?aux?|?grep?httpd?|?awk?'{print?$2}'?

管道限制

? ?1)管道是半雙工的，數據只能向一個方向流動；需要雙方通信時，需要建立起兩個管道;

? ?2)匿名管道只能用于具有共同祖先的進程(如父進程與fork出的子進程)之間進行通信,?原因是pipe創建的是兩個文件描述符,?不同進程直接無法直接獲得;[通常，一個管道由一個進程創建，然后該進程調用fork，此后父子進程共享該管道]

匿名管道pipe

#include <unistd.h> int pipe(int pipefd[2]);

創建一無名管道

參數

? ?Pipefd:文件描述符數組,其中pipefd[0]表示讀端,pipefd[1]表示寫端

?

管道創建示意圖

/**示例: 從子進程向父進程發送數據 管道示意圖如上面第二副圖 **/ int main() {int fd[2];if (pipe(fd) == -1)err_exit("pipe error");pid_t pid = fork();if (pid == -1)err_exit("fork error");if (pid == 0) //子進程: 向管道中寫入數據{close(fd[0]); //關閉讀端string str("message from child process!");write(fd[1], str.c_str(), str.size()); //向寫端fd[1]寫入數據close(fd[1]);exit(EXIT_SUCCESS);}//父進程: 從管道中讀出數據close(fd[1]); //關閉寫端char buf[BUFSIZ] = {0};read(fd[0], buf, sizeof(buf));close(fd[0]);cout << buf << endl; } /**示例: 用管道模擬: ls | wc -w的運行 1.子進程運行ls 2.父進程運行wc -w 3.通過管道, 將子進程的輸出發送到wc的輸入 **/ int main() {int pipefd[2];if (pipe(pipefd) == -1)err_exit("pipe error");pid_t pid = fork();if (pid == -1)err_exit("fork error");if (pid == 0) //子進程{close(pipefd[0]); //關閉讀端//使得STDOUT_FILENO也指向pipefd[1],亦即ls命令的輸出將打印到管道中dup2(pipefd[1], STDOUT_FILENO); //此時可以關閉管道寫端close(pipefd[1]);execlp("/bin/ls", "ls", NULL);//如果進程映像替換失敗,則打印下面出錯信息cerr << "child execlp error" << endl;;exit(EXIT_FAILURE);}//父進程close(pipefd[1]); //關閉寫端//使得STDIN_FILENO也指向pipefd[2],亦即wc命令將從管道中讀取輸入dup2(pipefd[0], STDIN_FILENO);close(pipefd[0]);execlp("/usr/bin/wc", "wc", "-w", NULL);cerr << "parent execlp error" << endl;exit(EXIT_FAILURE); }

匿名管道讀寫規則

規則?1)管道空時

? ?O_NONBLOCK?disable：read調用阻塞，即進程暫停執行，一直等到有數據來到為止。

? ?O_NONBLOCK?enable：read調用返回-1，errno值為EAGAIN。

//驗證 int main() {int pipefd[2];if (pipe(pipefd) != 0)err_exit("pipe error");pid_t pid = fork();if (pid == -1)err_exit("fork error");if (pid == 0) //In Child, Write pipe{sleep(10);close(pipefd[0]); //Close Read pipestring str("I Can Write Pipe from Child!");write(pipefd[1],str.c_str(),str.size()); //Write to pipeclose(pipefd[1]);exit(EXIT_SUCCESS);}//In Parent, Read pipeclose(pipefd[1]); //Close Write pipechar buf[1024] = {0};//Set Read pipefd UnBlock! 查看在下面四行語句注釋的前后有什么區別 // int flags = fcntl(pipefd[0],F_GETFL, 0); // flags |= O_NONBLOCK; // if (fcntl(pipefd[0],F_SETFL,flags) == -1) // err_exit("Set UnBlock error");int readCount = read(pipefd[0],buf,sizeof(buf)); //Read from pipeif (readCount < 0)//read立刻返回,不再等待子進程發送數據err_exit("read error");cout << "Read from pipe: " << buf << endl;close(pipefd[0]); }

規則?2)管道滿時

? ?O_NONBLOCK?disable：?write調用阻塞，直到有進程讀走數據

? ?O_NONBLOCK?enable：調用返回-1，errno值為EAGAIN

/** 驗證規則2) 同時測試管道的容量 **/ int main() {if (signal(SIGPIPE, handler) == SIG_ERR)err_exit("signal error");int pipefd[2];if (pipe(pipefd) != 0)err_exit("pipe error");// 將管道的寫端設置成為非阻塞模式// 將下面三行注釋之后查看效果int flags = fcntl(pipefd[1], F_GETFL, 0);if (fcntl(pipefd[1], F_SETFL, flags|O_NONBLOCK) == -1)err_exit("fcntl set error");int count = 0;while (true){if (write(pipefd[1], "A", 1) == -1){cerr << "write pipe error: " << strerror(errno) << endl;break;}++ count;}cout << "pipe size = " << count << endl; }

3)如果所有管道寫端對應的文件描述符被關閉，則read返回0

//驗證規則3) int main() {int pipefd[2];if (pipe(pipefd) != 0)err_exit("pipe error");pid_t pid = fork();if (pid == -1)err_exit("fork error");else if (pid == 0){close(pipefd[1]);exit(EXIT_SUCCESS);}close(pipefd[1]);sleep(2);char buf[2];if (read(pipefd[0], buf, sizeof(buf)) == 0)cout << "sure" << endl; }

4)如果所有管道讀端對應的文件描述符被關閉，則write操作會產生信號SIGPIPE

//驗證規則4) int main() {if (signal(SIGPIPE, handler) == SIG_ERR)err_exit("signal error");int pipefd[2];if (pipe(pipefd) != 0)err_exit("pipe error");pid_t pid = fork();if (pid == -1)err_exit("fork error");else if (pid == 0){close(pipefd[0]);exit(EXIT_SUCCESS);}close(pipefd[0]);sleep(2);char test;if (write(pipefd[1], &test, sizeof(test)) < 0)err_exit("write error"); }

Linux?PIPE特征

? ?1)當要寫入的數據量不大于PIPE_BUF時，Linux將保證寫入的原子性。

? ?2)當要寫入的數據量大于PIPE_BUF時，Linux將不再保證寫入的原子性。

man說明:

POSIX.1-2001?says?that?write(2)s?of?less?than?PIPE_BUF?bytes?must?be?atomic:??

the??output?data?is?written?to?the?pipe?as?a?contiguous?sequence.??

Writes?of?more?than?PIPE_BUF?bytes?may?be?nonatomic:?

the?kernel?may?interleave?the?data?with??data??written??by??other??processes.??

POSIX.1-2001?requires?PIPE_BUF?to?be?at?least?512?bytes.??

(On?Linux,?PIPE_BUF?is?4096?bytes.?在Linux當中,?PIPE_BUF為4字節).?

The?precise?semantics?depend?on?whether?the?file?descriptor?is??non-blocking(O_NONBLOCK),??

whether??there??are??multiple?writers?to?the?pipe,?and?on?n,?the?number?of?bytes?to?be?written:

O_NONBLOCK?disabled(阻塞),?n?<=?PIPE_BUF

? ?All?n?bytes?are?written?atomically;?write(2)?may?block?if?there?is?not?room?for??n?bytes?to?be?written?immediately

O_NONBLOCK?enabled(非阻塞),?n?<=?PIPE_BUF

? ?If?there?is?room?to?write?n?bytes?to?the?pipe,?then?write(2)?succeeds?immediately,?writing?all?n?bytes;?

otherwise?write(2)?fails,?with?errno?set?to?EAGAIN(注意:?如果空間不足以寫入數據,?則一個字節也不寫入,?直接出錯返回).

?

O_NONBLOCK?disabled,?n?>?PIPE_BUF

? ?The?write?is?nonatomic:?the??data??given??to??write(2)??may??be??interleaved??with?write(2)s?by?other?process;?

the?write(2)?blocks?until?n?bytes?have?been?written.

O_NONBLOCK?enabled,?n?>?PIPE_BUF

? ?If??the??pipe??is?full,?then?write(2)?fails,?with?errno?set?to?EAGAIN(此時也是沒有一個字符寫入管道).??

Otherwise,?from?1?to?n?bytes?may?be?written?(i.e.,?a?"partial?write"?may??occur;??

the??caller?should??check??the??return?value?from?write(2)?to?see?how?many?bytes?were?actually?written),?

and?these?bytes?may?be?interleaved?with?writes?by?other?processes.

/** 驗證: 已知管道的PIPE_BUF為4K, 我們啟動兩個進程A, B向管道中各自寫入68K的內容, 然后我們以4K為一組, 查看管道最后一個字節的內容, 多運行該程序幾次, 就會發現這68K的數據會有交叉寫入的情況 **/ int main() {const int TEST_BUF = 68 * 1024; //設置寫入的數據量為68Kchar bufA[TEST_BUF];char bufB[TEST_BUF];memset(bufA, 'A', sizeof(bufA));memset(bufB, 'B', sizeof(bufB));int pipefd[2];if (pipe(pipefd) != 0)err_exit("pipe error");pid_t pid;if ((pid = fork()) == -1)err_exit("first fork error");else if (pid == 0) //第一個子進程A, 向管道寫入bufA{close(pipefd[0]);int writeBytes = write(pipefd[1], bufA, sizeof(bufA));cout << "A Process " << getpid() << ", write "<< writeBytes << " bytes to pipe" << endl;exit(EXIT_SUCCESS);}if ((pid = fork()) == -1)err_exit("second fork error");else if (pid == 0) //第二個子進程B, 向管道寫入bufB{close(pipefd[0]);int writeBytes = write(pipefd[1], bufB, sizeof(bufB));cout << "B Process " << getpid() << ", write "<< writeBytes << " bytes to pipe" << endl;exit(EXIT_SUCCESS);}// 父進程close(pipefd[1]);sleep(2); //等待兩個子進程寫完char buf[4 * 1024]; //申請一個4K的bufint fd = open("save.txt", O_WRONLY|O_TRUNC|O_CREAT, 0666);if (fd == -1)err_exit("file open error");while (true){int readBytes = read(pipefd[0], buf, sizeof(buf));if (readBytes == 0)break;if (write(fd, buf, readBytes) == -1)err_exit("write file error");cout << "Parent Process " << getpid() << " read " << readBytes<< " bytes from pipe, buf[4095] = " << buf[4095] << endl;} }

附-管道容量查詢

man?7?pipe

?

注意:?管道的容量不一定就等于PIPE_BUF,?如在Ubuntu中,?管道容量為64K,?而PIPE_BUF為4K.

總結

以上是生活随笔為你收集整理的Linux IPC实践(2) --匿名PIPE的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。