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  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  * 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         cfs_page_t             *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(cfs_page_t) + \
  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         int rc, stripe = 0;
 199         ENTRY;
 200
 201         if (inode == NULL)
 202                 GOTO(out, rc = -ELDLM_NO_LOCK_DATA);
 203         lli = llu_i2info(inode);
 204         if (lli == NULL)
 205                 GOTO(iput, rc = -ELDLM_NO_LOCK_DATA);
 206         if (lli->lli_smd == NULL)
 207                 GOTO(iput, rc = -ELDLM_NO_LOCK_DATA);
 208
 209         /* First, find out which stripe index this lock corresponds to. */
 210         if (lli->lli_smd->lsm_stripe_count > 1)
 211                 stripe = llu_lock_to_stripe_offset(inode, lock);
 212
 213         req_capsule_extend(&req->rq_pill, &RQF_LDLM_GL_CALLBACK);
 214         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
 215                              sizeof(*lvb));
 216         rc = req_capsule_server_pack(&req->rq_pill);
 217         if (rc) {
 218                 CERROR("failed pack reply: %d\n", rc);
 219                 GOTO(iput, rc);
 220         }
 221
 222         lvb = req_capsule_server_get(&req->rq_pill, &RMF_DLM_LVB);
 223         lvb->lvb_size = lli->lli_smd->lsm_oinfo[stripe]->loi_kms;
 224
 225         LDLM_DEBUG(lock, "i_size: "LPU64" -> stripe number %u -> kms "LPU64,
 226                    (__u64)llu_i2stat(inode)->st_size, stripe,lvb->lvb_size);
 227  iput:
 228         I_RELE(inode);
 229  out:
 230         /* These errors are normal races, so we don't want to fill the console
 231          * with messages by calling ptlrpc_error() */
 232         if (rc == -ELDLM_NO_LOCK_DATA)
 233                 lustre_pack_reply(req, 1, NULL, NULL);
 234
 235         req->rq_status = rc;
 236         return rc;
 237 }
 238
 239 static int llu_merge_lvb(struct inode *inode)
 240 {
 241         struct llu_inode_info *lli = llu_i2info(inode);
 242         struct llu_sb_info *sbi = llu_i2sbi(inode);
 243         struct intnl_stat *st = llu_i2stat(inode);
 244         struct ost_lvb lvb;
 245         int rc;
 246         ENTRY;
 247
 248         inode_init_lvb(inode, &lvb);
 249         rc = obd_merge_lvb(sbi->ll_dt_exp, lli->lli_smd, &lvb, 0);
 250         st->st_size = lvb.lvb_size;
 251         st->st_blocks = lvb.lvb_blocks;
 252         /* handle st_blocks overflow gracefully */
 253         if (st->st_blocks < lvb.lvb_blocks)
 254                 st->st_blocks = ~0UL;
 255         st->st_mtime = lvb.lvb_mtime;
 256         st->st_atime = lvb.lvb_atime;
 257         st->st_ctime = lvb.lvb_ctime;
 258
 259         RETURN(rc);
 260 }
 261
 262 int llu_local_size(struct inode *inode)
 263 {
 264         ldlm_policy_data_t policy = { .l_extent = { 0, OBD_OBJECT_EOF } };
 265         struct llu_inode_info *lli = llu_i2info(inode);
 266         struct llu_sb_info *sbi = llu_i2sbi(inode);
 267         struct lustre_handle lockh = { 0 };
 268         int flags = 0;
 269         int rc;
 270         ENTRY;
 271
 272         if (lli->lli_smd->lsm_stripe_count == 0)
 273                 RETURN(0);
 274
 275         rc = obd_match(sbi->ll_dt_exp, lli->lli_smd, LDLM_EXTENT,
 276                        &policy, LCK_PR, &flags, inode, &lockh);
 277         if (rc < 0)
 278                 RETURN(rc);
 279         else if (rc == 0)
 280                 RETURN(-ENODATA);
 281
 282         rc = llu_merge_lvb(inode);
 283         obd_cancel(sbi->ll_dt_exp, lli->lli_smd, LCK_PR, &lockh);
 284         RETURN(rc);
 285 }
 286
 287 /* NB: lov_merge_size will prefer locally cached writes if they extend the
 288  * file (because it prefers KMS over RSS when larger) */
 289 int llu_glimpse_size(struct inode *inode)
 290 {
 291         struct llu_inode_info *lli = llu_i2info(inode);
 292         struct intnl_stat *st = llu_i2stat(inode);
 293         struct llu_sb_info *sbi = llu_i2sbi(inode);
 294         struct lustre_handle lockh = { 0 };
 295         struct ldlm_enqueue_info einfo = { 0 };
 296         struct obd_info oinfo = { { { 0 } } };
 297         int rc;
 298         ENTRY;
 299
 300         /* If size is cached on the mds, skip glimpse. */
 301         if (lli->lli_flags & LLIF_MDS_SIZE_LOCK)
 302                 RETURN(0);
 303
 304         CDEBUG(D_DLMTRACE, "Glimpsing inode "LPU64"\n", (__u64)st->st_ino);
 305
 306         if (!lli->lli_smd) {
 307                 CDEBUG(D_DLMTRACE, "No objects for inode "LPU64"\n",
 308                        (__u64)st->st_ino);
 309                 RETURN(0);
 310         }
 311
 312         einfo.ei_type = LDLM_EXTENT;
 313         einfo.ei_mode = LCK_PR;
 314         einfo.ei_cb_bl = osc_extent_blocking_cb;
 315         einfo.ei_cb_cp = ldlm_completion_ast;
 316         einfo.ei_cb_gl = llu_glimpse_callback;
 317         einfo.ei_cbdata = inode;
 318
 319         oinfo.oi_policy.l_extent.end = OBD_OBJECT_EOF;
 320         oinfo.oi_lockh = &lockh;
 321         oinfo.oi_md = lli->lli_smd;
 322         oinfo.oi_flags = LDLM_FL_HAS_INTENT;
 323
 324         rc = obd_enqueue_rqset(sbi->ll_dt_exp, &oinfo, &einfo);
 325         if (rc) {
 326                 CERROR("obd_enqueue returned rc %d, returning -EIO\n", rc);
 327                 RETURN(rc > 0 ? -EIO : rc);
 328         }
 329
 330         rc = llu_merge_lvb(inode);
 331         CDEBUG(D_DLMTRACE, "glimpse: size: "LPU64", blocks: "LPU64"\n",
 332                (__u64)st->st_size, (__u64)st->st_blocks);
 333
 334         RETURN(rc);
 335 }
 336
 337 int llu_extent_lock(struct ll_file_data *fd, struct inode *inode,
 338                     struct lov_stripe_md *lsm, int mode,
 339                     ldlm_policy_data_t *policy, struct lustre_handle *lockh,
 340                     int ast_flags)
 341 {
 342         struct llu_sb_info *sbi = llu_i2sbi(inode);
 343         struct intnl_stat *st = llu_i2stat(inode);
 344         struct ldlm_enqueue_info einfo = { 0 };
 345         struct obd_info oinfo = { { { 0 } } };
 346         struct ost_lvb lvb;
 347         int rc;
 348         ENTRY;
 349
 350         LASSERT(!lustre_handle_is_used(lockh));
 351         CLASSERT(ELDLM_OK == 0);
 352
 353         /* XXX phil: can we do this?  won't it screw the file size up? */
 354         if ((fd && (fd->fd_flags & LL_FILE_IGNORE_LOCK)) ||
 355             (sbi->ll_flags & LL_SBI_NOLCK) || mode == LCK_NL)
 356                 RETURN(0);
 357
 358         CDEBUG(D_DLMTRACE, "Locking inode %llu, start "LPU64" end "LPU64"\n",
 359                (unsigned long long)st->st_ino, policy->l_extent.start,
 360                policy->l_extent.end);
 361
 362         einfo.ei_type = LDLM_EXTENT;
 363         einfo.ei_mode = mode;
 364         einfo.ei_cb_bl = osc_extent_blocking_cb;
 365         einfo.ei_cb_cp = ldlm_completion_ast;
 366         einfo.ei_cb_gl = llu_glimpse_callback;
 367         einfo.ei_cbdata = inode;
 368
 369         oinfo.oi_policy = *policy;
 370         oinfo.oi_lockh = lockh;
 371         oinfo.oi_md = lsm;
 372         oinfo.oi_flags = ast_flags;
 373
 374         rc = obd_enqueue(sbi->ll_dt_exp, &oinfo, &einfo, NULL);
 375         *policy = oinfo.oi_policy;
 376         if (rc > 0)
 377                 rc = -EIO;
 378
 379         inode_init_lvb(inode, &lvb);
 380         obd_merge_lvb(sbi->ll_dt_exp, lsm, &lvb, 1);
 381         if (policy->l_extent.start == 0 &&
 382             policy->l_extent.end == OBD_OBJECT_EOF)
 383                 st->st_size = lvb.lvb_size;
 384
 385         if (rc == 0) {
 386                 st->st_mtime = lvb.lvb_mtime;
 387                 st->st_atime = lvb.lvb_atime;
 388                 st->st_ctime = lvb.lvb_ctime;
 389         }
 390
 391         RETURN(rc);
 392 }
 393
 394 int llu_extent_unlock(struct ll_file_data *fd, struct inode *inode,
 395                 struct lov_stripe_md *lsm, int mode,
 396                 struct lustre_handle *lockh)
 397 {
 398         struct llu_sb_info *sbi = llu_i2sbi(inode);
 399         int rc;
 400         ENTRY;
 401
 402         CLASSERT(ELDLM_OK == 0);
 403
 404         /* XXX phil: can we do this?  won't it screw the file size up? */
 405         if ((fd && (fd->fd_flags & LL_FILE_IGNORE_LOCK)) ||
 406             (sbi->ll_flags & LL_SBI_NOLCK) || mode == LCK_NL)
 407                 RETURN(0);
 408
 409         rc = obd_cancel(sbi->ll_dt_exp, lsm, mode, lockh);
 410
 411         RETURN(rc);
 412 }
 413
 414 #define LLAP_MAGIC 12346789
 415
 416 struct ll_async_page {
 417         int             llap_magic;
 418         void           *llap_cookie;
 419         int             llap_queued;
 420         cfs_page_t     *llap_page;
 421         struct inode   *llap_inode;
 422 };
 423
 424 static inline struct ll_async_page *llap_from_cookie(void *ptr)
 425 {
 426         struct ll_async_page *ap = ptr;
 427         LASSERT(ap->llap_magic == LLAP_MAGIC);
 428         return ap;
 429 }
 430
 431 static void llu_ap_fill_obdo(void *data, int cmd, struct obdo *oa)
 432 {
 433         struct ll_async_page *llap;
 434         struct inode *inode;
 435         struct lov_stripe_md *lsm;
 436         obd_flag valid_flags;
 437         ENTRY;
 438
 439         llap = llap_from_cookie(data);
 440         inode = llap->llap_inode;
 441         lsm = llu_i2info(inode)->lli_smd;
 442
 443         oa->o_id = lsm->lsm_object_id;
 444         oa->o_valid = OBD_MD_FLID;
 445         valid_flags = OBD_MD_FLTYPE | OBD_MD_FLATIME;
 446         if (cmd & OBD_BRW_WRITE)
 447                 valid_flags |= OBD_MD_FLMTIME | OBD_MD_FLCTIME |
 448                         OBD_MD_FLUID | OBD_MD_FLGID |
 449                         OBD_MD_FLFID | OBD_MD_FLGENER;
 450
 451         obdo_from_inode(oa, inode, valid_flags);
 452         EXIT;
 453 }
 454
 455 static void llu_ap_update_obdo(void *data, int cmd, struct obdo *oa,
 456                                obd_valid valid)
 457 {
 458         struct ll_async_page *llap;
 459         ENTRY;
 460
 461         llap = llap_from_cookie(data);
 462         obdo_from_inode(oa, llap->llap_inode, valid);
 463
 464         EXIT;
 465 }
 466
 467 /* called for each page in a completed rpc.*/
 468 static int llu_ap_completion(void *data, int cmd, struct obdo *oa, int rc)
 469 {
 470         struct ll_async_page *llap;
 471         cfs_page_t *page;
 472         ENTRY;
 473
 474         llap = llap_from_cookie(data);
 475         llap->llap_queued = 0;
 476         page = llap->llap_page;
 477
 478         if (rc != 0) {
 479                 if (cmd & OBD_BRW_WRITE)
 480                         CERROR("writeback error on page %p index %ld: %d\n",
 481                                page, page->index, rc);
 482         }
 483         RETURN(0);
 484 }
 485
 486 static struct obd_capa * llu_ap_lookup_capa(void *data, int cmd)
 487 {
 488         return NULL;
 489 }
 490
 491 static struct obd_async_page_ops llu_async_page_ops = {
 492         .ap_make_ready =        NULL,
 493         .ap_refresh_count =     NULL,
 494         .ap_fill_obdo =         llu_ap_fill_obdo,
 495         .ap_update_obdo =       llu_ap_update_obdo,
 496         .ap_completion =        llu_ap_completion,
 497         .ap_lookup_capa =       llu_ap_lookup_capa,
 498 };
 499
 500 static int llu_queue_pio(int cmd, struct llu_io_group *group,
 501                          char *buf, size_t count, loff_t pos)
 502 {
 503         struct llu_inode_info *lli = llu_i2info(group->lig_inode);
 504         struct intnl_stat *st = llu_i2stat(group->lig_inode);
 505         struct lov_stripe_md *lsm = lli->lli_smd;
 506         struct obd_export *exp = llu_i2obdexp(group->lig_inode);
 507         cfs_page_t *pages = &group->lig_pages[group->lig_npages],*page = pages;
 508         struct ll_async_page *llap = &group->lig_llaps[group->lig_npages];
 509         void *llap_cookie = group->lig_llap_cookies +
 510                 llap_cookie_size * group->lig_npages;
 511         int i, rc, npages = 0, ret_bytes = 0;
 512         int local_lock;
 513         ENTRY;
 514
 515         if (!exp)
 516                 RETURN(-EINVAL);
 517
 518         local_lock = group->lig_params->lrp_lock_mode != LCK_NL;
 519         /* prepare the pages array */
 520         do {
 521                 unsigned long index, offset, bytes;
 522
 523                 offset = (pos & ~CFS_PAGE_MASK);
 524                 index = pos >> CFS_PAGE_SHIFT;
 525                 bytes = CFS_PAGE_SIZE - offset;
 526                 if (bytes > count)
 527                         bytes = count;
 528
 529                 /* prevent read beyond file range */
 530                 if (/* local_lock && */
 531                     cmd == OBD_BRW_READ && pos + bytes >= st->st_size) {
 532                         if (pos >= st->st_size)
 533                                 break;
 534                         bytes = st->st_size - pos;
 535                 }
 536
 537                 /* prepare page for this index */
 538                 page->index = index;
 539                 page->addr = buf - offset;
 540
 541                 page->_offset = offset;
 542                 page->_count = bytes;
 543
 544                 page++;
 545                 npages++;
 546                 count -= bytes;
 547                 pos += bytes;
 548                 buf += bytes;
 549
 550                 group->lig_rwcount += bytes;
 551                 ret_bytes += bytes;
 552         } while (count);
 553
 554         group->lig_npages += npages;
 555
 556         for (i = 0, page = pages; i < npages;
 557              i++, page++, llap++, llap_cookie += llap_cookie_size){
 558                 llap->llap_magic = LLAP_MAGIC;
 559                 llap->llap_cookie = llap_cookie;
 560                 rc = obd_prep_async_page(exp, lsm, NULL, page,
 561                                          (obd_off)page->index << CFS_PAGE_SHIFT,
 562                                          &llu_async_page_ops,
 563                                          llap, &llap->llap_cookie,
 564                                          1 /* no cache in liblustre at all */,
 565                                          NULL);
 566                 if (rc) {
 567                         LASSERT(rc < 0);
 568                         llap->llap_cookie = NULL;
 569                         RETURN(rc);
 570                 }
 571
 572                 CDEBUG(D_CACHE, "llap %p page %p group %p obj off "LPU64"\n",
 573                        llap, page, llap->llap_cookie,
 574                        (obd_off)pages->index << CFS_PAGE_SHIFT);
 575                 page->private = (unsigned long)llap;
 576                 llap->llap_page = page;
 577                 llap->llap_inode = group->lig_inode;
 578
 579                 rc = obd_queue_group_io(exp, lsm, NULL, group->lig_oig,
 580                                         llap->llap_cookie, cmd,
 581                                         page->_offset, page->_count,
 582                                         group->lig_params->lrp_brw_flags,
 583                                         ASYNC_READY | ASYNC_URGENT |
 584                                         ASYNC_COUNT_STABLE | ASYNC_GROUP_SYNC);
 585                 if (!local_lock && cmd == OBD_BRW_READ) {
 586                         /*
 587                          * In OST-side locking case short reads cannot be
 588                          * detected properly.
 589                          *
 590                          * The root of the problem is that
 591                          *
 592                          * kms = lov_merge_size(lsm, 1);
 593                          * if (end >= kms)
 594                          *         glimpse_size(inode);
 595                          * else
 596                          *         st->st_size = kms;
 597                          *
 598                          * logic in the read code (both llite and liblustre)
 599                          * only works correctly when client holds DLM lock on
 600                          * [start, end]. Without DLM lock KMS can be
 601                          * completely out of date, and client can either make
 602                          * spurious short-read (missing concurrent write), or
 603                          * return stale data (missing concurrent
 604                          * truncate). For llite client this is fatal, because
 605                          * incorrect data are cached and can be later sent
 606                          * back to the server (vide bug 5047). This is hard to
 607                          * fix by handling short-reads on the server, as there
 608                          * is no easy way to communicate file size (or amount
 609                          * of bytes read/written) back to the client,
 610                          * _especially_ because OSC pages can be sliced and
 611                          * dices into multiple RPCs arbitrary. Fortunately,
 612                          * liblustre doesn't cache data and the worst case is
 613                          * that we get race with concurrent write or truncate.
 614                          */
 615                 }
 616                 if (rc) {
 617                         LASSERT(rc < 0);
 618                         RETURN(rc);
 619                 }
 620
 621                 llap->llap_queued = 1;
 622         }
 623
 624         RETURN(ret_bytes);
 625 }
 626
 627 static
 628 struct llu_io_group * get_io_group(struct inode *inode, int maxpages,
 629                                    struct lustre_rw_params *params)
 630 {
 631         struct llu_io_group *group;
 632         int rc;
 633
 634         if (!llap_cookie_size)
 635                 llap_cookie_size = obd_prep_async_page(llu_i2obdexp(inode),
 636                                                        NULL, NULL, NULL, 0,
 637                                                        NULL, NULL, NULL, 0,
 638                                                        NULL);
 639
 640         OBD_ALLOC(group, LLU_IO_GROUP_SIZE(maxpages));
 641         if (!group)
 642                 return ERR_PTR(-ENOMEM);
 643
 644         I_REF(inode);
 645         group->lig_inode = inode;
 646         group->lig_maxpages = maxpages;
 647         group->lig_params = params;
 648         group->lig_llaps = (struct ll_async_page *)(group + 1);
 649         group->lig_pages = (cfs_page_t *)(&group->lig_llaps[maxpages]);
 650         group->lig_llap_cookies = (void *)(&group->lig_pages[maxpages]);
 651
 652         rc = oig_init(&group->lig_oig);
 653         if (rc) {
 654                 OBD_FREE(group, LLU_IO_GROUP_SIZE(maxpages));
 655                 return ERR_PTR(rc);
 656         }
 657
 658         return group;
 659 }
 660
 661 static int max_io_pages(ssize_t len, int iovlen)
 662 {
 663         return (((len + CFS_PAGE_SIZE -1) / CFS_PAGE_SIZE) + 2 + iovlen - 1);
 664 }
 665
 666 static
 667 void put_io_group(struct llu_io_group *group)
 668 {
 669         struct lov_stripe_md *lsm = llu_i2info(group->lig_inode)->lli_smd;
 670         struct obd_export *exp = llu_i2obdexp(group->lig_inode);
 671         struct ll_async_page *llap = group->lig_llaps;
 672         int i;
 673
 674         for (i = 0; i < group->lig_npages; i++, llap++) {
 675                 if (llap->llap_cookie)
 676                         obd_teardown_async_page(exp, lsm, NULL,
 677                                                 llap->llap_cookie);
 678         }
 679
 680         I_RELE(group->lig_inode);
 681
 682         oig_release(group->lig_oig);
 683         OBD_FREE(group, LLU_IO_GROUP_SIZE(group->lig_maxpages));
 684 }
 685
 686 static
 687 ssize_t llu_file_prwv(const struct iovec *iovec, int iovlen,
 688                         _SYSIO_OFF_T pos, ssize_t len,
 689                         void *private)
 690 {
 691         struct llu_io_session *session = (struct llu_io_session *) private;
 692         struct inode *inode = session->lis_inode;
 693         struct llu_inode_info *lli = llu_i2info(inode);
 694         struct intnl_stat *st = llu_i2stat(inode);
 695         struct ll_file_data *fd = lli->lli_file_data;
 696         struct lustre_handle lockh = {0};
 697         struct lov_stripe_md *lsm = lli->lli_smd;
 698         struct obd_export *exp = NULL;
 699         struct llu_io_group *iogroup;
 700         struct lustre_rw_params p;
 701         struct ost_lvb lvb;
 702         __u64 kms;
 703         int err, is_read, iovidx, ret;
 704         int local_lock;
 705         ssize_t ret_len = len;
 706         ENTRY;
 707
 708         /* in a large iov read/write we'll be repeatedly called.
 709          * so give a chance to answer cancel ast here
 710          */
 711         liblustre_wait_event(0);
 712
 713         exp = llu_i2obdexp(inode);
 714         if (exp == NULL)
 715                 RETURN(-EINVAL);
 716
 717         if (len == 0 || iovlen == 0)
 718                 RETURN(0);
 719
 720         if (pos + len > lli->lli_maxbytes)
 721                 RETURN(-ERANGE);
 722
 723         lustre_build_lock_params(session->lis_cmd, lli->lli_open_flags,
 724                                  lli->lli_sbi->ll_lco.lco_flags,
 725                                  pos, len, &p);
 726
 727         iogroup = get_io_group(inode, max_io_pages(len, iovlen), &p);
 728         if (IS_ERR(iogroup))
 729                 RETURN(PTR_ERR(iogroup));
 730
 731         local_lock = p.lrp_lock_mode != LCK_NL;
 732
 733         err = llu_extent_lock(fd, inode, lsm, p.lrp_lock_mode, &p.lrp_policy,
 734                               &lockh, p.lrp_ast_flags);
 735         if (err != ELDLM_OK)
 736                 GOTO(err_put, err);
 737
 738         is_read = (session->lis_cmd == OBD_BRW_READ);
 739         if (is_read) {
 740                 /*
 741                  * If OST-side locking is used, KMS can be completely out of
 742                  * date, and, hence, cannot be used for short-read
 743                  * detection. Rely in OST to handle short reads in that case.
 744                  */
 745                 inode_init_lvb(inode, &lvb);
 746                 obd_merge_lvb(exp, lsm, &lvb, 1);
 747                 kms = lvb.lvb_size;
 748                 /* extent.end is last byte of the range */
 749                 if (p.lrp_policy.l_extent.end >= kms) {
 750                         /* A glimpse is necessary to determine whether
 751                          * we return a short read or some zeroes at
 752                          * the end of the buffer
 753                          *
 754                          * In the case of OST-side locking KMS can be
 755                          * completely out of date and short-reads maybe
 756                          * mishandled. See llu_queue_pio() for more detailed
 757                          * comment.
 758                          */
 759                         if ((err = llu_glimpse_size(inode))) {
 760                                 GOTO(err_unlock, err);
 761                         }
 762                 } else {
 763                         st->st_size = kms;
 764                 }
 765         } else if (lli->lli_open_flags & O_APPEND) {
 766                 pos = st->st_size;
 767         }
 768
 769         for (iovidx = 0; iovidx < iovlen; iovidx++) {
 770                 char *buf = (char *) iovec[iovidx].iov_base;
 771                 size_t count = iovec[iovidx].iov_len;
 772
 773                 if (!count)
 774                         continue;
 775                 if (len < count)
 776                         count = len;
 777                 if (IS_BAD_PTR(buf) || IS_BAD_PTR(buf + count)) {
 778                         GOTO(err_unlock, err = -EFAULT);
 779                 }
 780
 781                 if (is_read) {
 782                         if (/* local_lock && */ pos >= st->st_size)
 783                                 break;
 784                 } else {
 785                         if (pos >= lli->lli_maxbytes) {
 786                                 GOTO(err_unlock, err = -EFBIG);
 787                         }
 788                         if (pos + count >= lli->lli_maxbytes)
 789                                 count = lli->lli_maxbytes - pos;
 790                 }
 791
 792                 ret = llu_queue_pio(session->lis_cmd, iogroup, buf, count, pos);
 793                 if (ret < 0) {
 794                         GOTO(err_unlock, err = ret);
 795                 } else {
 796                         pos += ret;
 797                         if (!is_read) {
 798                                 LASSERT(ret == count);
 799                                 obd_adjust_kms(exp, lsm, pos, 0);
 800                                 /* file size grow immediately */
 801                                 if (pos > st->st_size)
 802                                         st->st_size = pos;
 803                         }
 804                         len -= ret;
 805                         if (!len)
 806                                 break;
 807                 }
 808         }
 809         LASSERT(len == 0 || is_read); /* libsysio should guarantee this */
 810
 811         err = obd_trigger_group_io(exp, lsm, NULL, iogroup->lig_oig);
 812         if (err)
 813                 GOTO(err_unlock, err);
 814
 815         err = oig_wait(iogroup->lig_oig);
 816         if (err) {
 817                 CERROR("%s error: %s\n", is_read ? "read" : "write", strerror(-err));
 818                 GOTO(err_unlock, err);
 819         }
 820
 821         ret = llu_extent_unlock(fd, inode, lsm, p.lrp_lock_mode, &lockh);
 822         if (ret)
 823                 CERROR("extent unlock error %d\n", ret);
 824
 825         session->lis_groups[session->lis_ngroups++] = iogroup;
 826         RETURN(ret_len);
 827
 828 err_unlock:
 829         llu_extent_unlock(fd, inode, lsm, p.lrp_lock_mode, &lockh);
 830 err_put:
 831         put_io_group(iogroup);
 832         RETURN((ssize_t)err);
 833 }
 834
 835 static
 836 struct llu_io_session *get_io_session(struct inode *ino, int ngroups, int cmd)
 837 {
 838         struct llu_io_session *session;
 839
 840         OBD_ALLOC(session, LLU_IO_SESSION_SIZE(ngroups));
 841         if (!session)
 842                 return NULL;
 843
 844         I_REF(ino);
 845         session->lis_inode = ino;
 846         session->lis_max_groups = ngroups;
 847         session->lis_cmd = cmd;
 848         return session;
 849 }
 850
 851 static void put_io_session(struct llu_io_session *session)
 852 {
 853         int i;
 854
 855         for (i = 0; i < session->lis_ngroups; i++) {
 856                 if (session->lis_groups[i]) {
 857                         put_io_group(session->lis_groups[i]);
 858                         session->lis_groups[i] = NULL;
 859                 }
 860         }
 861
 862         I_RELE(session->lis_inode);
 863         OBD_FREE(session, LLU_IO_SESSION_SIZE(session->lis_max_groups));
 864 }
 865
 866 static int llu_file_rwx(struct inode *ino,
 867                         struct ioctx *ioctx,
 868                         int read)
 869 {
 870         struct llu_io_session *session;
 871         ssize_t cc;
 872         int cmd = read ? OBD_BRW_READ : OBD_BRW_WRITE;
 873         ENTRY;
 874
 875         LASSERT(ioctx->ioctx_xtvlen >= 0);
 876         LASSERT(ioctx->ioctx_iovlen >= 0);
 877
 878         liblustre_wait_event(0);
 879
 880         if (!ioctx->ioctx_xtvlen)
 881                 RETURN(0);
 882
 883         /* XXX consider other types later */
 884         if (S_ISDIR(llu_i2stat(ino)->st_mode))
 885                 RETURN(-EISDIR);
 886         if (!S_ISREG(llu_i2stat(ino)->st_mode))
 887                 RETURN(-EOPNOTSUPP);
 888
 889         session = get_io_session(ino, ioctx->ioctx_xtvlen * 2, cmd);
 890         if (!session)
 891                 RETURN(-ENOMEM);
 892
 893         cc = _sysio_enumerate_extents(ioctx->ioctx_xtv, ioctx->ioctx_xtvlen,
 894                                       ioctx->ioctx_iov, ioctx->ioctx_iovlen,
 895                                       llu_file_prwv, session);
 896
 897         if (cc >= 0) {
 898                 LASSERT(!ioctx->ioctx_cc);
 899                 ioctx->ioctx_private = session;
 900                 cc = 0;
 901         } else {
 902                 put_io_session(session);
 903         }
 904
 905         liblustre_wait_event(0);
 906         RETURN(cc);
 907 }
 908
 909 int llu_iop_read(struct inode *ino,
 910                  struct ioctx *ioctx)
 911 {
 912         /* BUG: 5972 */
 913         struct intnl_stat *st = llu_i2stat(ino);
 914         st->st_atime = CURRENT_TIME;
 915
 916         return llu_file_rwx(ino, ioctx, 1);
 917 }
 918
 919 int llu_iop_write(struct inode *ino,
 920                   struct ioctx *ioctx)
 921 {
 922         struct intnl_stat *st = llu_i2stat(ino);
 923         st->st_mtime = st->st_ctime = CURRENT_TIME;
 924
 925         return llu_file_rwx(ino, ioctx, 0);
 926 }
 927
 928 int llu_iop_iodone(struct ioctx *ioctx)
 929 {
 930         struct llu_io_session *session;
 931         struct llu_io_group *group;
 932         int i, err = 0, rc = 0;
 933         ENTRY;
 934
 935         liblustre_wait_event(0);
 936
 937         session = (struct llu_io_session *) ioctx->ioctx_private;
 938         LASSERT(session);
 939         LASSERT(!IS_ERR(session));
 940
 941         for (i = 0; i < session->lis_ngroups; i++) {
 942                 group = session->lis_groups[i];
 943                 if (group) {
 944                         if (!rc) {
 945                                 err = oig_wait(group->lig_oig);
 946                                 if (err)
 947                                         rc = err;
 948                         }
 949                         if (!rc)
 950                                 ioctx->ioctx_cc += group->lig_rwcount;
 951                         put_io_group(group);
 952                         session->lis_groups[i] = NULL;
 953                 }
 954         }
 955
 956         if (rc) {
 957                 LASSERT(rc < 0);
 958                 ioctx->ioctx_cc = -1;
 959                 ioctx->ioctx_errno = -rc;
 960         }
 961
 962         put_io_session(session);
 963         ioctx->ioctx_private = NULL;
 964         liblustre_wait_event(0);
 965
 966         RETURN(1);
 967 }