splice.c

Go to the documentation of this file.

/*
 * "splice": joining two ropes together by interweaving their strands.
 *
 * This is the "extended pipe" functionality, where a pipe is used as
 * an arbitrary in-memory buffer. Think of a pipe as a small kernel
 * buffer that you can use to transfer data from one end to the other.
 *
 * The traditional unix read/write is extended with a "splice()" operation
 * that transfers data buffers to or from a pipe buffer.
 *
 * Named by Larry McVoy, original implementation from Linus, extended by
 * Jens to support splicing to files, network, direct splicing, etc and
 * fixing lots of bugs.
 *
 * Copyright (C) 2005-2006 Jens Axboe <[email protected]>
 * Copyright (C) 2005-2006 Linus Torvalds <[email protected]>
 * Copyright (C) 2006 Ingo Molnar <[email protected]>
 *
 */
#include <linux/fs.h>
#include <linux/file.h>
#include <linux/pagemap.h>
#include <linux/splice.h>
#include <linux/memcontrol.h>
#include <linux/mm_inline.h>
#include <linux/swap.h>
#include <linux/writeback.h>
#include <linux/export.h>
#include <linux/syscalls.h>
#include <linux/uio.h>
#include <linux/security.h>
#include <linux/gfp.h>
#include <linux/socket.h>

/*
 * Attempt to steal a page from a pipe buffer. This should perhaps go into
 * a vm helper function, it's already simplified quite a bit by the
 * addition of remove_mapping(). If success is returned, the caller may
 * attempt to reuse this page for another destination.
 */
static int page_cache_pipe_buf_steal(struct pipe_inode_info *pipe,
                     struct pipe_buffer *buf)
{
    struct page *page = buf->page;
    struct address_space *mapping;

    lock_page(page);

    mapping = page_mapping(page);
    if (mapping) {
        WARN_ON(!PageUptodate(page));

        /*
         * At least for ext2 with nobh option, we need to wait on
         * writeback completing on this page, since we'll remove it
         * from the pagecache.  Otherwise truncate wont wait on the
         * page, allowing the disk blocks to be reused by someone else
         * before we actually wrote our data to them. fs corruption
         * ensues.
         */
        wait_on_page_writeback(page);

        if (page_has_private(page) &&
            !try_to_release_page(page, GFP_KERNEL))
            goto out_unlock;

        /*
         * If we succeeded in removing the mapping, set LRU flag
         * and return good.
         */
        if (remove_mapping(mapping, page)) {
            buf->flags |= PIPE_BUF_FLAG_LRU;
            return 0;
        }
    }

    /*
     * Raced with truncate or failed to remove page from current
     * address space, unlock and return failure.
     */
out_unlock:
    unlock_page(page);
    return 1;
}

static void page_cache_pipe_buf_release(struct pipe_inode_info *pipe,
                    struct pipe_buffer *buf)
{
    page_cache_release(buf->page);
    buf->flags &= ~PIPE_BUF_FLAG_LRU;
}

/*
 * Check whether the contents of buf is OK to access. Since the content
 * is a page cache page, IO may be in flight.
 */
static int page_cache_pipe_buf_confirm(struct pipe_inode_info *pipe,
                       struct pipe_buffer *buf)
{
    struct page *page = buf->page;
    int err;

    if (!PageUptodate(page)) {
        lock_page(page);

        /*
         * Page got truncated/unhashed. This will cause a 0-byte
         * splice, if this is the first page.
         */
        if (!page->mapping) {
            err = -ENODATA;
            goto error;
        }

        /*
         * Uh oh, read-error from disk.
         */
        if (!PageUptodate(page)) {
            err = -EIO;
            goto error;
        }

        /*
         * Page is ok afterall, we are done.
         */
        unlock_page(page);
    }

    return 0;
error:
    unlock_page(page);
    return err;
}

const struct pipe_buf_operations page_cache_pipe_buf_ops = {
    .can_merge = 0,
    .map = generic_pipe_buf_map,
    .unmap = generic_pipe_buf_unmap,
    .confirm = page_cache_pipe_buf_confirm,
    .release = page_cache_pipe_buf_release,
    .steal = page_cache_pipe_buf_steal,
    .get = generic_pipe_buf_get,
};

static int user_page_pipe_buf_steal(struct pipe_inode_info *pipe,
                    struct pipe_buffer *buf)
{
    if (!(buf->flags & PIPE_BUF_FLAG_GIFT))
        return 1;

    buf->flags |= PIPE_BUF_FLAG_LRU;
    return generic_pipe_buf_steal(pipe, buf);
}

static const struct pipe_buf_operations user_page_pipe_buf_ops = {
    .can_merge = 0,
    .map = generic_pipe_buf_map,
    .unmap = generic_pipe_buf_unmap,
    .confirm = generic_pipe_buf_confirm,
    .release = page_cache_pipe_buf_release,
    .steal = user_page_pipe_buf_steal,
    .get = generic_pipe_buf_get,
};

static void wakeup_pipe_readers(struct pipe_inode_info *pipe)
{
    smp_mb();
    if (waitqueue_active(&pipe->wait))
        wake_up_interruptible(&pipe->wait);
    kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
}

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];
            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++;
            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;
}

void spd_release_page(struct splice_pipe_desc *spd, unsigned int i)
{
    page_cache_release(spd->pages[i]);
}

/*
 * Check if we need to grow the arrays holding pages and partial page
 * descriptions.
 */
int splice_grow_spd(const struct pipe_inode_info *pipe, struct splice_pipe_desc *spd)
{
    unsigned int buffers = ACCESS_ONCE(pipe->buffers);

    spd->nr_pages_max = buffers;
    if (buffers <= PIPE_DEF_BUFFERS)
        return 0;

    spd->pages = kmalloc(buffers * sizeof(struct page *), GFP_KERNEL);
    spd->partial = kmalloc(buffers * sizeof(struct partial_page), GFP_KERNEL);

    if (spd->pages && spd->partial)
        return 0;

    kfree(spd->pages);
    kfree(spd->partial);
    return -ENOMEM;
}

void splice_shrink_spd(struct splice_pipe_desc *spd)
{
    if (spd->nr_pages_max <= PIPE_DEF_BUFFERS)
        return;

    kfree(spd->pages);
    kfree(spd->partial);
}

static int
__generic_file_splice_read(struct file *in, loff_t *ppos,
               struct pipe_inode_info *pipe, size_t len,
               unsigned int flags)
{
    struct address_space *mapping = in->f_mapping;
    unsigned int loff, nr_pages, req_pages;
    struct page *pages[PIPE_DEF_BUFFERS];
    struct partial_page partial[PIPE_DEF_BUFFERS];
    struct page *page;
    pgoff_t index, end_index;
    loff_t isize;
    int error, page_nr;
    struct splice_pipe_desc spd = {
        .pages = pages,
        .partial = partial,
        .nr_pages_max = PIPE_DEF_BUFFERS,
        .flags = flags,
        .ops = &page_cache_pipe_buf_ops,
        .spd_release = spd_release_page,
    };

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

    index = *ppos >> PAGE_CACHE_SHIFT;
    loff = *ppos & ~PAGE_CACHE_MASK;
    req_pages = (len + loff + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT;
    nr_pages = min(req_pages, spd.nr_pages_max);

    /*
     * Lookup the (hopefully) full range of pages we need.
     */
    spd.nr_pages = find_get_pages_contig(mapping, index, nr_pages, spd.pages);
    index += spd.nr_pages;

    /*
     * If find_get_pages_contig() returned fewer pages than we needed,
     * readahead/allocate the rest and fill in the holes.
     */
    if (spd.nr_pages < nr_pages)
        page_cache_sync_readahead(mapping, &in->f_ra, in,
                index, req_pages - spd.nr_pages);

    error = 0;
    while (spd.nr_pages < nr_pages) {
        /*
         * Page could be there, find_get_pages_contig() breaks on
         * the first hole.
         */
        page = find_get_page(mapping, index);
        if (!page) {
            /*
             * page didn't exist, allocate one.
             */
            page = page_cache_alloc_cold(mapping);
            if (!page)
                break;

            error = add_to_page_cache_lru(page, mapping, index,
                        GFP_KERNEL);
            if (unlikely(error)) {
                page_cache_release(page);
                if (error == -EEXIST)
                    continue;
                break;
            }
            /*
             * add_to_page_cache() locks the page, unlock it
             * to avoid convoluting the logic below even more.
             */
            unlock_page(page);
        }

        spd.pages[spd.nr_pages++] = page;
        index++;
    }

    /*
     * Now loop over the map and see if we need to start IO on any
     * pages, fill in the partial map, etc.
     */
    index = *ppos >> PAGE_CACHE_SHIFT;
    nr_pages = spd.nr_pages;
    spd.nr_pages = 0;
    for (page_nr = 0; page_nr < nr_pages; page_nr++) {
        unsigned int this_len;

        if (!len)
            break;

        /*
         * this_len is the max we'll use from this page
         */
        this_len = min_t(unsigned long, len, PAGE_CACHE_SIZE - loff);
        page = spd.pages[page_nr];

        if (PageReadahead(page))
            page_cache_async_readahead(mapping, &in->f_ra, in,
                    page, index, req_pages - page_nr);

        /*
         * If the page isn't uptodate, we may need to start io on it
         */
        if (!PageUptodate(page)) {
            lock_page(page);

            /*
             * Page was truncated, or invalidated by the
             * filesystem.  Redo the find/create, but this time the
             * page is kept locked, so there's no chance of another
             * race with truncate/invalidate.
             */
            if (!page->mapping) {
                unlock_page(page);
                page = find_or_create_page(mapping, index,
                        mapping_gfp_mask(mapping));

                if (!page) {
                    error = -ENOMEM;
                    break;
                }
                page_cache_release(spd.pages[page_nr]);
                spd.pages[page_nr] = page;
            }
            /*
             * page was already under io and is now done, great
             */
            if (PageUptodate(page)) {
                unlock_page(page);
                goto fill_it;
            }

            /*
             * need to read in the page
             */
            error = mapping->a_ops->readpage(in, page);
            if (unlikely(error)) {
                /*
                 * We really should re-lookup the page here,
                 * but it complicates things a lot. Instead
                 * lets just do what we already stored, and
                 * we'll get it the next time we are called.
                 */
                if (error == AOP_TRUNCATED_PAGE)
                    error = 0;

                break;
            }
        }
fill_it:
        /*
         * i_size must be checked after PageUptodate.
         */
        isize = i_size_read(mapping->host);
        end_index = (isize - 1) >> PAGE_CACHE_SHIFT;
        if (unlikely(!isize || index > end_index))
            break;

        /*
         * if this is the last page, see if we need to shrink
         * the length and stop
         */
        if (end_index == index) {
            unsigned int plen;

            /*
             * max good bytes in this page
             */
            plen = ((isize - 1) & ~PAGE_CACHE_MASK) + 1;
            if (plen <= loff)
                break;

            /*
             * force quit after adding this page
             */
            this_len = min(this_len, plen - loff);
            len = this_len;
        }

        spd.partial[page_nr].offset = loff;
        spd.partial[page_nr].len = this_len;
        len -= this_len;
        loff = 0;
        spd.nr_pages++;
        index++;
    }

    /*
     * Release any pages at the end, if we quit early. 'page_nr' is how far
     * we got, 'nr_pages' is how many pages are in the map.
     */
    while (page_nr < nr_pages)
        page_cache_release(spd.pages[page_nr++]);
    in->f_ra.prev_pos = (loff_t)index << PAGE_CACHE_SHIFT;

    if (spd.nr_pages)
        error = splice_to_pipe(pipe, &spd);

    splice_shrink_spd(&spd);
    return error;
}

ssize_t generic_file_splice_read(struct file *in, loff_t *ppos,
                 struct pipe_inode_info *pipe, size_t len,
                 unsigned int flags)
{
    loff_t isize, left;
    int ret;

    isize = i_size_read(in->f_mapping->host);
    if (unlikely(*ppos >= isize))
        return 0;

    left = isize - *ppos;
    if (unlikely(left < len))
        len = left;

    ret = __generic_file_splice_read(in, ppos, pipe, len, flags);
    if (ret > 0) {
        *ppos += ret;
        file_accessed(in);
    }

    return ret;
}
EXPORT_SYMBOL(generic_file_splice_read);

static const struct pipe_buf_operations default_pipe_buf_ops = {
    .can_merge = 0,
    .map = generic_pipe_buf_map,
    .unmap = generic_pipe_buf_unmap,
    .confirm = generic_pipe_buf_confirm,
    .release = generic_pipe_buf_release,
    .steal = generic_pipe_buf_steal,
    .get = generic_pipe_buf_get,
};

static ssize_t kernel_readv(struct file *file, const struct iovec *vec,
                unsigned long vlen, loff_t offset)
{
    mm_segment_t old_fs;
    loff_t pos = offset;
    ssize_t res;

    old_fs = get_fs();
    set_fs(get_ds());
    /* The cast to a user pointer is valid due to the set_fs() */
    res = vfs_readv(file, (const struct iovec __user *)vec, vlen, &pos);
    set_fs(old_fs);

    return res;
}

static ssize_t kernel_write(struct file *file, const char *buf, size_t count,
                loff_t pos)
{
    mm_segment_t old_fs;
    ssize_t res;

    old_fs = get_fs();
    set_fs(get_ds());
    /* The cast to a user pointer is valid due to the set_fs() */
    res = vfs_write(file, (const char __user *)buf, count, &pos);
    set_fs(old_fs);

    return res;
}

ssize_t default_file_splice_read(struct file *in, loff_t *ppos,
                 struct pipe_inode_info *pipe, size_t len,
                 unsigned int flags)
{
    unsigned int nr_pages;
    unsigned int nr_freed;
    size_t offset;
    struct page *pages[PIPE_DEF_BUFFERS];
    struct partial_page partial[PIPE_DEF_BUFFERS];
    struct iovec *vec, __vec[PIPE_DEF_BUFFERS];
    ssize_t res;
    size_t this_len;
    int error;
    int i;
    struct splice_pipe_desc spd = {
        .pages = pages,
        .partial = partial,
        .nr_pages_max = PIPE_DEF_BUFFERS,
        .flags = flags,
        .ops = &default_pipe_buf_ops,
        .spd_release = spd_release_page,
    };

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

    res = -ENOMEM;
    vec = __vec;
    if (spd.nr_pages_max > PIPE_DEF_BUFFERS) {
        vec = kmalloc(spd.nr_pages_max * sizeof(struct iovec), GFP_KERNEL);
        if (!vec)
            goto shrink_ret;
    }

    offset = *ppos & ~PAGE_CACHE_MASK;
    nr_pages = (len + offset + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT;

    for (i = 0; i < nr_pages && i < spd.nr_pages_max && len; i++) {
        struct page *page;

        page = alloc_page(GFP_USER);
        error = -ENOMEM;
        if (!page)
            goto err;

        this_len = min_t(size_t, len, PAGE_CACHE_SIZE - offset);
        vec[i].iov_base = (void __user *) page_address(page);
        vec[i].iov_len = this_len;
        spd.pages[i] = page;
        spd.nr_pages++;
        len -= this_len;
        offset = 0;
    }

    res = kernel_readv(in, vec, spd.nr_pages, *ppos);
    if (res < 0) {
        error = res;
        goto err;
    }

    error = 0;
    if (!res)
        goto err;

    nr_freed = 0;
    for (i = 0; i < spd.nr_pages; i++) {
        this_len = min_t(size_t, vec[i].iov_len, res);
        spd.partial[i].offset = 0;
        spd.partial[i].len = this_len;
        if (!this_len) {
            __free_page(spd.pages[i]);
            spd.pages[i] = NULL;
            nr_freed++;
        }
        res -= this_len;
    }
    spd.nr_pages -= nr_freed;

    res = splice_to_pipe(pipe, &spd);
    if (res > 0)
        *ppos += res;

shrink_ret:
    if (vec != __vec)
        kfree(vec);
    splice_shrink_spd(&spd);
    return res;

err:
    for (i = 0; i < spd.nr_pages; i++)
        __free_page(spd.pages[i]);

    res = error;
    goto shrink_ret;
}
EXPORT_SYMBOL(default_file_splice_read);

/*
 * Send 'sd->len' bytes to socket from 'sd->file' at position 'sd->pos'
 * using sendpage(). Return the number of bytes sent.
 */
static int pipe_to_sendpage(struct pipe_inode_info *pipe,
                struct pipe_buffer *buf, struct splice_desc *sd)
{
    struct file *file = sd->u.file;
    loff_t pos = sd->pos;
    int more;

    if (!likely(file->f_op && file->f_op->sendpage))
        return -EINVAL;

    more = (sd->flags & SPLICE_F_MORE) ? MSG_MORE : 0;
    if (sd->len < sd->total_len)
        more |= MSG_SENDPAGE_NOTLAST;
    return file->f_op->sendpage(file, buf->page, buf->offset,
                    sd->len, &pos, more);
}

/*
 * This is a little more tricky than the file -> pipe splicing. There are
 * basically three cases:
 *
 *  - Destination page already exists in the address space and there
 *    are users of it. For that case we have no other option that
 *    copying the data. Tough luck.
 *  - Destination page already exists in the address space, but there
 *    are no users of it. Make sure it's uptodate, then drop it. Fall
 *    through to last case.
 *  - Destination page does not exist, we can add the pipe page to
 *    the page cache and avoid the copy.
 *
 * If asked to move pages to the output file (SPLICE_F_MOVE is set in
 * sd->flags), we attempt to migrate pages from the pipe to the output
 * file address space page cache. This is possible if no one else has
 * the pipe page referenced outside of the pipe and page cache. If
 * SPLICE_F_MOVE isn't set, or we cannot move the page, we simply create
 * a new page in the output file page cache and fill/dirty that.
 */
int pipe_to_file(struct pipe_inode_info *pipe, struct pipe_buffer *buf,
         struct splice_desc *sd)
{
    struct file *file = sd->u.file;
    struct address_space *mapping = file->f_mapping;
    unsigned int offset, this_len;
    struct page *page;
    void *fsdata;
    int ret;

    offset = sd->pos & ~PAGE_CACHE_MASK;

    this_len = sd->len;
    if (this_len + offset > PAGE_CACHE_SIZE)
        this_len = PAGE_CACHE_SIZE - offset;

    ret = pagecache_write_begin(file, mapping, sd->pos, this_len,
                AOP_FLAG_UNINTERRUPTIBLE, &page, &fsdata);
    if (unlikely(ret))
        goto out;

    if (buf->page != page) {
        char *src = buf->ops->map(pipe, buf, 1);
        char *dst = kmap_atomic(page);

        memcpy(dst + offset, src + buf->offset, this_len);
        flush_dcache_page(page);
        kunmap_atomic(dst);
        buf->ops->unmap(pipe, buf, src);
    }
    ret = pagecache_write_end(file, mapping, sd->pos, this_len, this_len,
                page, fsdata);
out:
    return ret;
}
EXPORT_SYMBOL(pipe_to_file);

static void wakeup_pipe_writers(struct pipe_inode_info *pipe)
{
    smp_mb();
    if (waitqueue_active(&pipe->wait))
        wake_up_interruptible(&pipe->wait);
    kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
}

int splice_from_pipe_feed(struct pipe_inode_info *pipe, struct splice_desc *sd,
              splice_actor *actor)
{
    int ret;

    while (pipe->nrbufs) {
        struct pipe_buffer *buf = pipe->bufs + pipe->curbuf;
        const struct pipe_buf_operations *ops = buf->ops;

        sd->len = buf->len;
        if (sd->len > sd->total_len)
            sd->len = sd->total_len;

        ret = buf->ops->confirm(pipe, buf);
        if (unlikely(ret)) {
            if (ret == -ENODATA)
                ret = 0;
            return ret;
        }

        ret = actor(pipe, buf, sd);
        if (ret <= 0)
            return ret;

        buf->offset += ret;
        buf->len -= ret;

        sd->num_spliced += ret;
        sd->len -= ret;
        sd->pos += ret;
        sd->total_len -= ret;

        if (!buf->len) {
            buf->ops = NULL;
            ops->release(pipe, buf);
            pipe->curbuf = (pipe->curbuf + 1) & (pipe->buffers - 1);
            pipe->nrbufs--;
            if (pipe->inode)
                sd->need_wakeup = true;
        }

        if (!sd->total_len)
            return 0;
    }

    return 1;
}
EXPORT_SYMBOL(splice_from_pipe_feed);

int splice_from_pipe_next(struct pipe_inode_info *pipe, struct splice_desc *sd)
{
    while (!pipe->nrbufs) {
        if (!pipe->writers)
            return 0;

        if (!pipe->waiting_writers && sd->num_spliced)
            return 0;

        if (sd->flags & SPLICE_F_NONBLOCK)
            return -EAGAIN;

        if (signal_pending(current))
            return -ERESTARTSYS;

        if (sd->need_wakeup) {
            wakeup_pipe_writers(pipe);
            sd->need_wakeup = false;
        }

        pipe_wait(pipe);
    }

    return 1;
}
EXPORT_SYMBOL(splice_from_pipe_next);

void splice_from_pipe_begin(struct splice_desc *sd)
{
    sd->num_spliced = 0;
    sd->need_wakeup = false;
}
EXPORT_SYMBOL(splice_from_pipe_begin);

void splice_from_pipe_end(struct pipe_inode_info *pipe, struct splice_desc *sd)
{
    if (sd->need_wakeup)
        wakeup_pipe_writers(pipe);
}
EXPORT_SYMBOL(splice_from_pipe_end);

ssize_t __splice_from_pipe(struct pipe_inode_info *pipe, struct splice_desc *sd,
               splice_actor *actor)
{
    int ret;

    splice_from_pipe_begin(sd);
    do {
        ret = splice_from_pipe_next(pipe, sd);
        if (ret > 0)
            ret = splice_from_pipe_feed(pipe, sd, actor);
    } while (ret > 0);
    splice_from_pipe_end(pipe, sd);

    return sd->num_spliced ? sd->num_spliced : ret;
}
EXPORT_SYMBOL(__splice_from_pipe);

ssize_t splice_from_pipe(struct pipe_inode_info *pipe, struct file *out,
             loff_t *ppos, size_t len, unsigned int flags,
             splice_actor *actor)
{
    ssize_t ret;
    struct splice_desc sd = {
        .total_len = len,
        .flags = flags,
        .pos = *ppos,
        .u.file = out,
    };

    pipe_lock(pipe);
    ret = __splice_from_pipe(pipe, &sd, actor);
    pipe_unlock(pipe);

    return ret;
}

ssize_t
generic_file_splice_write(struct pipe_inode_info *pipe, struct file *out,
              loff_t *ppos, size_t len, unsigned int flags)
{
    struct address_space *mapping = out->f_mapping;
    struct inode *inode = mapping->host;
    struct splice_desc sd = {
        .total_len = len,
        .flags = flags,
        .pos = *ppos,
        .u.file = out,
    };
    ssize_t ret;

    sb_start_write(inode->i_sb);

    pipe_lock(pipe);

    splice_from_pipe_begin(&sd);
    do {
        ret = splice_from_pipe_next(pipe, &sd);
        if (ret <= 0)
            break;

        mutex_lock_nested(&inode->i_mutex, I_MUTEX_CHILD);
        ret = file_remove_suid(out);
        if (!ret) {
            ret = file_update_time(out);
            if (!ret)
                ret = splice_from_pipe_feed(pipe, &sd,
                                pipe_to_file);
        }
        mutex_unlock(&inode->i_mutex);
    } while (ret > 0);
    splice_from_pipe_end(pipe, &sd);

    pipe_unlock(pipe);

    if (sd.num_spliced)
        ret = sd.num_spliced;

    if (ret > 0) {
        unsigned long nr_pages;
        int err;

        nr_pages = (ret + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT;

        err = generic_write_sync(out, *ppos, ret);
        if (err)
            ret = err;
        else
            *ppos += ret;
        balance_dirty_pages_ratelimited_nr(mapping, nr_pages);
    }
    sb_end_write(inode->i_sb);

    return ret;
}

EXPORT_SYMBOL(generic_file_splice_write);

static int write_pipe_buf(struct pipe_inode_info *pipe, struct pipe_buffer *buf,
              struct splice_desc *sd)
{
    int ret;
    void *data;

    data = buf->ops->map(pipe, buf, 0);
    ret = kernel_write(sd->u.file, data + buf->offset, sd->len, sd->pos);
    buf->ops->unmap(pipe, buf, data);

    return ret;
}

static ssize_t default_file_splice_write(struct pipe_inode_info *pipe,
                     struct file *out, loff_t *ppos,
                     size_t len, unsigned int flags)
{
    ssize_t ret;

    ret = splice_from_pipe(pipe, out, ppos, len, flags, write_pipe_buf);
    if (ret > 0)
        *ppos += ret;

    return ret;
}

ssize_t generic_splice_sendpage(struct pipe_inode_info *pipe, struct file *out,
                loff_t *ppos, size_t len, unsigned int flags)
{
    return splice_from_pipe(pipe, out, ppos, len, flags, pipe_to_sendpage);
}

EXPORT_SYMBOL(generic_splice_sendpage);

/*
 * Attempt to initiate a splice from pipe to file.
 */
static long do_splice_from(struct pipe_inode_info *pipe, struct file *out,
               loff_t *ppos, size_t len, unsigned int flags)
{
    ssize_t (*splice_write)(struct pipe_inode_info *, struct file *,
                loff_t *, size_t, unsigned int);
    int ret;

    if (unlikely(!(out->f_mode & FMODE_WRITE)))
        return -EBADF;

    if (unlikely(out->f_flags & O_APPEND))
        return -EINVAL;

    ret = rw_verify_area(WRITE, out, ppos, len);
    if (unlikely(ret < 0))
        return ret;

    if (out->f_op && out->f_op->splice_write)
        splice_write = out->f_op->splice_write;
    else
        splice_write = default_file_splice_write;

    return splice_write(pipe, out, ppos, len, flags);
}

/*
 * Attempt to initiate a splice from a file to a pipe.
 */
static long do_splice_to(struct file *in, loff_t *ppos,
             struct pipe_inode_info *pipe, size_t len,
             unsigned int flags)
{
    ssize_t (*splice_read)(struct file *, loff_t *,
                   struct pipe_inode_info *, size_t, unsigned int);
    int ret;

    if (unlikely(!(in->f_mode & FMODE_READ)))
        return -EBADF;

    ret = rw_verify_area(READ, in, ppos, len);
    if (unlikely(ret < 0))
        return ret;

    if (in->f_op && in->f_op->splice_read)
        splice_read = in->f_op->splice_read;
    else
        splice_read = default_file_splice_read;

    return splice_read(in, ppos, pipe, len, flags);
}

ssize_t splice_direct_to_actor(struct file *in, struct splice_desc *sd,
                   splice_direct_actor *actor)
{
    struct pipe_inode_info *pipe;
    long ret, bytes;
    umode_t i_mode;
    size_t len;
    int i, flags;

    /*
     * We require the input being a regular file, as we don't want to
     * randomly drop data for eg socket -> socket splicing. Use the
     * piped splicing for that!
     */
    i_mode = in->f_path.dentry->d_inode->i_mode;
    if (unlikely(!S_ISREG(i_mode) && !S_ISBLK(i_mode)))
        return -EINVAL;

    /*
     * neither in nor out is a pipe, setup an internal pipe attached to
     * 'out' and transfer the wanted data from 'in' to 'out' through that
     */
    pipe = current->splice_pipe;
    if (unlikely(!pipe)) {
        pipe = alloc_pipe_info(NULL);
        if (!pipe)
            return -ENOMEM;

        /*
         * We don't have an immediate reader, but we'll read the stuff
         * out of the pipe right after the splice_to_pipe(). So set
         * PIPE_READERS appropriately.
         */
        pipe->readers = 1;

        current->splice_pipe = pipe;
    }

    /*
     * Do the splice.
     */
    ret = 0;
    bytes = 0;
    len = sd->total_len;
    flags = sd->flags;

    /*
     * Don't block on output, we have to drain the direct pipe.
     */
    sd->flags &= ~SPLICE_F_NONBLOCK;

    while (len) {
        size_t read_len;
        loff_t pos = sd->pos, prev_pos = pos;

        ret = do_splice_to(in, &pos, pipe, len, flags);
        if (unlikely(ret <= 0))
            goto out_release;

        read_len = ret;
        sd->total_len = read_len;

        /*
         * NOTE: nonblocking mode only applies to the input. We
         * must not do the output in nonblocking mode as then we
         * could get stuck data in the internal pipe:
         */
        ret = actor(pipe, sd);
        if (unlikely(ret <= 0)) {
            sd->pos = prev_pos;
            goto out_release;
        }

        bytes += ret;
        len -= ret;
        sd->pos = pos;

        if (ret < read_len) {
            sd->pos = prev_pos + ret;
            goto out_release;
        }
    }

done:
    pipe->nrbufs = pipe->curbuf = 0;
    file_accessed(in);
    return bytes;

out_release:
    /*
     * If we did an incomplete transfer we must release
     * the pipe buffers in question:
     */
    for (i = 0; i < pipe->buffers; i++) {
        struct pipe_buffer *buf = pipe->bufs + i;

        if (buf->ops) {
            buf->ops->release(pipe, buf);
            buf->ops = NULL;
        }
    }

    if (!bytes)
        bytes = ret;

    goto done;
}
EXPORT_SYMBOL(splice_direct_to_actor);

static int direct_splice_actor(struct pipe_inode_info *pipe,
                   struct splice_desc *sd)
{
    struct file *file = sd->u.file;

    return do_splice_from(pipe, file, &file->f_pos, sd->total_len,
                  sd->flags);
}

long do_splice_direct(struct file *in, loff_t *ppos, struct file *out,
              size_t len, unsigned int flags)
{
    struct splice_desc sd = {
        .len        = len,
        .total_len  = len,
        .flags      = flags,
        .pos        = *ppos,
        .u.file     = out,
    };
    long ret;

    ret = splice_direct_to_actor(in, &sd, direct_splice_actor);
    if (ret > 0)
        *ppos = sd.pos;

    return ret;
}

static int splice_pipe_to_pipe(struct pipe_inode_info *ipipe,
                   struct pipe_inode_info *opipe,
                   size_t len, unsigned int flags);

/*
 * Determine where to splice to/from.
 */
static 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, *off;
    long ret;

    ipipe = get_pipe_info(in);
    opipe = get_pipe_info(out);

    if (ipipe && opipe) {
        if (off_in || off_out)
            return -ESPIPE;

        if (!(in->f_mode & FMODE_READ))
            return -EBADF;

        if (!(out->f_mode & FMODE_WRITE))
            return -EBADF;

        /* Splicing to self would be fun, but... */
        if (ipipe == opipe)
            return -EINVAL;

        return splice_pipe_to_pipe(ipipe, opipe, len, flags);
    }

    if (ipipe) {
        if (off_in)
            return -ESPIPE;
        if (off_out) {
            if (!(out->f_mode & FMODE_PWRITE))
                return -EINVAL;
            if (copy_from_user(&offset, off_out, sizeof(loff_t)))
                return -EFAULT;
            off = &offset;
        } else
            off = &out->f_pos;

        ret = do_splice_from(ipipe, out, off, len, flags);

        if (off_out && copy_to_user(off_out, off, sizeof(loff_t)))
            ret = -EFAULT;

        return ret;
    }

    if (opipe) {
        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;
            off = &offset;
        } else
            off = &in->f_pos;

        ret = do_splice_to(in, off, opipe, len, flags);

        if (off_in && copy_to_user(off_in, off, sizeof(loff_t)))
            ret = -EFAULT;

        return ret;
    }

    return -EINVAL;
}

/*
 * Map an iov into an array of pages and offset/length tupples. With the
 * partial_page structure, we can map several non-contiguous ranges into
 * our ones pages[] map instead of splitting that operation into pieces.
 * Could easily be exported as a generic helper for other users, in which
 * case one would probably want to add a 'max_nr_pages' parameter as well.
 */
static int get_iovec_page_array(const struct iovec __user *iov,
                unsigned int nr_vecs, struct page **pages,
                struct partial_page *partial, bool aligned,
                unsigned int pipe_buffers)
{
    int buffers = 0, error = 0;

    while (nr_vecs) {
        unsigned long off, npages;
        struct iovec entry;
        void __user *base;
        size_t len;
        int i;

        error = -EFAULT;
        if (copy_from_user(&entry, iov, sizeof(entry)))
            break;

        base = entry.iov_base;
        len = entry.iov_len;

        /*
         * Sanity check this iovec. 0 read succeeds.
         */
        error = 0;
        if (unlikely(!len))
            break;
        error = -EFAULT;
        if (!access_ok(VERIFY_READ, base, len))
            break;

        /*
         * Get this base offset and number of pages, then map
         * in the user pages.
         */
        off = (unsigned long) base & ~PAGE_MASK;

        /*
         * If asked for alignment, the offset must be zero and the
         * length a multiple of the PAGE_SIZE.
         */
        error = -EINVAL;
        if (aligned && (off || len & ~PAGE_MASK))
            break;

        npages = (off + len + PAGE_SIZE - 1) >> PAGE_SHIFT;
        if (npages > pipe_buffers - buffers)
            npages = pipe_buffers - buffers;

        error = get_user_pages_fast((unsigned long)base, npages,
                    0, &pages[buffers]);

        if (unlikely(error <= 0))
            break;

        /*
         * Fill this contiguous range into the partial page map.
         */
        for (i = 0; i < error; i++) {
            const int plen = min_t(size_t, len, PAGE_SIZE - off);

            partial[buffers].offset = off;
            partial[buffers].len = plen;

            off = 0;
            len -= plen;
            buffers++;
        }

        /*
         * We didn't complete this iov, stop here since it probably
         * means we have to move some of this into a pipe to
         * be able to continue.
         */
        if (len)
            break;

        /*
         * Don't continue if we mapped fewer pages than we asked for,
         * or if we mapped the max number of pages that we have
         * room for.
         */
        if (error < npages || buffers == pipe_buffers)
            break;

        nr_vecs--;
        iov++;
    }

    if (buffers)
        return buffers;

    return error;
}

static int pipe_to_user(struct pipe_inode_info *pipe, struct pipe_buffer *buf,
            struct splice_desc *sd)
{
    char *src;
    int ret;

    /*
     * See if we can use the atomic maps, by prefaulting in the
     * pages and doing an atomic copy
     */
    if (!fault_in_pages_writeable(sd->u.userptr, sd->len)) {
        src = buf->ops->map(pipe, buf, 1);
        ret = __copy_to_user_inatomic(sd->u.userptr, src + buf->offset,
                            sd->len);
        buf->ops->unmap(pipe, buf, src);
        if (!ret) {
            ret = sd->len;
            goto out;
        }
    }

    /*
     * No dice, use slow non-atomic map and copy
     */
    src = buf->ops->map(pipe, buf, 0);

    ret = sd->len;
    if (copy_to_user(sd->u.userptr, src + buf->offset, sd->len))
        ret = -EFAULT;

    buf->ops->unmap(pipe, buf, src);
out:
    if (ret > 0)
        sd->u.userptr += ret;
    return ret;
}

/*
 * For lack of a better implementation, implement vmsplice() to userspace
 * as a simple copy of the pipes pages to the user iov.
 */
static long vmsplice_to_user(struct file *file, const struct iovec __user *iov,
                 unsigned long nr_segs, unsigned int flags)
{
    struct pipe_inode_info *pipe;
    struct splice_desc sd;
    ssize_t size;
    int error;
    long ret;

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

    pipe_lock(pipe);

    error = ret = 0;
    while (nr_segs) {
        void __user *base;
        size_t len;

        /*
         * Get user address base and length for this iovec.
         */
        error = get_user(base, &iov->iov_base);
        if (unlikely(error))
            break;
        error = get_user(len, &iov->iov_len);
        if (unlikely(error))
            break;

        /*
         * Sanity check this iovec. 0 read succeeds.
         */
        if (unlikely(!len))
            break;
        if (unlikely(!base)) {
            error = -EFAULT;
            break;
        }

        if (unlikely(!access_ok(VERIFY_WRITE, base, len))) {
            error = -EFAULT;
            break;
        }

        sd.len = 0;
        sd.total_len = len;
        sd.flags = flags;
        sd.u.userptr = base;
        sd.pos = 0;

        size = __splice_from_pipe(pipe, &sd, pipe_to_user);
        if (size < 0) {
            if (!ret)
                ret = size;

            break;
        }

        ret += size;

        if (size < len)
            break;

        nr_segs--;
        iov++;
    }

    pipe_unlock(pipe);

    if (!ret)
        ret = error;

    return ret;
}

/*
 * vmsplice splices a user address range into a pipe. It can be thought of
 * as splice-from-memory, where the regular splice is splice-from-file (or
 * to file). In both cases the output is a pipe, naturally.
 */
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;
}

/*
 * Note that vmsplice only really supports true splicing _from_ user memory
 * to a pipe, not the other way around. Splicing from user memory is a simple
 * operation that can be supported without any funky alignment restrictions
 * or nasty vm tricks. We simply map in the user memory and fill them into
 * a pipe. The reverse isn't quite as easy, though. There are two possible
 * solutions for that:
 *
 *  - memcpy() the data internally, at which point we might as well just
 *    do a regular read() on the buffer anyway.
 *  - Lots of nasty vm tricks, that are neither fast nor flexible (it
 *    has restriction limitations on both ends of the pipe).
 *
 * Currently we punt and implement it as a normal copy, see pipe_to_user().
 *
 */
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;
}

SYSCALL_DEFINE6(splice, int, fd_in, loff_t __user *, off_in,
        int, fd_out, loff_t __user *, off_out,
        size_t, len, unsigned int, flags)
{
    struct fd in, out;
    long error;

    if (unlikely(!len))
        return 0;

    error = -EBADF;
    in = fdget(fd_in);
    if (in.file) {
        if (in.file->f_mode & FMODE_READ) {
            out = fdget(fd_out);
            if (out.file) {
                if (out.file->f_mode & FMODE_WRITE)
                    error = do_splice(in.file, off_in,
                              out.file, off_out,
                              len, flags);
                fdput(out);
            }
        }
        fdput(in);
    }
    return error;
}

/*
 * Make sure there's data to read. Wait for input if we can, otherwise
 * return an appropriate error.
 */
static int ipipe_prep(struct pipe_inode_info *pipe, unsigned int flags)
{
    int ret;

    /*
     * Check ->nrbufs without the inode lock first. This function
     * is speculative anyways, so missing one is ok.
     */
    if (pipe->nrbufs)
        return 0;

    ret = 0;
    pipe_lock(pipe);

    while (!pipe->nrbufs) {
        if (signal_pending(current)) {
            ret = -ERESTARTSYS;
            break;
        }
        if (!pipe->writers)
            break;
        if (!pipe->waiting_writers) {
            if (flags & SPLICE_F_NONBLOCK) {
                ret = -EAGAIN;
                break;
            }
        }
        pipe_wait(pipe);
    }

    pipe_unlock(pipe);
    return ret;
}

/*
 * Make sure there's writeable room. Wait for room if we can, otherwise
 * return an appropriate error.
 */
static int opipe_prep(struct pipe_inode_info *pipe, unsigned int flags)
{
    int ret;

    /*
     * Check ->nrbufs without the inode lock first. This function
     * is speculative anyways, so missing one is ok.
     */
    if (pipe->nrbufs < pipe->buffers)
        return 0;

    ret = 0;
    pipe_lock(pipe);

    while (pipe->nrbufs >= pipe->buffers) {
        if (!pipe->readers) {
            send_sig(SIGPIPE, current, 0);
            ret = -EPIPE;
            break;
        }
        if (flags & SPLICE_F_NONBLOCK) {
            ret = -EAGAIN;
            break;
        }
        if (signal_pending(current)) {
            ret = -ERESTARTSYS;
            break;
        }
        pipe->waiting_writers++;
        pipe_wait(pipe);
        pipe->waiting_writers--;
    }

    pipe_unlock(pipe);
    return ret;
}

/*
 * Splice contents of ipipe to opipe.
 */
static int splice_pipe_to_pipe(struct pipe_inode_info *ipipe,
                   struct pipe_inode_info *opipe,
                   size_t len, unsigned int flags)
{
    struct pipe_buffer *ibuf, *obuf;
    int ret = 0, nbuf;
    bool input_wakeup = false;


retry:
    ret = ipipe_prep(ipipe, flags);
    if (ret)
        return ret;

    ret = opipe_prep(opipe, flags);
    if (ret)
        return ret;

    /*
     * Potential ABBA deadlock, work around it by ordering lock
     * grabbing by pipe info address. Otherwise two different processes
     * could deadlock (one doing tee from A -> B, the other from B -> A).
     */
    pipe_double_lock(ipipe, opipe);

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

        if (!ipipe->nrbufs && !ipipe->writers)
            break;

        /*
         * Cannot make any progress, because either the input
         * pipe is empty or the output pipe is full.
         */
        if (!ipipe->nrbufs || opipe->nrbufs >= opipe->buffers) {
            /* Already processed some buffers, break */
            if (ret)
                break;

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

            /*
             * We raced with another reader/writer and haven't
             * managed to process any buffers.  A zero return
             * value means EOF, so retry instead.
             */
            pipe_unlock(ipipe);
            pipe_unlock(opipe);
            goto retry;
        }

        ibuf = ipipe->bufs + ipipe->curbuf;
        nbuf = (opipe->curbuf + opipe->nrbufs) & (opipe->buffers - 1);
        obuf = opipe->bufs + nbuf;

        if (len >= ibuf->len) {
            /*
             * Simply move the whole buffer from ipipe to opipe
             */
            *obuf = *ibuf;
            ibuf->ops = NULL;
            opipe->nrbufs++;
            ipipe->curbuf = (ipipe->curbuf + 1) & (ipipe->buffers - 1);
            ipipe->nrbufs--;
            input_wakeup = true;
        } else {
            /*
             * Get a reference to this pipe buffer,
             * so we can copy the contents over.
             */
            ibuf->ops->get(ipipe, ibuf);
            *obuf = *ibuf;

            /*
             * Don't inherit the gift flag, we need to
             * prevent multiple steals of this page.
             */
            obuf->flags &= ~PIPE_BUF_FLAG_GIFT;

            obuf->len = len;
            opipe->nrbufs++;
            ibuf->offset += obuf->len;
            ibuf->len -= obuf->len;
        }
        ret += obuf->len;
        len -= obuf->len;
    } while (len);

    pipe_unlock(ipipe);
    pipe_unlock(opipe);

    /*
     * If we put data in the output pipe, wakeup any potential readers.
     */
    if (ret > 0)
        wakeup_pipe_readers(opipe);

    if (input_wakeup)
        wakeup_pipe_writers(ipipe);

    return ret;
}

/*
 * Link contents of ipipe to opipe.
 */
static int link_pipe(struct pipe_inode_info *ipipe,
             struct pipe_inode_info *opipe,
             size_t len, unsigned int flags)
{
    struct pipe_buffer *ibuf, *obuf;
    int ret = 0, i = 0, nbuf;

    /*
     * Potential ABBA deadlock, work around it by ordering lock
     * grabbing by pipe info address. Otherwise two different processes
     * could deadlock (one doing tee from A -> B, the other from B -> A).
     */
    pipe_double_lock(ipipe, opipe);

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

        /*
         * If we have iterated all input buffers or ran out of
         * output room, break.
         */
        if (i >= ipipe->nrbufs || opipe->nrbufs >= opipe->buffers)
            break;

        ibuf = ipipe->bufs + ((ipipe->curbuf + i) & (ipipe->buffers-1));
        nbuf = (opipe->curbuf + opipe->nrbufs) & (opipe->buffers - 1);

        /*
         * Get a reference to this pipe buffer,
         * so we can copy the contents over.
         */
        ibuf->ops->get(ipipe, ibuf);

        obuf = opipe->bufs + nbuf;
        *obuf = *ibuf;

        /*
         * Don't inherit the gift flag, we need to
         * prevent multiple steals of this page.
         */
        obuf->flags &= ~PIPE_BUF_FLAG_GIFT;

        if (obuf->len > len)
            obuf->len = len;

        opipe->nrbufs++;
        ret += obuf->len;
        len -= obuf->len;
        i++;
    } while (len);

    /*
     * return EAGAIN if we have the potential of some data in the
     * future, otherwise just return 0
     */
    if (!ret && ipipe->waiting_writers && (flags & SPLICE_F_NONBLOCK))
        ret = -EAGAIN;

    pipe_unlock(ipipe);
    pipe_unlock(opipe);

    /*
     * If we put data in the output pipe, wakeup any potential readers.
     */
    if (ret > 0)
        wakeup_pipe_readers(opipe);

    return ret;
}

/*
 * This is a tee(1) implementation that works on pipes. It doesn't copy
 * any data, it simply references the 'in' pages on the 'out' pipe.
 * The 'flags' used are the SPLICE_F_* variants, currently the only
 * applicable one is SPLICE_F_NONBLOCK.
 */
static long do_tee(struct file *in, struct file *out, size_t len,
           unsigned int flags)
{
    struct pipe_inode_info *ipipe = get_pipe_info(in);
    struct pipe_inode_info *opipe = get_pipe_info(out);
    int ret = -EINVAL;

    /*
     * Duplicate the contents of ipipe to opipe without actually
     * copying the data.
     */
    if (ipipe && opipe && ipipe != opipe) {
        /*
         * Keep going, unless we encounter an error. The ipipe/opipe
         * ordering doesn't really matter.
         */
        ret = ipipe_prep(ipipe, flags);
        if (!ret) {
            ret = opipe_prep(opipe, flags);
            if (!ret)
                ret = link_pipe(ipipe, opipe, len, flags);
        }
    }

    return ret;
}

SYSCALL_DEFINE4(tee, int, fdin, int, fdout, size_t, len, unsigned int, flags)
{
    struct fd in;
    int error;

    if (unlikely(!len))
        return 0;

    error = -EBADF;
    in = fdget(fdin);
    if (in.file) {
        if (in.file->f_mode & FMODE_READ) {
            struct fd out = fdget(fdout);
            if (out.file) {
                if (out.file->f_mode & FMODE_WRITE)
                    error = do_tee(in.file, out.file,
                            len, flags);
                fdput(out);
            }
        }
        fdput(in);
    }

    return error;
}