lustre/lclient/lcommon_cl.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  * cl code shared between vvp and liblustre (and other Lustre clients in the
  37  * future).
  38  *
  39  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  40  */
  41
  42 #define DEBUG_SUBSYSTEM S_LLITE
  43
  44 #ifdef __KERNEL__
  45 # include <libcfs/libcfs.h>
  46 # include <linux/fs.h>
  47 # include <linux/sched.h>
  48 # include <linux/mm.h>
  49 # include <linux/smp_lock.h>
  50 # include <linux/quotaops.h>
  51 # include <linux/highmem.h>
  52 # include <linux/pagemap.h>
  53 # include <linux/rbtree.h>
  54 #else /* __KERNEL__ */
  55 #include <stdlib.h>
  56 #include <string.h>
  57 #include <assert.h>
  58 #include <time.h>
  59 #include <sys/types.h>
  60 #include <sys/stat.h>
  61 #include <sys/queue.h>
  62 #include <fcntl.h>
  63 # include <sysio.h>
  64 # ifdef HAVE_XTIO_H
  65 #  include <xtio.h>
  66 # endif
  67 # include <fs.h>
  68 # include <mount.h>
  69 # include <inode.h>
  70 # ifdef HAVE_FILE_H
  71 #  include <file.h>
  72 # endif
  73 # include <liblustre.h>
  74 #endif
  75
  76 #include <obd.h>
  77 #include <obd_support.h>
  78 #include <lustre_fid.h>
  79 #include <lustre_lite.h>
  80 #include <lustre_dlm.h>
  81 #include <lustre_ver.h>
  82 #include <lustre_mdc.h>
  83 #include <cl_object.h>
  84
  85 #include <lclient.h>
  86
  87 #ifdef __KERNEL__
  88 #include "../llite/llite_internal.h"
  89 #else
  90 #include "../liblustre/llite_lib.h"
  91 #endif
  92
  93 const struct cl_req_operations ccc_req_ops;
  94
  95 /*
  96  * ccc_ prefix stands for "Common Client Code".
  97  */
  98
  99 static cfs_mem_cache_t *ccc_lock_kmem;
 100 static cfs_mem_cache_t *ccc_object_kmem;
 101 static cfs_mem_cache_t *ccc_thread_kmem;
 102 static cfs_mem_cache_t *ccc_session_kmem;
 103 static cfs_mem_cache_t *ccc_req_kmem;
 104
 105 static struct lu_kmem_descr ccc_caches[] = {
 106         {
 107                 .ckd_cache = &ccc_lock_kmem,
 108                 .ckd_name  = "ccc_lock_kmem",
 109                 .ckd_size  = sizeof (struct ccc_lock)
 110         },
 111         {
 112                 .ckd_cache = &ccc_object_kmem,
 113                 .ckd_name  = "ccc_object_kmem",
 114                 .ckd_size  = sizeof (struct ccc_object)
 115         },
 116         {
 117                 .ckd_cache = &ccc_thread_kmem,
 118                 .ckd_name  = "ccc_thread_kmem",
 119                 .ckd_size  = sizeof (struct ccc_thread_info),
 120         },
 121         {
 122                 .ckd_cache = &ccc_session_kmem,
 123                 .ckd_name  = "ccc_session_kmem",
 124                 .ckd_size  = sizeof (struct ccc_session)
 125         },
 126         {
 127                 .ckd_cache = &ccc_req_kmem,
 128                 .ckd_name  = "ccc_req_kmem",
 129                 .ckd_size  = sizeof (struct ccc_req)
 130         },
 131         {
 132                 .ckd_cache = NULL
 133         }
 134 };
 135
 136 /*****************************************************************************
 137  *
 138  * Vvp device and device type functions.
 139  *
 140  */
 141
 142 void *ccc_key_init(const struct lu_context *ctx,
 143                           struct lu_context_key *key)
 144 {
 145         struct ccc_thread_info *info;
 146
 147         OBD_SLAB_ALLOC_PTR_GFP(info, ccc_thread_kmem, CFS_ALLOC_IO);
 148         if (info == NULL)
 149                 info = ERR_PTR(-ENOMEM);
 150         return info;
 151 }
 152
 153 void ccc_key_fini(const struct lu_context *ctx,
 154                          struct lu_context_key *key, void *data)
 155 {
 156         struct ccc_thread_info *info = data;
 157         OBD_SLAB_FREE_PTR(info, ccc_thread_kmem);
 158 }
 159
 160 void *ccc_session_key_init(const struct lu_context *ctx,
 161                                   struct lu_context_key *key)
 162 {
 163         struct ccc_session *session;
 164
 165         OBD_SLAB_ALLOC_PTR_GFP(session, ccc_session_kmem, CFS_ALLOC_IO);
 166         if (session == NULL)
 167                 session = ERR_PTR(-ENOMEM);
 168         return session;
 169 }
 170
 171 void ccc_session_key_fini(const struct lu_context *ctx,
 172                                  struct lu_context_key *key, void *data)
 173 {
 174         struct ccc_session *session = data;
 175         OBD_SLAB_FREE_PTR(session, ccc_session_kmem);
 176 }
 177
 178 struct lu_context_key ccc_key = {
 179         .lct_tags = LCT_CL_THREAD,
 180         .lct_init = ccc_key_init,
 181         .lct_fini = ccc_key_fini
 182 };
 183
 184 struct lu_context_key ccc_session_key = {
 185         .lct_tags = LCT_SESSION,
 186         .lct_init = ccc_session_key_init,
 187         .lct_fini = ccc_session_key_fini
 188 };
 189
 190
 191 /* type constructor/destructor: ccc_type_{init,fini,start,stop}(). */
 192 // LU_TYPE_INIT_FINI(ccc, &ccc_key, &ccc_session_key);
 193
 194 int ccc_device_init(const struct lu_env *env, struct lu_device *d,
 195                            const char *name, struct lu_device *next)
 196 {
 197         struct ccc_device  *vdv;
 198         int rc;
 199         ENTRY;
 200
 201         vdv = lu2ccc_dev(d);
 202         vdv->cdv_next = lu2cl_dev(next);
 203
 204         LASSERT(d->ld_site != NULL && next->ld_type != NULL);
 205         next->ld_site = d->ld_site;
 206         rc = next->ld_type->ldt_ops->ldto_device_init(
 207                         env, next, next->ld_type->ldt_name, NULL);
 208         if (rc == 0) {
 209                 lu_device_get(next);
 210                 lu_ref_add(&next->ld_reference, "lu-stack", &lu_site_init);
 211         }
 212         RETURN(rc);
 213 }
 214
 215 struct lu_device *ccc_device_fini(const struct lu_env *env,
 216                                          struct lu_device *d)
 217 {
 218         return cl2lu_dev(lu2ccc_dev(d)->cdv_next);
 219 }
 220
 221 struct lu_device *ccc_device_alloc(const struct lu_env *env,
 222                                    struct lu_device_type *t,
 223                                    struct lustre_cfg *cfg,
 224                                    const struct lu_device_operations *luops,
 225                                    const struct cl_device_operations *clops)
 226 {
 227         struct ccc_device *vdv;
 228         struct lu_device  *lud;
 229         struct cl_site    *site;
 230         int rc;
 231         ENTRY;
 232
 233         OBD_ALLOC_PTR(vdv);
 234         if (vdv == NULL)
 235                 RETURN(ERR_PTR(-ENOMEM));
 236
 237         lud = &vdv->cdv_cl.cd_lu_dev;
 238         cl_device_init(&vdv->cdv_cl, t);
 239         ccc2lu_dev(vdv)->ld_ops = luops;
 240         vdv->cdv_cl.cd_ops = clops;
 241
 242         OBD_ALLOC_PTR(site);
 243         if (site != NULL) {
 244                 rc = cl_site_init(site, &vdv->cdv_cl);
 245                 if (rc == 0)
 246                         rc = lu_site_init_finish(&site->cs_lu);
 247                 else {
 248                         LASSERT(lud->ld_site == NULL);
 249                         CERROR("Cannot init lu_site, rc %d.\n", rc);
 250                         OBD_FREE_PTR(site);
 251                 }
 252         } else
 253                 rc = -ENOMEM;
 254         if (rc != 0) {
 255                 ccc_device_free(env, lud);
 256                 lud = ERR_PTR(rc);
 257         }
 258         RETURN(lud);
 259 }
 260
 261 struct lu_device *ccc_device_free(const struct lu_env *env,
 262                                          struct lu_device *d)
 263 {
 264         struct ccc_device *vdv  = lu2ccc_dev(d);
 265         struct cl_site    *site = lu2cl_site(d->ld_site);
 266         struct lu_device  *next = cl2lu_dev(vdv->cdv_next);
 267
 268         if (d->ld_site != NULL) {
 269                 cl_site_fini(site);
 270                 OBD_FREE_PTR(site);
 271         }
 272         cl_device_fini(lu2cl_dev(d));
 273         OBD_FREE_PTR(vdv);
 274         return next;
 275 }
 276
 277 int ccc_req_init(const struct lu_env *env, struct cl_device *dev,
 278                         struct cl_req *req)
 279 {
 280         struct ccc_req *vrq;
 281         int result;
 282
 283         OBD_SLAB_ALLOC_PTR_GFP(vrq, ccc_req_kmem, CFS_ALLOC_IO);
 284         if (vrq != NULL) {
 285                 cl_req_slice_add(req, &vrq->crq_cl, dev, &ccc_req_ops);
 286                 result = 0;
 287         } else
 288                 result = -ENOMEM;
 289         return result;
 290 }
 291
 292 /**
 293  * An `emergency' environment used by ccc_inode_fini() when cl_env_get()
 294  * fails. Access to this environment is serialized by ccc_inode_fini_guard
 295  * mutex.
 296  */
 297 static struct lu_env *ccc_inode_fini_env = NULL;
 298
 299 /**
 300  * A mutex serializing calls to slp_inode_fini() under extreme memory
 301  * pressure, when environments cannot be allocated.
 302  */
 303 static DEFINE_MUTEX(ccc_inode_fini_guard);
 304 static int dummy_refcheck;
 305
 306 int ccc_global_init(struct lu_device_type *device_type)
 307 {
 308         int result;
 309
 310         result = lu_kmem_init(ccc_caches);
 311         if (result == 0) {
 312                 result = lu_device_type_init(device_type);
 313                 ccc_inode_fini_env = cl_env_alloc(&dummy_refcheck,
 314                                                   LCT_REMEMBER|LCT_NOREF);
 315                 if (IS_ERR(ccc_inode_fini_env))
 316                         result = PTR_ERR(ccc_inode_fini_env);
 317                 else
 318                         ccc_inode_fini_env->le_ctx.lc_cookie = 0x4;
 319         }
 320         return result;
 321 }
 322
 323 void ccc_global_fini(struct lu_device_type *device_type)
 324 {
 325         if (ccc_inode_fini_env != NULL) {
 326                 cl_env_put(ccc_inode_fini_env, &dummy_refcheck);
 327                 ccc_inode_fini_env = NULL;
 328         }
 329         lu_device_type_fini(device_type);
 330         lu_kmem_fini(ccc_caches);
 331 }
 332
 333 /*****************************************************************************
 334  *
 335  * Object operations.
 336  *
 337  */
 338
 339 struct lu_object *ccc_object_alloc(const struct lu_env *env,
 340                                    const struct lu_object_header *_,
 341                                    struct lu_device *dev,
 342                                    const struct cl_object_operations *clops,
 343                                    const struct lu_object_operations *luops)
 344 {
 345         struct ccc_object *vob;
 346         struct lu_object  *obj;
 347
 348         OBD_SLAB_ALLOC_PTR_GFP(vob, ccc_object_kmem, CFS_ALLOC_IO);
 349         if (vob != NULL) {
 350                 struct cl_object_header *hdr;
 351
 352                 obj = ccc2lu(vob);
 353                 hdr = &vob->cob_header;
 354                 cl_object_header_init(hdr);
 355                 lu_object_init(obj, &hdr->coh_lu, dev);
 356                 lu_object_add_top(&hdr->coh_lu, obj);
 357
 358                 vob->cob_cl.co_ops = clops;
 359                 obj->lo_ops = luops;
 360         } else
 361                 obj = NULL;
 362         return obj;
 363 }
 364
 365 int ccc_object_init0(const struct lu_env *env,
 366                             struct ccc_object *vob,
 367                             const struct cl_object_conf *conf)
 368 {
 369         vob->cob_inode = conf->coc_inode;
 370         vob->cob_transient_pages = 0;
 371         return 0;
 372 }
 373
 374 int ccc_object_init(const struct lu_env *env, struct lu_object *obj,
 375                            const struct lu_object_conf *conf)
 376 {
 377         struct ccc_device *dev = lu2ccc_dev(obj->lo_dev);
 378         struct ccc_object *vob = lu2ccc(obj);
 379         struct lu_object  *below;
 380         struct lu_device  *under;
 381         int result;
 382
 383         under = &dev->cdv_next->cd_lu_dev;
 384         below = under->ld_ops->ldo_object_alloc(env, obj->lo_header, under);
 385         if (below != NULL) {
 386                 const struct cl_object_conf *cconf;
 387
 388                 cconf = lu2cl_conf(conf);
 389                 CFS_INIT_LIST_HEAD(&vob->cob_pending_list);
 390                 lu_object_add(obj, below);
 391                 result = ccc_object_init0(env, vob, cconf);
 392         } else
 393                 result = -ENOMEM;
 394         return result;
 395 }
 396
 397 void ccc_object_free(const struct lu_env *env, struct lu_object *obj)
 398 {
 399         struct ccc_object *vob = lu2ccc(obj);
 400
 401         lu_object_fini(obj);
 402         lu_object_header_fini(obj->lo_header);
 403         OBD_SLAB_FREE_PTR(vob, ccc_object_kmem);
 404 }
 405
 406 int ccc_lock_init(const struct lu_env *env,
 407                   struct cl_object *obj, struct cl_lock *lock,
 408                   const struct cl_io *_,
 409                   const struct cl_lock_operations *lkops)
 410 {
 411         struct ccc_lock *clk;
 412         int result;
 413
 414         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
 415
 416         OBD_SLAB_ALLOC_PTR_GFP(clk, ccc_lock_kmem, CFS_ALLOC_IO);
 417         if (clk != NULL) {
 418                 cl_lock_slice_add(lock, &clk->clk_cl, obj, lkops);
 419                 result = 0;
 420         } else
 421                 result = -ENOMEM;
 422         return result;
 423 }
 424
 425 int ccc_attr_set(const struct lu_env *env, struct cl_object *obj,
 426                  const struct cl_attr *attr, unsigned valid)
 427 {
 428         return 0;
 429 }
 430
 431 int ccc_object_glimpse(const struct lu_env *env,
 432                        const struct cl_object *obj, struct ost_lvb *lvb)
 433 {
 434         struct inode *inode = ccc_object_inode(obj);
 435
 436         ENTRY;
 437         lvb->lvb_mtime = cl_inode_mtime(inode);
 438         lvb->lvb_atime = cl_inode_atime(inode);
 439         lvb->lvb_ctime = cl_inode_ctime(inode);
 440         RETURN(0);
 441 }
 442
 443
 444
 445 int ccc_conf_set(const struct lu_env *env, struct cl_object *obj,
 446                         const struct cl_object_conf *conf)
 447 {
 448         /* TODO: destroy all pages attached to this object. */
 449         return 0;
 450 }
 451
 452 /*****************************************************************************
 453  *
 454  * Page operations.
 455  *
 456  */
 457
 458 cfs_page_t *ccc_page_vmpage(const struct lu_env *env,
 459                             const struct cl_page_slice *slice)
 460 {
 461         return cl2vm_page(slice);
 462 }
 463
 464 int ccc_page_is_under_lock(const struct lu_env *env,
 465                            const struct cl_page_slice *slice,
 466                            struct cl_io *io)
 467 {
 468         struct ccc_io        *vio  = ccc_env_io(env);
 469         struct cl_lock_descr *desc = &ccc_env_info(env)->cti_descr;
 470         struct cl_page       *page = slice->cpl_page;
 471
 472         int result;
 473
 474         ENTRY;
 475
 476         if (io->ci_type == CIT_READ || io->ci_type == CIT_WRITE ||
 477             io->ci_type == CIT_FAULT) {
 478                 if (vio->cui_fd->fd_flags & LL_FILE_GROUP_LOCKED)
 479                         result = -EBUSY;
 480                 else {
 481                         desc->cld_start = page->cp_index;
 482                         desc->cld_end   = page->cp_index;
 483                         desc->cld_obj   = page->cp_obj;
 484                         desc->cld_mode  = CLM_READ;
 485                         result = cl_queue_match(&io->ci_lockset.cls_done,
 486                                                 desc) ? -EBUSY : 0;
 487                 }
 488         } else
 489                 result = 0;
 490         RETURN(result);
 491 }
 492
 493 int ccc_fail(const struct lu_env *env, const struct cl_page_slice *slice)
 494 {
 495         /*
 496          * Cached read?
 497          */
 498         LBUG();
 499         return 0;
 500 }
 501
 502 void ccc_transient_page_verify(const struct cl_page *page)
 503 {
 504 }
 505
 506 void ccc_transient_page_own(const struct lu_env *env,
 507                                    const struct cl_page_slice *slice,
 508                                    struct cl_io *_)
 509 {
 510         ccc_transient_page_verify(slice->cpl_page);
 511 }
 512
 513 void ccc_transient_page_assume(const struct lu_env *env,
 514                                       const struct cl_page_slice *slice,
 515                                       struct cl_io *_)
 516 {
 517         ccc_transient_page_verify(slice->cpl_page);
 518 }
 519
 520 void ccc_transient_page_unassume(const struct lu_env *env,
 521                                         const struct cl_page_slice *slice,
 522                                         struct cl_io *_)
 523 {
 524         ccc_transient_page_verify(slice->cpl_page);
 525 }
 526
 527 void ccc_transient_page_disown(const struct lu_env *env,
 528                                       const struct cl_page_slice *slice,
 529                                       struct cl_io *_)
 530 {
 531         ccc_transient_page_verify(slice->cpl_page);
 532 }
 533
 534 void ccc_transient_page_discard(const struct lu_env *env,
 535                                        const struct cl_page_slice *slice,
 536                                        struct cl_io *_)
 537 {
 538         struct cl_page *page = slice->cpl_page;
 539
 540         ccc_transient_page_verify(slice->cpl_page);
 541
 542         /*
 543          * For transient pages, remove it from the radix tree.
 544          */
 545         cl_page_delete(env, page);
 546 }
 547
 548 int ccc_transient_page_prep(const struct lu_env *env,
 549                                    const struct cl_page_slice *slice,
 550                                    struct cl_io *_)
 551 {
 552         ENTRY;
 553         /* transient page should always be sent. */
 554         RETURN(0);
 555 }
 556
 557 /*****************************************************************************
 558  *
 559  * Lock operations.
 560  *
 561  */
 562
 563 void ccc_lock_fini(const struct lu_env *env, struct cl_lock_slice *slice)
 564 {
 565         struct ccc_lock *clk = cl2ccc_lock(slice);
 566
 567         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
 568         OBD_SLAB_FREE_PTR(clk, ccc_lock_kmem);
 569 }
 570
 571 int ccc_lock_enqueue(const struct lu_env *env,
 572                      const struct cl_lock_slice *slice,
 573                      struct cl_io *_, __u32 enqflags)
 574 {
 575         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
 576         return 0;
 577 }
 578
 579 int ccc_lock_unuse(const struct lu_env *env, const struct cl_lock_slice *slice)
 580 {
 581         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
 582         return 0;
 583 }
 584
 585 int ccc_lock_wait(const struct lu_env *env, const struct cl_lock_slice *slice)
 586 {
 587         CLOBINVRNT(env, slice->cls_obj, ccc_object_invariant(slice->cls_obj));
 588         return 0;
 589 }
 590
 591 /**
 592  * Implementation of cl_lock_operations::clo_fits_into() methods for ccc
 593  * layer. This function is executed every time io finds an existing lock in
 594  * the lock cache while creating new lock. This function has to decide whether
 595  * cached lock "fits" into io.
 596  *
 597  * \param slice lock to be checked
 598  *
 599  * \param io    IO that wants a lock.
 600  *
 601  * \see lov_lock_fits_into().
 602  */
 603 int ccc_lock_fits_into(const struct lu_env *env,
 604                        const struct cl_lock_slice *slice,
 605                        const struct cl_lock_descr *need,
 606                        const struct cl_io *io)
 607 {
 608         const struct cl_lock       *lock  = slice->cls_lock;
 609         const struct cl_lock_descr *descr = &lock->cll_descr;
 610         const struct ccc_io        *cio   = ccc_env_io(env);
 611         int                         result;
 612
 613         ENTRY;
 614         /*
 615          * Work around DLM peculiarity: it assumes that glimpse
 616          * (LDLM_FL_HAS_INTENT) lock is always LCK_PR, and returns reads lock
 617          * when asked for LCK_PW lock with LDLM_FL_HAS_INTENT flag set. Make
 618          * sure that glimpse doesn't get CLM_WRITE top-lock, so that it
 619          * doesn't enqueue CLM_WRITE sub-locks.
 620          */
 621         if (cio->cui_glimpse)
 622                 result = descr->cld_mode != CLM_WRITE;
 623         /*
 624          * Also, don't match incomplete write locks for read, otherwise read
 625          * would enqueue missing sub-locks in the write mode.
 626          *
 627          * XXX this is a candidate for generic locking policy, to be moved
 628          * into cl_lock_lookup().
 629          */
 630         else if (need->cld_mode != descr->cld_mode)
 631                 result = lock->cll_state >= CLS_ENQUEUED;
 632         else
 633                 result = 1;
 634         RETURN(result);
 635 }
 636
 637 /**
 638  * Implements cl_lock_operations::clo_state() method for ccc layer, invoked
 639  * whenever lock state changes. Transfers object attributes, that might be
 640  * updated as a result of lock acquiring into inode.
 641  */
 642 void ccc_lock_state(const struct lu_env *env,
 643                     const struct cl_lock_slice *slice,
 644                     enum cl_lock_state state)
 645 {
 646         struct cl_lock   *lock;
 647         struct cl_object *obj;
 648         struct inode     *inode;
 649         struct cl_attr   *attr;
 650
 651         ENTRY;
 652         lock = slice->cls_lock;
 653
 654         /*
 655          * Refresh inode attributes when the lock is moving into CLS_HELD
 656          * state, and only when this is a result of real enqueue, rather than
 657          * of finding lock in the cache.
 658          */
 659         if (state == CLS_HELD && lock->cll_state < CLS_HELD) {
 660                 int rc;
 661
 662                 obj   = slice->cls_obj;
 663                 inode = ccc_object_inode(obj);
 664                 attr  = &ccc_env_info(env)->cti_attr;
 665
 666                 /* vmtruncate()->ll_truncate() first sets the i_size and then
 667                  * the kms under both a DLM lock and the
 668                  * ll_inode_size_lock().  If we don't get the
 669                  * ll_inode_size_lock() here we can match the DLM lock and
 670                  * reset i_size from the kms before the truncating path has
 671                  * updated the kms.  generic_file_write can then trust the
 672                  * stale i_size when doing appending writes and effectively
 673                  * cancel the result of the truncate.  Getting the
 674                  * ll_inode_size_lock() after the enqueue maintains the DLM
 675                  * -> ll_inode_size_lock() acquiring order. */
 676                 cl_isize_lock(inode, 0);
 677                 cl_object_attr_lock(obj);
 678                 rc = cl_object_attr_get(env, obj, attr);
 679                 if (rc == 0) {
 680                         if (lock->cll_descr.cld_start == 0 &&
 681                             lock->cll_descr.cld_end == CL_PAGE_EOF) {
 682                                 cl_isize_write(inode, attr->cat_kms);
 683                                 CDEBUG(D_INODE, DFID" updating i_size %llu\n",
 684                                        PFID(lu_object_fid(&obj->co_lu)),
 685                                        (__u64)cl_isize_read(inode));
 686                         }
 687                         cl_inode_mtime(inode) = attr->cat_mtime;
 688                         cl_inode_atime(inode) = attr->cat_atime;
 689                         cl_inode_ctime(inode) = attr->cat_ctime;
 690                 } else
 691                         CL_LOCK_DEBUG(D_ERROR, env, lock, "attr_get: %i\n", rc);
 692                 cl_object_attr_unlock(obj);
 693                 cl_isize_unlock(inode, 0);
 694         }
 695         EXIT;
 696 }
 697
 698 /*****************************************************************************
 699  *
 700  * io operations.
 701  *
 702  */
 703
 704 void ccc_io_fini(const struct lu_env *env, const struct cl_io_slice *ios)
 705 {
 706         struct cl_io *io = ios->cis_io;
 707
 708         CLOBINVRNT(env, io->ci_obj, ccc_object_invariant(io->ci_obj));
 709 }
 710
 711 int ccc_io_one_lock_index(const struct lu_env *env, struct cl_io *io,
 712                           __u32 enqflags, enum cl_lock_mode mode,
 713                           pgoff_t start, pgoff_t end)
 714 {
 715         struct ccc_io          *vio   = ccc_env_io(env);
 716         struct cl_lock_descr   *descr = &vio->cui_link.cill_descr;
 717         struct cl_object       *obj   = io->ci_obj;
 718
 719         CLOBINVRNT(env, obj, ccc_object_invariant(obj));
 720         ENTRY;
 721
 722         CDEBUG(D_VFSTRACE, "lock: %i [%lu, %lu]\n", mode, start, end);
 723
 724         memset(&vio->cui_link, 0, sizeof vio->cui_link);
 725         descr->cld_mode  = mode;
 726         descr->cld_obj   = obj;
 727         descr->cld_start = start;
 728         descr->cld_end   = end;
 729
 730         vio->cui_link.cill_enq_flags = enqflags;
 731         cl_io_lock_add(env, io, &vio->cui_link);
 732         RETURN(0);
 733 }
 734
 735 int ccc_io_one_lock(const struct lu_env *env, struct cl_io *io,
 736                     __u32 enqflags, enum cl_lock_mode mode,
 737                     loff_t start, loff_t end)
 738 {
 739         struct cl_object *obj = io->ci_obj;
 740
 741         return ccc_io_one_lock_index(env, io, enqflags, mode,
 742                                      cl_index(obj, start), cl_index(obj, end));
 743 }
 744
 745 void ccc_io_end(const struct lu_env *env, const struct cl_io_slice *ios)
 746 {
 747         CLOBINVRNT(env, ios->cis_io->ci_obj,
 748                    ccc_object_invariant(ios->cis_io->ci_obj));
 749 }
 750
 751 static void ccc_object_size_lock(struct cl_object *obj, int vfslock)
 752 {
 753         struct inode *inode = ccc_object_inode(obj);
 754
 755         if (vfslock)
 756                 cl_isize_lock(inode, 0);
 757         cl_object_attr_lock(obj);
 758 }
 759
 760 static void ccc_object_size_unlock(struct cl_object *obj, int vfslock)
 761 {
 762         struct inode *inode = ccc_object_inode(obj);
 763
 764         cl_object_attr_unlock(obj);
 765         if (vfslock)
 766                 cl_isize_unlock(inode, 0);
 767 }
 768
 769 /**
 770  * Helper function that if necessary adjusts file size (inode->i_size), when
 771  * position at the offset \a pos is accessed. File size can be arbitrary stale
 772  * on a Lustre client, but client at least knows KMS. If accessed area is
 773  * inside [0, KMS], set file size to KMS, otherwise glimpse file size.
 774  *
 775  * Locking: cl_isize_lock is used to serialize changes to inode size and to
 776  * protect consistency between inode size and cl_object
 777  * attributes. cl_object_size_lock() protects consistency between cl_attr's of
 778  * top-object and sub-objects.
 779  *
 780  * In page fault path cl_isize_lock cannot be taken, client has to live with
 781  * the resulting races.
 782  */
 783 int ccc_prep_size(const struct lu_env *env, struct cl_object *obj,
 784                   struct cl_io *io, loff_t start, size_t count, int vfslock,
 785                   int *exceed)
 786 {
 787         struct cl_attr *attr  = &ccc_env_info(env)->cti_attr;
 788         struct inode   *inode = ccc_object_inode(obj);
 789         loff_t          pos   = start + count - 1;
 790         loff_t kms;
 791         int result;
 792
 793         /*
 794          * Consistency guarantees: following possibilities exist for the
 795          * relation between region being accessed and real file size at this
 796          * moment:
 797          *
 798          *  (A): the region is completely inside of the file;
 799          *
 800          *  (B-x): x bytes of region are inside of the file, the rest is
 801          *  outside;
 802          *
 803          *  (C): the region is completely outside of the file.
 804          *
 805          * This classification is stable under DLM lock already acquired by
 806          * the caller, because to change the class, other client has to take
 807          * DLM lock conflicting with our lock. Also, any updates to ->i_size
 808          * by other threads on this client are serialized by
 809          * ll_inode_size_lock(). This guarantees that short reads are handled
 810          * correctly in the face of concurrent writes and truncates.
 811          */
 812         ccc_object_size_lock(obj, vfslock);
 813         result = cl_object_attr_get(env, obj, attr);
 814         if (result == 0) {
 815                 kms = attr->cat_kms;
 816                 if (pos > kms) {
 817                         /*
 818                          * A glimpse is necessary to determine whether we
 819                          * return a short read (B) or some zeroes at the end
 820                          * of the buffer (C)
 821                          */
 822                         ccc_object_size_unlock(obj, vfslock);
 823                         result = cl_glimpse_lock(env, io, inode, obj);
 824                         if (result == 0 && exceed != NULL) {
 825                                 /* If objective page index exceed end-of-file
 826                                  * page index, return directly. Do not expect
 827                                  * kernel will check such case correctly.
 828                                  * linux-2.6.18-128.1.1 miss to do that.
 829                                  * --bug 17336 */
 830                                 size_t size = cl_isize_read(inode);
 831                                 unsigned long cur_index = start >> CFS_PAGE_SHIFT;
 832
 833                                 if ((size == 0 && cur_index != 0) ||
 834                                     (((size - 1) >> CFS_PAGE_SHIFT) < cur_index))
 835                                 *exceed = 1;
 836                         }
 837                         return result;
 838                 } else {
 839                         /*
 840                          * region is within kms and, hence, within real file
 841                          * size (A). We need to increase i_size to cover the
 842                          * read region so that generic_file_read() will do its
 843                          * job, but that doesn't mean the kms size is
 844                          * _correct_, it is only the _minimum_ size. If
 845                          * someone does a stat they will get the correct size
 846                          * which will always be >= the kms value here.
 847                          * b=11081
 848                          */
 849                         /*
 850                          * XXX in a page fault path, change inode size without
 851                          * ll_inode_size_lock() held!  there is a race
 852                          * condition with truncate path. (see ll_extent_lock)
 853                          */
 854                         /*
 855                          * XXX i_size_write() is not used because it is not
 856                          * safe to take the ll_inode_size_lock() due to a
 857                          * potential lock inversion (bug 6077).  And since
 858                          * it's not safe to use i_size_write() without a
 859                          * covering mutex we do the assignment directly.  It
 860                          * is not critical that the size be correct.
 861                          */
 862                         if (cl_isize_read(inode) < kms) {
 863                                 if (vfslock)
 864                                         cl_isize_write(inode, kms);
 865                                 else
 866                                         cl_isize_write_nolock(inode, kms);
 867                         }
 868                 }
 869         }
 870         ccc_object_size_unlock(obj, vfslock);
 871         return result;
 872 }
 873
 874 /*****************************************************************************
 875  *
 876  * Transfer operations.
 877  *
 878  */
 879
 880 void ccc_req_completion(const struct lu_env *env,
 881                         const struct cl_req_slice *slice, int ioret)
 882 {
 883         struct ccc_req *vrq;
 884
 885         vrq = cl2ccc_req(slice);
 886         OBD_SLAB_FREE_PTR(vrq, ccc_req_kmem);
 887 }
 888
 889 /**
 890  * Implementation of struct cl_req_operations::cro_attr_set() for ccc
 891  * layer. ccc is responsible for
 892  *
 893  *    - o_[mac]time
 894  *
 895  *    - o_mode
 896  *
 897  *    - o_fid (filled with inode number?!)
 898  *
 899  *    - o_[ug]id
 900  *
 901  *    - o_generation
 902  *
 903  *    - and IO epoch (stored in o_easize),
 904  *
 905  *  and capability.
 906  */
 907 void ccc_req_attr_set(const struct lu_env *env,
 908                       const struct cl_req_slice *slice,
 909                       const struct cl_object *obj,
 910                       struct cl_req_attr *attr, obd_valid flags)
 911 {
 912         struct inode *inode;
 913         struct obdo  *oa;
 914         obd_flag      valid_flags;
 915
 916         oa = attr->cra_oa;
 917         inode = ccc_object_inode(obj);
 918         valid_flags = OBD_MD_FLTYPE|OBD_MD_FLATIME;
 919
 920         if (flags != (obd_valid)~0ULL)
 921                 valid_flags |= OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME;
 922         else {
 923                 LASSERT(attr->cra_capa == NULL);
 924                 attr->cra_capa = cl_capa_lookup(inode,
 925                                                 slice->crs_req->crq_type);
 926         }
 927
 928         if (slice->crs_req->crq_type == CRT_WRITE) {
 929                 if (flags & OBD_MD_FLEPOCH) {
 930                         oa->o_valid |= OBD_MD_FLEPOCH;
 931                         oa->o_easize = cl_i2info(inode)->lli_ioepoch;
 932                         valid_flags |= OBD_MD_FLMTIME|OBD_MD_FLCTIME|
 933                                 OBD_MD_FLUID|OBD_MD_FLGID|
 934                                 OBD_MD_FLFID|OBD_MD_FLGENER;
 935                 }
 936         }
 937         obdo_from_inode(oa, inode, valid_flags & flags);
 938 }
 939
 940 const struct cl_req_operations ccc_req_ops = {
 941         .cro_attr_set   = ccc_req_attr_set,
 942         .cro_completion = ccc_req_completion
 943 };
 944
 945 /* Setattr helpers */
 946 int cl_setattr_do_truncate(struct inode *inode, loff_t size,
 947                            struct obd_capa *capa)
 948 {
 949         struct lu_env *env;
 950         struct cl_io  *io;
 951         int            result;
 952         int            refcheck;
 953
 954         ENTRY;
 955
 956         env = cl_env_get(&refcheck);
 957         if (IS_ERR(env))
 958                 RETURN(PTR_ERR(env));
 959
 960         io = &ccc_env_info(env)->cti_io;
 961         io->ci_obj = cl_i2info(inode)->lli_clob;
 962         io->u.ci_truncate.tr_size = size;
 963         io->u.ci_truncate.tr_capa = capa;
 964         if (cl_io_init(env, io, CIT_TRUNC, io->ci_obj) == 0)
 965                 result = cl_io_loop(env, io);
 966         else
 967                 result = io->ci_result;
 968         cl_io_fini(env, io);
 969         cl_env_put(env, &refcheck);
 970         RETURN(result);
 971 }
 972
 973 int cl_setattr_ost(struct inode *inode, struct obd_capa *capa)
 974 {
 975         struct cl_inode_info *lli = cl_i2info(inode);
 976         struct lov_stripe_md *lsm = lli->lli_smd;
 977         int rc;
 978         obd_flag flags;
 979         struct obd_info oinfo = { { { 0 } } };
 980         struct obdo *oa;
 981
 982         OBDO_ALLOC(oa);
 983         if (oa) {
 984                 oa->o_id = lsm->lsm_object_id;
 985                 oa->o_gr = lsm->lsm_object_gr;
 986                 oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
 987
 988                 flags = OBD_MD_FLTYPE | OBD_MD_FLATIME |
 989                         OBD_MD_FLMTIME | OBD_MD_FLCTIME |
 990                         OBD_MD_FLFID | OBD_MD_FLGENER |
 991                         OBD_MD_FLGROUP;
 992
 993                 obdo_from_inode(oa, inode, flags);
 994
 995                 oinfo.oi_oa = oa;
 996                 oinfo.oi_md = lsm;
 997
 998                 /* XXX: this looks unnecessary now. */
 999                 rc = obd_setattr_rqset(cl_i2sbi(inode)->ll_dt_exp, &oinfo,
1000                                        NULL);
1001                 if (rc)
1002                         CERROR("obd_setattr_async fails: rc=%d\n", rc);
1003                 OBDO_FREE(oa);
1004         } else {
1005                 rc = -ENOMEM;
1006         }
1007         return rc;
1008 }
1009
1010
1011 /*****************************************************************************
1012  *
1013  * Type conversions.
1014  *
1015  */
1016
1017 struct lu_device *ccc2lu_dev(struct ccc_device *vdv)
1018 {
1019         return &vdv->cdv_cl.cd_lu_dev;
1020 }
1021
1022 struct ccc_device *lu2ccc_dev(const struct lu_device *d)
1023 {
1024         return container_of0(d, struct ccc_device, cdv_cl.cd_lu_dev);
1025 }
1026
1027 struct ccc_device *cl2ccc_dev(const struct cl_device *d)
1028 {
1029         return container_of0(d, struct ccc_device, cdv_cl);
1030 }
1031
1032 struct lu_object *ccc2lu(struct ccc_object *vob)
1033 {
1034         return &vob->cob_cl.co_lu;
1035 }
1036
1037 struct ccc_object *lu2ccc(const struct lu_object *obj)
1038 {
1039         return container_of0(obj, struct ccc_object, cob_cl.co_lu);
1040 }
1041
1042 struct ccc_object *cl2ccc(const struct cl_object *obj)
1043 {
1044         return container_of0(obj, struct ccc_object, cob_cl);
1045 }
1046
1047 struct ccc_lock *cl2ccc_lock(const struct cl_lock_slice *slice)
1048 {
1049         return container_of(slice, struct ccc_lock, clk_cl);
1050 }
1051
1052 struct ccc_io *cl2ccc_io(const struct lu_env *env,
1053                          const struct cl_io_slice *slice)
1054 {
1055         struct ccc_io *cio;
1056
1057         cio = container_of(slice, struct ccc_io, cui_cl);
1058         LASSERT(cio == ccc_env_io(env));
1059         return cio;
1060 }
1061
1062 struct ccc_req *cl2ccc_req(const struct cl_req_slice *slice)
1063 {
1064         return container_of0(slice, struct ccc_req, crq_cl);
1065 }
1066
1067 cfs_page_t *cl2vm_page(const struct cl_page_slice *slice)
1068 {
1069         return cl2ccc_page(slice)->cpg_page;
1070 }
1071
1072 /*****************************************************************************
1073  *
1074  * Accessors.
1075  *
1076  */
1077 int ccc_object_invariant(const struct cl_object *obj)
1078 {
1079         struct inode         *inode = ccc_object_inode(obj);
1080         struct cl_inode_info *lli   = cl_i2info(inode);
1081
1082         return (S_ISREG(cl_inode_mode(inode)) ||
1083                 /* i_mode of unlinked inode is zeroed. */
1084                 cl_inode_mode(inode) == 0) && lli->lli_clob == obj;
1085 }
1086
1087 struct inode *ccc_object_inode(const struct cl_object *obj)
1088 {
1089         return cl2ccc(obj)->cob_inode;
1090 }
1091
1092 /**
1093  * Returns a pointer to cl_page associated with \a vmpage, without acquiring
1094  * additional reference to the resulting page. This is an unsafe version of
1095  * cl_vmpage_page() that can only be used under vmpage lock.
1096  */
1097 struct cl_page *ccc_vmpage_page_transient(cfs_page_t *vmpage)
1098 {
1099         KLASSERT(PageLocked(vmpage));
1100         return (struct cl_page *)vmpage->private;
1101 }
1102
1103 /**
1104  * Initializes or updates CLIO part when new meta-data arrives from the
1105  * server.
1106  *
1107  *     - allocates cl_object if necessary,
1108  *     - updated layout, if object was already here.
1109  */
1110 int cl_inode_init(struct inode *inode, struct lustre_md *md)
1111 {
1112         struct lu_env        *env;
1113         struct cl_inode_info *lli;
1114         struct cl_object     *clob;
1115         struct lu_site       *site;
1116         struct lu_fid        *fid;
1117         const struct cl_object_conf conf = {
1118                 .coc_inode = inode,
1119                 .u = {
1120                         .coc_md    = md
1121                 }
1122         };
1123         int result = 0;
1124         int refcheck;
1125
1126         /* LASSERT(inode->i_state & I_NEW); */
1127         LASSERT(md->body->valid & OBD_MD_FLID);
1128
1129         if (!S_ISREG(cl_inode_mode(inode)))
1130                 return 0;
1131
1132         env = cl_env_get(&refcheck);
1133         if (IS_ERR(env))
1134                 return PTR_ERR(env);
1135
1136         site = cl_i2sbi(inode)->ll_site;
1137         lli  = cl_i2info(inode);
1138         fid  = &lli->lli_fid;
1139         LASSERT(fid_is_sane(fid));
1140
1141         if (lli->lli_clob == NULL) {
1142                 clob = cl_object_find(env, lu2cl_dev(site->ls_top_dev),
1143                                       fid, &conf);
1144                 if (!IS_ERR(clob)) {
1145                         /*
1146                          * No locking is necessary, as new inode is
1147                          * locked by I_NEW bit.
1148                          *
1149                          * XXX not true for call from ll_update_inode().
1150                          */
1151                         lli->lli_clob = clob;
1152                         lu_object_ref_add(&clob->co_lu, "inode", inode);
1153                 } else
1154                         result = PTR_ERR(clob);
1155         } else
1156                 result = cl_conf_set(env, lli->lli_clob, &conf);
1157         cl_env_put(env, &refcheck);
1158
1159         if (result != 0)
1160                 CERROR("Failure to initialize cl object "DFID": %d\n",
1161                        PFID(fid), result);
1162         return result;
1163 }
1164
1165 void cl_inode_fini(struct inode *inode)
1166 {
1167         struct lu_env           *env;
1168         struct cl_inode_info    *lli  = cl_i2info(inode);
1169         struct cl_object        *clob = lli->lli_clob;
1170         int refcheck;
1171         int emergency;
1172
1173         if (clob != NULL) {
1174                 struct lu_object_header *head = clob->co_lu.lo_header;
1175                 void                    *cookie;
1176
1177                 cookie = cl_env_reenter();
1178                 env = cl_env_get(&refcheck);
1179                 emergency = IS_ERR(env);
1180                 if (emergency) {
1181                         mutex_lock(&ccc_inode_fini_guard);
1182                         LASSERT(ccc_inode_fini_env != NULL);
1183                         cl_env_implant(ccc_inode_fini_env, &refcheck);
1184                         env = ccc_inode_fini_env;
1185                 }
1186                 /*
1187                  * cl_object cache is a slave to inode cache (which, in turn
1188                  * is a slave to dentry cache), don't keep cl_object in memory
1189                  * when its master is evicted.
1190                  */
1191                 cl_object_kill(env, clob);
1192                 lu_object_ref_del(&clob->co_lu, "inode", inode);
1193                 /* XXX temporary: this is racy */
1194                 LASSERT(atomic_read(&head->loh_ref) == 1);
1195                 cl_object_put(env, clob);
1196                 lli->lli_clob = NULL;
1197                 if (emergency) {
1198                         cl_env_unplant(ccc_inode_fini_env, &refcheck);
1199                         mutex_unlock(&ccc_inode_fini_guard);
1200                 } else
1201                         cl_env_put(env, &refcheck);
1202                 cl_env_reexit(cookie);
1203         }
1204 }
1205
1206 /**
1207  * return IF_* type for given lu_dirent entry.
1208  * IF_* flag shld be converted to particular OS file type in
1209  * platform llite module.
1210  */
1211 __u16 ll_dirent_type_get(struct lu_dirent *ent)
1212 {
1213         __u16 type = 0;
1214         struct luda_type *lt;
1215         int len = 0;
1216
1217         if (le32_to_cpu(ent->lde_attrs) & LUDA_TYPE) {
1218                 const unsigned align = sizeof(struct luda_type) - 1;
1219
1220                 len = le16_to_cpu(ent->lde_namelen);
1221                 len = (len + align) & ~align;
1222                 lt = (void *) ent->lde_name + len;
1223                 type = CFS_IFTODT(le16_to_cpu(lt->lt_type));
1224         }
1225         return type;
1226 }