进程间通信（IPC）系列管道（pipe）

　　今天从源码的角度分析下进程间通信之管道(pipe )。
　　什么是管道 ？
　　所谓管道，是指用于连接一个读进程和一个写进程，以实现它们之间通信的共享文件，又称 pipe  文件。
　　向管道（共享文件）提供输入的发送进程（即写进程），以字符流形式将大量的数据送入管道；而接收管道输出的接收进程（即读进程），可从管道中接收数据。由于发送进程和接收进程是利用管道进行通信的，故又称管道通信。
　　为了协调双方的通信，管道通信机制必须提供以下3 方面的协调能力。互斥。当一个进程正在对 pipe  进行读/写操作时，另一个进程必须等待。同步。当写（输入）进程把一定数量（如4KB）数据写入 pipe  后，便去睡眠等待，直到读（输出）进程取走数据后，再把它唤醒。当读进程读到一空 pipe  时，也应睡眠等待，直至写进程将数据写入管道后，才将它唤醒。对方是否存在。只有确定对方已存在时，才能进行通信。
　　管道的应用
　　管道是利用 pipe()  系统调用而不是利用 open()  系统调用建立的。pipe()   调用的原型是：int pipe（int fd[2]）
　　我们看到，有两个文件描述符与管道结合在一起，一个文件描述符用于管道的read()  端，一个文件描述符用于管道的 write()  端。
　　由于一个函数调用不能返回两个值，pipe()  的参数是指向两个元素的整型数组的指针，它将由调用两个所要求的文件描述符填入。
　　fd[0]  元素将含有管道 read()  端的文件描述符，而 fd[1]  含有管道 write()  端的文件描述符。系统可根据 fd[0]  和 fd[1]  分别找到对应的 file  结构。
　　注意，在 pipe  的参数中，没有路径名，这表明，创建管道并不像创建文件一样，要为它创建一个目录连接。这样做的好处是，其他现存的进程无法得到该管道的文件描述符，从而不能访问它。
　　那么，两个进程如何使用一个管道来通信呢？
　　我们知道，fork()  和 exec()  系统调用可以保证文件描述符的复制品既可供双亲进程使用，也可供它的子女进程使用。也就是说，一个进程用 pipe()  系统调用创建管道，然后用 fork()  调用创建一个或多个进程，那么，管道的文件描述符将可供所有这些进程使用。
　　这里更明确的含义是：一个普通的管道仅可供具有共同祖先的两个进程之间共享，并且这个祖先必须已经建立了供它们使用的管道。
　　注意，在管道中的数据始终以和写数据相同的次序来进行读，这表示 lseek() 系统调用对管道不起作用。
　　下面给出在两个进程之间设置和使用管道的简单程序：
　　#include  #include  #include   int main（void） { int fd[2], nbytes; pid_t childpid; char string[] = ＂Hello, world! ＂; char readbuffer[80];  pipe(fd);  if((childpid = fork()) == -1） { printf(＂Error:fork＂); exit(1); }  if(childpid == 0) /* 子进程是管道的读进程 */ { close(fd[1]); /*关闭管道的写端 */ nbytes = read(fd[0], readbuffer, sizeof(readbuffer)); printf(＂Received string: %s＂, readbuffer); exit(0); } else /* 父进程是管道的写进程 */ { close(fd[0]); /*关闭管道的读端 */ write(fd[1], string, strlen(string));  }  return(0); }
　　注意，在这个例子中，为什么这两个进程都关闭它所不需的管道端呢？
　　这是因为写进程完全关闭管道端时，文件结束的条件被正确地传递给读进程。而读进程完全关闭管道端时，写进程无需等待继续写数据。
　　阻塞读和写分别成为对空和满管道的默认操作，这些默认操作也可以改变，这就需要调用 fcntl()  系统调用，对管道文件描述符设置 O_NONBLOCK  标志可以忽略默认操作：#include   fcntl（fd,F_SETFL,O_NONBlOCK）;
　　上述例子如下图：
　　管道的源码分析
　　pipefs 初始化
　　pipefs  是一种简单的、虚拟的文件系统类型，因为它没有对应的物理设备，因此其安装时不需要块设备。大部分文件系统是以模块的形式来实现的。该文件系统相关的代码在 fs/pipe.c  中：
　　static struct file_system_type pipe_fs_type = { .name = ＂pipefs＂, .get_sb = pipefs_get_sb, .kill_sb = kill_anon_super, };   static int __init init_pipe_fs(void) { /* pipe_fs_type 链接到file_systems 链表可以通过读/proc/filesystems 找到  ＂pipefs＂入口点，在那里，＂nodev＂标志表示没有设置 FS_REQUIRES_DEV 标志，即该文件系统没有对应的物理设备。 */ int err = register_filesystem(&pipe_fs_type); if (!err) { //安装pipefs 文件系统 pipe_mnt = kern_mount(&pipe_fs_type); if (IS_ERR(pipe_mnt)) { err = PTR_ERR(pipe_mnt); unregister_filesystem(&pipe_fs_type); } }  return err; }   //pipefs 文件系统是作为一个模块来安装的 fs_initcall(init_pipe_fs); module_exit(exit_pipe_fs); //模块卸载函数
　　上述就是初始化时注册 pipefs  文件系统的过程，操作如下：把 pipe_fs_type  链接到 file_systems  链表中。创建一个struct vfsmount  结构，然后把该结构赋值给全局变量pipe_mnt 。该 vfsmount  结构通过 super_block  与 pipefs  文件系统进行了关联。
　　pipe 的创建
　　pipefs  文件系统的入口点就是pipe() 系统调用，其内核实现函数为sys_pipe() ，而真正的工作是调用 do_pipe()  函数来完成的，其代码在 fs/pipe.c  中： int do_pipe(int *fd) {     struct file *fw, *fr;     int fdw, fdr;      //创建管道写端的file结构     fw = create_write_pipe();         //在写端的file结构基础上构建读端     fr = create_read_pipe(fw);      //创建读端fd     fdr = get_unused_fd();      //创建写端fd     fdw = get_unused_fd();      //fd 和 file进行关联     fd_install(fdr, fr);     fd_install(fdw, fw);      //返回读写端fd     fd[0] = fdr;     fd[1] = fdw;      ...      return 0; }   struct file *create_write_pipe(void) {     ...      struct qstr name = { .name = ＂＂ };      //创建 file 结构     f = get_empty_filp();      //创建一个pipe相关的 inode     inode = get_pipe_inode();      //创建一个dentry结构     dentry = d_alloc(pipe_mnt->mnt_sb->s_root, &name);      //inode 和 dentry 相关联     d_instantiate(dentry, inode);      // pipe 和 pipe_mnt 关联     f->f_path.mnt = mntget(pipe_mnt);      //file 与 dentry 相关联     f->f_path.dentry = dentry;      //该file是只写的     f->f_flags = O_WRONLY;      //该pipe的可操作方法     f->f_op = &write_pipe_fops;      ...      return f; }    struct file *create_read_pipe(struct file *wrf) {     //创建一个file结构用于读     struct file *f = get_empty_filp();      //file 与 已有的dentry、inode、struct vfsmount 相关联     f->f_path.mnt = mntget(wrf->f_path.mnt);     f->f_path.dentry = dget(wrf->f_path.dentry);     f->f_mapping = wrf->f_path.dentry->d_inode->i_mapping;      //该file是只读的     f->f_flags = O_RDONLY;     f->f_op = &read_pipe_fops;      f->f_mode = FMODE_READ;     f->f_version = 0;      return f; }
　　do_pipe()  的操作也很简单，操作如下：创建管道的读写端关联的 file 、inode 、dentry  结构。把 fd  和 file  进行联系，并返回给用户关于管道的读写描述符。创建一个 pipe_inode_info  对象，该结构指向具体的数据页内存缓冲区。后续所有向管道的读写均操作于该数据页内存缓冲区中。
　　上述 pipe  的初始化中，创建的 pipe_inode_info  对象记录了 pipe  读写过程中所有的得数据。pipe_inode_info  对象的结构如下struct pipe_inode_info { wait_queue_head_t wait;//存储等待读写进程的等待队列 /* nrbufs: 写入但还未被读取的数据占用缓冲区的页数 curbuf：当前正在读取环形缓冲区中的页节点 */ unsigned int nrbufs, curbuf; struct page *tmp_page; //临时缓冲区页面 unsigned int readers; //正在读取pipe的读进程数目 unsigned int writers; //正在写pipe的写进程数目 unsigned int waiting_writers; //等待管道可以写的进程数目 ... struct inode *inode; //pipe 对应的inode结构 struct pipe_buffer bufs[PIPE_BUFFERS]; //环形缓冲区，每个元素对应一个内存页 };
　　经过上述的一系列初始化，整个管道的内存结构如下图所示
　　pipe_read
　　static ssize_t pipe_read(struct kiocb *iocb, const struct iovec *_iov, unsigned long nr_segs, loff_t pos) { ...  //要读的数据长度 total_len = iov_length(iov, nr_segs);  do_wakeup = 0; ret = 0; //读之前先加锁 mutex_lock(&inode->i_mutex);  //循环读数据 for (;;) { int bufs = pipe->nrbufs; if (bufs) { int curbuf = pipe->curbuf; struct pipe_buffer *buf = pipe->bufs + curbuf; size_t chars = buf->len;  //待读的数据比要读的数据多，则设置要读的长度 if (chars > total_len) chars = total_len;  error = ops->confirm(pipe, buf); if (error) { if (!ret) error = ret; break; }  atomic = !iov_fault_in_pages_write(iov, chars); redo:  //把数据拷贝到用户缓冲区 addr = ops->map(pipe, buf, atomic); error = pipe_iov_copy_to_user(iov, addr + buf->offset, chars, atomic); ops->unmap(pipe, buf, addr); if (unlikely(error)) { ... }  ret += chars; buf->offset += chars; buf->len -= chars; //该页的数据全部读完，释放该page if (!buf->len) { buf->ops = NULL; ops->release(pipe, buf); curbuf = (curbuf + 1) & (PIPE_BUFFERS-1); pipe->curbuf = curbuf; pipe->nrbufs = --bufs; do_wakeup = 1; } total_len -= chars; if (!total_len) break; /* common path: read succeeded */ } //if (bufs) //若该页没读完，继续循环读，若该page读完，则读下一个page if (bufs) /* More to do? */ continue;  if (!pipe->writers) break;  if (!pipe->waiting_writers) { if (ret) break; //非阻塞跳出循环，不进行休眠 if (filp->f_flags & O_NONBLOCK) { ret = -EAGAIN; break; } }  if (signal_pending(current)) { if (!ret) ret = -ERESTARTSYS; break; } //唤醒写进程 if (do_wakeup) { wake_up_interruptible_sync(&pipe->wait); kill_fasync(&pipe->fasync_writers, SIGIO,POLL_OUT); }  //让出cpu，进行休眠，等待条件唤醒 pipe_wait(pipe);  } // for  mutex_unlock(&inode->i_mutex);  if (do_wakeup) { wake_up_interruptible_sync(&pipe->wait); kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT); } if (ret > 0) file_accessed(filp);  return ret; }
　　读的操作很简单，操作如下：读之前先锁定内存。从缓冲区中获取数据页，把页中数据拷贝到用户缓冲区中。数据全部被读完，则发送信号唤醒写进程，同时读进程让出 CPU  进行休眠。
　　读取数据的过程如下
　　pipe_write
　　static ssize_t pipe_write(struct kiocb *iocb, const struct iovec *_iov, unsigned long nr_segs, loff_t ppos) { ...  //需要写入数据的长度 total_len = iov_length(iov, nr_segs);  do_wakeup = 0;  mutex_lock(&inode->i_mutex);  //管道读端已关闭，返回SIGPIPE信号 if (!pipe->readers) { send_sig(SIGPIPE, current, 0); ret = -EPIPE; goto out; }  /* We try to merge small writes  curbuf：当前的pipe缓冲节点 nrbufs：非空的pipe缓冲节点数目 buffers：buf缓冲区总数目 buf->offset：页内可用数据的偏移量 buf->len ：可用数据的长度 buf->offset + buf->len ：页内可以往有效数据后追加数据的下标  */ //获取完整页之外的数量 chars = total_len & (PAGE_SIZE-1); /* size of the last buffer */ if (pipe->nrbufs && chars != 0) { //获取下一个可用的缓冲区，比如缓冲区是0~5， 有效数据起始buff是3，有效数据是4，那么存储buff数据依次为3,4,5,0，下一个可用的buff为1 ((3+4-1)/5) int lastbuf = (pipe->curbuf + pipe->nrbufs - 1) & (PIPE_BUFFERS-1); //获取下一个可用的buff，lastbuf为获取到的索引 struct pipe_buffer *buf = pipe->bufs + lastbuf;  const struct pipe_buf_operations *ops = buf->ops; /*offset: page中数据的偏移量。len：page中数据的长度 目前数据在page中的有效起始地址 + 有效数据长度 = 下一个可存放数据的地址。 管道是从前往后读的，并没规定读写大小，有可能只读取了page的前一部分，中间部分尚未读取，但是写的时候必须从中间有效数据后继续写 */ int offset = buf->offset + buf->len; //当前需要写入的数据 + 已有的数据若没有超过PAGE_SIZE大小，则拷贝到page中 if (ops->can_merge && offset + chars <= PAGE_SIZE) { ... redo1: addr = ops->map(pipe, buf, atomic); //将用户数据拷贝到page中 error = pipe_iov_copy_from_user(offset + addr, iov, chars, atomic); ops->unmap(pipe, buf, addr); ret = error; do_wakeup = 1; if (error) { ... } //更新有效数据 buf->len += chars; total_len -= chars; ret = chars; //全拷贝完则跳出 if (!total_len) goto out; } }  for (;;) { int bufs;  if (!pipe->readers) { ... } /*获取管道还有多少有效的buffer缓冲区*/ bufs = pipe->nrbufs; if (bufs < PIPE_BUFFERS) { //有效的bufs小于缓冲区总数 int newbuf = (pipe->curbuf + bufs) & (PIPE_BUFFERS-1); //获取下一个可用的buf struct pipe_buffer *buf = pipe->bufs + newbuf; struct page *page = pipe->tmp_page;  if (!page) { page = alloc_page(GFP_HIGHUSER); if (unlikely(!page)) { ... } pipe->tmp_page = page; }  do_wakeup = 1; chars = PAGE_SIZE; if (chars > total_len) chars = total_len;  iov_fault_in_pages_read(iov, chars); redo2:  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;  total_len -= chars; //数据写完，跳出循环结束 if (!total_len) break; } //还有可用的缓冲区，继续写 if (bufs < PIPE_BUFFERS) continue; //缓冲区全部写完了，则判断是否需要阻塞休眠 if (filp->f_flags & O_NONBLOCK) { if (!ret) ret = -EAGAIN; break; }  if (signal_pending(current)) { if (!ret) ret = -ERESTARTSYS; break; } //唤醒读进程 if (do_wakeup) { wake_up_interruptible_sync(&pipe->wait); kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN); do_wakeup = 0; } //写进程休眠 pipe->waiting_writers++; pipe_wait(pipe);//将当前任务加入到等待列表,释放锁，让出CPU pipe->waiting_writers--; } out: ...  return ret; }
　　管道写操作流程如下：写之前先锁定内存。获取可写的缓冲区，若可写则循环写。若不可写，则唤醒读进程，同时写进程进行休眠，让出 CPU 。
　　往管道中写数据的过程如下
　　管道中最重要的2个方法就是管道的读写。从上述的分析来看，读写进程共同操作内核中的数据缓冲区，若有缓冲区可写，则进程往缓冲区中写，若条件不允许写，则进程休眠让出 CPU 。读操作同理。
　　从上述管道读写操作可知，父子进程之所以能够通过 pipe  进行通信，是因为在内核中共同指向了同一个pipe_inode_info  对象，共同操作同一个内存页。
　　总结
　　管道也称无名管道，是 UNIX  系统中进程间通信(IPC )中的一种。管道由于是无名管道，因此只能在有亲缘关系的进程间使用。管道不是普通的文件，它是基于内存的。管道属于半双工，数据只能从一方流向另一方，也即数据只能从一端写，从另一端读。管道中读数据是一次性的操作，数据读取后就会释放空间，让出空间供更多的数据写。管道写数据遵循先入先出的原则。