Whamcloud - gitweb
7adac1702a03316029fce53b79011f66115f41c8
[fs/lustre-release.git] / lustre / llite / vvp_io.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
30  * Use is subject to license terms.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * Implementation of cl_io for VVP layer.
37  *
38  *   Author: Nikita Danilov <nikita.danilov@sun.com>
39  */
40
41 #define DEBUG_SUBSYSTEM S_LLITE
42
43 #ifndef __KERNEL__
44 # error This file is kernel only.
45 #endif
46
47 #include <obd.h>
48 #include <lustre_lite.h>
49
50 #include "vvp_internal.h"
51
52 static struct vvp_io *cl2vvp_io(const struct lu_env *env,
53                                 const struct cl_io_slice *slice);
54
55 /*****************************************************************************
56  *
57  * io operations.
58  *
59  */
60
61 static int vvp_io_fault_iter_init(const struct lu_env *env,
62                                   const struct cl_io_slice *ios)
63 {
64         struct vvp_io *vio   = cl2vvp_io(env, ios);
65         struct inode  *inode = ccc_object_inode(ios->cis_obj);
66
67         LASSERT(inode ==
68                 cl2ccc_io(env, ios)->cui_fd->fd_file->f_dentry->d_inode);
69         vio->u.fault.ft_mtime = LTIME_S(inode->i_mtime);
70         return 0;
71 }
72
73 static void vvp_io_fini(const struct lu_env *env, const struct cl_io_slice *ios)
74 {
75         struct cl_io     *io  = ios->cis_io;
76         struct cl_object *obj = io->ci_obj;
77
78         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
79         if (io->ci_type == CIT_WRITE)
80                 up(&ll_i2info(ccc_object_inode(obj))->lli_write_sem);
81         else {
82                 struct vvp_io     *vio  = cl2vvp_io(env, ios);
83                 struct ccc_io     *cio  = cl2ccc_io(env, ios);
84
85                 if (vio->cui_ra_window_set)
86                         ll_ra_read_ex(cio->cui_fd->fd_file, &vio->cui_bead);
87         }
88
89 }
90
91 static void vvp_io_fault_fini(const struct lu_env *env,
92                               const struct cl_io_slice *ios)
93 {
94         struct cl_io   *io   = ios->cis_io;
95         struct cl_page *page = io->u.ci_fault.ft_page;
96
97         CLOBINVRNT(env, io->ci_obj, ccc_object_invariant(io->ci_obj));
98
99         if (page != NULL) {
100                 lu_ref_del(&page->cp_reference, "fault", io);
101                 cl_page_put(env, page);
102                 io->u.ci_fault.ft_page = NULL;
103         }
104         vvp_io_fini(env, ios);
105 }
106
107 enum cl_lock_mode vvp_mode_from_vma(struct vm_area_struct *vma)
108 {
109         /*
110          * we only want to hold PW locks if the mmap() can generate
111          * writes back to the file and that only happens in shared
112          * writable vmas
113          */
114         if ((vma->vm_flags & VM_SHARED) && (vma->vm_flags & VM_WRITE))
115                 return CLM_WRITE;
116         return CLM_READ;
117 }
118
119 static int vvp_mmap_locks(const struct lu_env *env,
120                           struct ccc_io *vio, struct cl_io *io)
121 {
122         struct ccc_thread_info *cti = ccc_env_info(env);
123         struct vm_area_struct  *vma;
124         struct cl_lock_descr   *descr = &cti->cti_descr;
125         ldlm_policy_data_t      policy;
126         struct inode           *inode;
127         unsigned long           addr;
128         unsigned long           seg;
129         ssize_t                 count;
130         int                     result;
131         ENTRY;
132
133         LASSERT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE);
134
135         if (cl_io_is_sendfile(io))
136                 RETURN(0);
137
138         for (seg = 0; seg < vio->cui_nrsegs; seg++) {
139                 const struct iovec *iv = &vio->cui_iov[seg];
140
141                 addr = (unsigned long)iv->iov_base;
142                 count = iv->iov_len;
143                 if (count == 0)
144                         continue;
145
146                 count += addr & (~CFS_PAGE_MASK);
147                 addr &= CFS_PAGE_MASK;
148                 while((vma = our_vma(addr, count)) != NULL) {
149                         struct file *file = vma->vm_file;
150                         struct ll_file_data *fd;
151
152                         LASSERT(file);
153                         fd = LUSTRE_FPRIVATE(file);
154
155                         inode = file->f_dentry->d_inode;
156                         if (!(fd->fd_flags & LL_FILE_IGNORE_LOCK ||
157                             ll_i2sbi(inode)->ll_flags & LL_SBI_NOLCK))
158                                 goto cont;
159
160                         /*
161                          * XXX: Required lock mode can be weakened: CIT_WRITE
162                          * io only ever reads user level buffer, and CIT_READ
163                          * only writes on it.
164                          */
165                         policy_from_vma(&policy, vma, addr, count);
166                         descr->cld_mode = vvp_mode_from_vma(vma);
167                         descr->cld_obj = ll_i2info(inode)->lli_clob;
168                         descr->cld_start = cl_index(descr->cld_obj,
169                                                     policy.l_extent.start);
170                         descr->cld_end = cl_index(descr->cld_obj,
171                                                   policy.l_extent.end);
172                         result = cl_io_lock_alloc_add(env, io, descr, CEF_MUST);
173                         if (result < 0)
174                                 RETURN(result);
175
176                 cont:
177                         if (vma->vm_end - addr >= count)
178                                 break;
179                         count -= vma->vm_end - addr;
180                         addr = vma->vm_end;
181                 }
182         }
183         RETURN(0);
184 }
185
186 static void vvp_io_update_iov(const struct lu_env *env,
187                               struct ccc_io *vio, struct cl_io *io)
188 {
189         int i;
190         size_t size = io->u.ci_rw.crw_count;
191
192         vio->cui_iov_olen = 0;
193         if (cl_io_is_sendfile(io) || size == vio->cui_tot_count)
194                 return;
195
196         if (vio->cui_tot_nrsegs == 0)
197                 vio->cui_tot_nrsegs =  vio->cui_nrsegs;
198
199         for (i = 0; i < vio->cui_tot_nrsegs; i++) {
200                 struct iovec *iv = &vio->cui_iov[i];
201
202                 if (iv->iov_len < size)
203                         size -= iv->iov_len;
204                 else {
205                         if (iv->iov_len > size) {
206                                 vio->cui_iov_olen = iv->iov_len;
207                                 iv->iov_len = size;
208                         }
209                         break;
210                 }
211         }
212
213         vio->cui_nrsegs = i + 1;
214 }
215
216 static int vvp_io_rw_lock(const struct lu_env *env, struct cl_io *io,
217                           enum cl_lock_mode mode, loff_t start, loff_t end)
218 {
219         struct ccc_io *cio = ccc_env_io(env);
220         int result;
221         int ast_flags = 0;
222
223         LASSERT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE);
224         LASSERT(vvp_env_io(env)->cui_oneshot == 0);
225         ENTRY;
226
227         vvp_io_update_iov(env, cio, io);
228
229         if (io->u.ci_rw.crw_nonblock)
230                 ast_flags |= CEF_NONBLOCK;
231         result = vvp_mmap_locks(env, cio, io);
232         if (result == 0)
233                 result = ccc_io_one_lock(env, io, ast_flags, mode, start, end);
234         RETURN(result);
235 }
236
237 static int vvp_io_read_lock(const struct lu_env *env,
238                             const struct cl_io_slice *ios)
239 {
240         struct cl_io         *io  = ios->cis_io;
241         struct ll_inode_info *lli = ll_i2info(ccc_object_inode(io->ci_obj));
242         int result;
243
244         ENTRY;
245         /* XXX: Layer violation, we shouldn't see lsm at llite level. */
246         if (lli->lli_smd != NULL) /* lsm-less file, don't need to lock */
247                 result = vvp_io_rw_lock(env, io, CLM_READ,
248                                         io->u.ci_rd.rd.crw_pos,
249                                         io->u.ci_rd.rd.crw_pos +
250                                         io->u.ci_rd.rd.crw_count - 1);
251         else
252                 result = 0;
253         RETURN(result);
254 }
255
256 static int vvp_io_fault_lock(const struct lu_env *env,
257                              const struct cl_io_slice *ios)
258 {
259         struct cl_io *io   = ios->cis_io;
260         struct vvp_io *vio = cl2vvp_io(env, ios);
261         /*
262          * XXX LDLM_FL_CBPENDING
263          */
264         return ccc_io_one_lock_index
265                 (env, io, 0, vvp_mode_from_vma(vio->u.fault.ft_vma),
266                  io->u.ci_fault.ft_index, io->u.ci_fault.ft_index);
267 }
268
269 static int vvp_io_write_lock(const struct lu_env *env,
270                              const struct cl_io_slice *ios)
271 {
272         struct cl_io *io = ios->cis_io;
273         loff_t start;
274         loff_t end;
275         int    result;
276
277         if (cl2vvp_io(env, ios)->cui_oneshot == 0) {
278                 if (io->u.ci_wr.wr_append) {
279                         start = 0;
280                         end   = OBD_OBJECT_EOF;
281                 } else {
282                         start = io->u.ci_wr.wr.crw_pos;
283                         end   = start + io->u.ci_wr.wr.crw_count - 1;
284                 }
285                 result = vvp_io_rw_lock(env, io, CLM_WRITE, start, end);
286         } else
287                 result = 0;
288         return result;
289 }
290
291 static int vvp_io_trunc_iter_init(const struct lu_env *env,
292                                   const struct cl_io_slice *ios)
293 {
294         struct ccc_io *vio   = cl2ccc_io(env, ios);
295         struct inode  *inode = ccc_object_inode(ios->cis_obj);
296
297         /*
298          * We really need to get our PW lock before we change inode->i_size.
299          * If we don't we can race with other i_size updaters on our node,
300          * like ll_file_read.  We can also race with i_size propogation to
301          * other nodes through dirtying and writeback of final cached pages.
302          * This last one is especially bad for racing o_append users on other
303          * nodes.
304          */
305
306         UNLOCK_INODE_MUTEX(inode);
307         UP_WRITE_I_ALLOC_SEM(inode);
308         vio->u.trunc.cui_locks_released = 1;
309         return 0;
310 }
311
312 /**
313  * Implementation of cl_io_operations::vio_lock() method for CIT_TRUNC io.
314  *
315  * Handles "lockless io" mode when extent locking is done by server.
316  */
317 static int vvp_io_trunc_lock(const struct lu_env *env,
318                              const struct cl_io_slice *ios)
319 {
320         struct ccc_io     *vio   = cl2ccc_io(env, ios);
321         struct cl_io      *io    = ios->cis_io;
322         size_t new_size          = io->u.ci_truncate.tr_size;
323         __u32 enqflags = new_size == 0 ? CEF_DISCARD_DATA : 0;
324         int result;
325
326         vio->u.trunc.cui_local_lock = TRUNC_EXTENT;
327         result = ccc_io_one_lock(env, io, enqflags, CLM_WRITE,
328                                  new_size, OBD_OBJECT_EOF);
329         return result;
330 }
331
332 static int vvp_io_trunc_start(const struct lu_env *env,
333                               const struct cl_io_slice *ios)
334 {
335         struct ccc_io        *cio   = cl2ccc_io(env, ios);
336         struct vvp_io        *vio   = cl2vvp_io(env, ios);
337         struct cl_io         *io    = ios->cis_io;
338         struct inode         *inode = ccc_object_inode(io->ci_obj);
339         struct cl_object     *obj   = ios->cis_obj;
340         size_t                size  = io->u.ci_truncate.tr_size;
341         pgoff_t               start = cl_index(obj, size);
342         int                   result;
343
344         LASSERT(cio->u.trunc.cui_locks_released);
345         LASSERT(vio->cui_oneshot == 0);
346
347         LOCK_INODE_MUTEX(inode);
348         DOWN_WRITE_I_ALLOC_SEM(inode);
349         cio->u.trunc.cui_locks_released = 0;
350
351         /*
352          * Only ll_inode_size_lock is taken at this level. lov_stripe_lock()
353          * is grabbed by ll_truncate() only over call to obd_adjust_kms().  If
354          * vmtruncate returns 0, then ll_truncate dropped ll_inode_size_lock()
355          */
356         ll_inode_size_lock(inode, 0);
357         result = vmtruncate(inode, size);
358         if (result != 0)
359                 ll_inode_size_unlock(inode, 0);
360         /*
361          * If a page is partially truncated, keep it owned across truncate to
362          * prevent... races.
363          *
364          * XXX this properly belongs to osc, because races in question are OST
365          * specific.
366          */
367         if (cl_offset(obj, start) != size) {
368                 struct cl_object_header *hdr;
369
370                 hdr = cl_object_header(obj);
371                 spin_lock(&hdr->coh_page_guard);
372                 vio->cui_partpage = cl_page_lookup(hdr, start);
373                 spin_unlock(&hdr->coh_page_guard);
374
375                 if (vio->cui_partpage != NULL)
376                         /*
377                          * Wait for the transfer completion for a partially
378                          * truncated page to avoid dead-locking an OST with
379                          * the concurrent page-wise overlapping WRITE and
380                          * PUNCH requests. BUG:17397.
381                          *
382                          * Partial page is disowned in vvp_io_trunc_end().
383                          */
384                         cl_page_own(env, io, vio->cui_partpage);
385         } else
386                 vio->cui_partpage = NULL;
387         return result;
388 }
389
390 static void vvp_io_trunc_end(const struct lu_env *env,
391                              const struct cl_io_slice *ios)
392 {
393         struct vvp_io *vio = cl2vvp_io(env, ios);
394
395         if (vio->cui_partpage != NULL) {
396                 cl_page_disown(env, ios->cis_io, vio->cui_partpage);
397                 cl_page_put(env, vio->cui_partpage);
398                 vio->cui_partpage = NULL;
399         }
400 }
401
402 static void vvp_io_trunc_fini(const struct lu_env *env,
403                               const struct cl_io_slice *ios)
404 {
405         struct ccc_io *cio   = ccc_env_io(env);
406         struct inode  *inode = ccc_object_inode(ios->cis_io->ci_obj);
407
408         if (cio->u.trunc.cui_locks_released) {
409                 LOCK_INODE_MUTEX(inode);
410                 DOWN_WRITE_I_ALLOC_SEM(inode);
411                 cio->u.trunc.cui_locks_released = 0;
412         }
413         vvp_io_fini(env, ios);
414 }
415
416 #ifdef HAVE_FILE_READV
417 static ssize_t lustre_generic_file_read(struct file *file,
418                                         struct ccc_io *vio, loff_t *ppos)
419 {
420         return generic_file_readv(file, vio->cui_iov, vio->cui_nrsegs, ppos);
421 }
422
423 static ssize_t lustre_generic_file_write(struct file *file,
424                                          struct ccc_io *vio, loff_t *ppos)
425 {
426         return generic_file_writev(file, vio->cui_iov, vio->cui_nrsegs, ppos);
427 }
428 #else
429 static ssize_t lustre_generic_file_read(struct file *file,
430                                         struct ccc_io *vio, loff_t *ppos)
431 {
432         return generic_file_aio_read(vio->cui_iocb, vio->cui_iov,
433                                      vio->cui_nrsegs, *ppos);
434 }
435
436 static ssize_t lustre_generic_file_write(struct file *file,
437                                         struct ccc_io *vio, loff_t *ppos)
438 {
439         return generic_file_aio_write(vio->cui_iocb, vio->cui_iov,
440                                       vio->cui_nrsegs, *ppos);
441 }
442 #endif
443
444 static int vvp_io_read_start(const struct lu_env *env,
445                              const struct cl_io_slice *ios)
446 {
447         struct vvp_io     *vio   = cl2vvp_io(env, ios);
448         struct ccc_io     *cio   = cl2ccc_io(env, ios);
449         struct cl_io      *io    = ios->cis_io;
450         struct cl_object  *obj   = io->ci_obj;
451         struct inode      *inode = ccc_object_inode(obj);
452         struct ll_ra_read *bead  = &vio->cui_bead;
453         struct file       *file  = cio->cui_fd->fd_file;
454
455         int     result;
456         loff_t  pos = io->u.ci_rd.rd.crw_pos;
457         size_t  cnt = io->u.ci_rd.rd.crw_count;
458         size_t  tot = cio->cui_tot_count;
459
460         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
461         LASSERT(vio->cui_oneshot == 0);
462
463         CDEBUG(D_VFSTRACE, "read: -> [%lli, %lli)\n", pos, pos + cnt);
464
465         result = ccc_prep_size(env, obj, io, pos + tot - 1, 1);
466         if (result != 0)
467                 return result;
468
469         LU_OBJECT_HEADER(D_INODE, env, &obj->co_lu,
470                         "Read ino %lu, "LPSZ" bytes, offset %lld, size %llu\n",
471                         inode->i_ino, cnt, pos, i_size_read(inode));
472
473         /* turn off the kernel's read-ahead */
474         cio->cui_fd->fd_file->f_ra.ra_pages = 0;
475
476         /* initialize read-ahead window once per syscall */
477         if (!vio->cui_ra_window_set) {
478                 vio->cui_ra_window_set = 1;
479                 bead->lrr_start = cl_index(obj, pos);
480                 /*
481                  * XXX: explicit CFS_PAGE_SIZE
482                  */
483                 bead->lrr_count = cl_index(obj, tot + CFS_PAGE_SIZE - 1);
484                 ll_ra_read_in(file, bead);
485         }
486
487         /* BUG: 5972 */
488         file_accessed(file);
489         if (cl_io_is_sendfile(io)) {
490                 result = generic_file_sendfile(file, &pos, cnt,
491                                 vio->u.read.cui_actor, vio->u.read.cui_target);
492         } else {
493                 result = lustre_generic_file_read(file, cio, &pos);
494         }
495
496         if (result >= 0) {
497                 if (result < cnt)
498                         io->ci_continue = 0;
499                 io->ci_nob += result;
500                 ll_rw_stats_tally(ll_i2sbi(inode), current->pid,
501                                   cio->cui_fd, pos, result, 0);
502                 result = 0;
503         }
504         return result;
505 }
506
507 static int vvp_io_write_start(const struct lu_env *env,
508                               const struct cl_io_slice *ios)
509 {
510         struct ccc_io      *cio   = cl2ccc_io(env, ios);
511         struct cl_io       *io    = ios->cis_io;
512         struct cl_object   *obj   = io->ci_obj;
513         struct inode       *inode = ccc_object_inode(obj);
514         struct file        *file  = cio->cui_fd->fd_file;
515         ssize_t result = 0;
516         loff_t pos = io->u.ci_wr.wr.crw_pos;
517         size_t cnt = io->u.ci_wr.wr.crw_count;
518
519         ENTRY;
520
521         if (cl_io_is_append(io))
522                 /*
523                  * PARALLEL IO This has to be changed for parallel IO doing
524                  * out-of-order writes.
525                  */
526                 pos = io->u.ci_wr.wr.crw_pos = i_size_read(inode);
527
528         CDEBUG(D_VFSTRACE, "write: [%lli, %lli)\n", pos, pos + cnt);
529
530         if (cl2vvp_io(env, ios)->cui_oneshot > 0)
531                 result = 0;
532         else
533                 result = lustre_generic_file_write(file, cio, &pos);
534
535         if (result > 0) {
536                 if (result < cnt)
537                         io->ci_continue = 0;
538                 io->ci_nob += result;
539                 ll_rw_stats_tally(ll_i2sbi(inode), current->pid,
540                                   cio->cui_fd, pos, result, 0);
541                 result = 0;
542         }
543         RETURN(result);
544 }
545
546 static int vvp_io_fault_start(const struct lu_env *env,
547                               const struct cl_io_slice *ios)
548 {
549         struct vvp_io       *vio     = cl2vvp_io(env, ios);
550         struct cl_io        *io      = ios->cis_io;
551         struct cl_object    *obj     = io->ci_obj;
552         struct inode        *inode   = ccc_object_inode(obj);
553         struct cl_fault_io  *fio     = &io->u.ci_fault;
554         struct vvp_fault_io *cfio    = &vio->u.fault;
555         cfs_page_t          *vmpage;
556         loff_t               offset;
557         int                  result  = 0;
558
559         LASSERT(vio->cui_oneshot == 0);
560
561         if (fio->ft_executable &&
562             LTIME_S(inode->i_mtime) != vio->u.fault.ft_mtime)
563                 CWARN("binary "DFID
564                       " changed while waiting for the page fault lock\n",
565                       PFID(lu_object_fid(&obj->co_lu)));
566
567         /* offset of the last byte on the page */
568         offset = cl_offset(obj, fio->ft_index + 1) - 1;
569         LASSERT(cl_index(obj, offset) == fio->ft_index);
570         result = ccc_prep_size(env, obj, io, offset, 0);
571         if (result != 0)
572                 return result;
573
574         vmpage = filemap_nopage(cfio->ft_vma, cfio->ft_address, cfio->ft_type);
575         if (vmpage != NOPAGE_SIGBUS && vmpage != NOPAGE_OOM)
576                 LL_CDEBUG_PAGE(D_PAGE, vmpage,
577                                "got addr %lu type %lx\n",
578                                cfio->ft_address, (long)cfio->ft_type);
579         else
580                 CDEBUG(D_PAGE, "got addr %lu type %lx - SIGBUS\n",
581                        cfio->ft_address, (long)cfio->ft_type);
582
583         if (vmpage == NOPAGE_SIGBUS)
584                 result = -EFAULT;
585         else if (vmpage == NOPAGE_OOM)
586                 result = -ENOMEM;
587         else {
588                 struct cl_page *page;
589                 loff_t          size;
590                 pgoff_t         last; /* last page in a file data region */
591
592                 /* Temporarily lock vmpage to keep cl_page_find() happy. */
593                 lock_page(vmpage);
594                 page = cl_page_find(env, obj, fio->ft_index, vmpage,
595                                     CPT_CACHEABLE);
596                 unlock_page(vmpage);
597                 if (!IS_ERR(page)) {
598                         size = i_size_read(inode);
599                         last = cl_index(obj, size - 1);
600                         if (fio->ft_index == last)
601                                 /*
602                                  * Last page is mapped partially.
603                                  */
604                                 fio->ft_nob = size - cl_offset(obj,
605                                                                fio->ft_index);
606                         else
607                                 fio->ft_nob = cl_page_size(obj);
608                         lu_ref_add(&page->cp_reference, "fault", io);
609                         fio->ft_page = page;
610                         /*
611                          * Certain 2.6 kernels return not-NULL from
612                          * filemap_nopage() when page is beyond the file size,
613                          * on the grounds that "An external ptracer can access
614                          * pages that normally aren't accessible.." Don't
615                          * propagate such page fault to the lower layers to
616                          * avoid side-effects like KMS updates.
617                          */
618                         if (fio->ft_index > last)
619                                 result = +1;
620                 } else
621                         result = PTR_ERR(page);
622         }
623         return result;
624 }
625
626 static void vvp_io_advance(const struct lu_env *env,
627                            const struct cl_io_slice *ios, size_t nob)
628 {
629         struct ccc_io    *vio = cl2ccc_io(env, ios);
630         struct cl_io     *io  = ios->cis_io;
631         struct cl_object *obj = ios->cis_io->ci_obj;
632
633         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
634
635         if (!cl_io_is_sendfile(io) && io->ci_continue) {
636                 /* update the iov */
637                 LASSERT(vio->cui_tot_nrsegs >= vio->cui_nrsegs);
638                 LASSERT(vio->cui_tot_count  >= nob);
639
640                 vio->cui_iov        += vio->cui_nrsegs;
641                 vio->cui_tot_nrsegs -= vio->cui_nrsegs;
642                 vio->cui_tot_count  -= nob;
643
644                 if (vio->cui_iov_olen) {
645                         struct iovec *iv;
646
647                         vio->cui_iov--;
648                         vio->cui_tot_nrsegs++;
649                         iv = &vio->cui_iov[0];
650                         iv->iov_base += iv->iov_len;
651                         LASSERT(vio->cui_iov_olen > iv->iov_len);
652                         iv->iov_len = vio->cui_iov_olen - iv->iov_len;
653                 }
654         }
655 }
656
657 static int vvp_io_read_page(const struct lu_env *env,
658                             const struct cl_io_slice *ios,
659                             const struct cl_page_slice *slice)
660 {
661         struct cl_io              *io     = ios->cis_io;
662         struct cl_object          *obj    = slice->cpl_obj;
663         struct ccc_page           *cp     = cl2ccc_page(slice);
664         struct cl_page            *page   = slice->cpl_page;
665         struct inode              *inode  = ccc_object_inode(obj);
666         struct ll_sb_info         *sbi    = ll_i2sbi(inode);
667         struct ll_file_data       *fd     = cl2ccc_io(env, ios)->cui_fd;
668         struct ll_readahead_state *ras    = &fd->fd_ras;
669         cfs_page_t                *vmpage = cp->cpg_page;
670         struct cl_2queue          *queue  = &io->ci_queue;
671
672         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
673         LASSERT(cl2vvp_io(env, ios)->cui_oneshot == 0);
674         LASSERT(slice->cpl_obj == obj);
675
676         ENTRY;
677
678         if (sbi->ll_ra_info.ra_max_pages)
679                 ras_update(sbi, inode, ras, page->cp_index,
680                            cp->cpg_defer_uptodate);
681
682         /* Sanity check whether the page is protected by a lock. */
683         if (likely(!(fd->fd_flags & LL_FILE_IGNORE_LOCK))) {
684                 int rc;
685
686                 rc = cl_page_is_under_lock(env, io, page);
687                 if (rc != -EBUSY) {
688                         CL_PAGE_HEADER(D_WARNING, env, page, "%s: %i\n",
689                                        rc == -ENODATA ? "without a lock" :
690                                        "match failed", rc);
691                         if (rc != -ENODATA)
692                                 RETURN(rc);
693                 }
694         }
695
696         if (cp->cpg_defer_uptodate) {
697                 cp->cpg_ra_used = 1;
698                 cl_page_export(env, page);
699         }
700         /*
701          * Add page into the queue even when it is marked uptodate above.
702          * this will unlock it automatically as part of cl_page_list_disown().
703          */
704         cl_2queue_add(queue, page);
705         if (sbi->ll_ra_info.ra_max_pages)
706                 ll_readahead(env, io, ras,
707                              vmpage->mapping, &queue->c2_qin, fd->fd_flags);
708
709         RETURN(0);
710 }
711
712 static int vvp_page_sync_io(const struct lu_env *env, struct cl_io *io,
713                             struct cl_page *page, struct ccc_page *cp,
714                             int to, enum cl_req_type crt)
715 {
716         struct cl_2queue  *queue;
717         struct ccc_object *cobo   = cl2ccc(page->cp_obj);
718         struct cl_sync_io *anchor = &ccc_env_info(env)->cti_sync_io;
719
720         int writing = io->ci_type == CIT_WRITE;
721         int result;
722
723         LASSERT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE);
724
725         queue = &io->ci_queue;
726
727         cl_2queue_init_page(queue, page);
728
729         if (writing)
730                 /* Do not pass llap here as it is sync write. */
731                 vvp_write_pending(cobo, cp);
732
733         cl_sync_io_init(anchor, 1);
734         cp->cpg_sync_io = anchor;
735         cl_page_clip(env, page, 0, to);
736         result = cl_io_submit_rw(env, io, crt, queue, CRP_NORMAL);
737         if (result == 0)
738                 result = cl_sync_io_wait(env, io, &queue->c2_qout, anchor);
739         else
740                 cp->cpg_sync_io = NULL;
741         LASSERT(cl_page_is_owned(page, io));
742         cl_page_clip(env, page, 0, CFS_PAGE_SIZE);
743
744         if (crt == CRT_READ)
745                 /*
746                  * in CRT_WRITE case page is left locked even in case of
747                  * error.
748                  */
749                 cl_page_list_disown(env, io, &queue->c2_qin);
750         cl_2queue_fini(env, queue);
751
752         return result;
753 }
754
755 /**
756  * Prepare partially written-to page for a write.
757  */
758 static int vvp_io_prepare_partial(const struct lu_env *env, struct cl_io *io,
759                                   struct cl_object *obj, struct cl_page *pg,
760                                   struct ccc_page *cp,
761                                   unsigned from, unsigned to)
762 {
763         struct cl_attr *attr   = &ccc_env_info(env)->cti_attr;
764         loff_t          offset = cl_offset(obj, pg->cp_index);
765         int             result;
766
767         cl_object_attr_lock(obj);
768         result = cl_object_attr_get(env, obj, attr);
769         cl_object_attr_unlock(obj);
770         if (result == 0) {
771                 /*
772                  * If are writing to a new page, no need to read old data.
773                  * The extent locking will have updated the KMS, and for our
774                  * purposes here we can treat it like i_size.
775                  */
776                 if (attr->cat_kms <= offset) {
777                         char *kaddr = kmap_atomic(cp->cpg_page, KM_USER0);
778
779                         memset(kaddr, 0, cl_page_size(obj));
780                         kunmap_atomic(kaddr, KM_USER0);
781                 } else if (cp->cpg_defer_uptodate)
782                         cp->cpg_ra_used = 1;
783                 else
784                         result = vvp_page_sync_io(env, io, pg, cp,
785                                                   CFS_PAGE_SIZE, CRT_READ);
786                 /*
787                  * In older implementations, obdo_refresh_inode is called here
788                  * to update the inode because the write might modify the
789                  * object info at OST. However, this has been proven useless,
790                  * since LVB functions will be called when user space program
791                  * tries to retrieve inode attribute.  Also, see bug 15909 for
792                  * details. -jay
793                  */
794                 if (result == 0)
795                         cl_page_export(env, pg);
796         }
797         return result;
798 }
799
800 static int vvp_io_prepare_write(const struct lu_env *env,
801                                 const struct cl_io_slice *ios,
802                                 const struct cl_page_slice *slice,
803                                 unsigned from, unsigned to)
804 {
805         struct cl_object *obj    = slice->cpl_obj;
806         struct ccc_page  *cp     = cl2ccc_page(slice);
807         struct cl_page   *pg     = slice->cpl_page;
808         cfs_page_t       *vmpage = cp->cpg_page;
809
810         int result;
811
812         ENTRY;
813
814         LINVRNT(cl_page_is_vmlocked(env, pg));
815         LASSERT(vmpage->mapping->host == ccc_object_inode(obj));
816
817         result = 0;
818
819         CL_PAGE_HEADER(D_PAGE, env, pg, "preparing: [%d, %d]\n", from, to);
820         if (!PageUptodate(vmpage)) {
821                 /*
822                  * We're completely overwriting an existing page, so _don't_
823                  * set it up to date until commit_write
824                  */
825                 if (from == 0 && to == CFS_PAGE_SIZE) {
826                         CL_PAGE_HEADER(D_PAGE, env, pg, "full page write\n");
827                         POISON_PAGE(page, 0x11);
828                 } else
829                         result = vvp_io_prepare_partial(env, ios->cis_io, obj,
830                                                         pg, cp, from, to);
831         } else
832                 CL_PAGE_HEADER(D_PAGE, env, pg, "uptodate\n");
833         RETURN(result);
834 }
835
836 static int vvp_io_commit_write(const struct lu_env *env,
837                                const struct cl_io_slice *ios,
838                                const struct cl_page_slice *slice,
839                                unsigned from, unsigned to)
840 {
841         struct cl_object  *obj    = slice->cpl_obj;
842         struct cl_io      *io     = ios->cis_io;
843         struct ccc_page   *cp     = cl2ccc_page(slice);
844         struct cl_page    *pg     = slice->cpl_page;
845         struct inode      *inode  = ccc_object_inode(obj);
846         struct ll_sb_info *sbi    = ll_i2sbi(inode);
847         cfs_page_t        *vmpage = cp->cpg_page;
848
849         int    result;
850         int    tallyop;
851         loff_t size;
852
853         ENTRY;
854
855         LINVRNT(cl_page_is_vmlocked(env, pg));
856         LASSERT(vmpage->mapping->host == inode);
857
858         LU_OBJECT_HEADER(D_INODE, env, &obj->co_lu, "commiting page write\n");
859         CL_PAGE_HEADER(D_PAGE, env, pg, "committing: [%d, %d]\n", from, to);
860
861         /*
862          * queue a write for some time in the future the first time we
863          * dirty the page.
864          *
865          * This is different from what other file systems do: they usually
866          * just mark page (and some of its buffers) dirty and rely on
867          * balance_dirty_pages() to start a write-back. Lustre wants write-back
868          * to be started earlier for the following reasons:
869          *
870          *     (1) with a large number of clients we need to limit the amount
871          *     of cached data on the clients a lot;
872          *
873          *     (2) large compute jobs generally want compute-only then io-only
874          *     and the IO should complete as quickly as possible;
875          *
876          *     (3) IO is batched up to the RPC size and is async until the
877          *     client max cache is hit
878          *     (/proc/fs/lustre/osc/OSC.../max_dirty_mb)
879          *
880          */
881         if (!PageDirty(vmpage)) {
882                 tallyop = LPROC_LL_DIRTY_MISSES;
883                 vvp_write_pending(cl2ccc(obj), cp);
884                 set_page_dirty(vmpage);
885                 result = cl_page_cache_add(env, io, pg, CRT_WRITE);
886                 if (result == -EDQUOT)
887                         /*
888                          * Client ran out of disk space grant. Possible
889                          * strategies are:
890                          *
891                          *     (a) do a sync write, renewing grant;
892                          *
893                          *     (b) stop writing on this stripe, switch to the
894                          *     next one.
895                          *
896                          * (b) is a part of "parallel io" design that is the
897                          * ultimate goal. (a) is what "old" client did, and
898                          * what the new code continues to do for the time
899                          * being.
900                          */
901                         result = vvp_page_sync_io(env, io, pg, cp,
902                                                   to, CRT_WRITE);
903         } else {
904                 tallyop = LPROC_LL_DIRTY_HITS;
905                 result = 0;
906         }
907         ll_stats_ops_tally(sbi, tallyop, 1);
908
909         size = cl_offset(obj, pg->cp_index) + to;
910
911         if (result == 0) {
912                 if (size > i_size_read(inode))
913                         i_size_write(inode, size);
914                 cl_page_export(env, pg);
915         } else if (size > i_size_read(inode))
916                 cl_page_discard(env, io, pg);
917         RETURN(result);
918 }
919
920 static const struct cl_io_operations vvp_io_ops = {
921         .op = {
922                 [CIT_READ] = {
923                         .cio_fini      = vvp_io_fini,
924                         .cio_lock      = vvp_io_read_lock,
925                         .cio_start     = vvp_io_read_start,
926                         .cio_advance   = vvp_io_advance
927                 },
928                 [CIT_WRITE] = {
929                         .cio_fini      = vvp_io_fini,
930                         .cio_lock      = vvp_io_write_lock,
931                         .cio_start     = vvp_io_write_start,
932                         .cio_advance   = vvp_io_advance
933                 },
934                 [CIT_TRUNC] = {
935                         .cio_fini       = vvp_io_trunc_fini,
936                         .cio_iter_init  = vvp_io_trunc_iter_init,
937                         .cio_lock       = vvp_io_trunc_lock,
938                         .cio_start      = vvp_io_trunc_start,
939                         .cio_end        = vvp_io_trunc_end
940                 },
941                 [CIT_FAULT] = {
942                         .cio_fini      = vvp_io_fault_fini,
943                         .cio_iter_init = vvp_io_fault_iter_init,
944                         .cio_lock      = vvp_io_fault_lock,
945                         .cio_start     = vvp_io_fault_start,
946                         .cio_end       = ccc_io_end
947                 },
948                 [CIT_MISC] = {
949                         .cio_fini   = vvp_io_fini
950                 }
951         },
952         .cio_read_page     = vvp_io_read_page,
953         .cio_prepare_write = vvp_io_prepare_write,
954         .cio_commit_write  = vvp_io_commit_write
955 };
956
957 int vvp_io_init(const struct lu_env *env, struct cl_object *obj,
958                 struct cl_io *io)
959 {
960         struct vvp_io      *vio   = vvp_env_io(env);
961         struct ccc_io      *cio   = ccc_env_io(env);
962         struct inode       *inode = ccc_object_inode(obj);
963         struct ll_sb_info  *sbi   = ll_i2sbi(inode);
964         int                 result;
965
966         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
967         ENTRY;
968
969         CL_IO_SLICE_CLEAN(cio, cui_cl);
970         cl_io_slice_add(io, &cio->cui_cl, obj, &vvp_io_ops);
971         vio->cui_oneshot = 0;
972         vio->cui_ra_window_set = 0;
973         result = 0;
974         if (io->ci_type == CIT_READ || io->ci_type == CIT_WRITE) {
975                 int    op;
976                 size_t count;
977
978                 count = io->u.ci_rw.crw_count;
979                 op    = io->ci_type == CIT_READ ?
980                         LPROC_LL_READ_BYTES : LPROC_LL_WRITE_BYTES;
981                 if (io->ci_type == CIT_WRITE)
982                         down(&ll_i2info(inode)->lli_write_sem);
983                 /* "If nbyte is 0, read() will return 0 and have no other
984                  *  results."  -- Single Unix Spec */
985                 if (count == 0)
986                         result = 1;
987                 else {
988                         cio->cui_tot_count = count;
989                         cio->cui_tot_nrsegs = 0;
990                         ll_stats_ops_tally(sbi, op, count);
991                 }
992         } else if (io->ci_type == CIT_TRUNC) {
993                 /* lockless truncate? */
994                 ll_stats_ops_tally(sbi, LPROC_LL_TRUNC, 1);
995         }
996         RETURN(result);
997 }
998
999 static struct vvp_io *cl2vvp_io(const struct lu_env *env,
1000                                 const struct cl_io_slice *slice)
1001 {
1002         /* Caling just for assertion */
1003         cl2ccc_io(env, slice);
1004         return vvp_env_io(env);
1005 }
1006