/** *  struct pipe_inode_info - a linux kernel pipe *  @mutex: mutex protecting the whole thing *  @rd_wait: reader wait point in case of empty pipe *  @wr_wait: writer wait point in case of full pipe *  @head: The point of buffer production *  @tail: The point of buffer consumption *  @note_loss: The next read() should insert a data-lost message *  @max_usage: The maximum number of slots that may be used in the ring *  @ring_size: total number of buffers (should be a power of 2) *  @nr_accounted: The amount this pipe accounts for in user->pipe_bufs *  @tmp_page: cached released page *  @readers: number of current readers of this pipe *  @writers: number of current writers of this pipe *  @files: number of struct file referring this pipe (protected by ->i_lock) *  @r_counter: reader counter *  @w_counter: writer counter *  @fasync_readers: reader side fasync *  @fasync_writers: writer side fasync *  @bufs: the circular array of pipe buffers *  @user: the user who created this pipe *  @watch_queue: If this pipe is a watch_queue, this is the stuff for that **/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};

/** *  struct pipe_buffer - a linux kernel pipe buffer *  @page: the page containing the data for the pipe buffer *  @offset: offset of data inside the @page *  @len: length of data inside the @page *  @ops: operations associated with this buffer. See @pipe_buf_operations. *  @flags: pipe buffer flags. See above. *  @private: private data owned by the ops. **/struct pipe_buffer {    struct page *page;    unsigned int offset, len;    const struct pipe_buf_operations *ops;    unsigned int flags;    unsigned long private;};

static ssize_tpipe_write(struct kiocb *iocb, struct iov_iter *from){    struct file *filp = iocb->ki_filp;    struct pipe_inode_info *pipe = filp->private_data;    unsigned int head;    ssize_t ret = 0;    size_t total_len = iov_iter_count(from);    ssize_t chars;    bool was_empty = false;    bool wake_next_writer = false;
    /* Null write succeeds. */    if (unlikely(total_len == 0))        return 0;
    __pipe_lock(pipe);
    // 确保读者数量不为0    if (!pipe->readers) {        send_sig(SIGPIPE, current, 0);        ret = -EPIPE;        goto out;    }
#ifdef CONFIG_WATCH_QUEUE    if (pipe->watch_queue) {        ret = -EXDEV;        goto out;    }#endif
    /*     * Only wake up if the pipe started out empty, since     * otherwise there should be no readers waiting.     *     * If it wasn't empty we try to merge new data into     * the last buffer.     *     * That naturally merges small writes, but it also     * page-aligs the rest of the writes for large writes     * spanning multiple pages.     */    head = pipe->head;    was_empty = pipe_empty(head, pipe->tail);    chars = total_len & (PAGE_SIZE-1); // 要写入的数据的大小相对页帧大小的余数    // 如果余数不为0，且pipe不为空    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;
        // 如果PIPE_BUF_FLAG_CAN_MERGE被置位，且buf能容下chars大小的数据        if ((buf->flags & PIPE_BUF_FLAG_CAN_MERGE) &&            offset + chars <= PAGE_SIZE) {            ret = pipe_buf_confirm(pipe, buf);            if (ret)                goto out;
            // 将chars大小的数据写入缓冲区            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 (;;) {        // 确保对着数量不为0        if (!pipe->readers) {            send_sig(SIGPIPE, current, 0);            if (!ret)                ret = -EPIPE;            break;        }
        head = pipe->head;        // 如果pipe没被填满        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; // tmp_page用来临时存数据            int copied;
            // 如果tmp_page还未分配，则用alloc_page分配一个page并赋值            if (!page) {                page = alloc_page(GFP_HIGHUSER | __GFP_ACCOUNT);                if (unlikely(!page)) {                    ret = ret ? : -ENOMEM;                    break;                }                pipe->tmp_page = page;            }
            /* Allocate a slot in the ring in advance and attach an             * empty buffer.  If we fault or otherwise fail to use             * it, either the reader will consume it or it'll still             * be there for the next write.             */            // 自旋锁锁住读者等待队列            spin_lock_irq(&pipe->rd_wait.lock);
            head = pipe->head;            // 如果pipe已经被填满则进入下一次循环            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; //将之前分配的tmp_page赋值给buf->page            buf->ops = &anon_pipe_buf_ops;            buf->offset = 0;            buf->len = 0;            // 如果创建pipe时没有指定O_DIRECT选项，则将flags设置为PIPE_BUF_FLAG_CAN_MERGE            // 所以只要创建pipe时不指定flags，就能将buffer的PIPE_BUF_FLAG_CAN_MERGE置位            if (is_packetized(filp))                buf->flags = PIPE_BUF_FLAG_PACKET;            else                buf->flags = PIPE_BUF_FLAG_CAN_MERGE;            pipe->tmp_page = NULL; // tmp_page置空
            // 拷贝一页大小的数据到page里            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;        }        ......}

/* * Determine where to splice to/from. */long do_splice(struct file *in, loff_t __user *off_in,        struct file *out, loff_t __user *off_out,        size_t len, unsigned int flags){    struct pipe_inode_info *ipipe;    struct pipe_inode_info *opipe;    loff_t offset;    long ret;
    if (unlikely(!(in->f_mode & FMODE_READ) ||             !(out->f_mode & FMODE_WRITE)))        return -EBADF;
    ipipe = get_pipe_info(in, true);    opipe = get_pipe_info(out, true);
    // in和out都是pipe    if (ipipe && opipe) {        ......    }
    // 只有in是pipe    if (ipipe) {        ......    }    // 只有out是pipe    if (opipe) {        // 处理in和out的偏移        if (off_out)            return -ESPIPE;        if (off_in) {            if (!(in->f_mode & FMODE_PREAD))                return -EINVAL;            if (copy_from_user(&offset, off_in, sizeof(loff_t)))                return -EFAULT;        } else {            offset = in->f_pos;        }
        if (out->f_flags & O_NONBLOCK)            flags |= SPLICE_F_NONBLOCK;
        pipe_lock(opipe);        // 等待pipe有可用的缓冲区        ret = wait_for_space(opipe, flags);        if (!ret) {            unsigned int p_space;
            /* Don't try to read more the pipe has space for. */            p_space = opipe->max_usage - pipe_occupancy(opipe->head, opipe->tail);// pipe可用空间            len = min_t(size_t, len, p_space << PAGE_SHIFT);// 实际读取长度不能超过pipe可用空间
            ret = do_splice_to(in, &offset, opipe, len, flags); // 调用do_splice_to完成主要工作        }        pipe_unlock(opipe);        if (ret > 0)            wakeup_pipe_readers(opipe);        if (!off_in)            in->f_pos = offset;        else if (copy_to_user(off_in, &offset, sizeof(loff_t)))            ret = -EFAULT;
        return ret;    }
    return -EINVAL;}

tatic size_t copy_page_to_iter_pipe(struct page *page, size_t offset, size_t bytes,             struct iov_iter *i){    struct pipe_inode_info *pipe = i->pipe;    struct pipe_buffer *buf;    unsigned int p_tail = pipe->tail;    unsigned int p_mask = pipe->ring_size - 1;    unsigned int i_head = i->head;    size_t off;
    if (unlikely(bytes > i->count))        bytes = i->count;
    if (unlikely(!bytes))        return 0;
    if (!sanity(i))        return 0;
    off = i->iov_offset;    buf = &pipe->bufs[i_head & p_mask];    if (off) {        // 如果要求的offset和实际的offset相同，且头部的buffer指向的就是当前的page cache        // 则直接移动offset即可        if (offset == off && buf->page == page) {            /* merge with the last one */            buf->len += bytes;            i->iov_offset += bytes;            goto out;        }        i_head++;        buf = &pipe->bufs[i_head & p_mask];    }    if (pipe_full(i_head, p_tail, pipe->max_usage))        return 0;
    buf->ops = &page_cache_pipe_buf_ops;    // 增加page的应用计数    get_page(page);    // 将pipe缓冲区的page指针指向文件的page cache    buf->page = page;    buf->offset = offset;    buf->len = bytes;
    pipe->head = i_head + 1;    i->iov_offset = offset + bytes;    i->head = i_head;out:    i->count -= bytes;    return bytes;}