[PATCH] splice: LRU fixups
[linux-2.6.git] / 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 *info,
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 *info,
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 *info,
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 *info,
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(info, 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 *info, 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(info, 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(info, 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(info, 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(info, 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 }