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(info, ccc_thread_kmem);
 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(session, ccc_session_kmem);
 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(vrq, ccc_req_kmem);
 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(vob, ccc_object_kmem);
 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(clk, ccc_lock_kmem);
 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 pos, int vfslock)
 785 {
 786         struct cl_attr *attr  = &ccc_env_info(env)->cti_attr;
 787         struct inode   *inode = ccc_object_inode(obj);
 788         loff_t kms;
 789         int result;
 790
 791         /*
 792          * Consistency guarantees: following possibilities exist for the
 793          * relation between region being accessed and real file size at this
 794          * moment:
 795          *
 796          *  (A): the region is completely inside of the file;
 797          *
 798          *  (B-x): x bytes of region are inside of the file, the rest is
 799          *  outside;
 800          *
 801          *  (C): the region is completely outside of the file.
 802          *
 803          * This classification is stable under DLM lock already acquired by
 804          * the caller, because to change the class, other client has to take
 805          * DLM lock conflicting with our lock. Also, any updates to ->i_size
 806          * by other threads on this client are serialized by
 807          * ll_inode_size_lock(). This guarantees that short reads are handled
 808          * correctly in the face of concurrent writes and truncates.
 809          */
 810         ccc_object_size_lock(obj, vfslock);
 811         result = cl_object_attr_get(env, obj, attr);
 812         if (result == 0) {
 813                 kms = attr->cat_kms;
 814                 if (pos > kms) {
 815                         /*
 816                          * A glimpse is necessary to determine whether we
 817                          * return a short read (B) or some zeroes at the end
 818                          * of the buffer (C)
 819                          */
 820                         ccc_object_size_unlock(obj, vfslock);
 821                         return cl_glimpse_lock(env, io, inode, obj);
 822                 } else {
 823                         /*
 824                          * region is within kms and, hence, within real file
 825                          * size (A). We need to increase i_size to cover the
 826                          * read region so that generic_file_read() will do its
 827                          * job, but that doesn't mean the kms size is
 828                          * _correct_, it is only the _minimum_ size. If
 829                          * someone does a stat they will get the correct size
 830                          * which will always be >= the kms value here.
 831                          * b=11081
 832                          */
 833                         /*
 834                          * XXX in a page fault path, change inode size without
 835                          * ll_inode_size_lock() held!  there is a race
 836                          * condition with truncate path. (see ll_extent_lock)
 837                          */
 838                         /*
 839                          * XXX i_size_write() is not used because it is not
 840                          * safe to take the ll_inode_size_lock() due to a
 841                          * potential lock inversion (bug 6077).  And since
 842                          * it's not safe to use i_size_write() without a
 843                          * covering mutex we do the assignment directly.  It
 844                          * is not critical that the size be correct.
 845                          */
 846                         if (cl_isize_read(inode) < kms) {
 847                                 if (vfslock)
 848                                         cl_isize_write(inode, kms);
 849                                 else
 850                                         cl_isize_write_nolock(inode, kms);
 851                         }
 852                 }
 853         }
 854         ccc_object_size_unlock(obj, vfslock);
 855         return result;
 856 }
 857
 858 /*****************************************************************************
 859  *
 860  * Transfer operations.
 861  *
 862  */
 863
 864 void ccc_req_completion(const struct lu_env *env,
 865                         const struct cl_req_slice *slice, int ioret)
 866 {
 867         struct ccc_req *vrq;
 868
 869         vrq = cl2ccc_req(slice);
 870         OBD_SLAB_FREE_PTR(vrq, ccc_req_kmem);
 871 }
 872
 873 /**
 874  * Implementation of struct cl_req_operations::cro_attr_set() for ccc
 875  * layer. ccc is responsible for
 876  *
 877  *    - o_[mac]time
 878  *
 879  *    - o_mode
 880  *
 881  *    - o_fid (filled with inode number?!)
 882  *
 883  *    - o_[ug]id
 884  *
 885  *    - o_generation
 886  *
 887  *    - and IO epoch (stored in o_easize),
 888  *
 889  *  and capability.
 890  */
 891 void ccc_req_attr_set(const struct lu_env *env,
 892                       const struct cl_req_slice *slice,
 893                       const struct cl_object *obj,
 894                       struct cl_req_attr *attr, obd_valid flags)
 895 {
 896         struct inode *inode;
 897         struct obdo  *oa;
 898         obd_flag      valid_flags;
 899
 900         oa = attr->cra_oa;
 901         inode = ccc_object_inode(obj);
 902         valid_flags = OBD_MD_FLTYPE|OBD_MD_FLATIME;
 903
 904         if (flags != (obd_valid)~0ULL)
 905                 valid_flags |= OBD_MD_FLMTIME|OBD_MD_FLCTIME|OBD_MD_FLATIME;
 906         else {
 907                 LASSERT(attr->cra_capa == NULL);
 908                 attr->cra_capa = cl_capa_lookup(inode,
 909                                                 slice->crs_req->crq_type);
 910         }
 911
 912         if (slice->crs_req->crq_type == CRT_WRITE) {
 913                 if (flags & OBD_MD_FLEPOCH) {
 914                         oa->o_valid |= OBD_MD_FLEPOCH;
 915                         oa->o_easize = cl_i2info(inode)->lli_ioepoch;
 916                         valid_flags |= OBD_MD_FLMTIME|OBD_MD_FLCTIME|
 917                                 OBD_MD_FLUID|OBD_MD_FLGID|
 918                                 OBD_MD_FLFID|OBD_MD_FLGENER;
 919                 }
 920         }
 921         obdo_from_inode(oa, inode, valid_flags & flags);
 922 }
 923
 924 const struct cl_req_operations ccc_req_ops = {
 925         .cro_attr_set   = ccc_req_attr_set,
 926         .cro_completion = ccc_req_completion
 927 };
 928
 929 /* Setattr helpers */
 930 int cl_setattr_do_truncate(struct inode *inode, loff_t size,
 931                            struct obd_capa *capa)
 932 {
 933         struct lu_env *env;
 934         struct cl_io  *io;
 935         int            result;
 936         int            refcheck;
 937
 938         ENTRY;
 939
 940         env = cl_env_get(&refcheck);
 941         if (IS_ERR(env))
 942                 RETURN(PTR_ERR(env));
 943
 944         io = &ccc_env_info(env)->cti_io;
 945         io->ci_obj = cl_i2info(inode)->lli_clob;
 946         io->u.ci_truncate.tr_size = size;
 947         io->u.ci_truncate.tr_capa = capa;
 948         if (cl_io_init(env, io, CIT_TRUNC, io->ci_obj) == 0)
 949                 result = cl_io_loop(env, io);
 950         else
 951                 result = io->ci_result;
 952         cl_io_fini(env, io);
 953         cl_env_put(env, &refcheck);
 954         RETURN(result);
 955 }
 956
 957 int cl_setattr_ost(struct inode *inode, struct obd_capa *capa)
 958 {
 959         struct cl_inode_info *lli = cl_i2info(inode);
 960         struct lov_stripe_md *lsm = lli->lli_smd;
 961         int rc;
 962         obd_flag flags;
 963         struct obd_info oinfo = { { { 0 } } };
 964         struct obdo *oa;
 965
 966         OBDO_ALLOC(oa);
 967         if (oa) {
 968                 oa->o_id = lsm->lsm_object_id;
 969                 oa->o_gr = lsm->lsm_object_gr;
 970                 oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
 971
 972                 flags = OBD_MD_FLTYPE | OBD_MD_FLATIME |
 973                         OBD_MD_FLMTIME | OBD_MD_FLCTIME |
 974                         OBD_MD_FLFID | OBD_MD_FLGENER |
 975                         OBD_MD_FLGROUP;
 976
 977                 obdo_from_inode(oa, inode, flags);
 978
 979                 oinfo.oi_oa = oa;
 980                 oinfo.oi_md = lsm;
 981
 982                 /* XXX: this looks unnecessary now. */
 983                 rc = obd_setattr_rqset(cl_i2sbi(inode)->ll_dt_exp, &oinfo,
 984                                        NULL);
 985                 if (rc)
 986                         CERROR("obd_setattr_async fails: rc=%d\n", rc);
 987                 OBDO_FREE(oa);
 988         } else {
 989                 rc = -ENOMEM;
 990         }
 991         return rc;
 992 }
 993
 994
 995 /*****************************************************************************
 996  *
 997  * Type conversions.
 998  *
 999  */
1000
1001 struct lu_device *ccc2lu_dev(struct ccc_device *vdv)
1002 {
1003         return &vdv->cdv_cl.cd_lu_dev;
1004 }
1005
1006 struct ccc_device *lu2ccc_dev(const struct lu_device *d)
1007 {
1008         return container_of0(d, struct ccc_device, cdv_cl.cd_lu_dev);
1009 }
1010
1011 struct ccc_device *cl2ccc_dev(const struct cl_device *d)
1012 {
1013         return container_of0(d, struct ccc_device, cdv_cl);
1014 }
1015
1016 struct lu_object *ccc2lu(struct ccc_object *vob)
1017 {
1018         return &vob->cob_cl.co_lu;
1019 }
1020
1021 struct ccc_object *lu2ccc(const struct lu_object *obj)
1022 {
1023         return container_of0(obj, struct ccc_object, cob_cl.co_lu);
1024 }
1025
1026 struct ccc_object *cl2ccc(const struct cl_object *obj)
1027 {
1028         return container_of0(obj, struct ccc_object, cob_cl);
1029 }
1030
1031 struct ccc_lock *cl2ccc_lock(const struct cl_lock_slice *slice)
1032 {
1033         return container_of(slice, struct ccc_lock, clk_cl);
1034 }
1035
1036 struct ccc_io *cl2ccc_io(const struct lu_env *env,
1037                          const struct cl_io_slice *slice)
1038 {
1039         struct ccc_io *cio;
1040
1041         cio = container_of(slice, struct ccc_io, cui_cl);
1042         LASSERT(cio == ccc_env_io(env));
1043         return cio;
1044 }
1045
1046 struct ccc_req *cl2ccc_req(const struct cl_req_slice *slice)
1047 {
1048         return container_of0(slice, struct ccc_req, crq_cl);
1049 }
1050
1051 cfs_page_t *cl2vm_page(const struct cl_page_slice *slice)
1052 {
1053         return cl2ccc_page(slice)->cpg_page;
1054 }
1055
1056 /*****************************************************************************
1057  *
1058  * Accessors.
1059  *
1060  */
1061 int ccc_object_invariant(const struct cl_object *obj)
1062 {
1063         struct inode         *inode = ccc_object_inode(obj);
1064         struct cl_inode_info *lli   = cl_i2info(inode);
1065
1066         return (S_ISREG(cl_inode_mode(inode)) ||
1067                 /* i_mode of unlinked inode is zeroed. */
1068                 cl_inode_mode(inode) == 0) && lli->lli_clob == obj;
1069 }
1070
1071 struct inode *ccc_object_inode(const struct cl_object *obj)
1072 {
1073         return cl2ccc(obj)->cob_inode;
1074 }
1075
1076 /**
1077  * Returns a pointer to cl_page associated with \a vmpage, without acquiring
1078  * additional reference to the resulting page. This is an unsafe version of
1079  * cl_vmpage_page() that can only be used under vmpage lock.
1080  */
1081 struct cl_page *ccc_vmpage_page_transient(cfs_page_t *vmpage)
1082 {
1083         KLASSERT(PageLocked(vmpage));
1084         return (struct cl_page *)vmpage->private;
1085 }
1086
1087 /**
1088  * Initializes or updates CLIO part when new meta-data arrives from the
1089  * server.
1090  *
1091  *     - allocates cl_object if necessary,
1092  *     - updated layout, if object was already here.
1093  */
1094 int cl_inode_init(struct inode *inode, struct lustre_md *md)
1095 {
1096         struct lu_env        *env;
1097         struct cl_inode_info *lli;
1098         struct cl_object     *clob;
1099         struct lu_site       *site;
1100         struct lu_fid        *fid;
1101         const struct cl_object_conf conf = {
1102                 .coc_inode = inode,
1103                 .u = {
1104                         .coc_md    = md
1105                 }
1106         };
1107         int result = 0;
1108         int refcheck;
1109
1110         /* LASSERT(inode->i_state & I_NEW); */
1111         LASSERT(md->body->valid & OBD_MD_FLID);
1112
1113         if (!S_ISREG(cl_inode_mode(inode)))
1114                 return 0;
1115
1116         env = cl_env_get(&refcheck);
1117         if (IS_ERR(env))
1118                 return PTR_ERR(env);
1119
1120         site = cl_i2sbi(inode)->ll_site;
1121         lli  = cl_i2info(inode);
1122         fid  = &lli->lli_fid;
1123         LASSERT(fid_is_sane(fid));
1124
1125         if (lli->lli_clob == NULL) {
1126                 clob = cl_object_find(env, lu2cl_dev(site->ls_top_dev),
1127                                       fid, &conf);
1128                 if (!IS_ERR(clob)) {
1129                         /*
1130                          * No locking is necessary, as new inode is
1131                          * locked by I_NEW bit.
1132                          *
1133                          * XXX not true for call from ll_update_inode().
1134                          */
1135                         lli->lli_clob = clob;
1136                         lu_object_ref_add(&clob->co_lu, "inode", inode);
1137                 } else
1138                         result = PTR_ERR(clob);
1139         } else
1140                 result = cl_conf_set(env, lli->lli_clob, &conf);
1141         cl_env_put(env, &refcheck);
1142
1143         if (result != 0)
1144                 CERROR("Failure to initialize cl object "DFID": %d\n",
1145                        PFID(fid), result);
1146         return result;
1147 }
1148
1149 void cl_inode_fini(struct inode *inode)
1150 {
1151         struct lu_env           *env;
1152         struct cl_inode_info    *lli  = cl_i2info(inode);
1153         struct cl_object        *clob = lli->lli_clob;
1154         int refcheck;
1155         int emergency;
1156
1157         if (clob != NULL) {
1158                 struct lu_object_header *head = clob->co_lu.lo_header;
1159                 void                    *cookie;
1160
1161                 cookie = cl_env_reenter();
1162                 env = cl_env_get(&refcheck);
1163                 emergency = IS_ERR(env);
1164                 if (emergency) {
1165                         mutex_lock(&ccc_inode_fini_guard);
1166                         LASSERT(ccc_inode_fini_env != NULL);
1167                         cl_env_implant(ccc_inode_fini_env, &refcheck);
1168                         env = ccc_inode_fini_env;
1169                 }
1170                 /*
1171                  * cl_object cache is a slave to inode cache (which, in turn
1172                  * is a slave to dentry cache), don't keep cl_object in memory
1173                  * when its master is evicted.
1174                  */
1175                 cl_object_kill(env, clob);
1176                 lu_object_ref_del(&clob->co_lu, "inode", inode);
1177                 /* XXX temporary: this is racy */
1178                 LASSERT(atomic_read(&head->loh_ref) == 1);
1179                 cl_object_put(env, clob);
1180                 lli->lli_clob = NULL;
1181                 if (emergency) {
1182                         cl_env_unplant(ccc_inode_fini_env, &refcheck);
1183                         mutex_unlock(&ccc_inode_fini_guard);
1184                 } else
1185                         cl_env_put(env, &refcheck);
1186                 cl_env_reexit(cookie);
1187         }
1188 }