lustre/lod/lod_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,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License version 2 for more details.  A copy is
  14  * included in the COPYING file that accompanied this code.
  15
  16  * You should have received a copy of the GNU General Public License
  17  * along with this program; if not, write to the Free Software
  18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright  2009 Sun Microsystems, Inc. All rights reserved
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2013, Intel Corporation.
  27  */
  28 /*
  29  * lustre/lod/lod_object.c
  30  *
  31  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  32  */
  33
  34 #define DEBUG_SUBSYSTEM S_MDS
  35
  36 #include <obd.h>
  37 #include <obd_class.h>
  38 #include <lustre_ver.h>
  39 #include <obd_support.h>
  40 #include <lprocfs_status.h>
  41
  42 #include <lustre_fid.h>
  43 #include <lustre_param.h>
  44 #include <lustre_fid.h>
  45 #include <obd_lov.h>
  46
  47 #include "lod_internal.h"
  48
  49 extern struct kmem_cache *lod_object_kmem;
  50 static const struct dt_body_operations lod_body_lnk_ops;
  51
  52 static int lod_index_lookup(const struct lu_env *env, struct dt_object *dt,
  53                             struct dt_rec *rec, const struct dt_key *key,
  54                             struct lustre_capa *capa)
  55 {
  56         struct dt_object *next = dt_object_child(dt);
  57         return next->do_index_ops->dio_lookup(env, next, rec, key, capa);
  58 }
  59
  60 static int lod_declare_index_insert(const struct lu_env *env,
  61                                     struct dt_object *dt,
  62                                     const struct dt_rec *rec,
  63                                     const struct dt_key *key,
  64                                     struct thandle *handle)
  65 {
  66         return dt_declare_insert(env, dt_object_child(dt), rec, key, handle);
  67 }
  68
  69 static int lod_index_insert(const struct lu_env *env,
  70                             struct dt_object *dt,
  71                             const struct dt_rec *rec,
  72                             const struct dt_key *key,
  73                             struct thandle *th,
  74                             struct lustre_capa *capa,
  75                             int ign)
  76 {
  77         return dt_insert(env, dt_object_child(dt), rec, key, th, capa, ign);
  78 }
  79
  80 static int lod_declare_index_delete(const struct lu_env *env,
  81                                     struct dt_object *dt,
  82                                     const struct dt_key *key,
  83                                     struct thandle *th)
  84 {
  85         return dt_declare_delete(env, dt_object_child(dt), key, th);
  86 }
  87
  88 static int lod_index_delete(const struct lu_env *env,
  89                             struct dt_object *dt,
  90                             const struct dt_key *key,
  91                             struct thandle *th,
  92                             struct lustre_capa *capa)
  93 {
  94         return dt_delete(env, dt_object_child(dt), key, th, capa);
  95 }
  96
  97 static struct dt_it *lod_it_init(const struct lu_env *env,
  98                                  struct dt_object *dt, __u32 attr,
  99                                  struct lustre_capa *capa)
 100 {
 101         struct dt_object        *next = dt_object_child(dt);
 102         struct lod_it           *it = &lod_env_info(env)->lti_it;
 103         struct dt_it            *it_next;
 104
 105
 106         it_next = next->do_index_ops->dio_it.init(env, next, attr, capa);
 107         if (IS_ERR(it_next))
 108                 return it_next;
 109
 110         /* currently we do not use more than one iterator per thread
 111          * so we store it in thread info. if at some point we need
 112          * more active iterators in a single thread, we can allocate
 113          * additional ones */
 114         LASSERT(it->lit_obj == NULL);
 115
 116         it->lit_it = it_next;
 117         it->lit_obj = next;
 118
 119         return (struct dt_it *)it;
 120 }
 121
 122 #define LOD_CHECK_IT(env, it)                                   \
 123 {                                                               \
 124         LASSERT((it)->lit_obj != NULL);                         \
 125         LASSERT((it)->lit_it != NULL);                          \
 126 } while(0)
 127
 128 void lod_it_fini(const struct lu_env *env, struct dt_it *di)
 129 {
 130         struct lod_it *it = (struct lod_it *)di;
 131
 132         LOD_CHECK_IT(env, it);
 133         it->lit_obj->do_index_ops->dio_it.fini(env, it->lit_it);
 134
 135         /* the iterator not in use any more */
 136         it->lit_obj = NULL;
 137         it->lit_it = NULL;
 138 }
 139
 140 int lod_it_get(const struct lu_env *env, struct dt_it *di,
 141                const struct dt_key *key)
 142 {
 143         const struct lod_it *it = (const struct lod_it *)di;
 144
 145         LOD_CHECK_IT(env, it);
 146         return it->lit_obj->do_index_ops->dio_it.get(env, it->lit_it, key);
 147 }
 148
 149 void lod_it_put(const struct lu_env *env, struct dt_it *di)
 150 {
 151         struct lod_it *it = (struct lod_it *)di;
 152
 153         LOD_CHECK_IT(env, it);
 154         return it->lit_obj->do_index_ops->dio_it.put(env, it->lit_it);
 155 }
 156
 157 int lod_it_next(const struct lu_env *env, struct dt_it *di)
 158 {
 159         struct lod_it *it = (struct lod_it *)di;
 160
 161         LOD_CHECK_IT(env, it);
 162         return it->lit_obj->do_index_ops->dio_it.next(env, it->lit_it);
 163 }
 164
 165 struct dt_key *lod_it_key(const struct lu_env *env, const struct dt_it *di)
 166 {
 167         const struct lod_it *it = (const struct lod_it *)di;
 168
 169         LOD_CHECK_IT(env, it);
 170         return it->lit_obj->do_index_ops->dio_it.key(env, it->lit_it);
 171 }
 172
 173 int lod_it_key_size(const struct lu_env *env, const struct dt_it *di)
 174 {
 175         struct lod_it *it = (struct lod_it *)di;
 176
 177         LOD_CHECK_IT(env, it);
 178         return it->lit_obj->do_index_ops->dio_it.key_size(env, it->lit_it);
 179 }
 180
 181 int lod_it_rec(const struct lu_env *env, const struct dt_it *di,
 182                struct dt_rec *rec, __u32 attr)
 183 {
 184         const struct lod_it *it = (const struct lod_it *)di;
 185
 186         LOD_CHECK_IT(env, it);
 187         return it->lit_obj->do_index_ops->dio_it.rec(env, it->lit_it, rec, attr);
 188 }
 189
 190 __u64 lod_it_store(const struct lu_env *env, const struct dt_it *di)
 191 {
 192         const struct lod_it *it = (const struct lod_it *)di;
 193
 194         LOD_CHECK_IT(env, it);
 195         return it->lit_obj->do_index_ops->dio_it.store(env, it->lit_it);
 196 }
 197
 198 int lod_it_load(const struct lu_env *env, const struct dt_it *di, __u64 hash)
 199 {
 200         const struct lod_it *it = (const struct lod_it *)di;
 201
 202         LOD_CHECK_IT(env, it);
 203         return it->lit_obj->do_index_ops->dio_it.load(env, it->lit_it, hash);
 204 }
 205
 206 int lod_it_key_rec(const struct lu_env *env, const struct dt_it *di,
 207                    void* key_rec)
 208 {
 209         const struct lod_it *it = (const struct lod_it *)di;
 210
 211         LOD_CHECK_IT(env, it);
 212         return it->lit_obj->do_index_ops->dio_it.key_rec(env, it->lit_it, key_rec);
 213 }
 214
 215 static struct dt_index_operations lod_index_ops = {
 216         .dio_lookup             = lod_index_lookup,
 217         .dio_declare_insert     = lod_declare_index_insert,
 218         .dio_insert             = lod_index_insert,
 219         .dio_declare_delete     = lod_declare_index_delete,
 220         .dio_delete             = lod_index_delete,
 221         .dio_it = {
 222                 .init           = lod_it_init,
 223                 .fini           = lod_it_fini,
 224                 .get            = lod_it_get,
 225                 .put            = lod_it_put,
 226                 .next           = lod_it_next,
 227                 .key            = lod_it_key,
 228                 .key_size       = lod_it_key_size,
 229                 .rec            = lod_it_rec,
 230                 .store          = lod_it_store,
 231                 .load           = lod_it_load,
 232                 .key_rec        = lod_it_key_rec,
 233         }
 234 };
 235
 236 static void lod_object_read_lock(const struct lu_env *env,
 237                                  struct dt_object *dt, unsigned role)
 238 {
 239         dt_read_lock(env, dt_object_child(dt), role);
 240 }
 241
 242 static void lod_object_write_lock(const struct lu_env *env,
 243                                   struct dt_object *dt, unsigned role)
 244 {
 245         dt_write_lock(env, dt_object_child(dt), role);
 246 }
 247
 248 static void lod_object_read_unlock(const struct lu_env *env,
 249                                    struct dt_object *dt)
 250 {
 251         dt_read_unlock(env, dt_object_child(dt));
 252 }
 253
 254 static void lod_object_write_unlock(const struct lu_env *env,
 255                                     struct dt_object *dt)
 256 {
 257         dt_write_unlock(env, dt_object_child(dt));
 258 }
 259
 260 static int lod_object_write_locked(const struct lu_env *env,
 261                                    struct dt_object *dt)
 262 {
 263         return dt_write_locked(env, dt_object_child(dt));
 264 }
 265
 266 static int lod_attr_get(const struct lu_env *env,
 267                         struct dt_object *dt,
 268                         struct lu_attr *attr,
 269                         struct lustre_capa *capa)
 270 {
 271         return dt_attr_get(env, dt_object_child(dt), attr, capa);
 272 }
 273
 274 static int lod_declare_attr_set(const struct lu_env *env,
 275                                 struct dt_object *dt,
 276                                 const struct lu_attr *attr,
 277                                 struct thandle *handle)
 278 {
 279         struct dt_object  *next = dt_object_child(dt);
 280         struct lod_object *lo = lod_dt_obj(dt);
 281         int                rc, i;
 282         ENTRY;
 283
 284         /*
 285          * declare setattr on the local object
 286          */
 287         rc = dt_declare_attr_set(env, next, attr, handle);
 288         if (rc)
 289                 RETURN(rc);
 290
 291         /* osp_declare_attr_set() ignores all attributes other than
 292          * UID, GID, and size, and osp_attr_set() ignores all but UID
 293          * and GID.  Declaration of size attr setting happens through
 294          * lod_declare_init_size(), and not through this function.
 295          * Therefore we need not load striping unless ownership is
 296          * changing.  This should save memory and (we hope) speed up
 297          * rename(). */
 298         if (!(attr->la_valid & (LA_UID | LA_GID)))
 299                 RETURN(rc);
 300
 301         /*
 302          * load striping information, notice we don't do this when object
 303          * is being initialized as we don't need this information till
 304          * few specific cases like destroy, chown
 305          */
 306         rc = lod_load_striping(env, lo);
 307         if (rc)
 308                 RETURN(rc);
 309
 310         /*
 311          * if object is striped declare changes on the stripes
 312          */
 313         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
 314         for (i = 0; i < lo->ldo_stripenr; i++) {
 315                 LASSERT(lo->ldo_stripe[i]);
 316                 rc = dt_declare_attr_set(env, lo->ldo_stripe[i], attr, handle);
 317                 if (rc) {
 318                         CERROR("failed declaration: %d\n", rc);
 319                         break;
 320                 }
 321         }
 322
 323         RETURN(rc);
 324 }
 325
 326 static int lod_attr_set(const struct lu_env *env,
 327                         struct dt_object *dt,
 328                         const struct lu_attr *attr,
 329                         struct thandle *handle,
 330                         struct lustre_capa *capa)
 331 {
 332         struct dt_object  *next = dt_object_child(dt);
 333         struct lod_object *lo = lod_dt_obj(dt);
 334         int                rc, i;
 335         ENTRY;
 336
 337         /*
 338          * apply changes to the local object
 339          */
 340         rc = dt_attr_set(env, next, attr, handle, capa);
 341         if (rc)
 342                 RETURN(rc);
 343
 344         if (!(attr->la_valid & (LA_UID | LA_GID)))
 345                 RETURN(rc);
 346
 347         /*
 348          * if object is striped, apply changes to all the stripes
 349          */
 350         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
 351         for (i = 0; i < lo->ldo_stripenr; i++) {
 352                 LASSERT(lo->ldo_stripe[i]);
 353                 rc = dt_attr_set(env, lo->ldo_stripe[i], attr, handle, capa);
 354                 if (rc) {
 355                         CERROR("failed declaration: %d\n", rc);
 356                         break;
 357                 }
 358         }
 359
 360         RETURN(rc);
 361 }
 362
 363 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
 364                          struct lu_buf *buf, const char *name,
 365                          struct lustre_capa *capa)
 366 {
 367         struct lod_thread_info  *info = lod_env_info(env);
 368         struct lod_device       *dev = lu2lod_dev(dt->do_lu.lo_dev);
 369         int                      rc, is_root;
 370         ENTRY;
 371
 372         rc = dt_xattr_get(env, dt_object_child(dt), buf, name, capa);
 373         if (rc != -ENODATA || !S_ISDIR(dt->do_lu.lo_header->loh_attr & S_IFMT))
 374                 RETURN(rc);
 375
 376         /*
 377          * lod returns default striping on the real root of the device
 378          * this is like the root stores default striping for the whole
 379          * filesystem. historically we've been using a different approach
 380          * and store it in the config.
 381          */
 382         dt_root_get(env, dev->lod_child, &info->lti_fid);
 383         is_root = lu_fid_eq(&info->lti_fid, lu_object_fid(&dt->do_lu));
 384
 385         if (is_root && strcmp(XATTR_NAME_LOV, name) == 0) {
 386                 struct lov_user_md *lum = buf->lb_buf;
 387                 struct lov_desc    *desc = &dev->lod_desc;
 388
 389                 if (buf->lb_buf == NULL) {
 390                         rc = sizeof(struct lov_user_md_v1);
 391                 } else if (buf->lb_len >= sizeof(struct lov_user_md_v1)) {
 392                         lum->lmm_magic = LOV_USER_MAGIC_V1;
 393                         lmm_oi_set_seq(&lum->lmm_oi, FID_SEQ_LOV_DEFAULT);
 394                         lum->lmm_pattern = desc->ld_pattern;
 395                         lum->lmm_stripe_size = desc->ld_default_stripe_size;
 396                         lum->lmm_stripe_count = desc->ld_default_stripe_count;
 397                         lum->lmm_stripe_offset = desc->ld_default_stripe_offset;
 398                         rc = sizeof(struct lov_user_md_v1);
 399                 } else {
 400                         rc = -ERANGE;
 401                 }
 402         }
 403
 404         RETURN(rc);
 405 }
 406
 407 /*
 408  * LOV xattr is a storage for striping, and LOD owns this xattr.
 409  * but LOD allows others to control striping to some extent
 410  * - to reset strping
 411  * - to set new defined striping
 412  * - to set new semi-defined striping
 413  *   - number of stripes is defined
 414  *   - number of stripes + osts are defined
 415  *   - ??
 416  */
 417 static int lod_declare_xattr_set(const struct lu_env *env,
 418                                  struct dt_object *dt,
 419                                  const struct lu_buf *buf,
 420                                  const char *name, int fl,
 421                                  struct thandle *th)
 422 {
 423         struct dt_object *next = dt_object_child(dt);
 424         struct lu_attr   *attr = &lod_env_info(env)->lti_attr;
 425         __u32             mode;
 426         int               rc;
 427         ENTRY;
 428
 429         /*
 430          * allow to declare predefined striping on a new (!mode) object
 431          * which is supposed to be replay of regular file creation
 432          * (when LOV setting is declared)
 433          * LU_XATTR_REPLACE is set to indicate a layout swap
 434          */
 435         mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
 436         if ((S_ISREG(mode) || !mode) && !strcmp(name, XATTR_NAME_LOV) &&
 437              !(fl & LU_XATTR_REPLACE)) {
 438                 /*
 439                  * this is a request to manipulate object's striping
 440                  */
 441                 if (dt_object_exists(dt)) {
 442                         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
 443                         if (rc)
 444                                 RETURN(rc);
 445                 } else {
 446                         memset(attr, 0, sizeof(*attr));
 447                         attr->la_valid = LA_TYPE | LA_MODE;
 448                         attr->la_mode = S_IFREG;
 449                 }
 450                 rc = lod_declare_striped_object(env, dt, attr, buf, th);
 451                 if (rc)
 452                         RETURN(rc);
 453         }
 454
 455         rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
 456
 457         RETURN(rc);
 458 }
 459
 460 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
 461                                     struct dt_object *dt,
 462                                     const struct lu_buf *buf,
 463                                     const char *name, int fl,
 464                                     struct thandle *th,
 465                                     struct lustre_capa *capa)
 466 {
 467         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
 468         struct dt_object        *next = dt_object_child(dt);
 469         struct lod_object       *l = lod_dt_obj(dt);
 470         struct lov_user_md_v1   *lum;
 471         struct lov_user_md_v3   *v3 = NULL;
 472         int                      rc;
 473         ENTRY;
 474
 475         LASSERT(l->ldo_stripe == NULL);
 476         l->ldo_striping_cached = 0;
 477         l->ldo_def_striping_set = 0;
 478         lod_object_set_pool(l, NULL);
 479         l->ldo_def_stripe_size = 0;
 480         l->ldo_def_stripenr = 0;
 481
 482         LASSERT(buf);
 483         LASSERT(buf->lb_buf);
 484         lum = buf->lb_buf;
 485
 486         rc = lod_verify_striping(d, buf, 0);
 487         if (rc)
 488                 RETURN(rc);
 489
 490         if (lum->lmm_magic == LOV_USER_MAGIC_V3)
 491                 v3 = buf->lb_buf;
 492
 493         /* if { size, offset, count } = { 0, -1, 0 } and no pool
 494          * (i.e. all default values specified) then delete default
 495          * striping from dir. */
 496         CDEBUG(D_OTHER,
 497                 "set default striping: sz %u # %u offset %d %s %s\n",
 498                 (unsigned)lum->lmm_stripe_size,
 499                 (unsigned)lum->lmm_stripe_count,
 500                 (int)lum->lmm_stripe_offset,
 501                 v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
 502
 503         if (LOVEA_DELETE_VALUES((lum->lmm_stripe_size),
 504                                 (lum->lmm_stripe_count),
 505                                 (lum->lmm_stripe_offset)) &&
 506                         lum->lmm_magic == LOV_USER_MAGIC_V1) {
 507                 rc = dt_xattr_del(env, next, name, th, capa);
 508                 if (rc == -ENODATA)
 509                         rc = 0;
 510         } else {
 511                 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
 512         }
 513
 514         RETURN(rc);
 515 }
 516
 517 static int lod_xattr_set(const struct lu_env *env,
 518                          struct dt_object *dt, const struct lu_buf *buf,
 519                          const char *name, int fl, struct thandle *th,
 520                          struct lustre_capa *capa)
 521 {
 522         struct dt_object        *next = dt_object_child(dt);
 523         __u32                    attr;
 524         int                      rc;
 525         ENTRY;
 526
 527         attr = dt->do_lu.lo_header->loh_attr & S_IFMT;
 528         if (S_ISDIR(attr)) {
 529                 if (strncmp(name, XATTR_NAME_LOV, strlen(XATTR_NAME_LOV)) == 0)
 530                         rc = lod_xattr_set_lov_on_dir(env, dt, buf, name,
 531                                                       fl, th, capa);
 532                 else
 533                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
 534
 535         } else if (S_ISREG(attr) && !strcmp(name, XATTR_NAME_LOV)) {
 536                 /* in case of lov EA swap, just set it
 537                  * if not, it is a replay so check striping match what we
 538                  * already have during req replay, declare_xattr_set()
 539                  * defines striping, then create() does the work
 540                 */
 541                 if (fl & LU_XATTR_REPLACE) {
 542                         /* free stripes, then update disk */
 543                         lod_object_free_striping(env, lod_dt_obj(dt));
 544                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
 545                 } else {
 546                         rc = lod_striping_create(env, dt, NULL, NULL, th);
 547                 }
 548                 RETURN(rc);
 549         } else {
 550                 /*
 551                  * behave transparantly for all other EAs
 552                  */
 553                 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
 554         }
 555
 556         RETURN(rc);
 557 }
 558
 559 static int lod_declare_xattr_del(const struct lu_env *env,
 560                                  struct dt_object *dt, const char *name,
 561                                  struct thandle *th)
 562 {
 563         return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
 564 }
 565
 566 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
 567                          const char *name, struct thandle *th,
 568                          struct lustre_capa *capa)
 569 {
 570         if (!strcmp(name, XATTR_NAME_LOV))
 571                 lod_object_free_striping(env, lod_dt_obj(dt));
 572         return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
 573 }
 574
 575 static int lod_xattr_list(const struct lu_env *env,
 576                           struct dt_object *dt, struct lu_buf *buf,
 577                           struct lustre_capa *capa)
 578 {
 579         return dt_xattr_list(env, dt_object_child(dt), buf, capa);
 580 }
 581
 582 int lod_object_set_pool(struct lod_object *o, char *pool)
 583 {
 584         int len;
 585
 586         if (o->ldo_pool) {
 587                 len = strlen(o->ldo_pool);
 588                 OBD_FREE(o->ldo_pool, len + 1);
 589                 o->ldo_pool = NULL;
 590         }
 591         if (pool) {
 592                 len = strlen(pool);
 593                 OBD_ALLOC(o->ldo_pool, len + 1);
 594                 if (o->ldo_pool == NULL)
 595                         return -ENOMEM;
 596                 strcpy(o->ldo_pool, pool);
 597         }
 598         return 0;
 599 }
 600
 601 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
 602 {
 603         return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
 604 }
 605
 606 static int lod_cache_parent_striping(const struct lu_env *env,
 607                                      struct lod_object *lp)
 608 {
 609         struct lov_user_md_v1   *v1 = NULL;
 610         struct lov_user_md_v3   *v3 = NULL;
 611         int                      rc;
 612         ENTRY;
 613
 614         /* dt_ah_init() is called from MDD without parent being write locked
 615          * lock it here */
 616         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
 617         if (lp->ldo_striping_cached)
 618                 GOTO(unlock, rc = 0);
 619
 620         rc = lod_get_lov_ea(env, lp);
 621         if (rc < 0)
 622                 GOTO(unlock, rc);
 623
 624         if (rc < sizeof(struct lov_user_md)) {
 625                 /* don't lookup for non-existing or invalid striping */
 626                 lp->ldo_def_striping_set = 0;
 627                 lp->ldo_striping_cached = 1;
 628                 lp->ldo_def_stripe_size = 0;
 629                 lp->ldo_def_stripenr = 0;
 630                 lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1);
 631                 GOTO(unlock, rc = 0);
 632         }
 633
 634         v1 = (struct lov_user_md_v1 *)lod_env_info(env)->lti_ea_store;
 635         if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1))
 636                 lustre_swab_lov_user_md_v1(v1);
 637         else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3))
 638                 lustre_swab_lov_user_md_v3(v3);
 639
 640         if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1)
 641                 GOTO(unlock, rc = 0);
 642
 643         if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
 644                 GOTO(unlock, rc = 0);
 645
 646         lp->ldo_def_stripenr = v1->lmm_stripe_count;
 647         lp->ldo_def_stripe_size = v1->lmm_stripe_size;
 648         lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
 649         lp->ldo_striping_cached = 1;
 650         lp->ldo_def_striping_set = 1;
 651
 652         if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
 653                 /* XXX: sanity check here */
 654                 v3 = (struct lov_user_md_v3 *) v1;
 655                 if (v3->lmm_pool_name[0])
 656                         lod_object_set_pool(lp, v3->lmm_pool_name);
 657         }
 658
 659         CDEBUG(D_OTHER, "def. striping: # %d, sz %d, off %d %s%s on "DFID"\n",
 660                lp->ldo_def_stripenr, lp->ldo_def_stripe_size,
 661                lp->ldo_def_stripe_offset, v3 ? "from " : "",
 662                v3 ? lp->ldo_pool : "", PFID(lu_object_fid(&lp->ldo_obj.do_lu)));
 663
 664         EXIT;
 665 unlock:
 666         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
 667         return rc;
 668 }
 669
 670 /**
 671  * used to transfer default striping data to the object being created
 672  */
 673 static void lod_ah_init(const struct lu_env *env,
 674                         struct dt_allocation_hint *ah,
 675                         struct dt_object *parent,
 676                         struct dt_object *child,
 677                         cfs_umode_t child_mode)
 678 {
 679         struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
 680         struct dt_object  *nextp = NULL;
 681         struct dt_object  *nextc;
 682         struct lod_object *lp = NULL;
 683         struct lod_object *lc;
 684         struct lov_desc   *desc;
 685         ENTRY;
 686
 687         LASSERT(child);
 688
 689         if (likely(parent)) {
 690                 nextp = dt_object_child(parent);
 691                 lp = lod_dt_obj(parent);
 692         }
 693
 694         nextc = dt_object_child(child);
 695         lc = lod_dt_obj(child);
 696
 697         LASSERT(lc->ldo_stripenr == 0);
 698         LASSERT(lc->ldo_stripe == NULL);
 699
 700         /*
 701          * local object may want some hints
 702          * in case of late striping creation, ->ah_init()
 703          * can be called with local object existing
 704          */
 705         if (!dt_object_exists(nextc) || dt_object_remote(nextc))
 706                 nextc->do_ops->do_ah_init(env, ah, nextp, nextc, child_mode);
 707
 708         if (S_ISDIR(child_mode)) {
 709                 if (lp->ldo_striping_cached == 0) {
 710                         /* we haven't tried to get default striping for
 711                          * the directory yet, let's cache it in the object */
 712                         lod_cache_parent_striping(env, lp);
 713                 }
 714                 /* transfer defaults to new directory */
 715                 if (lp->ldo_striping_cached) {
 716                         if (lp->ldo_pool)
 717                                 lod_object_set_pool(lc, lp->ldo_pool);
 718                         lc->ldo_def_stripenr = lp->ldo_def_stripenr;
 719                         lc->ldo_def_stripe_size = lp->ldo_def_stripe_size;
 720                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
 721                         lc->ldo_striping_cached = 1;
 722                         lc->ldo_def_striping_set = 1;
 723                         CDEBUG(D_OTHER, "inherite EA sz:%d off:%d nr:%d\n",
 724                                (int)lc->ldo_def_stripenr,
 725                                (int)lc->ldo_def_stripe_size,
 726                                (int)lc->ldo_def_stripe_offset);
 727                 }
 728                 return;
 729         }
 730
 731         /*
 732          * if object is going to be striped over OSTs, transfer default
 733          * striping information to the child, so that we can use it
 734          * during declaration and creation
 735          */
 736         if (!lod_object_will_be_striped(S_ISREG(child_mode),
 737                                         lu_object_fid(&child->do_lu)))
 738                 return;
 739
 740         /*
 741          * try from the parent
 742          */
 743         if (likely(parent)) {
 744                 if (lp->ldo_striping_cached == 0) {
 745                         /* we haven't tried to get default striping for
 746                          * the directory yet, let's cache it in the object */
 747                         lod_cache_parent_striping(env, lp);
 748                 }
 749
 750                 lc->ldo_def_stripe_offset = (__u16) -1;
 751
 752                 if (lp->ldo_def_striping_set) {
 753                         if (lp->ldo_pool)
 754                                 lod_object_set_pool(lc, lp->ldo_pool);
 755                         lc->ldo_stripenr = lp->ldo_def_stripenr;
 756                         lc->ldo_stripe_size = lp->ldo_def_stripe_size;
 757                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
 758                         CDEBUG(D_OTHER, "striping from parent: #%d, sz %d %s\n",
 759                                lc->ldo_stripenr, lc->ldo_stripe_size,
 760                                lp->ldo_pool ? lp->ldo_pool : "");
 761                 }
 762         }
 763
 764         /*
 765          * if the parent doesn't provide with specific pattern, grab fs-wide one
 766          */
 767         desc = &d->lod_desc;
 768         if (lc->ldo_stripenr == 0)
 769                 lc->ldo_stripenr = desc->ld_default_stripe_count;
 770         if (lc->ldo_stripe_size == 0)
 771                 lc->ldo_stripe_size = desc->ld_default_stripe_size;
 772         CDEBUG(D_OTHER, "final striping: # %d stripes, sz %d from %s\n",
 773                lc->ldo_stripenr, lc->ldo_stripe_size,
 774                lc->ldo_pool ? lc->ldo_pool : "");
 775
 776         EXIT;
 777 }
 778
 779 #define ll_do_div64(aaa,bbb)    do_div((aaa), (bbb))
 780 /*
 781  * this function handles a special case when truncate was done
 782  * on a stripeless object and now striping is being created
 783  * we can't lose that size, so we have to propagate it to newly
 784  * created object
 785  */
 786 static int lod_declare_init_size(const struct lu_env *env,
 787                                  struct dt_object *dt, struct thandle *th)
 788 {
 789         struct dt_object   *next = dt_object_child(dt);
 790         struct lod_object  *lo = lod_dt_obj(dt);
 791         struct lu_attr     *attr = &lod_env_info(env)->lti_attr;
 792         uint64_t            size, offs;
 793         int                 rc, stripe;
 794         ENTRY;
 795
 796         /* XXX: we support the simplest (RAID0) striping so far */
 797         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
 798         LASSERT(lo->ldo_stripe_size > 0);
 799
 800         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
 801         LASSERT(attr->la_valid & LA_SIZE);
 802         if (rc)
 803                 RETURN(rc);
 804
 805         size = attr->la_size;
 806         if (size == 0)
 807                 RETURN(0);
 808
 809         /* ll_do_div64(a, b) returns a % b, and a = a / b */
 810         ll_do_div64(size, (__u64) lo->ldo_stripe_size);
 811         stripe = ll_do_div64(size, (__u64) lo->ldo_stripenr);
 812
 813         size = size * lo->ldo_stripe_size;
 814         offs = attr->la_size;
 815         size += ll_do_div64(offs, lo->ldo_stripe_size);
 816
 817         attr->la_valid = LA_SIZE;
 818         attr->la_size = size;
 819
 820         rc = dt_declare_attr_set(env, lo->ldo_stripe[stripe], attr, th);
 821
 822         RETURN(rc);
 823 }
 824
 825
 826 /**
 827  * Create declaration of striped object
 828  */
 829 int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt,
 830                                struct lu_attr *attr,
 831                                const struct lu_buf *lovea, struct thandle *th)
 832 {
 833         struct lod_thread_info  *info = lod_env_info(env);
 834         struct dt_object        *next = dt_object_child(dt);
 835         struct lod_object       *lo = lod_dt_obj(dt);
 836         int                      rc;
 837         ENTRY;
 838
 839         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) {
 840                 /* failed to create striping, let's reset
 841                  * config so that others don't get confused */
 842                 lod_object_free_striping(env, lo);
 843                 GOTO(out, rc = -ENOMEM);
 844         }
 845
 846         /* choose OST and generate appropriate objects */
 847         rc = lod_qos_prep_create(env, lo, attr, lovea, th);
 848         if (rc) {
 849                 /* failed to create striping, let's reset
 850                  * config so that others don't get confused */
 851                 lod_object_free_striping(env, lo);
 852                 GOTO(out, rc);
 853         }
 854
 855         /*
 856          * declare storage for striping data
 857          */
 858         info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
 859                                 lo->ldo_pool ?  LOV_MAGIC_V3 : LOV_MAGIC_V1);
 860         rc = dt_declare_xattr_set(env, next, &info->lti_buf, XATTR_NAME_LOV,
 861                                   0, th);
 862         if (rc)
 863                 GOTO(out, rc);
 864
 865         /*
 866          * if striping is created with local object's size > 0,
 867          * we have to propagate this size to specific object
 868          * the case is possible only when local object was created previously
 869          */
 870         if (dt_object_exists(next))
 871                 rc = lod_declare_init_size(env, dt, th);
 872
 873 out:
 874         RETURN(rc);
 875 }
 876
 877 static int lod_declare_object_create(const struct lu_env *env,
 878                                      struct dt_object *dt,
 879                                      struct lu_attr *attr,
 880                                      struct dt_allocation_hint *hint,
 881                                      struct dt_object_format *dof,
 882                                      struct thandle *th)
 883 {
 884         struct dt_object   *next = dt_object_child(dt);
 885         struct lod_object  *lo = lod_dt_obj(dt);
 886         int                 rc;
 887         ENTRY;
 888
 889         LASSERT(dof);
 890         LASSERT(attr);
 891         LASSERT(th);
 892
 893         /*
 894          * first of all, we declare creation of local object
 895          */
 896         rc = dt_declare_create(env, next, attr, hint, dof, th);
 897         if (rc)
 898                 GOTO(out, rc);
 899
 900         if (dof->dof_type == DFT_SYM)
 901                 dt->do_body_ops = &lod_body_lnk_ops;
 902
 903         /*
 904          * it's lod_ah_init() who has decided the object will striped
 905          */
 906         if (dof->dof_type == DFT_REGULAR) {
 907                 /* callers don't want stripes */
 908                 /* XXX: all tricky interactions with ->ah_make_hint() decided
 909                  * to use striping, then ->declare_create() behaving differently
 910                  * should be cleaned */
 911                 if (dof->u.dof_reg.striped == 0)
 912                         lo->ldo_stripenr = 0;
 913                 if (lo->ldo_stripenr > 0)
 914                         rc = lod_declare_striped_object(env, dt, attr,
 915                                                         NULL, th);
 916         } else if (dof->dof_type == DFT_DIR && lo->ldo_striping_cached) {
 917                 struct lod_thread_info *info = lod_env_info(env);
 918
 919                 struct lov_user_md_v3 *v3;
 920
 921                 if (LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
 922                                         lo->ldo_def_stripenr,
 923                                         lo->ldo_def_stripe_offset))
 924                         RETURN(0);
 925
 926                 OBD_ALLOC_PTR(v3);
 927                 if (v3 == NULL)
 928                         RETURN(-ENOMEM);
 929
 930                 v3->lmm_magic = cpu_to_le32(LOV_MAGIC_V3);
 931                 v3->lmm_pattern = cpu_to_le32(LOV_PATTERN_RAID0);
 932                 fid_to_lmm_oi(lu_object_fid(&dt->do_lu), &v3->lmm_oi);
 933                 lmm_oi_cpu_to_le(&v3->lmm_oi, &v3->lmm_oi);
 934                 v3->lmm_stripe_size = cpu_to_le32(lo->ldo_def_stripe_size);
 935                 v3->lmm_stripe_count = cpu_to_le32(lo->ldo_def_stripenr);
 936                 v3->lmm_stripe_offset = cpu_to_le16(lo->ldo_def_stripe_offset);
 937                 if (lo->ldo_pool)
 938                         strncpy(v3->lmm_pool_name, lo->ldo_pool,
 939                                 LOV_MAXPOOLNAME);
 940
 941                 info->lti_buf.lb_buf = v3;
 942                 info->lti_buf.lb_len = sizeof(*v3);
 943
 944                 /* to transfer default striping from the parent */
 945                 rc = dt_declare_xattr_set(env, next, &info->lti_buf,
 946                                           XATTR_NAME_LOV, 0, th);
 947                 OBD_FREE_PTR(v3);
 948         }
 949
 950 out:
 951         RETURN(rc);
 952 }
 953
 954 int lod_striping_create(const struct lu_env *env, struct dt_object *dt,
 955                         struct lu_attr *attr, struct dt_object_format *dof,
 956                         struct thandle *th)
 957 {
 958         struct lod_object *lo = lod_dt_obj(dt);
 959         int                rc = 0, i;
 960         ENTRY;
 961
 962         LASSERT(lo->ldo_striping_cached == 0);
 963
 964         /* create all underlying objects */
 965         for (i = 0; i < lo->ldo_stripenr; i++) {
 966                 LASSERT(lo->ldo_stripe[i]);
 967                 rc = dt_create(env, lo->ldo_stripe[i], attr, NULL, dof, th);
 968
 969                 if (rc)
 970                         break;
 971         }
 972         if (rc == 0)
 973                 rc = lod_generate_and_set_lovea(env, lo, th);
 974
 975         RETURN(rc);
 976 }
 977
 978 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
 979                              struct lu_attr *attr,
 980                              struct dt_allocation_hint *hint,
 981                              struct dt_object_format *dof, struct thandle *th)
 982 {
 983         struct dt_object   *next = dt_object_child(dt);
 984         struct lod_object  *lo = lod_dt_obj(dt);
 985         int                 rc;
 986         ENTRY;
 987
 988         /* create local object */
 989         rc = dt_create(env, next, attr, hint, dof, th);
 990
 991         if (rc == 0) {
 992                 if (S_ISDIR(dt->do_lu.lo_header->loh_attr))
 993                         rc = lod_store_def_striping(env, dt, th);
 994                 else if (lo->ldo_stripe)
 995                         rc = lod_striping_create(env, dt, attr, dof, th);
 996         }
 997
 998         RETURN(rc);
 999 }
1000
1001 static int lod_declare_object_destroy(const struct lu_env *env,
1002                                       struct dt_object *dt,
1003                                       struct thandle *th)
1004 {
1005         struct dt_object   *next = dt_object_child(dt);
1006         struct lod_object  *lo = lod_dt_obj(dt);
1007         int                 rc, i;
1008         ENTRY;
1009
1010         /*
1011          * we declare destroy for the local object
1012          */
1013         rc = dt_declare_destroy(env, next, th);
1014         if (rc)
1015                 RETURN(rc);
1016
1017         /*
1018          * load striping information, notice we don't do this when object
1019          * is being initialized as we don't need this information till
1020          * few specific cases like destroy, chown
1021          */
1022         rc = lod_load_striping(env, lo);
1023         if (rc)
1024                 RETURN(rc);
1025
1026         /* declare destroy for all underlying objects */
1027         for (i = 0; i < lo->ldo_stripenr; i++) {
1028                 LASSERT(lo->ldo_stripe[i]);
1029                 rc = dt_declare_destroy(env, lo->ldo_stripe[i], th);
1030
1031                 if (rc)
1032                         break;
1033         }
1034
1035         RETURN(rc);
1036 }
1037
1038 static int lod_object_destroy(const struct lu_env *env,
1039                 struct dt_object *dt, struct thandle *th)
1040 {
1041         struct dt_object  *next = dt_object_child(dt);
1042         struct lod_object *lo = lod_dt_obj(dt);
1043         int                rc, i;
1044         ENTRY;
1045
1046         /* destroy local object */
1047         rc = dt_destroy(env, next, th);
1048         if (rc)
1049                 RETURN(rc);
1050
1051         /* destroy all underlying objects */
1052         for (i = 0; i < lo->ldo_stripenr; i++) {
1053                 LASSERT(lo->ldo_stripe[i]);
1054                 rc = dt_destroy(env, lo->ldo_stripe[i], th);
1055                 if (rc)
1056                         break;
1057         }
1058
1059         RETURN(rc);
1060 }
1061
1062 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
1063                          const struct dt_index_features *feat)
1064 {
1065         struct dt_object *next = dt_object_child(dt);
1066         int               rc;
1067         ENTRY;
1068
1069         LASSERT(next->do_ops);
1070         LASSERT(next->do_ops->do_index_try);
1071
1072         rc = next->do_ops->do_index_try(env, next, feat);
1073         if (next->do_index_ops && dt->do_index_ops == NULL)
1074                 dt->do_index_ops = &lod_index_ops;
1075
1076         RETURN(rc);
1077 }
1078
1079 static int lod_declare_ref_add(const struct lu_env *env,
1080                                struct dt_object *dt, struct thandle *th)
1081 {
1082         return dt_declare_ref_add(env, dt_object_child(dt), th);
1083 }
1084
1085 static int lod_ref_add(const struct lu_env *env,
1086                        struct dt_object *dt, struct thandle *th)
1087 {
1088         return dt_ref_add(env, dt_object_child(dt), th);
1089 }
1090
1091 static int lod_declare_ref_del(const struct lu_env *env,
1092                                struct dt_object *dt, struct thandle *th)
1093 {
1094         return dt_declare_ref_del(env, dt_object_child(dt), th);
1095 }
1096
1097 static int lod_ref_del(const struct lu_env *env,
1098                        struct dt_object *dt, struct thandle *th)
1099 {
1100         return dt_ref_del(env, dt_object_child(dt), th);
1101 }
1102
1103 static struct obd_capa *lod_capa_get(const struct lu_env *env,
1104                                      struct dt_object *dt,
1105                                      struct lustre_capa *old, __u64 opc)
1106 {
1107         return dt_capa_get(env, dt_object_child(dt), old, opc);
1108 }
1109
1110 static int lod_object_sync(const struct lu_env *env, struct dt_object *dt)
1111 {
1112         return dt_object_sync(env, dt_object_child(dt));
1113 }
1114
1115 static int lod_object_lock(const struct lu_env *env,
1116                            struct dt_object *dt, struct lustre_handle *lh,
1117                            struct ldlm_enqueue_info *einfo,
1118                            void *policy)
1119 {
1120         struct dt_object   *next = dt_object_child(dt);
1121         int              rc;
1122         ENTRY;
1123
1124         /*
1125          * declare setattr on the local object
1126          */
1127         rc = dt_object_lock(env, next, lh, einfo, policy);
1128
1129         RETURN(rc);
1130 }
1131
1132 struct dt_object_operations lod_obj_ops = {
1133         .do_read_lock           = lod_object_read_lock,
1134         .do_write_lock          = lod_object_write_lock,
1135         .do_read_unlock         = lod_object_read_unlock,
1136         .do_write_unlock        = lod_object_write_unlock,
1137         .do_write_locked        = lod_object_write_locked,
1138         .do_attr_get            = lod_attr_get,
1139         .do_declare_attr_set    = lod_declare_attr_set,
1140         .do_attr_set            = lod_attr_set,
1141         .do_xattr_get           = lod_xattr_get,
1142         .do_declare_xattr_set   = lod_declare_xattr_set,
1143         .do_xattr_set           = lod_xattr_set,
1144         .do_declare_xattr_del   = lod_declare_xattr_del,
1145         .do_xattr_del           = lod_xattr_del,
1146         .do_xattr_list          = lod_xattr_list,
1147         .do_ah_init             = lod_ah_init,
1148         .do_declare_create      = lod_declare_object_create,
1149         .do_create              = lod_object_create,
1150         .do_declare_destroy     = lod_declare_object_destroy,
1151         .do_destroy             = lod_object_destroy,
1152         .do_index_try           = lod_index_try,
1153         .do_declare_ref_add     = lod_declare_ref_add,
1154         .do_ref_add             = lod_ref_add,
1155         .do_declare_ref_del     = lod_declare_ref_del,
1156         .do_ref_del             = lod_ref_del,
1157         .do_capa_get            = lod_capa_get,
1158         .do_object_sync         = lod_object_sync,
1159         .do_object_lock         = lod_object_lock,
1160 };
1161
1162 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
1163                         struct lu_buf *buf, loff_t *pos,
1164                         struct lustre_capa *capa)
1165 {
1166         struct dt_object *next = dt_object_child(dt);
1167         return next->do_body_ops->dbo_read(env, next, buf, pos, capa);
1168 }
1169
1170 static ssize_t lod_declare_write(const struct lu_env *env,
1171                                  struct dt_object *dt,
1172                                  const loff_t size, loff_t pos,
1173                                  struct thandle *th)
1174 {
1175         return dt_declare_record_write(env, dt_object_child(dt),
1176                                        size, pos, th);
1177 }
1178
1179 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,
1180                          const struct lu_buf *buf, loff_t *pos,
1181                          struct thandle *th, struct lustre_capa *capa, int iq)
1182 {
1183         struct dt_object *next = dt_object_child(dt);
1184         LASSERT(next);
1185         return next->do_body_ops->dbo_write(env, next, buf, pos, th, capa, iq);
1186 }
1187
1188 static const struct dt_body_operations lod_body_lnk_ops = {
1189         .dbo_read               = lod_read,
1190         .dbo_declare_write      = lod_declare_write,
1191         .dbo_write              = lod_write
1192 };
1193
1194 static int lod_object_init(const struct lu_env *env, struct lu_object *o,
1195                            const struct lu_object_conf *conf)
1196 {
1197         struct lod_device *d = lu2lod_dev(o->lo_dev);
1198         struct lu_object  *below;
1199         struct lu_device  *under;
1200         ENTRY;
1201
1202         /*
1203          * create local object
1204          */
1205         under = &d->lod_child->dd_lu_dev;
1206         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
1207         if (below == NULL)
1208                 RETURN(-ENOMEM);
1209
1210         lu_object_add(o, below);
1211
1212         RETURN(0);
1213 }
1214
1215 void lod_object_free_striping(const struct lu_env *env, struct lod_object *lo)
1216 {
1217         int i;
1218
1219         if (lo->ldo_stripe) {
1220                 LASSERT(lo->ldo_stripes_allocated > 0);
1221
1222                 for (i = 0; i < lo->ldo_stripenr; i++) {
1223                         if (lo->ldo_stripe[i])
1224                                 lu_object_put(env, &lo->ldo_stripe[i]->do_lu);
1225                 }
1226
1227                 i = sizeof(struct dt_object *) * lo->ldo_stripes_allocated;
1228                 OBD_FREE(lo->ldo_stripe, i);
1229                 lo->ldo_stripe = NULL;
1230                 lo->ldo_stripes_allocated = 0;
1231         }
1232         lo->ldo_stripenr = 0;
1233         lo->ldo_pattern = 0;
1234 }
1235
1236 /*
1237  * ->start is called once all slices are initialized, including header's
1238  * cache for mode (object type). using the type we can initialize ops
1239  */
1240 static int lod_object_start(const struct lu_env *env, struct lu_object *o)
1241 {
1242         if (S_ISLNK(o->lo_header->loh_attr & S_IFMT))
1243                 lu2lod_obj(o)->ldo_obj.do_body_ops = &lod_body_lnk_ops;
1244         return 0;
1245 }
1246
1247 static void lod_object_free(const struct lu_env *env, struct lu_object *o)
1248 {
1249         struct lod_object *mo = lu2lod_obj(o);
1250
1251         /*
1252          * release all underlying object pinned
1253          */
1254
1255         lod_object_free_striping(env, mo);
1256
1257         lod_object_set_pool(mo, NULL);
1258
1259         lu_object_fini(o);
1260         OBD_SLAB_FREE_PTR(mo, lod_object_kmem);
1261 }
1262
1263 static void lod_object_release(const struct lu_env *env, struct lu_object *o)
1264 {
1265         /* XXX: shouldn't we release everything here in case if object
1266          * creation failed before? */
1267 }
1268
1269 static int lod_object_print(const struct lu_env *env, void *cookie,
1270                             lu_printer_t p, const struct lu_object *l)
1271 {
1272         struct lod_object *o = lu2lod_obj((struct lu_object *) l);
1273
1274         return (*p)(env, cookie, LUSTRE_LOD_NAME"-object@%p", o);
1275 }
1276
1277 struct lu_object_operations lod_lu_obj_ops = {
1278         .loo_object_init        = lod_object_init,
1279         .loo_object_start       = lod_object_start,
1280         .loo_object_free        = lod_object_free,
1281         .loo_object_release     = lod_object_release,
1282         .loo_object_print       = lod_object_print,
1283 };
1284
1285 /**
1286  * Init remote lod object
1287  */
1288 static int lod_robject_init(const struct lu_env *env, struct lu_object *lo,
1289                             const struct lu_object_conf *conf)
1290 {
1291         struct lod_device *lod = lu2lod_dev(lo->lo_dev);
1292         struct lod_tgt_descs *ltd = &lod->lod_mdt_descs;
1293         struct lu_device  *c_dev = NULL;
1294         struct lu_object  *c_obj;
1295         int i;
1296         ENTRY;
1297
1298         lod_getref(ltd);
1299         if (ltd->ltd_tgts_size > 0) {
1300                 cfs_foreach_bit(ltd->ltd_tgt_bitmap, i) {
1301                         struct lod_tgt_desc *tgt;
1302                         tgt = LTD_TGT(ltd, i);
1303                         LASSERT(tgt && tgt->ltd_tgt);
1304                         if (tgt->ltd_index ==
1305                             lu2lod_obj(lo)->ldo_mds_num) {
1306                                 c_dev = &(tgt->ltd_tgt->dd_lu_dev);
1307                                 break;
1308                         }
1309                 }
1310         }
1311         lod_putref(lod, ltd);
1312
1313         if (unlikely(c_dev == NULL))
1314                 RETURN(-ENOENT);
1315
1316         c_obj = c_dev->ld_ops->ldo_object_alloc(env, lo->lo_header, c_dev);
1317         if (unlikely(c_obj == NULL))
1318                 RETURN(-ENOMEM);
1319
1320         lu_object_add(lo, c_obj);
1321
1322         RETURN(0);
1323 }
1324
1325 struct lu_object_operations lod_lu_robj_ops = {
1326         .loo_object_init      = lod_robject_init,
1327         .loo_object_start     = lod_object_start,
1328         .loo_object_free      = lod_object_free,
1329         .loo_object_release   = lod_object_release,
1330         .loo_object_print     = lod_object_print,
1331 };