lustre/lov/lov_object.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2014, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * Implementation of cl_object for LOV layer.
  37  *
  38  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  39  *   Author: Jinshan Xiong <jinshan.xiong@whamcloud.com>
  40  */
  41
  42 #define DEBUG_SUBSYSTEM S_LOV
  43
  44 #include "lov_cl_internal.h"
  45
  46 /** \addtogroup lov
  47  *  @{
  48  */
  49
  50 /*****************************************************************************
  51  *
  52  * Layout operations.
  53  *
  54  */
  55
  56 struct lov_layout_operations {
  57         int (*llo_init)(const struct lu_env *env, struct lov_device *dev,
  58                         struct lov_object *lov,
  59                         const struct cl_object_conf *conf,
  60                         union lov_layout_state *state);
  61         int (*llo_delete)(const struct lu_env *env, struct lov_object *lov,
  62                            union lov_layout_state *state);
  63         void (*llo_fini)(const struct lu_env *env, struct lov_object *lov,
  64                          union lov_layout_state *state);
  65         void (*llo_install)(const struct lu_env *env, struct lov_object *lov,
  66                             union lov_layout_state *state);
  67         int  (*llo_print)(const struct lu_env *env, void *cookie,
  68                           lu_printer_t p, const struct lu_object *o);
  69         int  (*llo_page_init)(const struct lu_env *env, struct cl_object *obj,
  70                               struct cl_page *page, pgoff_t index);
  71         int  (*llo_lock_init)(const struct lu_env *env,
  72                               struct cl_object *obj, struct cl_lock *lock,
  73                               const struct cl_io *io);
  74         int  (*llo_io_init)(const struct lu_env *env,
  75                             struct cl_object *obj, struct cl_io *io);
  76         int  (*llo_getattr)(const struct lu_env *env, struct cl_object *obj,
  77                             struct cl_attr *attr);
  78 };
  79
  80 static int lov_layout_wait(const struct lu_env *env, struct lov_object *lov);
  81
  82 struct lov_stripe_md *lov_lsm_get(struct cl_object *clobj)
  83 {
  84         struct lu_object *luobj;
  85         struct lov_stripe_md *lsm = NULL;
  86
  87         if (clobj == NULL)
  88                 return NULL;
  89
  90         luobj = lu_object_locate(&cl_object_header(clobj)->coh_lu,
  91                                  &lov_device_type);
  92         if (luobj != NULL)
  93                 lsm = lov_lsm_addref(lu2lov(luobj));
  94         return lsm;
  95 }
  96 EXPORT_SYMBOL(lov_lsm_get);
  97
  98 void lov_lsm_put(struct cl_object *unused, struct lov_stripe_md *lsm)
  99 {
 100         if (lsm != NULL)
 101                 lov_free_memmd(&lsm);
 102 }
 103 EXPORT_SYMBOL(lov_lsm_put);
 104
 105 /*****************************************************************************
 106  *
 107  * Lov object layout operations.
 108  *
 109  */
 110
 111 static void lov_install_empty(const struct lu_env *env,
 112                               struct lov_object *lov,
 113                               union  lov_layout_state *state)
 114 {
 115         /*
 116          * File without objects.
 117          */
 118 }
 119
 120 static int lov_init_empty(const struct lu_env *env,
 121                           struct lov_device *dev, struct lov_object *lov,
 122                           const struct cl_object_conf *conf,
 123                           union  lov_layout_state *state)
 124 {
 125         return 0;
 126 }
 127
 128 static void lov_install_raid0(const struct lu_env *env,
 129                               struct lov_object *lov,
 130                               union  lov_layout_state *state)
 131 {
 132 }
 133
 134 static struct cl_object *lov_sub_find(const struct lu_env *env,
 135                                       struct cl_device *dev,
 136                                       const struct lu_fid *fid,
 137                                       const struct cl_object_conf *conf)
 138 {
 139         struct lu_object *o;
 140
 141         ENTRY;
 142         o = lu_object_find_at(env, cl2lu_dev(dev), fid, &conf->coc_lu);
 143         LASSERT(ergo(!IS_ERR(o), o->lo_dev->ld_type == &lovsub_device_type));
 144         RETURN(lu2cl(o));
 145 }
 146
 147 static int lov_init_sub(const struct lu_env *env, struct lov_object *lov,
 148                         struct cl_object *stripe, struct lov_layout_raid0 *r0,
 149                         int idx)
 150 {
 151         struct cl_object_header *hdr;
 152         struct cl_object_header *subhdr;
 153         struct cl_object_header *parent;
 154         struct lov_oinfo        *oinfo;
 155         int result;
 156
 157         if (OBD_FAIL_CHECK(OBD_FAIL_LOV_INIT)) {
 158                 /* For sanity:test_206.
 159                  * Do not leave the object in cache to avoid accessing
 160                  * freed memory. This is because osc_object is referring to
 161                  * lov_oinfo of lsm_stripe_data which will be freed due to
 162                  * this failure. */
 163                 cl_object_kill(env, stripe);
 164                 cl_object_put(env, stripe);
 165                 return -EIO;
 166         }
 167
 168         hdr    = cl_object_header(lov2cl(lov));
 169         subhdr = cl_object_header(stripe);
 170
 171         oinfo = lov->lo_lsm->lsm_oinfo[idx];
 172         CDEBUG(D_INODE, DFID"@%p[%d] -> "DFID"@%p: ostid: "DOSTID
 173                " idx: %d gen: %d\n",
 174                PFID(&subhdr->coh_lu.loh_fid), subhdr, idx,
 175                PFID(&hdr->coh_lu.loh_fid), hdr, POSTID(&oinfo->loi_oi),
 176                oinfo->loi_ost_idx, oinfo->loi_ost_gen);
 177
 178         /* reuse ->coh_attr_guard to protect coh_parent change */
 179         spin_lock(&subhdr->coh_attr_guard);
 180         parent = subhdr->coh_parent;
 181         if (parent == NULL) {
 182                 subhdr->coh_parent = hdr;
 183                 spin_unlock(&subhdr->coh_attr_guard);
 184                 subhdr->coh_nesting = hdr->coh_nesting + 1;
 185                 lu_object_ref_add(&stripe->co_lu, "lov-parent", lov);
 186                 r0->lo_sub[idx] = cl2lovsub(stripe);
 187                 r0->lo_sub[idx]->lso_super = lov;
 188                 r0->lo_sub[idx]->lso_index = idx;
 189                 result = 0;
 190         } else {
 191                 struct lu_object  *old_obj;
 192                 struct lov_object *old_lov;
 193                 unsigned int mask = D_INODE;
 194
 195                 spin_unlock(&subhdr->coh_attr_guard);
 196                 old_obj = lu_object_locate(&parent->coh_lu, &lov_device_type);
 197                 LASSERT(old_obj != NULL);
 198                 old_lov = cl2lov(lu2cl(old_obj));
 199                 if (old_lov->lo_layout_invalid) {
 200                         /* the object's layout has already changed but isn't
 201                          * refreshed */
 202                         lu_object_unhash(env, &stripe->co_lu);
 203                         result = -EAGAIN;
 204                 } else {
 205                         mask = D_ERROR;
 206                         result = -EIO;
 207                 }
 208
 209                 LU_OBJECT_DEBUG(mask, env, &stripe->co_lu,
 210                                 "stripe %d is already owned.\n", idx);
 211                 LU_OBJECT_DEBUG(mask, env, old_obj, "owned.\n");
 212                 LU_OBJECT_HEADER(mask, env, lov2lu(lov), "try to own.\n");
 213                 cl_object_put(env, stripe);
 214         }
 215         return result;
 216 }
 217
 218 static int lov_page_slice_fixup(struct lov_object *lov,
 219                                 struct cl_object *stripe)
 220 {
 221         struct cl_object_header *hdr = cl_object_header(&lov->lo_cl);
 222         struct cl_object *o;
 223
 224         if (stripe == NULL)
 225                 return hdr->coh_page_bufsize - lov->lo_cl.co_slice_off -
 226                        cfs_size_round(sizeof(struct lov_page));
 227
 228         cl_object_for_each(o, stripe)
 229                 o->co_slice_off += hdr->coh_page_bufsize;
 230
 231         return cl_object_header(stripe)->coh_page_bufsize;
 232 }
 233
 234 static int lov_init_raid0(const struct lu_env *env,
 235                           struct lov_device *dev, struct lov_object *lov,
 236                           const struct cl_object_conf *conf,
 237                           union  lov_layout_state *state)
 238 {
 239         int result;
 240         int i;
 241
 242         struct cl_object        *stripe;
 243         struct lov_thread_info  *lti     = lov_env_info(env);
 244         struct cl_object_conf   *subconf = &lti->lti_stripe_conf;
 245         struct lov_stripe_md    *lsm     = conf->u.coc_md->lsm;
 246         struct lu_fid           *ofid    = &lti->lti_fid;
 247         struct lov_layout_raid0 *r0      = &state->raid0;
 248
 249         ENTRY;
 250
 251         if (lsm->lsm_magic != LOV_MAGIC_V1 && lsm->lsm_magic != LOV_MAGIC_V3) {
 252                 dump_lsm(D_ERROR, lsm);
 253                 LASSERTF(0, "magic mismatch, expected %d/%d, actual %d.\n",
 254                          LOV_MAGIC_V1, LOV_MAGIC_V3, lsm->lsm_magic);
 255         }
 256
 257         LASSERT(lov->lo_lsm == NULL);
 258         lov->lo_lsm = lsm_addref(lsm);
 259         r0->lo_nr  = lsm->lsm_stripe_count;
 260         LASSERT(r0->lo_nr <= lov_targets_nr(dev));
 261
 262         lov->lo_layout_invalid = true;
 263
 264         OBD_ALLOC_LARGE(r0->lo_sub, r0->lo_nr * sizeof r0->lo_sub[0]);
 265         if (r0->lo_sub != NULL) {
 266                 int psz = 0;
 267
 268                 result = 0;
 269                 subconf->coc_inode = conf->coc_inode;
 270                 spin_lock_init(&r0->lo_sub_lock);
 271                 /*
 272                  * Create stripe cl_objects.
 273                  */
 274                 for (i = 0; i < r0->lo_nr && result == 0; ++i) {
 275                         struct cl_device *subdev;
 276                         struct lov_oinfo *oinfo = lsm->lsm_oinfo[i];
 277                         int ost_idx = oinfo->loi_ost_idx;
 278
 279                         if (lov_oinfo_is_dummy(oinfo))
 280                                 continue;
 281
 282                         result = ostid_to_fid(ofid, &oinfo->loi_oi,
 283                                               oinfo->loi_ost_idx);
 284                         if (result != 0)
 285                                 GOTO(out, result);
 286
 287                         subdev = lovsub2cl_dev(dev->ld_target[ost_idx]);
 288                         subconf->u.coc_oinfo = oinfo;
 289                         LASSERTF(subdev != NULL, "not init ost %d\n", ost_idx);
 290                         /* In the function below, .hs_keycmp resolves to
 291                          * lu_obj_hop_keycmp() */
 292                         /* coverity[overrun-buffer-val] */
 293                         stripe = lov_sub_find(env, subdev, ofid, subconf);
 294                         if (!IS_ERR(stripe)) {
 295                                 result = lov_init_sub(env, lov, stripe, r0, i);
 296                                 if (result == -EAGAIN) { /* try again */
 297                                         --i;
 298                                         result = 0;
 299                                         continue;
 300                                 }
 301                         } else {
 302                                 result = PTR_ERR(stripe);
 303                         }
 304
 305                         if (result == 0) {
 306                                 int sz = lov_page_slice_fixup(lov, stripe);
 307                                 LASSERT(ergo(psz > 0, psz == sz));
 308                                 psz = sz;
 309                         }
 310                 }
 311                 if (result == 0)
 312                         cl_object_header(&lov->lo_cl)->coh_page_bufsize += psz;
 313         } else
 314                 result = -ENOMEM;
 315 out:
 316         RETURN(result);
 317 }
 318
 319 static int lov_init_released(const struct lu_env *env,
 320                         struct lov_device *dev, struct lov_object *lov,
 321                         const struct cl_object_conf *conf,
 322                         union  lov_layout_state *state)
 323 {
 324         struct lov_stripe_md *lsm = conf->u.coc_md->lsm;
 325
 326         LASSERT(lsm != NULL);
 327         LASSERT(lsm_is_released(lsm));
 328         LASSERT(lov->lo_lsm == NULL);
 329
 330         lov->lo_lsm = lsm_addref(lsm);
 331         return 0;
 332 }
 333
 334 static int lov_delete_empty(const struct lu_env *env, struct lov_object *lov,
 335                             union lov_layout_state *state)
 336 {
 337         LASSERT(lov->lo_type == LLT_EMPTY || lov->lo_type == LLT_RELEASED);
 338
 339         lov_layout_wait(env, lov);
 340         return 0;
 341 }
 342
 343 static void lov_subobject_kill(const struct lu_env *env, struct lov_object *lov,
 344                                struct lovsub_object *los, int idx)
 345 {
 346         struct cl_object        *sub;
 347         struct lov_layout_raid0 *r0;
 348         struct lu_site          *site;
 349         struct lu_site_bkt_data *bkt;
 350         wait_queue_t          *waiter;
 351
 352         r0  = &lov->u.raid0;
 353         LASSERT(r0->lo_sub[idx] == los);
 354
 355         sub  = lovsub2cl(los);
 356         site = sub->co_lu.lo_dev->ld_site;
 357         bkt  = lu_site_bkt_from_fid(site, &sub->co_lu.lo_header->loh_fid);
 358
 359         cl_object_kill(env, sub);
 360         /* release a reference to the sub-object and ... */
 361         lu_object_ref_del(&sub->co_lu, "lov-parent", lov);
 362         cl_object_put(env, sub);
 363
 364         /* ... wait until it is actually destroyed---sub-object clears its
 365          * ->lo_sub[] slot in lovsub_object_fini() */
 366         if (r0->lo_sub[idx] == los) {
 367                 waiter = &lov_env_info(env)->lti_waiter;
 368                 init_waitqueue_entry_current(waiter);
 369                 add_wait_queue(&bkt->lsb_marche_funebre, waiter);
 370                 set_current_state(TASK_UNINTERRUPTIBLE);
 371                 while (1) {
 372                         /* this wait-queue is signaled at the end of
 373                          * lu_object_free(). */
 374                         set_current_state(TASK_UNINTERRUPTIBLE);
 375                         spin_lock(&r0->lo_sub_lock);
 376                         if (r0->lo_sub[idx] == los) {
 377                                 spin_unlock(&r0->lo_sub_lock);
 378                                 waitq_wait(waiter, TASK_UNINTERRUPTIBLE);
 379                         } else {
 380                                 spin_unlock(&r0->lo_sub_lock);
 381                                 set_current_state(TASK_RUNNING);
 382                                 break;
 383                         }
 384                 }
 385                 remove_wait_queue(&bkt->lsb_marche_funebre, waiter);
 386         }
 387         LASSERT(r0->lo_sub[idx] == NULL);
 388 }
 389
 390 static int lov_delete_raid0(const struct lu_env *env, struct lov_object *lov,
 391                             union lov_layout_state *state)
 392 {
 393         struct lov_layout_raid0 *r0 = &state->raid0;
 394         struct lov_stripe_md    *lsm = lov->lo_lsm;
 395         int i;
 396
 397         ENTRY;
 398
 399         dump_lsm(D_INODE, lsm);
 400
 401         lov_layout_wait(env, lov);
 402         if (r0->lo_sub != NULL) {
 403                 for (i = 0; i < r0->lo_nr; ++i) {
 404                         struct lovsub_object *los = r0->lo_sub[i];
 405
 406                         if (los != NULL) {
 407                                 cl_object_prune(env, &los->lso_cl);
 408                                 /*
 409                                  * If top-level object is to be evicted from
 410                                  * the cache, so are its sub-objects.
 411                                  */
 412                                 lov_subobject_kill(env, lov, los, i);
 413                         }
 414                 }
 415         }
 416         RETURN(0);
 417 }
 418
 419 static void lov_fini_empty(const struct lu_env *env, struct lov_object *lov,
 420                            union lov_layout_state *state)
 421 {
 422         LASSERT(lov->lo_type == LLT_EMPTY || lov->lo_type == LLT_RELEASED);
 423 }
 424
 425 static void lov_fini_raid0(const struct lu_env *env, struct lov_object *lov,
 426                            union lov_layout_state *state)
 427 {
 428         struct lov_layout_raid0 *r0 = &state->raid0;
 429         ENTRY;
 430
 431         if (r0->lo_sub != NULL) {
 432                 OBD_FREE_LARGE(r0->lo_sub, r0->lo_nr * sizeof r0->lo_sub[0]);
 433                 r0->lo_sub = NULL;
 434         }
 435
 436         dump_lsm(D_INODE, lov->lo_lsm);
 437         lov_free_memmd(&lov->lo_lsm);
 438
 439         EXIT;
 440 }
 441
 442 static void lov_fini_released(const struct lu_env *env, struct lov_object *lov,
 443                                 union lov_layout_state *state)
 444 {
 445         ENTRY;
 446         dump_lsm(D_INODE, lov->lo_lsm);
 447         lov_free_memmd(&lov->lo_lsm);
 448         EXIT;
 449 }
 450
 451 static int lov_print_empty(const struct lu_env *env, void *cookie,
 452                            lu_printer_t p, const struct lu_object *o)
 453 {
 454         (*p)(env, cookie, "empty %d\n", lu2lov(o)->lo_layout_invalid);
 455         return 0;
 456 }
 457
 458 static int lov_print_raid0(const struct lu_env *env, void *cookie,
 459                            lu_printer_t p, const struct lu_object *o)
 460 {
 461         struct lov_object       *lov = lu2lov(o);
 462         struct lov_layout_raid0 *r0  = lov_r0(lov);
 463         struct lov_stripe_md    *lsm = lov->lo_lsm;
 464         int                      i;
 465
 466         (*p)(env, cookie, "stripes: %d, %s, lsm{%p 0x%08X %d %u %u}:\n",
 467                 r0->lo_nr, lov->lo_layout_invalid ? "invalid" : "valid", lsm,
 468                 lsm->lsm_magic, atomic_read(&lsm->lsm_refc),
 469                 lsm->lsm_stripe_count, lsm->lsm_layout_gen);
 470         for (i = 0; i < r0->lo_nr; ++i) {
 471                 struct lu_object *sub;
 472
 473                 if (r0->lo_sub[i] != NULL) {
 474                         sub = lovsub2lu(r0->lo_sub[i]);
 475                         lu_object_print(env, cookie, p, sub);
 476                 } else {
 477                         (*p)(env, cookie, "sub %d absent\n", i);
 478                 }
 479         }
 480         return 0;
 481 }
 482
 483 static int lov_print_released(const struct lu_env *env, void *cookie,
 484                                 lu_printer_t p, const struct lu_object *o)
 485 {
 486         struct lov_object       *lov = lu2lov(o);
 487         struct lov_stripe_md    *lsm = lov->lo_lsm;
 488
 489         (*p)(env, cookie,
 490                 "released: %s, lsm{%p 0x%08X %d %u %u}:\n",
 491                 lov->lo_layout_invalid ? "invalid" : "valid", lsm,
 492                 lsm->lsm_magic, atomic_read(&lsm->lsm_refc),
 493                 lsm->lsm_stripe_count, lsm->lsm_layout_gen);
 494         return 0;
 495 }
 496
 497 /**
 498  * Implements cl_object_operations::coo_attr_get() method for an object
 499  * without stripes (LLT_EMPTY layout type).
 500  *
 501  * The only attributes this layer is authoritative in this case is
 502  * cl_attr::cat_blocks---it's 0.
 503  */
 504 static int lov_attr_get_empty(const struct lu_env *env, struct cl_object *obj,
 505                               struct cl_attr *attr)
 506 {
 507         attr->cat_blocks = 0;
 508         return 0;
 509 }
 510
 511 static int lov_attr_get_raid0(const struct lu_env *env, struct cl_object *obj,
 512                               struct cl_attr *attr)
 513 {
 514         struct lov_object       *lov = cl2lov(obj);
 515         struct lov_layout_raid0 *r0 = lov_r0(lov);
 516         struct cl_attr          *lov_attr = &r0->lo_attr;
 517         int                      result = 0;
 518
 519         ENTRY;
 520
 521         /* this is called w/o holding type guard mutex, so it must be inside
 522          * an on going IO otherwise lsm may be replaced.
 523          * LU-2117: it turns out there exists one exception. For mmaped files,
 524          * the lock of those files may be requested in the other file's IO
 525          * context, and this function is called in ccc_lock_state(), it will
 526          * hit this assertion.
 527          * Anyway, it's still okay to call attr_get w/o type guard as layout
 528          * can't go if locks exist. */
 529         /* LASSERT(atomic_read(&lsm->lsm_refc) > 1); */
 530
 531         if (!r0->lo_attr_valid) {
 532                 struct lov_stripe_md    *lsm = lov->lo_lsm;
 533                 struct ost_lvb          *lvb = &lov_env_info(env)->lti_lvb;
 534                 __u64                    kms = 0;
 535
 536                 memset(lvb, 0, sizeof(*lvb));
 537                 /* XXX: timestamps can be negative by sanity:test_39m,
 538                  * how can it be? */
 539                 lvb->lvb_atime = LLONG_MIN;
 540                 lvb->lvb_ctime = LLONG_MIN;
 541                 lvb->lvb_mtime = LLONG_MIN;
 542
 543                 /*
 544                  * XXX that should be replaced with a loop over sub-objects,
 545                  * doing cl_object_attr_get() on them. But for now, let's
 546                  * reuse old lov code.
 547                  */
 548
 549                 /*
 550                  * XXX take lsm spin-lock to keep lov_merge_lvb_kms()
 551                  * happy. It's not needed, because new code uses
 552                  * ->coh_attr_guard spin-lock to protect consistency of
 553                  * sub-object attributes.
 554                  */
 555                 lov_stripe_lock(lsm);
 556                 result = lov_merge_lvb_kms(lsm, lvb, &kms);
 557                 lov_stripe_unlock(lsm);
 558                 if (result == 0) {
 559                         cl_lvb2attr(lov_attr, lvb);
 560                         lov_attr->cat_kms = kms;
 561                         r0->lo_attr_valid = 1;
 562                 }
 563         }
 564         if (result == 0) { /* merge results */
 565                 attr->cat_blocks = lov_attr->cat_blocks;
 566                 attr->cat_size = lov_attr->cat_size;
 567                 attr->cat_kms = lov_attr->cat_kms;
 568                 if (attr->cat_atime < lov_attr->cat_atime)
 569                         attr->cat_atime = lov_attr->cat_atime;
 570                 if (attr->cat_ctime < lov_attr->cat_ctime)
 571                         attr->cat_ctime = lov_attr->cat_ctime;
 572                 if (attr->cat_mtime < lov_attr->cat_mtime)
 573                         attr->cat_mtime = lov_attr->cat_mtime;
 574         }
 575         RETURN(result);
 576 }
 577
 578 const static struct lov_layout_operations lov_dispatch[] = {
 579         [LLT_EMPTY] = {
 580                 .llo_init      = lov_init_empty,
 581                 .llo_delete    = lov_delete_empty,
 582                 .llo_fini      = lov_fini_empty,
 583                 .llo_install   = lov_install_empty,
 584                 .llo_print     = lov_print_empty,
 585                 .llo_page_init = lov_page_init_empty,
 586                 .llo_lock_init = lov_lock_init_empty,
 587                 .llo_io_init   = lov_io_init_empty,
 588                 .llo_getattr   = lov_attr_get_empty
 589         },
 590         [LLT_RAID0] = {
 591                 .llo_init      = lov_init_raid0,
 592                 .llo_delete    = lov_delete_raid0,
 593                 .llo_fini      = lov_fini_raid0,
 594                 .llo_install   = lov_install_raid0,
 595                 .llo_print     = lov_print_raid0,
 596                 .llo_page_init = lov_page_init_raid0,
 597                 .llo_lock_init = lov_lock_init_raid0,
 598                 .llo_io_init   = lov_io_init_raid0,
 599                 .llo_getattr   = lov_attr_get_raid0
 600         },
 601         [LLT_RELEASED] = {
 602                 .llo_init      = lov_init_released,
 603                 .llo_delete    = lov_delete_empty,
 604                 .llo_fini      = lov_fini_released,
 605                 .llo_install   = lov_install_empty,
 606                 .llo_print     = lov_print_released,
 607                 .llo_page_init = lov_page_init_empty,
 608                 .llo_lock_init = lov_lock_init_empty,
 609                 .llo_io_init   = lov_io_init_released,
 610                 .llo_getattr   = lov_attr_get_empty
 611         }
 612 };
 613
 614 /**
 615  * Performs a double-dispatch based on the layout type of an object.
 616  */
 617 #define LOV_2DISPATCH_NOLOCK(obj, op, ...)                              \
 618 ({                                                                      \
 619         struct lov_object                      *__obj = (obj);          \
 620         enum lov_layout_type                    __llt;                  \
 621                                                                         \
 622         __llt = __obj->lo_type;                                         \
 623         LASSERT(0 <= __llt && __llt < ARRAY_SIZE(lov_dispatch));        \
 624         lov_dispatch[__llt].op(__VA_ARGS__);                            \
 625 })
 626
 627 /**
 628  * Return lov_layout_type associated with a given lsm
 629  */
 630 static enum lov_layout_type lov_type(struct lov_stripe_md *lsm)
 631 {
 632         if (lsm == NULL)
 633                 return LLT_EMPTY;
 634         if (lsm_is_released(lsm))
 635                 return LLT_RELEASED;
 636         return LLT_RAID0;
 637 }
 638
 639 static inline void lov_conf_freeze(struct lov_object *lov)
 640 {
 641         if (lov->lo_owner != current)
 642                 down_read(&lov->lo_type_guard);
 643 }
 644
 645 static inline void lov_conf_thaw(struct lov_object *lov)
 646 {
 647         if (lov->lo_owner != current)
 648                 up_read(&lov->lo_type_guard);
 649 }
 650
 651 #define LOV_2DISPATCH_MAYLOCK(obj, op, lock, ...)                       \
 652 ({                                                                      \
 653         struct lov_object                      *__obj = (obj);          \
 654         int                                     __lock = !!(lock);      \
 655         typeof(lov_dispatch[0].op(__VA_ARGS__)) __result;               \
 656                                                                         \
 657         if (__lock)                                                     \
 658                 lov_conf_freeze(__obj);                                 \
 659         __result = LOV_2DISPATCH_NOLOCK(obj, op, __VA_ARGS__);          \
 660         if (__lock)                                                     \
 661                 lov_conf_thaw(__obj);                                   \
 662         __result;                                                       \
 663 })
 664
 665 /**
 666  * Performs a locked double-dispatch based on the layout type of an object.
 667  */
 668 #define LOV_2DISPATCH(obj, op, ...)                     \
 669         LOV_2DISPATCH_MAYLOCK(obj, op, 1, __VA_ARGS__)
 670
 671 #define LOV_2DISPATCH_VOID(obj, op, ...)                                \
 672 do {                                                                    \
 673         struct lov_object                      *__obj = (obj);          \
 674         enum lov_layout_type                    __llt;                  \
 675                                                                         \
 676         lov_conf_freeze(__obj);                                         \
 677         __llt = __obj->lo_type;                                         \
 678         LASSERT(0 <= __llt && __llt < ARRAY_SIZE(lov_dispatch));        \
 679         lov_dispatch[__llt].op(__VA_ARGS__);                            \
 680         lov_conf_thaw(__obj);                                           \
 681 } while (0)
 682
 683 static void lov_conf_lock(struct lov_object *lov)
 684 {
 685         LASSERT(lov->lo_owner != current);
 686         down_write(&lov->lo_type_guard);
 687         LASSERT(lov->lo_owner == NULL);
 688         lov->lo_owner = current;
 689 }
 690
 691 static void lov_conf_unlock(struct lov_object *lov)
 692 {
 693         lov->lo_owner = NULL;
 694         up_write(&lov->lo_type_guard);
 695 }
 696
 697 static int lov_layout_wait(const struct lu_env *env, struct lov_object *lov)
 698 {
 699         struct l_wait_info lwi = { 0 };
 700         ENTRY;
 701
 702         while (atomic_read(&lov->lo_active_ios) > 0) {
 703                 CDEBUG(D_INODE, "file:"DFID" wait for active IO, now: %d.\n",
 704                         PFID(lu_object_fid(lov2lu(lov))),
 705                         atomic_read(&lov->lo_active_ios));
 706
 707                 l_wait_event(lov->lo_waitq,
 708                              atomic_read(&lov->lo_active_ios) == 0, &lwi);
 709         }
 710         RETURN(0);
 711 }
 712
 713 static int lov_layout_change(const struct lu_env *unused,
 714                              struct lov_object *lov,
 715                              const struct cl_object_conf *conf)
 716 {
 717         int result;
 718         enum lov_layout_type llt = LLT_EMPTY;
 719         union lov_layout_state *state = &lov->u;
 720         const struct lov_layout_operations *old_ops;
 721         const struct lov_layout_operations *new_ops;
 722
 723         void *cookie;
 724         struct lu_env *env;
 725         int refcheck;
 726         ENTRY;
 727
 728         LASSERT(0 <= lov->lo_type && lov->lo_type < ARRAY_SIZE(lov_dispatch));
 729
 730         if (conf->u.coc_md != NULL)
 731                 llt = lov_type(conf->u.coc_md->lsm);
 732         LASSERT(0 <= llt && llt < ARRAY_SIZE(lov_dispatch));
 733
 734         cookie = cl_env_reenter();
 735         env = cl_env_get(&refcheck);
 736         if (IS_ERR(env)) {
 737                 cl_env_reexit(cookie);
 738                 RETURN(PTR_ERR(env));
 739         }
 740
 741         CDEBUG(D_INODE, DFID" from %s to %s\n",
 742                PFID(lu_object_fid(lov2lu(lov))),
 743                llt2str(lov->lo_type), llt2str(llt));
 744
 745         old_ops = &lov_dispatch[lov->lo_type];
 746         new_ops = &lov_dispatch[llt];
 747
 748         result = cl_object_prune(env, &lov->lo_cl);
 749         if (result != 0)
 750                 GOTO(out, result);
 751
 752         result = old_ops->llo_delete(env, lov, &lov->u);
 753         if (result == 0) {
 754                 old_ops->llo_fini(env, lov, &lov->u);
 755
 756                 LASSERT(atomic_read(&lov->lo_active_ios) == 0);
 757
 758                 lov->lo_type = LLT_EMPTY;
 759                 /* page bufsize fixup */
 760                 cl_object_header(&lov->lo_cl)->coh_page_bufsize -=
 761                         lov_page_slice_fixup(lov, NULL);
 762
 763                 result = new_ops->llo_init(env,
 764                                         lu2lov_dev(lov->lo_cl.co_lu.lo_dev),
 765                                         lov, conf, state);
 766                 if (result == 0) {
 767                         new_ops->llo_install(env, lov, state);
 768                         lov->lo_type = llt;
 769                 } else {
 770                         new_ops->llo_delete(env, lov, state);
 771                         new_ops->llo_fini(env, lov, state);
 772                         /* this file becomes an EMPTY file. */
 773                 }
 774         }
 775
 776 out:
 777         cl_env_put(env, &refcheck);
 778         cl_env_reexit(cookie);
 779         RETURN(result);
 780 }
 781
 782 /*****************************************************************************
 783  *
 784  * Lov object operations.
 785  *
 786  */
 787 int lov_object_init(const struct lu_env *env, struct lu_object *obj,
 788                     const struct lu_object_conf *conf)
 789 {
 790         struct lov_device            *dev   = lu2lov_dev(obj->lo_dev);
 791         struct lov_object            *lov   = lu2lov(obj);
 792         const struct cl_object_conf  *cconf = lu2cl_conf(conf);
 793         union  lov_layout_state      *set   = &lov->u;
 794         const struct lov_layout_operations *ops;
 795         int result;
 796
 797         ENTRY;
 798         init_rwsem(&lov->lo_type_guard);
 799         atomic_set(&lov->lo_active_ios, 0);
 800         init_waitqueue_head(&lov->lo_waitq);
 801
 802         cl_object_page_init(lu2cl(obj), sizeof(struct lov_page));
 803
 804         /* no locking is necessary, as object is being created */
 805         lov->lo_type = lov_type(cconf->u.coc_md->lsm);
 806         ops = &lov_dispatch[lov->lo_type];
 807         result = ops->llo_init(env, dev, lov, cconf, set);
 808         if (result == 0)
 809                 ops->llo_install(env, lov, set);
 810         RETURN(result);
 811 }
 812
 813 static int lov_conf_set(const struct lu_env *env, struct cl_object *obj,
 814                         const struct cl_object_conf *conf)
 815 {
 816         struct lov_stripe_md    *lsm = NULL;
 817         struct lov_object       *lov = cl2lov(obj);
 818         int                      result = 0;
 819         ENTRY;
 820
 821         lov_conf_lock(lov);
 822         if (conf->coc_opc == OBJECT_CONF_INVALIDATE) {
 823                 lov->lo_layout_invalid = true;
 824                 GOTO(out, result = 0);
 825         }
 826
 827         if (conf->coc_opc == OBJECT_CONF_WAIT) {
 828                 if (lov->lo_layout_invalid &&
 829                     atomic_read(&lov->lo_active_ios) > 0) {
 830                         lov_conf_unlock(lov);
 831                         result = lov_layout_wait(env, lov);
 832                         lov_conf_lock(lov);
 833                 }
 834                 GOTO(out, result);
 835         }
 836
 837         LASSERT(conf->coc_opc == OBJECT_CONF_SET);
 838
 839         if (conf->u.coc_md != NULL)
 840                 lsm = conf->u.coc_md->lsm;
 841         if ((lsm == NULL && lov->lo_lsm == NULL) ||
 842             ((lsm != NULL && lov->lo_lsm != NULL) &&
 843              (lov->lo_lsm->lsm_layout_gen == lsm->lsm_layout_gen) &&
 844              (lov->lo_lsm->lsm_pattern == lsm->lsm_pattern))) {
 845                 /* same version of layout */
 846                 lov->lo_layout_invalid = false;
 847                 GOTO(out, result = 0);
 848         }
 849
 850         /* will change layout - check if there still exists active IO. */
 851         if (atomic_read(&lov->lo_active_ios) > 0) {
 852                 lov->lo_layout_invalid = true;
 853                 GOTO(out, result = -EBUSY);
 854         }
 855
 856         result = lov_layout_change(env, lov, conf);
 857         lov->lo_layout_invalid = result != 0;
 858         EXIT;
 859
 860 out:
 861         lov_conf_unlock(lov);
 862         CDEBUG(D_INODE, DFID" lo_layout_invalid=%d\n",
 863                PFID(lu_object_fid(lov2lu(lov))), lov->lo_layout_invalid);
 864         RETURN(result);
 865 }
 866
 867 static void lov_object_delete(const struct lu_env *env, struct lu_object *obj)
 868 {
 869         struct lov_object *lov = lu2lov(obj);
 870
 871         ENTRY;
 872         LOV_2DISPATCH_VOID(lov, llo_delete, env, lov, &lov->u);
 873         EXIT;
 874 }
 875
 876 static void lov_object_free(const struct lu_env *env, struct lu_object *obj)
 877 {
 878         struct lov_object *lov = lu2lov(obj);
 879
 880         ENTRY;
 881         LOV_2DISPATCH_VOID(lov, llo_fini, env, lov, &lov->u);
 882         lu_object_fini(obj);
 883         OBD_SLAB_FREE_PTR(lov, lov_object_kmem);
 884         EXIT;
 885 }
 886
 887 static int lov_object_print(const struct lu_env *env, void *cookie,
 888                             lu_printer_t p, const struct lu_object *o)
 889 {
 890         return LOV_2DISPATCH_NOLOCK(lu2lov(o), llo_print, env, cookie, p, o);
 891 }
 892
 893 int lov_page_init(const struct lu_env *env, struct cl_object *obj,
 894                   struct cl_page *page, pgoff_t index)
 895 {
 896         return LOV_2DISPATCH_NOLOCK(cl2lov(obj), llo_page_init, env, obj, page,
 897                                     index);
 898 }
 899
 900 /**
 901  * Implements cl_object_operations::clo_io_init() method for lov
 902  * layer. Dispatches to the appropriate layout io initialization method.
 903  */
 904 int lov_io_init(const struct lu_env *env, struct cl_object *obj,
 905                 struct cl_io *io)
 906 {
 907         CL_IO_SLICE_CLEAN(lov_env_io(env), lis_cl);
 908         return LOV_2DISPATCH_MAYLOCK(cl2lov(obj), llo_io_init,
 909                                      !io->ci_ignore_layout, env, obj, io);
 910 }
 911
 912 /**
 913  * An implementation of cl_object_operations::clo_attr_get() method for lov
 914  * layer. For raid0 layout this collects and merges attributes of all
 915  * sub-objects.
 916  */
 917 static int lov_attr_get(const struct lu_env *env, struct cl_object *obj,
 918                         struct cl_attr *attr)
 919 {
 920         /* do not take lock, as this function is called under a
 921          * spin-lock. Layout is protected from changing by ongoing IO. */
 922         return LOV_2DISPATCH_NOLOCK(cl2lov(obj), llo_getattr, env, obj, attr);
 923 }
 924
 925 static int lov_attr_update(const struct lu_env *env, struct cl_object *obj,
 926                            const struct cl_attr *attr, unsigned valid)
 927 {
 928         /*
 929          * No dispatch is required here, as no layout implements this.
 930          */
 931         return 0;
 932 }
 933
 934 int lov_lock_init(const struct lu_env *env, struct cl_object *obj,
 935                   struct cl_lock *lock, const struct cl_io *io)
 936 {
 937         /* No need to lock because we've taken one refcount of layout.  */
 938         return LOV_2DISPATCH_NOLOCK(cl2lov(obj), llo_lock_init, env, obj, lock,
 939                                     io);
 940 }
 941
 942 static int lov_object_getstripe(const struct lu_env *env, struct cl_object *obj,
 943                                 struct lov_user_md __user *lum)
 944 {
 945         struct lov_object       *lov = cl2lov(obj);
 946         struct lov_stripe_md    *lsm;
 947         int                     rc = 0;
 948         ENTRY;
 949
 950         lsm = lov_lsm_addref(lov);
 951         if (lsm == NULL)
 952                 RETURN(-ENODATA);
 953
 954         rc = lov_getstripe(cl2lov(obj), lsm, lum);
 955         lov_lsm_put(obj, lsm);
 956         RETURN(rc);
 957 }
 958
 959 static const struct cl_object_operations lov_ops = {
 960         .coo_page_init    = lov_page_init,
 961         .coo_lock_init    = lov_lock_init,
 962         .coo_io_init      = lov_io_init,
 963         .coo_attr_get     = lov_attr_get,
 964         .coo_attr_update  = lov_attr_update,
 965         .coo_conf_set     = lov_conf_set,
 966         .coo_getstripe    = lov_object_getstripe
 967 };
 968
 969 static const struct lu_object_operations lov_lu_obj_ops = {
 970         .loo_object_init      = lov_object_init,
 971         .loo_object_delete    = lov_object_delete,
 972         .loo_object_release   = NULL,
 973         .loo_object_free      = lov_object_free,
 974         .loo_object_print     = lov_object_print,
 975         .loo_object_invariant = NULL
 976 };
 977
 978 struct lu_object *lov_object_alloc(const struct lu_env *env,
 979                                    const struct lu_object_header *unused,
 980                                    struct lu_device *dev)
 981 {
 982         struct lov_object *lov;
 983         struct lu_object  *obj;
 984
 985         ENTRY;
 986         OBD_SLAB_ALLOC_PTR_GFP(lov, lov_object_kmem, GFP_NOFS);
 987         if (lov != NULL) {
 988                 obj = lov2lu(lov);
 989                 lu_object_init(obj, NULL, dev);
 990                 lov->lo_cl.co_ops = &lov_ops;
 991                 lov->lo_type = -1; /* invalid, to catch uninitialized type */
 992                 /*
 993                  * object io operation vector (cl_object::co_iop) is installed
 994                  * later in lov_object_init(), as different vectors are used
 995                  * for object with different layouts.
 996                  */
 997                 obj->lo_ops = &lov_lu_obj_ops;
 998         } else
 999                 obj = NULL;
1000         RETURN(obj);
1001 }
1002
1003 struct lov_stripe_md *lov_lsm_addref(struct lov_object *lov)
1004 {
1005         struct lov_stripe_md *lsm = NULL;
1006
1007         lov_conf_freeze(lov);
1008         if (lov->lo_lsm != NULL) {
1009                 lsm = lsm_addref(lov->lo_lsm);
1010                 CDEBUG(D_INODE, "lsm %p addref %d/%d by %p.\n",
1011                         lsm, atomic_read(&lsm->lsm_refc),
1012                         lov->lo_layout_invalid, current);
1013         }
1014         lov_conf_thaw(lov);
1015         return lsm;
1016 }
1017
1018 int lov_read_and_clear_async_rc(struct cl_object *clob)
1019 {
1020         struct lu_object *luobj;
1021         int rc = 0;
1022         ENTRY;
1023
1024         luobj = lu_object_locate(&cl_object_header(clob)->coh_lu,
1025                                  &lov_device_type);
1026         if (luobj != NULL) {
1027                 struct lov_object *lov = lu2lov(luobj);
1028
1029                 lov_conf_freeze(lov);
1030                 switch (lov->lo_type) {
1031                 case LLT_RAID0: {
1032                         struct lov_stripe_md *lsm;
1033                         int i;
1034
1035                         lsm = lov->lo_lsm;
1036                         LASSERT(lsm != NULL);
1037                         for (i = 0; i < lsm->lsm_stripe_count; i++) {
1038                                 struct lov_oinfo *loi = lsm->lsm_oinfo[i];
1039
1040                                 if (lov_oinfo_is_dummy(loi))
1041                                         continue;
1042
1043                                 if (loi->loi_ar.ar_rc && !rc)
1044                                         rc = loi->loi_ar.ar_rc;
1045                                 loi->loi_ar.ar_rc = 0;
1046                         }
1047                 }
1048                 case LLT_RELEASED:
1049                 case LLT_EMPTY:
1050                         break;
1051                 default:
1052                         LBUG();
1053                 }
1054                 lov_conf_thaw(lov);
1055         }
1056         RETURN(rc);
1057 }
1058 EXPORT_SYMBOL(lov_read_and_clear_async_rc);
1059
1060 /** @} lov */