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