lustre/liblustre/rw.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 (c) 2003, 2010, Oracle and/or its affiliates. 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  * lustre/liblustre/rw.c
  37  *
  38  * Lustre Light block IO
  39  */
  40
  41 #define DEBUG_SUBSYSTEM S_LLITE
  42
  43 #include <stdlib.h>
  44 #include <string.h>
  45 #include <assert.h>
  46 #include <time.h>
  47 #include <sys/types.h>
  48 #include <sys/stat.h>
  49 #include <sys/queue.h>
  50 #include <fcntl.h>
  51 #include <sys/uio.h>
  52
  53 #include <sysio.h>
  54 #ifdef HAVE_XTIO_H
  55 #include <xtio.h>
  56 #endif
  57 #include <fs.h>
  58 #include <mount.h>
  59 #include <inode.h>
  60 #ifdef HAVE_FILE_H
  61 #include <file.h>
  62 #endif
  63
  64 #undef LIST_HEAD
  65
  66 #include "llite_lib.h"
  67
  68 struct llu_io_group
  69 {
  70         struct obd_io_group    *lig_oig;
  71         struct inode           *lig_inode;
  72         struct lustre_rw_params *lig_params;
  73         int                     lig_maxpages;
  74         int                     lig_npages;
  75         __u64                   lig_rwcount;
  76         struct ll_async_page   *lig_llaps;
  77         struct page            *lig_pages;
  78         void                   *lig_llap_cookies;
  79 };
  80
  81 #define LLU_IO_GROUP_SIZE(x) \
  82         (sizeof(struct llu_io_group) + \
  83          (sizeof(struct ll_async_page) + \
  84           sizeof(struct page) + \
  85           llap_cookie_size) * (x))
  86
  87 struct llu_io_session
  88 {
  89         struct inode           *lis_inode;
  90         int                     lis_cmd;
  91         int                     lis_max_groups;
  92         int                     lis_ngroups;
  93         struct llu_io_group    *lis_groups[0];
  94 };
  95 #define LLU_IO_SESSION_SIZE(x)  \
  96         (sizeof(struct llu_io_session) + (x) * 2 * sizeof(void *))
  97
  98
  99 typedef ssize_t llu_file_piov_t(const struct iovec *iovec, int iovlen,
 100                                 _SYSIO_OFF_T pos, ssize_t len,
 101                                 void *private);
 102
 103 size_t llap_cookie_size;
 104
 105 static int llu_lock_to_stripe_offset(struct inode *inode, struct ldlm_lock *lock)
 106 {
 107         struct llu_inode_info *lli = llu_i2info(inode);
 108         struct lov_stripe_md *lsm = lli->lli_smd;
 109         struct obd_export *exp = llu_i2obdexp(inode);
 110         struct {
 111                 char name[16];
 112                 struct ldlm_lock *lock;
 113         } key = { .name = KEY_LOCK_TO_STRIPE, .lock = lock };
 114         __u32 stripe, vallen = sizeof(stripe);
 115         int rc;
 116         ENTRY;
 117
 118         if (lsm->lsm_stripe_count == 1)
 119                 RETURN(0);
 120
 121         /* get our offset in the lov */
 122         rc = obd_get_info(exp, sizeof(key), &key, &vallen, &stripe, lsm);
 123         if (rc != 0) {
 124                 CERROR("obd_get_info: rc = %d\n", rc);
 125                 LBUG();
 126         }
 127         LASSERT(stripe < lsm->lsm_stripe_count);
 128         RETURN(stripe);
 129 }
 130
 131 int llu_extent_lock_cancel_cb(struct ldlm_lock *lock,
 132                                     struct ldlm_lock_desc *new, void *data,
 133                                     int flag)
 134 {
 135         struct lustre_handle lockh = { 0 };
 136         int rc;
 137         ENTRY;
 138
 139         if ((unsigned long)data > 0 && (unsigned long)data < 0x1000) {
 140                 LDLM_ERROR(lock, "cancelling lock with bad data %p", data);
 141                 LBUG();
 142         }
 143
 144         switch (flag) {
 145         case LDLM_CB_BLOCKING:
 146                 ldlm_lock2handle(lock, &lockh);
 147                 rc = ldlm_cli_cancel(&lockh);
 148                 if (rc != ELDLM_OK)
 149                         CERROR("ldlm_cli_cancel failed: %d\n", rc);
 150                 break;
 151         case LDLM_CB_CANCELING: {
 152                 struct inode *inode;
 153                 struct llu_inode_info *lli;
 154                 struct lov_stripe_md *lsm;
 155                 __u32 stripe;
 156                 __u64 kms;
 157
 158                 /* This lock wasn't granted, don't try to evict pages */
 159                 if (lock->l_req_mode != lock->l_granted_mode)
 160                         RETURN(0);
 161
 162                 inode = llu_inode_from_lock(lock);
 163                 if (!inode)
 164                         RETURN(0);
 165                 lli= llu_i2info(inode);
 166                 if (!lli)
 167                         goto iput;
 168                 if (!lli->lli_smd)
 169                         goto iput;
 170                 lsm = lli->lli_smd;
 171
 172                 stripe = llu_lock_to_stripe_offset(inode, lock);
 173                 lock_res_and_lock(lock);
 174                 kms = ldlm_extent_shift_kms(lock,
 175                                             lsm->lsm_oinfo[stripe]->loi_kms);
 176                 unlock_res_and_lock(lock);
 177                 if (lsm->lsm_oinfo[stripe]->loi_kms != kms)
 178                         LDLM_DEBUG(lock, "updating kms from "LPU64" to "LPU64,
 179                                    lsm->lsm_oinfo[stripe]->loi_kms, kms);
 180                 lsm->lsm_oinfo[stripe]->loi_kms = kms;
 181 iput:
 182                 I_RELE(inode);
 183                 break;
 184         }
 185         default:
 186                 LBUG();
 187         }
 188
 189         RETURN(0);
 190 }
 191
 192 static int llu_glimpse_callback(struct ldlm_lock *lock, void *reqp)
 193 {
 194         struct ptlrpc_request *req = reqp;
 195         struct inode *inode = llu_inode_from_lock(lock);
 196         struct llu_inode_info *lli;
 197         struct ost_lvb *lvb;
 198         __u32 size[2] = { sizeof(struct ptlrpc_body), sizeof(*lvb) };
 199         int rc, stripe = 0;
 200         ENTRY;
 201
 202         if (inode == NULL)
 203                 GOTO(out, rc = -ELDLM_NO_LOCK_DATA);
 204         lli = llu_i2info(inode);
 205         if (lli == NULL)
 206                 GOTO(iput, rc = -ELDLM_NO_LOCK_DATA);
 207         if (lli->lli_smd == NULL)
 208                 GOTO(iput, rc = -ELDLM_NO_LOCK_DATA);
 209
 210         /* First, find out which stripe index this lock corresponds to. */
 211         if (lli->lli_smd->lsm_stripe_count > 1)
 212                 stripe = llu_lock_to_stripe_offset(inode, lock);
 213
 214         rc = lustre_pack_reply(req, 2, size, NULL);
 215         if (rc)
 216                 GOTO(iput, rc);
 217
 218         lvb = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*lvb));
 219         lvb->lvb_size = lli->lli_smd->lsm_oinfo[stripe]->loi_kms;
 220
 221         LDLM_DEBUG(lock, "i_size: %llu -> stripe number %u -> kms "LPU64,
 222                    (long long)llu_i2stat(inode)->st_size, stripe,lvb->lvb_size);
 223  iput:
 224         I_RELE(inode);
 225  out:
 226         /* These errors are normal races, so we don't want to fill the console
 227          * with messages by calling ptlrpc_error() */
 228         if (rc == -ELDLM_NO_LOCK_DATA)
 229                 lustre_pack_reply(req, 1, NULL, NULL);
 230
 231         req->rq_status = rc;
 232         return rc;
 233 }
 234
 235 /* NB: lov_merge_size will prefer locally cached writes if they extend the
 236  * file (because it prefers KMS over RSS when larger) */
 237 int llu_glimpse_size(struct inode *inode)
 238 {
 239         struct llu_inode_info *lli = llu_i2info(inode);
 240         struct intnl_stat *st = llu_i2stat(inode);
 241         struct llu_sb_info *sbi = llu_i2sbi(inode);
 242         struct lustre_handle lockh = { 0 };
 243         struct ldlm_enqueue_info einfo = { 0 };
 244         struct obd_info oinfo = { { { 0 } } };
 245         struct ost_lvb lvb;
 246         int rc;
 247         ENTRY;
 248
 249         CDEBUG(D_DLMTRACE, "Glimpsing inode %llu\n", (long long)st->st_ino);
 250
 251         if (!lli->lli_smd) {
 252                 CDEBUG(D_DLMTRACE, "No objects for inode %llu\n",
 253                        (long long)st->st_ino);
 254                 RETURN(0);
 255         }
 256
 257         einfo.ei_type = LDLM_EXTENT;
 258         einfo.ei_mode = LCK_PR;
 259         einfo.ei_cb_bl = osc_extent_blocking_cb;
 260         einfo.ei_cb_cp = ldlm_completion_ast;
 261         einfo.ei_cb_gl = llu_glimpse_callback;
 262         einfo.ei_cbdata = inode;
 263
 264         oinfo.oi_policy.l_extent.end = OBD_OBJECT_EOF;
 265         oinfo.oi_lockh = &lockh;
 266         oinfo.oi_md = lli->lli_smd;
 267         oinfo.oi_flags = LDLM_FL_HAS_INTENT;
 268
 269         rc = obd_enqueue_rqset(sbi->ll_osc_exp, &oinfo, &einfo);
 270         if (rc) {
 271                 CERROR("obd_enqueue returned rc %d, returning -EIO\n", rc);
 272                 RETURN(rc > 0 ? -EIO : rc);
 273         }
 274
 275         lov_stripe_lock(lli->lli_smd);
 276         inode_init_lvb(inode, &lvb);
 277         /* merge timestamps the most recently obtained from mds with
 278            timestamps obtained from osts */
 279         lvb = lli->lli_lvb;
 280         rc = obd_merge_lvb(sbi->ll_osc_exp, lli->lli_smd, &lvb, 0);
 281         st->st_size = lvb.lvb_size;
 282         st->st_blocks = lvb.lvb_blocks;
 283         /* handle st_blocks overflow gracefully */
 284         if (st->st_blocks < lvb.lvb_blocks)
 285                 st->st_blocks = ~0UL;
 286         st->st_mtime = lvb.lvb_mtime;
 287         st->st_atime = lvb.lvb_atime;
 288         st->st_ctime = lvb.lvb_ctime;
 289         lov_stripe_unlock(lli->lli_smd);
 290
 291         CDEBUG(D_DLMTRACE, "glimpse: size: "LPU64", blocks: "LPU64"\n",
 292                (__u64)st->st_size, (__u64)st->st_blocks);
 293
 294         RETURN(rc);
 295 }
 296
 297 int llu_extent_lock(struct ll_file_data *fd, struct inode *inode,
 298                     struct lov_stripe_md *lsm, int mode,
 299                     ldlm_policy_data_t *policy, struct lustre_handle *lockh,
 300                     int ast_flags)
 301 {
 302         struct llu_sb_info *sbi = llu_i2sbi(inode);
 303         struct intnl_stat *st = llu_i2stat(inode);
 304         struct ldlm_enqueue_info einfo = { 0 };
 305         struct obd_info oinfo = { { { 0 } } };
 306         struct ost_lvb lvb;
 307         int rc;
 308         ENTRY;
 309
 310         LASSERT(!lustre_handle_is_used(lockh));
 311         CLASSERT(ELDLM_OK == 0);
 312
 313         /* XXX phil: can we do this?  won't it screw the file size up? */
 314         if ((fd && (fd->fd_flags & LL_FILE_IGNORE_LOCK)) ||
 315             (sbi->ll_flags & LL_SBI_NOLCK) || mode == LCK_NL)
 316                 RETURN(0);
 317
 318         CDEBUG(D_DLMTRACE, "Locking inode %llu, start "LPU64" end "LPU64"\n",
 319                (long long)st->st_ino, policy->l_extent.start,
 320                policy->l_extent.end);
 321
 322         einfo.ei_type = LDLM_EXTENT;
 323         einfo.ei_mode = mode;
 324         einfo.ei_cb_bl = osc_extent_blocking_cb;
 325         einfo.ei_cb_cp = ldlm_completion_ast;
 326         einfo.ei_cb_gl = llu_glimpse_callback;
 327         einfo.ei_cbdata = inode;
 328
 329         oinfo.oi_policy = *policy;
 330         oinfo.oi_lockh = lockh;
 331         oinfo.oi_md = lsm;
 332         oinfo.oi_flags = ast_flags;
 333
 334         rc = obd_enqueue(sbi->ll_osc_exp, &oinfo, &einfo, NULL);
 335         *policy = oinfo.oi_policy;
 336         if (rc > 0)
 337                 rc = -EIO;
 338
 339         inode_init_lvb(inode, &lvb);
 340         obd_merge_lvb(sbi->ll_osc_exp, lsm, &lvb, 1);
 341         if (policy->l_extent.start == 0 &&
 342             policy->l_extent.end == OBD_OBJECT_EOF)
 343                 st->st_size = lvb.lvb_size;
 344
 345         if (rc == 0) {
 346                 st->st_mtime = lvb.lvb_mtime;
 347                 st->st_atime = lvb.lvb_atime;
 348                 st->st_ctime = lvb.lvb_ctime;
 349         }
 350
 351         RETURN(rc);
 352 }
 353
 354 int llu_extent_unlock(struct ll_file_data *fd, struct inode *inode,
 355                 struct lov_stripe_md *lsm, int mode,
 356                 struct lustre_handle *lockh)
 357 {
 358         struct llu_sb_info *sbi = llu_i2sbi(inode);
 359         int rc;
 360         ENTRY;
 361
 362         CLASSERT(ELDLM_OK == 0);
 363
 364         /* XXX phil: can we do this?  won't it screw the file size up? */
 365         if ((fd && (fd->fd_flags & LL_FILE_IGNORE_LOCK)) ||
 366             (sbi->ll_flags & LL_SBI_NOLCK) || mode == LCK_NL)
 367                 RETURN(0);
 368
 369         rc = obd_cancel(sbi->ll_osc_exp, lsm, mode, lockh, 0, 0);
 370
 371         RETURN(rc);
 372 }
 373
 374 #define LLAP_MAGIC 12346789
 375
 376 struct ll_async_page {
 377         int             llap_magic;
 378         void           *llap_cookie;
 379         int             llap_queued;
 380         struct page    *llap_page;
 381         struct inode   *llap_inode;
 382 };
 383
 384 static void llu_ap_fill_obdo(void *data, int cmd, struct obdo *oa)
 385 {
 386         struct ll_async_page *llap;
 387         struct inode *inode;
 388         struct lov_stripe_md *lsm;
 389         obd_flag valid_flags;
 390         ENTRY;
 391
 392         llap = LLAP_FROM_COOKIE(data);
 393         inode = llap->llap_inode;
 394         lsm = llu_i2info(inode)->lli_smd;
 395
 396         oa->o_id = lsm->lsm_object_id;
 397         oa->o_valid = OBD_MD_FLID;
 398         valid_flags = OBD_MD_FLTYPE | OBD_MD_FLATIME;
 399         if (cmd & OBD_BRW_WRITE)
 400                 valid_flags |= OBD_MD_FLMTIME | OBD_MD_FLCTIME |
 401                         OBD_MD_FLUID | OBD_MD_FLGID |
 402                         OBD_MD_FLFID | OBD_MD_FLGENER;
 403
 404         obdo_from_inode(oa, inode, valid_flags);
 405         EXIT;
 406 }
 407
 408 static void llu_ap_update_obdo(void *data, int cmd, struct obdo *oa,
 409                                obd_valid valid)
 410 {
 411         struct ll_async_page *llap;
 412         ENTRY;
 413
 414         llap = LLAP_FROM_COOKIE(data);
 415         obdo_from_inode(oa, llap->llap_inode, valid);
 416
 417         EXIT;
 418 }
 419
 420 /* called for each page in a completed rpc.*/
 421 static int llu_ap_completion(void *data, int cmd, struct obdo *oa, int rc)
 422 {
 423         struct ll_async_page *llap;
 424         struct page *page;
 425         ENTRY;
 426
 427         llap = LLAP_FROM_COOKIE(data);
 428         llap->llap_queued = 0;
 429         page = llap->llap_page;
 430
 431         if (rc != 0) {
 432                 if (cmd & OBD_BRW_WRITE)
 433                         CERROR("writeback error on page %p index %ld: %d\n",
 434                                page, page->index, rc);
 435         }
 436         RETURN(0);
 437 }
 438
 439 static struct obd_async_page_ops llu_async_page_ops = {
 440         .ap_make_ready =        NULL,
 441         .ap_refresh_count =     NULL,
 442         .ap_fill_obdo =         llu_ap_fill_obdo,
 443         .ap_update_obdo =       llu_ap_update_obdo,
 444         .ap_completion =        llu_ap_completion,
 445 };
 446
 447 static int llu_queue_pio(int cmd, struct llu_io_group *group,
 448                          char *buf, size_t count, loff_t pos)
 449 {
 450         struct llu_inode_info *lli = llu_i2info(group->lig_inode);
 451         struct intnl_stat *st = llu_i2stat(group->lig_inode);
 452         struct lov_stripe_md *lsm = lli->lli_smd;
 453         struct obd_export *exp = llu_i2obdexp(group->lig_inode);
 454         struct page *pages = &group->lig_pages[group->lig_npages],*page = pages;
 455         struct ll_async_page *llap = &group->lig_llaps[group->lig_npages];
 456         void *llap_cookie = group->lig_llap_cookies +
 457                 llap_cookie_size * group->lig_npages;
 458         int i, rc, npages = 0, ret_bytes = 0;
 459         int local_lock;
 460         ENTRY;
 461
 462         if (!exp)
 463                 RETURN(-EINVAL);
 464
 465         local_lock = group->lig_params->lrp_lock_mode != LCK_NL;
 466         /* prepare the pages array */
 467         do {
 468                 unsigned long index, offset, bytes;
 469
 470                 offset = (pos & ~CFS_PAGE_MASK);
 471                 index = pos >> CFS_PAGE_SHIFT;
 472                 bytes = CFS_PAGE_SIZE - offset;
 473                 if (bytes > count)
 474                         bytes = count;
 475
 476                 /* prevent read beyond file range */
 477                 if (/* local_lock && */
 478                     cmd == OBD_BRW_READ && pos + bytes >= st->st_size) {
 479                         if (pos >= st->st_size)
 480                                 break;
 481                         bytes = st->st_size - pos;
 482                 }
 483
 484                 /* prepare page for this index */
 485                 page->index = index;
 486                 page->addr = buf - offset;
 487
 488                 page->_offset = offset;
 489                 page->_count = bytes;
 490
 491                 page++;
 492                 npages++;
 493                 count -= bytes;
 494                 pos += bytes;
 495                 buf += bytes;
 496
 497                 group->lig_rwcount += bytes;
 498                 ret_bytes += bytes;
 499         } while (count);
 500
 501         group->lig_npages += npages;
 502
 503         for (i = 0, page = pages; i < npages;
 504              i++, page++, llap++, llap_cookie += llap_cookie_size){
 505                 llap->llap_magic = LLAP_MAGIC;
 506                 llap->llap_cookie = llap_cookie;
 507                 rc = obd_prep_async_page(exp, lsm, NULL, page,
 508                                          (obd_off)page->index << CFS_PAGE_SHIFT,
 509                                          &llu_async_page_ops,
 510                                          llap, &llap->llap_cookie,
 511                                          /* no cache in liblustre at all */
 512                                          OBD_PAGE_NO_CACHE,
 513                                          NULL);
 514                 if (rc) {
 515                         LASSERT(rc < 0);
 516                         llap->llap_cookie = NULL;
 517                         RETURN(rc);
 518                 }
 519                 CDEBUG(D_CACHE, "llap %p page %p group %p obj off "LPU64"\n",
 520                        llap, page, llap->llap_cookie,
 521                        (obd_off)pages->index << CFS_PAGE_SHIFT);
 522                 page->private = (unsigned long)llap;
 523                 llap->llap_page = page;
 524                 llap->llap_inode = group->lig_inode;
 525
 526                 rc = obd_queue_group_io(exp, lsm, NULL, group->lig_oig,
 527                                         llap->llap_cookie, cmd,
 528                                         page->_offset, page->_count,
 529                                         group->lig_params->lrp_brw_flags,
 530                                         ASYNC_READY | ASYNC_URGENT |
 531                                         ASYNC_COUNT_STABLE | ASYNC_GROUP_SYNC);
 532                 if (!local_lock && cmd == OBD_BRW_READ) {
 533                         /*
 534                          * In OST-side locking case short reads cannot be
 535                          * detected properly.
 536                          *
 537                          * The root of the problem is that
 538                          *
 539                          * kms = lov_merge_size(lsm, 1);
 540                          * if (end >= kms)
 541                          *         glimpse_size(inode);
 542                          * else
 543                          *         st->st_size = kms;
 544                          *
 545                          * logic in the read code (both llite and liblustre)
 546                          * only works correctly when client holds DLM lock on
 547                          * [start, end]. Without DLM lock KMS can be
 548                          * completely out of date, and client can either make
 549                          * spurious short-read (missing concurrent write), or
 550                          * return stale data (missing concurrent
 551                          * truncate). For llite client this is fatal, because
 552                          * incorrect data are cached and can be later sent
 553                          * back to the server (vide bug 5047). This is hard to
 554                          * fix by handling short-reads on the server, as there
 555                          * is no easy way to communicate file size (or amount
 556                          * of bytes read/written) back to the client,
 557                          * _especially_ because OSC pages can be sliced and
 558                          * dices into multiple RPCs arbitrary. Fortunately,
 559                          * liblustre doesn't cache data and the worst case is
 560                          * that we get race with concurrent write or truncate.
 561                          */
 562                 }
 563                 if (rc) {
 564                         LASSERT(rc < 0);
 565                         RETURN(rc);
 566                 }
 567
 568                 llap->llap_queued = 1;
 569         }
 570
 571         RETURN(ret_bytes);
 572 }
 573
 574 static
 575 struct llu_io_group * get_io_group(struct inode *inode, int maxpages,
 576                                    struct lustre_rw_params *params)
 577 {
 578         struct llu_io_group *group;
 579         int rc;
 580
 581         if (!llap_cookie_size)
 582                 llap_cookie_size = obd_prep_async_page(llu_i2obdexp(inode),
 583                                                        NULL, NULL, NULL, 0,
 584                                                        NULL, NULL, NULL, 0,
 585                                                        NULL);
 586
 587         OBD_ALLOC(group, LLU_IO_GROUP_SIZE(maxpages));
 588         if (!group)
 589                 return ERR_PTR(-ENOMEM);
 590
 591         I_REF(inode);
 592         group->lig_inode = inode;
 593         group->lig_maxpages = maxpages;
 594         group->lig_params = params;
 595         group->lig_llaps = (struct ll_async_page *)(group + 1);
 596         group->lig_pages = (struct page *)(&group->lig_llaps[maxpages]);
 597         group->lig_llap_cookies = (void *)(&group->lig_pages[maxpages]);
 598
 599         rc = oig_init(&group->lig_oig);
 600         if (rc) {
 601                 OBD_FREE(group, LLU_IO_GROUP_SIZE(maxpages));
 602                 return ERR_PTR(rc);
 603         }
 604
 605         return group;
 606 }
 607
 608 static int max_io_pages(ssize_t len, int iovlen)
 609 {
 610         return (((len + CFS_PAGE_SIZE -1) >> CFS_PAGE_SHIFT) + 2 + iovlen - 1);
 611 }
 612
 613 static
 614 void put_io_group(struct llu_io_group *group)
 615 {
 616         struct lov_stripe_md *lsm = llu_i2info(group->lig_inode)->lli_smd;
 617         struct obd_export *exp = llu_i2obdexp(group->lig_inode);
 618         struct ll_async_page *llap = group->lig_llaps;
 619         int i;
 620
 621         for (i = 0; i < group->lig_npages; i++, llap++) {
 622                 if (llap->llap_cookie)
 623                         obd_teardown_async_page(exp, lsm, NULL,
 624                                                 llap->llap_cookie);
 625         }
 626
 627         I_RELE(group->lig_inode);
 628
 629         oig_release(group->lig_oig);
 630         OBD_FREE(group, LLU_IO_GROUP_SIZE(group->lig_maxpages));
 631 }
 632
 633 static
 634 ssize_t llu_file_prwv(const struct iovec *iovec, int iovlen,
 635                         _SYSIO_OFF_T pos, ssize_t len,
 636                         void *private)
 637 {
 638         struct llu_io_session *session = (struct llu_io_session *) private;
 639         struct inode *inode = session->lis_inode;
 640         struct llu_inode_info *lli = llu_i2info(inode);
 641         struct intnl_stat *st = llu_i2stat(inode);
 642         struct ll_file_data *fd = lli->lli_file_data;
 643         struct lustre_handle lockh = {0};
 644         struct lov_stripe_md *lsm = lli->lli_smd;
 645         struct obd_export *exp = NULL;
 646         struct llu_io_group *iogroup;
 647         struct lustre_rw_params p;
 648         struct ost_lvb lvb;
 649         __u64 kms;
 650         int err, is_read, iovidx, ret;
 651         int local_lock;
 652         ssize_t ret_len = len;
 653         ENTRY;
 654
 655         /* in a large iov read/write we'll be repeatedly called.
 656          * so give a chance to answer cancel ast here
 657          */
 658         liblustre_wait_event(0);
 659
 660         exp = llu_i2obdexp(inode);
 661         if (exp == NULL)
 662                 RETURN(-EINVAL);
 663
 664         if (len == 0 || iovlen == 0)
 665                 RETURN(0);
 666
 667         if (pos + len > lli->lli_maxbytes)
 668                 RETURN(-ERANGE);
 669
 670         lustre_build_lock_params(session->lis_cmd, lli->lli_open_flags,
 671                                  lli->lli_sbi->ll_lco.lco_flags,
 672                                  pos, len, &p);
 673
 674         iogroup = get_io_group(inode, max_io_pages(len, iovlen), &p);
 675         if (IS_ERR(iogroup))
 676                 RETURN(PTR_ERR(iogroup));
 677
 678         local_lock = p.lrp_lock_mode != LCK_NL;
 679
 680         err = llu_extent_lock(fd, inode, lsm, p.lrp_lock_mode, &p.lrp_policy,
 681                               &lockh, p.lrp_ast_flags);
 682         if (err != ELDLM_OK)
 683                 GOTO(err_put, err);
 684
 685         is_read = (session->lis_cmd == OBD_BRW_READ);
 686         if (is_read) {
 687                 /*
 688                  * If OST-side locking is used, KMS can be completely out of
 689                  * date, and, hence, cannot be used for short-read
 690                  * detection. Rely in OST to handle short reads in that case.
 691                  */
 692                 inode_init_lvb(inode, &lvb);
 693                 obd_merge_lvb(exp, lsm, &lvb, 1);
 694                 kms = lvb.lvb_size;
 695                 /* extent.end is last byte of the range */
 696                 if (p.lrp_policy.l_extent.end >= kms) {
 697                         /* A glimpse is necessary to determine whether
 698                          * we return a short read or some zeroes at
 699                          * the end of the buffer
 700                          *
 701                          * In the case of OST-side locking KMS can be
 702                          * completely out of date and short-reads maybe
 703                          * mishandled. See llu_queue_pio() for more detailed
 704                          * comment.
 705                          */
 706                         if ((err = llu_glimpse_size(inode))) {
 707                                 GOTO(err_unlock, err);
 708                         } else {
 709                                 /* If objective page index exceed end-of-file
 710                                  * page index, return directly. --bug 17336 */
 711                                 loff_t size = st->st_size;
 712                                 unsigned long cur_index = pos >> CFS_PAGE_SHIFT;
 713
 714                                 if ((size == 0 && cur_index != 0) ||
 715                                     (((size - 1) >> CFS_PAGE_SHIFT) < cur_index))
 716                                         GOTO(err_unlock, err);
 717                         }
 718                 } else {
 719                         st->st_size = kms;
 720                 }
 721         } else if (lli->lli_open_flags & O_APPEND) {
 722                 pos = st->st_size;
 723         }
 724
 725         if (local_lock) {
 726                 struct ost_lvb xtimes;
 727
 728                 lov_stripe_lock(lsm);
 729                 /* inode might mtime and ctime set earlier in race with stat
 730                  * which merged into inode timestamps obtained from mds and
 731                  * osts */
 732                 st->st_atime = st->st_mtime = st->st_ctime = CURRENT_TIME;
 733                 xtimes.lvb_atime = st->st_atime;
 734                 xtimes.lvb_mtime = st->st_mtime;
 735                 xtimes.lvb_ctime = st->st_ctime;
 736                 obd_update_lvb(exp, lsm, &xtimes,
 737                                is_read ? OBD_MD_FLATIME :
 738                                (OBD_MD_FLMTIME | OBD_MD_FLCTIME));
 739                 lov_stripe_unlock(lsm);
 740         }
 741
 742         for (iovidx = 0; iovidx < iovlen; iovidx++) {
 743                 char *buf = (char *) iovec[iovidx].iov_base;
 744                 size_t count = iovec[iovidx].iov_len;
 745
 746                 if (!count)
 747                         continue;
 748                 if (len < count)
 749                         count = len;
 750                 if (IS_BAD_PTR(buf) || IS_BAD_PTR(buf + count)) {
 751                         GOTO(err_unlock, err = -EFAULT);
 752                 }
 753
 754                 if (is_read) {
 755                         if (/* local_lock && */ pos >= st->st_size)
 756                                 break;
 757                 } else {
 758                         if (pos >= lli->lli_maxbytes) {
 759                                 GOTO(err_unlock, err = -EFBIG);
 760                         }
 761                         if (pos + count >= lli->lli_maxbytes)
 762                                 count = lli->lli_maxbytes - pos;
 763                 }
 764
 765                 ret = llu_queue_pio(session->lis_cmd, iogroup, buf, count, pos);
 766                 if (ret < 0) {
 767                         GOTO(err_unlock, err = ret);
 768                 } else {
 769                         pos += ret;
 770                         if (!is_read) {
 771                                 LASSERT(ret == count);
 772                                 obd_adjust_kms(exp, lsm, pos, 0);
 773                                 /* file size grow immediately */
 774                                 if (pos > st->st_size)
 775                                         st->st_size = pos;
 776                         }
 777                         len -= ret;
 778                         if (!len)
 779                                 break;
 780                 }
 781         }
 782         LASSERT(len == 0 || is_read); /* libsysio should guarantee this */
 783
 784         err = obd_trigger_group_io(exp, lsm, NULL, iogroup->lig_oig);
 785         if (err)
 786                 GOTO(err_unlock, err);
 787
 788         err = oig_wait(iogroup->lig_oig);
 789         if (err) {
 790                 CERROR("%s error: %s\n", is_read ? "read" : "write", strerror(-err));
 791                 GOTO(err_unlock, err);
 792         }
 793
 794         ret = llu_extent_unlock(fd, inode, lsm, p.lrp_lock_mode, &lockh);
 795         if (ret)
 796                 CERROR("extent unlock error %d\n", ret);
 797
 798         session->lis_groups[session->lis_ngroups++] = iogroup;
 799         RETURN(ret_len);
 800
 801 err_unlock:
 802         llu_extent_unlock(fd, inode, lsm, p.lrp_lock_mode, &lockh);
 803 err_put:
 804         put_io_group(iogroup);
 805         RETURN((ssize_t)err);
 806 }
 807
 808 static
 809 struct llu_io_session *get_io_session(struct inode *ino, int ngroups, int cmd)
 810 {
 811         struct llu_io_session *session;
 812
 813         OBD_ALLOC(session, LLU_IO_SESSION_SIZE(ngroups));
 814         if (!session)
 815                 return NULL;
 816
 817         I_REF(ino);
 818         session->lis_inode = ino;
 819         session->lis_max_groups = ngroups;
 820         session->lis_cmd = cmd;
 821         return session;
 822 }
 823
 824 static void put_io_session(struct llu_io_session *session)
 825 {
 826         int i;
 827
 828         for (i = 0; i < session->lis_ngroups; i++) {
 829                 if (session->lis_groups[i]) {
 830                         put_io_group(session->lis_groups[i]);
 831                         session->lis_groups[i] = NULL;
 832                 }
 833         }
 834
 835         I_RELE(session->lis_inode);
 836         OBD_FREE(session, LLU_IO_SESSION_SIZE(session->lis_max_groups));
 837 }
 838
 839 static int llu_file_rwx(struct inode *ino,
 840                         struct ioctx *ioctx,
 841                         int read)
 842 {
 843         struct llu_io_session *session;
 844         ssize_t cc;
 845         int cmd = read ? OBD_BRW_READ : OBD_BRW_WRITE;
 846         ENTRY;
 847
 848         LASSERT(ioctx->ioctx_xtvlen >= 0);
 849         LASSERT(ioctx->ioctx_iovlen >= 0);
 850
 851         liblustre_wait_event(0);
 852
 853         if (!ioctx->ioctx_xtvlen)
 854                 RETURN(0);
 855
 856         /* XXX consider other types later */
 857         if (S_ISDIR(llu_i2stat(ino)->st_mode))
 858                 RETURN(-EISDIR);
 859         if (!S_ISREG(llu_i2stat(ino)->st_mode))
 860                 RETURN(-EOPNOTSUPP);
 861
 862         session = get_io_session(ino, ioctx->ioctx_xtvlen * 2, cmd);
 863         if (!session)
 864                 RETURN(-ENOMEM);
 865
 866         cc = _sysio_enumerate_extents(ioctx->ioctx_xtv, ioctx->ioctx_xtvlen,
 867                                       ioctx->ioctx_iov, ioctx->ioctx_iovlen,
 868                                       llu_file_prwv, session);
 869
 870         if (cc >= 0) {
 871                 LASSERT(!ioctx->ioctx_cc);
 872                 ioctx->ioctx_private = session;
 873                 cc = 0;
 874         } else {
 875                 put_io_session(session);
 876         }
 877
 878         liblustre_wait_event(0);
 879         RETURN(cc);
 880 }
 881
 882 int llu_iop_read(struct inode *ino,
 883                  struct ioctx *ioctx)
 884 {
 885         /* BUG: 5972 */
 886         struct intnl_stat *st = llu_i2stat(ino);
 887         st->st_atime = CURRENT_TIME;
 888
 889         return llu_file_rwx(ino, ioctx, 1);
 890 }
 891
 892 int llu_iop_write(struct inode *ino,
 893                   struct ioctx *ioctx)
 894 {
 895         struct intnl_stat *st = llu_i2stat(ino);
 896         st->st_mtime = st->st_ctime = CURRENT_TIME;
 897
 898         return llu_file_rwx(ino, ioctx, 0);
 899 }
 900
 901 int llu_iop_iodone(struct ioctx *ioctx)
 902 {
 903         struct llu_io_session *session;
 904         struct llu_io_group *group;
 905         int i, err = 0, rc = 0;
 906         ENTRY;
 907
 908         liblustre_wait_event(0);
 909
 910         session = (struct llu_io_session *) ioctx->ioctx_private;
 911         LASSERT(session);
 912         LASSERT(!IS_ERR(session));
 913
 914         for (i = 0; i < session->lis_ngroups; i++) {
 915                 group = session->lis_groups[i];
 916                 if (group) {
 917                         if (!rc) {
 918                                 err = oig_wait(group->lig_oig);
 919                                 if (err)
 920                                         rc = err;
 921                         }
 922                         if (!rc)
 923                                 ioctx->ioctx_cc += group->lig_rwcount;
 924                         put_io_group(group);
 925                         session->lis_groups[i] = NULL;
 926                 }
 927         }
 928
 929         if (rc) {
 930                 LASSERT(rc < 0);
 931                 ioctx->ioctx_cc = -1;
 932                 ioctx->ioctx_errno = -rc;
 933         }
 934
 935         put_io_session(session);
 936         ioctx->ioctx_private = NULL;
 937         liblustre_wait_event(0);
 938
 939         RETURN(1);
 940 }