Whamcloud - gitweb
2dc2f16032b4cbb5509c2edce07a7288c46061c7
[fs/lustre-release.git] / lustre / llite / rw26.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2011, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  *
32  * lustre/lustre/llite/rw26.c
33  *
34  * Lustre Lite I/O page cache routines for the 2.5/2.6 kernel version
35  */
36
37 #include <linux/buffer_head.h>
38 #include <linux/errno.h>
39 #include <linux/fs.h>
40 #include <linux/kernel.h>
41 #include <linux/mm.h>
42 #include <linux/mpage.h>
43 #include <linux/pagemap.h>
44 #include <linux/string.h>
45 #include <linux/unistd.h>
46 #include <linux/writeback.h>
47 #include <linux/migrate.h>
48
49 #define DEBUG_SUBSYSTEM S_LLITE
50
51 #include "llite_internal.h"
52 #include <lustre_compat.h>
53
54 /**
55  * Implements Linux VM address_space::invalidatepage() method. This method is
56  * called when the page is truncate from a file, either as a result of
57  * explicit truncate, or when inode is removed from memory (as a result of
58  * final iput(), umount, or memory pressure induced icache shrinking).
59  *
60  * [0, offset] bytes of the page remain valid (this is for a case of not-page
61  * aligned truncate). Lustre leaves partially truncated page in the cache,
62  * relying on struct inode::i_size to limit further accesses.
63  */
64 static void ll_invalidatepage(struct page *vmpage,
65 #ifdef HAVE_INVALIDATE_RANGE
66                                 unsigned int offset, unsigned int length
67 #else
68                                 unsigned long offset
69 #endif
70                              )
71 {
72         struct inode     *inode;
73         struct lu_env    *env;
74         struct cl_page   *page;
75         struct cl_object *obj;
76
77         LASSERT(PageLocked(vmpage));
78         LASSERT(!PageWriteback(vmpage));
79
80         /*
81          * It is safe to not check anything in invalidatepage/releasepage
82          * below because they are run with page locked and all our io is
83          * happening with locked page too
84          */
85 #ifdef HAVE_INVALIDATE_RANGE
86         if (offset == 0 && length == PAGE_SIZE) {
87 #else
88         if (offset == 0) {
89 #endif
90                 /* See the comment in ll_releasepage() */
91                 env = cl_env_percpu_get();
92                 LASSERT(!IS_ERR(env));
93
94                 inode = vmpage->mapping->host;
95                 obj = ll_i2info(inode)->lli_clob;
96                 if (obj != NULL) {
97                         page = cl_vmpage_page(vmpage, obj);
98                         if (page != NULL) {
99                                 cl_page_delete(env, page);
100                                 cl_page_put(env, page);
101                         }
102                 } else
103                         LASSERT(vmpage->private == 0);
104
105                 cl_env_percpu_put(env);
106         }
107 }
108
109 #ifdef HAVE_RELEASEPAGE_WITH_INT
110 #define RELEASEPAGE_ARG_TYPE int
111 #else
112 #define RELEASEPAGE_ARG_TYPE gfp_t
113 #endif
114 static int ll_releasepage(struct page *vmpage, RELEASEPAGE_ARG_TYPE gfp_mask)
115 {
116         struct lu_env           *env;
117         struct cl_object        *obj;
118         struct cl_page          *page;
119         struct address_space    *mapping;
120         int result = 0;
121
122         LASSERT(PageLocked(vmpage));
123         if (PageWriteback(vmpage) || PageDirty(vmpage))
124                 return 0;
125
126         mapping = vmpage->mapping;
127         if (mapping == NULL)
128                 return 1;
129
130         obj = ll_i2info(mapping->host)->lli_clob;
131         if (obj == NULL)
132                 return 1;
133
134         page = cl_vmpage_page(vmpage, obj);
135         if (page == NULL)
136                 return 1;
137
138         env = cl_env_percpu_get();
139         LASSERT(!IS_ERR(env));
140
141         if (!cl_page_in_use(page)) {
142                 result = 1;
143                 cl_page_delete(env, page);
144         }
145
146         /* To use percpu env array, the call path can not be rescheduled;
147          * otherwise percpu array will be messed if ll_releaspage() called
148          * again on the same CPU.
149          *
150          * If this page holds the last refc of cl_object, the following
151          * call path may cause reschedule:
152          *   cl_page_put -> cl_page_free -> cl_object_put ->
153          *     lu_object_put -> lu_object_free -> lov_delete_raid0.
154          *
155          * However, the kernel can't get rid of this inode until all pages have
156          * been cleaned up. Now that we hold page lock here, it's pretty safe
157          * that we won't get into object delete path.
158          */
159         LASSERT(cl_object_refc(obj) > 1);
160         cl_page_put(env, page);
161
162         cl_env_percpu_put(env);
163         return result;
164 }
165
166 #if defined(HAVE_DIRECTIO_ITER) || defined(HAVE_IOV_ITER_RW) || \
167         defined(HAVE_DIRECTIO_2ARGS)
168 #define HAVE_DIO_ITER 1
169 #endif
170
171 /*
172  * ll_free_user_pages - tear down page struct array
173  * @pages: array of page struct pointers underlying target buffer
174  */
175 static void ll_free_user_pages(struct page **pages, int npages)
176 {
177         int i;
178
179         for (i = 0; i < npages; i++) {
180                 if (!pages[i])
181                         break;
182                 put_page(pages[i]);
183         }
184
185 #if defined(HAVE_DIO_ITER)
186         kvfree(pages);
187 #else
188         OBD_FREE_PTR_ARRAY_LARGE(pages, npages);
189 #endif
190 }
191
192 static ssize_t ll_get_user_pages(int rw, struct iov_iter *iter,
193                                 struct page ***pages, ssize_t *npages,
194                                 size_t maxsize)
195 {
196 #if defined(HAVE_DIO_ITER)
197         size_t start;
198         size_t result;
199
200         /*
201          * iov_iter_get_pages_alloc() is introduced in 3.16 similar
202          * to HAVE_DIO_ITER.
203          */
204         result = iov_iter_get_pages_alloc(iter, pages, maxsize, &start);
205         if (result > 0)
206                 *npages = DIV_ROUND_UP(result + start, PAGE_SIZE);
207
208         return result;
209 #else
210         unsigned long addr;
211         size_t page_count;
212         size_t size;
213         long result;
214
215         if (!maxsize)
216                 return 0;
217
218         if (!iter->nr_segs)
219                 return 0;
220
221         addr = (unsigned long)iter->iov->iov_base + iter->iov_offset;
222         if (addr & ~PAGE_MASK)
223                 return -EINVAL;
224
225         size = min_t(size_t, maxsize, iter->iov->iov_len);
226         page_count = (size + PAGE_SIZE - 1) >> PAGE_SHIFT;
227         OBD_ALLOC_PTR_ARRAY_LARGE(*pages, page_count);
228         if (*pages == NULL)
229                 return -ENOMEM;
230
231         down_read(&current->mm->mmap_sem);
232         result = get_user_pages(current, current->mm, addr, page_count,
233                                 rw == READ, 0, *pages, NULL);
234         up_read(&current->mm->mmap_sem);
235
236         if (unlikely(result != page_count)) {
237                 ll_free_user_pages(*pages, page_count);
238                 *pages = NULL;
239
240                 if (result >= 0)
241                         return -EFAULT;
242
243                 return result;
244         }
245         *npages = page_count;
246
247         return size;
248 #endif
249 }
250
251 /* iov_iter_alignment() is introduced in 3.16 similar to HAVE_DIO_ITER */
252 #if defined(HAVE_DIO_ITER)
253 static unsigned long ll_iov_iter_alignment(const struct iov_iter *i)
254 {
255         return iov_iter_alignment(i);
256 }
257 #else /* copied from alignment_iovec() */
258 static unsigned long ll_iov_iter_alignment(const struct iov_iter *i)
259 {
260         const struct iovec *iov = i->iov;
261         unsigned long res;
262         size_t size = i->count;
263         size_t n;
264
265         if (!size)
266                 return 0;
267
268         res = (unsigned long)iov->iov_base + i->iov_offset;
269         n = iov->iov_len - i->iov_offset;
270         if (n >= size)
271                 return res | size;
272
273         size -= n;
274         res |= n;
275         while (size > (++iov)->iov_len) {
276                 res |= (unsigned long)iov->iov_base | iov->iov_len;
277                 size -= iov->iov_len;
278         }
279         res |= (unsigned long)iov->iov_base | size;
280
281         return res;
282 }
283 #endif
284
285 /** direct IO pages */
286 struct ll_dio_pages {
287         struct cl_dio_aio       *ldp_aio;
288         /*
289          * page array to be written. we don't support
290          * partial pages except the last one.
291          */
292         struct page             **ldp_pages;
293         /** # of pages in the array. */
294         size_t                  ldp_count;
295         /* the file offset of the first page. */
296         loff_t                  ldp_file_offset;
297 };
298
299 static int
300 ll_direct_rw_pages(const struct lu_env *env, struct cl_io *io, size_t size,
301                    int rw, struct inode *inode, struct ll_dio_pages *pv)
302 {
303         struct cl_page    *page;
304         struct cl_2queue  *queue = &io->ci_queue;
305         struct cl_object  *obj = io->ci_obj;
306         struct cl_sync_io *anchor = &pv->ldp_aio->cda_sync;
307         loff_t offset   = pv->ldp_file_offset;
308         int io_pages    = 0;
309         size_t page_size = cl_page_size(obj);
310         int i;
311         ssize_t rc = 0;
312
313         ENTRY;
314
315         cl_2queue_init(queue);
316         for (i = 0; i < pv->ldp_count; i++) {
317                 LASSERT(!(offset & (PAGE_SIZE - 1)));
318                 page = cl_page_find(env, obj, cl_index(obj, offset),
319                                     pv->ldp_pages[i], CPT_TRANSIENT);
320                 if (IS_ERR(page)) {
321                         rc = PTR_ERR(page);
322                         break;
323                 }
324                 LASSERT(page->cp_type == CPT_TRANSIENT);
325                 rc = cl_page_own(env, io, page);
326                 if (rc) {
327                         cl_page_put(env, page);
328                         break;
329                 }
330
331                 page->cp_sync_io = anchor;
332                 cl_2queue_add(queue, page);
333                 /*
334                  * Set page clip to tell transfer formation engine
335                  * that page has to be sent even if it is beyond KMS.
336                  */
337                 cl_page_clip(env, page, 0, min(size, page_size));
338                 ++io_pages;
339
340                 /* drop the reference count for cl_page_find */
341                 cl_page_put(env, page);
342                 offset += page_size;
343                 size -= page_size;
344         }
345         if (rc == 0 && io_pages > 0) {
346                 int iot = rw == READ ? CRT_READ : CRT_WRITE;
347
348                 atomic_add(io_pages, &anchor->csi_sync_nr);
349                 rc = cl_io_submit_rw(env, io, iot, queue);
350                 if (rc == 0) {
351                         cl_page_list_splice(&queue->c2_qout,
352                                         &pv->ldp_aio->cda_pages);
353                 } else {
354                         atomic_add(-queue->c2_qin.pl_nr,
355                                    &anchor->csi_sync_nr);
356                         cl_page_list_for_each(page, &queue->c2_qin)
357                                 page->cp_sync_io = NULL;
358                 }
359                 /* handle partially submitted reqs */
360                 if (queue->c2_qin.pl_nr > 0) {
361                         CERROR(DFID " failed to submit %d dio pages: %zd\n",
362                                PFID(lu_object_fid(&obj->co_lu)),
363                                queue->c2_qin.pl_nr, rc);
364                         if (rc == 0)
365                                 rc = -EIO;
366                 }
367         }
368
369         cl_2queue_discard(env, io, queue);
370         cl_2queue_disown(env, io, queue);
371         cl_2queue_fini(env, queue);
372         RETURN(rc);
373 }
374
375 #ifdef KMALLOC_MAX_SIZE
376 #define MAX_MALLOC KMALLOC_MAX_SIZE
377 #else
378 #define MAX_MALLOC (128 * 1024)
379 #endif
380
381 /* This is the maximum size of a single O_DIRECT request, based on the
382  * kmalloc limit.  We need to fit all of the brw_page structs, each one
383  * representing PAGE_SIZE worth of user data, into a single buffer, and
384  * then truncate this to be a full-sized RPC.  For 4kB PAGE_SIZE this is
385  * up to 22MB for 128kB kmalloc and up to 682MB for 4MB kmalloc. */
386 #define MAX_DIO_SIZE ((MAX_MALLOC / sizeof(struct brw_page) * PAGE_SIZE) & \
387                       ~(DT_MAX_BRW_SIZE - 1))
388
389 static ssize_t
390 ll_direct_IO_impl(struct kiocb *iocb, struct iov_iter *iter, int rw)
391 {
392         struct ll_cl_context *lcc;
393         const struct lu_env *env;
394         struct cl_io *io;
395         struct file *file = iocb->ki_filp;
396         struct inode *inode = file->f_mapping->host;
397         struct cl_dio_aio *aio;
398         size_t count = iov_iter_count(iter);
399         ssize_t tot_bytes = 0, result = 0;
400         loff_t file_offset = iocb->ki_pos;
401
402         /* Check EOF by ourselves */
403         if (rw == READ && file_offset >= i_size_read(inode))
404                 return 0;
405
406         /* FIXME: io smaller than PAGE_SIZE is broken on ia64 ??? */
407         if ((file_offset & ~PAGE_MASK) || (count & ~PAGE_MASK))
408                 return -EINVAL;
409
410         CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p), size=%zd (max %lu), "
411                "offset=%lld=%llx, pages %zd (max %lu)\n",
412                PFID(ll_inode2fid(inode)), inode, count, MAX_DIO_SIZE,
413                file_offset, file_offset, count >> PAGE_SHIFT,
414                MAX_DIO_SIZE >> PAGE_SHIFT);
415
416         /* Check that all user buffers are aligned as well */
417         if (ll_iov_iter_alignment(iter) & ~PAGE_MASK)
418                 return -EINVAL;
419
420         lcc = ll_cl_find(file);
421         if (lcc == NULL)
422                 RETURN(-EIO);
423
424         env = lcc->lcc_env;
425         LASSERT(!IS_ERR(env));
426         io = lcc->lcc_io;
427         LASSERT(io != NULL);
428
429         aio = cl_aio_alloc(iocb);
430         if (!aio)
431                 RETURN(-ENOMEM);
432
433         /* 0. Need locking between buffered and direct access. and race with
434          *    size changing by concurrent truncates and writes.
435          * 1. Need inode mutex to operate transient pages.
436          */
437         if (rw == READ)
438                 inode_lock(inode);
439
440         while (iov_iter_count(iter)) {
441                 struct ll_dio_pages pvec = { .ldp_aio = aio };
442                 struct page **pages;
443
444                 count = min_t(size_t, iov_iter_count(iter), MAX_DIO_SIZE);
445                 if (rw == READ) {
446                         if (file_offset >= i_size_read(inode))
447                                 break;
448
449                         if (file_offset + count > i_size_read(inode))
450                                 count = i_size_read(inode) - file_offset;
451                 }
452
453                 result = ll_get_user_pages(rw, iter, &pages,
454                                            &pvec.ldp_count, count);
455                 if (unlikely(result <= 0))
456                         GOTO(out, result);
457
458                 count = result;
459                 pvec.ldp_file_offset = file_offset;
460                 pvec.ldp_pages = pages;
461
462                 result = ll_direct_rw_pages(env, io, count,
463                                             rw, inode, &pvec);
464                 ll_free_user_pages(pages, pvec.ldp_count);
465
466                 if (unlikely(result < 0))
467                         GOTO(out, result);
468
469                 iov_iter_advance(iter, count);
470                 tot_bytes += count;
471                 file_offset += count;
472         }
473
474 out:
475         aio->cda_bytes = tot_bytes;
476         cl_sync_io_note(env, &aio->cda_sync, result);
477
478         if (is_sync_kiocb(iocb)) {
479                 ssize_t rc2;
480
481                 rc2 = cl_sync_io_wait(env, &aio->cda_sync, 0);
482                 if (result == 0 && rc2)
483                         result = rc2;
484
485                 if (result == 0) {
486                         struct vvp_io *vio = vvp_env_io(env);
487                         /* no commit async for direct IO */
488                         vio->u.write.vui_written += tot_bytes;
489                         result = tot_bytes;
490                 }
491                 OBD_FREE_PTR(aio);
492
493         } else {
494                 result = -EIOCBQUEUED;
495         }
496
497         if (rw == READ)
498                 inode_unlock(inode);
499
500         return result;
501 }
502
503 #if defined(HAVE_DIO_ITER)
504 static ssize_t ll_direct_IO(
505 #ifndef HAVE_IOV_ITER_RW
506              int rw,
507 #endif
508              struct kiocb *iocb, struct iov_iter *iter
509 #ifndef HAVE_DIRECTIO_2ARGS
510              , loff_t file_offset
511 #endif
512              )
513 {
514         int nrw;
515
516 #ifndef HAVE_IOV_ITER_RW
517         nrw = rw;
518 #else
519         nrw = iov_iter_rw(iter);
520 #endif
521
522         return ll_direct_IO_impl(iocb, iter, nrw);
523 }
524
525 #else /* !defined(HAVE_DIO_ITER) */
526
527 static ssize_t
528 ll_direct_IO(int rw, struct kiocb *iocb, const struct iovec *iov,
529              loff_t file_offset, unsigned long nr_segs)
530 {
531         struct iov_iter iter;
532
533         iov_iter_init(&iter, iov, nr_segs, iov_length(iov, nr_segs), 0);
534         return ll_direct_IO_impl(iocb, &iter, rw);
535 }
536
537 #endif /* !defined(HAVE_DIO_ITER) */
538
539 /**
540  * Prepare partially written-to page for a write.
541  * @pg is owned when passed in and disowned when it returns non-zero result to
542  * the caller.
543  */
544 static int ll_prepare_partial_page(const struct lu_env *env, struct cl_io *io,
545                                    struct cl_page *pg, struct file *file)
546 {
547         struct cl_attr *attr   = vvp_env_thread_attr(env);
548         struct cl_object *obj  = io->ci_obj;
549         struct vvp_page *vpg   = cl_object_page_slice(obj, pg);
550         loff_t          offset = cl_offset(obj, vvp_index(vpg));
551         int             result;
552         ENTRY;
553
554         cl_object_attr_lock(obj);
555         result = cl_object_attr_get(env, obj, attr);
556         cl_object_attr_unlock(obj);
557         if (result) {
558                 cl_page_disown(env, io, pg);
559                 GOTO(out, result);
560         }
561
562         /*
563          * If are writing to a new page, no need to read old data.
564          * The extent locking will have updated the KMS, and for our
565          * purposes here we can treat it like i_size.
566          */
567         if (attr->cat_kms <= offset) {
568                 char *kaddr = kmap_atomic(vpg->vpg_page);
569
570                 memset(kaddr, 0, cl_page_size(obj));
571                 kunmap_atomic(kaddr);
572                 GOTO(out, result = 0);
573         }
574
575         if (vpg->vpg_defer_uptodate) {
576                 vpg->vpg_ra_used = 1;
577                 GOTO(out, result = 0);
578         }
579
580         result = ll_io_read_page(env, io, pg, file);
581         if (result)
582                 GOTO(out, result);
583
584         /* ll_io_read_page() disowns the page */
585         result = cl_page_own(env, io, pg);
586         if (!result) {
587                 if (!PageUptodate(cl_page_vmpage(pg))) {
588                         cl_page_disown(env, io, pg);
589                         result = -EIO;
590                 }
591         } else if (result == -ENOENT) {
592                 /* page was truncated */
593                 result = -EAGAIN;
594         }
595         EXIT;
596
597 out:
598         return result;
599 }
600
601 static int ll_tiny_write_begin(struct page *vmpage, struct address_space *mapping)
602 {
603         /* Page must be present, up to date, dirty, and not in writeback. */
604         if (!vmpage || !PageUptodate(vmpage) || !PageDirty(vmpage) ||
605             PageWriteback(vmpage) || vmpage->mapping != mapping)
606                 return -ENODATA;
607
608         return 0;
609 }
610
611 static int ll_write_begin(struct file *file, struct address_space *mapping,
612                           loff_t pos, unsigned len, unsigned flags,
613                           struct page **pagep, void **fsdata)
614 {
615         struct ll_cl_context *lcc = NULL;
616         const struct lu_env  *env = NULL;
617         struct cl_io   *io = NULL;
618         struct cl_page *page = NULL;
619
620         struct cl_object *clob = ll_i2info(mapping->host)->lli_clob;
621         pgoff_t index = pos >> PAGE_SHIFT;
622         struct page *vmpage = NULL;
623         unsigned from = pos & (PAGE_SIZE - 1);
624         unsigned to = from + len;
625         int result = 0;
626         ENTRY;
627
628         CDEBUG(D_VFSTRACE, "Writing %lu of %d to %d bytes\n", index, from, len);
629
630         lcc = ll_cl_find(file);
631         if (lcc == NULL) {
632                 vmpage = grab_cache_page_nowait(mapping, index);
633                 result = ll_tiny_write_begin(vmpage, mapping);
634                 GOTO(out, result);
635         }
636
637         env = lcc->lcc_env;
638         io  = lcc->lcc_io;
639
640         if (file->f_flags & O_DIRECT) {
641                 /* direct IO failed because it couldn't clean up cached pages,
642                  * this causes a problem for mirror write because the cached
643                  * page may belong to another mirror, which will result in
644                  * problem submitting the I/O. */
645                 if (io->ci_designated_mirror > 0)
646                         GOTO(out, result = -EBUSY);
647
648                 /**
649                  * Direct read can fall back to buffered read, but DIO is done
650                  * with lockless i/o, and buffered requires LDLM locking, so
651                  * in this case we must restart without lockless.
652                  */
653                 if (!io->ci_ignore_lockless) {
654                         io->ci_ignore_lockless = 1;
655                         io->ci_need_restart = 1;
656                         GOTO(out, result = -ENOLCK);
657                 }
658         }
659 again:
660         /* To avoid deadlock, try to lock page first. */
661         vmpage = grab_cache_page_nowait(mapping, index);
662
663         if (unlikely(vmpage == NULL ||
664                      PageDirty(vmpage) || PageWriteback(vmpage))) {
665                 struct vvp_io *vio = vvp_env_io(env);
666                 struct cl_page_list *plist = &vio->u.write.vui_queue;
667
668                 /* if the page is already in dirty cache, we have to commit
669                  * the pages right now; otherwise, it may cause deadlock
670                  * because it holds page lock of a dirty page and request for
671                  * more grants. It's okay for the dirty page to be the first
672                  * one in commit page list, though. */
673                 if (vmpage != NULL && plist->pl_nr > 0) {
674                         unlock_page(vmpage);
675                         put_page(vmpage);
676                         vmpage = NULL;
677                 }
678
679                 /* commit pages and then wait for page lock */
680                 result = vvp_io_write_commit(env, io);
681                 if (result < 0)
682                         GOTO(out, result);
683
684                 if (vmpage == NULL) {
685                         vmpage = grab_cache_page_write_begin(mapping, index,
686                                                              flags);
687                         if (vmpage == NULL)
688                                 GOTO(out, result = -ENOMEM);
689                 }
690         }
691
692         /* page was truncated */
693         if (mapping != vmpage->mapping) {
694                 CDEBUG(D_VFSTRACE, "page: %lu was truncated\n", index);
695                 unlock_page(vmpage);
696                 put_page(vmpage);
697                 vmpage = NULL;
698                 goto again;
699         }
700
701         page = cl_page_find(env, clob, vmpage->index, vmpage, CPT_CACHEABLE);
702         if (IS_ERR(page))
703                 GOTO(out, result = PTR_ERR(page));
704
705         lcc->lcc_page = page;
706         lu_ref_add(&page->cp_reference, "cl_io", io);
707
708         cl_page_assume(env, io, page);
709         if (!PageUptodate(vmpage)) {
710                 /*
711                  * We're completely overwriting an existing page,
712                  * so _don't_ set it up to date until commit_write
713                  */
714                 if (from == 0 && to == PAGE_SIZE) {
715                         CL_PAGE_HEADER(D_PAGE, env, page, "full page write\n");
716                         POISON_PAGE(vmpage, 0x11);
717                 } else {
718                         /* TODO: can be optimized at OSC layer to check if it
719                          * is a lockless IO. In that case, it's not necessary
720                          * to read the data. */
721                         result = ll_prepare_partial_page(env, io, page, file);
722                         if (result) {
723                                 /* vmpage should have been unlocked */
724                                 put_page(vmpage);
725                                 vmpage = NULL;
726
727                                 if (result == -EAGAIN)
728                                         goto again;
729                                 GOTO(out, result);
730                         }
731                 }
732         }
733         EXIT;
734 out:
735         if (result < 0) {
736                 if (vmpage != NULL) {
737                         unlock_page(vmpage);
738                         put_page(vmpage);
739                 }
740                 /* On tiny_write failure, page and io are always null. */
741                 if (!IS_ERR_OR_NULL(page)) {
742                         lu_ref_del(&page->cp_reference, "cl_io", io);
743                         cl_page_put(env, page);
744                 }
745                 if (io)
746                         io->ci_result = result;
747         } else {
748                 *pagep = vmpage;
749                 *fsdata = lcc;
750         }
751         RETURN(result);
752 }
753
754 static int ll_tiny_write_end(struct file *file, struct address_space *mapping,
755                              loff_t pos, unsigned int len, unsigned int copied,
756                              struct page *vmpage)
757 {
758         struct cl_page *clpage = (struct cl_page *) vmpage->private;
759         loff_t kms = pos+copied;
760         loff_t to = kms & (PAGE_SIZE-1) ? kms & (PAGE_SIZE-1) : PAGE_SIZE;
761         __u16 refcheck;
762         struct lu_env *env = cl_env_get(&refcheck);
763         int rc = 0;
764
765         ENTRY;
766
767         if (IS_ERR(env)) {
768                 rc = PTR_ERR(env);
769                 goto out;
770         }
771
772         /* This page is dirty in cache, so it should have a cl_page pointer
773          * set in vmpage->private.
774          */
775         LASSERT(clpage != NULL);
776
777         if (copied == 0)
778                 goto out_env;
779
780         /* Update the underlying size information in the OSC/LOV objects this
781          * page is part of.
782          */
783         cl_page_touch(env, clpage, to);
784
785 out_env:
786         cl_env_put(env, &refcheck);
787
788 out:
789         /* Must return page unlocked. */
790         unlock_page(vmpage);
791
792         RETURN(rc);
793 }
794
795 static int ll_write_end(struct file *file, struct address_space *mapping,
796                         loff_t pos, unsigned len, unsigned copied,
797                         struct page *vmpage, void *fsdata)
798 {
799         struct ll_cl_context *lcc = fsdata;
800         const struct lu_env *env;
801         struct cl_io *io;
802         struct vvp_io *vio;
803         struct cl_page *page;
804         unsigned from = pos & (PAGE_SIZE - 1);
805         bool unplug = false;
806         int result = 0;
807         ENTRY;
808
809         put_page(vmpage);
810
811         CDEBUG(D_VFSTRACE, "pos %llu, len %u, copied %u\n", pos, len, copied);
812
813         if (lcc == NULL) {
814                 result = ll_tiny_write_end(file, mapping, pos, len, copied,
815                                            vmpage);
816                 GOTO(out, result);
817         }
818
819         LASSERT(lcc != NULL);
820         env  = lcc->lcc_env;
821         page = lcc->lcc_page;
822         io   = lcc->lcc_io;
823         vio  = vvp_env_io(env);
824
825         LASSERT(cl_page_is_owned(page, io));
826         if (copied > 0) {
827                 struct cl_page_list *plist = &vio->u.write.vui_queue;
828
829                 lcc->lcc_page = NULL; /* page will be queued */
830
831                 /* Add it into write queue */
832                 cl_page_list_add(plist, page);
833                 if (plist->pl_nr == 1) /* first page */
834                         vio->u.write.vui_from = from;
835                 else
836                         LASSERT(from == 0);
837                 vio->u.write.vui_to = from + copied;
838
839                 /* To address the deadlock in balance_dirty_pages() where
840                  * this dirty page may be written back in the same thread. */
841                 if (PageDirty(vmpage))
842                         unplug = true;
843
844                 /* We may have one full RPC, commit it soon */
845                 if (plist->pl_nr >= PTLRPC_MAX_BRW_PAGES)
846                         unplug = true;
847
848                 CL_PAGE_DEBUG(D_VFSTRACE, env, page,
849                               "queued page: %d.\n", plist->pl_nr);
850         } else {
851                 cl_page_disown(env, io, page);
852
853                 lcc->lcc_page = NULL;
854                 lu_ref_del(&page->cp_reference, "cl_io", io);
855                 cl_page_put(env, page);
856
857                 /* page list is not contiguous now, commit it now */
858                 unplug = true;
859         }
860         if (unplug || io->u.ci_wr.wr_sync)
861                 result = vvp_io_write_commit(env, io);
862
863         if (result < 0)
864                 io->ci_result = result;
865
866
867 out:
868         RETURN(result >= 0 ? copied : result);
869 }
870
871 #ifdef CONFIG_MIGRATION
872 static int ll_migratepage(struct address_space *mapping,
873                           struct page *newpage, struct page *page,
874                           enum migrate_mode mode)
875 {
876         /* Always fail page migration until we have a proper implementation */
877         return -EIO;
878 }
879 #endif
880
881 const struct address_space_operations ll_aops = {
882         .readpage       = ll_readpage,
883         .direct_IO      = ll_direct_IO,
884         .writepage      = ll_writepage,
885         .writepages     = ll_writepages,
886         .set_page_dirty = __set_page_dirty_nobuffers,
887         .write_begin    = ll_write_begin,
888         .write_end      = ll_write_end,
889         .invalidatepage = ll_invalidatepage,
890         .releasepage    = (void *)ll_releasepage,
891 #ifdef CONFIG_MIGRATION
892         .migratepage    = ll_migratepage,
893 #endif
894 };