Rev.2を表示中。最新版はこちら。

書込みpipeの vmspliceは引数データをpipeバッファに複写するのでなく、struct iovec.iov_base引数データのpageをpipeバッファpageとし、データのpage内のオフセット/サイズをpipeバッファの属性のoffset/lenに設定。故にvmsplice後に設定したユーザバッファデータ変更すれば、pipeバッファを更新した事になる。

pipeのwriteは、書込データがpage領域を超えるとpipe->nrbufs++とし次書込みは次バッファの別ページで、page領域を超えなければ同pageに追加され、pipe_buffernのlenに追加サイズが加算されるが、vmspliceはデータがpageサイズを超えなくても設定毎にpipe->nrbufs++。従ってvmspliceは設定バッファに関係なくデフォルトでPIPE_DEF_BUFFERSの16回しか実行できない。小さいサイズバッファを多々扱う場合、配列によるバッファを配列[0]のアドレスをvmspliceすべきである。

読込pipeのvmspliceは、書込みpipeの係る実装でなく、pipeのreadそのものである。そもそも読/書pipeバッファは共有、読込pipeバッファを設定できるとすると、書込pipeバッファにも設定した事になり、読込pipeなのに書込みしたことになってしまう。

サンプル

[root@north a]# cat w-vmsplice.c

#define _GNU_SOURCE

#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <sys/syscall.h>

void    pread1(int pfd);
void    pread2(int pfd);
void    pbuff_read(void);

char    buf1[20], buf2[20], buf3[20];
void main(int argc, char *argv[])
{
    int     cnt, fd, pw, pr, pfd[2];
    struct iovec     iov[3];

    pipe2(pfd, O_NONBLOCK);
    pr = pfd[0]; pw = pfd[1];

    strcpy(buf1, "0123456789");
    strcpy(buf2, "0123456789");
    strcpy(buf3, "0123456789");

    iov[0].iov_base = buf1; iov[0].iov_len  = 5;
    iov[1].iov_base = buf2; iov[1].iov_len  = 8;
    iov[2].iov_base = buf3; iov[2].iov_len  = 10;

    if (argc == 1) { 
        cnt = syscall(SYS_vmsplice, pw, iov, 3, O_NONBLOCK);
        printf("vmspliced  :%d\n", cnt);

        buf1[0] = 'a';  buf2[0] = 'b';  buf3[0] = 'c';          <- pipeバッファに設定されたbuf1/buf2/buf3を更新
        pread1(pr);
    }
    else {
        cnt = 0;
        while (1) {
            if (syscall(SYS_vmsplice, pw, &iov[0], 1, SPLICE_F_NONBLOCK) <= 0) {
                break;
               }
            cnt++;
            }
         printf("vmspliced cnt:%d [PIPE_DEF_BUFFERS=16]\n", cnt);
    }
}

void     pread1(int pfd)
{
    char     buf[32];

    printf("pipe read:");
    while(1) {
         buf[1] = 0;
         if (read(pfd, buf, 1) > 0) {
              printf("%s", buf);
            }
         else {
              break;
         }
    }
    printf("\n");
}

[root@north a]# ./w-vmsplice.out
vmsplice :23
pipe read:a1234b1234567c123456789

[root@north a]# ./w-vmsplice.out max-do_vmsplice
vmspliced cnt:16 [PIPE_DEF_BUFFERS=16]

カーネル

pipeはpageをバッファとするstruct pipe_bufferを16有するデフォルト領域で、pipeのwriteは、pageサイズを超て、次のstruct pipe_bufferへ書き込まれるが、vmspliceは同じページであっても設定毎に新規struct pipe_bufferへ設定される。

#define PIPE_DEF_BUFFERS        16
struct inode {
 :
       union {
               struct pipe_inode_info  *i_pipe;
               struct block_device     *i_bdev;
               struct cdev             *i_cdev;
       };
 :
};

struct pipe_inode_info {
       wait_queue_head_t wait;
       unsigned int nrbufs, curbuf, buffers;
       unsigned int readers;
       unsigned int writers;
       unsigned int waiting_writers;
       unsigned int r_counter;
       unsigned int w_counter;
       struct page *tmp_page;
       struct fasync_struct *fasync_readers;
       struct fasync_struct *fasync_writers;
       struct inode *inode;
       struct pipe_buffer *bufs;               <-- bufs[PIPE_DEF_BUFFERS]
};

struct pipe_buffer {
       struct page *page;
       unsigned int offset, len;
       const struct pipe_buf_operations *ops;
       unsigned int flags;
       unsigned long private;
};
読込page(bufs->page) bufs = inode->i_pipe->bufs + inode->i_pipe->curbuf
書込page(bufs->page) bufs = inode->i_pipe->bufs + inode->i_pipe->curbuf + inode->i_pipe->nrbufs

pipe作成のデフォルトメモリサイズはpage*PIPE_DEF_BUFFERS(4096*16)
struct pipe_inode_info * alloc_pipe_info(struct inode *inode)
{
      struct pipe_inode_info *pipe;

      pipe = kzalloc(sizeof(struct pipe_inode_info), GFP_KERNEL);
      if (pipe) {
              pipe->bufs = kzalloc(sizeof(struct pipe_buffer) * PIPE_DEF_BUFFERS, GFP_KERNEL);
              if (pipe->bufs) {
                      init_waitqueue_head(&pipe->wait);
                      pipe->r_counter = pipe->w_counter = 1;
                      pipe->inode = inode;
                      pipe->buffers = PIPE_DEF_BUFFERS;
                      return pipe;
              }
              kfree(pipe);
      }
      return NULL;
}

SYSCALL_DEFINE4(vmsplice, int, fd, const struct iovec __user *, iov,
               unsigned long, nr_segs, unsigned int, flags)
{
       struct fd f;
       long error;

       if (unlikely(nr_segs > UIO_MAXIOV))
               return -EINVAL;
       else if (unlikely(!nr_segs))
               return 0;

       error = -EBADF;
       f = fdget(fd);
       if (f.file) {
               if (f.file->f_mode & FMODE_WRITE)
                       error = vmsplice_to_pipe(f.file, iov, nr_segs, flags);
               else if (f.file->f_mode & FMODE_READ)
                       error = vmsplice_to_user(f.file, iov, nr_segs, flags);

               fdput(f);
       }

       return error;
}

static long vmsplice_to_pipe(struct file *file, const struct iovec __user *iov,
                            unsigned long nr_segs, unsigned int flags)
{
       struct pipe_inode_info *pipe;
       struct page *pages[PIPE_DEF_BUFFERS];
       struct partial_page partial[PIPE_DEF_BUFFERS];
       struct splice_pipe_desc spd = {
               .pages = pages,
               .partial = partial,
               .nr_pages_max = PIPE_DEF_BUFFERS,
               .flags = flags,
               .ops = &user_page_pipe_buf_ops,
               .spd_release = spd_release_page,
       };
       long ret;

       pipe = get_pipe_info(file);
       if (!pipe)
               return -EBADF;

       if (splice_grow_spd(pipe, &spd))
               return -ENOMEM;

       spd.nr_pages = get_iovec_page_array(iov, nr_segs, spd.pages,
                                           spd.partial, false,
                                           spd.nr_pages_max);
       if (spd.nr_pages <= 0)
               ret = spd.nr_pages;
       else
               ret = splice_to_pipe(pipe, &spd);

       splice_shrink_spd(&spd);
       return ret;
}

get_iovec_page_array()でユーザ引数のiovec __user *iovのデータのpageと係るoffset/len等の属性が*spdに設定される。
ssize_t splice_to_pipe(struct pipe_inode_info *pipe, struct splice_pipe_desc *spd)
{
    unsigned int spd_pages = spd->nr_pages;
    int ret, do_wakeup, page_nr;

    ret = 0;
    do_wakeup = 0;
    page_nr = 0;

    pipe_lock(pipe);

    for (;;) {
        if (!pipe->readers) {
            send_sig(SIGPIPE, current, 0);
            if (!ret)
                ret = -EPIPE;
                break;
          }

        if (pipe->nrbufs < pipe->buffers) {
              int newbuf = (pipe->curbuf + pipe->nrbufs) & (pipe->buffers - 1);
              struct pipe_buffer *buf = pipe->bufs + newbuf;

              buf->page = spd->pages[page_nr];             <-設定バッファのページ（page_nrはiov[]の配列数に係る引数)
              buf->offset = spd->partial[page_nr].offset;  <-設定バッファのページ内のオフセット
              buf->len = spd->partial[page_nr].len;        <-設定バッファの長さ
              buf->private = spd->partial[page_nr].private;
              buf->ops = spd->ops;
              if (spd->flags & SPLICE_F_GIFT)
                  buf->flags |= PIPE_BUF_FLAG_GIFT;

              pipe->nrbufs++;       <-設定バッファがページサイズを超えなくても、次pipe書込みは新規pipeバッファ
              page_nr++;
              ret += buf->len;

              if (pipe->inode)
                   do_wakeup = 1;

              if (!--spd->nr_pages)
                   break;
              if (pipe->nrbufs < pipe->buffers)
                   continue;

              break;
          }

        if (spd->flags & SPLICE_F_NONBLOCK) {
            if (!ret)
                ret = -EAGAIN;
                break;
               }

            if (signal_pending(current)) {
                if (!ret)
                     ret = -ERESTARTSYS;
                break;
               }

            if (do_wakeup) {
                smp_mb();
                if (waitqueue_active(&pipe->wait))
                    wake_up_interruptible_sync(&pipe->wait);
                kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
                do_wakeup = 0;
          }

        pipe->waiting_writers++;
        pipe_wait(pipe);
        pipe->waiting_writers--;
     }

    pipe_unlock(pipe);

    if (do_wakeup)
        wakeup_pipe_readers(pipe);

    while (page_nr < spd_pages)
        spd->spd_release(spd, page_nr++);

    return ret;
}