lustre/llite/vvp_io.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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2013, Intel Corporation.
  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  *   Author: Jinshan Xiong <jinshan.xiong@whamcloud.com>
  40  */
  41
  42 #define DEBUG_SUBSYSTEM S_LLITE
  43
  44
  45 #include <obd.h>
  46 #include "vvp_internal.h"
  47
  48 static struct vvp_io *cl2vvp_io(const struct lu_env *env,
  49                                 const struct cl_io_slice *slice);
  50
  51 /**
  52  * True, if \a io is a normal io, False for splice_{read,write}
  53  */
  54 int cl_is_normalio(const struct lu_env *env, const struct cl_io *io)
  55 {
  56         struct vvp_io *vio = vvp_env_io(env);
  57
  58         LASSERT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE);
  59
  60         return vio->cui_io_subtype == IO_NORMAL;
  61 }
  62
  63 /**
  64  * For swapping layout. The file's layout may have changed.
  65  * To avoid populating pages to a wrong stripe, we have to verify the
  66  * correctness of layout. It works because swapping layout processes
  67  * have to acquire group lock.
  68  */
  69 static bool can_populate_pages(const struct lu_env *env, struct cl_io *io,
  70                                 struct inode *inode)
  71 {
  72         struct ll_inode_info    *lli = ll_i2info(inode);
  73         struct ccc_io           *cio = ccc_env_io(env);
  74         bool rc = true;
  75
  76         switch (io->ci_type) {
  77         case CIT_READ:
  78         case CIT_WRITE:
  79                 /* don't need lock here to check lli_layout_gen as we have held
  80                  * extent lock and GROUP lock has to hold to swap layout */
  81                 if (ll_layout_version_get(lli) != cio->cui_layout_gen) {
  82                         io->ci_need_restart = 1;
  83                         /* this will return application a short read/write */
  84                         io->ci_continue = 0;
  85                         rc = false;
  86                 }
  87         case CIT_FAULT:
  88                 /* fault is okay because we've already had a page. */
  89         default:
  90                 break;
  91         }
  92
  93         return rc;
  94 }
  95
  96 /*****************************************************************************
  97  *
  98  * io operations.
  99  *
 100  */
 101
 102 static int vvp_io_write_iter_init(const struct lu_env *env,
 103                                   const struct cl_io_slice *ios)
 104 {
 105         struct ccc_io *cio = cl2ccc_io(env, ios);
 106
 107         cl_page_list_init(&cio->u.write.cui_queue);
 108         cio->u.write.cui_written = 0;
 109         cio->u.write.cui_from = 0;
 110         cio->u.write.cui_to = PAGE_SIZE;
 111
 112         return 0;
 113 }
 114
 115 static void vvp_io_write_iter_fini(const struct lu_env *env,
 116                                    const struct cl_io_slice *ios)
 117 {
 118         struct ccc_io *cio = cl2ccc_io(env, ios);
 119
 120         LASSERT(cio->u.write.cui_queue.pl_nr == 0);
 121 }
 122
 123 static int vvp_io_fault_iter_init(const struct lu_env *env,
 124                                   const struct cl_io_slice *ios)
 125 {
 126         struct vvp_io *vio   = cl2vvp_io(env, ios);
 127         struct inode  *inode = ccc_object_inode(ios->cis_obj);
 128
 129         LASSERT(inode ==
 130                 cl2ccc_io(env, ios)->cui_fd->fd_file->f_dentry->d_inode);
 131         vio->u.fault.ft_mtime = LTIME_S(inode->i_mtime);
 132         return 0;
 133 }
 134
 135 static void vvp_io_fini(const struct lu_env *env, const struct cl_io_slice *ios)
 136 {
 137         struct cl_io     *io  = ios->cis_io;
 138         struct cl_object *obj = io->ci_obj;
 139         struct ccc_io    *cio = cl2ccc_io(env, ios);
 140         struct inode     *inode = ccc_object_inode(obj);
 141
 142         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
 143
 144         CDEBUG(D_VFSTRACE, DFID" ignore/verify layout %d/%d, layout version %d "
 145                            "restore needed %d\n",
 146                PFID(lu_object_fid(&obj->co_lu)),
 147                io->ci_ignore_layout, io->ci_verify_layout,
 148                cio->cui_layout_gen, io->ci_restore_needed);
 149
 150         if (io->ci_restore_needed == 1) {
 151                 int     rc;
 152
 153                 /* file was detected release, we need to restore it
 154                  * before finishing the io
 155                  */
 156                 rc = ll_layout_restore(inode, 0, OBD_OBJECT_EOF);
 157                 /* if restore registration failed, no restart,
 158                  * we will return -ENODATA */
 159                 /* The layout will change after restore, so we need to
 160                  * block on layout lock hold by the MDT
 161                  * as MDT will not send new layout in lvb (see LU-3124)
 162                  * we have to explicitly fetch it, all this will be done
 163                  * by ll_layout_refresh()
 164                  */
 165                 if (rc == 0) {
 166                         io->ci_restore_needed = 0;
 167                         io->ci_need_restart = 1;
 168                         io->ci_verify_layout = 1;
 169                 } else {
 170                         io->ci_restore_needed = 1;
 171                         io->ci_need_restart = 0;
 172                         io->ci_verify_layout = 0;
 173                         io->ci_result = rc;
 174                 }
 175         }
 176
 177         if (!io->ci_ignore_layout && io->ci_verify_layout) {
 178                 __u32 gen = 0;
 179
 180                 /* check layout version */
 181                 ll_layout_refresh(inode, &gen);
 182                 io->ci_need_restart = cio->cui_layout_gen != gen;
 183                 if (io->ci_need_restart) {
 184                         CDEBUG(D_VFSTRACE,
 185                                DFID" layout changed from %d to %d.\n",
 186                                PFID(lu_object_fid(&obj->co_lu)),
 187                                cio->cui_layout_gen, gen);
 188                         /* today successful restore is the only possible
 189                          * case */
 190                         /* restore was done, clear restoring state */
 191                         ll_i2info(ccc_object_inode(obj))->lli_flags &=
 192                                 ~LLIF_FILE_RESTORING;
 193                 }
 194         }
 195 }
 196
 197 static void vvp_io_fault_fini(const struct lu_env *env,
 198                               const struct cl_io_slice *ios)
 199 {
 200         struct cl_io   *io   = ios->cis_io;
 201         struct cl_page *page = io->u.ci_fault.ft_page;
 202
 203         CLOBINVRNT(env, io->ci_obj, ccc_object_invariant(io->ci_obj));
 204
 205         if (page != NULL) {
 206                 lu_ref_del(&page->cp_reference, "fault", io);
 207                 cl_page_put(env, page);
 208                 io->u.ci_fault.ft_page = NULL;
 209         }
 210         vvp_io_fini(env, ios);
 211 }
 212
 213 static enum cl_lock_mode vvp_mode_from_vma(struct vm_area_struct *vma)
 214 {
 215         /*
 216          * we only want to hold PW locks if the mmap() can generate
 217          * writes back to the file and that only happens in shared
 218          * writable vmas
 219          */
 220         if ((vma->vm_flags & VM_SHARED) && (vma->vm_flags & VM_WRITE))
 221                 return CLM_WRITE;
 222         return CLM_READ;
 223 }
 224
 225 static int vvp_mmap_locks(const struct lu_env *env,
 226                           struct ccc_io *vio, struct cl_io *io)
 227 {
 228         struct ccc_thread_info *cti = ccc_env_info(env);
 229         struct mm_struct       *mm = current->mm;
 230         struct vm_area_struct  *vma;
 231         struct cl_lock_descr   *descr = &cti->cti_descr;
 232         ldlm_policy_data_t      policy;
 233         unsigned long           addr;
 234         unsigned long           seg;
 235         ssize_t                 count;
 236         int                     result = 0;
 237         ENTRY;
 238
 239         LASSERT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE);
 240
 241         if (!cl_is_normalio(env, io))
 242                 RETURN(0);
 243
 244         if (vio->cui_iov == NULL) /* nfs or loop back device write */
 245                 RETURN(0);
 246
 247         /* No MM (e.g. NFS)? No vmas too. */
 248         if (mm == NULL)
 249                 RETURN(0);
 250
 251         for (seg = 0; seg < vio->cui_nrsegs; seg++) {
 252                 const struct iovec *iv = &vio->cui_iov[seg];
 253
 254                 addr = (unsigned long)iv->iov_base;
 255                 count = iv->iov_len;
 256                 if (count == 0)
 257                         continue;
 258
 259                 count += addr & (~CFS_PAGE_MASK);
 260                 addr &= CFS_PAGE_MASK;
 261
 262                 down_read(&mm->mmap_sem);
 263                 while((vma = our_vma(mm, addr, count)) != NULL) {
 264                         struct inode *inode = vma->vm_file->f_dentry->d_inode;
 265                         int flags = CEF_MUST;
 266
 267                         if (ll_file_nolock(vma->vm_file)) {
 268                                 /*
 269                                  * For no lock case is not allowed for mmap
 270                                  */
 271                                 result = -EINVAL;
 272                                 break;
 273                         }
 274
 275                         /*
 276                          * XXX: Required lock mode can be weakened: CIT_WRITE
 277                          * io only ever reads user level buffer, and CIT_READ
 278                          * only writes on it.
 279                          */
 280                         policy_from_vma(&policy, vma, addr, count);
 281                         descr->cld_mode = vvp_mode_from_vma(vma);
 282                         descr->cld_obj = ll_i2info(inode)->lli_clob;
 283                         descr->cld_start = cl_index(descr->cld_obj,
 284                                                     policy.l_extent.start);
 285                         descr->cld_end = cl_index(descr->cld_obj,
 286                                                   policy.l_extent.end);
 287                         descr->cld_enq_flags = flags;
 288                         result = cl_io_lock_alloc_add(env, io, descr);
 289
 290                         CDEBUG(D_VFSTRACE, "lock: %d: [%lu, %lu]\n",
 291                                descr->cld_mode, descr->cld_start,
 292                                descr->cld_end);
 293
 294                         if (result < 0)
 295                                 break;
 296
 297                         if (vma->vm_end - addr >= count)
 298                                 break;
 299
 300                         count -= vma->vm_end - addr;
 301                         addr = vma->vm_end;
 302                 }
 303                 up_read(&mm->mmap_sem);
 304                 if (result < 0)
 305                         break;
 306         }
 307         RETURN(result);
 308 }
 309
 310 static int vvp_io_rw_lock(const struct lu_env *env, struct cl_io *io,
 311                           enum cl_lock_mode mode, loff_t start, loff_t end)
 312 {
 313         struct ccc_io *cio = ccc_env_io(env);
 314         int result;
 315         int ast_flags = 0;
 316
 317         LASSERT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE);
 318         ENTRY;
 319
 320         ccc_io_update_iov(env, cio, io);
 321
 322         if (io->u.ci_rw.crw_nonblock)
 323                 ast_flags |= CEF_NONBLOCK;
 324         result = vvp_mmap_locks(env, cio, io);
 325         if (result == 0)
 326                 result = ccc_io_one_lock(env, io, ast_flags, mode, start, end);
 327         RETURN(result);
 328 }
 329
 330 static int vvp_io_read_lock(const struct lu_env *env,
 331                             const struct cl_io_slice *ios)
 332 {
 333         struct cl_io            *io  = ios->cis_io;
 334         struct cl_io_rw_common  *rd = &io->u.ci_rd.rd;
 335         int result;
 336
 337         ENTRY;
 338         result = vvp_io_rw_lock(env, io, CLM_READ, rd->crw_pos,
 339                                 rd->crw_pos + rd->crw_count - 1);
 340         RETURN(result);
 341 }
 342
 343 static int vvp_io_fault_lock(const struct lu_env *env,
 344                              const struct cl_io_slice *ios)
 345 {
 346         struct cl_io *io   = ios->cis_io;
 347         struct vvp_io *vio = cl2vvp_io(env, ios);
 348         /*
 349          * XXX LDLM_FL_CBPENDING
 350          */
 351         return ccc_io_one_lock_index
 352                 (env, io, 0, vvp_mode_from_vma(vio->u.fault.ft_vma),
 353                  io->u.ci_fault.ft_index, io->u.ci_fault.ft_index);
 354 }
 355
 356 static int vvp_io_write_lock(const struct lu_env *env,
 357                              const struct cl_io_slice *ios)
 358 {
 359         struct cl_io *io = ios->cis_io;
 360         loff_t start;
 361         loff_t end;
 362
 363         if (io->u.ci_wr.wr_append) {
 364                 start = 0;
 365                 end   = OBD_OBJECT_EOF;
 366         } else {
 367                 start = io->u.ci_wr.wr.crw_pos;
 368                 end   = start + io->u.ci_wr.wr.crw_count - 1;
 369         }
 370         return vvp_io_rw_lock(env, io, CLM_WRITE, start, end);
 371 }
 372
 373 static int vvp_io_setattr_iter_init(const struct lu_env *env,
 374                                     const struct cl_io_slice *ios)
 375 {
 376         return 0;
 377 }
 378
 379 /**
 380  * Implementation of cl_io_operations::cio_lock() method for CIT_SETATTR io.
 381  *
 382  * Handles "lockless io" mode when extent locking is done by server.
 383  */
 384 static int vvp_io_setattr_lock(const struct lu_env *env,
 385                                const struct cl_io_slice *ios)
 386 {
 387         struct ccc_io *cio = ccc_env_io(env);
 388         struct cl_io  *io  = ios->cis_io;
 389         __u64 new_size;
 390         __u32 enqflags = 0;
 391
 392         if (cl_io_is_trunc(io)) {
 393                 new_size = io->u.ci_setattr.sa_attr.lvb_size;
 394                 if (new_size == 0)
 395                         enqflags = CEF_DISCARD_DATA;
 396         } else {
 397                 if ((io->u.ci_setattr.sa_attr.lvb_mtime >=
 398                      io->u.ci_setattr.sa_attr.lvb_ctime) ||
 399                     (io->u.ci_setattr.sa_attr.lvb_atime >=
 400                      io->u.ci_setattr.sa_attr.lvb_ctime))
 401                         return 0;
 402                 new_size = 0;
 403         }
 404         cio->u.setattr.cui_local_lock = SETATTR_EXTENT_LOCK;
 405         return ccc_io_one_lock(env, io, enqflags, CLM_WRITE,
 406                                new_size, OBD_OBJECT_EOF);
 407 }
 408
 409 static int vvp_do_vmtruncate(struct inode *inode, size_t size)
 410 {
 411         int     result;
 412
 413         /*
 414          * Only ll_inode_size_lock is taken at this level.
 415          */
 416         ll_inode_size_lock(inode);
 417         result = inode_newsize_ok(inode, size);
 418         if (result < 0) {
 419                 ll_inode_size_unlock(inode);
 420                 return result;
 421         }
 422         i_size_write(inode, size);
 423
 424         ll_truncate_pagecache(inode, size);
 425         ll_inode_size_unlock(inode);
 426         return result;
 427 }
 428
 429 static int vvp_io_setattr_trunc(const struct lu_env *env,
 430                                 const struct cl_io_slice *ios,
 431                                 struct inode *inode, loff_t size)
 432 {
 433         inode_dio_wait(inode);
 434         return 0;
 435 }
 436
 437 static int vvp_io_setattr_time(const struct lu_env *env,
 438                                const struct cl_io_slice *ios)
 439 {
 440         struct cl_io       *io    = ios->cis_io;
 441         struct cl_object   *obj   = io->ci_obj;
 442         struct cl_attr     *attr  = ccc_env_thread_attr(env);
 443         int result;
 444         unsigned valid = CAT_CTIME;
 445
 446         cl_object_attr_lock(obj);
 447         attr->cat_ctime = io->u.ci_setattr.sa_attr.lvb_ctime;
 448         if (io->u.ci_setattr.sa_valid & ATTR_ATIME_SET) {
 449                 attr->cat_atime = io->u.ci_setattr.sa_attr.lvb_atime;
 450                 valid |= CAT_ATIME;
 451         }
 452         if (io->u.ci_setattr.sa_valid & ATTR_MTIME_SET) {
 453                 attr->cat_mtime = io->u.ci_setattr.sa_attr.lvb_mtime;
 454                 valid |= CAT_MTIME;
 455         }
 456         result = cl_object_attr_set(env, obj, attr, valid);
 457         cl_object_attr_unlock(obj);
 458
 459         return result;
 460 }
 461
 462 static int vvp_io_setattr_start(const struct lu_env *env,
 463                                 const struct cl_io_slice *ios)
 464 {
 465         struct cl_io    *io    = ios->cis_io;
 466         struct inode    *inode = ccc_object_inode(io->ci_obj);
 467         int result = 0;
 468
 469         mutex_lock(&inode->i_mutex);
 470         if (cl_io_is_trunc(io))
 471                 result = vvp_io_setattr_trunc(env, ios, inode,
 472                                         io->u.ci_setattr.sa_attr.lvb_size);
 473         if (result == 0)
 474                 result = vvp_io_setattr_time(env, ios);
 475         return result;
 476 }
 477
 478 static void vvp_io_setattr_end(const struct lu_env *env,
 479                                const struct cl_io_slice *ios)
 480 {
 481         struct cl_io *io    = ios->cis_io;
 482         struct inode *inode = ccc_object_inode(io->ci_obj);
 483
 484         if (cl_io_is_trunc(io)) {
 485                 /* Truncate in memory pages - they must be clean pages
 486                  * because osc has already notified to destroy osc_extents. */
 487                 vvp_do_vmtruncate(inode, io->u.ci_setattr.sa_attr.lvb_size);
 488                 inode_dio_write_done(inode);
 489         }
 490         mutex_unlock(&inode->i_mutex);
 491 }
 492
 493 static void vvp_io_setattr_fini(const struct lu_env *env,
 494                                 const struct cl_io_slice *ios)
 495 {
 496         vvp_io_fini(env, ios);
 497 }
 498
 499 static int vvp_io_read_start(const struct lu_env *env,
 500                              const struct cl_io_slice *ios)
 501 {
 502         struct vvp_io     *vio   = cl2vvp_io(env, ios);
 503         struct ccc_io     *cio   = cl2ccc_io(env, ios);
 504         struct cl_io      *io    = ios->cis_io;
 505         struct cl_object  *obj   = io->ci_obj;
 506         struct inode      *inode = ccc_object_inode(obj);
 507         struct ll_ra_read *bead  = &vio->cui_bead;
 508         struct file       *file  = cio->cui_fd->fd_file;
 509
 510         int     result;
 511         loff_t  pos = io->u.ci_rd.rd.crw_pos;
 512         long    cnt = io->u.ci_rd.rd.crw_count;
 513         long    tot = cio->cui_tot_count;
 514         int     exceed = 0;
 515
 516         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
 517
 518         CDEBUG(D_VFSTRACE, "read: -> [%lli, %lli)\n", pos, pos + cnt);
 519
 520         if (!can_populate_pages(env, io, inode))
 521                 return 0;
 522
 523         result = ccc_prep_size(env, obj, io, pos, tot, &exceed);
 524         if (result != 0)
 525                 return result;
 526         else if (exceed != 0)
 527                 goto out;
 528
 529         LU_OBJECT_HEADER(D_INODE, env, &obj->co_lu,
 530                         "Read ino %lu, %lu bytes, offset %lld, size %llu\n",
 531                         inode->i_ino, cnt, pos, i_size_read(inode));
 532
 533         /* turn off the kernel's read-ahead */
 534         cio->cui_fd->fd_file->f_ra.ra_pages = 0;
 535
 536         /* initialize read-ahead window once per syscall */
 537         if (!vio->cui_ra_window_set) {
 538                 vio->cui_ra_window_set = 1;
 539                 bead->lrr_start = cl_index(obj, pos);
 540                 bead->lrr_count = cl_index(obj, tot + PAGE_CACHE_SIZE - 1);
 541                 ll_ra_read_in(file, bead);
 542         }
 543
 544         /* BUG: 5972 */
 545         file_accessed(file);
 546         switch (vio->cui_io_subtype) {
 547         case IO_NORMAL:
 548                 LASSERT(cio->cui_iocb->ki_pos == pos);
 549                 result = generic_file_aio_read(cio->cui_iocb,
 550                                                cio->cui_iov, cio->cui_nrsegs,
 551                                                cio->cui_iocb->ki_pos);
 552                 break;
 553         case IO_SPLICE:
 554                 result = generic_file_splice_read(file, &pos,
 555                                 vio->u.splice.cui_pipe, cnt,
 556                                 vio->u.splice.cui_flags);
 557                 /* LU-1109: do splice read stripe by stripe otherwise if it
 558                  * may make nfsd stuck if this read occupied all internal pipe
 559                  * buffers. */
 560                 io->ci_continue = 0;
 561                 break;
 562         default:
 563                 CERROR("Wrong IO type %u\n", vio->cui_io_subtype);
 564                 LBUG();
 565         }
 566
 567 out:
 568         if (result >= 0) {
 569                 if (result < cnt)
 570                         io->ci_continue = 0;
 571                 io->ci_nob += result;
 572                 ll_rw_stats_tally(ll_i2sbi(inode), current->pid, cio->cui_fd,
 573                                   pos, result, READ);
 574                 result = 0;
 575         }
 576
 577         return result;
 578 }
 579
 580 static void vvp_io_read_fini(const struct lu_env *env, const struct cl_io_slice *ios)
 581 {
 582         struct vvp_io *vio = cl2vvp_io(env, ios);
 583         struct ccc_io *cio = cl2ccc_io(env, ios);
 584
 585         if (vio->cui_ra_window_set)
 586                 ll_ra_read_ex(cio->cui_fd->fd_file, &vio->cui_bead);
 587
 588         vvp_io_fini(env, ios);
 589 }
 590
 591 static int vvp_io_commit_sync(const struct lu_env *env, struct cl_io *io,
 592                               struct cl_page_list *plist, int from, int to)
 593 {
 594         struct cl_2queue *queue = &io->ci_queue;
 595         struct cl_page *page;
 596         unsigned int bytes = 0;
 597         int rc = 0;
 598         ENTRY;
 599
 600         if (plist->pl_nr == 0)
 601                 RETURN(0);
 602
 603         if (from > 0 || to != PAGE_SIZE) {
 604                 page = cl_page_list_first(plist);
 605                 if (plist->pl_nr == 1) {
 606                         cl_page_clip(env, page, from, to);
 607                 } else {
 608                         if (from > 0)
 609                                 cl_page_clip(env, page, from, PAGE_SIZE);
 610                         if (to != PAGE_SIZE) {
 611                                 page = cl_page_list_last(plist);
 612                                 cl_page_clip(env, page, 0, to);
 613                         }
 614                 }
 615         }
 616
 617         cl_2queue_init(queue);
 618         cl_page_list_splice(plist, &queue->c2_qin);
 619         rc = cl_io_submit_sync(env, io, CRT_WRITE, queue, 0);
 620
 621         /* plist is not sorted any more */
 622         cl_page_list_splice(&queue->c2_qin, plist);
 623         cl_page_list_splice(&queue->c2_qout, plist);
 624         cl_2queue_fini(env, queue);
 625
 626         if (rc == 0) {
 627                 /* calculate bytes */
 628                 bytes = plist->pl_nr << PAGE_SHIFT;
 629                 bytes -= from + PAGE_SIZE - to;
 630
 631                 while (plist->pl_nr > 0) {
 632                         page = cl_page_list_first(plist);
 633                         cl_page_list_del(env, plist, page);
 634
 635                         cl_page_clip(env, page, 0, PAGE_SIZE);
 636
 637                         SetPageUptodate(cl_page_vmpage(page));
 638                         cl_page_disown(env, io, page);
 639
 640                         /* held in ll_cl_init() */
 641                         lu_ref_del(&page->cp_reference, "cl_io", io);
 642                         cl_page_put(env, page);
 643                 }
 644         }
 645
 646         RETURN(bytes > 0 ? bytes : rc);
 647 }
 648
 649 static void write_commit_callback(const struct lu_env *env, struct cl_io *io,
 650                                 struct cl_page *page)
 651 {
 652         struct ccc_page *cp;
 653         struct page *vmpage = page->cp_vmpage;
 654         struct cl_object *clob = cl_io_top(io)->ci_obj;
 655
 656         SetPageUptodate(vmpage);
 657         set_page_dirty(vmpage);
 658
 659         cp = cl2ccc_page(cl_object_page_slice(clob, page));
 660         vvp_write_pending(cl2ccc(clob), cp);
 661
 662         cl_page_disown(env, io, page);
 663
 664         /* held in ll_cl_init() */
 665         lu_ref_del(&page->cp_reference, "cl_io", cl_io_top(io));
 666         cl_page_put(env, page);
 667 }
 668
 669 /* make sure the page list is contiguous */
 670 static bool page_list_sanity_check(struct cl_object *obj,
 671                                    struct cl_page_list *plist)
 672 {
 673         struct cl_page *page;
 674         pgoff_t index = CL_PAGE_EOF;
 675
 676         cl_page_list_for_each(page, plist) {
 677                 struct ccc_page *cp = cl_object_page_slice(obj, page);
 678
 679                 if (index == CL_PAGE_EOF) {
 680                         index = ccc_index(cp);
 681                         continue;
 682                 }
 683
 684                 ++index;
 685                 if (index == ccc_index(cp))
 686                         continue;
 687
 688                 return false;
 689         }
 690         return true;
 691 }
 692
 693 /* Return how many bytes have queued or written */
 694 int vvp_io_write_commit(const struct lu_env *env, struct cl_io *io)
 695 {
 696         struct cl_object *obj = io->ci_obj;
 697         struct inode *inode = ccc_object_inode(obj);
 698         struct ccc_io *cio = ccc_env_io(env);
 699         struct cl_page_list *queue = &cio->u.write.cui_queue;
 700         struct cl_page *page;
 701         int rc = 0;
 702         int bytes = 0;
 703         unsigned int npages = cio->u.write.cui_queue.pl_nr;
 704         ENTRY;
 705
 706         if (npages == 0)
 707                 RETURN(0);
 708
 709         CDEBUG(D_VFSTRACE, "commit async pages: %d, from %d, to %d\n",
 710                 npages, cio->u.write.cui_from, cio->u.write.cui_to);
 711
 712         LASSERT(page_list_sanity_check(obj, queue));
 713
 714         /* submit IO with async write */
 715         rc = cl_io_commit_async(env, io, queue,
 716                                 cio->u.write.cui_from, cio->u.write.cui_to,
 717                                 write_commit_callback);
 718         npages -= queue->pl_nr; /* already committed pages */
 719         if (npages > 0) {
 720                 /* calculate how many bytes were written */
 721                 bytes = npages << PAGE_SHIFT;
 722
 723                 /* first page */
 724                 bytes -= cio->u.write.cui_from;
 725                 if (queue->pl_nr == 0) /* last page */
 726                         bytes -= PAGE_SIZE - cio->u.write.cui_to;
 727                 LASSERTF(bytes > 0, "bytes = %d, pages = %d\n", bytes, npages);
 728
 729                 cio->u.write.cui_written += bytes;
 730
 731                 CDEBUG(D_VFSTRACE, "Committed %d pages %d bytes, tot: %ld\n",
 732                         npages, bytes, cio->u.write.cui_written);
 733
 734                 /* the first page must have been written. */
 735                 cio->u.write.cui_from = 0;
 736         }
 737         LASSERT(page_list_sanity_check(obj, queue));
 738         LASSERT(ergo(rc == 0, queue->pl_nr == 0));
 739
 740         /* out of quota, try sync write */
 741         if (rc == -EDQUOT && !cl_io_is_mkwrite(io)) {
 742                 rc = vvp_io_commit_sync(env, io, queue,
 743                                         cio->u.write.cui_from,
 744                                         cio->u.write.cui_to);
 745                 if (rc > 0) {
 746                         cio->u.write.cui_written += rc;
 747                         rc = 0;
 748                 }
 749         }
 750
 751         /* update inode size */
 752         ll_merge_lvb(env, inode);
 753
 754         /* Now the pages in queue were failed to commit, discard them
 755          * unless they were dirtied before. */
 756         while (queue->pl_nr > 0) {
 757                 page = cl_page_list_first(queue);
 758                 cl_page_list_del(env, queue, page);
 759
 760                 if (!PageDirty(cl_page_vmpage(page)))
 761                         cl_page_discard(env, io, page);
 762
 763                 cl_page_disown(env, io, page);
 764
 765                 /* held in ll_cl_init() */
 766                 lu_ref_del(&page->cp_reference, "cl_io", io);
 767                 cl_page_put(env, page);
 768         }
 769         cl_page_list_fini(env, queue);
 770
 771         RETURN(rc);
 772 }
 773
 774 static int vvp_io_write_start(const struct lu_env *env,
 775                               const struct cl_io_slice *ios)
 776 {
 777         struct ccc_io      *cio   = cl2ccc_io(env, ios);
 778         struct cl_io       *io    = ios->cis_io;
 779         struct cl_object   *obj   = io->ci_obj;
 780         struct inode       *inode = ccc_object_inode(obj);
 781         ssize_t result = 0;
 782         loff_t pos = io->u.ci_wr.wr.crw_pos;
 783         size_t cnt = io->u.ci_wr.wr.crw_count;
 784
 785         ENTRY;
 786
 787         if (!can_populate_pages(env, io, inode))
 788                 RETURN(0);
 789
 790         if (cl_io_is_append(io)) {
 791                 /*
 792                  * PARALLEL IO This has to be changed for parallel IO doing
 793                  * out-of-order writes.
 794                  */
 795                 ll_merge_lvb(env, inode);
 796                 pos = io->u.ci_wr.wr.crw_pos = i_size_read(inode);
 797                 cio->cui_iocb->ki_pos = pos;
 798         } else {
 799                 LASSERT(cio->cui_iocb->ki_pos == pos);
 800         }
 801
 802         CDEBUG(D_VFSTRACE, "write: [%lli, %lli)\n", pos, pos + (long long)cnt);
 803
 804         if (cio->cui_iov == NULL) {
 805                 /* from a temp io in ll_cl_init(). */
 806                 result = 0;
 807         } else {
 808                 /*
 809                  * When using the locked AIO function (generic_file_aio_write())
 810                  * testing has shown the inode mutex to be a limiting factor
 811                  * with multi-threaded single shared file performance. To get
 812                  * around this, we now use the lockless version. To maintain
 813                  * consistency, proper locking to protect against writes,
 814                  * trucates, etc. is handled in the higher layers of lustre.
 815                  */
 816                 result = __generic_file_aio_write(cio->cui_iocb,
 817                                                   cio->cui_iov, cio->cui_nrsegs,
 818                                                   &cio->cui_iocb->ki_pos);
 819                 if (result > 0 || result == -EIOCBQUEUED) {
 820                         ssize_t err;
 821
 822                         err = generic_write_sync(cio->cui_iocb->ki_filp,
 823                                                  pos, result);
 824                         if (err < 0 && result > 0)
 825                                 result = err;
 826                 }
 827
 828         }
 829         if (result > 0) {
 830                 result = vvp_io_write_commit(env, io);
 831                 if (cio->u.write.cui_written > 0) {
 832                         result = cio->u.write.cui_written;
 833                         io->ci_nob += result;
 834
 835                         CDEBUG(D_VFSTRACE, "write: nob %zd, result: %zd\n",
 836                                 io->ci_nob, result);
 837                 }
 838         }
 839         if (result > 0) {
 840                 struct ll_inode_info *lli = ll_i2info(inode);
 841
 842                 spin_lock(&lli->lli_lock);
 843                 lli->lli_flags |= LLIF_DATA_MODIFIED;
 844                 spin_unlock(&lli->lli_lock);
 845
 846                 if (result < cnt)
 847                         io->ci_continue = 0;
 848                 ll_rw_stats_tally(ll_i2sbi(inode), current->pid,
 849                                   cio->cui_fd, pos, result, WRITE);
 850                 result = 0;
 851         }
 852
 853         RETURN(result);
 854 }
 855
 856 static int vvp_io_kernel_fault(struct vvp_fault_io *cfio)
 857 {
 858         struct vm_fault *vmf = cfio->fault.ft_vmf;
 859
 860         cfio->fault.ft_flags = filemap_fault(cfio->ft_vma, vmf);
 861         cfio->fault.ft_flags_valid = 1;
 862
 863         if (vmf->page) {
 864                 LL_CDEBUG_PAGE(D_PAGE, vmf->page, "got addr %p type NOPAGE\n",
 865                                vmf->virtual_address);
 866                 if (unlikely(!(cfio->fault.ft_flags & VM_FAULT_LOCKED))) {
 867                         lock_page(vmf->page);
 868                         cfio->fault.ft_flags |= VM_FAULT_LOCKED;
 869                 }
 870
 871                 cfio->ft_vmpage = vmf->page;
 872                 return 0;
 873         }
 874
 875         if (cfio->fault.ft_flags & VM_FAULT_SIGBUS) {
 876                 CDEBUG(D_PAGE, "got addr %p - SIGBUS\n", vmf->virtual_address);
 877                 return -EFAULT;
 878         }
 879
 880         if (cfio->fault.ft_flags & VM_FAULT_OOM) {
 881                 CDEBUG(D_PAGE, "got addr %p - OOM\n", vmf->virtual_address);
 882                 return -ENOMEM;
 883         }
 884
 885         if (cfio->fault.ft_flags & VM_FAULT_RETRY)
 886                 return -EAGAIN;
 887
 888         CERROR("unknow error in page fault %d!\n", cfio->fault.ft_flags);
 889         return -EINVAL;
 890 }
 891
 892 static void mkwrite_commit_callback(const struct lu_env *env, struct cl_io *io,
 893                                     struct cl_page *page)
 894 {
 895         struct ccc_page *cp;
 896         struct cl_object *clob = cl_io_top(io)->ci_obj;
 897
 898         set_page_dirty(page->cp_vmpage);
 899
 900         cp = cl2ccc_page(cl_object_page_slice(clob, page));
 901         vvp_write_pending(cl2ccc(clob), cp);
 902 }
 903
 904 static int vvp_io_fault_start(const struct lu_env *env,
 905                               const struct cl_io_slice *ios)
 906 {
 907         struct vvp_io       *vio     = cl2vvp_io(env, ios);
 908         struct cl_io        *io      = ios->cis_io;
 909         struct cl_object    *obj     = io->ci_obj;
 910         struct inode        *inode   = ccc_object_inode(obj);
 911         struct cl_fault_io  *fio     = &io->u.ci_fault;
 912         struct vvp_fault_io *cfio    = &vio->u.fault;
 913         loff_t               offset;
 914         int                  result  = 0;
 915         struct page          *vmpage  = NULL;
 916         struct cl_page      *page;
 917         loff_t               size;
 918         pgoff_t              last_index;
 919         ENTRY;
 920
 921         if (fio->ft_executable &&
 922             LTIME_S(inode->i_mtime) != vio->u.fault.ft_mtime)
 923                 CWARN("binary "DFID
 924                       " changed while waiting for the page fault lock\n",
 925                       PFID(lu_object_fid(&obj->co_lu)));
 926
 927         /* offset of the last byte on the page */
 928         offset = cl_offset(obj, fio->ft_index + 1) - 1;
 929         LASSERT(cl_index(obj, offset) == fio->ft_index);
 930         result = ccc_prep_size(env, obj, io, 0, offset + 1, NULL);
 931         if (result != 0)
 932                 RETURN(result);
 933
 934         /* must return locked page */
 935         if (fio->ft_mkwrite) {
 936                 LASSERT(cfio->ft_vmpage != NULL);
 937                 lock_page(cfio->ft_vmpage);
 938         } else {
 939                 result = vvp_io_kernel_fault(cfio);
 940                 if (result != 0)
 941                         RETURN(result);
 942         }
 943
 944         vmpage = cfio->ft_vmpage;
 945         LASSERT(PageLocked(vmpage));
 946
 947         if (OBD_FAIL_CHECK(OBD_FAIL_LLITE_FAULT_TRUNC_RACE))
 948                 ll_invalidate_page(vmpage);
 949
 950         size = i_size_read(inode);
 951         /* Though we have already held a cl_lock upon this page, but
 952          * it still can be truncated locally. */
 953         if (unlikely((vmpage->mapping != inode->i_mapping) ||
 954                      (page_offset(vmpage) > size))) {
 955                 CDEBUG(D_PAGE, "llite: fault and truncate race happened!\n");
 956
 957                 /* return +1 to stop cl_io_loop() and ll_fault() will catch
 958                  * and retry. */
 959                 GOTO(out, result = +1);
 960         }
 961
 962         last_index = cl_index(obj, size - 1);
 963
 964         if (fio->ft_mkwrite ) {
 965                 /*
 966                  * Capture the size while holding the lli_trunc_sem from above
 967                  * we want to make sure that we complete the mkwrite action
 968                  * while holding this lock. We need to make sure that we are
 969                  * not past the end of the file.
 970                  */
 971                 if (last_index < fio->ft_index) {
 972                         CDEBUG(D_PAGE,
 973                                 "llite: mkwrite and truncate race happened: "
 974                                 "%p: 0x%lx 0x%lx\n",
 975                                 vmpage->mapping,fio->ft_index,last_index);
 976                         /*
 977                          * We need to return if we are
 978                          * passed the end of the file. This will propagate
 979                          * up the call stack to ll_page_mkwrite where
 980                          * we will return VM_FAULT_NOPAGE. Any non-negative
 981                          * value returned here will be silently
 982                          * converted to 0. If the vmpage->mapping is null
 983                          * the error code would be converted back to ENODATA
 984                          * in ll_page_mkwrite0. Thus we return -ENODATA
 985                          * to handle both cases
 986                          */
 987                         GOTO(out, result = -ENODATA);
 988                 }
 989         }
 990
 991         page = cl_page_find(env, obj, fio->ft_index, vmpage, CPT_CACHEABLE);
 992         if (IS_ERR(page))
 993                 GOTO(out, result = PTR_ERR(page));
 994
 995         /* if page is going to be written, we should add this page into cache
 996          * earlier. */
 997         if (fio->ft_mkwrite) {
 998                 wait_on_page_writeback(vmpage);
 999                 if (!PageDirty(vmpage)) {
1000                         struct cl_page_list *plist = &io->ci_queue.c2_qin;
1001                         struct ccc_page *cp = cl_object_page_slice(obj, page);
1002                         int to = PAGE_SIZE;
1003
1004                         /* vvp_page_assume() calls wait_on_page_writeback(). */
1005                         cl_page_assume(env, io, page);
1006
1007                         cl_page_list_init(plist);
1008                         cl_page_list_add(plist, page);
1009
1010                         /* size fixup */
1011                         if (last_index == ccc_index(cp))
1012                                 to = size & ~CFS_PAGE_MASK;
1013
1014                         /* Do not set Dirty bit here so that in case IO is
1015                          * started before the page is really made dirty, we
1016                          * still have chance to detect it. */
1017                         result = cl_io_commit_async(env, io, plist, 0, to,
1018                                                     mkwrite_commit_callback);
1019                         LASSERT(cl_page_is_owned(page, io));
1020                         cl_page_list_fini(env, plist);
1021
1022                         vmpage = NULL;
1023                         if (result < 0) {
1024                                 cl_page_discard(env, io, page);
1025                                 cl_page_disown(env, io, page);
1026
1027                                 cl_page_put(env, page);
1028
1029                                 /* we're in big trouble, what can we do now? */
1030                                 if (result == -EDQUOT)
1031                                         result = -ENOSPC;
1032                                 GOTO(out, result);
1033                         } else
1034                                 cl_page_disown(env, io, page);
1035                 }
1036         }
1037
1038         /*
1039          * The ft_index is only used in the case of
1040          * a mkwrite action. We need to check
1041          * our assertions are correct, since
1042          * we should have caught this above
1043          */
1044         LASSERT(!fio->ft_mkwrite || fio->ft_index <= last_index);
1045         if (fio->ft_index == last_index)
1046                 /*
1047                  * Last page is mapped partially.
1048                  */
1049                 fio->ft_nob = size - cl_offset(obj, fio->ft_index);
1050         else
1051                 fio->ft_nob = cl_page_size(obj);
1052
1053         lu_ref_add(&page->cp_reference, "fault", io);
1054         fio->ft_page = page;
1055         EXIT;
1056
1057 out:
1058         /* return unlocked vmpage to avoid deadlocking */
1059         if (vmpage != NULL)
1060                 unlock_page(vmpage);
1061         cfio->fault.ft_flags &= ~VM_FAULT_LOCKED;
1062         return result;
1063 }
1064
1065 static int vvp_io_fsync_start(const struct lu_env *env,
1066                               const struct cl_io_slice *ios)
1067 {
1068         /* we should mark TOWRITE bit to each dirty page in radix tree to
1069          * verify pages have been written, but this is difficult because of
1070          * race. */
1071         return 0;
1072 }
1073
1074 static int vvp_io_read_page(const struct lu_env *env,
1075                             const struct cl_io_slice *ios,
1076                             const struct cl_page_slice *slice)
1077 {
1078         struct cl_io              *io     = ios->cis_io;
1079         struct ccc_page           *cp     = cl2ccc_page(slice);
1080         struct cl_page            *page   = slice->cpl_page;
1081         struct inode              *inode  = ccc_object_inode(slice->cpl_obj);
1082         struct ll_sb_info         *sbi    = ll_i2sbi(inode);
1083         struct ll_file_data       *fd     = cl2ccc_io(env, ios)->cui_fd;
1084         struct ll_readahead_state *ras    = &fd->fd_ras;
1085         struct cl_2queue          *queue  = &io->ci_queue;
1086
1087         ENTRY;
1088
1089         if (sbi->ll_ra_info.ra_max_pages_per_file > 0 &&
1090             sbi->ll_ra_info.ra_max_pages > 0)
1091                 ras_update(sbi, inode, ras, ccc_index(cp),
1092                            cp->cpg_defer_uptodate);
1093
1094         if (cp->cpg_defer_uptodate) {
1095                 cp->cpg_ra_used = 1;
1096                 cl_page_export(env, page, 1);
1097         }
1098
1099         /*
1100          * Add page into the queue even when it is marked uptodate above.
1101          * this will unlock it automatically as part of cl_page_list_disown().
1102          */
1103         cl_2queue_add(queue, page);
1104         if (sbi->ll_ra_info.ra_max_pages_per_file > 0 &&
1105             sbi->ll_ra_info.ra_max_pages > 0)
1106                 ll_readahead(env, io, &queue->c2_qin, ras,
1107                              cp->cpg_defer_uptodate);
1108
1109         RETURN(0);
1110 }
1111
1112 static const struct cl_io_operations vvp_io_ops = {
1113         .op = {
1114                 [CIT_READ] = {
1115                         .cio_fini      = vvp_io_read_fini,
1116                         .cio_lock      = vvp_io_read_lock,
1117                         .cio_start     = vvp_io_read_start,
1118                         .cio_advance   = ccc_io_advance
1119                 },
1120                 [CIT_WRITE] = {
1121                         .cio_fini      = vvp_io_fini,
1122                         .cio_iter_init = vvp_io_write_iter_init,
1123                         .cio_iter_fini = vvp_io_write_iter_fini,
1124                         .cio_lock      = vvp_io_write_lock,
1125                         .cio_start     = vvp_io_write_start,
1126                         .cio_advance   = ccc_io_advance
1127                 },
1128                 [CIT_SETATTR] = {
1129                         .cio_fini       = vvp_io_setattr_fini,
1130                         .cio_iter_init  = vvp_io_setattr_iter_init,
1131                         .cio_lock       = vvp_io_setattr_lock,
1132                         .cio_start      = vvp_io_setattr_start,
1133                         .cio_end        = vvp_io_setattr_end
1134                 },
1135                 [CIT_FAULT] = {
1136                         .cio_fini      = vvp_io_fault_fini,
1137                         .cio_iter_init = vvp_io_fault_iter_init,
1138                         .cio_lock      = vvp_io_fault_lock,
1139                         .cio_start     = vvp_io_fault_start,
1140                         .cio_end       = ccc_io_end
1141                 },
1142                 [CIT_FSYNC] = {
1143                         .cio_start  = vvp_io_fsync_start,
1144                         .cio_fini   = vvp_io_fini
1145                 },
1146                 [CIT_MISC] = {
1147                         .cio_fini   = vvp_io_fini
1148                 }
1149         },
1150         .cio_read_page     = vvp_io_read_page,
1151 };
1152
1153 int vvp_io_init(const struct lu_env *env, struct cl_object *obj,
1154                 struct cl_io *io)
1155 {
1156         struct vvp_io      *vio   = vvp_env_io(env);
1157         struct ccc_io      *cio   = ccc_env_io(env);
1158         struct inode       *inode = ccc_object_inode(obj);
1159         int                 result;
1160
1161         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
1162         ENTRY;
1163
1164         CDEBUG(D_VFSTRACE, DFID" ignore/verify layout %d/%d, layout version %d "
1165                            "restore needed %d\n",
1166                PFID(lu_object_fid(&obj->co_lu)),
1167                io->ci_ignore_layout, io->ci_verify_layout,
1168                cio->cui_layout_gen, io->ci_restore_needed);
1169
1170         CL_IO_SLICE_CLEAN(cio, cui_cl);
1171         cl_io_slice_add(io, &cio->cui_cl, obj, &vvp_io_ops);
1172         vio->cui_ra_window_set = 0;
1173         result = 0;
1174         if (io->ci_type == CIT_READ || io->ci_type == CIT_WRITE) {
1175                 size_t count;
1176                 struct ll_inode_info *lli = ll_i2info(inode);
1177
1178                 count = io->u.ci_rw.crw_count;
1179                 /* "If nbyte is 0, read() will return 0 and have no other
1180                  *  results."  -- Single Unix Spec */
1181                 if (count == 0)
1182                         result = 1;
1183                 else {
1184                         cio->cui_tot_count = count;
1185                         cio->cui_tot_nrsegs = 0;
1186                 }
1187
1188                 /* for read/write, we store the jobid in the inode, and
1189                  * it'll be fetched by osc when building RPC.
1190                  *
1191                  * it's not accurate if the file is shared by different
1192                  * jobs.
1193                  */
1194                 lustre_get_jobid(lli->lli_jobid);
1195         } else if (io->ci_type == CIT_SETATTR) {
1196                 if (!cl_io_is_trunc(io))
1197                         io->ci_lockreq = CILR_MANDATORY;
1198         }
1199
1200         /* ignore layout change for generic CIT_MISC but not for glimpse.
1201          * io context for glimpse must set ci_verify_layout to true,
1202          * see cl_glimpse_size0() for details. */
1203         if (io->ci_type == CIT_MISC && !io->ci_verify_layout)
1204                 io->ci_ignore_layout = 1;
1205
1206         /* Enqueue layout lock and get layout version. We need to do this
1207          * even for operations requiring to open file, such as read and write,
1208          * because it might not grant layout lock in IT_OPEN. */
1209         if (result == 0 && !io->ci_ignore_layout) {
1210                 result = ll_layout_refresh(inode, &cio->cui_layout_gen);
1211                 if (result == -ENOENT)
1212                         /* If the inode on MDS has been removed, but the objects
1213                          * on OSTs haven't been destroyed (async unlink), layout
1214                          * fetch will return -ENOENT, we'd ingore this error
1215                          * and continue with dirty flush. LU-3230. */
1216                         result = 0;
1217                 if (result < 0)
1218                         CERROR("%s: refresh file layout " DFID " error %d.\n",
1219                                 ll_get_fsname(inode->i_sb, NULL, 0),
1220                                 PFID(lu_object_fid(&obj->co_lu)), result);
1221         }
1222
1223         RETURN(result);
1224 }
1225
1226 static struct vvp_io *cl2vvp_io(const struct lu_env *env,
1227                                 const struct cl_io_slice *slice)
1228 {
1229         /* Caling just for assertion */
1230         cl2ccc_io(env, slice);
1231         return vvp_env_io(env);
1232 }