一、pipe特点
每个进程各自有不同的用户地址空间，任何一个进程的全局变量在另一个进程中都看不到，所以进程之间要交换数据必须通过内核，内核中用于进程间通信的技术有很多，管道是其中的一种。
有两种类型的管道，一种是匿名管道，也是类Unix系统上传统的管道，用与将一个进程的输出作为另一个进程的输入。另一种则是命名管道（又叫FIFO），是对类Unix系统上匿名管道概念的扩展，并且是进程间通信（IPC）的方法之一
匿名管道仅在过程中持续使用，过程结束后管道会被释放。命名管道则可以在系统启动时一直运行到被删除为止，遵循先进先出机制，是一种的特殊文件，但可以像普通文件一样使用它。如果不再使用它，也可以将其删除。通常，命名管道以文件的形式出现，并且通常会附加用于IPC的进程。命名管道利用了filesystem，可用于将信息从一个应用程序传输到另一个应用程序，而无需使用中间临时文件。
fifo和pipe之间的关系：
- pipe是匿名管道，只能用于两个拥有pipe读写两端fd的进程通信；
- fifo在文件系统中有自己的名称，操作fifo与操作普通文件几无差别，可以用于两个没有关系的进程间通信
- fifo和pipe在kernel层面上都实现在fs/pipe.c中，本质上是一个东西。
- pipe作为linux文件系统的一部分，与epoll一样，都是在向kernel注册了自己的文件系统，可以使用VFS提供的通用接口，比如open、read和write等操作
- pipe的容量不是无限大的，早期linux版本中pipe容量只能是4KB大小，新版本可以在运行时动态扩大到64KB
实现机制：
管道是由内核管理的一个缓冲区，其一端连接一个进程的输出，这个进程会向管道中放入信息，而另一端连接一个进程的输入，这个进程取出被放入管道的信息。这个缓冲区不需要很大，因而被设计成为环形数据结构，以便管道可以被循环利用。当管道中没有信息的话，从管道中读取的进程会等待，直到另一端的进程放入信息。当管道被放满信息的时候，尝试放入信息的进程会等待，直到另一端的进程取出信息。当两个进程都终结的时候，管道也自动消失。
二、数据结构
pipe buffer在内存中保存pipe的缓存，其中page为真正内存对应的page结构，offset、len则为page的起始地址偏移和page的长度，即多少个page，而ops则是buffer操作的方法，private指向下面的管道描述符
/* 用来描述pipe所使用的page结构 */
struct pipe_buffer {
struct page *page;
unsigned int offset, len;
const struct pipe_buf_operations *ops;
unsigned int flags;
unsigned long private;
};在pipe_inode_info结构中有很多的限制方法来对pipe进行限制，例如mutex、counter、readers和writers等，都是为了限制对pipe所对应的临界区访问
/* 管道描述符，表示一个管道，存储管道相应的信息 */
struct pipe_inode_info {
struct mutex mutex;
wait_queue_head_t rd_wait, wr_wait;
unsigned int head;
unsigned int tail;
unsigned int max_usage;
unsigned int ring_size;
#ifdef CONFIG_WATCH_QUEUE
bool note_loss;
#endif
unsigned int nr_accounted;
unsigned int readers;
unsigned int writers;
unsigned int files;
unsigned int r_counter;
unsigned int w_counter;
struct page *tmp_page;
struct fasync_struct *fasync_readers;
struct fasync_struct *fasync_writers;
struct pipe_buffer *bufs;
struct user_struct *user;
#ifdef CONFIG_WATCH_QUEUE
struct watch_queue *watch_queue;
#endif
};
三、内核源码实现
首先在应用层使用pipe会通过pipe系统调用去调用pipefs中的方法，所以首先需要初始化pipefs。需要先创建一个file_system_type对象，然后在初始化的时候注册并挂载。
static struct file_system_type pipe_fs_type = {
.name = "pipefs",
.init_fs_context = pipefs_init_fs_context,
.kill_sb = kill_anon_super,
};
static int __init init_pipe_fs(void)
{
int err = register_filesystem(&pipe_fs_type); //注册
if (!err) {
pipe_mnt = kern_mount(&pipe_fs_type); //挂载
...
}
return err;
}注册完pipefs就可以使用其中的方法，定义在
static const struct super_operations pipefs_ops = {
.destroy_inode = free_inode_nonrcu,
.statfs = simple_statfs,
};其方法是释放inode和获取pipefs的状态信息，用于在pipe缓冲使用完毕需要进行回收时使用。pipe的调用都是通过系统调用进而进入内核空间，在内核中由do_pipe2这个函数去完成标志的检测和file描述符的分配，同时在分配file描述符的同时为pipe分配内存空间，pipe缓冲区的分配则是在create_pipe_files函数中，pipe实现在file system的inode、superblock、file基础结构之上，一个pipe就是一个特殊的文件，使用方法与其他文件一样，都需要open，之后去调用其读写方法
static int __do_pipe_flags(int *fd, struct file **files, int flags)
{
...
if (flags & ~(O_CLOEXEC | O_NONBLOCK | O_DIRECT | O_NOTIFICATION_PIPE))
return -EINVAL;
error = create_pipe_files(files, flags);
if (error)
return error;
...
}在__do_pipe_flags函数中会创建管道并返回读写端文件描述符，首先会检查传进来的flag，O_CLOEXEC、O_NONBLOCK、O_DIRECT、O_NOTIFICATION_PIPE四种标志都可以创建管道，分别代表发生clone、没有映射的block、直接磁盘传输以及无通知的pipe，进而去创建读写端的文件句柄
创建pipe的文件描述符会先通过superblock找到一个inode节点，同时这个inode对应的缓冲区不能发生回写，因为它只存在与内存，并将这个inode标记为dirty，所以说pipe其实是一个特殊的文件，然后分配一个pipe_inode_info，将其初始化并将地址保存在inode的i_pipe成员中
int create_pipe_files(struct file **res, int flags)
{
...
f = alloc_file_pseudo(inode, pipe_mnt, "",
O_WRONLY | (flags & (O_NONBLOCK | O_DIRECT)),
&pipefifo_fops);
f->private_data = inode->i_pipe;
res[0] = alloc_file_clone(f, O_RDONLY | (flags & O_NONBLOCK),
&pipefifo_fops);
res[0]->private_data = inode->i_pipe;
res[1] = f;
...
}alloc_file_pseudo函数则会将pipefifo_fops里面的方法与file结构和inode绑定，之后就能够通过open打开文件句柄，并使用其中的方法，file结构中的private_data也会被初始化成inode中的i_pipe，可以看到res[0]和res[1]都绑定了pipefifo_fops，只是res[0]的分配是从res[1]clone过来的，最后指定stream方式的打开，最后还需要检测一下flags，其中fd[0]被设置为读端口，fd[1]设置为写端口
3.2 读写
pipe的读写就是去操作缓冲区中的数据
读：
static ssize_t
pipe_read(struct kiocb *iocb, struct iov_iter *to)
{
...
/* 只有当pipe中的数据已满时并且reader开始读取数据才唤醒writer */
was_full = pipe_full(pipe->head, pipe->tail, pipe->max_usage);
for (;;) {
if (!pipe_empty(head, tail)) {
struct pipe_buffer *buf = &pipe->bufs[tail & mask];
/* 最大buffer限制，不能超过这个最大值 */
if (chars > total_len) {
chars = total_len;
}
/* 验证缓冲区中的内容，是buffer方法中的一种 */
error = pipe_buf_confirm(pipe, buf);
written = copy_page_to_iter(buf->page, buf->offset, chars, to);
/* 包缓冲区则退出 */
if (buf->flags & PIPE_BUF_FLAG_PACKET) {
total_len = chars;
buf->len = 0;
}
/* 长度为0则释放pipe buffer， */
if (!buf->len) {
pipe_buf_release(pipe, buf);
}
/* 总长度减掉 */
total_len -= chars;
if (!total_len)
break; /* 读取成功直接break */
if (!pipe_empty(head, tail)) /* 没读取完则再来一次 */
continue;
}
}
/* 结束，为pipe解锁 */
__pipe_unlock(pipe);
/* 仍然满，需要去挂起writer，并将writer的同步模式清除 */
if (unlikely(was_full)) {
wake_up_interruptible_sync_poll(&pipe->wr_wait, EPOLLOUT | EPOLLWRNORM);
kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
}
__pipe_lock(pipe);
/* 为下一次判断pipe buffer是否为满 */
was_full = pipe_full(pipe->head, pipe->tail, pipe->max_usage);
/* 下一次读取标志 */
wake_next_reader = true;
}
/* pipe buffer已经为空，不需要再去读 */
if (pipe_empty(pipe->head, pipe->tail))
wake_next_reader = false;
__pipe_unlock(pipe);
/* 到这里不论是读还是写都需要挂起 */
if (was_full) {
wake_up_interruptible_sync_poll(&pipe->wr_wait, EPOLLOUT | EPOLLWRNORM);
kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
}
if (wake_next_reader)
wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM);
...
}读过程就是在一个循环中，将缓冲区中数据通过copy_page_to_iter函数写到用户态空间缓冲区中，需要判断读完成标志和为full状态
static ssize_t
pipe_write(struct kiocb *iocb, struct iov_iter *from)
{
...
was_empty = pipe_empty(head, pipe->tail);
chars = total_len & (PAGE_SIZE-1);
if (chars && !was_empty) {
unsigned int mask = pipe->ring_size - 1;
struct pipe_buffer *buf = &pipe->bufs[(head - 1) & mask];
int offset = buf->offset + buf->len;
if ((buf->flags & PIPE_BUF_FLAG_CAN_MERGE) &&
offset + chars <= PAGE_SIZE) {
ret = pipe_buf_confirm(pipe, buf);
if (ret)
goto out;
ret = copy_page_from_iter(buf->page, offset, chars, from);
if (unlikely(ret < chars)) {
ret = -EFAULT;
goto out;
}
buf->len += ret;
if (!iov_iter_count(from))
goto out;
}
}
for (;;) {
head = pipe->head;
if (!pipe_full(head, pipe->tail, pipe->max_usage)) {
unsigned int mask = pipe->ring_size - 1;
struct pipe_buffer *buf = &pipe->bufs[head & mask];
struct page *page = pipe->tmp_page;
int copied;
if (!page) {
page = alloc_page(GFP_HIGHUSER | __GFP_ACCOUNT);
if (unlikely(!page)) {
ret = ret ? : -ENOMEM;
break;
}
pipe->tmp_page = page;
}
spin_lock_irq(&pipe->rd_wait.lock);
head = pipe->head;
if (pipe_full(head, pipe->tail, pipe->max_usage)) {
spin_unlock_irq(&pipe->rd_wait.lock);
continue;
}
pipe->head = head + 1;
spin_unlock_irq(&pipe->rd_wait.lock);
/* Insert it into the buffer array */
buf = &pipe->bufs[head & mask];
buf->page = page;
buf->ops = &anon_pipe_buf_ops;
buf->offset = 0;
buf->len = 0;
if (is_packetized(filp))
buf->flags = PIPE_BUF_FLAG_PACKET;
else
buf->flags = PIPE_BUF_FLAG_CAN_MERGE;
pipe->tmp_page = NULL;
copied = copy_page_from_iter(page, 0, PAGE_SIZE, from);
if (unlikely(copied < PAGE_SIZE && iov_iter_count(from))) {
if (!ret)
ret = -EFAULT;
break;
}
ret += copied;
buf->offset = 0;
buf->len = copied;
if (!iov_iter_count(from))
break;
}
if (!pipe_full(head, pipe->tail, pipe->max_usage))
continue;
/* Wait for buffer space to become available. */
if (filp->f_flags & O_NONBLOCK) {
if (!ret)
ret = -EAGAIN;
break;
}
if (signal_pending(current)) {
if (!ret)
ret = -ERESTARTSYS;
break;
}
__pipe_unlock(pipe);
if (was_empty) {
wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM);
kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
}
wait_event_interruptible_exclusive(pipe->wr_wait, pipe_writable(pipe));
__pipe_lock(pipe);
was_empty = pipe_empty(pipe->head, pipe->tail);
wake_next_writer = true;
}
out:
if (pipe_full(pipe->head, pipe->tail, pipe->max_usage))
wake_next_writer = false;
__pipe_unlock(pipe);
if (was_empty) {
wake_up_interruptible_sync_poll(&pipe->rd_wait, EPOLLIN | EPOLLRDNORM);
kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
}
if (wake_next_writer)
wake_up_interruptible_sync_poll(&pipe->wr_wait, EPOLLOUT | EPOLLWRNORM);
if (ret > 0 && sb_start_write_trylock(file_inode(filp)->i_sb)) {
int err = file_update_time(filp);
if (err)
ret = err;
sb_end_write(file_inode(filp)->i_sb);
}
return ret;
}写缓冲区则是先分配page，属于虚拟内存空间，然后将数据写入到这块内存