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