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