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 [sun.com URL with a
  20  * copy of GPLv2].
  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         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         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         struct page    *llap_page;
 421         struct inode   *llap_inode;
 422 };
 423
 424 static void llu_ap_fill_obdo(void *data, int cmd, struct obdo *oa)
 425 {
 426         struct ll_async_page *llap;
 427         struct inode *inode;
 428         struct lov_stripe_md *lsm;
 429         obd_flag valid_flags;
 430         ENTRY;
 431
 432         llap = LLAP_FROM_COOKIE(data);
 433         inode = llap->llap_inode;
 434         lsm = llu_i2info(inode)->lli_smd;
 435
 436         oa->o_id = lsm->lsm_object_id;
 437         oa->o_valid = OBD_MD_FLID;
 438         valid_flags = OBD_MD_FLTYPE | OBD_MD_FLATIME;
 439         if (cmd & OBD_BRW_WRITE)
 440                 valid_flags |= OBD_MD_FLMTIME | OBD_MD_FLCTIME |
 441                         OBD_MD_FLUID | OBD_MD_FLGID |
 442                         OBD_MD_FLFID | OBD_MD_FLGENER;
 443
 444         obdo_from_inode(oa, inode, valid_flags);
 445         EXIT;
 446 }
 447
 448 static void llu_ap_update_obdo(void *data, int cmd, struct obdo *oa,
 449                                obd_valid valid)
 450 {
 451         struct ll_async_page *llap;
 452         ENTRY;
 453
 454         llap = LLAP_FROM_COOKIE(data);
 455         obdo_from_inode(oa, llap->llap_inode, valid);
 456
 457         EXIT;
 458 }
 459
 460 /* called for each page in a completed rpc.*/
 461 static int llu_ap_completion(void *data, int cmd, struct obdo *oa, int rc)
 462 {
 463         struct ll_async_page *llap;
 464         struct page *page;
 465         ENTRY;
 466
 467         llap = LLAP_FROM_COOKIE(data);
 468         llap->llap_queued = 0;
 469         page = llap->llap_page;
 470
 471         if (rc != 0) {
 472                 if (cmd & OBD_BRW_WRITE)
 473                         CERROR("writeback error on page %p index %ld: %d\n",
 474                                page, page->index, rc);
 475         }
 476         RETURN(0);
 477 }
 478
 479 static struct obd_capa * llu_ap_lookup_capa(void *data, int cmd)
 480 {
 481         return NULL;
 482 }
 483
 484 static struct obd_async_page_ops llu_async_page_ops = {
 485         .ap_make_ready =        NULL,
 486         .ap_refresh_count =     NULL,
 487         .ap_fill_obdo =         llu_ap_fill_obdo,
 488         .ap_update_obdo =       llu_ap_update_obdo,
 489         .ap_completion =        llu_ap_completion,
 490         .ap_lookup_capa =       llu_ap_lookup_capa,
 491 };
 492
 493 static int llu_queue_pio(int cmd, struct llu_io_group *group,
 494                          char *buf, size_t count, loff_t pos)
 495 {
 496         struct llu_inode_info *lli = llu_i2info(group->lig_inode);
 497         struct intnl_stat *st = llu_i2stat(group->lig_inode);
 498         struct lov_stripe_md *lsm = lli->lli_smd;
 499         struct obd_export *exp = llu_i2obdexp(group->lig_inode);
 500         struct page *pages = &group->lig_pages[group->lig_npages],*page = pages;
 501         struct ll_async_page *llap = &group->lig_llaps[group->lig_npages];
 502         void *llap_cookie = group->lig_llap_cookies +
 503                 llap_cookie_size * group->lig_npages;
 504         int i, rc, npages = 0, ret_bytes = 0;
 505         int local_lock;
 506         ENTRY;
 507
 508         if (!exp)
 509                 RETURN(-EINVAL);
 510
 511         local_lock = group->lig_params->lrp_lock_mode != LCK_NL;
 512         /* prepare the pages array */
 513         do {
 514                 unsigned long index, offset, bytes;
 515
 516                 offset = (pos & ~CFS_PAGE_MASK);
 517                 index = pos >> CFS_PAGE_SHIFT;
 518                 bytes = CFS_PAGE_SIZE - offset;
 519                 if (bytes > count)
 520                         bytes = count;
 521
 522                 /* prevent read beyond file range */
 523                 if (/* local_lock && */
 524                     cmd == OBD_BRW_READ && pos + bytes >= st->st_size) {
 525                         if (pos >= st->st_size)
 526                                 break;
 527                         bytes = st->st_size - pos;
 528                 }
 529
 530                 /* prepare page for this index */
 531                 page->index = index;
 532                 page->addr = buf - offset;
 533
 534                 page->_offset = offset;
 535                 page->_count = bytes;
 536
 537                 page++;
 538                 npages++;
 539                 count -= bytes;
 540                 pos += bytes;
 541                 buf += bytes;
 542
 543                 group->lig_rwcount += bytes;
 544                 ret_bytes += bytes;
 545         } while (count);
 546
 547         group->lig_npages += npages;
 548
 549         for (i = 0, page = pages; i < npages;
 550              i++, page++, llap++, llap_cookie += llap_cookie_size){
 551                 llap->llap_magic = LLAP_MAGIC;
 552                 llap->llap_cookie = llap_cookie;
 553                 rc = obd_prep_async_page(exp, lsm, NULL, page,
 554                                          (obd_off)page->index << CFS_PAGE_SHIFT,
 555                                          &llu_async_page_ops,
 556                                          llap, &llap->llap_cookie,
 557                                          1 /* no cache in liblustre at all */,
 558                                          NULL);
 559                 if (rc) {
 560                         LASSERT(rc < 0);
 561                         llap->llap_cookie = NULL;
 562                         RETURN(rc);
 563                 }
 564
 565                 CDEBUG(D_CACHE, "llap %p page %p group %p obj off "LPU64"\n",
 566                        llap, page, llap->llap_cookie,
 567                        (obd_off)pages->index << CFS_PAGE_SHIFT);
 568                 page->private = (unsigned long)llap;
 569                 llap->llap_page = page;
 570                 llap->llap_inode = group->lig_inode;
 571
 572                 rc = obd_queue_group_io(exp, lsm, NULL, group->lig_oig,
 573                                         llap->llap_cookie, cmd,
 574                                         page->_offset, page->_count,
 575                                         group->lig_params->lrp_brw_flags,
 576                                         ASYNC_READY | ASYNC_URGENT |
 577                                         ASYNC_COUNT_STABLE | ASYNC_GROUP_SYNC);
 578                 if (!local_lock && cmd == OBD_BRW_READ) {
 579                         /*
 580                          * In OST-side locking case short reads cannot be
 581                          * detected properly.
 582                          *
 583                          * The root of the problem is that
 584                          *
 585                          * kms = lov_merge_size(lsm, 1);
 586                          * if (end >= kms)
 587                          *         glimpse_size(inode);
 588                          * else
 589                          *         st->st_size = kms;
 590                          *
 591                          * logic in the read code (both llite and liblustre)
 592                          * only works correctly when client holds DLM lock on
 593                          * [start, end]. Without DLM lock KMS can be
 594                          * completely out of date, and client can either make
 595                          * spurious short-read (missing concurrent write), or
 596                          * return stale data (missing concurrent
 597                          * truncate). For llite client this is fatal, because
 598                          * incorrect data are cached and can be later sent
 599                          * back to the server (vide bug 5047). This is hard to
 600                          * fix by handling short-reads on the server, as there
 601                          * is no easy way to communicate file size (or amount
 602                          * of bytes read/written) back to the client,
 603                          * _especially_ because OSC pages can be sliced and
 604                          * dices into multiple RPCs arbitrary. Fortunately,
 605                          * liblustre doesn't cache data and the worst case is
 606                          * that we get race with concurrent write or truncate.
 607                          */
 608                 }
 609                 if (rc) {
 610                         LASSERT(rc < 0);
 611                         RETURN(rc);
 612                 }
 613
 614                 llap->llap_queued = 1;
 615         }
 616
 617         RETURN(ret_bytes);
 618 }
 619
 620 static
 621 struct llu_io_group * get_io_group(struct inode *inode, int maxpages,
 622                                    struct lustre_rw_params *params)
 623 {
 624         struct llu_io_group *group;
 625         int rc;
 626
 627         if (!llap_cookie_size)
 628                 llap_cookie_size = obd_prep_async_page(llu_i2obdexp(inode),
 629                                                        NULL, NULL, NULL, 0,
 630                                                        NULL, NULL, NULL, 0,
 631                                                        NULL);
 632
 633         OBD_ALLOC(group, LLU_IO_GROUP_SIZE(maxpages));
 634         if (!group)
 635                 return ERR_PTR(-ENOMEM);
 636
 637         I_REF(inode);
 638         group->lig_inode = inode;
 639         group->lig_maxpages = maxpages;
 640         group->lig_params = params;
 641         group->lig_llaps = (struct ll_async_page *)(group + 1);
 642         group->lig_pages = (struct page *)(&group->lig_llaps[maxpages]);
 643         group->lig_llap_cookies = (void *)(&group->lig_pages[maxpages]);
 644
 645         rc = oig_init(&group->lig_oig);
 646         if (rc) {
 647                 OBD_FREE(group, LLU_IO_GROUP_SIZE(maxpages));
 648                 return ERR_PTR(rc);
 649         }
 650
 651         return group;
 652 }
 653
 654 static int max_io_pages(ssize_t len, int iovlen)
 655 {
 656         return (((len + CFS_PAGE_SIZE -1) / CFS_PAGE_SIZE) + 2 + iovlen - 1);
 657 }
 658
 659 static
 660 void put_io_group(struct llu_io_group *group)
 661 {
 662         struct lov_stripe_md *lsm = llu_i2info(group->lig_inode)->lli_smd;
 663         struct obd_export *exp = llu_i2obdexp(group->lig_inode);
 664         struct ll_async_page *llap = group->lig_llaps;
 665         int i;
 666
 667         for (i = 0; i < group->lig_npages; i++, llap++) {
 668                 if (llap->llap_cookie)
 669                         obd_teardown_async_page(exp, lsm, NULL,
 670                                                 llap->llap_cookie);
 671         }
 672
 673         I_RELE(group->lig_inode);
 674
 675         oig_release(group->lig_oig);
 676         OBD_FREE(group, LLU_IO_GROUP_SIZE(group->lig_maxpages));
 677 }
 678
 679 static
 680 ssize_t llu_file_prwv(const struct iovec *iovec, int iovlen,
 681                         _SYSIO_OFF_T pos, ssize_t len,
 682                         void *private)
 683 {
 684         struct llu_io_session *session = (struct llu_io_session *) private;
 685         struct inode *inode = session->lis_inode;
 686         struct llu_inode_info *lli = llu_i2info(inode);
 687         struct intnl_stat *st = llu_i2stat(inode);
 688         struct ll_file_data *fd = lli->lli_file_data;
 689         struct lustre_handle lockh = {0};
 690         struct lov_stripe_md *lsm = lli->lli_smd;
 691         struct obd_export *exp = NULL;
 692         struct llu_io_group *iogroup;
 693         struct lustre_rw_params p;
 694         struct ost_lvb lvb;
 695         __u64 kms;
 696         int err, is_read, iovidx, ret;
 697         int local_lock;
 698         ssize_t ret_len = len;
 699         ENTRY;
 700
 701         /* in a large iov read/write we'll be repeatedly called.
 702          * so give a chance to answer cancel ast here
 703          */
 704         liblustre_wait_event(0);
 705
 706         exp = llu_i2obdexp(inode);
 707         if (exp == NULL)
 708                 RETURN(-EINVAL);
 709
 710         if (len == 0 || iovlen == 0)
 711                 RETURN(0);
 712
 713         if (pos + len > lli->lli_maxbytes)
 714                 RETURN(-ERANGE);
 715
 716         lustre_build_lock_params(session->lis_cmd, lli->lli_open_flags,
 717                                  lli->lli_sbi->ll_lco.lco_flags,
 718                                  pos, len, &p);
 719
 720         iogroup = get_io_group(inode, max_io_pages(len, iovlen), &p);
 721         if (IS_ERR(iogroup))
 722                 RETURN(PTR_ERR(iogroup));
 723
 724         local_lock = p.lrp_lock_mode != LCK_NL;
 725
 726         err = llu_extent_lock(fd, inode, lsm, p.lrp_lock_mode, &p.lrp_policy,
 727                               &lockh, p.lrp_ast_flags);
 728         if (err != ELDLM_OK)
 729                 GOTO(err_put, err);
 730
 731         is_read = (session->lis_cmd == OBD_BRW_READ);
 732         if (is_read) {
 733                 /*
 734                  * If OST-side locking is used, KMS can be completely out of
 735                  * date, and, hence, cannot be used for short-read
 736                  * detection. Rely in OST to handle short reads in that case.
 737                  */
 738                 inode_init_lvb(inode, &lvb);
 739                 obd_merge_lvb(exp, lsm, &lvb, 1);
 740                 kms = lvb.lvb_size;
 741                 /* extent.end is last byte of the range */
 742                 if (p.lrp_policy.l_extent.end >= kms) {
 743                         /* A glimpse is necessary to determine whether
 744                          * we return a short read or some zeroes at
 745                          * the end of the buffer
 746                          *
 747                          * In the case of OST-side locking KMS can be
 748                          * completely out of date and short-reads maybe
 749                          * mishandled. See llu_queue_pio() for more detailed
 750                          * comment.
 751                          */
 752                         if ((err = llu_glimpse_size(inode))) {
 753                                 GOTO(err_unlock, err);
 754                         }
 755                 } else {
 756                         st->st_size = kms;
 757                 }
 758         } else if (lli->lli_open_flags & O_APPEND) {
 759                 pos = st->st_size;
 760         }
 761
 762         for (iovidx = 0; iovidx < iovlen; iovidx++) {
 763                 char *buf = (char *) iovec[iovidx].iov_base;
 764                 size_t count = iovec[iovidx].iov_len;
 765
 766                 if (!count)
 767                         continue;
 768                 if (len < count)
 769                         count = len;
 770                 if (IS_BAD_PTR(buf) || IS_BAD_PTR(buf + count)) {
 771                         GOTO(err_unlock, err = -EFAULT);
 772                 }
 773
 774                 if (is_read) {
 775                         if (/* local_lock && */ pos >= st->st_size)
 776                                 break;
 777                 } else {
 778                         if (pos >= lli->lli_maxbytes) {
 779                                 GOTO(err_unlock, err = -EFBIG);
 780                         }
 781                         if (pos + count >= lli->lli_maxbytes)
 782                                 count = lli->lli_maxbytes - pos;
 783                 }
 784
 785                 ret = llu_queue_pio(session->lis_cmd, iogroup, buf, count, pos);
 786                 if (ret < 0) {
 787                         GOTO(err_unlock, err = ret);
 788                 } else {
 789                         pos += ret;
 790                         if (!is_read) {
 791                                 LASSERT(ret == count);
 792                                 obd_adjust_kms(exp, lsm, pos, 0);
 793                                 /* file size grow immediately */
 794                                 if (pos > st->st_size)
 795                                         st->st_size = pos;
 796                         }
 797                         len -= ret;
 798                         if (!len)
 799                                 break;
 800                 }
 801         }
 802         LASSERT(len == 0 || is_read); /* libsysio should guarantee this */
 803
 804         err = obd_trigger_group_io(exp, lsm, NULL, iogroup->lig_oig);
 805         if (err)
 806                 GOTO(err_unlock, err);
 807
 808         err = oig_wait(iogroup->lig_oig);
 809         if (err) {
 810                 CERROR("%s error: %s\n", is_read ? "read" : "write", strerror(-err));
 811                 GOTO(err_unlock, err);
 812         }
 813
 814         ret = llu_extent_unlock(fd, inode, lsm, p.lrp_lock_mode, &lockh);
 815         if (ret)
 816                 CERROR("extent unlock error %d\n", ret);
 817
 818         session->lis_groups[session->lis_ngroups++] = iogroup;
 819         RETURN(ret_len);
 820
 821 err_unlock:
 822         llu_extent_unlock(fd, inode, lsm, p.lrp_lock_mode, &lockh);
 823 err_put:
 824         put_io_group(iogroup);
 825         RETURN((ssize_t)err);
 826 }
 827
 828 static
 829 struct llu_io_session *get_io_session(struct inode *ino, int ngroups, int cmd)
 830 {
 831         struct llu_io_session *session;
 832
 833         OBD_ALLOC(session, LLU_IO_SESSION_SIZE(ngroups));
 834         if (!session)
 835                 return NULL;
 836
 837         I_REF(ino);
 838         session->lis_inode = ino;
 839         session->lis_max_groups = ngroups;
 840         session->lis_cmd = cmd;
 841         return session;
 842 }
 843
 844 static void put_io_session(struct llu_io_session *session)
 845 {
 846         int i;
 847
 848         for (i = 0; i < session->lis_ngroups; i++) {
 849                 if (session->lis_groups[i]) {
 850                         put_io_group(session->lis_groups[i]);
 851                         session->lis_groups[i] = NULL;
 852                 }
 853         }
 854
 855         I_RELE(session->lis_inode);
 856         OBD_FREE(session, LLU_IO_SESSION_SIZE(session->lis_max_groups));
 857 }
 858
 859 static int llu_file_rwx(struct inode *ino,
 860                         struct ioctx *ioctx,
 861                         int read)
 862 {
 863         struct llu_io_session *session;
 864         ssize_t cc;
 865         int cmd = read ? OBD_BRW_READ : OBD_BRW_WRITE;
 866         ENTRY;
 867
 868         LASSERT(ioctx->ioctx_xtvlen >= 0);
 869         LASSERT(ioctx->ioctx_iovlen >= 0);
 870
 871         liblustre_wait_event(0);
 872
 873         if (!ioctx->ioctx_xtvlen)
 874                 RETURN(0);
 875
 876         /* XXX consider other types later */
 877         if (S_ISDIR(llu_i2stat(ino)->st_mode))
 878                 RETURN(-EISDIR);
 879         if (!S_ISREG(llu_i2stat(ino)->st_mode))
 880                 RETURN(-EOPNOTSUPP);
 881
 882         session = get_io_session(ino, ioctx->ioctx_xtvlen * 2, cmd);
 883         if (!session)
 884                 RETURN(-ENOMEM);
 885
 886         cc = _sysio_enumerate_extents(ioctx->ioctx_xtv, ioctx->ioctx_xtvlen,
 887                                       ioctx->ioctx_iov, ioctx->ioctx_iovlen,
 888                                       llu_file_prwv, session);
 889
 890         if (cc >= 0) {
 891                 LASSERT(!ioctx->ioctx_cc);
 892                 ioctx->ioctx_private = session;
 893                 cc = 0;
 894         } else {
 895                 put_io_session(session);
 896         }
 897
 898         liblustre_wait_event(0);
 899         RETURN(cc);
 900 }
 901
 902 int llu_iop_read(struct inode *ino,
 903                  struct ioctx *ioctx)
 904 {
 905         /* BUG: 5972 */
 906         struct intnl_stat *st = llu_i2stat(ino);
 907         st->st_atime = CURRENT_TIME;
 908
 909         return llu_file_rwx(ino, ioctx, 1);
 910 }
 911
 912 int llu_iop_write(struct inode *ino,
 913                   struct ioctx *ioctx)
 914 {
 915         struct intnl_stat *st = llu_i2stat(ino);
 916         st->st_mtime = st->st_ctime = CURRENT_TIME;
 917
 918         return llu_file_rwx(ino, ioctx, 0);
 919 }
 920
 921 int llu_iop_iodone(struct ioctx *ioctx)
 922 {
 923         struct llu_io_session *session;
 924         struct llu_io_group *group;
 925         int i, err = 0, rc = 0;
 926         ENTRY;
 927
 928         liblustre_wait_event(0);
 929
 930         session = (struct llu_io_session *) ioctx->ioctx_private;
 931         LASSERT(session);
 932         LASSERT(!IS_ERR(session));
 933
 934         for (i = 0; i < session->lis_ngroups; i++) {
 935                 group = session->lis_groups[i];
 936                 if (group) {
 937                         if (!rc) {
 938                                 err = oig_wait(group->lig_oig);
 939                                 if (err)
 940                                         rc = err;
 941                         }
 942                         if (!rc)
 943                                 ioctx->ioctx_cc += group->lig_rwcount;
 944                         put_io_group(group);
 945                         session->lis_groups[i] = NULL;
 946                 }
 947         }
 948
 949         if (rc) {
 950                 LASSERT(rc < 0);
 951                 ioctx->ioctx_cc = -1;
 952                 ioctx->ioctx_errno = -rc;
 953         }
 954
 955         put_io_session(session);
 956         ioctx->ioctx_private = NULL;
 957         liblustre_wait_event(0);
 958
 959         RETURN(1);
 960 }