[PATCH] splice: add support for sys_tee()

Basically an in-kernel implementation of tee, which uses splice and the
pipe buffers as an intelligent way to pass data around by reference.

Where the user space tee consumes the input and produces a stdout and
file output, this syscall merely duplicates the data inside a pipe to
another pipe. No data is copied, the output just grabs a reference to the
input pipe data.

Signed-off-by: Jens Axboe <axboe@suse.de>

11 files changed, 208 insertions, 4 deletions

diff --git a/arch/i386/kernel/syscall_table.S b/arch/i386/kernel/syscall_table.S
index 4f58b9c0ef..f48bef15b4 100644
--- a/arch/i386/kernel/syscall_table.S
+++ b/arch/i386/kernel/syscall_table.S
@@ -314,3 +314,4 @@ ENTRY(sys_call_table)
         .long sys_get_robust_list
         .long sys_splice
         .long sys_sync_file_range
+        .long sys_tee                   /* 315 */
diff --git a/arch/ia64/kernel/entry.S b/arch/ia64/kernel/entry.S
index 6e16f6b35b..e307988112 100644
--- a/arch/ia64/kernel/entry.S
+++ b/arch/ia64/kernel/entry.S
@@ -1609,5 +1609,6 @@ sys_call_table:
         data8 sys_set_robust_list
         data8 sys_get_robust_list
         data8 sys_sync_file_range               // 1300
+        data8 sys_tee
 
         .org sys_call_table + 8*NR_syscalls     // guard against failures to increase NR_syscalls
diff --git a/arch/powerpc/kernel/systbl.S b/arch/powerpc/kernel/systbl.S
index 1424eab450..a14c964038 100644
--- a/arch/powerpc/kernel/systbl.S
+++ b/arch/powerpc/kernel/systbl.S
@@ -323,3 +323,4 @@ COMPAT_SYS(pselect6)
 COMPAT_SYS(ppoll)
 SYSCALL(unshare)
 SYSCALL(splice)
+SYSCALL(tee)
diff --git a/fs/pipe.c b/fs/pipe.c
index e984beb93a..7fefb10db8 100644
--- a/fs/pipe.c
+++ b/fs/pipe.c
@@ -131,12 +131,19 @@ static int anon_pipe_buf_steal(struct pipe_inode_info *pipe,
         return 0;
 }
 
+static void anon_pipe_buf_get(struct pipe_inode_info *info,
+                              struct pipe_buffer *buf)
+{
+        page_cache_get(buf->page);
+}
+
 static struct pipe_buf_operations anon_pipe_buf_ops = {
         .can_merge = 1,
         .map = anon_pipe_buf_map,
         .unmap = anon_pipe_buf_unmap,
         .release = anon_pipe_buf_release,
         .steal = anon_pipe_buf_steal,
+        .get = anon_pipe_buf_get,
 };
 
 static ssize_t
diff --git a/fs/splice.c b/fs/splice.c
index 5d3eda6470..8d57e89924 100644
--- a/fs/splice.c
+++ b/fs/splice.c
@@ -125,12 +125,19 @@ static void page_cache_pipe_buf_unmap(struct pipe_inode_info *info,
         kunmap(buf->page);
 }
 
+static void page_cache_pipe_buf_get(struct pipe_inode_info *info,
+                                    struct pipe_buffer *buf)
+{
+        page_cache_get(buf->page);
+}
+
 static struct pipe_buf_operations page_cache_pipe_buf_ops = {
         .can_merge = 0,
         .map = page_cache_pipe_buf_map,
         .unmap = page_cache_pipe_buf_unmap,
         .release = page_cache_pipe_buf_release,
         .steal = page_cache_pipe_buf_steal,
+        .get = page_cache_pipe_buf_get,
 };
 
 /*
@@ -963,3 +970,182 @@ asmlinkage long sys_splice(int fd_in, loff_t __user *off_in,
 
         return error;
 }
+
+/*
+ * 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, do_wakeup = 0, i;
+
+        /*
+         * Potential ABBA deadlock, work around it by ordering lock
+         * grabbing by inode address. Otherwise two different processes
+         * could deadlock (one doing tee from A -> B, the other from B -> A).
+         */
+        if (ipipe->inode < opipe->inode) {
+                mutex_lock(&ipipe->inode->i_mutex);
+                mutex_lock(&opipe->inode->i_mutex);
+        } else {
+                mutex_lock(&opipe->inode->i_mutex);
+                mutex_lock(&ipipe->inode->i_mutex);
+        }
+
+        for (i = 0;; i++) {
+                if (!opipe->readers) {
+                        send_sig(SIGPIPE, current, 0);
+                        if (!ret)
+                                ret = -EPIPE;
+                        break;
+                }
+                if (ipipe->nrbufs - i) {
+                        ibuf = ipipe->bufs + ((ipipe->curbuf + i) & (PIPE_BUFFERS - 1));
+
+                        /*
+                         * If we have room, fill this buffer
+                         */
+                        if (opipe->nrbufs < PIPE_BUFFERS) {
+                                int nbuf = (opipe->curbuf + opipe->nrbufs) & (PIPE_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;
+
+                                if (obuf->len > len)
+                                        obuf->len = len;
+
+                                opipe->nrbufs++;
+                                do_wakeup = 1;
+                                ret += obuf->len;
+                                len -= obuf->len;
+
+                                if (!len)
+                                        break;
+                                if (opipe->nrbufs < PIPE_BUFFERS)
+                                        continue;
+                        }
+
+                        /*
+                         * We have input available, but no output room.
+                         * If we already copied data, return that.
+                         */
+                        if (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(&opipe->wait))
+                                        wake_up_interruptible(&opipe->wait);
+                                kill_fasync(&opipe->fasync_readers, SIGIO, POLL_IN);
+                                do_wakeup = 0;
+                        }
+
+                        opipe->waiting_writers++;
+                        pipe_wait(opipe);
+                        opipe->waiting_writers--;
+                        continue;
+                }
+
+                /*
+                 * No input buffers, do the usual checks for available
+                 * writers and blocking and wait if necessary
+                 */
+                if (!ipipe->writers)
+                        break;
+                if (!ipipe->waiting_writers) {
+                        if (ret)
+                                break;
+                }
+                if (flags & SPLICE_F_NONBLOCK) {
+                        if (!ret)
+                                ret = -EAGAIN;
+                        break;
+                }
+                if (signal_pending(current)) {
+                        if (!ret)
+                                ret = -ERESTARTSYS;
+                        break;
+                }
+
+                if (waitqueue_active(&ipipe->wait))
+                        wake_up_interruptible_sync(&ipipe->wait);
+                kill_fasync(&ipipe->fasync_writers, SIGIO, POLL_OUT);
+
+                pipe_wait(ipipe);
+        }
+
+        mutex_unlock(&ipipe->inode->i_mutex);
+        mutex_unlock(&opipe->inode->i_mutex);
+
+        if (do_wakeup) {
+                smp_mb();