IPC研究（2） --- pipes

==================================================================== IPC ---- Pipes Related system calls: popen  (high-level, implemented by actually invoking shell and then command) pclose (redirect shell commands input and output, you can use these two calls to write tests) pipe exec dup mkfifo unlink Basics: redirect input and output (parent process) -->file_pipes[1] --> [PIPE] --> file_pipes[0] --> (child process) The child process may use the exec call to become a different program. You've got to pass the file descriptor as a parameter to the new program. pipe is unidirectional By default, file descriptors remain open across an execve().  File descriptors that  are  marked  close-on-exec are  closed;  see  the description of FD_CLOEXEC in fcntl(2).  也就是说，用execve或者exec-like functions来替换掉原进程（A）执行新进程（B）时，A的fd会被保留下来。此时，新进程（B）可使用这些fd。好比Example 1中，子进程被替换掉后，新进程其实是有子进程的fd的，之因此要传参数给新进程，是由于它没法知道哪一个fd是pipe的read end。若是事先知道的话，是不须要传参数的；好比，咱们知道此程序中pipe的read end是3，那么在procB.c中，read(fd, buf, BUFSIZ);改为read(3, buf, BUFSIZ);也是彻底正确的。 在全部的fd中，有3个比较特别，0，1，2；它们的语义是预约义的，分别为标准输入，标准输出和标准错误输出。 由以上两点，咱们能够想到，若是利用dup函数，复制一个fd，而dup本身产生的fd是0（咱们只要先close（0）就能够了），那么，此时标准输入实际上是从这个fd指向的文件（inode）中来读取东西。若是以后，咱们用execve来替换掉这个进程执行新进程，那么新进程的标准输入，实际上是fd指向的文件。若是这个文件是一个pipe，那么新进程的标准输入其实就是pipe的read end。 由上，咱们能够在不改变原来某些程序的状况下，编写新程序（pipe，fork, close, dup, execve），来创建一个pipe。 新程序 --》 （pipe） --》 原程序 （note：每一个进程有独立的地址空间和fd table，因此此时若是系统运行其余程序，打开一个文件，其fd依然是3，只是这个fd指向的是系统中另一个inode）。 mkfifo filename (create a named pipe in command line) 命名管道其实是在系统的file system中建立了一个named file,它的类型是pipe。因为此FIFO存在于file system中，不相关的程序就能够经过它来进行数据交互（不像unnamed pipe对数据交互双方有很强的耦合要求。） 注意，管道是单向的。[procA] --- named pipe ---> [procB] 若是想让数据双向传输，有如下两种方法： 1. [procA] --- named pipe1 ---> [procB]    [procA] <---named pipe2 ---- [procB] 2. [procA] --- named pipe1 ---> [procB]     双方关闭管道，从新打开以下    [procB] --- named pipe1 ---> [procA] 咱们一般使用方法1. 注意，打开fifo是可能会block的，若是没有O_NONBLOCK标志的话。 若是有多个writer和一个reader，会出现什么状况呢？ 【procA】| 【procB】||=== named pipe ===> 【procD】 【procC】| ABC的write request就可又能会交错。好比A的一个write request写了一半，B的write request就写进来了。 如何避免这种状况呢？让write request < PIPE_BUF字节，系统就能保证，每一个write行为都是atomic的。 Analysis: ==> Unnamed Pipes pipe IPC机制总算是比signal要强大一点。由于它能够传输的信息量比较大，而不只仅只是一个integer。 可是，unnamed pipe的IPC机制的使用颇有限制，由于它本质上是经过一对fd来进行通信的。因此，它的应用基本只能局限在如下两种状况： 1. 父子进程，由于子进程有parent进程的fd拷贝。 2. 子进程exec成另一个进程，同时经过arg将fd传递给新进程。 通讯双方的耦合程度不言而喻，基本上比signal还要差。由于signal还有办法经过程序名字得到进程名，从而能够经过kill调用来进行通信。pipe IPC机制而言，你如何去获得fd？/proc底下某进程有一堆fd，你咋知道哪一个是哪一个？若是进程A只有一对pipe fd，那么在/proc目录下还能够经过readlink来进行判断，那两个fd属于pipe；但要是A有好几个pipe呢？因此，unnamed pipe的应用场景，基本上就是以上两种。 3. pipe做为标准输入输出使用，达到如下效果【新程序 --》 （pipe） --》 原程序】。分析见上。 （注意：这个应用场景颇有用！好比，咱们本身写了一个程序，从标准输入读东西来进行处理的，那么咱们能够用以上方法，创建pipe在另一个程序中，使其输入自动化而且能够对输入进行控制。同理，咱们能够用dup来替换掉标准输出，从而对某些程序的输出进程自动化控制和分析。） 不过pipe这种单刀直入的思想，在不少应用场景下，颇有效果。 ==> Named Pipes Named Pipe机制比unnamed pipe机制要强大一点。缘由是它的reader和writer的耦合程度相对较低（只须要知道fifo的存在位置，如/tmp/my_fifo）。另外，从上面的分析能够知道，若是多个writer的write request的字节数少于PIPE_BUF的话，每一个write行为都是atomic的。这一点，使named pipe能够做为client/server的通信。 由此，named pipe的应用场景要比unnamed pipe要广。 能够以下 [client1] | [client2] | ---- serv_pipe ----> [server] [client3] |               | --- cli_pipe_1 --> [client1] [server] ---- | --- cli_pipe_2 --> [client2]               | --- cli_pipe_3 --> [client3] 只要定义好协议格式就能够进行处理了。 另外，open pipe若是没有O_NONBLOCK，是会block的。 好比open(pipe_fd, O_RDONLY);会一种block到另一个进程open(pipe_fd, O_WRONLY). 同理open(pipe_fd, O_WRONLY);会一种block到另一个进程open(pipe_fd, O_RDONLY). 因此，在写client/server时要注意对pipe操做的顺序，以避免发生两边都blcok了。             Examples: Example 1 ---- unamed pipes (procA sends some data to procB) /**  * procA.c  *  * write to pipe  **/ #include <unistd.h> #include <stdio.h> #include <stdlib.h> #include <string.h> void main() {     int file_pipes[2];     const char some_data[] = "Hello! I'm process A!";     char buf[BUFSIZ+1];     pid_t pid;     int ret;     memset(buf, 0, sizeof(buf));     ret = pipe(file_pipes);     if (ret < 0)     {         perror("create pipe failed");         exit(-1);     }     else            /* ret == 0 */     {         pid = fork();         if (pid <0)         {             perror("fork() failed");             exit(-1);         }         else if (pid == 0) /* child process */         {             sprintf(buf, "%d", file_pipes[0]);             execl("procB", "procB", buf, NULL);             exit(-1); /* we should not get here */         }         else        /* parent process */         {             write(file_pipes[1], some_data, strlen(some_data));             printf("[%d] ---- write finished! \n", getpid());         }     } } /**  * procB.c  *  * read from a pipe, display its contents  **/ #include <unistd.h> #include <stdio.h> #include <stdlib.h> #include <string.h> void main(int argc, char *argv[]) {   int fd;   char buf[BUFSIZ+1];   memset(buf, 0, sizeof(buf));   sscanf(argv[1], "%d", &fd);   read(fd, buf, BUFSIZ);   printf("[%d] ---- read data finish: %s \n", getpid(), buf); } Example 2 ---- pipes as standard input and standard output /**  * pipe_as_stdin.c  * procA --[pipe] --> procB  * procA passes a string to procB through a pipe  * procB is 'wc' program, it receives the string as if it was passed in from stdin  **/ #include <unistd.h> #include <stdio.h> #include <stdlib.h> #include <string.h> void main() {     char data[] = "Hello! This is a string to be passed to wc as an input!";     int file_pipes[2];     pid_t pid;     int ret;     ret = pipe(file_pipes);     if (ret < 0)     {         perror("create pipes failed\n");         exit(-1);     }     else     {         pid = fork();         if (pid < 0)         {             perror("fork failed");             exit(-1);         }         else if (pid == 0) /* child process */         {             close(0);             dup(file_pipes[0]);             close(file_pipes[1]);             close(file_pipes[0]);             execlp("wc", "wc", NULL);             exit(-1); /* we should never get here */         }         else         /* parent process */         {             close(file_pipes[0]);             write(file_pipes[1], data, sizeof(data));             close(file_pipes[1]);             printf("[%d] ---- write finished!\n", getpid());         }     } } /**  * pipe_as_stdout.c  * analyse how many hidden files there are in this directory  * 'ls -a' --[pipe]--> my program  **/ #include <unistd.h> #include <stdio.h> #include <stdlib.h> #include <string.h> void main() {     char buf[BUFSIZ+1];     int file_pipes[2];     pid_t pid;     int ret;     ret = pipe(file_pipes);     if (ret < 0)     {         perror("create pipes failed\n");         exit(-1);     }     else     {         pid = fork();         if (pid < 0)         {             perror("fork failed");             exit(-1);         }         else if (pid == 0) /* child process */         {             close(1);             dup(file_pipes[1]);             close(file_pipes[0]);             close(file_pipes[1]);             execlp("ls", "ls", "-a", NULL);             exit(-1); /* we should never get here */         }         else         /* parent process */         {             close(file_pipes[1]);             memset(buf, 0, sizeof(buf));             read(file_pipes[0], buf, BUFSIZ);             close(file_pipes[0]);             printf("[%d] ---- read finished -- \n %s \n", getpid(), buf);         }     } } Example 3 --- name pipes (client/server using fifo) #ifndef _CS_PIPE_COMMON_H #define _CS_PIPE_COMMON_H #include <unistd.h> #include <stdlib.h> #include <string.h> #include <stdio.h> #include <fcntl.h> #include <limits.h> #include <sys/types.h> #include <sys/stat.h> #include <ctype.h> #define SERVER_FIFO_NAME "/tmp/serv_fifo" #define CLIENT_FIFO_NAME "/tmp/cli_%d_fifo" #define ACTION_LEN 32 #define DATA_LEN 1024 #define BUFFER_SIZE PIPE_BUF struct cs_packet_t {     pid_t cli_pid;     char action[ACTION_LEN];     char data[DATA_LEN]; }; #endif /**  * client.c  *  * client/server using named pipes (FIFO)  **/ #include "common.h" static void print_pkt(struct cs_packet_t *pkt) {     printf("==========packet=================\n");     printf("client pid = %d \n", pkt->cli_pid);     printf("client action = %s \n", pkt->action);     printf("client data = %s \n", pkt->data);     printf("\n"); } void main() {     struct    cs_packet_t pkt;     pkt.cli_pid = getpid();     /* fifo initialization */     int serv_fd;     int cli_fd;     serv_fd = open(SERVER_FIFO_NAME, O_WRONLY);     if (serv_fd < 0)     {         perror("open server fifo failed");         exit(-1);     }     char client_fifo[32];     sprintf(client_fifo, CLIENT_FIFO_NAME, pkt.cli_pid);     int ret = mkfifo(client_fifo, 0777);     if (ret < 0)     {         perror ("create client fifo failed");         exit(-1);     }     /* loop to send the server uppercase requests and down case requests */     for (;;)     {         printf("<<debugging client>> --- in for loop\n");         sprintf(pkt.action, "upcase");         sprintf(pkt.data, "Data From Client %d", pkt.cli_pid);         write(serv_fd, &pkt, sizeof(pkt));         printf("sent packet finished: \n");         print_pkt(&pkt);         cli_fd = open(client_fifo, O_RDONLY); /* open client fifo for result */         struct cs_packet_t ret_pkt;         int read_res = read(cli_fd, &ret_pkt, sizeof(ret_pkt));         if (read_res > 0)         {             printf("receive packet form server: \n");             print_pkt(&ret_pkt);         }         close(cli_fd);         sleep(2);         sprintf(pkt.action, "downcase");         sprintf(pkt.data, "Data From Client %d", pkt.cli_pid);         write(serv_fd, &pkt, sizeof(pkt));         printf("sent packet finished: \n");         print_pkt(&pkt);         cli_fd = open(client_fifo, O_RDONLY); /* open client fifo for result */         read_res = read(cli_fd, &ret_pkt, sizeof(ret_pkt));         if (read_res > 0)         {             printf("receive packet form server: \n");             print_pkt(&ret_pkt);         }         close(cli_fd);         sleep(2);     } } /**  * server.c  *  * client/server using named pipes (FIFO)  **/ #include "common.h" #include <pthread.h> static void print_pkt(struct cs_packet_t *pkt) {     printf("==========packet=================\n");     printf("client pid = %d \n", pkt->cli_pid);     printf("client action = %s \n", pkt->action);     printf("client data = %s \n", pkt->data);     printf("\n"); } /**  * function: thread_handle_requests  *  * thread which handles client requests, sends back the result to client through cli_pipe_%d pipe and then exits  * client1 --(pkt1)--> |                  | --> pkt1 (thread1)  * client2 --(pkt2)--> | --> serv_buf --> | --> pkt2 (thread2)  * client3 --(pkt3)--> |                  | --> pkt3 (thread3)  * This is not a very good design. For practical server, we should queue the requests.  * N threads may loop to take the requests out of the queue and then handles it.  * This avoids the overhead of thead creation and destruction.  **/ void *thread_handle_requests(void *arg) /* arg is of type struct cs_packet_t */ {     struct cs_packet_t *pkt = (struct cs_packet_t*)malloc(sizeof(struct cs_packet_t));     if (pkt == NULL)     {         perror("Not Enough Memory In Heap!");         exit(-1);     }     memcpy(pkt, arg, sizeof(struct cs_packet_t));     printf("<<debugging server>> --- \n");     print_pkt(pkt);     pid_t cli_pid = pkt->cli_pid;     char *action = pkt->action;     char *data = pkt->data;     char client_fifo[32];    /* name of this client's fifo, cli_fifo_%d */     int client_fifo_fd;     memset(client_fifo, 0, sizeof(client_fifo));     sprintf(client_fifo, CLIENT_FIFO_NAME, cli_pid);     client_fifo_fd = open(client_fifo, O_WRONLY);     if (client_fifo_fd == -1)     {         perror("open client pipe failed");         exit(-1);     }     if (!strcmp(pkt->action, "upcase"))     {         char *tmp_char_ptr = data;         while (*tmp_char_ptr)         {             *tmp_char_ptr = toupper(*tmp_char_ptr);             tmp_char_ptr++;         }         write(client_fifo_fd, pkt, sizeof(struct cs_packet_t));     }     else if (!strcmp(pkt->action, "downcase"))     {         char *tmp_char_ptr = data;         while (*tmp_char_ptr)         {             *tmp_char_ptr = tolower(*tmp_char_ptr);             tmp_char_ptr++;         }         write(client_fifo_fd, pkt, sizeof(struct cs_packet_t));     }     else     {         sprintf(data, "Action %s not supported", pkt->action);         write(client_fifo_fd, pkt, sizeof(struct cs_packet_t));     }     close(client_fifo_fd);     free(pkt);     return NULL; } void main() {     char serv_buf[BUFFER_SIZE];     struct cs_packet_t pkt;     int server_fifo_fd;     pthread_t th;     /* make server fifo */     mkfifo(SERVER_FIFO_NAME, 0777);     server_fifo_fd = open(SERVER_FIFO_NAME, O_RDONLY);     if (server_fifo_fd == -1)     {         perror("create server fifo failed");         exit(-1);     }     /* read from server fifo */     for (;;)     {         int read_res = read(server_fifo_fd, &pkt, sizeof(pkt));         if (read_res > 0)         {             printf("<<debugging server>>----received a packet from %d\n", pkt.cli_pid);             /* create a thread to handle this request */             int ret = pthread_create(&th, NULL, thread_handle_requests, (void*)&pkt);             if (ret < 0)             {                 perror("create thread failed");                 exit(-1);             }         }     } }