本文转载自微信公众号《Linux内核那些事儿》，作者songsong001。转载本文请联系Linux内核那些事儿公众号。为了安全起见，不同进程的内存空间是相互隔离的，也就是说进程A不能访问进程B的内存空间，反之亦然。如果不同的进程可以互相访问和修改对方的内存，则当前进程的内存可能会被其他进程非法修改，造成安全隐患。不同的进程就像大海中的孤岛，它们之间不能直接通信，如下图：但是在某些场景下，不同的进程之间需要通信，例如：进程A负责处理用户请求，而进程AB负责保存处理后的数据。那么当A进程处理完请求后，需要将处理后的数据提交给B进程存储。此时，进程A需要与进程B进行通信，如下图所示：由于不同进程之间是相互隔离的，因此必须借助内核作为相互通信的桥梁。内核相当于岛屿之间的一艘船，如下图所示：内核提供了多种进程间通信方式。如：共享内存、信号、消息队列和管道（pipe）等。本文主要介绍管道的原理和实现。1.管道的使用管道一般用于父子进程之间的相互通信。一般用法如下：父进程使用管道系统调用创建管道。然后父进程使用fork系统调用创建一个子进程。由于子进程继承了父进程打开的文件句柄，父子进程可以通过新建的管道进行通信。原理如下图所示：由于管道分为读端和写端，所以需要两个文件描述符来管理管道：fd[0]是读端，fd[1]是写端结尾。下面的代码描述了如何使用管道系统调用来创建管道：#include#include#include#include#includeintmain(){intret=-1;intfd[2];//用于管理管道的文件描述符pid_tpid;charbuf[512]={0};char*msg="helloworld";//创建一个managementret=pipe(fd);if(-1==ret){printf("failedtocreatepipe\n");return-1;}pid=fork();//创建子进程if(0==pid){//Subprocessclose(fd[0]);//关闭管道的读端ret=write(fd[1],msg,strlen(msg));//向管道的写端写入数据exit(0);}else{//父进程close(fd[1]);//关闭管道的写端ret=read(fd[0],buf,sizeof(buf));//从管道读取数据readendofthepipeprintf("parentread%dbytesdata:%s\n",ret,buf);}return0;}编译代码：[root@localhostpipe]#gcc-gpipe.c-opipe运行代码，输出如下：[root@localhostpipe]#./pipeparentread11bytesdata:helloworld二、执行pipeline的用户空间每个进程都是独立的，但内核空间是共享的。因此，进程间通信必须由内核提供服务。前面介绍了管道（pipe）的使用，接下来介绍管道在内核中的实现。本文以Linux-2.6.23内核为分析对象。1.RingBuffer在内核中，pipeline使用一个ringbuffer来存储数据。环形缓冲区的原理是：把一个缓冲区看成一个端到端的环，通过读指针和写指针记录读操作和写操作的位置。如下图所示：在Linux内核中，16个内存页被用作一个环形缓冲区，所以这个环形缓冲区的大小为64KB（16*4KB）。向管道写入数据时，从写指针指向的位置开始写，将写指针向前移动。从管道读取数据时，从读指针开始读取，将读指针前移。当对没有数据可读的管道进行读操作时，当前进程将被阻塞。写入没有可用空间的管道也会阻塞当前进程。注意：管道文件描述符可以设置为非阻塞的，这样当前进程对管道进行读写操作就不会被阻塞。2.管道对象在Linux内核中，管道是使用pipe_inode_info对象来管理的。我们看一下pipe_inode_info对象的定义，如下：structpipe_inode_info{wait_queue_head_twait;unsignedintnrbufs,unsignedintcurbuf;...unsignedintreaders;unsignedintwriters;unsignedintwaiting_writers;...structinode*inode;node_bufferbufs[16];各字段的作用：wait：等待队列，用于存放等待管道可读或可写的进程。bufs：环形缓冲区，由16个pipe_buffer对象组成，每个pipe_buffer对象有一个内存页，后面会介绍。nrbufs：表示ringbuffer中有多少内存页已经被未读数据占用。curbuf：表示当前正在读取数据的是ringbuffer的哪个内存页。readers：表示正在读取管道的进程数。writers：表示正在写入管道的进程数。waiting_writers：表示等待管道可写的进程数。inode：与管道关联的inode对象。由于ringbuffer由16个pipe_buffer对象组成，我们先看一下pipe_buffer对象的定义：structpipe_buffer{structpage*page;unsignedintoffset;unsignedintlen;...};下面介绍一下pipe_buffer对象的各个字段的作用：page：指向pipe_buffer对象占用的内存页。offset：如果进程正在读取当前内存页的数据，那么offset指向当前正在读取的内存页的偏移量。len：表示当前内存页中未读数据的长度。下图显示了pipe_inode_info对象和pipe_buffer对象的关系：pipeline的ringbuffer的实现与经典ringbuffer的实现略有不同。经典的环形缓冲区一般都是先申请一块地址连续的内存块，然后通过读指针和写指针来定位读写操作。但是为了减少内存的使用，内核在创建管道时并没有申请64K的内存块，而是在进程向管道写入数据时按需申请内存。那么当进程从管道中读取数据时，内核是如何处理的呢？下面看一下管道读操作的实现。3、读操作从经典环形缓冲区读取数据时，首先通过读指针定位到读取数据的起始地址，然后判断环形缓冲区中是否有要读取的数据，如果有，则从ringbuffer将数据读入用户空间的缓冲区。如下图所示：管道的环形缓冲区与经典的环形缓冲区略有不同。管道的环形缓冲区的读指针由pipe_inode_info对象的curbuf字段和pipe_buffer对象的offset字段组成：pipe_inode_info对象的curbuf字段表示read操作将从bufs数组中的哪个pipe_buffer读取数据。pipe_buffer对象的offset字段表示read操作从内存页开始读取数据的位置。读取数据的过程如下图所示：从buffer中读取n个字节的数据后，读指针的位置会相应移动n个字节（即pipe_buffer对象的offset字段会增加),n个字节将被减少。以字节为单位的可读数据长度（即减少pipe_buffer对象的len字段）。当pipe_buffer对象的len字段变为0时，表示当前pipe_buffer没有可读数据，则pipe_inode_info对象的curbuf字段将移动一位，其nrbufs字段将减1。下面看一下管道读操作的代码实现，由pipe_read函数完成。为了突出重点，我们只列出关键代码，如下：staticssize_tpipe_read(structkiocb*iocb,conststructiovec*_iov,unsignedlongnr_segs,loff_tpos){...structpipe_inode_info*pipe;//1.获取管道对象pipe=inode->i_pipe;for(;;){//2。获取pipeline未读数据占用多少内存页Readdataintcurbuf=pipe->curbuf;structpipe_buffer*buf=pipe->bufs+curbuf;.../*4.通过pipe_buffer的offset字段获取真正的读指针，*从pipe中读取数据到userbufferArea。*/error=pipe_iov_copy_to_user(iov,addr+buf->offset,chars,atomic);...ret+=chars;buf->offset+=chars;//增加pipe_buffer对象的offset字段的值buf->len-=chars;//减少pipe_buffer对象的len字段的值/*5.如果当前内存页的数据已经被读取*/if(!buf->len){...curbuf=(curbuf+1)&(PIPE_BUFFERS-1);pipe->curbuf=curbuf;//移动curbufpointerofthepipe_inode_infoobjectpipe->nrbufs=--bufs;//减少pipe_inode_info对象的nrbufs字段do_wakeup=1;}total_len-=chars;//6.如果读取到用户期望的数据长度，则退出循环if(!total_len)break;}...}...returnret;}以上代码可以归纳为以下步骤：管道的对象。通过pipe_inode_info对象的nrbufs字段获取管道未读数据占用的内存页数。通过pipe_inode_info对象的curbuf字段获取read操作应该从ringbuffer的哪个内存页读取数据。通过pipe_buffer对象的offset字段获取真正的读指针，从管道中读取数据到用户缓冲区。如果当前内存页的数据已被读取，则移动pipe_inode_info对象的curbuf指针，减少其nrbufs字段的值。如果读取到用户期望的数据长度，则退出循环。4.写操作分析完流水线读操作的实现，接下来，我们分析流水线写操作的实现。在经典的ringbuffer中写入数据时，首先通过写指针定位到要写入的内存地址，然后判断ringbuffer的空间是否足够，然后将数据写入ringbuffer。如下图所示：但是pipeline的ringbuffer并没有保存writepointer，而是通过readpointer计算出来的。那么如何通过读指针计算出写指针呢？其实很简单，就是：写指针=读指针+未读数据长度下面我们看一下向管道写入200字节数据的过程示意图，如下图：如上图所示如图，向管道写入数据时：首先通过pipe_inode_info的curbuf字段和nrbufs字段定位应该写入哪个pipe_buffer。然后通过pipe_buffer对象的offset域和len域定位内存页应该写入的位置。我们通过源码来分析一下，write操作是如何实现的，代码如下（为了突出重点，代码已经删掉）：...structpipe_inode_info*pipe;...pipe=inode->i_pipe;...chars=total_len&(PAGE_SIZE-1);/*sizeofthelastbuffer*///1。如果最后写入的pipe_buffer还有空闲空间if(pipe->nrbufs&&chars!=0){//获取写入数据的位置intlastbuf=(pipe->curbuf+pipe->nrbufs-1)&(PIPE_BUFFERS-1);structpipe_buffer*buf=pipe->bufs+lastbuf;conststructpipe_buf_operations*ops=buf->ops;intoffset=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;//如果所有要写入的数据都写入成功，则退出循环if(!total_len)gotoout;}}//2.如果最后写入的pipe_buffer中没有足够的空闲空间，则申请一个新的内存页来存放数据for(;;){intbufs;...bufs=pipe->nrbufs;if(bufscurbuf+bufs)&(PIPE_BUFFERS-1);structpipe_buffer*buf=pipe->bufs+newbuf;...//申请一个新的内存页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;//如果所有要写入的数据都写入成功,退出循环total_len-=chars;if(!total_len)break;}...}out:...returnret;}上面的代码有点长，但是逻辑很简单。主要操作如下：如果上次的写操作write如果导入的pipe_buffer仍有空闲空间，则将数据写入这个pipe_buffer并增加其len字段的值。如果上次写操作写入的pipe_buffer没有足够的空闲空间，则申请新的内存页。并将数据保存到一个新的内存页，并增加pipe_inode_info的nrbufs字段的值。如果写入的数据全部写入成功，则退出写入操作。3、想想管道读写操作的实现。分析已经完成。现在让我们思考以下问题。1、为什么父子进程可以通过管道进行通信？这是因为父子进程通过管道系统调用打开的管道指向内核空间中同一个管道对象（pipe_inode_info）。所以父子进程共享同一个管道对象，那么它们就可以通过这个共享的管道对象进行通信。2、为什么内核要用16个内存页来存储数据？这是为了减少内存使用。因为当使用管道系统调用打开管道时，并不是立即请求内存页，而是当进程向管道写入数据时，才按需请求内存页。当内存页的数据被读取时，内核会回收内存页以减少流水线的内存占用。
