git.oblomov.eu Git - linux-2.6/blob - fs/splice.c

   1 /*
   2  * "splice": joining two ropes together by interweaving their strands.
   3  *
   4  * This is the "extended pipe" functionality, where a pipe is used as
   5  * an arbitrary in-memory buffer. Think of a pipe as a small kernel
   6  * buffer that you can use to transfer data from one end to the other.
   7  *
   8  * The traditional unix read/write is extended with a "splice()" operation
   9  * that transfers data buffers to or from a pipe buffer.
  10  *
  11  * Named by Larry McVoy, original implementation from Linus, extended by
  12  * Jens to support splicing to files, network, direct splicing, etc and
  13  * fixing lots of bugs.
  14  *
  15  * Copyright (C) 2005-2006 Jens Axboe <axboe@suse.de>
  16  * Copyright (C) 2005-2006 Linus Torvalds <torvalds@osdl.org>
  17  * Copyright (C) 2006 Ingo Molnar <mingo@elte.hu>
  18  *
  19  */
  20 #include <linux/fs.h>
  21 #include <linux/file.h>
  22 #include <linux/pagemap.h>
  23 #include <linux/pipe_fs_i.h>
  24 #include <linux/mm_inline.h>
  25 #include <linux/swap.h>
  26 #include <linux/writeback.h>
  27 #include <linux/buffer_head.h>
  28 #include <linux/module.h>
  29 #include <linux/syscalls.h>
  30 #include <linux/uio.h>
  31
  32 struct partial_page {
  33         unsigned int offset;
  34         unsigned int len;
  35 };
  36
  37 /*
  38  * Passed to splice_to_pipe
  39  */
  40 struct splice_pipe_desc {
  41         struct page **pages;            /* page map */
  42         struct partial_page *partial;   /* pages[] may not be contig */
  43         int nr_pages;                   /* number of pages in map */
  44         unsigned int flags;             /* splice flags */
  45         struct pipe_buf_operations *ops;/* ops associated with output pipe */
  46 };
  47
  48 /*
  49  * Attempt to steal a page from a pipe buffer. This should perhaps go into
  50  * a vm helper function, it's already simplified quite a bit by the
  51  * addition of remove_mapping(). If success is returned, the caller may
  52  * attempt to reuse this page for another destination.
  53  */
  54 static int page_cache_pipe_buf_steal(struct pipe_inode_info *pipe,
  55                                      struct pipe_buffer *buf)
  56 {
  57         struct page *page = buf->page;
  58         struct address_space *mapping = page_mapping(page);
  59
  60         lock_page(page);
  61
  62         WARN_ON(!PageUptodate(page));
  63
  64         /*
  65          * At least for ext2 with nobh option, we need to wait on writeback
  66          * completing on this page, since we'll remove it from the pagecache.
  67          * Otherwise truncate wont wait on the page, allowing the disk
  68          * blocks to be reused by someone else before we actually wrote our
  69          * data to them. fs corruption ensues.
  70          */
  71         wait_on_page_writeback(page);
  72
  73         if (PagePrivate(page))
  74                 try_to_release_page(page, mapping_gfp_mask(mapping));
  75
  76         if (!remove_mapping(mapping, page)) {
  77                 unlock_page(page);
  78                 return 1;
  79         }
  80
  81         buf->flags |= PIPE_BUF_FLAG_LRU;
  82         return 0;
  83 }
  84
  85 static void page_cache_pipe_buf_release(struct pipe_inode_info *pipe,
  86                                         struct pipe_buffer *buf)
  87 {
  88         page_cache_release(buf->page);
  89         buf->flags &= ~PIPE_BUF_FLAG_LRU;
  90 }
  91
  92 static int page_cache_pipe_buf_pin(struct pipe_inode_info *pipe,
  93                                    struct pipe_buffer *buf)
  94 {
  95         struct page *page = buf->page;
  96         int err;
  97
  98         if (!PageUptodate(page)) {
  99                 lock_page(page);
 100
 101                 /*
 102                  * Page got truncated/unhashed. This will cause a 0-byte
 103                  * splice, if this is the first page.
 104                  */
 105                 if (!page->mapping) {
 106                         err = -ENODATA;
 107                         goto error;
 108                 }
 109
 110                 /*
 111                  * Uh oh, read-error from disk.
 112                  */
 113                 if (!PageUptodate(page)) {
 114                         err = -EIO;
 115                         goto error;
 116                 }
 117
 118                 /*
 119                  * Page is ok afterall, we are done.
 120                  */
 121                 unlock_page(page);
 122         }
 123
 124         return 0;
 125 error:
 126         unlock_page(page);
 127         return err;
 128 }
 129
 130 static struct pipe_buf_operations page_cache_pipe_buf_ops = {
 131         .can_merge = 0,
 132         .map = generic_pipe_buf_map,
 133         .unmap = generic_pipe_buf_unmap,
 134         .pin = page_cache_pipe_buf_pin,
 135         .release = page_cache_pipe_buf_release,
 136         .steal = page_cache_pipe_buf_steal,
 137         .get = generic_pipe_buf_get,
 138 };
 139
 140 static int user_page_pipe_buf_steal(struct pipe_inode_info *pipe,
 141                                     struct pipe_buffer *buf)
 142 {
 143         if (!(buf->flags & PIPE_BUF_FLAG_GIFT))
 144                 return 1;
 145
 146         buf->flags |= PIPE_BUF_FLAG_LRU;
 147         return generic_pipe_buf_steal(pipe, buf);
 148 }
 149
 150 static struct pipe_buf_operations user_page_pipe_buf_ops = {
 151         .can_merge = 0,
 152         .map = generic_pipe_buf_map,
 153         .unmap = generic_pipe_buf_unmap,
 154         .pin = generic_pipe_buf_pin,
 155         .release = page_cache_pipe_buf_release,
 156         .steal = user_page_pipe_buf_steal,
 157         .get = generic_pipe_buf_get,
 158 };
 159
 160 /*
 161  * Pipe output worker. This sets up our pipe format with the page cache
 162  * pipe buffer operations. Otherwise very similar to the regular pipe_writev().
 163  */
 164 static ssize_t splice_to_pipe(struct pipe_inode_info *pipe,
 165                               struct splice_pipe_desc *spd)
 166 {
 167         int ret, do_wakeup, page_nr;
 168
 169         ret = 0;
 170         do_wakeup = 0;
 171         page_nr = 0;
 172
 173         if (pipe->inode)
 174                 mutex_lock(&pipe->inode->i_mutex);
 175
 176         for (;;) {
 177                 if (!pipe->readers) {
 178                         send_sig(SIGPIPE, current, 0);
 179                         if (!ret)
 180                                 ret = -EPIPE;
 181                         break;
 182                 }
 183
 184                 if (pipe->nrbufs < PIPE_BUFFERS) {
 185                         int newbuf = (pipe->curbuf + pipe->nrbufs) & (PIPE_BUFFERS - 1);
 186                         struct pipe_buffer *buf = pipe->bufs + newbuf;
 187
 188                         buf->page = spd->pages[page_nr];
 189                         buf->offset = spd->partial[page_nr].offset;
 190                         buf->len = spd->partial[page_nr].len;
 191                         buf->ops = spd->ops;
 192                         if (spd->flags & SPLICE_F_GIFT)
 193                                 buf->flags |= PIPE_BUF_FLAG_GIFT;
 194
 195                         pipe->nrbufs++;
 196                         page_nr++;
 197                         ret += buf->len;
 198
 199                         if (pipe->inode)
 200                                 do_wakeup = 1;
 201
 202                         if (!--spd->nr_pages)
 203                                 break;
 204                         if (pipe->nrbufs < PIPE_BUFFERS)
 205                                 continue;
 206
 207                         break;
 208                 }
 209
 210                 if (spd->flags & SPLICE_F_NONBLOCK) {
 211                         if (!ret)
 212                                 ret = -EAGAIN;
 213                         break;
 214                 }
 215
 216                 if (signal_pending(current)) {
 217                         if (!ret)
 218                                 ret = -ERESTARTSYS;
 219                         break;
 220                 }
 221
 222                 if (do_wakeup) {
 223                         smp_mb();
 224                         if (waitqueue_active(&pipe->wait))
 225                                 wake_up_interruptible_sync(&pipe->wait);
 226                         kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
 227                         do_wakeup = 0;
 228                 }
 229
 230                 pipe->waiting_writers++;
 231                 pipe_wait(pipe);
 232                 pipe->waiting_writers--;
 233         }
 234
 235         if (pipe->inode)
 236                 mutex_unlock(&pipe->inode->i_mutex);
 237
 238         if (do_wakeup) {
 239                 smp_mb();
 240                 if (waitqueue_active(&pipe->wait))
 241                         wake_up_interruptible(&pipe->wait);
 242                 kill_fasync(&pipe->fasync_readers, SIGIO, POLL_IN);
 243         }
 244
 245         while (page_nr < spd->nr_pages)
 246                 page_cache_release(spd->pages[page_nr++]);
 247
 248         return ret;
 249 }
 250
 251 static int
 252 __generic_file_splice_read(struct file *in, loff_t *ppos,
 253                            struct pipe_inode_info *pipe, size_t len,
 254                            unsigned int flags)
 255 {
 256         struct address_space *mapping = in->f_mapping;
 257         unsigned int loff, nr_pages;
 258         struct page *pages[PIPE_BUFFERS];
 259         struct partial_page partial[PIPE_BUFFERS];
 260         struct page *page;
 261         pgoff_t index, end_index;
 262         loff_t isize;
 263         size_t total_len;
 264         int error, page_nr;
 265         struct splice_pipe_desc spd = {
 266                 .pages = pages,
 267                 .partial = partial,
 268                 .flags = flags,
 269                 .ops = &page_cache_pipe_buf_ops,
 270         };
 271
 272         index = *ppos >> PAGE_CACHE_SHIFT;
 273         loff = *ppos & ~PAGE_CACHE_MASK;
 274         nr_pages = (len + loff + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT;
 275
 276         if (nr_pages > PIPE_BUFFERS)
 277                 nr_pages = PIPE_BUFFERS;
 278
 279         /*
 280          * Initiate read-ahead on this page range. however, don't call into
 281          * read-ahead if this is a non-zero offset (we are likely doing small
 282          * chunk splice and the page is already there) for a single page.
 283          */
 284         if (!loff || nr_pages > 1)
 285                 page_cache_readahead(mapping, &in->f_ra, in, index, nr_pages);
 286
 287         /*
 288          * Now fill in the holes:
 289          */
 290         error = 0;
 291         total_len = 0;
 292
 293         /*
 294          * Lookup the (hopefully) full range of pages we need.
 295          */
 296         spd.nr_pages = find_get_pages_contig(mapping, index, nr_pages, pages);
 297
 298         /*
 299          * If find_get_pages_contig() returned fewer pages than we needed,
 300          * allocate the rest.
 301          */
 302         index += spd.nr_pages;
 303         while (spd.nr_pages < nr_pages) {
 304                 /*
 305                  * Page could be there, find_get_pages_contig() breaks on
 306                  * the first hole.
 307                  */
 308                 page = find_get_page(mapping, index);
 309                 if (!page) {
 310                         /*
 311                          * Make sure the read-ahead engine is notified
 312                          * about this failure.
 313                          */
 314                         handle_ra_miss(mapping, &in->f_ra, index);
 315
 316                         /*
 317                          * page didn't exist, allocate one.
 318                          */
 319                         page = page_cache_alloc_cold(mapping);
 320                         if (!page)
 321                                 break;
 322
 323                         error = add_to_page_cache_lru(page, mapping, index,
 324                                               mapping_gfp_mask(mapping));
 325                         if (unlikely(error)) {
 326                                 page_cache_release(page);
 327                                 break;
 328                         }
 329                         /*
 330                          * add_to_page_cache() locks the page, unlock it
 331                          * to avoid convoluting the logic below even more.
 332                          */
 333                         unlock_page(page);
 334                 }
 335
 336                 pages[spd.nr_pages++] = page;
 337                 index++;
 338         }
 339
 340         /*
 341          * Now loop over the map and see if we need to start IO on any
 342          * pages, fill in the partial map, etc.
 343          */
 344         index = *ppos >> PAGE_CACHE_SHIFT;
 345         nr_pages = spd.nr_pages;
 346         spd.nr_pages = 0;
 347         for (page_nr = 0; page_nr < nr_pages; page_nr++) {
 348                 unsigned int this_len;
 349
 350                 if (!len)
 351                         break;
 352
 353                 /*
 354                  * this_len is the max we'll use from this page
 355                  */
 356                 this_len = min_t(unsigned long, len, PAGE_CACHE_SIZE - loff);
 357                 page = pages[page_nr];
 358
 359                 /*
 360                  * If the page isn't uptodate, we may need to start io on it
 361                  */
 362                 if (!PageUptodate(page)) {
 363                         /*
 364                          * If in nonblock mode then dont block on waiting
 365                          * for an in-flight io page
 366                          */
 367                         if (flags & SPLICE_F_NONBLOCK)
 368                                 break;
 369
 370                         lock_page(page);
 371
 372                         /*
 373                          * page was truncated, stop here. if this isn't the
 374                          * first page, we'll just complete what we already
 375                          * added
 376                          */
 377                         if (!page->mapping) {
 378                                 unlock_page(page);
 379                                 break;
 380                         }
 381                         /*
 382                          * page was already under io and is now done, great
 383                          */
 384                         if (PageUptodate(page)) {
 385                                 unlock_page(page);
 386                                 goto fill_it;
 387                         }
 388
 389                         /*
 390                          * need to read in the page
 391                          */
 392                         error = mapping->a_ops->readpage(in, page);
 393                         if (unlikely(error)) {
 394                                 /*
 395                                  * We really should re-lookup the page here,
 396                                  * but it complicates things a lot. Instead
 397                                  * lets just do what we already stored, and
 398                                  * we'll get it the next time we are called.
 399                                  */
 400                                 if (error == AOP_TRUNCATED_PAGE)
 401                                         error = 0;
 402
 403                                 break;
 404                         }
 405
 406                         /*
 407                          * i_size must be checked after ->readpage().
 408                          */
 409                         isize = i_size_read(mapping->host);
 410                         end_index = (isize - 1) >> PAGE_CACHE_SHIFT;
 411                         if (unlikely(!isize || index > end_index))
 412                                 break;
 413
 414                         /*
 415                          * if this is the last page, see if we need to shrink
 416                          * the length and stop
 417                          */
 418                         if (end_index == index) {
 419                                 loff = PAGE_CACHE_SIZE - (isize & ~PAGE_CACHE_MASK);
 420                                 if (total_len + loff > isize)
 421                                         break;
 422                                 /*
 423                                  * force quit after adding this page
 424                                  */
 425                                 len = this_len;
 426                                 this_len = min(this_len, loff);
 427                                 loff = 0;
 428                         }
 429                 }
 430 fill_it:
 431                 partial[page_nr].offset = loff;
 432                 partial[page_nr].len = this_len;
 433                 len -= this_len;
 434                 total_len += this_len;
 435                 loff = 0;
 436                 spd.nr_pages++;
 437                 index++;
 438         }
 439
 440         /*
 441          * Release any pages at the end, if we quit early. 'i' is how far
 442          * we got, 'nr_pages' is how many pages are in the map.
 443          */
 444         while (page_nr < nr_pages)
 445                 page_cache_release(pages[page_nr++]);
 446
 447         if (spd.nr_pages)
 448                 return splice_to_pipe(pipe, &spd);
 449
 450         return error;
 451 }
 452
 453 /**
 454  * generic_file_splice_read - splice data from file to a pipe
 455  * @in:         file to splice from
 456  * @pipe:       pipe to splice to
 457  * @len:        number of bytes to splice
 458  * @flags:      splice modifier flags
 459  *
 460  * Will read pages from given file and fill them into a pipe.
 461  */
 462 ssize_t generic_file_splice_read(struct file *in, loff_t *ppos,
 463                                  struct pipe_inode_info *pipe, size_t len,
 464                                  unsigned int flags)
 465 {
 466         ssize_t spliced;
 467         int ret;
 468
 469         ret = 0;
 470         spliced = 0;
 471
 472         while (len) {
 473                 ret = __generic_file_splice_read(in, ppos, pipe, len, flags);
 474
 475                 if (ret < 0)
 476                         break;
 477                 else if (!ret) {
 478                         if (spliced)
 479                                 break;
 480                         if (flags & SPLICE_F_NONBLOCK) {
 481                                 ret = -EAGAIN;
 482                                 break;
 483                         }
 484                 }
 485
 486                 *ppos += ret;
 487                 len -= ret;
 488                 spliced += ret;
 489         }
 490
 491         if (spliced)
 492                 return spliced;
 493
 494         return ret;
 495 }
 496
 497 EXPORT_SYMBOL(generic_file_splice_read);
 498
 499 /*
 500  * Send 'sd->len' bytes to socket from 'sd->file' at position 'sd->pos'
 501  * using sendpage(). Return the number of bytes sent.
 502  */
 503 static int pipe_to_sendpage(struct pipe_inode_info *pipe,
 504                             struct pipe_buffer *buf, struct splice_desc *sd)
 505 {
 506         struct file *file = sd->file;
 507         loff_t pos = sd->pos;
 508         int ret, more;
 509
 510         ret = buf->ops->pin(pipe, buf);
 511         if (!ret) {
 512                 more = (sd->flags & SPLICE_F_MORE) || sd->len < sd->total_len;
 513
 514                 ret = file->f_op->sendpage(file, buf->page, buf->offset,
 515                                            sd->len, &pos, more);
 516         }
 517
 518         return ret;
 519 }
 520
 521 /*
 522  * This is a little more tricky than the file -> pipe splicing. There are
 523  * basically three cases:
 524  *
 525  *      - Destination page already exists in the address space and there
 526  *        are users of it. For that case we have no other option that
 527  *        copying the data. Tough luck.
 528  *      - Destination page already exists in the address space, but there
 529  *        are no users of it. Make sure it's uptodate, then drop it. Fall
 530  *        through to last case.
 531  *      - Destination page does not exist, we can add the pipe page to
 532  *        the page cache and avoid the copy.
 533  *
 534  * If asked to move pages to the output file (SPLICE_F_MOVE is set in
 535  * sd->flags), we attempt to migrate pages from the pipe to the output
 536  * file address space page cache. This is possible if no one else has
 537  * the pipe page referenced outside of the pipe and page cache. If
 538  * SPLICE_F_MOVE isn't set, or we cannot move the page, we simply create
 539  * a new page in the output file page cache and fill/dirty that.
 540  */
 541 static int pipe_to_file(struct pipe_inode_info *pipe, struct pipe_buffer *buf,
 542                         struct splice_desc *sd)
 543 {
 544         struct file *file = sd->file;
 545         struct address_space *mapping = file->f_mapping;
 546         gfp_t gfp_mask = mapping_gfp_mask(mapping);
 547         unsigned int offset, this_len;
 548         struct page *page;
 549         pgoff_t index;
 550         int ret;
 551
 552         /*
 553          * make sure the data in this buffer is uptodate
 554          */
 555         ret = buf->ops->pin(pipe, buf);
 556         if (unlikely(ret))
 557                 return ret;
 558
 559         index = sd->pos >> PAGE_CACHE_SHIFT;
 560         offset = sd->pos & ~PAGE_CACHE_MASK;
 561
 562         this_len = sd->len;
 563         if (this_len + offset > PAGE_CACHE_SIZE)
 564                 this_len = PAGE_CACHE_SIZE - offset;
 565
 566         /*
 567          * Reuse buf page, if SPLICE_F_MOVE is set and we are doing a full
 568          * page.
 569          */
 570         if ((sd->flags & SPLICE_F_MOVE) && this_len == PAGE_CACHE_SIZE) {
 571                 /*
 572                  * If steal succeeds, buf->page is now pruned from the
 573                  * pagecache and we can reuse it. The page will also be
 574                  * locked on successful return.
 575                  */
 576                 if (buf->ops->steal(pipe, buf))
 577                         goto find_page;
 578
 579                 page = buf->page;
 580                 if (add_to_page_cache(page, mapping, index, gfp_mask)) {
 581                         unlock_page(page);
 582                         goto find_page;
 583                 }
 584
 585                 page_cache_get(page);
 586
 587                 if (!(buf->flags & PIPE_BUF_FLAG_LRU))
 588                         lru_cache_add(page);
 589         } else {
 590 find_page:
 591                 page = find_lock_page(mapping, index);
 592                 if (!page) {
 593                         ret = -ENOMEM;
 594                         page = page_cache_alloc_cold(mapping);
 595                         if (unlikely(!page))
 596                                 goto out_nomem;
 597
 598                         /*
 599                          * This will also lock the page
 600                          */
 601                         ret = add_to_page_cache_lru(page, mapping, index,
 602                                                     gfp_mask);
 603                         if (unlikely(ret))
 604                                 goto out;
 605                 }
 606
 607                 /*
 608                  * We get here with the page locked. If the page is also
 609                  * uptodate, we don't need to do more. If it isn't, we
 610                  * may need to bring it in if we are not going to overwrite
 611                  * the full page.
 612                  */
 613                 if (!PageUptodate(page)) {
 614                         if (this_len < PAGE_CACHE_SIZE) {
 615                                 ret = mapping->a_ops->readpage(file, page);
 616                                 if (unlikely(ret))
 617                                         goto out;
 618
 619                                 lock_page(page);
 620
 621                                 if (!PageUptodate(page)) {
 622                                         /*
 623                                          * Page got invalidated, repeat.
 624                                          */
 625                                         if (!page->mapping) {
 626                                                 unlock_page(page);
 627                                                 page_cache_release(page);
 628                                                 goto find_page;
 629                                         }
 630                                         ret = -EIO;
 631                                         goto out;
 632                                 }
 633                         } else
 634                                 SetPageUptodate(page);
 635                 }
 636         }
 637
 638         ret = mapping->a_ops->prepare_write(file, page, offset, offset+this_len);
 639         if (unlikely(ret)) {
 640                 loff_t isize = i_size_read(mapping->host);
 641
 642                 if (ret != AOP_TRUNCATED_PAGE)
 643                         unlock_page(page);
 644                 page_cache_release(page);
 645                 if (ret == AOP_TRUNCATED_PAGE)
 646                         goto find_page;
 647
 648                 /*
 649                  * prepare_write() may have instantiated a few blocks
 650                  * outside i_size.  Trim these off again.
 651                  */
 652                 if (sd->pos + this_len > isize)
 653                         vmtruncate(mapping->host, isize);
 654
 655                 goto out;
 656         }
 657
 658         if (buf->page != page) {
 659                 /*
 660                  * Careful, ->map() uses KM_USER0!
 661                  */
 662                 char *src = buf->ops->map(pipe, buf, 1);
 663                 char *dst = kmap_atomic(page, KM_USER1);
 664
 665                 memcpy(dst + offset, src + buf->offset, this_len);
 666                 flush_dcache_page(page);
 667                 kunmap_atomic(dst, KM_USER1);
 668                 buf->ops->unmap(pipe, buf, src);
 669         }
 670
 671         ret = mapping->a_ops->commit_write(file, page, offset, offset+this_len);
 672         if (!ret) {
 673                 /*
 674                  * Return the number of bytes written and mark page as
 675                  * accessed, we are now done!
 676                  */
 677                 ret = this_len;
 678                 mark_page_accessed(page);
 679                 balance_dirty_pages_ratelimited(mapping);
 680         } else if (ret == AOP_TRUNCATED_PAGE) {
 681                 page_cache_release(page);
 682                 goto find_page;
 683         }
 684 out:
 685         page_cache_release(page);
 686         unlock_page(page);
 687 out_nomem:
 688         return ret;
 689 }
 690
 691 /*
 692  * Pipe input worker. Most of this logic works like a regular pipe, the
 693  * key here is the 'actor' worker passed in that actually moves the data
 694  * to the wanted destination. See pipe_to_file/pipe_to_sendpage above.
 695  */
 696 ssize_t splice_from_pipe(struct pipe_inode_info *pipe, struct file *out,
 697                          loff_t *ppos, size_t len, unsigned int flags,
 698                          splice_actor *actor)
 699 {
 700         int ret, do_wakeup, err;
 701         struct splice_desc sd;
 702
 703         ret = 0;
 704         do_wakeup = 0;
 705
 706         sd.total_len = len;
 707         sd.flags = flags;
 708         sd.file = out;
 709         sd.pos = *ppos;
 710
 711         if (pipe->inode)
 712                 mutex_lock(&pipe->inode->i_mutex);
 713
 714         for (;;) {
 715                 if (pipe->nrbufs) {
 716                         struct pipe_buffer *buf = pipe->bufs + pipe->curbuf;
 717                         struct pipe_buf_operations *ops = buf->ops;
 718
 719                         sd.len = buf->len;
 720                         if (sd.len > sd.total_len)
 721                                 sd.len = sd.total_len;
 722
 723                         err = actor(pipe, buf, &sd);
 724                         if (err <= 0) {
 725                                 if (!ret && err != -ENODATA)
 726                                         ret = err;
 727
 728                                 break;
 729                         }
 730
 731                         ret += err;
 732                         buf->offset += err;
 733                         buf->len -= err;
 734
 735                         sd.len -= err;
 736                         sd.pos += err;
 737                         sd.total_len -= err;
 738                         if (sd.len)
 739                                 continue;
 740
 741                         if (!buf->len) {
 742                                 buf->ops = NULL;
 743                                 ops->release(pipe, buf);
 744                                 pipe->curbuf = (pipe->curbuf + 1) & (PIPE_BUFFERS - 1);
 745                                 pipe->nrbufs--;
 746                                 if (pipe->inode)
 747                                         do_wakeup = 1;
 748                         }
 749
 750                         if (!sd.total_len)
 751                                 break;
 752                 }
 753
 754                 if (pipe->nrbufs)
 755                         continue;
 756                 if (!pipe->writers)
 757                         break;
 758                 if (!pipe->waiting_writers) {
 759                         if (ret)
 760                                 break;
 761                 }
 762
 763                 if (flags & SPLICE_F_NONBLOCK) {
 764                         if (!ret)
 765                                 ret = -EAGAIN;
 766                         break;
 767                 }
 768
 769                 if (signal_pending(current)) {
 770                         if (!ret)
 771                                 ret = -ERESTARTSYS;
 772                         break;
 773                 }
 774
 775                 if (do_wakeup) {
 776                         smp_mb();
 777                         if (waitqueue_active(&pipe->wait))
 778                                 wake_up_interruptible_sync(&pipe->wait);
 779                         kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
 780                         do_wakeup = 0;
 781                 }
 782
 783                 pipe_wait(pipe);
 784         }
 785
 786         if (pipe->inode)
 787                 mutex_unlock(&pipe->inode->i_mutex);
 788
 789         if (do_wakeup) {
 790                 smp_mb();
 791                 if (waitqueue_active(&pipe->wait))
 792                         wake_up_interruptible(&pipe->wait);
 793                 kill_fasync(&pipe->fasync_writers, SIGIO, POLL_OUT);
 794         }
 795
 796         return ret;
 797 }
 798
 799 /**
 800  * generic_file_splice_write - splice data from a pipe to a file
 801  * @pipe:       pipe info
 802  * @out:        file to write to
 803  * @len:        number of bytes to splice
 804  * @flags:      splice modifier flags
 805  *
 806  * Will either move or copy pages (determined by @flags options) from
 807  * the given pipe inode to the given file.
 808  *
 809  */
 810 ssize_t
 811 generic_file_splice_write(struct pipe_inode_info *pipe, struct file *out,
 812                           loff_t *ppos, size_t len, unsigned int flags)
 813 {
 814         struct address_space *mapping = out->f_mapping;
 815         ssize_t ret;
 816
 817         ret = splice_from_pipe(pipe, out, ppos, len, flags, pipe_to_file);
 818         if (ret > 0) {
 819                 struct inode *inode = mapping->host;
 820
 821                 *ppos += ret;
 822
 823                 /*
 824                  * If file or inode is SYNC and we actually wrote some data,
 825                  * sync it.
 826                  */
 827                 if (unlikely((out->f_flags & O_SYNC) || IS_SYNC(inode))) {
 828                         int err;
 829
 830                         mutex_lock(&inode->i_mutex);
 831                         err = generic_osync_inode(inode, mapping,
 832                                                   OSYNC_METADATA|OSYNC_DATA);
 833                         mutex_unlock(&inode->i_mutex);
 834
 835                         if (err)
 836                                 ret = err;
 837                 }
 838         }
 839
 840         return ret;
 841 }
 842
 843 EXPORT_SYMBOL(generic_file_splice_write);
 844
 845 /**
 846  * generic_splice_sendpage - splice data from a pipe to a socket
 847  * @inode:      pipe inode
 848  * @out:        socket to write to
 849  * @len:        number of bytes to splice
 850  * @flags:      splice modifier flags
 851  *
 852  * Will send @len bytes from the pipe to a network socket. No data copying
 853  * is involved.
 854  *
 855  */
 856 ssize_t generic_splice_sendpage(struct pipe_inode_info *pipe, struct file *out,
 857                                 loff_t *ppos, size_t len, unsigned int flags)
 858 {
 859         return splice_from_pipe(pipe, out, ppos, len, flags, pipe_to_sendpage);
 860 }
 861
 862 EXPORT_SYMBOL(generic_splice_sendpage);
 863
 864 /*
 865  * Attempt to initiate a splice from pipe to file.
 866  */
 867 static long do_splice_from(struct pipe_inode_info *pipe, struct file *out,
 868                            loff_t *ppos, size_t len, unsigned int flags)
 869 {
 870         int ret;
 871
 872         if (unlikely(!out->f_op || !out->f_op->splice_write))
 873                 return -EINVAL;
 874
 875         if (unlikely(!(out->f_mode & FMODE_WRITE)))
 876                 return -EBADF;
 877
 878         ret = rw_verify_area(WRITE, out, ppos, len);
 879         if (unlikely(ret < 0))
 880                 return ret;
 881
 882         return out->f_op->splice_write(pipe, out, ppos, len, flags);
 883 }
 884
 885 /*
 886  * Attempt to initiate a splice from a file to a pipe.
 887  */
 888 static long do_splice_to(struct file *in, loff_t *ppos,
 889                          struct pipe_inode_info *pipe, size_t len,
 890                          unsigned int flags)
 891 {
 892         loff_t isize, left;
 893         int ret;
 894
 895         if (unlikely(!in->f_op || !in->f_op->splice_read))
 896                 return -EINVAL;
 897
 898         if (unlikely(!(in->f_mode & FMODE_READ)))
 899                 return -EBADF;
 900
 901         ret = rw_verify_area(READ, in, ppos, len);
 902         if (unlikely(ret < 0))
 903                 return ret;
 904
 905         isize = i_size_read(in->f_mapping->host);
 906         if (unlikely(*ppos >= isize))
 907                 return 0;
 908
 909         left = isize - *ppos;
 910         if (unlikely(left < len))
 911                 len = left;
 912
 913         return in->f_op->splice_read(in, ppos, pipe, len, flags);
 914 }
 915
 916 long do_splice_direct(struct file *in, loff_t *ppos, struct file *out,
 917                       size_t len, unsigned int flags)
 918 {
 919         struct pipe_inode_info *pipe;
 920         long ret, bytes;
 921         loff_t out_off;
 922         umode_t i_mode;
 923         int i;
 924
 925         /*
 926          * We require the input being a regular file, as we don't want to
 927          * randomly drop data for eg socket -> socket splicing. Use the
 928          * piped splicing for that!
 929          */
 930         i_mode = in->f_dentry->d_inode->i_mode;
 931         if (unlikely(!S_ISREG(i_mode) && !S_ISBLK(i_mode)))
 932                 return -EINVAL;
 933
 934         /*
 935          * neither in nor out is a pipe, setup an internal pipe attached to
 936          * 'out' and transfer the wanted data from 'in' to 'out' through that
 937          */
 938         pipe = current->splice_pipe;
 939         if (unlikely(!pipe)) {
 940                 pipe = alloc_pipe_info(NULL);
 941                 if (!pipe)
 942                         return -ENOMEM;
 943
 944                 /*
 945                  * We don't have an immediate reader, but we'll read the stuff
 946                  * out of the pipe right after the splice_to_pipe(). So set
 947                  * PIPE_READERS appropriately.
 948                  */
 949                 pipe->readers = 1;
 950
 951                 current->splice_pipe = pipe;
 952         }
 953
 954         /*
 955          * Do the splice.
 956          */
 957         ret = 0;
 958         bytes = 0;
 959         out_off = 0;
 960
 961         while (len) {
 962                 size_t read_len, max_read_len;
 963
 964                 /*
 965                  * Do at most PIPE_BUFFERS pages worth of transfer:
 966                  */
 967                 max_read_len = min(len, (size_t)(PIPE_BUFFERS*PAGE_SIZE));
 968
 969                 ret = do_splice_to(in, ppos, pipe, max_read_len, flags);
 970                 if (unlikely(ret < 0))
 971                         goto out_release;
 972
 973                 read_len = ret;
 974
 975                 /*
 976                  * NOTE: nonblocking mode only applies to the input. We
 977                  * must not do the output in nonblocking mode as then we
 978                  * could get stuck data in the internal pipe:
 979                  */
 980                 ret = do_splice_from(pipe, out, &out_off, read_len,
 981                                      flags & ~SPLICE_F_NONBLOCK);
 982                 if (unlikely(ret < 0))
 983                         goto out_release;
 984
 985                 bytes += ret;
 986                 len -= ret;
 987
 988                 /*
 989                  * In nonblocking mode, if we got back a short read then
 990                  * that was due to either an IO error or due to the
 991                  * pagecache entry not being there. In the IO error case
 992                  * the _next_ splice attempt will produce a clean IO error
 993                  * return value (not a short read), so in both cases it's
 994                  * correct to break out of the loop here:
 995                  */
 996                 if ((flags & SPLICE_F_NONBLOCK) && (read_len < max_read_len))
 997                         break;
 998         }
 999
1000         pipe->nrbufs = pipe->curbuf = 0;
1001
1002         return bytes;
1003
1004 out_release:
1005         /*
1006          * If we did an incomplete transfer we must release
1007          * the pipe buffers in question:
1008          */
1009         for (i = 0; i < PIPE_BUFFERS; i++) {
1010                 struct pipe_buffer *buf = pipe->bufs + i;
1011
1012                 if (buf->ops) {
1013                         buf->ops->release(pipe, buf);
1014                         buf->ops = NULL;
1015                 }
1016         }
1017         pipe->nrbufs = pipe->curbuf = 0;
1018
1019         /*
1020          * If we transferred some data, return the number of bytes:
1021          */
1022         if (bytes > 0)
1023                 return bytes;
1024
1025         return ret;
1026 }
1027
1028 EXPORT_SYMBOL(do_splice_direct);
1029
1030 /*
1031  * Determine where to splice to/from.
1032  */
1033 static long do_splice(struct file *in, loff_t __user *off_in,
1034                       struct file *out, loff_t __user *off_out,
1035                       size_t len, unsigned int flags)
1036 {
1037         struct pipe_inode_info *pipe;
1038         loff_t offset, *off;
1039         long ret;
1040
1041         pipe = in->f_dentry->d_inode->i_pipe;
1042         if (pipe) {
1043                 if (off_in)
1044                         return -ESPIPE;
1045                 if (off_out) {
1046                         if (out->f_op->llseek == no_llseek)
1047                                 return -EINVAL;
1048                         if (copy_from_user(&offset, off_out, sizeof(loff_t)))
1049                                 return -EFAULT;
1050                         off = &offset;
1051                 } else
1052                         off = &out->f_pos;
1053
1054                 ret = do_splice_from(pipe, out, off, len, flags);
1055
1056                 if (off_out && copy_to_user(off_out, off, sizeof(loff_t)))
1057                         ret = -EFAULT;
1058
1059                 return ret;
1060         }
1061
1062         pipe = out->f_dentry->d_inode->i_pipe;
1063         if (pipe) {
1064                 if (off_out)
1065                         return -ESPIPE;
1066                 if (off_in) {
1067                         if (in->f_op->llseek == no_llseek)
1068                                 return -EINVAL;
1069                         if (copy_from_user(&offset, off_in, sizeof(loff_t)))
1070                                 return -EFAULT;
1071                         off = &offset;
1072                 } else
1073                         off = &in->f_pos;
1074
1075                 ret = do_splice_to(in, off, pipe, len, flags);
1076
1077                 if (off_in && copy_to_user(off_in, off, sizeof(loff_t)))
1078                         ret = -EFAULT;
1079
1080                 return ret;
1081         }
1082
1083         return -EINVAL;
1084 }
1085
1086 /*
1087  * Map an iov into an array of pages and offset/length tupples. With the
1088  * partial_page structure, we can map several non-contiguous ranges into
1089  * our ones pages[] map instead of splitting that operation into pieces.
1090  * Could easily be exported as a generic helper for other users, in which
1091  * case one would probably want to add a 'max_nr_pages' parameter as well.
1092  */
1093 static int get_iovec_page_array(const struct iovec __user *iov,
1094                                 unsigned int nr_vecs, struct page **pages,
1095                                 struct partial_page *partial, int aligned)
1096 {
1097         int buffers = 0, error = 0;
1098
1099         /*
1100          * It's ok to take the mmap_sem for reading, even
1101          * across a "get_user()".
1102          */
1103         down_read(&current->mm->mmap_sem);
1104
1105         while (nr_vecs) {
1106                 unsigned long off, npages;
1107                 void __user *base;
1108                 size_t len;
1109                 int i;
1110
1111                 /*
1112                  * Get user address base and length for this iovec.
1113                  */
1114                 error = get_user(base, &iov->iov_base);
1115                 if (unlikely(error))
1116                         break;
1117                 error = get_user(len, &iov->iov_len);
1118                 if (unlikely(error))
1119                         break;
1120
1121                 /*
1122                  * Sanity check this iovec. 0 read succeeds.
1123                  */
1124                 if (unlikely(!len))
1125                         break;
1126                 error = -EFAULT;
1127                 if (unlikely(!base))
1128                         break;
1129
1130                 /*
1131                  * Get this base offset and number of pages, then map
1132                  * in the user pages.
1133                  */
1134                 off = (unsigned long) base & ~PAGE_MASK;
1135
1136                 /*
1137                  * If asked for alignment, the offset must be zero and the
1138                  * length a multiple of the PAGE_SIZE.
1139                  */
1140                 error = -EINVAL;
1141                 if (aligned && (off || len & ~PAGE_MASK))
1142                         break;
1143
1144                 npages = (off + len + PAGE_SIZE - 1) >> PAGE_SHIFT;
1145                 if (npages > PIPE_BUFFERS - buffers)
1146                         npages = PIPE_BUFFERS - buffers;
1147
1148                 error = get_user_pages(current, current->mm,
1149                                        (unsigned long) base, npages, 0, 0,
1150                                        &pages[buffers], NULL);
1151
1152                 if (unlikely(error <= 0))
1153                         break;
1154
1155                 /*
1156                  * Fill this contiguous range into the partial page map.
1157                  */
1158                 for (i = 0; i < error; i++) {
1159                         const int plen = min_t(size_t, len, PAGE_SIZE - off);
1160
1161                         partial[buffers].offset = off;
1162                         partial[buffers].len = plen;
1163
1164                         off = 0;
1165                         len -= plen;
1166                         buffers++;
1167                 }
1168
1169                 /*
1170                  * We didn't complete this iov, stop here since it probably
1171                  * means we have to move some of this into a pipe to
1172                  * be able to continue.
1173                  */
1174                 if (len)
1175                         break;
1176
1177                 /*
1178                  * Don't continue if we mapped fewer pages than we asked for,
1179                  * or if we mapped the max number of pages that we have
1180                  * room for.
1181                  */
1182                 if (error < npages || buffers == PIPE_BUFFERS)
1183                         break;
1184
1185                 nr_vecs--;
1186                 iov++;
1187         }
1188
1189         up_read(&current->mm->mmap_sem);
1190
1191         if (buffers)
1192                 return buffers;
1193
1194         return error;
1195 }
1196
1197 /*
1198  * vmsplice splices a user address range into a pipe. It can be thought of
1199  * as splice-from-memory, where the regular splice is splice-from-file (or
1200  * to file). In both cases the output is a pipe, naturally.
1201  *
1202  * Note that vmsplice only supports splicing _from_ user memory to a pipe,
1203  * not the other way around. Splicing from user memory is a simple operation
1204  * that can be supported without any funky alignment restrictions or nasty
1205  * vm tricks. We simply map in the user memory and fill them into a pipe.
1206  * The reverse isn't quite as easy, though. There are two possible solutions
1207  * for that:
1208  *
1209  *      - memcpy() the data internally, at which point we might as well just
1210  *        do a regular read() on the buffer anyway.
1211  *      - Lots of nasty vm tricks, that are neither fast nor flexible (it
1212  *        has restriction limitations on both ends of the pipe).
1213  *
1214  * Alas, it isn't here.
1215  *
1216  */
1217 static long do_vmsplice(struct file *file, const struct iovec __user *iov,
1218                         unsigned long nr_segs, unsigned int flags)
1219 {
1220         struct pipe_inode_info *pipe = file->f_dentry->d_inode->i_pipe;
1221         struct page *pages[PIPE_BUFFERS];
1222         struct partial_page partial[PIPE_BUFFERS];
1223         struct splice_pipe_desc spd = {
1224                 .pages = pages,
1225                 .partial = partial,
1226                 .flags = flags,
1227                 .ops = &user_page_pipe_buf_ops,
1228         };
1229
1230         if (unlikely(!pipe))
1231                 return -EBADF;
1232         if (unlikely(nr_segs > UIO_MAXIOV))
1233                 return -EINVAL;
1234         else if (unlikely(!nr_segs))
1235                 return 0;
1236
1237         spd.nr_pages = get_iovec_page_array(iov, nr_segs, pages, partial,
1238                                             flags & SPLICE_F_GIFT);
1239         if (spd.nr_pages <= 0)
1240                 return spd.nr_pages;
1241
1242         return splice_to_pipe(pipe, &spd);
1243 }
1244
1245 asmlinkage long sys_vmsplice(int fd, const struct iovec __user *iov,
1246                              unsigned long nr_segs, unsigned int flags)
1247 {
1248         struct file *file;
1249         long error;
1250         int fput;
1251
1252         error = -EBADF;
1253         file = fget_light(fd, &fput);
1254         if (file) {
1255                 if (file->f_mode & FMODE_WRITE)
1256                         error = do_vmsplice(file, iov, nr_segs, flags);
1257
1258                 fput_light(file, fput);
1259         }
1260
1261         return error;
1262 }
1263
1264 asmlinkage long sys_splice(int fd_in, loff_t __user *off_in,
1265                            int fd_out, loff_t __user *off_out,
1266                            size_t len, unsigned int flags)
1267 {
1268         long error;
1269         struct file *in, *out;
1270         int fput_in, fput_out;
1271
1272         if (unlikely(!len))
1273                 return 0;
1274
1275         error = -EBADF;
1276         in = fget_light(fd_in, &fput_in);
1277         if (in) {
1278                 if (in->f_mode & FMODE_READ) {
1279                         out = fget_light(fd_out, &fput_out);
1280                         if (out) {
1281                                 if (out->f_mode & FMODE_WRITE)
1282                                         error = do_splice(in, off_in,
1283                                                           out, off_out,
1284                                                           len, flags);
1285                                 fput_light(out, fput_out);
1286                         }
1287                 }
1288
1289                 fput_light(in, fput_in);
1290         }
1291
1292         return error;
1293 }
1294
1295 /*
1296  * Link contents of ipipe to opipe.
1297  */
1298 static int link_pipe(struct pipe_inode_info *ipipe,
1299                      struct pipe_inode_info *opipe,
1300                      size_t len, unsigned int flags)
1301 {
1302         struct pipe_buffer *ibuf, *obuf;
1303         int ret, do_wakeup, i, ipipe_first;
1304
1305         ret = do_wakeup = ipipe_first = 0;
1306
1307         /*
1308          * Potential ABBA deadlock, work around it by ordering lock
1309          * grabbing by inode address. Otherwise two different processes
1310          * could deadlock (one doing tee from A -> B, the other from B -> A).
1311          */
1312         if (ipipe->inode < opipe->inode) {
1313                 ipipe_first = 1;
1314                 mutex_lock(&ipipe->inode->i_mutex);
1315                 mutex_lock(&opipe->inode->i_mutex);
1316         } else {
1317                 mutex_lock(&opipe->inode->i_mutex);
1318                 mutex_lock(&ipipe->inode->i_mutex);
1319         }
1320
1321         for (i = 0;; i++) {
1322                 if (!opipe->readers) {
1323                         send_sig(SIGPIPE, current, 0);
1324                         if (!ret)
1325                                 ret = -EPIPE;
1326                         break;
1327                 }
1328                 if (ipipe->nrbufs - i) {
1329                         ibuf = ipipe->bufs + ((ipipe->curbuf + i) & (PIPE_BUFFERS - 1));
1330
1331                         /*
1332                          * If we have room, fill this buffer
1333                          */
1334                         if (opipe->nrbufs < PIPE_BUFFERS) {
1335                                 int nbuf = (opipe->curbuf + opipe->nrbufs) & (PIPE_BUFFERS - 1);
1336
1337                                 /*
1338                                  * Get a reference to this pipe buffer,
1339                                  * so we can copy the contents over.
1340                                  */
1341                                 ibuf->ops->get(ipipe, ibuf);
1342
1343                                 obuf = opipe->bufs + nbuf;
1344                                 *obuf = *ibuf;
1345
1346                                 /*
1347                                  * Don't inherit the gift flag, we need to
1348                                  * prevent multiple steals of this page.
1349                                  */
1350                                 obuf->flags &= ~PIPE_BUF_FLAG_GIFT;
1351
1352                                 if (obuf->len > len)
1353                                         obuf->len = len;
1354
1355                                 opipe->nrbufs++;
1356                                 do_wakeup = 1;
1357                                 ret += obuf->len;
1358                                 len -= obuf->len;
1359
1360                                 if (!len)
1361                                         break;
1362                                 if (opipe->nrbufs < PIPE_BUFFERS)
1363                                         continue;
1364                         }
1365
1366                         /*
1367                          * We have input available, but no output room.
1368                          * If we already copied data, return that. If we
1369                          * need to drop the opipe lock, it must be ordered
1370                          * last to avoid deadlocks.
1371                          */
1372                         if ((flags & SPLICE_F_NONBLOCK) || !ipipe_first) {
1373                                 if (!ret)
1374                                         ret = -EAGAIN;
1375                                 break;
1376                         }
1377                         if (signal_pending(current)) {
1378                                 if (!ret)
1379                                         ret = -ERESTARTSYS;
1380                                 break;
1381                         }
1382                         if (do_wakeup) {
1383                                 smp_mb();
1384                                 if (waitqueue_active(&opipe->wait))
1385                                         wake_up_interruptible(&opipe->wait);
1386                                 kill_fasync(&opipe->fasync_readers, SIGIO, POLL_IN);
1387                                 do_wakeup = 0;
1388                         }
1389
1390                         opipe->waiting_writers++;
1391                         pipe_wait(opipe);
1392                         opipe->waiting_writers--;
1393                         continue;
1394                 }
1395
1396                 /*
1397                  * No input buffers, do the usual checks for available
1398                  * writers and blocking and wait if necessary
1399                  */
1400                 if (!ipipe->writers)
1401                         break;
1402                 if (!ipipe->waiting_writers) {
1403                         if (ret)
1404                                 break;
1405                 }
1406                 /*
1407                  * pipe_wait() drops the ipipe mutex. To avoid deadlocks
1408                  * with another process, we can only safely do that if
1409                  * the ipipe lock is ordered last.
1410                  */
1411                 if ((flags & SPLICE_F_NONBLOCK) || ipipe_first) {
1412                         if (!ret)
1413                                 ret = -EAGAIN;
1414                         break;
1415                 }
1416                 if (signal_pending(current)) {
1417                         if (!ret)
1418                                 ret = -ERESTARTSYS;
1419                         break;
1420                 }
1421
1422                 if (waitqueue_active(&ipipe->wait))
1423                         wake_up_interruptible_sync(&ipipe->wait);
1424                 kill_fasync(&ipipe->fasync_writers, SIGIO, POLL_OUT);
1425
1426                 pipe_wait(ipipe);
1427         }
1428
1429         mutex_unlock(&ipipe->inode->i_mutex);
1430         mutex_unlock(&opipe->inode->i_mutex);
1431
1432         if (do_wakeup) {
1433                 smp_mb();
1434                 if (waitqueue_active(&opipe->wait))
1435                         wake_up_interruptible(&opipe->wait);
1436                 kill_fasync(&opipe->fasync_readers, SIGIO, POLL_IN);
1437         }
1438
1439         return ret;
1440 }
1441
1442 /*
1443  * This is a tee(1) implementation that works on pipes. It doesn't copy
1444  * any data, it simply references the 'in' pages on the 'out' pipe.
1445  * The 'flags' used are the SPLICE_F_* variants, currently the only
1446  * applicable one is SPLICE_F_NONBLOCK.
1447  */
1448 static long do_tee(struct file *in, struct file *out, size_t len,
1449                    unsigned int flags)
1450 {
1451         struct pipe_inode_info *ipipe = in->f_dentry->d_inode->i_pipe;
1452         struct pipe_inode_info *opipe = out->f_dentry->d_inode->i_pipe;
1453
1454         /*
1455          * Link ipipe to the two output pipes, consuming as we go along.
1456          */
1457         if (ipipe && opipe)
1458                 return link_pipe(ipipe, opipe, len, flags);
1459
1460         return -EINVAL;
1461 }
1462
1463 asmlinkage long sys_tee(int fdin, int fdout, size_t len, unsigned int flags)
1464 {
1465         struct file *in;
1466         int error, fput_in;
1467
1468         if (unlikely(!len))
1469                 return 0;
1470
1471         error = -EBADF;
1472         in = fget_light(fdin, &fput_in);
1473         if (in) {
1474                 if (in->f_mode & FMODE_READ) {
1475                         int fput_out;
1476                         struct file *out = fget_light(fdout, &fput_out);
1477
1478                         if (out) {
1479                                 if (out->f_mode & FMODE_WRITE)
1480                                         error = do_tee(in, out, len, flags);
1481                                 fput_light(out, fput_out);
1482                         }
1483                 }
1484                 fput_light(in, fput_in);
1485         }
1486
1487         return error;
1488 }