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 <lustre_lmv.h>
  46 #include <obd_lov.h>
  47 #include <md_object.h>
  48
  49 #include "lod_internal.h"
  50
  51 static const char dot[] = ".";
  52 static const char dotdot[] = "..";
  53
  54 extern struct kmem_cache *lod_object_kmem;
  55 static const struct dt_body_operations lod_body_lnk_ops;
  56
  57 static int lod_index_lookup(const struct lu_env *env, struct dt_object *dt,
  58                             struct dt_rec *rec, const struct dt_key *key,
  59                             struct lustre_capa *capa)
  60 {
  61         struct dt_object *next = dt_object_child(dt);
  62         return next->do_index_ops->dio_lookup(env, next, rec, key, capa);
  63 }
  64
  65 static int lod_declare_index_insert(const struct lu_env *env,
  66                                     struct dt_object *dt,
  67                                     const struct dt_rec *rec,
  68                                     const struct dt_key *key,
  69                                     struct thandle *handle)
  70 {
  71         return dt_declare_insert(env, dt_object_child(dt), rec, key, handle);
  72 }
  73
  74 static int lod_index_insert(const struct lu_env *env,
  75                             struct dt_object *dt,
  76                             const struct dt_rec *rec,
  77                             const struct dt_key *key,
  78                             struct thandle *th,
  79                             struct lustre_capa *capa,
  80                             int ign)
  81 {
  82         return dt_insert(env, dt_object_child(dt), rec, key, th, capa, ign);
  83 }
  84
  85 static int lod_declare_index_delete(const struct lu_env *env,
  86                                     struct dt_object *dt,
  87                                     const struct dt_key *key,
  88                                     struct thandle *th)
  89 {
  90         return dt_declare_delete(env, dt_object_child(dt), key, th);
  91 }
  92
  93 static int lod_index_delete(const struct lu_env *env,
  94                             struct dt_object *dt,
  95                             const struct dt_key *key,
  96                             struct thandle *th,
  97                             struct lustre_capa *capa)
  98 {
  99         return dt_delete(env, dt_object_child(dt), key, th, capa);
 100 }
 101
 102 static struct dt_it *lod_it_init(const struct lu_env *env,
 103                                  struct dt_object *dt, __u32 attr,
 104                                  struct lustre_capa *capa)
 105 {
 106         struct dt_object        *next = dt_object_child(dt);
 107         struct lod_it           *it = &lod_env_info(env)->lti_it;
 108         struct dt_it            *it_next;
 109
 110
 111         it_next = next->do_index_ops->dio_it.init(env, next, attr, capa);
 112         if (IS_ERR(it_next))
 113                 return it_next;
 114
 115         /* currently we do not use more than one iterator per thread
 116          * so we store it in thread info. if at some point we need
 117          * more active iterators in a single thread, we can allocate
 118          * additional ones */
 119         LASSERT(it->lit_obj == NULL);
 120
 121         it->lit_it = it_next;
 122         it->lit_obj = next;
 123
 124         return (struct dt_it *)it;
 125 }
 126
 127 #define LOD_CHECK_IT(env, it)                                   \
 128 {                                                               \
 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         /* osp_declare_attr_set() ignores all attributes other than
 297          * UID, GID, and size, and osp_attr_set() ignores all but UID
 298          * and GID.  Declaration of size attr setting happens through
 299          * lod_declare_init_size(), and not through this function.
 300          * Therefore we need not load striping unless ownership is
 301          * changing.  This should save memory and (we hope) speed up
 302          * rename(). */
 303         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
 304                 if (!(attr->la_valid & (LA_UID | LA_GID)))
 305                         RETURN(rc);
 306
 307                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
 308                         RETURN(0);
 309         } else {
 310                 if (!(attr->la_valid & (LA_UID | LA_GID | LA_MODE |
 311                                         LA_ATIME | LA_MTIME | LA_CTIME)))
 312                         RETURN(rc);
 313         }
 314         /*
 315          * load striping information, notice we don't do this when object
 316          * is being initialized as we don't need this information till
 317          * few specific cases like destroy, chown
 318          */
 319         rc = lod_load_striping(env, lo);
 320         if (rc)
 321                 RETURN(rc);
 322
 323         if (lo->ldo_stripenr == 0)
 324                 RETURN(0);
 325
 326         if (!(attr->la_valid & ~(LA_ATIME | LA_MTIME | LA_CTIME))) {
 327                 struct lu_attr   *la = &lod_env_info(env)->lti_attr;
 328                 bool             setattr_time = false;
 329
 330                 rc = dt_attr_get(env, dt_object_child(dt), la,
 331                                  BYPASS_CAPA);
 332                 if (rc != 0)
 333                         RETURN(rc);
 334
 335                 /* If it will only setattr time, it will only set
 336                  * time < current_time */
 337                 if ((attr->la_valid & LA_ATIME &&
 338                      attr->la_atime < la->la_atime) ||
 339                     (attr->la_valid & LA_CTIME &&
 340                      attr->la_ctime < la->la_ctime) ||
 341                     (attr->la_valid & LA_MTIME &&
 342                      attr->la_mtime < la->la_mtime))
 343                         setattr_time = true;
 344
 345                 if (!setattr_time)
 346                         RETURN(0);
 347         }
 348         /*
 349          * if object is striped declare changes on the stripes
 350          */
 351         LASSERT(lo->ldo_stripe);
 352         for (i = 0; i < lo->ldo_stripenr; i++) {
 353                 LASSERT(lo->ldo_stripe[i]);
 354
 355                 rc = dt_declare_attr_set(env, lo->ldo_stripe[i], attr, handle);
 356                 if (rc) {
 357                         CERROR("failed declaration: %d\n", rc);
 358                         break;
 359                 }
 360         }
 361
 362         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE) &&
 363             dt_object_exists(next) != 0 &&
 364             dt_object_remote(next) == 0)
 365                 dt_declare_xattr_del(env, next, XATTR_NAME_LOV, handle);
 366
 367         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE) &&
 368             dt_object_exists(next) &&
 369             dt_object_remote(next) == 0 && S_ISREG(attr->la_mode)) {
 370                 struct lod_thread_info *info = lod_env_info(env);
 371                 struct lu_buf *buf = &info->lti_buf;
 372
 373                 buf->lb_buf = info->lti_ea_store;
 374                 buf->lb_len = info->lti_ea_store_size;
 375                 dt_declare_xattr_set(env, next, buf, XATTR_NAME_LOV,
 376                                      LU_XATTR_REPLACE, handle);
 377         }
 378
 379         RETURN(rc);
 380 }
 381
 382 static int lod_attr_set(const struct lu_env *env,
 383                         struct dt_object *dt,
 384                         const struct lu_attr *attr,
 385                         struct thandle *handle,
 386                         struct lustre_capa *capa)
 387 {
 388         struct dt_object  *next = dt_object_child(dt);
 389         struct lod_object *lo = lod_dt_obj(dt);
 390         int                rc, i;
 391         ENTRY;
 392
 393         /*
 394          * apply changes to the local object
 395          */
 396         rc = dt_attr_set(env, next, attr, handle, capa);
 397         if (rc)
 398                 RETURN(rc);
 399
 400         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
 401                 if (!(attr->la_valid & (LA_UID | LA_GID)))
 402                         RETURN(rc);
 403
 404                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
 405                         RETURN(0);
 406         } else {
 407                 if (!(attr->la_valid & (LA_UID | LA_GID | LA_MODE |
 408                                         LA_ATIME | LA_MTIME | LA_CTIME)))
 409                         RETURN(rc);
 410         }
 411
 412         if (lo->ldo_stripenr == 0)
 413                 RETURN(0);
 414
 415         if (!(attr->la_valid & ~(LA_ATIME | LA_MTIME | LA_CTIME))) {
 416                 struct lu_attr   *la = &lod_env_info(env)->lti_attr;
 417                 bool             setattr_time = false;
 418
 419                 rc = dt_attr_get(env, dt_object_child(dt), la,
 420                                  BYPASS_CAPA);
 421                 if (rc != 0)
 422                         RETURN(rc);
 423
 424                 /* If it will only setattr time, it will only set
 425                  * time < current_time */
 426                 if ((attr->la_valid & LA_ATIME &&
 427                      attr->la_atime < la->la_atime) ||
 428                     (attr->la_valid & LA_CTIME &&
 429                      attr->la_ctime < la->la_ctime) ||
 430                     (attr->la_valid & LA_MTIME &&
 431                      attr->la_mtime < la->la_mtime))
 432                         setattr_time = true;
 433
 434                 if (!setattr_time)
 435                         RETURN(0);
 436         }
 437
 438         /*
 439          * if object is striped, apply changes to all the stripes
 440          */
 441         LASSERT(lo->ldo_stripe);
 442         for (i = 0; i < lo->ldo_stripenr; i++) {
 443                 LASSERT(lo->ldo_stripe[i]);
 444                 rc = dt_attr_set(env, lo->ldo_stripe[i], attr, handle, capa);
 445                 if (rc) {
 446                         CERROR("failed declaration: %d\n", rc);
 447                         break;
 448                 }
 449         }
 450
 451         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE) &&
 452             dt_object_exists(next) != 0 &&
 453             dt_object_remote(next) == 0)
 454                 dt_xattr_del(env, next, XATTR_NAME_LOV, handle, BYPASS_CAPA);
 455
 456         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE) &&
 457             dt_object_exists(next) &&
 458             dt_object_remote(next) == 0 && S_ISREG(attr->la_mode)) {
 459                 struct lod_thread_info *info = lod_env_info(env);
 460                 struct lu_buf *buf = &info->lti_buf;
 461                 struct ost_id *oi = &info->lti_ostid;
 462                 struct lu_fid *fid = &info->lti_fid;
 463                 struct lov_mds_md_v1 *lmm;
 464                 struct lov_ost_data_v1 *objs;
 465                 __u32 magic;
 466                 int rc1;
 467
 468                 rc1 = lod_get_lov_ea(env, lo);
 469                 if (rc1  <= 0)
 470                         RETURN(rc);
 471
 472                 buf->lb_buf = info->lti_ea_store;
 473                 buf->lb_len = info->lti_ea_store_size;
 474                 lmm = info->lti_ea_store;
 475                 magic = le32_to_cpu(lmm->lmm_magic);
 476                 if (magic == LOV_MAGIC_V1)
 477                         objs = &(lmm->lmm_objects[0]);
 478                 else
 479                         objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
 480                 ostid_le_to_cpu(&objs->l_ost_oi, oi);
 481                 ostid_to_fid(fid, oi, le32_to_cpu(objs->l_ost_idx));
 482                 fid->f_oid--;
 483                 fid_to_ostid(fid, oi);
 484                 ostid_cpu_to_le(oi, &objs->l_ost_oi);
 485                 dt_xattr_set(env, next, buf, XATTR_NAME_LOV,
 486                              LU_XATTR_REPLACE, handle, BYPASS_CAPA);
 487         }
 488
 489         RETURN(rc);
 490 }
 491
 492 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
 493                          struct lu_buf *buf, const char *name,
 494                          struct lustre_capa *capa)
 495 {
 496         struct lod_thread_info  *info = lod_env_info(env);
 497         struct lod_device       *dev = lu2lod_dev(dt->do_lu.lo_dev);
 498         int                      rc, is_root;
 499         ENTRY;
 500
 501         rc = dt_xattr_get(env, dt_object_child(dt), buf, name, capa);
 502         if (rc != -ENODATA || !S_ISDIR(dt->do_lu.lo_header->loh_attr & S_IFMT))
 503                 RETURN(rc);
 504
 505         /*
 506          * lod returns default striping on the real root of the device
 507          * this is like the root stores default striping for the whole
 508          * filesystem. historically we've been using a different approach
 509          * and store it in the config.
 510          */
 511         dt_root_get(env, dev->lod_child, &info->lti_fid);
 512         is_root = lu_fid_eq(&info->lti_fid, lu_object_fid(&dt->do_lu));
 513
 514         if (is_root && strcmp(XATTR_NAME_LOV, name) == 0) {
 515                 struct lov_user_md *lum = buf->lb_buf;
 516                 struct lov_desc    *desc = &dev->lod_desc;
 517
 518                 if (buf->lb_buf == NULL) {
 519                         rc = sizeof(*lum);
 520                 } else if (buf->lb_len >= sizeof(*lum)) {
 521                         lum->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V1);
 522                         lmm_oi_set_seq(&lum->lmm_oi, FID_SEQ_LOV_DEFAULT);
 523                         lmm_oi_set_id(&lum->lmm_oi, 0);
 524                         lmm_oi_cpu_to_le(&lum->lmm_oi, &lum->lmm_oi);
 525                         lum->lmm_pattern = cpu_to_le32(desc->ld_pattern);
 526                         lum->lmm_stripe_size = cpu_to_le32(
 527                                                 desc->ld_default_stripe_size);
 528                         lum->lmm_stripe_count = cpu_to_le16(
 529                                                 desc->ld_default_stripe_count);
 530                         lum->lmm_stripe_offset = cpu_to_le16(
 531                                                 desc->ld_default_stripe_offset);
 532                         rc = sizeof(*lum);
 533                 } else {
 534                         rc = -ERANGE;
 535                 }
 536         }
 537
 538         RETURN(rc);
 539 }
 540
 541 static int lod_verify_md_striping(struct lod_device *lod,
 542                                   const struct lmv_user_md_v1 *lum)
 543 {
 544         int     rc = 0;
 545         ENTRY;
 546
 547         if (unlikely(le32_to_cpu(lum->lum_magic) != LMV_USER_MAGIC))
 548                 GOTO(out, rc = -EINVAL);
 549
 550         if (unlikely(le32_to_cpu(lum->lum_stripe_count) == 0))
 551                 GOTO(out, rc = -EINVAL);
 552
 553         if (unlikely(le32_to_cpu(lum->lum_stripe_count) >
 554                                 lod->lod_remote_mdt_count + 1))
 555                 GOTO(out, rc = -EINVAL);
 556 out:
 557         if (rc != 0)
 558                 CERROR("%s: invalid lmv_user_md: magic = %x, "
 559                        "stripe_offset = %d, stripe_count = %u: rc = %d\n",
 560                        lod2obd(lod)->obd_name, le32_to_cpu(lum->lum_magic),
 561                        (int)le32_to_cpu(lum->lum_stripe_offset),
 562                        le32_to_cpu(lum->lum_stripe_count), rc);
 563         return rc;
 564 }
 565
 566 int lod_prep_lmv_md(const struct lu_env *env, struct dt_object *dt,
 567                     struct lu_buf *lmv_buf)
 568 {
 569         struct lod_thread_info  *info = lod_env_info(env);
 570         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
 571         struct lod_object       *lo = lod_dt_obj(dt);
 572         struct lmv_mds_md_v1    *lmm1;
 573         int                     stripe_count;
 574         int                     lmm_size;
 575         int                     type = LU_SEQ_RANGE_ANY;
 576         int                     i;
 577         int                     rc;
 578         __u32                   mdtidx;
 579         ENTRY;
 580
 581         LASSERT(lo->ldo_dir_striped != 0);
 582         LASSERT(lo->ldo_stripenr > 0);
 583         stripe_count = lo->ldo_stripenr + 1;
 584         lmm_size = lmv_mds_md_size(stripe_count, LMV_MAGIC);
 585         if (info->lti_ea_store_size < lmm_size) {
 586                 rc = lod_ea_store_resize(info, lmm_size);
 587                 if (rc != 0)
 588                         RETURN(rc);
 589         }
 590
 591         lmm1 = (struct lmv_mds_md_v1 *)info->lti_ea_store;
 592         lmm1->lmv_magic = cpu_to_le32(LMV_MAGIC);
 593         lmm1->lmv_stripe_count = cpu_to_le32(stripe_count);
 594         lmm1->lmv_hash_type = cpu_to_le32(lo->ldo_dir_hash_type);
 595         rc = lod_fld_lookup(env, lod, lu_object_fid(&dt->do_lu),
 596                             &mdtidx, &type);
 597         if (rc != 0)
 598                 RETURN(rc);
 599
 600         lmm1->lmv_master_mdt_index = cpu_to_le32(mdtidx);
 601         fid_cpu_to_le(&lmm1->lmv_stripe_fids[0], lu_object_fid(&dt->do_lu));
 602         for (i = 0; i < lo->ldo_stripenr; i++) {
 603                 struct dt_object *dto;
 604
 605                 dto = lo->ldo_stripe[i];
 606                 LASSERT(dto != NULL);
 607                 fid_cpu_to_le(&lmm1->lmv_stripe_fids[i + 1],
 608                               lu_object_fid(&dto->do_lu));
 609         }
 610
 611         lmv_buf->lb_buf = info->lti_ea_store;
 612         lmv_buf->lb_len = lmm_size;
 613         lo->ldo_dir_striping_cached = 1;
 614
 615         RETURN(rc);
 616 }
 617
 618 int lod_parse_dir_striping(const struct lu_env *env, struct lod_object *lo,
 619                            const struct lu_buf *buf)
 620 {
 621         struct lod_thread_info  *info = lod_env_info(env);
 622         struct lod_device       *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
 623         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
 624         struct dt_object        **stripe;
 625         union lmv_mds_md        *lmm = buf->lb_buf;
 626         struct lmv_mds_md_v1    *lmv1 = &lmm->lmv_md_v1;
 627         struct lu_fid           *fid = &info->lti_fid;
 628         int                     i;
 629         int                     rc = 0;
 630         ENTRY;
 631
 632         if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_MIGRATE)
 633                 RETURN(0);
 634
 635         if (le32_to_cpu(lmv1->lmv_magic) != LMV_MAGIC_V1)
 636                 RETURN(-EINVAL);
 637
 638         if (le32_to_cpu(lmv1->lmv_stripe_count) <= 1)
 639                 RETURN(0);
 640
 641         fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[0]);
 642         /* Do not load striping information for slave inode */
 643         if (!lu_fid_eq(fid, lu_object_fid(&lo->ldo_obj.do_lu))) {
 644                 lo->ldo_dir_slave_stripe = 1;
 645                 RETURN(0);
 646         }
 647
 648         LASSERT(lo->ldo_stripe == NULL);
 649         OBD_ALLOC(stripe, sizeof(stripe[0]) *
 650                   (le32_to_cpu(lmv1->lmv_stripe_count) - 1));
 651         if (stripe == NULL)
 652                 RETURN(-ENOMEM);
 653
 654         /* skip master stripe */
 655         for (i = 1; i < le32_to_cpu(lmv1->lmv_stripe_count); i++) {
 656                 struct lod_tgt_desc     *tgt;
 657                 int                     idx;
 658                 int                     type = LU_SEQ_RANGE_ANY;
 659                 struct dt_object        *dto;
 660
 661                 fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[i]);
 662                 rc = lod_fld_lookup(env, lod, fid, &idx, &type);
 663                 if (rc != 0)
 664                         GOTO(out, rc);
 665
 666                 tgt = LTD_TGT(ltd, idx);
 667                 if (tgt == NULL)
 668                         GOTO(out, rc = -ESTALE);
 669
 670                 dto = dt_locate_at(env, tgt->ltd_tgt, fid,
 671                                   lo->ldo_obj.do_lu.lo_dev->ld_site->ls_top_dev,
 672                                   NULL);
 673                 if (IS_ERR(dto))
 674                         GOTO(out, rc = PTR_ERR(dto));
 675
 676                 stripe[i - 1] = dto;
 677         }
 678 out:
 679         lo->ldo_stripe = stripe;
 680         lo->ldo_stripenr = le32_to_cpu(lmv1->lmv_stripe_count) - 1;
 681         lo->ldo_stripes_allocated = le32_to_cpu(lmv1->lmv_stripe_count) - 1;
 682         if (rc != 0)
 683                 lod_object_free_striping(env, lo);
 684
 685         RETURN(rc);
 686 }
 687
 688 static int lod_prep_md_striped_create(const struct lu_env *env,
 689                                       struct dt_object *dt,
 690                                       struct lu_attr *attr,
 691                                       const struct lmv_user_md_v1 *lum,
 692                                       struct thandle *th)
 693 {
 694         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
 695         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
 696         struct lod_object       *lo = lod_dt_obj(dt);
 697         struct dt_object        **stripe;
 698         struct lu_buf           lmv_buf;
 699         int                     stripe_count;
 700         int                     *idx_array;
 701         int                     rc = 0;
 702         int                     i;
 703         int                     j;
 704         ENTRY;
 705
 706         /* The lum has been verifed in lod_verify_md_striping */
 707         LASSERT(le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC);
 708         LASSERT(le32_to_cpu(lum->lum_stripe_count) > 0);
 709
 710         /* Do not need allocated master stripe */
 711         stripe_count = le32_to_cpu(lum->lum_stripe_count);
 712         OBD_ALLOC(stripe, sizeof(stripe[0]) * (stripe_count - 1));
 713         if (stripe == NULL)
 714                 RETURN(-ENOMEM);
 715
 716         OBD_ALLOC(idx_array, sizeof(idx_array[0]) * stripe_count);
 717         if (idx_array == NULL)
 718                 GOTO(out_free, rc = -ENOMEM);
 719
 720         idx_array[0] = le32_to_cpu(lum->lum_stripe_offset);
 721         for (i = 1; i < stripe_count; i++) {
 722                 struct lod_tgt_desc     *tgt;
 723                 struct dt_object        *dto;
 724                 struct lu_fid           fid;
 725                 int                     idx;
 726                 struct lu_object_conf   conf = { 0 };
 727
 728                 idx = (idx_array[i - 1] + 1) % (lod->lod_remote_mdt_count + 1);
 729
 730                 for (j = 0; j < lod->lod_remote_mdt_count;
 731                      j++, idx = (idx + 1) % (lod->lod_remote_mdt_count + 1)) {
 732                         bool already_allocated = false;
 733                         int k;
 734
 735                         CDEBUG(D_INFO, "try idx %d, mdt cnt %d,"
 736                                " allocated %d, last allocated %d\n", idx,
 737                                lod->lod_remote_mdt_count, i, idx_array[i - 1]);
 738
 739                         /* Find next avaible target */
 740                         if (!cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx))
 741                                 continue;
 742
 743                         /* check whether the idx already exists
 744                          * in current allocated array */
 745                         for (k = 0; k < i; k++) {
 746                                 if (idx_array[k] == idx) {
 747                                         already_allocated = true;
 748                                         break;
 749                                 }
 750                         }
 751
 752                         if (already_allocated)
 753                                 continue;
 754
 755                         break;
 756                 }
 757
 758                 /* Can not allocate more stripes */
 759                 if (j == lod->lod_remote_mdt_count) {
 760                         CDEBUG(D_INFO, "%s: require stripes %d only get %d\n",
 761                                lod2obd(lod)->obd_name, stripe_count, i - 1);
 762                         break;
 763                 }
 764
 765                 CDEBUG(D_INFO, "idx %d, mdt cnt %d,"
 766                        " allocated %d, last allocated %d\n", idx,
 767                        lod->lod_remote_mdt_count, i, idx_array[i - 1]);
 768
 769                 tgt = LTD_TGT(ltd, idx);
 770                 LASSERT(tgt != NULL);
 771
 772                 rc = obd_fid_alloc(tgt->ltd_exp, &fid, NULL);
 773                 if (rc < 0)
 774                         GOTO(out_put, rc);
 775                 rc = 0;
 776
 777                 conf.loc_flags = LOC_F_NEW;
 778                 dto = dt_locate_at(env, tgt->ltd_tgt, &fid,
 779                                   dt->do_lu.lo_dev->ld_site->ls_top_dev, &conf);
 780                 if (IS_ERR(dto))
 781                         GOTO(out_put, rc = PTR_ERR(dto));
 782                 stripe[i - 1] = dto;
 783                 idx_array[i] = idx;
 784         }
 785
 786         lo->ldo_dir_striped = 1;
 787         lo->ldo_stripe = stripe;
 788         lo->ldo_stripenr = i - 1;
 789         lo->ldo_stripes_allocated = stripe_count - 1;
 790
 791         if (lo->ldo_stripenr == 0)
 792                 GOTO(out_put, rc = -ENOSPC);
 793
 794         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
 795         if (rc != 0)
 796                 GOTO(out_put, rc);
 797
 798         for (i = 0; i < lo->ldo_stripenr; i++) {
 799                 struct dt_object *dto;
 800
 801                 dto = stripe[i];
 802                 /* only create slave striped object */
 803                 rc = dt_declare_create(env, dto, attr, NULL, NULL, th);
 804                 if (rc != 0)
 805                         GOTO(out_put, rc);
 806
 807                 if (!dt_try_as_dir(env, dto))
 808                         GOTO(out_put, rc = -EINVAL);
 809
 810                 rc = dt_declare_insert(env, dto,
 811                      (const struct dt_rec *)lu_object_fid(&dto->do_lu),
 812                      (const struct dt_key *)dot, th);
 813                 if (rc != 0)
 814                         GOTO(out_put, rc);
 815
 816                 /* master stripe FID will be put to .. */
 817                 rc = dt_declare_insert(env, dto,
 818                      (const struct dt_rec *)lu_object_fid(&dt->do_lu),
 819                      (const struct dt_key *)dotdot, th);
 820                 if (rc != 0)
 821                         GOTO(out_put, rc);
 822
 823                 /* probably nothing to inherite */
 824                 if (lo->ldo_striping_cached &&
 825                     !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
 826                                          lo->ldo_def_stripenr,
 827                                          lo->ldo_def_stripe_offset)) {
 828                         struct lod_thread_info  *info;
 829                         struct lov_user_md_v3   *v3;
 830
 831                         /* sigh, lti_ea_store has been used for lmv_buf,
 832                          * so we have to allocate buffer for default
 833                          * stripe EA */
 834                         OBD_ALLOC_PTR(v3);
 835                         if (v3 == NULL)
 836                                 GOTO(out_put, rc = -ENOMEM);
 837
 838                         memset(v3, 0, sizeof(*v3));
 839                         v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
 840                         v3->lmm_stripe_count =
 841                                 cpu_to_le32(lo->ldo_def_stripenr);
 842                         v3->lmm_stripe_offset =
 843                                 cpu_to_le32(lo->ldo_def_stripe_offset);
 844                         v3->lmm_stripe_size =
 845                                 cpu_to_le32(lo->ldo_def_stripe_size);
 846                         if (lo->ldo_pool)
 847                                 strncpy(v3->lmm_pool_name, lo->ldo_pool,
 848                                         LOV_MAXPOOLNAME);
 849
 850                         info = lod_env_info(env);
 851                         info->lti_buf.lb_buf = v3;
 852                         info->lti_buf.lb_len = sizeof(*v3);
 853                         rc = dt_declare_xattr_set(env, dto,
 854                                                   &info->lti_buf,
 855                                                   XATTR_NAME_LOV,
 856                                                   0, th);
 857                         OBD_FREE_PTR(v3);
 858                         if (rc != 0)
 859                                 GOTO(out_put, rc);
 860                 }
 861                 rc = dt_declare_xattr_set(env, dto, &lmv_buf, XATTR_NAME_LMV, 0,
 862                                           th);
 863                 if (rc != 0)
 864                         GOTO(out_put, rc);
 865         }
 866
 867         rc = dt_declare_xattr_set(env, dt_object_child(dt), &lmv_buf,
 868                                   XATTR_NAME_LMV, 0, th);
 869         if (rc != 0)
 870                 GOTO(out_put, rc);
 871
 872 out_put:
 873         if (rc < 0) {
 874                 for (i = 0; i < stripe_count - 1; i++)
 875                         if (stripe[i] != NULL)
 876                                 lu_object_put(env, &stripe[i]->do_lu);
 877                 OBD_FREE(stripe, sizeof(stripe[0]) * (stripe_count - 1));
 878         }
 879
 880 out_free:
 881         if (idx_array != NULL)
 882                 OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count);
 883
 884         RETURN(rc);
 885 }
 886
 887 /**
 888  * Declare create striped md object.
 889  */
 890 static int lod_declare_xattr_set_lmv(const struct lu_env *env,
 891                                      struct dt_object *dt,
 892                                      struct lu_attr *attr,
 893                                      const struct lu_buf *lum_buf,
 894                                      struct thandle *th)
 895 {
 896         struct lod_object       *lo = lod_dt_obj(dt);
 897         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
 898         struct lmv_user_md_v1   *lum;
 899         int                     rc;
 900         ENTRY;
 901
 902         lum = lum_buf->lb_buf;
 903         LASSERT(lum != NULL);
 904
 905         CDEBUG(D_INFO, "lum magic = %x count = %u offset = %d\n",
 906                le32_to_cpu(lum->lum_magic), le32_to_cpu(lum->lum_stripe_count),
 907                (int)le32_to_cpu(lum->lum_stripe_offset));
 908
 909         if (le32_to_cpu(lum->lum_stripe_count) <= 1)
 910                 GOTO(out, rc = 0);
 911
 912         rc = lod_verify_md_striping(lod, lum);
 913         if (rc != 0)
 914                 GOTO(out, rc);
 915
 916         /* prepare dir striped objects */
 917         rc = lod_prep_md_striped_create(env, dt, attr, lum, th);
 918         if (rc != 0) {
 919                 /* failed to create striping, let's reset
 920                  * config so that others don't get confused */
 921                 lod_object_free_striping(env, lo);
 922                 GOTO(out, rc);
 923         }
 924 out:
 925         RETURN(rc);
 926 }
 927
 928 /*
 929  * LOV xattr is a storage for striping, and LOD owns this xattr.
 930  * but LOD allows others to control striping to some extent
 931  * - to reset strping
 932  * - to set new defined striping
 933  * - to set new semi-defined striping
 934  *   - number of stripes is defined
 935  *   - number of stripes + osts are defined
 936  *   - ??
 937  */
 938 static int lod_declare_xattr_set(const struct lu_env *env,
 939                                  struct dt_object *dt,
 940                                  const struct lu_buf *buf,
 941                                  const char *name, int fl,
 942                                  struct thandle *th)
 943 {
 944         struct dt_object *next = dt_object_child(dt);
 945         struct lu_attr   *attr = &lod_env_info(env)->lti_attr;
 946         __u32             mode;
 947         int               rc;
 948         ENTRY;
 949
 950         /*
 951          * allow to declare predefined striping on a new (!mode) object
 952          * which is supposed to be replay of regular file creation
 953          * (when LOV setting is declared)
 954          * LU_XATTR_REPLACE is set to indicate a layout swap
 955          */
 956         mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
 957         if ((S_ISREG(mode) || mode == 0) && strcmp(name, XATTR_NAME_LOV) == 0 &&
 958              !(fl & LU_XATTR_REPLACE)) {
 959                 /*
 960                  * this is a request to manipulate object's striping
 961                  */
 962                 if (dt_object_exists(dt)) {
 963                         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
 964                         if (rc)
 965                                 RETURN(rc);
 966                 } else {
 967                         memset(attr, 0, sizeof(*attr));
 968                         attr->la_valid = LA_TYPE | LA_MODE;
 969                         attr->la_mode = S_IFREG;
 970                 }
 971                 rc = lod_declare_striped_object(env, dt, attr, buf, th);
 972         } else if (S_ISDIR(mode)) {
 973                 struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
 974                 struct lod_object       *lo = lod_dt_obj(dt);
 975                 int                     i;
 976
 977                 if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
 978                         struct lmv_user_md_v1 *lum;
 979
 980                         LASSERT(buf != NULL && buf->lb_buf != NULL);
 981                         lum = buf->lb_buf;
 982                         rc = lod_verify_md_striping(d, lum);
 983                         if (rc != 0)
 984                                 RETURN(rc);
 985                 }
 986
 987                 rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
 988                 if (rc != 0)
 989                         RETURN(rc);
 990
 991                 /* set xattr to each stripes, if needed */
 992                 rc = lod_load_striping(env, lo);
 993                 if (rc != 0)
 994                         RETURN(rc);
 995
 996                 if (lo->ldo_stripenr == 0)
 997                         RETURN(rc);
 998
 999                 for (i = 0; i < lo->ldo_stripenr; i++) {
1000                         LASSERT(lo->ldo_stripe[i]);
1001                         rc = dt_declare_xattr_set(env, lo->ldo_stripe[i], buf,
1002                                                   name, fl, th);
1003                         if (rc != 0)
1004                                 break;
1005                 }
1006         } else {
1007                 rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
1008         }
1009
1010         RETURN(rc);
1011 }
1012
1013 static void lod_lov_stripe_cache_clear(struct lod_object *lo)
1014 {
1015         lo->ldo_striping_cached = 0;
1016         lo->ldo_def_striping_set = 0;
1017         lod_object_set_pool(lo, NULL);
1018         lo->ldo_def_stripe_size = 0;
1019         lo->ldo_def_stripenr = 0;
1020 }
1021
1022 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
1023                                     struct dt_object *dt,
1024                                     const struct lu_buf *buf,
1025                                     const char *name, int fl,
1026                                     struct thandle *th,
1027                                     struct lustre_capa *capa)
1028 {
1029         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
1030         struct dt_object        *next = dt_object_child(dt);
1031         struct lod_object       *l = lod_dt_obj(dt);
1032         struct lov_user_md_v1   *lum;
1033         struct lov_user_md_v3   *v3 = NULL;
1034         int                      rc;
1035         ENTRY;
1036
1037         /* If it is striped dir, we should clear the stripe cache for
1038          * slave stripe as well, but there are no effective way to
1039          * notify the LOD on the slave MDT, so we do not cache stripe
1040          * information for slave stripe for now. XXX*/
1041         lod_lov_stripe_cache_clear(l);
1042         LASSERT(buf != NULL && buf->lb_buf != NULL);
1043         lum = buf->lb_buf;
1044
1045         rc = lod_verify_striping(d, buf, 0);
1046         if (rc)
1047                 RETURN(rc);
1048
1049         if (lum->lmm_magic == LOV_USER_MAGIC_V3)
1050                 v3 = buf->lb_buf;
1051
1052         /* if { size, offset, count } = { 0, -1, 0 } and no pool
1053          * (i.e. all default values specified) then delete default
1054          * striping from dir. */
1055         CDEBUG(D_OTHER,
1056                 "set default striping: sz %u # %u offset %d %s %s\n",
1057                 (unsigned)lum->lmm_stripe_size,
1058                 (unsigned)lum->lmm_stripe_count,
1059                 (int)lum->lmm_stripe_offset,
1060                 v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
1061
1062         if (LOVEA_DELETE_VALUES((lum->lmm_stripe_size),
1063                                 (lum->lmm_stripe_count),
1064                                 (lum->lmm_stripe_offset)) &&
1065                         lum->lmm_magic == LOV_USER_MAGIC_V1) {
1066                 rc = dt_xattr_del(env, next, name, th, capa);
1067                 if (rc == -ENODATA)
1068                         rc = 0;
1069         } else {
1070                 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
1071         }
1072
1073         RETURN(rc);
1074 }
1075
1076 static int lod_xattr_set_default_lmv_on_dir(const struct lu_env *env,
1077                                             struct dt_object *dt,
1078                                             const struct lu_buf *buf,
1079                                             const char *name, int fl,
1080                                             struct thandle *th,
1081                                             struct lustre_capa *capa)
1082 {
1083         struct dt_object        *next = dt_object_child(dt);
1084         struct lod_object       *l = lod_dt_obj(dt);
1085         struct lmv_user_md_v1   *lum;
1086         int                      rc;
1087         ENTRY;
1088
1089         LASSERT(buf != NULL && buf->lb_buf != NULL);
1090         lum = buf->lb_buf;
1091
1092         CDEBUG(D_OTHER, "set default stripe_count # %u stripe_offset %d\n",
1093               le32_to_cpu(lum->lum_stripe_count),
1094               (int)le32_to_cpu(lum->lum_stripe_offset));
1095
1096         if (LMVEA_DELETE_VALUES((le32_to_cpu(lum->lum_stripe_count)),
1097                                  le32_to_cpu(lum->lum_stripe_offset)) &&
1098                                 le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC) {
1099                 rc = dt_xattr_del(env, next, name, th, capa);
1100                 if (rc == -ENODATA)
1101                         rc = 0;
1102         } else {
1103                 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
1104                 if (rc != 0)
1105                         RETURN(rc);
1106
1107                 /* Update default stripe cache */
1108                 if (l->ldo_dir_stripe == NULL) {
1109                         OBD_ALLOC_PTR(l->ldo_dir_stripe);
1110                         if (l->ldo_dir_stripe == NULL)
1111                                 RETURN(-ENOMEM);
1112                 }
1113
1114                 l->ldo_dir_striping_cached = 0;
1115                 l->ldo_dir_def_striping_set = 1;
1116                 l->ldo_dir_def_stripenr =
1117                         le32_to_cpu(lum->lum_stripe_count) - 1;
1118         }
1119
1120         RETURN(rc);
1121 }
1122
1123 static int lod_xattr_set_lmv(const struct lu_env *env, struct dt_object *dt,
1124                              const struct lu_buf *buf, const char *name,
1125                              int fl, struct thandle *th,
1126                              struct lustre_capa *capa)
1127 {
1128         struct lod_object       *lo = lod_dt_obj(dt);
1129         struct lu_buf           lmv_buf;
1130         int                     i;
1131         int                     rc;
1132         ENTRY;
1133
1134         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
1135                 RETURN(-ENOTDIR);
1136
1137         /* The stripes are supposed to be allocated in declare phase,
1138          * if there are no stripes being allocated, it will skip */
1139         if (lo->ldo_stripenr == 0)
1140                 RETURN(0);
1141
1142         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
1143         if (rc != 0)
1144                 RETURN(rc);
1145
1146         for (i = 0; i < lo->ldo_stripenr; i++) {
1147                 struct dt_object *dto;
1148                 struct lu_attr  *attr = &lod_env_info(env)->lti_attr;
1149
1150                 dto = lo->ldo_stripe[i];
1151                 memset(attr, 0, sizeof(*attr));
1152                 attr->la_valid = LA_TYPE | LA_MODE;
1153                 attr->la_mode = S_IFDIR;
1154                 rc = dt_create(env, dto, attr, NULL, NULL, th);
1155                 if (rc != 0)
1156                         RETURN(rc);
1157
1158                 rc = dt_insert(env, dto,
1159                               (const struct dt_rec *)lu_object_fid(&dto->do_lu),
1160                               (const struct dt_key *)dot, th, capa, 0);
1161                 if (rc != 0)
1162                         RETURN(rc);
1163
1164                 rc = dt_insert(env, dto,
1165                               (struct dt_rec *)lu_object_fid(&dt->do_lu),
1166                               (const struct dt_key *)dotdot, th, capa, 0);
1167                 if (rc != 0)
1168                         RETURN(rc);
1169
1170                 if (lo->ldo_striping_cached &&
1171                     !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
1172                                          lo->ldo_def_stripenr,
1173                                          lo->ldo_def_stripe_offset)) {
1174                         struct lod_thread_info  *info;
1175                         struct lov_user_md_v3   *v3;
1176
1177                         /* sigh, lti_ea_store has been used for lmv_buf,
1178                          * so we have to allocate buffer for default
1179                          * stripe EA */
1180                         OBD_ALLOC_PTR(v3);
1181                         if (v3 == NULL)
1182                                 RETURN(-ENOMEM);
1183
1184                         memset(v3, 0, sizeof(*v3));
1185                         v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
1186                         v3->lmm_stripe_count =
1187                                 cpu_to_le32(lo->ldo_def_stripenr);
1188                         v3->lmm_stripe_offset =
1189                                 cpu_to_le32(lo->ldo_def_stripe_offset);
1190                         v3->lmm_stripe_size =
1191                                 cpu_to_le32(lo->ldo_def_stripe_size);
1192                         if (lo->ldo_pool)
1193                                 strncpy(v3->lmm_pool_name, lo->ldo_pool,
1194                                         LOV_MAXPOOLNAME);
1195
1196                         info = lod_env_info(env);
1197                         info->lti_buf.lb_buf = v3;
1198                         info->lti_buf.lb_len = sizeof(*v3);
1199                         rc = dt_xattr_set(env, dto, &info->lti_buf,
1200                                           XATTR_NAME_LOV, 0, th, capa);
1201                         OBD_FREE_PTR(v3);
1202                         if (rc != 0)
1203                                 RETURN(rc);
1204                 }
1205
1206                 rc = dt_xattr_set(env, dto, &lmv_buf, XATTR_NAME_LMV, fl, th,
1207                                   capa);
1208         }
1209
1210         rc = dt_xattr_set(env, dt_object_child(dt), &lmv_buf, XATTR_NAME_LMV,
1211                           fl, th, capa);
1212
1213         RETURN(rc);
1214 }
1215
1216 static int lod_xattr_set(const struct lu_env *env,
1217                          struct dt_object *dt, const struct lu_buf *buf,
1218                          const char *name, int fl, struct thandle *th,
1219                          struct lustre_capa *capa)
1220 {
1221         struct lod_object       *lo = lod_dt_obj(dt);
1222         struct dt_object        *next = dt_object_child(dt);
1223         __u32                    attr;
1224         int                      rc;
1225         int                     i;
1226         ENTRY;
1227
1228         attr = dt->do_lu.lo_header->loh_attr & S_IFMT;
1229         if (S_ISDIR(attr) && strcmp(name, XATTR_NAME_LOV) == 0) {
1230                 rc = lod_xattr_set_lov_on_dir(env, dt, buf, name, fl, th, capa);
1231         } else if (S_ISREG(attr) && !strcmp(name, XATTR_NAME_LOV)) {
1232                 /* in case of lov EA swap, just set it
1233                  * if not, it is a replay so check striping match what we
1234                  * already have during req replay, declare_xattr_set()
1235                  * defines striping, then create() does the work
1236                 */
1237                 if (fl & LU_XATTR_REPLACE) {
1238                         /* free stripes, then update disk */
1239                         lod_object_free_striping(env, lod_dt_obj(dt));
1240                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
1241                 } else {
1242                         rc = lod_striping_create(env, dt, NULL, NULL, th);
1243                 }
1244         } else if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
1245                 if (!S_ISDIR(attr))
1246                         RETURN(-ENOTDIR);
1247                 rc = lod_xattr_set_default_lmv_on_dir(env, dt, buf, name, fl,
1248                                                       th, capa);
1249         } else {
1250                 /*
1251                  * behave transparantly for all other EAs
1252                  */
1253                 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
1254         }
1255
1256         if (rc != 0 || !S_ISDIR(attr))
1257                 RETURN(rc);
1258
1259         if (lo->ldo_stripenr == 0)
1260                 RETURN(rc);
1261
1262         for (i = 0; i < lo->ldo_stripenr; i++) {
1263                 LASSERT(lo->ldo_stripe[i]);
1264                 rc = dt_xattr_set(env, lo->ldo_stripe[i], buf, name, fl, th,
1265                                   capa);
1266                 if (rc != 0)
1267                         break;
1268         }
1269
1270         RETURN(rc);
1271 }
1272
1273 static int lod_declare_xattr_del(const struct lu_env *env,
1274                                  struct dt_object *dt, const char *name,
1275                                  struct thandle *th)
1276 {
1277         return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
1278 }
1279
1280 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
1281                          const char *name, struct thandle *th,
1282                          struct lustre_capa *capa)
1283 {
1284         if (!strcmp(name, XATTR_NAME_LOV))
1285                 lod_object_free_striping(env, lod_dt_obj(dt));
1286         return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
1287 }
1288
1289 static int lod_xattr_list(const struct lu_env *env,
1290                           struct dt_object *dt, struct lu_buf *buf,
1291                           struct lustre_capa *capa)
1292 {
1293         return dt_xattr_list(env, dt_object_child(dt), buf, capa);
1294 }
1295
1296 int lod_object_set_pool(struct lod_object *o, char *pool)
1297 {
1298         int len;
1299
1300         if (o->ldo_pool) {
1301                 len = strlen(o->ldo_pool);
1302                 OBD_FREE(o->ldo_pool, len + 1);
1303                 o->ldo_pool = NULL;
1304         }
1305         if (pool) {
1306                 len = strlen(pool);
1307                 OBD_ALLOC(o->ldo_pool, len + 1);
1308                 if (o->ldo_pool == NULL)
1309                         return -ENOMEM;
1310                 strcpy(o->ldo_pool, pool);
1311         }
1312         return 0;
1313 }
1314
1315 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
1316 {
1317         return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
1318 }
1319
1320
1321 static int lod_cache_parent_lov_striping(const struct lu_env *env,
1322                                          struct lod_object *lp)
1323 {
1324         struct lod_thread_info  *info = lod_env_info(env);
1325         struct lov_user_md_v1   *v1 = NULL;
1326         struct lov_user_md_v3   *v3 = NULL;
1327         int                      rc;
1328         ENTRY;
1329
1330         /* called from MDD without parent being write locked,
1331          * lock it here */
1332         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
1333         rc = lod_get_lov_ea(env, lp);
1334         if (rc < 0)
1335                 GOTO(unlock, rc);
1336
1337         if (rc < sizeof(struct lov_user_md)) {
1338                 /* don't lookup for non-existing or invalid striping */
1339                 lp->ldo_def_striping_set = 0;
1340                 lp->ldo_striping_cached = 1;
1341                 lp->ldo_def_stripe_size = 0;
1342                 lp->ldo_def_stripenr = 0;
1343                 lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1);
1344                 GOTO(unlock, rc = 0);
1345         }
1346
1347         rc = 0;
1348         v1 = info->lti_ea_store;
1349         if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1))
1350                 lustre_swab_lov_user_md_v1(v1);
1351         else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3))
1352                 lustre_swab_lov_user_md_v3(v3);
1353
1354         if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1)
1355                 GOTO(unlock, rc = 0);
1356
1357         if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
1358                 GOTO(unlock, rc = 0);
1359
1360         CDEBUG(D_INFO, DFID" stripe_count=%d stripe_size=%d stripe_offset=%d\n",
1361                PFID(lu_object_fid(&lp->ldo_obj.do_lu)),
1362                (int)v1->lmm_stripe_count,
1363                (int)v1->lmm_stripe_size, (int)v1->lmm_stripe_offset);
1364
1365         lp->ldo_def_stripenr = v1->lmm_stripe_count;
1366         lp->ldo_def_stripe_size = v1->lmm_stripe_size;
1367         lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
1368         lp->ldo_striping_cached = 1;
1369         lp->ldo_def_striping_set = 1;
1370         if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
1371                 /* XXX: sanity check here */
1372                 v3 = (struct lov_user_md_v3 *) v1;
1373                 if (v3->lmm_pool_name[0])
1374                         lod_object_set_pool(lp, v3->lmm_pool_name);
1375         }
1376         EXIT;
1377 unlock:
1378         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
1379         return rc;
1380 }
1381
1382
1383 static int lod_cache_parent_lmv_striping(const struct lu_env *env,
1384                                          struct lod_object *lp)
1385 {
1386         struct lod_thread_info  *info = lod_env_info(env);
1387         struct lmv_user_md_v1   *v1 = NULL;
1388         int                      rc;
1389         ENTRY;
1390
1391         /* called from MDD without parent being write locked,
1392          * lock it here */
1393         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
1394         rc = lod_get_default_lmv_ea(env, lp);
1395         if (rc < 0)
1396                 GOTO(unlock, rc);
1397
1398         if (rc < sizeof(struct lmv_user_md)) {
1399                 /* don't lookup for non-existing or invalid striping */
1400                 lp->ldo_dir_def_striping_set = 0;
1401                 lp->ldo_dir_striping_cached = 1;
1402                 lp->ldo_dir_def_stripenr = 0;
1403                 lp->ldo_dir_def_stripe_offset =
1404                                         (typeof(v1->lum_stripe_offset))(-1);
1405                 lp->ldo_dir_def_hash_type = LMV_HASH_TYPE_FNV_1A_64;
1406                 GOTO(unlock, rc = 0);
1407         }
1408
1409         rc = 0;
1410         v1 = info->lti_ea_store;
1411
1412         lp->ldo_dir_def_stripenr = le32_to_cpu(v1->lum_stripe_count) - 1;
1413         lp->ldo_dir_def_stripe_offset = le32_to_cpu(v1->lum_stripe_offset);
1414         lp->ldo_dir_def_hash_type = le32_to_cpu(v1->lum_hash_type);
1415         lp->ldo_dir_def_striping_set = 1;
1416         lp->ldo_dir_striping_cached = 1;
1417
1418         EXIT;
1419 unlock:
1420         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
1421         return rc;
1422 }
1423
1424 static int lod_cache_parent_striping(const struct lu_env *env,
1425                                      struct lod_object *lp,
1426                                      umode_t child_mode)
1427 {
1428         int rc = 0;
1429         ENTRY;
1430
1431         rc = lod_load_striping(env, lp);
1432         if (rc != 0)
1433                 RETURN(rc);
1434
1435         if (!lp->ldo_striping_cached) {
1436                 /* we haven't tried to get default striping for
1437                  * the directory yet, let's cache it in the object */
1438                 rc = lod_cache_parent_lov_striping(env, lp);
1439                 if (rc != 0)
1440                         RETURN(rc);
1441         }
1442
1443         if (S_ISDIR(child_mode) && !lp->ldo_dir_striping_cached)
1444                 rc = lod_cache_parent_lmv_striping(env, lp);
1445
1446         RETURN(rc);
1447 }
1448
1449 /**
1450  * used to transfer default striping data to the object being created
1451  */
1452 static void lod_ah_init(const struct lu_env *env,
1453                         struct dt_allocation_hint *ah,
1454                         struct dt_object *parent,
1455                         struct dt_object *child,
1456                         umode_t child_mode)
1457 {
1458         struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
1459         struct dt_object  *nextp = NULL;
1460         struct dt_object  *nextc;
1461         struct lod_object *lp = NULL;
1462         struct lod_object *lc;
1463         struct lov_desc   *desc;
1464         ENTRY;
1465
1466         LASSERT(child);
1467
1468         if (likely(parent)) {
1469                 nextp = dt_object_child(parent);
1470                 lp = lod_dt_obj(parent);
1471         }
1472
1473         nextc = dt_object_child(child);
1474         lc = lod_dt_obj(child);
1475
1476         LASSERT(lc->ldo_stripenr == 0);
1477         LASSERT(lc->ldo_stripe == NULL);
1478
1479         /*
1480          * local object may want some hints
1481          * in case of late striping creation, ->ah_init()
1482          * can be called with local object existing
1483          */
1484         if (!dt_object_exists(nextc) || dt_object_remote(nextc))
1485                 nextc->do_ops->do_ah_init(env, ah, dt_object_remote(nextp) ?
1486                                           NULL : nextp, nextc, child_mode);
1487
1488         if (S_ISDIR(child_mode)) {
1489                 int rc;
1490
1491                 if (lc->ldo_dir_stripe == NULL) {
1492                         OBD_ALLOC_PTR(lc->ldo_dir_stripe);
1493                         if (lc->ldo_dir_stripe == NULL)
1494                                 return;
1495                 }
1496
1497                 if (lp->ldo_dir_stripe == NULL) {
1498                         OBD_ALLOC_PTR(lp->ldo_dir_stripe);
1499                         if (lp->ldo_dir_stripe == NULL)
1500                                 return;
1501                 }
1502
1503                 rc = lod_cache_parent_striping(env, lp, child_mode);
1504                 if (rc != 0)
1505                         return;
1506
1507                 /* transfer defaults to new directory */
1508                 if (lp->ldo_striping_cached) {
1509                         if (lp->ldo_pool)
1510                                 lod_object_set_pool(lc, lp->ldo_pool);
1511                         lc->ldo_def_stripenr = lp->ldo_def_stripenr;
1512                         lc->ldo_def_stripe_size = lp->ldo_def_stripe_size;
1513                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
1514                         lc->ldo_striping_cached = 1;
1515                         lc->ldo_def_striping_set = 1;
1516                         CDEBUG(D_OTHER, "inherite EA sz:%d off:%d nr:%d\n",
1517                                (int)lc->ldo_def_stripe_size,
1518                                (int)lc->ldo_def_stripe_offset,
1519                                (int)lc->ldo_def_stripenr);
1520                 }
1521
1522                 /* transfer dir defaults to new directory */
1523                 if (lp->ldo_dir_striping_cached) {
1524                         lc->ldo_dir_def_stripenr = lp->ldo_dir_def_stripenr;
1525                         lc->ldo_dir_def_stripe_offset =
1526                                                   lp->ldo_dir_def_stripe_offset;
1527                         lc->ldo_dir_def_hash_type =
1528                                                   lp->ldo_dir_def_hash_type;
1529                         lc->ldo_dir_striping_cached = 1;
1530                         lc->ldo_dir_def_striping_set = 1;
1531                         CDEBUG(D_INFO, "inherit default EA nr:%d off:%d t%u\n",
1532                                (int)lc->ldo_dir_def_stripenr,
1533                                (int)lc->ldo_dir_def_stripe_offset,
1534                                lc->ldo_dir_def_hash_type);
1535                 }
1536
1537                 /* If the directory is specified with certain stripes */
1538                 if (ah->dah_eadata != NULL && ah->dah_eadata_len != 0) {
1539                         const struct lmv_user_md_v1 *lum1 = ah->dah_eadata;
1540                         int rc;
1541
1542                         rc = lod_verify_md_striping(d, lum1);
1543                         if (rc == 0 &&
1544                                 le32_to_cpu(lum1->lum_stripe_count) > 1) {
1545                                 /* Directory will be striped only if
1546                                  * stripe_count > 1 */
1547                                 lc->ldo_stripenr =
1548                                         le32_to_cpu(lum1->lum_stripe_count) - 1;
1549                                 lc->ldo_dir_stripe_offset =
1550                                         le32_to_cpu(lum1->lum_stripe_offset);
1551                                 lc->ldo_dir_hash_type =
1552                                         le32_to_cpu(lum1->lum_hash_type);
1553                                 CDEBUG(D_INFO, "set stripe EA nr:%hu off:%d\n",
1554                                        lc->ldo_stripenr,
1555                                        (int)lc->ldo_dir_stripe_offset);
1556                         }
1557                 } else if (lp->ldo_dir_def_striping_set) {
1558                         /* If there are default dir stripe from parent */
1559                         lc->ldo_stripenr = lp->ldo_dir_def_stripenr;
1560                         lc->ldo_dir_stripe_offset =
1561                                         lp->ldo_dir_def_stripe_offset;
1562                         lc->ldo_dir_hash_type =
1563                                         lp->ldo_dir_def_hash_type;
1564                         CDEBUG(D_INFO, "inherit EA nr:%hu off:%d\n",
1565                                lc->ldo_stripenr,
1566                                (int)lc->ldo_dir_stripe_offset);
1567                 } else {
1568                         /* set default stripe for this directory */
1569                         lc->ldo_stripenr = 0;
1570                         lc->ldo_dir_stripe_offset = -1;
1571                 }
1572
1573                 CDEBUG(D_INFO, "final striping count:%hu, offset:%d\n",
1574                        lc->ldo_stripenr, (int)lc->ldo_dir_stripe_offset);
1575
1576                 goto out;
1577         }
1578
1579         /*
1580          * if object is going to be striped over OSTs, transfer default
1581          * striping information to the child, so that we can use it
1582          * during declaration and creation
1583          */
1584         if (!lod_object_will_be_striped(S_ISREG(child_mode),
1585                                         lu_object_fid(&child->do_lu)))
1586                 goto out;
1587         /*
1588          * try from the parent
1589          */
1590         if (likely(parent)) {
1591                 lod_cache_parent_striping(env, lp, child_mode);
1592
1593                 lc->ldo_def_stripe_offset = (__u16) -1;
1594
1595                 if (lp->ldo_def_striping_set) {
1596                         if (lp->ldo_pool)
1597                                 lod_object_set_pool(lc, lp->ldo_pool);
1598                         lc->ldo_stripenr = lp->ldo_def_stripenr;
1599                         lc->ldo_stripe_size = lp->ldo_def_stripe_size;
1600                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
1601                         CDEBUG(D_OTHER, "striping from parent: #%d, sz %d %s\n",
1602                                lc->ldo_stripenr, lc->ldo_stripe_size,
1603                                lp->ldo_pool ? lp->ldo_pool : "");
1604                 }
1605         }
1606
1607         /*
1608          * if the parent doesn't provide with specific pattern, grab fs-wide one
1609          */
1610         desc = &d->lod_desc;
1611         if (lc->ldo_stripenr == 0)
1612                 lc->ldo_stripenr = desc->ld_default_stripe_count;
1613         if (lc->ldo_stripe_size == 0)
1614                 lc->ldo_stripe_size = desc->ld_default_stripe_size;
1615         CDEBUG(D_OTHER, "final striping: # %d stripes, sz %d from %s\n",
1616                lc->ldo_stripenr, lc->ldo_stripe_size,
1617                lc->ldo_pool ? lc->ldo_pool : "");
1618
1619 out:
1620         /* we do not cache stripe information for slave stripe, see
1621          * lod_xattr_set_lov_on_dir */
1622         if (lp != NULL && lp->ldo_dir_slave_stripe)
1623                 lod_lov_stripe_cache_clear(lp);
1624
1625         EXIT;
1626 }
1627
1628 #define ll_do_div64(aaa,bbb)    do_div((aaa), (bbb))
1629 /*
1630  * this function handles a special case when truncate was done
1631  * on a stripeless object and now striping is being created
1632  * we can't lose that size, so we have to propagate it to newly
1633  * created object
1634  */
1635 static int lod_declare_init_size(const struct lu_env *env,
1636                                  struct dt_object *dt, struct thandle *th)
1637 {
1638         struct dt_object   *next = dt_object_child(dt);
1639         struct lod_object  *lo = lod_dt_obj(dt);
1640         struct lu_attr     *attr = &lod_env_info(env)->lti_attr;
1641         uint64_t            size, offs;
1642         int                 rc, stripe;
1643         ENTRY;
1644
1645         /* XXX: we support the simplest (RAID0) striping so far */
1646         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
1647         LASSERT(lo->ldo_stripe_size > 0);
1648
1649         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
1650         LASSERT(attr->la_valid & LA_SIZE);
1651         if (rc)
1652                 RETURN(rc);
1653
1654         size = attr->la_size;
1655         if (size == 0)
1656                 RETURN(0);
1657
1658         /* ll_do_div64(a, b) returns a % b, and a = a / b */
1659         ll_do_div64(size, (__u64) lo->ldo_stripe_size);
1660         stripe = ll_do_div64(size, (__u64) lo->ldo_stripenr);
1661
1662         size = size * lo->ldo_stripe_size;
1663         offs = attr->la_size;
1664         size += ll_do_div64(offs, lo->ldo_stripe_size);
1665
1666         attr->la_valid = LA_SIZE;
1667         attr->la_size = size;
1668
1669         rc = dt_declare_attr_set(env, lo->ldo_stripe[stripe], attr, th);
1670
1671         RETURN(rc);
1672 }
1673
1674 /**
1675  * Create declaration of striped object
1676  */
1677 int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt,
1678                                struct lu_attr *attr,
1679                                const struct lu_buf *lovea, struct thandle *th)
1680 {
1681         struct lod_thread_info  *info = lod_env_info(env);
1682         struct dt_object        *next = dt_object_child(dt);
1683         struct lod_object       *lo = lod_dt_obj(dt);
1684         int                      rc;
1685         ENTRY;
1686
1687         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) {
1688                 /* failed to create striping, let's reset
1689                  * config so that others don't get confused */
1690                 lod_object_free_striping(env, lo);
1691                 GOTO(out, rc = -ENOMEM);
1692         }
1693
1694         if (!dt_object_remote(next)) {
1695                 /* choose OST and generate appropriate objects */
1696                 rc = lod_qos_prep_create(env, lo, attr, lovea, th);
1697                 if (rc) {
1698                         /* failed to create striping, let's reset
1699                          * config so that others don't get confused */
1700                         lod_object_free_striping(env, lo);
1701                         GOTO(out, rc);
1702                 }
1703
1704                 /*
1705                  * declare storage for striping data
1706                  */
1707                 info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
1708                                 lo->ldo_pool ?  LOV_MAGIC_V3 : LOV_MAGIC_V1);
1709         } else {
1710                 /* LOD can not choose OST objects for remote objects, i.e.
1711                  * stripes must be ready before that. Right now, it can only
1712                  * happen during migrate, i.e. migrate process needs to create
1713                  * remote regular file (mdd_migrate_create), then the migrate
1714                  * process will provide stripeEA. */
1715                 LASSERT(lovea != NULL);
1716                 info->lti_buf = *lovea;
1717         }
1718
1719         rc = dt_declare_xattr_set(env, next, &info->lti_buf,
1720                                   XATTR_NAME_LOV, 0, th);
1721         if (rc)
1722                 GOTO(out, rc);
1723
1724         /*
1725          * if striping is created with local object's size > 0,
1726          * we have to propagate this size to specific object
1727          * the case is possible only when local object was created previously
1728          */
1729         if (dt_object_exists(next))
1730                 rc = lod_declare_init_size(env, dt, th);
1731
1732 out:
1733         RETURN(rc);
1734 }
1735
1736 int lod_dir_striping_create_internal(const struct lu_env *env,
1737                                      struct dt_object *dt,
1738                                      struct lu_attr *attr,
1739                                      const struct dt_object_format *dof,
1740                                      struct thandle *th,
1741                                      bool declare)
1742 {
1743         struct lod_thread_info  *info = lod_env_info(env);
1744         struct dt_object        *next = dt_object_child(dt);
1745         struct lod_object       *lo = lod_dt_obj(dt);
1746         int                     rc;
1747         ENTRY;
1748
1749         if (lo->ldo_dir_def_striping_set &&
1750             !LMVEA_DELETE_VALUES(lo->ldo_stripenr,
1751                                  lo->ldo_dir_stripe_offset)) {
1752                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
1753                 int stripe_count = lo->ldo_stripenr + 1;
1754
1755                 if (info->lti_ea_store_size < sizeof(*v1)) {
1756                         rc = lod_ea_store_resize(info, sizeof(*v1));
1757                         if (rc != 0)
1758                                 RETURN(rc);
1759                         v1 = info->lti_ea_store;
1760                 }
1761
1762                 memset(v1, 0, sizeof(*v1));
1763                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
1764                 v1->lum_stripe_count = cpu_to_le32(stripe_count);
1765                 v1->lum_stripe_offset =
1766                                 cpu_to_le32(lo->ldo_dir_stripe_offset);
1767
1768                 info->lti_buf.lb_buf = v1;
1769                 info->lti_buf.lb_len = sizeof(*v1);
1770
1771                 if (declare)
1772                         rc = lod_declare_xattr_set_lmv(env, dt, attr,
1773                                                        &info->lti_buf, th);
1774                 else
1775                         rc = lod_xattr_set_lmv(env, dt, &info->lti_buf,
1776                                                XATTR_NAME_LMV, 0, th,
1777                                                BYPASS_CAPA);
1778                 if (rc != 0)
1779                         RETURN(rc);
1780         }
1781
1782         /* Transfer default LMV striping from the parent */
1783         if (lo->ldo_dir_striping_cached &&
1784             !LMVEA_DELETE_VALUES(lo->ldo_dir_def_stripenr,
1785                                  lo->ldo_dir_def_stripe_offset)) {
1786                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
1787                 int def_stripe_count = lo->ldo_dir_def_stripenr + 1;
1788
1789                 if (info->lti_ea_store_size < sizeof(*v1)) {
1790                         rc = lod_ea_store_resize(info, sizeof(*v1));
1791                         if (rc != 0)
1792                                 RETURN(rc);
1793                         v1 = info->lti_ea_store;
1794                 }
1795
1796                 memset(v1, 0, sizeof(*v1));
1797                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
1798                 v1->lum_stripe_count = cpu_to_le32(def_stripe_count);
1799                 v1->lum_stripe_offset =
1800                                 cpu_to_le32(lo->ldo_dir_def_stripe_offset);
1801                 v1->lum_hash_type =
1802                                 cpu_to_le32(lo->ldo_dir_def_hash_type);
1803
1804                 info->lti_buf.lb_buf = v1;
1805                 info->lti_buf.lb_len = sizeof(*v1);
1806                 if (declare)
1807                         rc = dt_declare_xattr_set(env, next, &info->lti_buf,
1808                                                   XATTR_NAME_DEFAULT_LMV, 0,
1809                                                   th);
1810                 else
1811                         rc = dt_xattr_set(env, next, &info->lti_buf,
1812                                            XATTR_NAME_DEFAULT_LMV, 0, th,
1813                                            BYPASS_CAPA);
1814                 if (rc != 0)
1815                         RETURN(rc);
1816         }
1817
1818         /* Transfer default LOV striping from the parent */
1819         if (lo->ldo_striping_cached &&
1820             !LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
1821                                  lo->ldo_def_stripenr,
1822                                  lo->ldo_def_stripe_offset)) {
1823                 struct lov_user_md_v3 *v3 = info->lti_ea_store;
1824
1825                 if (info->lti_ea_store_size < sizeof(*v3)) {
1826                         rc = lod_ea_store_resize(info, sizeof(*v3));
1827                         if (rc != 0)
1828                                 RETURN(rc);
1829                         v3 = info->lti_ea_store;
1830                 }
1831
1832                 memset(v3, 0, sizeof(*v3));
1833                 v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
1834                 v3->lmm_stripe_count = cpu_to_le16(lo->ldo_def_stripenr);
1835                 v3->lmm_stripe_offset = cpu_to_le16(lo->ldo_def_stripe_offset);
1836                 v3->lmm_stripe_size = cpu_to_le32(lo->ldo_def_stripe_size);
1837                 if (lo->ldo_pool)
1838                         strncpy(v3->lmm_pool_name, lo->ldo_pool,
1839                                 LOV_MAXPOOLNAME);
1840
1841                 info->lti_buf.lb_buf = v3;
1842                 info->lti_buf.lb_len = sizeof(*v3);
1843
1844                 if (declare)
1845                         rc = dt_declare_xattr_set(env, next, &info->lti_buf,
1846                                                   XATTR_NAME_LOV, 0, th);
1847                 else
1848                         rc = dt_xattr_set(env, next, &info->lti_buf,
1849                                           XATTR_NAME_LOV, 0, th,
1850                                           BYPASS_CAPA);
1851                 if (rc != 0)
1852                         RETURN(rc);
1853         }
1854
1855         RETURN(0);
1856 }
1857
1858 static int lod_declare_dir_striping_create(const struct lu_env *env,
1859                                            struct dt_object *dt,
1860                                            struct lu_attr *attr,
1861                                            struct dt_object_format *dof,
1862                                            struct thandle *th)
1863 {
1864         return lod_dir_striping_create_internal(env, dt, attr, dof, th, true);
1865 }
1866
1867 static int lod_dir_striping_create(const struct lu_env *env,
1868                                    struct dt_object *dt,
1869                                    struct lu_attr *attr,
1870                                    struct dt_object_format *dof,
1871                                    struct thandle *th)
1872 {
1873         return lod_dir_striping_create_internal(env, dt, attr, dof, th, false);
1874 }
1875
1876 static int lod_declare_object_create(const struct lu_env *env,
1877                                      struct dt_object *dt,
1878                                      struct lu_attr *attr,
1879                                      struct dt_allocation_hint *hint,
1880                                      struct dt_object_format *dof,
1881                                      struct thandle *th)
1882 {
1883         struct dt_object   *next = dt_object_child(dt);
1884         struct lod_object  *lo = lod_dt_obj(dt);
1885         int                 rc;
1886         ENTRY;
1887
1888         LASSERT(dof);
1889         LASSERT(attr);
1890         LASSERT(th);
1891
1892         /*
1893          * first of all, we declare creation of local object
1894          */
1895         rc = dt_declare_create(env, next, attr, hint, dof, th);
1896         if (rc)
1897                 GOTO(out, rc);
1898
1899         if (dof->dof_type == DFT_SYM)
1900                 dt->do_body_ops = &lod_body_lnk_ops;
1901
1902         /*
1903          * it's lod_ah_init() who has decided the object will striped
1904          */
1905         if (dof->dof_type == DFT_REGULAR) {
1906                 /* callers don't want stripes */
1907                 /* XXX: all tricky interactions with ->ah_make_hint() decided
1908                  * to use striping, then ->declare_create() behaving differently
1909                  * should be cleaned */
1910                 if (dof->u.dof_reg.striped == 0)
1911                         lo->ldo_stripenr = 0;
1912                 if (lo->ldo_stripenr > 0)
1913                         rc = lod_declare_striped_object(env, dt, attr,
1914                                                         NULL, th);
1915         } else if (dof->dof_type == DFT_DIR) {
1916                 /* Orphan object (like migrating object) does not have
1917                  * lod_dir_stripe, see lod_ah_init */
1918                 if (lo->ldo_dir_stripe != NULL)
1919                         rc = lod_declare_dir_striping_create(env, dt, attr,
1920                                                              dof, th);
1921         }
1922 out:
1923         RETURN(rc);
1924 }
1925
1926 int lod_striping_create(const struct lu_env *env, struct dt_object *dt,
1927                         struct lu_attr *attr, struct dt_object_format *dof,
1928                         struct thandle *th)
1929 {
1930         struct lod_object *lo = lod_dt_obj(dt);
1931         int                rc = 0, i;
1932         ENTRY;
1933
1934         LASSERT(lo->ldo_striping_cached == 0);
1935
1936         /* create all underlying objects */
1937         for (i = 0; i < lo->ldo_stripenr; i++) {
1938                 LASSERT(lo->ldo_stripe[i]);
1939                 rc = dt_create(env, lo->ldo_stripe[i], attr, NULL, dof, th);
1940
1941                 if (rc)
1942                         break;
1943         }
1944         if (rc == 0)
1945                 rc = lod_generate_and_set_lovea(env, lo, th);
1946
1947         RETURN(rc);
1948 }
1949
1950 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
1951                              struct lu_attr *attr,
1952                              struct dt_allocation_hint *hint,
1953                              struct dt_object_format *dof, struct thandle *th)
1954 {
1955         struct dt_object   *next = dt_object_child(dt);
1956         struct lod_object  *lo = lod_dt_obj(dt);
1957         int                 rc;
1958         ENTRY;
1959
1960         /* create local object */
1961         rc = dt_create(env, next, attr, hint, dof, th);
1962
1963         if (rc == 0) {
1964                 if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
1965                     lo->ldo_dir_stripe != NULL)
1966                         rc = lod_dir_striping_create(env, dt, attr, dof, th);
1967                 else if (lo->ldo_stripe && dof->u.dof_reg.striped != 0)
1968                         rc = lod_striping_create(env, dt, attr, dof, th);
1969         }
1970
1971         RETURN(rc);
1972 }
1973
1974 static int lod_declare_object_destroy(const struct lu_env *env,
1975                                       struct dt_object *dt,
1976                                       struct thandle *th)
1977 {
1978         struct dt_object   *next = dt_object_child(dt);
1979         struct lod_object  *lo = lod_dt_obj(dt);
1980         int                 rc, i;
1981         ENTRY;
1982
1983         /*
1984          * we declare destroy for the local object
1985          */
1986         rc = dt_declare_destroy(env, next, th);
1987         if (rc)
1988                 RETURN(rc);
1989
1990         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ))
1991                 RETURN(0);
1992
1993         /*
1994          * load striping information, notice we don't do this when object
1995          * is being initialized as we don't need this information till
1996          * few specific cases like destroy, chown
1997          */
1998         rc = lod_load_striping(env, lo);
1999         if (rc)
2000                 RETURN(rc);
2001
2002         /* declare destroy for all underlying objects */
2003         for (i = 0; i < lo->ldo_stripenr; i++) {
2004                 LASSERT(lo->ldo_stripe[i]);
2005                 rc = dt_declare_destroy(env, lo->ldo_stripe[i], th);
2006
2007                 if (rc)
2008                         break;
2009         }
2010
2011         RETURN(rc);
2012 }
2013
2014 static int lod_object_destroy(const struct lu_env *env,
2015                 struct dt_object *dt, struct thandle *th)
2016 {
2017         struct dt_object  *next = dt_object_child(dt);
2018         struct lod_object *lo = lod_dt_obj(dt);
2019         int                rc, i;
2020         ENTRY;
2021
2022         /* destroy local object */
2023         rc = dt_destroy(env, next, th);
2024         if (rc)
2025                 RETURN(rc);
2026
2027         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ))
2028                 RETURN(0);
2029
2030         /* destroy all underlying objects */
2031         for (i = 0; i < lo->ldo_stripenr; i++) {
2032                 LASSERT(lo->ldo_stripe[i]);
2033                 /* for striped directory, next == ldo_stripe[0] */
2034                 if (next != lo->ldo_stripe[i]) {
2035                         rc = dt_destroy(env, lo->ldo_stripe[i], th);
2036                         if (rc)
2037                                 break;
2038                 }
2039         }
2040
2041         RETURN(rc);
2042 }
2043
2044 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
2045                          const struct dt_index_features *feat)
2046 {
2047         struct dt_object *next = dt_object_child(dt);
2048         int               rc;
2049         ENTRY;
2050
2051         LASSERT(next->do_ops);
2052         LASSERT(next->do_ops->do_index_try);
2053
2054         rc = next->do_ops->do_index_try(env, next, feat);
2055         if (next->do_index_ops && dt->do_index_ops == NULL)
2056                 dt->do_index_ops = &lod_index_ops;
2057
2058         RETURN(rc);
2059 }
2060
2061 static int lod_declare_ref_add(const struct lu_env *env,
2062                                struct dt_object *dt, struct thandle *th)
2063 {
2064         return dt_declare_ref_add(env, dt_object_child(dt), th);
2065 }
2066
2067 static int lod_ref_add(const struct lu_env *env,
2068                        struct dt_object *dt, struct thandle *th)
2069 {
2070         return dt_ref_add(env, dt_object_child(dt), th);
2071 }
2072
2073 static int lod_declare_ref_del(const struct lu_env *env,
2074                                struct dt_object *dt, struct thandle *th)
2075 {
2076         return dt_declare_ref_del(env, dt_object_child(dt), th);
2077 }
2078
2079 static int lod_ref_del(const struct lu_env *env,
2080                        struct dt_object *dt, struct thandle *th)
2081 {
2082         return dt_ref_del(env, dt_object_child(dt), th);
2083 }
2084
2085 static struct obd_capa *lod_capa_get(const struct lu_env *env,
2086                                      struct dt_object *dt,
2087                                      struct lustre_capa *old, __u64 opc)
2088 {
2089         return dt_capa_get(env, dt_object_child(dt), old, opc);
2090 }
2091
2092 static int lod_object_sync(const struct lu_env *env, struct dt_object *dt)
2093 {
2094         return dt_object_sync(env, dt_object_child(dt));
2095 }
2096
2097 struct lod_slave_locks  {
2098         int                     lsl_lock_count;
2099         struct lustre_handle    lsl_handle[0];
2100 };
2101
2102 static int lod_object_unlock_internal(const struct lu_env *env,
2103                                       struct dt_object *dt,
2104                                       struct ldlm_enqueue_info *einfo,
2105                                       ldlm_policy_data_t *policy)
2106 {
2107         struct lod_object       *lo = lod_dt_obj(dt);
2108         struct lod_slave_locks  *slave_locks = einfo->ei_cbdata;
2109         int                     rc = 0;
2110         int                     i;
2111         ENTRY;
2112
2113         if (slave_locks == NULL)
2114                 RETURN(0);
2115
2116         for (i = 0; i < slave_locks->lsl_lock_count; i++) {
2117                 if (lustre_handle_is_used(&slave_locks->lsl_handle[i])) {
2118                         int     rc1;
2119
2120                         einfo->ei_cbdata = &slave_locks->lsl_handle[i];
2121                         rc1 = dt_object_unlock(env, lo->ldo_stripe[i], einfo,
2122                                                policy);
2123                         if (rc1 < 0)
2124                                 rc = rc == 0 ? rc1 : rc;
2125                 }
2126         }
2127
2128         RETURN(rc);
2129 }
2130
2131 static int lod_object_unlock(const struct lu_env *env, struct dt_object *dt,
2132                              struct ldlm_enqueue_info *einfo,
2133                              union ldlm_policy_data *policy)
2134 {
2135         struct lod_object       *lo = lod_dt_obj(dt);
2136         struct lod_slave_locks  *slave_locks = einfo->ei_cbdata;
2137         int                     slave_locks_size;
2138         int                     rc;
2139         ENTRY;
2140
2141         if (slave_locks == NULL)
2142                 RETURN(0);
2143
2144         rc = lod_load_striping(env, lo);
2145         if (rc != 0)
2146                 RETURN(rc);
2147
2148         /* Note: for remote lock for single stripe dir, MDT will cancel
2149          * the lock by lockh directly */
2150         if (lo->ldo_stripenr == 0 && dt_object_remote(dt_object_child(dt)))
2151                 RETURN(0);
2152
2153         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
2154                 RETURN(-ENOTDIR);
2155
2156         /* Only cancel slave lock for striped dir */
2157         rc = lod_object_unlock_internal(env, dt, einfo, policy);
2158
2159         slave_locks_size = sizeof(*slave_locks) + slave_locks->lsl_lock_count *
2160                            sizeof(slave_locks->lsl_handle[0]);
2161         OBD_FREE(slave_locks, slave_locks_size);
2162         einfo->ei_cbdata = NULL;
2163
2164         RETURN(rc);
2165 }
2166
2167 static int lod_object_lock(const struct lu_env *env,
2168                            struct dt_object *dt,
2169                            struct lustre_handle *lh,
2170                            struct ldlm_enqueue_info *einfo,
2171                            union ldlm_policy_data *policy)
2172 {
2173         struct lod_object       *lo = lod_dt_obj(dt);
2174         int                     rc = 0;
2175         int                     i;
2176         int                     slave_locks_size;
2177         struct lod_slave_locks  *slave_locks = NULL;
2178         ENTRY;
2179
2180         /* remote object lock */
2181         if (!einfo->ei_enq_slave) {
2182                 LASSERT(dt_object_remote(dt));
2183                 return dt_object_lock(env, dt_object_child(dt), lh, einfo,
2184                                       policy);
2185         }
2186
2187         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
2188                 RETURN(-ENOTDIR);
2189
2190         rc = lod_load_striping(env, lo);
2191         if (rc != 0)
2192                 RETURN(rc);
2193
2194         /* No stripes */
2195         if (lo->ldo_stripenr == 0)
2196                 RETURN(0);
2197
2198         slave_locks_size = sizeof(*slave_locks) + lo->ldo_stripenr *
2199                            sizeof(slave_locks->lsl_handle[0]);
2200         /* Freed in lod_object_unlock */
2201         OBD_ALLOC(slave_locks, slave_locks_size);
2202         if (slave_locks == NULL)
2203                 RETURN(-ENOMEM);
2204         slave_locks->lsl_lock_count = lo->ldo_stripenr;
2205
2206         /* striped directory lock */
2207         for (i = 0; i < lo->ldo_stripenr; i++) {
2208                 struct lustre_handle    lockh;
2209                 struct ldlm_res_id      *res_id;
2210
2211                 res_id = &lod_env_info(env)->lti_res_id;
2212                 fid_build_reg_res_name(lu_object_fid(&lo->ldo_stripe[i]->do_lu),
2213                                        res_id);
2214                 einfo->ei_res_id = res_id;
2215
2216                 LASSERT(lo->ldo_stripe[i]);
2217                 rc = dt_object_lock(env, lo->ldo_stripe[i], &lockh, einfo,
2218                                     policy);
2219                 if (rc != 0)
2220                         GOTO(out, rc);
2221
2222                 slave_locks->lsl_handle[i] = lockh;
2223         }
2224
2225         einfo->ei_cbdata = slave_locks;
2226
2227 out:
2228         if (rc != 0 && slave_locks != NULL) {
2229                 einfo->ei_cbdata = slave_locks;
2230                 lod_object_unlock_internal(env, dt, einfo, policy);
2231                 OBD_FREE(slave_locks, slave_locks_size);
2232                 einfo->ei_cbdata = NULL;
2233         }
2234
2235         RETURN(rc);
2236 }
2237
2238 struct dt_object_operations lod_obj_ops = {
2239         .do_read_lock           = lod_object_read_lock,
2240         .do_write_lock          = lod_object_write_lock,
2241         .do_read_unlock         = lod_object_read_unlock,
2242         .do_write_unlock        = lod_object_write_unlock,
2243         .do_write_locked        = lod_object_write_locked,
2244         .do_attr_get            = lod_attr_get,
2245         .do_declare_attr_set    = lod_declare_attr_set,
2246         .do_attr_set            = lod_attr_set,
2247         .do_xattr_get           = lod_xattr_get,
2248         .do_declare_xattr_set   = lod_declare_xattr_set,
2249         .do_xattr_set           = lod_xattr_set,
2250         .do_declare_xattr_del   = lod_declare_xattr_del,
2251         .do_xattr_del           = lod_xattr_del,
2252         .do_xattr_list          = lod_xattr_list,
2253         .do_ah_init             = lod_ah_init,
2254         .do_declare_create      = lod_declare_object_create,
2255         .do_create              = lod_object_create,
2256         .do_declare_destroy     = lod_declare_object_destroy,
2257         .do_destroy             = lod_object_destroy,
2258         .do_index_try           = lod_index_try,
2259         .do_declare_ref_add     = lod_declare_ref_add,
2260         .do_ref_add             = lod_ref_add,
2261         .do_declare_ref_del     = lod_declare_ref_del,
2262         .do_ref_del             = lod_ref_del,
2263         .do_capa_get            = lod_capa_get,
2264         .do_object_sync         = lod_object_sync,
2265         .do_object_lock         = lod_object_lock,
2266         .do_object_unlock       = lod_object_unlock,
2267 };
2268
2269 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
2270                         struct lu_buf *buf, loff_t *pos,
2271                         struct lustre_capa *capa)
2272 {
2273         struct dt_object *next = dt_object_child(dt);
2274         return next->do_body_ops->dbo_read(env, next, buf, pos, capa);
2275 }
2276
2277 static ssize_t lod_declare_write(const struct lu_env *env,
2278                                  struct dt_object *dt,
2279                                  const struct lu_buf *buf, loff_t pos,
2280                                  struct thandle *th)
2281 {
2282         return dt_declare_record_write(env, dt_object_child(dt),
2283                                        buf, pos, th);
2284 }
2285
2286 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,
2287                          const struct lu_buf *buf, loff_t *pos,
2288                          struct thandle *th, struct lustre_capa *capa, int iq)
2289 {
2290         struct dt_object *next = dt_object_child(dt);
2291         LASSERT(next);
2292         return next->do_body_ops->dbo_write(env, next, buf, pos, th, capa, iq);
2293 }
2294
2295 static const struct dt_body_operations lod_body_lnk_ops = {
2296         .dbo_read               = lod_read,
2297         .dbo_declare_write      = lod_declare_write,
2298         .dbo_write              = lod_write
2299 };
2300
2301 static int lod_object_init(const struct lu_env *env, struct lu_object *lo,
2302                            const struct lu_object_conf *conf)
2303 {
2304         struct lod_device       *lod    = lu2lod_dev(lo->lo_dev);
2305         struct lu_device        *cdev   = NULL;
2306         struct lu_object        *cobj;
2307         struct lod_tgt_descs    *ltd    = NULL;
2308         struct lod_tgt_desc     *tgt;
2309         mdsno_t                  idx    = 0;
2310         int                      type   = LU_SEQ_RANGE_ANY;
2311         int                      rc;
2312         ENTRY;
2313
2314         rc = lod_fld_lookup(env, lod, lu_object_fid(lo), &idx, &type);
2315         if (rc != 0)
2316                 RETURN(rc);
2317
2318         if (type == LU_SEQ_RANGE_MDT &&
2319             idx == lu_site2seq(lo->lo_dev->ld_site)->ss_node_id) {
2320                 cdev = &lod->lod_child->dd_lu_dev;
2321         } else if (type == LU_SEQ_RANGE_MDT) {
2322                 ltd = &lod->lod_mdt_descs;
2323                 lod_getref(ltd);
2324         } else if (type == LU_SEQ_RANGE_OST) {
2325                 ltd = &lod->lod_ost_descs;
2326                 lod_getref(ltd);
2327         } else {
2328                 LBUG();
2329         }
2330
2331         if (ltd != NULL) {
2332                 if (ltd->ltd_tgts_size > idx &&
2333                     cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx)) {
2334                         tgt = LTD_TGT(ltd, idx);
2335
2336                         LASSERT(tgt != NULL);
2337                         LASSERT(tgt->ltd_tgt != NULL);
2338
2339                         cdev = &(tgt->ltd_tgt->dd_lu_dev);
2340                 }
2341                 lod_putref(lod, ltd);
2342         }
2343
2344         if (unlikely(cdev == NULL))
2345                 RETURN(-ENOENT);
2346
2347         cobj = cdev->ld_ops->ldo_object_alloc(env, lo->lo_header, cdev);
2348         if (unlikely(cobj == NULL))
2349                 RETURN(-ENOMEM);
2350
2351         lu_object_add(lo, cobj);
2352
2353         RETURN(0);
2354 }
2355
2356 void lod_object_free_striping(const struct lu_env *env, struct lod_object *lo)
2357 {
2358         int i;
2359
2360         if (lo->ldo_dir_stripe != NULL) {
2361                 OBD_FREE_PTR(lo->ldo_dir_stripe);
2362                 lo->ldo_dir_stripe = NULL;
2363         }
2364
2365         if (lo->ldo_stripe) {
2366                 LASSERT(lo->ldo_stripes_allocated > 0);
2367
2368                 for (i = 0; i < lo->ldo_stripenr; i++) {
2369                         if (lo->ldo_stripe[i])
2370                                 lu_object_put(env, &lo->ldo_stripe[i]->do_lu);
2371                 }
2372
2373                 i = sizeof(struct dt_object *) * lo->ldo_stripes_allocated;
2374                 OBD_FREE(lo->ldo_stripe, i);
2375                 lo->ldo_stripe = NULL;
2376                 lo->ldo_stripes_allocated = 0;
2377         }
2378         lo->ldo_stripenr = 0;
2379         lo->ldo_pattern = 0;
2380 }
2381
2382 /*
2383  * ->start is called once all slices are initialized, including header's
2384  * cache for mode (object type). using the type we can initialize ops
2385  */
2386 static int lod_object_start(const struct lu_env *env, struct lu_object *o)
2387 {
2388         if (S_ISLNK(o->lo_header->loh_attr & S_IFMT))
2389                 lu2lod_obj(o)->ldo_obj.do_body_ops = &lod_body_lnk_ops;
2390         return 0;
2391 }
2392
2393 static void lod_object_free(const struct lu_env *env, struct lu_object *o)
2394 {
2395         struct lod_object *mo = lu2lod_obj(o);
2396
2397         /*
2398          * release all underlying object pinned
2399          */
2400
2401         lod_object_free_striping(env, mo);
2402
2403         lod_object_set_pool(mo, NULL);
2404
2405         lu_object_fini(o);
2406         OBD_SLAB_FREE_PTR(mo, lod_object_kmem);
2407 }
2408
2409 static void lod_object_release(const struct lu_env *env, struct lu_object *o)
2410 {
2411         /* XXX: shouldn't we release everything here in case if object
2412          * creation failed before? */
2413 }
2414
2415 static int lod_object_print(const struct lu_env *env, void *cookie,
2416                             lu_printer_t p, const struct lu_object *l)
2417 {
2418         struct lod_object *o = lu2lod_obj((struct lu_object *) l);
2419
2420         return (*p)(env, cookie, LUSTRE_LOD_NAME"-object@%p", o);
2421 }
2422
2423 struct lu_object_operations lod_lu_obj_ops = {
2424         .loo_object_init        = lod_object_init,
2425         .loo_object_start       = lod_object_start,
2426         .loo_object_free        = lod_object_free,
2427         .loo_object_release     = lod_object_release,
2428         .loo_object_print       = lod_object_print,
2429 };