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, 2017, Intel Corporation.
  27  */
  28 /*
  29  * lustre/lod/lod_object.c
  30  *
  31  * This file contains implementations of methods for the OSD API
  32  * for the Logical Object Device (LOD) layer, which provides a virtual
  33  * local OSD object interface to the MDD layer, and abstracts the
  34  * addressing of local (OSD) and remote (OSP) objects. The API is
  35  * described in the file lustre/include/dt_object.h and in
  36  * Documentation/osd-api.txt.
  37  *
  38  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  39  */
  40
  41 #define DEBUG_SUBSYSTEM S_MDS
  42
  43 #include <linux/random.h>
  44
  45 #include <obd.h>
  46 #include <obd_class.h>
  47 #include <obd_support.h>
  48
  49 #include <lustre_fid.h>
  50 #include <lustre_linkea.h>
  51 #include <lustre_lmv.h>
  52 #include <uapi/linux/lustre/lustre_param.h>
  53 #include <lustre_swab.h>
  54 #include <uapi/linux/lustre/lustre_ver.h>
  55 #include <lprocfs_status.h>
  56 #include <md_object.h>
  57
  58 #include "lod_internal.h"
  59
  60 static const char dot[] = ".";
  61 static const char dotdot[] = "..";
  62
  63 /**
  64  * Implementation of dt_index_operations::dio_lookup
  65  *
  66  * Used with regular (non-striped) objects.
  67  *
  68  * \see dt_index_operations::dio_lookup() in the API description for details.
  69  */
  70 static int lod_lookup(const struct lu_env *env, struct dt_object *dt,
  71                       struct dt_rec *rec, const struct dt_key *key)
  72 {
  73         struct dt_object *next = dt_object_child(dt);
  74         return next->do_index_ops->dio_lookup(env, next, rec, key);
  75 }
  76
  77 /**
  78  * Implementation of dt_index_operations::dio_declare_insert.
  79  *
  80  * Used with regular (non-striped) objects.
  81  *
  82  * \see dt_index_operations::dio_declare_insert() in the API description
  83  * for details.
  84  */
  85 static int lod_declare_insert(const struct lu_env *env, struct dt_object *dt,
  86                               const struct dt_rec *rec,
  87                               const struct dt_key *key, struct thandle *th)
  88 {
  89         return lod_sub_declare_insert(env, dt_object_child(dt), rec, key, th);
  90 }
  91
  92 /**
  93  * Implementation of dt_index_operations::dio_insert.
  94  *
  95  * Used with regular (non-striped) objects
  96  *
  97  * \see dt_index_operations::dio_insert() in the API description for details.
  98  */
  99 static int lod_insert(const struct lu_env *env, struct dt_object *dt,
 100                       const struct dt_rec *rec, const struct dt_key *key,
 101                       struct thandle *th, int ign)
 102 {
 103         return lod_sub_insert(env, dt_object_child(dt), rec, key, th, ign);
 104 }
 105
 106 /**
 107  * Implementation of dt_index_operations::dio_declare_delete.
 108  *
 109  * Used with regular (non-striped) objects.
 110  *
 111  * \see dt_index_operations::dio_declare_delete() in the API description
 112  * for details.
 113  */
 114 static int lod_declare_delete(const struct lu_env *env, struct dt_object *dt,
 115                               const struct dt_key *key, struct thandle *th)
 116 {
 117         return lod_sub_declare_delete(env, dt_object_child(dt), key, th);
 118 }
 119
 120 /**
 121  * Implementation of dt_index_operations::dio_delete.
 122  *
 123  * Used with regular (non-striped) objects.
 124  *
 125  * \see dt_index_operations::dio_delete() in the API description for details.
 126  */
 127 static int lod_delete(const struct lu_env *env, struct dt_object *dt,
 128                       const struct dt_key *key, struct thandle *th)
 129 {
 130         return lod_sub_delete(env, dt_object_child(dt), key, th);
 131 }
 132
 133 /**
 134  * Implementation of dt_it_ops::init.
 135  *
 136  * Used with regular (non-striped) objects.
 137  *
 138  * \see dt_it_ops::init() in the API description for details.
 139  */
 140 static struct dt_it *lod_it_init(const struct lu_env *env,
 141                                  struct dt_object *dt, __u32 attr)
 142 {
 143         struct dt_object        *next = dt_object_child(dt);
 144         struct lod_it           *it = &lod_env_info(env)->lti_it;
 145         struct dt_it            *it_next;
 146
 147         it_next = next->do_index_ops->dio_it.init(env, next, attr);
 148         if (IS_ERR(it_next))
 149                 return it_next;
 150
 151         /* currently we do not use more than one iterator per thread
 152          * so we store it in thread info. if at some point we need
 153          * more active iterators in a single thread, we can allocate
 154          * additional ones */
 155         LASSERT(it->lit_obj == NULL);
 156
 157         it->lit_it = it_next;
 158         it->lit_obj = next;
 159
 160         return (struct dt_it *)it;
 161 }
 162
 163 #define LOD_CHECK_IT(env, it)                                   \
 164 do {                                                            \
 165         LASSERT((it)->lit_obj != NULL);                         \
 166         LASSERT((it)->lit_it != NULL);                          \
 167 } while (0)
 168
 169 /**
 170  * Implementation of dt_index_operations::dio_it.fini.
 171  *
 172  * Used with regular (non-striped) objects.
 173  *
 174  * \see dt_index_operations::dio_it.fini() in the API description for details.
 175  */
 176 static void lod_it_fini(const struct lu_env *env, struct dt_it *di)
 177 {
 178         struct lod_it *it = (struct lod_it *)di;
 179
 180         LOD_CHECK_IT(env, it);
 181         it->lit_obj->do_index_ops->dio_it.fini(env, it->lit_it);
 182
 183         /* the iterator not in use any more */
 184         it->lit_obj = NULL;
 185         it->lit_it = NULL;
 186 }
 187
 188 /**
 189  * Implementation of dt_it_ops::get.
 190  *
 191  * Used with regular (non-striped) objects.
 192  *
 193  * \see dt_it_ops::get() in the API description for details.
 194  */
 195 static int lod_it_get(const struct lu_env *env, struct dt_it *di,
 196                       const struct dt_key *key)
 197 {
 198         const struct lod_it *it = (const struct lod_it *)di;
 199
 200         LOD_CHECK_IT(env, it);
 201         return it->lit_obj->do_index_ops->dio_it.get(env, it->lit_it, key);
 202 }
 203
 204 /**
 205  * Implementation of dt_it_ops::put.
 206  *
 207  * Used with regular (non-striped) objects.
 208  *
 209  * \see dt_it_ops::put() in the API description for details.
 210  */
 211 static void lod_it_put(const struct lu_env *env, struct dt_it *di)
 212 {
 213         struct lod_it *it = (struct lod_it *)di;
 214
 215         LOD_CHECK_IT(env, it);
 216         return it->lit_obj->do_index_ops->dio_it.put(env, it->lit_it);
 217 }
 218
 219 /**
 220  * Implementation of dt_it_ops::next.
 221  *
 222  * Used with regular (non-striped) objects
 223  *
 224  * \see dt_it_ops::next() in the API description for details.
 225  */
 226 static int lod_it_next(const struct lu_env *env, struct dt_it *di)
 227 {
 228         struct lod_it *it = (struct lod_it *)di;
 229
 230         LOD_CHECK_IT(env, it);
 231         return it->lit_obj->do_index_ops->dio_it.next(env, it->lit_it);
 232 }
 233
 234 /**
 235  * Implementation of dt_it_ops::key.
 236  *
 237  * Used with regular (non-striped) objects.
 238  *
 239  * \see dt_it_ops::key() in the API description for details.
 240  */
 241 static struct dt_key *lod_it_key(const struct lu_env *env,
 242                                  const struct dt_it *di)
 243 {
 244         const struct lod_it *it = (const struct lod_it *)di;
 245
 246         LOD_CHECK_IT(env, it);
 247         return it->lit_obj->do_index_ops->dio_it.key(env, it->lit_it);
 248 }
 249
 250 /**
 251  * Implementation of dt_it_ops::key_size.
 252  *
 253  * Used with regular (non-striped) objects.
 254  *
 255  * \see dt_it_ops::key_size() in the API description for details.
 256  */
 257 static int lod_it_key_size(const struct lu_env *env, const struct dt_it *di)
 258 {
 259         struct lod_it *it = (struct lod_it *)di;
 260
 261         LOD_CHECK_IT(env, it);
 262         return it->lit_obj->do_index_ops->dio_it.key_size(env, it->lit_it);
 263 }
 264
 265 /**
 266  * Implementation of dt_it_ops::rec.
 267  *
 268  * Used with regular (non-striped) objects.
 269  *
 270  * \see dt_it_ops::rec() in the API description for details.
 271  */
 272 static int lod_it_rec(const struct lu_env *env, const struct dt_it *di,
 273                       struct dt_rec *rec, __u32 attr)
 274 {
 275         const struct lod_it *it = (const struct lod_it *)di;
 276
 277         LOD_CHECK_IT(env, it);
 278         return it->lit_obj->do_index_ops->dio_it.rec(env, it->lit_it, rec,
 279                                                      attr);
 280 }
 281
 282 /**
 283  * Implementation of dt_it_ops::rec_size.
 284  *
 285  * Used with regular (non-striped) objects.
 286  *
 287  * \see dt_it_ops::rec_size() in the API description for details.
 288  */
 289 static int lod_it_rec_size(const struct lu_env *env, const struct dt_it *di,
 290                            __u32 attr)
 291 {
 292         const struct lod_it *it = (const struct lod_it *)di;
 293
 294         LOD_CHECK_IT(env, it);
 295         return it->lit_obj->do_index_ops->dio_it.rec_size(env, it->lit_it,
 296                                                           attr);
 297 }
 298
 299 /**
 300  * Implementation of dt_it_ops::store.
 301  *
 302  * Used with regular (non-striped) objects.
 303  *
 304  * \see dt_it_ops::store() in the API description for details.
 305  */
 306 static __u64 lod_it_store(const struct lu_env *env, const struct dt_it *di)
 307 {
 308         const struct lod_it *it = (const struct lod_it *)di;
 309
 310         LOD_CHECK_IT(env, it);
 311         return it->lit_obj->do_index_ops->dio_it.store(env, it->lit_it);
 312 }
 313
 314 /**
 315  * Implementation of dt_it_ops::load.
 316  *
 317  * Used with regular (non-striped) objects.
 318  *
 319  * \see dt_it_ops::load() in the API description for details.
 320  */
 321 static int lod_it_load(const struct lu_env *env, const struct dt_it *di,
 322                        __u64 hash)
 323 {
 324         const struct lod_it *it = (const struct lod_it *)di;
 325
 326         LOD_CHECK_IT(env, it);
 327         return it->lit_obj->do_index_ops->dio_it.load(env, it->lit_it, hash);
 328 }
 329
 330 /**
 331  * Implementation of dt_it_ops::key_rec.
 332  *
 333  * Used with regular (non-striped) objects.
 334  *
 335  * \see dt_it_ops::rec() in the API description for details.
 336  */
 337 static int lod_it_key_rec(const struct lu_env *env, const struct dt_it *di,
 338                           void *key_rec)
 339 {
 340         const struct lod_it *it = (const struct lod_it *)di;
 341
 342         LOD_CHECK_IT(env, it);
 343         return it->lit_obj->do_index_ops->dio_it.key_rec(env, it->lit_it,
 344                                                          key_rec);
 345 }
 346
 347 static struct dt_index_operations lod_index_ops = {
 348         .dio_lookup             = lod_lookup,
 349         .dio_declare_insert     = lod_declare_insert,
 350         .dio_insert             = lod_insert,
 351         .dio_declare_delete     = lod_declare_delete,
 352         .dio_delete             = lod_delete,
 353         .dio_it = {
 354                 .init           = lod_it_init,
 355                 .fini           = lod_it_fini,
 356                 .get            = lod_it_get,
 357                 .put            = lod_it_put,
 358                 .next           = lod_it_next,
 359                 .key            = lod_it_key,
 360                 .key_size       = lod_it_key_size,
 361                 .rec            = lod_it_rec,
 362                 .rec_size       = lod_it_rec_size,
 363                 .store          = lod_it_store,
 364                 .load           = lod_it_load,
 365                 .key_rec        = lod_it_key_rec,
 366         }
 367 };
 368
 369 /**
 370  * Implementation of dt_it_ops::init.
 371  *
 372  * Used with striped objects. Internally just initializes the iterator
 373  * on the first stripe.
 374  *
 375  * \see dt_it_ops::init() in the API description for details.
 376  */
 377 static struct dt_it *lod_striped_it_init(const struct lu_env *env,
 378                                          struct dt_object *dt, __u32 attr)
 379 {
 380         struct lod_object       *lo = lod_dt_obj(dt);
 381         struct dt_object        *next;
 382         struct lod_it           *it = &lod_env_info(env)->lti_it;
 383         struct dt_it            *it_next;
 384         ENTRY;
 385
 386         LASSERT(lo->ldo_dir_stripe_count > 0);
 387         next = lo->ldo_stripe[0];
 388         LASSERT(next != NULL);
 389         LASSERT(next->do_index_ops != NULL);
 390
 391         it_next = next->do_index_ops->dio_it.init(env, next, attr);
 392         if (IS_ERR(it_next))
 393                 return it_next;
 394
 395         /* currently we do not use more than one iterator per thread
 396          * so we store it in thread info. if at some point we need
 397          * more active iterators in a single thread, we can allocate
 398          * additional ones */
 399         LASSERT(it->lit_obj == NULL);
 400
 401         it->lit_stripe_index = 0;
 402         it->lit_attr = attr;
 403         it->lit_it = it_next;
 404         it->lit_obj = dt;
 405
 406         return (struct dt_it *)it;
 407 }
 408
 409 #define LOD_CHECK_STRIPED_IT(env, it, lo)                               \
 410 do {                                                                    \
 411         LASSERT((it)->lit_obj != NULL);                                 \
 412         LASSERT((it)->lit_it != NULL);                                  \
 413         LASSERT((lo)->ldo_dir_stripe_count > 0);                        \
 414         LASSERT((it)->lit_stripe_index < (lo)->ldo_dir_stripe_count);   \
 415 } while (0)
 416
 417 /**
 418  * Implementation of dt_it_ops::fini.
 419  *
 420  * Used with striped objects.
 421  *
 422  * \see dt_it_ops::fini() in the API description for details.
 423  */
 424 static void lod_striped_it_fini(const struct lu_env *env, struct dt_it *di)
 425 {
 426         struct lod_it           *it = (struct lod_it *)di;
 427         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
 428         struct dt_object        *next;
 429
 430         /* If lit_it == NULL, then it means the sub_it has been finished,
 431          * which only happens in failure cases, see lod_striped_it_next() */
 432         if (it->lit_it != NULL) {
 433                 LOD_CHECK_STRIPED_IT(env, it, lo);
 434
 435                 next = lo->ldo_stripe[it->lit_stripe_index];
 436                 LASSERT(next != NULL);
 437                 LASSERT(next->do_index_ops != NULL);
 438
 439                 next->do_index_ops->dio_it.fini(env, it->lit_it);
 440         }
 441
 442         /* the iterator not in use any more */
 443         it->lit_obj = NULL;
 444         it->lit_it = NULL;
 445         it->lit_stripe_index = 0;
 446 }
 447
 448 /**
 449  * Implementation of dt_it_ops::get.
 450  *
 451  * Right now it's not used widely, only to reset the iterator to the
 452  * initial position. It should be possible to implement a full version
 453  * which chooses a correct stripe to be able to position with any key.
 454  *
 455  * \see dt_it_ops::get() in the API description for details.
 456  */
 457 static int lod_striped_it_get(const struct lu_env *env, struct dt_it *di,
 458                               const struct dt_key *key)
 459 {
 460         const struct lod_it     *it = (const struct lod_it *)di;
 461         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
 462         struct dt_object        *next;
 463         ENTRY;
 464
 465         LOD_CHECK_STRIPED_IT(env, it, lo);
 466
 467         next = lo->ldo_stripe[it->lit_stripe_index];
 468         LASSERT(next != NULL);
 469         LASSERT(next->do_index_ops != NULL);
 470
 471         return next->do_index_ops->dio_it.get(env, it->lit_it, key);
 472 }
 473
 474 /**
 475  * Implementation of dt_it_ops::put.
 476  *
 477  * Used with striped objects.
 478  *
 479  * \see dt_it_ops::put() in the API description for details.
 480  */
 481 static void lod_striped_it_put(const struct lu_env *env, struct dt_it *di)
 482 {
 483         struct lod_it           *it = (struct lod_it *)di;
 484         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
 485         struct dt_object        *next;
 486
 487         LOD_CHECK_STRIPED_IT(env, it, lo);
 488
 489         next = lo->ldo_stripe[it->lit_stripe_index];
 490         LASSERT(next != NULL);
 491         LASSERT(next->do_index_ops != NULL);
 492
 493         return next->do_index_ops->dio_it.put(env, it->lit_it);
 494 }
 495
 496 /**
 497  * Implementation of dt_it_ops::next.
 498  *
 499  * Used with striped objects. When the end of the current stripe is
 500  * reached, the method takes the next stripe's iterator.
 501  *
 502  * \see dt_it_ops::next() in the API description for details.
 503  */
 504 static int lod_striped_it_next(const struct lu_env *env, struct dt_it *di)
 505 {
 506         struct lod_it           *it = (struct lod_it *)di;
 507         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
 508         struct dt_object        *next;
 509         struct dt_it            *it_next;
 510         int                     rc;
 511         ENTRY;
 512
 513         LOD_CHECK_STRIPED_IT(env, it, lo);
 514
 515         next = lo->ldo_stripe[it->lit_stripe_index];
 516         LASSERT(next != NULL);
 517         LASSERT(next->do_index_ops != NULL);
 518 again:
 519         rc = next->do_index_ops->dio_it.next(env, it->lit_it);
 520         if (rc < 0)
 521                 RETURN(rc);
 522
 523         if (rc == 0 && it->lit_stripe_index == 0)
 524                 RETURN(rc);
 525
 526         if (rc == 0 && it->lit_stripe_index > 0) {
 527                 struct lu_dirent *ent;
 528
 529                 ent = (struct lu_dirent *)lod_env_info(env)->lti_key;
 530
 531                 rc = next->do_index_ops->dio_it.rec(env, it->lit_it,
 532                                                     (struct dt_rec *)ent,
 533                                                     it->lit_attr);
 534                 if (rc != 0)
 535                         RETURN(rc);
 536
 537                 /* skip . and .. for slave stripe */
 538                 if ((strncmp(ent->lde_name, ".",
 539                              le16_to_cpu(ent->lde_namelen)) == 0 &&
 540                      le16_to_cpu(ent->lde_namelen) == 1) ||
 541                     (strncmp(ent->lde_name, "..",
 542                              le16_to_cpu(ent->lde_namelen)) == 0 &&
 543                      le16_to_cpu(ent->lde_namelen) == 2))
 544                         goto again;
 545
 546                 RETURN(rc);
 547         }
 548
 549         /* go to next stripe */
 550         if (it->lit_stripe_index + 1 >= lo->ldo_dir_stripe_count)
 551                 RETURN(1);
 552
 553         it->lit_stripe_index++;
 554
 555         next->do_index_ops->dio_it.put(env, it->lit_it);
 556         next->do_index_ops->dio_it.fini(env, it->lit_it);
 557         it->lit_it = NULL;
 558
 559         next = lo->ldo_stripe[it->lit_stripe_index];
 560         LASSERT(next != NULL);
 561         rc = next->do_ops->do_index_try(env, next, &dt_directory_features);
 562         if (rc != 0)
 563                 RETURN(rc);
 564
 565         LASSERT(next->do_index_ops != NULL);
 566
 567         it_next = next->do_index_ops->dio_it.init(env, next, it->lit_attr);
 568         if (!IS_ERR(it_next)) {
 569                 it->lit_it = it_next;
 570                 goto again;
 571         } else {
 572                 rc = PTR_ERR(it_next);
 573         }
 574
 575         RETURN(rc);
 576 }
 577
 578 /**
 579  * Implementation of dt_it_ops::key.
 580  *
 581  * Used with striped objects.
 582  *
 583  * \see dt_it_ops::key() in the API description for details.
 584  */
 585 static struct dt_key *lod_striped_it_key(const struct lu_env *env,
 586                                          const struct dt_it *di)
 587 {
 588         const struct lod_it     *it = (const struct lod_it *)di;
 589         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
 590         struct dt_object        *next;
 591
 592         LOD_CHECK_STRIPED_IT(env, it, lo);
 593
 594         next = lo->ldo_stripe[it->lit_stripe_index];
 595         LASSERT(next != NULL);
 596         LASSERT(next->do_index_ops != NULL);
 597
 598         return next->do_index_ops->dio_it.key(env, it->lit_it);
 599 }
 600
 601 /**
 602  * Implementation of dt_it_ops::key_size.
 603  *
 604  * Used with striped objects.
 605  *
 606  * \see dt_it_ops::size() in the API description for details.
 607  */
 608 static int lod_striped_it_key_size(const struct lu_env *env,
 609                                    const struct dt_it *di)
 610 {
 611         struct lod_it           *it = (struct lod_it *)di;
 612         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
 613         struct dt_object        *next;
 614
 615         LOD_CHECK_STRIPED_IT(env, it, lo);
 616
 617         next = lo->ldo_stripe[it->lit_stripe_index];
 618         LASSERT(next != NULL);
 619         LASSERT(next->do_index_ops != NULL);
 620
 621         return next->do_index_ops->dio_it.key_size(env, it->lit_it);
 622 }
 623
 624 /**
 625  * Implementation of dt_it_ops::rec.
 626  *
 627  * Used with striped objects.
 628  *
 629  * \see dt_it_ops::rec() in the API description for details.
 630  */
 631 static int lod_striped_it_rec(const struct lu_env *env, const struct dt_it *di,
 632                               struct dt_rec *rec, __u32 attr)
 633 {
 634         const struct lod_it     *it = (const struct lod_it *)di;
 635         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
 636         struct dt_object        *next;
 637
 638         LOD_CHECK_STRIPED_IT(env, it, lo);
 639
 640         next = lo->ldo_stripe[it->lit_stripe_index];
 641         LASSERT(next != NULL);
 642         LASSERT(next->do_index_ops != NULL);
 643
 644         return next->do_index_ops->dio_it.rec(env, it->lit_it, rec, attr);
 645 }
 646
 647 /**
 648  * Implementation of dt_it_ops::rec_size.
 649  *
 650  * Used with striped objects.
 651  *
 652  * \see dt_it_ops::rec_size() in the API description for details.
 653  */
 654 static int lod_striped_it_rec_size(const struct lu_env *env,
 655                                    const struct dt_it *di, __u32 attr)
 656 {
 657         struct lod_it           *it = (struct lod_it *)di;
 658         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
 659         struct dt_object        *next;
 660
 661         LOD_CHECK_STRIPED_IT(env, it, lo);
 662
 663         next = lo->ldo_stripe[it->lit_stripe_index];
 664         LASSERT(next != NULL);
 665         LASSERT(next->do_index_ops != NULL);
 666
 667         return next->do_index_ops->dio_it.rec_size(env, it->lit_it, attr);
 668 }
 669
 670 /**
 671  * Implementation of dt_it_ops::store.
 672  *
 673  * Used with striped objects.
 674  *
 675  * \see dt_it_ops::store() in the API description for details.
 676  */
 677 static __u64 lod_striped_it_store(const struct lu_env *env,
 678                                   const struct dt_it *di)
 679 {
 680         const struct lod_it     *it = (const struct lod_it *)di;
 681         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
 682         struct dt_object        *next;
 683
 684         LOD_CHECK_STRIPED_IT(env, it, lo);
 685
 686         next = lo->ldo_stripe[it->lit_stripe_index];
 687         LASSERT(next != NULL);
 688         LASSERT(next->do_index_ops != NULL);
 689
 690         return next->do_index_ops->dio_it.store(env, it->lit_it);
 691 }
 692
 693 /**
 694  * Implementation of dt_it_ops::load.
 695  *
 696  * Used with striped objects.
 697  *
 698  * \see dt_it_ops::load() in the API description for details.
 699  */
 700 static int lod_striped_it_load(const struct lu_env *env,
 701                                const struct dt_it *di, __u64 hash)
 702 {
 703         const struct lod_it     *it = (const struct lod_it *)di;
 704         struct lod_object       *lo = lod_dt_obj(it->lit_obj);
 705         struct dt_object        *next;
 706
 707         LOD_CHECK_STRIPED_IT(env, it, lo);
 708
 709         next = lo->ldo_stripe[it->lit_stripe_index];
 710         LASSERT(next != NULL);
 711         LASSERT(next->do_index_ops != NULL);
 712
 713         return next->do_index_ops->dio_it.load(env, it->lit_it, hash);
 714 }
 715
 716 static struct dt_index_operations lod_striped_index_ops = {
 717         .dio_lookup             = lod_lookup,
 718         .dio_declare_insert     = lod_declare_insert,
 719         .dio_insert             = lod_insert,
 720         .dio_declare_delete     = lod_declare_delete,
 721         .dio_delete             = lod_delete,
 722         .dio_it = {
 723                 .init           = lod_striped_it_init,
 724                 .fini           = lod_striped_it_fini,
 725                 .get            = lod_striped_it_get,
 726                 .put            = lod_striped_it_put,
 727                 .next           = lod_striped_it_next,
 728                 .key            = lod_striped_it_key,
 729                 .key_size       = lod_striped_it_key_size,
 730                 .rec            = lod_striped_it_rec,
 731                 .rec_size       = lod_striped_it_rec_size,
 732                 .store          = lod_striped_it_store,
 733                 .load           = lod_striped_it_load,
 734         }
 735 };
 736
 737 /**
 738  * Append the FID for each shard of the striped directory after the
 739  * given LMV EA header.
 740  *
 741  * To simplify striped directory and the consistency verification,
 742  * we only store the LMV EA header on disk, for both master object
 743  * and slave objects. When someone wants to know the whole LMV EA,
 744  * such as client readdir(), we can build the entrie LMV EA on the
 745  * MDT side (in RAM) via iterating the sub-directory entries that
 746  * are contained in the master object of the stripe directory.
 747  *
 748  * For the master object of the striped directroy, the valid name
 749  * for each shard is composed of the ${shard_FID}:${shard_idx}.
 750  *
 751  * There may be holes in the LMV EA if some shards' name entries
 752  * are corrupted or lost.
 753  *
 754  * \param[in] env       pointer to the thread context
 755  * \param[in] lo        pointer to the master object of the striped directory
 756  * \param[in] buf       pointer to the lu_buf which will hold the LMV EA
 757  * \param[in] resize    whether re-allocate the buffer if it is not big enough
 758  *
 759  * \retval              positive size of the LMV EA
 760  * \retval              0 for nothing to be loaded
 761  * \retval              negative error number on failure
 762  */
 763 int lod_load_lmv_shards(const struct lu_env *env, struct lod_object *lo,
 764                         struct lu_buf *buf, bool resize)
 765 {
 766         struct lu_dirent        *ent    =
 767                         (struct lu_dirent *)lod_env_info(env)->lti_key;
 768         struct lod_device       *lod    = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
 769         struct dt_object        *obj    = dt_object_child(&lo->ldo_obj);
 770         struct lmv_mds_md_v1    *lmv1   = buf->lb_buf;
 771         struct dt_it            *it;
 772         const struct dt_it_ops  *iops;
 773         __u32                    stripes;
 774         __u32                    magic  = le32_to_cpu(lmv1->lmv_magic);
 775         size_t                   lmv1_size;
 776         int                      rc;
 777         ENTRY;
 778
 779         /* If it is not a striped directory, then load nothing. */
 780         if (magic != LMV_MAGIC_V1)
 781                 RETURN(0);
 782
 783         /* If it is in migration (or failure), then load nothing. */
 784         if (le32_to_cpu(lmv1->lmv_hash_type) & LMV_HASH_FLAG_MIGRATION)
 785                 RETURN(0);
 786
 787         stripes = le32_to_cpu(lmv1->lmv_stripe_count);
 788         if (stripes < 1)
 789                 RETURN(0);
 790
 791         rc = lmv_mds_md_size(stripes, magic);
 792         if (rc < 0)
 793                 RETURN(rc);
 794         lmv1_size = rc;
 795         if (buf->lb_len < lmv1_size) {
 796                 struct lu_buf tbuf;
 797
 798                 if (!resize)
 799                         RETURN(-ERANGE);
 800
 801                 tbuf = *buf;
 802                 buf->lb_buf = NULL;
 803                 buf->lb_len = 0;
 804                 lu_buf_alloc(buf, lmv1_size);
 805                 lmv1 = buf->lb_buf;
 806                 if (lmv1 == NULL)
 807                         RETURN(-ENOMEM);
 808
 809                 memcpy(buf->lb_buf, tbuf.lb_buf, tbuf.lb_len);
 810         }
 811
 812         if (unlikely(!dt_try_as_dir(env, obj)))
 813                 RETURN(-ENOTDIR);
 814
 815         memset(&lmv1->lmv_stripe_fids[0], 0, stripes * sizeof(struct lu_fid));
 816         iops = &obj->do_index_ops->dio_it;
 817         it = iops->init(env, obj, LUDA_64BITHASH);
 818         if (IS_ERR(it))
 819                 RETURN(PTR_ERR(it));
 820
 821         rc = iops->load(env, it, 0);
 822         if (rc == 0)
 823                 rc = iops->next(env, it);
 824         else if (rc > 0)
 825                 rc = 0;
 826
 827         while (rc == 0) {
 828                 char             name[FID_LEN + 2] = "";
 829                 struct lu_fid    fid;
 830                 __u32            index;
 831                 int              len;
 832
 833                 rc = iops->rec(env, it, (struct dt_rec *)ent, LUDA_64BITHASH);
 834                 if (rc != 0)
 835                         break;
 836
 837                 rc = -EIO;
 838
 839                 fid_le_to_cpu(&fid, &ent->lde_fid);
 840                 ent->lde_namelen = le16_to_cpu(ent->lde_namelen);
 841                 if (ent->lde_name[0] == '.') {
 842                         if (ent->lde_namelen == 1)
 843                                 goto next;
 844
 845                         if (ent->lde_namelen == 2 && ent->lde_name[1] == '.')
 846                                 goto next;
 847                 }
 848
 849                 len = snprintf(name, sizeof(name),
 850                                DFID":", PFID(&ent->lde_fid));
 851                 /* The ent->lde_name is composed of ${FID}:${index} */
 852                 if (ent->lde_namelen < len + 1 ||
 853                     memcmp(ent->lde_name, name, len) != 0) {
 854                         CDEBUG(lod->lod_lmv_failout ? D_ERROR : D_INFO,
 855                                "%s: invalid shard name %.*s with the FID "DFID
 856                                " for the striped directory "DFID", %s\n",
 857                                lod2obd(lod)->obd_name, ent->lde_namelen,
 858                                ent->lde_name, PFID(&fid),
 859                                PFID(lu_object_fid(&obj->do_lu)),
 860                                lod->lod_lmv_failout ? "failout" : "skip");
 861
 862                         if (lod->lod_lmv_failout)
 863                                 break;
 864
 865                         goto next;
 866                 }
 867
 868                 index = 0;
 869                 do {
 870                         if (ent->lde_name[len] < '0' ||
 871                             ent->lde_name[len] > '9') {
 872                                 CDEBUG(lod->lod_lmv_failout ? D_ERROR : D_INFO,
 873                                        "%s: invalid shard name %.*s with the "
 874                                        "FID "DFID" for the striped directory "
 875                                        DFID", %s\n",
 876                                        lod2obd(lod)->obd_name, ent->lde_namelen,
 877                                        ent->lde_name, PFID(&fid),
 878                                        PFID(lu_object_fid(&obj->do_lu)),
 879                                        lod->lod_lmv_failout ?
 880                                        "failout" : "skip");
 881
 882                                 if (lod->lod_lmv_failout)
 883                                         break;
 884
 885                                 goto next;
 886                         }
 887
 888                         index = index * 10 + ent->lde_name[len++] - '0';
 889                 } while (len < ent->lde_namelen);
 890
 891                 if (len == ent->lde_namelen) {
 892                         /* Out of LMV EA range. */
 893                         if (index >= stripes) {
 894                                 CERROR("%s: the shard %.*s for the striped "
 895                                        "directory "DFID" is out of the known "
 896                                        "LMV EA range [0 - %u], failout\n",
 897                                        lod2obd(lod)->obd_name, ent->lde_namelen,
 898                                        ent->lde_name,
 899                                        PFID(lu_object_fid(&obj->do_lu)),
 900                                        stripes - 1);
 901
 902                                 break;
 903                         }
 904
 905                         /* The slot has been occupied. */
 906                         if (!fid_is_zero(&lmv1->lmv_stripe_fids[index])) {
 907                                 struct lu_fid fid0;
 908
 909                                 fid_le_to_cpu(&fid0,
 910                                         &lmv1->lmv_stripe_fids[index]);
 911                                 CERROR("%s: both the shard "DFID" and "DFID
 912                                        " for the striped directory "DFID
 913                                        " claim the same LMV EA slot at the "
 914                                        "index %d, failout\n",
 915                                        lod2obd(lod)->obd_name,
 916                                        PFID(&fid0), PFID(&fid),
 917                                        PFID(lu_object_fid(&obj->do_lu)), index);
 918
 919                                 break;
 920                         }
 921
 922                         /* stored as LE mode */
 923                         lmv1->lmv_stripe_fids[index] = ent->lde_fid;
 924
 925 next:
 926                         rc = iops->next(env, it);
 927                 }
 928         }
 929
 930         iops->put(env, it);
 931         iops->fini(env, it);
 932
 933         RETURN(rc > 0 ? lmv_mds_md_size(stripes, magic) : rc);
 934 }
 935
 936 /**
 937  * Implementation of dt_object_operations::do_index_try.
 938  *
 939  * \see dt_object_operations::do_index_try() in the API description for details.
 940  */
 941 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
 942                          const struct dt_index_features *feat)
 943 {
 944         struct lod_object       *lo = lod_dt_obj(dt);
 945         struct dt_object        *next = dt_object_child(dt);
 946         int                     rc;
 947         ENTRY;
 948
 949         LASSERT(next->do_ops);
 950         LASSERT(next->do_ops->do_index_try);
 951
 952         rc = lod_load_striping_locked(env, lo);
 953         if (rc != 0)
 954                 RETURN(rc);
 955
 956         rc = next->do_ops->do_index_try(env, next, feat);
 957         if (rc != 0)
 958                 RETURN(rc);
 959
 960         if (lo->ldo_dir_stripe_count > 0) {
 961                 int i;
 962
 963                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
 964                         if (dt_object_exists(lo->ldo_stripe[i]) == 0)
 965                                 continue;
 966                         rc = lo->ldo_stripe[i]->do_ops->do_index_try(env,
 967                                                 lo->ldo_stripe[i], feat);
 968                         if (rc != 0)
 969                                 RETURN(rc);
 970                 }
 971                 dt->do_index_ops = &lod_striped_index_ops;
 972         } else {
 973                 dt->do_index_ops = &lod_index_ops;
 974         }
 975
 976         RETURN(rc);
 977 }
 978
 979 /**
 980  * Implementation of dt_object_operations::do_read_lock.
 981  *
 982  * \see dt_object_operations::do_read_lock() in the API description for details.
 983  */
 984 static void lod_read_lock(const struct lu_env *env, struct dt_object *dt,
 985                           unsigned role)
 986 {
 987         dt_read_lock(env, dt_object_child(dt), role);
 988 }
 989
 990 /**
 991  * Implementation of dt_object_operations::do_write_lock.
 992  *
 993  * \see dt_object_operations::do_write_lock() in the API description for
 994  * details.
 995  */
 996 static void lod_write_lock(const struct lu_env *env, struct dt_object *dt,
 997                            unsigned role)
 998 {
 999         dt_write_lock(env, dt_object_child(dt), role);
1000 }
1001
1002 /**
1003  * Implementation of dt_object_operations::do_read_unlock.
1004  *
1005  * \see dt_object_operations::do_read_unlock() in the API description for
1006  * details.
1007  */
1008 static void lod_read_unlock(const struct lu_env *env, struct dt_object *dt)
1009 {
1010         dt_read_unlock(env, dt_object_child(dt));
1011 }
1012
1013 /**
1014  * Implementation of dt_object_operations::do_write_unlock.
1015  *
1016  * \see dt_object_operations::do_write_unlock() in the API description for
1017  * details.
1018  */
1019 static void lod_write_unlock(const struct lu_env *env, struct dt_object *dt)
1020 {
1021         dt_write_unlock(env, dt_object_child(dt));
1022 }
1023
1024 /**
1025  * Implementation of dt_object_operations::do_write_locked.
1026  *
1027  * \see dt_object_operations::do_write_locked() in the API description for
1028  * details.
1029  */
1030 static int lod_write_locked(const struct lu_env *env, struct dt_object *dt)
1031 {
1032         return dt_write_locked(env, dt_object_child(dt));
1033 }
1034
1035 /**
1036  * Implementation of dt_object_operations::do_attr_get.
1037  *
1038  * \see dt_object_operations::do_attr_get() in the API description for details.
1039  */
1040 static int lod_attr_get(const struct lu_env *env,
1041                         struct dt_object *dt,
1042                         struct lu_attr *attr)
1043 {
1044         /* Note: for striped directory, client will merge attributes
1045          * from all of the sub-stripes see lmv_merge_attr(), and there
1046          * no MDD logic depend on directory nlink/size/time, so we can
1047          * always use master inode nlink and size for now. */
1048         return dt_attr_get(env, dt_object_child(dt), attr);
1049 }
1050
1051 static inline void lod_adjust_stripe_info(struct lod_layout_component *comp,
1052                                           struct lov_desc *desc)
1053 {
1054         if (comp->llc_pattern != LOV_PATTERN_MDT) {
1055                 if (!comp->llc_stripe_count)
1056                         comp->llc_stripe_count =
1057                                 desc->ld_default_stripe_count;
1058         }
1059         if (comp->llc_stripe_size <= 0)
1060                 comp->llc_stripe_size = desc->ld_default_stripe_size;
1061 }
1062
1063 int lod_obj_for_each_stripe(const struct lu_env *env, struct lod_object *lo,
1064                             struct thandle *th,
1065                             struct lod_obj_stripe_cb_data *data)
1066 {
1067         struct lod_layout_component *lod_comp;
1068         int i, j, rc;
1069         ENTRY;
1070
1071         LASSERT(lo->ldo_comp_cnt != 0 && lo->ldo_comp_entries != NULL);
1072         for (i = 0; i < lo->ldo_comp_cnt; i++) {
1073                 lod_comp = &lo->ldo_comp_entries[i];
1074
1075                 if (lod_comp->llc_stripe == NULL)
1076                         continue;
1077
1078                 /* has stripe but not inited yet, this component has been
1079                  * declared to be created, but hasn't created yet.
1080                  */
1081                 if (!lod_comp_inited(lod_comp))
1082                         continue;
1083
1084                 if (data->locd_comp_skip_cb &&
1085                     data->locd_comp_skip_cb(env, lo, i, data))
1086                         continue;
1087
1088                 LASSERT(lod_comp->llc_stripe_count > 0);
1089                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
1090                         struct dt_object *dt = lod_comp->llc_stripe[j];
1091
1092                         if (dt == NULL)
1093                                 continue;
1094                         rc = data->locd_stripe_cb(env, lo, dt, th, i, j, data);
1095                         if (rc != 0)
1096                                 RETURN(rc);
1097                 }
1098         }
1099         RETURN(0);
1100 }
1101
1102 static bool lod_obj_attr_set_comp_skip_cb(const struct lu_env *env,
1103                 struct lod_object *lo, int comp_idx,
1104                 struct lod_obj_stripe_cb_data *data)
1105 {
1106         struct lod_layout_component *lod_comp = &lo->ldo_comp_entries[comp_idx];
1107         bool skipped = false;
1108
1109         if (!(data->locd_attr->la_valid & LA_LAYOUT_VERSION))
1110                 return skipped;
1111
1112         switch (lo->ldo_flr_state) {
1113         case LCM_FL_WRITE_PENDING: {
1114                 int i;
1115
1116                 /* skip stale components */
1117                 if (lod_comp->llc_flags & LCME_FL_STALE) {
1118                         skipped = true;
1119                         break;
1120                 }
1121
1122                 /* skip valid and overlapping components, therefore any
1123                  * attempts to write overlapped components will never succeed
1124                  * because client will get EINPROGRESS. */
1125                 for (i = 0; i < lo->ldo_comp_cnt; i++) {
1126                         if (i == comp_idx)
1127                                 continue;
1128
1129                         if (lo->ldo_comp_entries[i].llc_flags & LCME_FL_STALE)
1130                                 continue;
1131
1132                         if (lu_extent_is_overlapped(&lod_comp->llc_extent,
1133                                         &lo->ldo_comp_entries[i].llc_extent)) {
1134                                 skipped = true;
1135                                 break;
1136                         }
1137                 }
1138                 break;
1139         }
1140         default:
1141                 LASSERTF(0, "impossible: %d\n", lo->ldo_flr_state);
1142         case LCM_FL_SYNC_PENDING:
1143                 break;
1144         }
1145
1146         CDEBUG(D_LAYOUT, DFID": %s to set component %x to version: %u\n",
1147                PFID(lu_object_fid(&lo->ldo_obj.do_lu)),
1148                skipped ? "skipped" : "chose", lod_comp->llc_id,
1149                data->locd_attr->la_layout_version);
1150
1151         return skipped;
1152 }
1153
1154 static inline int
1155 lod_obj_stripe_attr_set_cb(const struct lu_env *env, struct lod_object *lo,
1156                            struct dt_object *dt, struct thandle *th,
1157                            int comp_idx, int stripe_idx,
1158                            struct lod_obj_stripe_cb_data *data)
1159 {
1160         if (data->locd_declare)
1161                 return lod_sub_declare_attr_set(env, dt, data->locd_attr, th);
1162
1163         if (data->locd_attr->la_valid & LA_LAYOUT_VERSION) {
1164                 CDEBUG(D_LAYOUT, DFID": set layout version: %u, comp_idx: %d\n",
1165                        PFID(lu_object_fid(&dt->do_lu)),
1166                        data->locd_attr->la_layout_version, comp_idx);
1167         }
1168
1169         return lod_sub_attr_set(env, dt, data->locd_attr, th);
1170 }
1171
1172 /**
1173  * Implementation of dt_object_operations::do_declare_attr_set.
1174  *
1175  * If the object is striped, then apply the changes to all the stripes.
1176  *
1177  * \see dt_object_operations::do_declare_attr_set() in the API description
1178  * for details.
1179  */
1180 static int lod_declare_attr_set(const struct lu_env *env,
1181                                 struct dt_object *dt,
1182                                 const struct lu_attr *attr,
1183                                 struct thandle *th)
1184 {
1185         struct dt_object  *next = dt_object_child(dt);
1186         struct lod_object *lo = lod_dt_obj(dt);
1187         int                rc, i;
1188         ENTRY;
1189
1190         /*
1191          * declare setattr on the local object
1192          */
1193         rc = lod_sub_declare_attr_set(env, next, attr, th);
1194         if (rc)
1195                 RETURN(rc);
1196
1197         /* osp_declare_attr_set() ignores all attributes other than
1198          * UID, GID, PROJID, and size, and osp_attr_set() ignores all
1199          * but UID, GID and PROJID. Declaration of size attr setting
1200          * happens through lod_declare_init_size(), and not through
1201          * this function. Therefore we need not load striping unless
1202          * ownership is changing.  This should save memory and (we hope)
1203          * speed up rename().
1204          */
1205         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1206                 if (!(attr->la_valid & LA_REMOTE_ATTR_SET))
1207                         RETURN(rc);
1208
1209                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
1210                         RETURN(0);
1211         } else {
1212                 if (!(attr->la_valid & (LA_UID | LA_GID | LA_PROJID | LA_MODE |
1213                                         LA_ATIME | LA_MTIME | LA_CTIME |
1214                                         LA_FLAGS)))
1215                         RETURN(rc);
1216         }
1217         /*
1218          * load striping information, notice we don't do this when object
1219          * is being initialized as we don't need this information till
1220          * few specific cases like destroy, chown
1221          */
1222         rc = lod_load_striping(env, lo);
1223         if (rc)
1224                 RETURN(rc);
1225
1226         if (!lod_obj_is_striped(dt))
1227                 RETURN(0);
1228
1229         /*
1230          * if object is striped declare changes on the stripes
1231          */
1232         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1233                 LASSERT(lo->ldo_stripe);
1234                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
1235                         if (lo->ldo_stripe[i] == NULL)
1236                                 continue;
1237                         rc = lod_sub_declare_attr_set(env, lo->ldo_stripe[i],
1238                                                       attr, th);
1239                         if (rc != 0)
1240                                 RETURN(rc);
1241                 }
1242         } else {
1243                 struct lod_obj_stripe_cb_data data = { { 0 } };
1244
1245                 data.locd_attr = attr;
1246                 data.locd_declare = true;
1247                 data.locd_stripe_cb = lod_obj_stripe_attr_set_cb;
1248                 rc = lod_obj_for_each_stripe(env, lo, th, &data);
1249         }
1250
1251         if (rc)
1252                 RETURN(rc);
1253
1254         if (!dt_object_exists(next) || dt_object_remote(next) ||
1255             !S_ISREG(attr->la_mode))
1256                 RETURN(0);
1257
1258         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE)) {
1259                 rc = lod_sub_declare_xattr_del(env, next, XATTR_NAME_LOV, th);
1260                 RETURN(rc);
1261         }
1262
1263         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE) ||
1264             OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PFL_RANGE)) {
1265                 struct lod_thread_info *info = lod_env_info(env);
1266                 struct lu_buf *buf = &info->lti_buf;
1267
1268                 buf->lb_buf = info->lti_ea_store;
1269                 buf->lb_len = info->lti_ea_store_size;
1270                 rc = lod_sub_declare_xattr_set(env, next, buf, XATTR_NAME_LOV,
1271                                                LU_XATTR_REPLACE, th);
1272         }
1273
1274         RETURN(rc);
1275 }
1276
1277 /**
1278  * Implementation of dt_object_operations::do_attr_set.
1279  *
1280  * If the object is striped, then apply the changes to all or subset of
1281  * the stripes depending on the object type and specific attributes.
1282  *
1283  * \see dt_object_operations::do_attr_set() in the API description for details.
1284  */
1285 static int lod_attr_set(const struct lu_env *env,
1286                         struct dt_object *dt,
1287                         const struct lu_attr *attr,
1288                         struct thandle *th)
1289 {
1290         struct dt_object        *next = dt_object_child(dt);
1291         struct lod_object       *lo = lod_dt_obj(dt);
1292         int                     rc, i;
1293         ENTRY;
1294
1295         /*
1296          * apply changes to the local object
1297          */
1298         rc = lod_sub_attr_set(env, next, attr, th);
1299         if (rc)
1300                 RETURN(rc);
1301
1302         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1303                 if (!(attr->la_valid & LA_REMOTE_ATTR_SET))
1304                         RETURN(rc);
1305
1306                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_OWNER))
1307                         RETURN(0);
1308         } else {
1309                 if (!(attr->la_valid & (LA_UID | LA_GID | LA_MODE | LA_PROJID |
1310                                         LA_ATIME | LA_MTIME | LA_CTIME |
1311                                         LA_FLAGS)))
1312                         RETURN(rc);
1313         }
1314
1315         /* FIXME: a tricky case in the code path of mdd_layout_change():
1316          * the in-memory striping information has been freed in lod_xattr_set()
1317          * due to layout change. It has to load stripe here again. It only
1318          * changes flags of layout so declare_attr_set() is still accurate */
1319         rc = lod_load_striping_locked(env, lo);
1320         if (rc)
1321                 RETURN(rc);
1322
1323         if (!lod_obj_is_striped(dt))
1324                 RETURN(0);
1325
1326         /*
1327          * if object is striped, apply changes to all the stripes
1328          */
1329         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
1330                 LASSERT(lo->ldo_stripe);
1331                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
1332                         if (unlikely(lo->ldo_stripe[i] == NULL))
1333                                 continue;
1334
1335                         if ((dt_object_exists(lo->ldo_stripe[i]) == 0))
1336                                 continue;
1337
1338                         rc = lod_sub_attr_set(env, lo->ldo_stripe[i], attr, th);
1339                         if (rc != 0)
1340                                 break;
1341                 }
1342         } else {
1343                 struct lod_obj_stripe_cb_data data = { { 0 } };
1344
1345                 data.locd_attr = attr;
1346                 data.locd_declare = false;
1347                 data.locd_comp_skip_cb = lod_obj_attr_set_comp_skip_cb;
1348                 data.locd_stripe_cb = lod_obj_stripe_attr_set_cb;
1349                 rc = lod_obj_for_each_stripe(env, lo, th, &data);
1350         }
1351
1352         if (rc)
1353                 RETURN(rc);
1354
1355         if (!dt_object_exists(next) || dt_object_remote(next) ||
1356             !S_ISREG(attr->la_mode))
1357                 RETURN(0);
1358
1359         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_STRIPE)) {
1360                 rc = lod_sub_xattr_del(env, next, XATTR_NAME_LOV, th);
1361                 RETURN(rc);
1362         }
1363
1364         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_CHANGE_STRIPE)) {
1365                 struct lod_thread_info *info = lod_env_info(env);
1366                 struct lu_buf *buf = &info->lti_buf;
1367                 struct ost_id *oi = &info->lti_ostid;
1368                 struct lu_fid *fid = &info->lti_fid;
1369                 struct lov_mds_md_v1 *lmm;
1370                 struct lov_ost_data_v1 *objs;
1371                 __u32 magic;
1372
1373                 rc = lod_get_lov_ea(env, lo);
1374                 if (rc <= 0)
1375                         RETURN(rc);
1376
1377                 buf->lb_buf = info->lti_ea_store;
1378                 buf->lb_len = info->lti_ea_store_size;
1379                 lmm = info->lti_ea_store;
1380                 magic = le32_to_cpu(lmm->lmm_magic);
1381                 if (magic == LOV_MAGIC_COMP_V1) {
1382                         struct lov_comp_md_v1 *lcm = buf->lb_buf;
1383                         struct lov_comp_md_entry_v1 *lcme =
1384                                                 &lcm->lcm_entries[0];
1385
1386                         lmm = buf->lb_buf + le32_to_cpu(lcme->lcme_offset);
1387                         magic = le32_to_cpu(lmm->lmm_magic);
1388                 }
1389
1390                 if (magic == LOV_MAGIC_V1)
1391                         objs = &(lmm->lmm_objects[0]);
1392                 else
1393                         objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
1394                 ostid_le_to_cpu(&objs->l_ost_oi, oi);
1395                 ostid_to_fid(fid, oi, le32_to_cpu(objs->l_ost_idx));
1396                 fid->f_oid--;
1397                 fid_to_ostid(fid, oi);
1398                 ostid_cpu_to_le(oi, &objs->l_ost_oi);
1399
1400                 rc = lod_sub_xattr_set(env, next, buf, XATTR_NAME_LOV,
1401                                        LU_XATTR_REPLACE, th);
1402         } else if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PFL_RANGE)) {
1403                 struct lod_thread_info *info = lod_env_info(env);
1404                 struct lu_buf *buf = &info->lti_buf;
1405                 struct lov_comp_md_v1 *lcm;
1406                 struct lov_comp_md_entry_v1 *lcme;
1407
1408                 rc = lod_get_lov_ea(env, lo);
1409                 if (rc <= 0)
1410                         RETURN(rc);
1411
1412                 buf->lb_buf = info->lti_ea_store;
1413                 buf->lb_len = info->lti_ea_store_size;
1414                 lcm = buf->lb_buf;
1415                 if (le32_to_cpu(lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
1416                         RETURN(-EINVAL);
1417
1418                 le32_add_cpu(&lcm->lcm_layout_gen, 1);
1419                 lcme = &lcm->lcm_entries[0];
1420                 le64_add_cpu(&lcme->lcme_extent.e_start, 1);
1421                 le64_add_cpu(&lcme->lcme_extent.e_end, -1);
1422
1423                 rc = lod_sub_xattr_set(env, next, buf, XATTR_NAME_LOV,
1424                                        LU_XATTR_REPLACE, th);
1425         }
1426
1427         RETURN(rc);
1428 }
1429
1430 /**
1431  * Implementation of dt_object_operations::do_xattr_get.
1432  *
1433  * If LOV EA is requested from the root object and it's not
1434  * found, then return default striping for the filesystem.
1435  *
1436  * \see dt_object_operations::do_xattr_get() in the API description for details.
1437  */
1438 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
1439                          struct lu_buf *buf, const char *name)
1440 {
1441         struct lod_thread_info *info = lod_env_info(env);
1442         struct lod_device *dev = lu2lod_dev(dt->do_lu.lo_dev);
1443         int is_root;
1444         int rc;
1445         ENTRY;
1446
1447         rc = dt_xattr_get(env, dt_object_child(dt), buf, name);
1448         if (strcmp(name, XATTR_NAME_LMV) == 0) {
1449                 struct lmv_mds_md_v1    *lmv1;
1450                 int                      rc1 = 0;
1451
1452                 if (rc > (typeof(rc))sizeof(*lmv1))
1453                         RETURN(rc);
1454
1455                 if (rc < (typeof(rc))sizeof(*lmv1))
1456                         RETURN(rc = rc > 0 ? -EINVAL : rc);
1457
1458                 if (buf->lb_buf == NULL || buf->lb_len == 0) {
1459                         CLASSERT(sizeof(*lmv1) <= sizeof(info->lti_key));
1460
1461                         info->lti_buf.lb_buf = info->lti_key;
1462                         info->lti_buf.lb_len = sizeof(*lmv1);
1463                         rc = dt_xattr_get(env, dt_object_child(dt),
1464                                           &info->lti_buf, name);
1465                         if (unlikely(rc != sizeof(*lmv1)))
1466                                 RETURN(rc = rc > 0 ? -EINVAL : rc);
1467
1468                         lmv1 = info->lti_buf.lb_buf;
1469                         /* The on-disk LMV EA only contains header, but the
1470                          * returned LMV EA size should contain the space for
1471                          * the FIDs of all shards of the striped directory. */
1472                         if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_V1)
1473                                 rc = lmv_mds_md_size(
1474                                         le32_to_cpu(lmv1->lmv_stripe_count),
1475                                         LMV_MAGIC_V1);
1476                 } else {
1477                         rc1 = lod_load_lmv_shards(env, lod_dt_obj(dt),
1478                                                   buf, false);
1479                 }
1480
1481                 RETURN(rc = rc1 != 0 ? rc1 : rc);
1482         }
1483
1484         if (rc != -ENODATA || !S_ISDIR(dt->do_lu.lo_header->loh_attr & S_IFMT))
1485                 RETURN(rc);
1486
1487         /*
1488          * XXX: Only used by lfsck
1489          *
1490          * lod returns default striping on the real root of the device
1491          * this is like the root stores default striping for the whole
1492          * filesystem. historically we've been using a different approach
1493          * and store it in the config.
1494          */
1495         dt_root_get(env, dev->lod_child, &info->lti_fid);
1496         is_root = lu_fid_eq(&info->lti_fid, lu_object_fid(&dt->do_lu));
1497
1498         if (is_root && strcmp(XATTR_NAME_LOV, name) == 0) {
1499                 struct lov_user_md *lum = buf->lb_buf;
1500                 struct lov_desc    *desc = &dev->lod_desc;
1501
1502                 if (buf->lb_buf == NULL) {
1503                         rc = sizeof(*lum);
1504                 } else if (buf->lb_len >= sizeof(*lum)) {
1505                         lum->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V1);
1506                         lmm_oi_set_seq(&lum->lmm_oi, FID_SEQ_LOV_DEFAULT);
1507                         lmm_oi_set_id(&lum->lmm_oi, 0);
1508                         lmm_oi_cpu_to_le(&lum->lmm_oi, &lum->lmm_oi);
1509                         lum->lmm_pattern = cpu_to_le32(desc->ld_pattern);
1510                         lum->lmm_stripe_size = cpu_to_le32(
1511                                                 desc->ld_default_stripe_size);
1512                         lum->lmm_stripe_count = cpu_to_le16(
1513                                                 desc->ld_default_stripe_count);
1514                         lum->lmm_stripe_offset = cpu_to_le16(
1515                                                 desc->ld_default_stripe_offset);
1516                         rc = sizeof(*lum);
1517                 } else {
1518                         rc = -ERANGE;
1519                 }
1520         }
1521
1522         RETURN(rc);
1523 }
1524
1525 /**
1526  * Verify LVM EA.
1527  *
1528  * Checks that the magic of the stripe is sane.
1529  *
1530  * \param[in] lod       lod device
1531  * \param[in] lum       a buffer storing LMV EA to verify
1532  *
1533  * \retval              0 if the EA is sane
1534  * \retval              negative otherwise
1535  */
1536 static int lod_verify_md_striping(struct lod_device *lod,
1537                                   const struct lmv_user_md_v1 *lum)
1538 {
1539         if (unlikely(le32_to_cpu(lum->lum_magic) != LMV_USER_MAGIC)) {
1540                 CERROR("%s: invalid lmv_user_md: magic = %x, "
1541                        "stripe_offset = %d, stripe_count = %u: rc = %d\n",
1542                        lod2obd(lod)->obd_name, le32_to_cpu(lum->lum_magic),
1543                        (int)le32_to_cpu(lum->lum_stripe_offset),
1544                        le32_to_cpu(lum->lum_stripe_count), -EINVAL);
1545                 return -EINVAL;
1546         }
1547
1548         return 0;
1549 }
1550
1551 /**
1552  * Initialize LMV EA for a slave.
1553  *
1554  * Initialize slave's LMV EA from the master's LMV EA.
1555  *
1556  * \param[in] master_lmv        a buffer containing master's EA
1557  * \param[out] slave_lmv        a buffer where slave's EA will be stored
1558  *
1559  */
1560 static void lod_prep_slave_lmv_md(struct lmv_mds_md_v1 *slave_lmv,
1561                                   const struct lmv_mds_md_v1 *master_lmv)
1562 {
1563         *slave_lmv = *master_lmv;
1564         slave_lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_STRIPE);
1565 }
1566
1567 /**
1568  * Generate LMV EA.
1569  *
1570  * Generate LMV EA from the object passed as \a dt. The object must have
1571  * the stripes created and initialized.
1572  *
1573  * \param[in] env       execution environment
1574  * \param[in] dt        object
1575  * \param[out] lmv_buf  buffer storing generated LMV EA
1576  *
1577  * \retval              0 on success
1578  * \retval              negative if failed
1579  */
1580 static int lod_prep_lmv_md(const struct lu_env *env, struct dt_object *dt,
1581                            struct lu_buf *lmv_buf)
1582 {
1583         struct lod_thread_info  *info = lod_env_info(env);
1584         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1585         struct lod_object       *lo = lod_dt_obj(dt);
1586         struct lmv_mds_md_v1    *lmm1;
1587         int                     stripe_count;
1588         int                     type = LU_SEQ_RANGE_ANY;
1589         int                     rc;
1590         __u32                   mdtidx;
1591         ENTRY;
1592
1593         LASSERT(lo->ldo_dir_striped != 0);
1594         LASSERT(lo->ldo_dir_stripe_count > 0);
1595         stripe_count = lo->ldo_dir_stripe_count;
1596         /* Only store the LMV EA heahder on the disk. */
1597         if (info->lti_ea_store_size < sizeof(*lmm1)) {
1598                 rc = lod_ea_store_resize(info, sizeof(*lmm1));
1599                 if (rc != 0)
1600                         RETURN(rc);
1601         } else {
1602                 memset(info->lti_ea_store, 0, sizeof(*lmm1));
1603         }
1604
1605         lmm1 = (struct lmv_mds_md_v1 *)info->lti_ea_store;
1606         lmm1->lmv_magic = cpu_to_le32(LMV_MAGIC);
1607         lmm1->lmv_stripe_count = cpu_to_le32(stripe_count);
1608         lmm1->lmv_hash_type = cpu_to_le32(lo->ldo_dir_hash_type);
1609         rc = lod_fld_lookup(env, lod, lu_object_fid(&dt->do_lu),
1610                             &mdtidx, &type);
1611         if (rc != 0)
1612                 RETURN(rc);
1613
1614         lmm1->lmv_master_mdt_index = cpu_to_le32(mdtidx);
1615         lmv_buf->lb_buf = info->lti_ea_store;
1616         lmv_buf->lb_len = sizeof(*lmm1);
1617
1618         RETURN(rc);
1619 }
1620
1621 /**
1622  * Create in-core represenation for a striped directory.
1623  *
1624  * Parse the buffer containing LMV EA and instantiate LU objects
1625  * representing the stripe objects. The pointers to the objects are
1626  * stored in ldo_stripe field of \a lo. This function is used when
1627  * we need to access an already created object (i.e. load from a disk).
1628  *
1629  * \param[in] env       execution environment
1630  * \param[in] lo        lod object
1631  * \param[in] buf       buffer containing LMV EA
1632  *
1633  * \retval              0 on success
1634  * \retval              negative if failed
1635  */
1636 int lod_parse_dir_striping(const struct lu_env *env, struct lod_object *lo,
1637                            const struct lu_buf *buf)
1638 {
1639         struct lod_thread_info  *info = lod_env_info(env);
1640         struct lod_device       *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
1641         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
1642         struct dt_object        **stripe;
1643         union lmv_mds_md        *lmm = buf->lb_buf;
1644         struct lmv_mds_md_v1    *lmv1 = &lmm->lmv_md_v1;
1645         struct lu_fid           *fid = &info->lti_fid;
1646         unsigned int            i;
1647         int                     rc = 0;
1648         ENTRY;
1649
1650         if (le32_to_cpu(lmv1->lmv_hash_type) & LMV_HASH_FLAG_MIGRATION)
1651                 RETURN(0);
1652
1653         if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_STRIPE) {
1654                 lo->ldo_dir_slave_stripe = 1;
1655                 RETURN(0);
1656         }
1657
1658         if (le32_to_cpu(lmv1->lmv_magic) != LMV_MAGIC_V1)
1659                 RETURN(-EINVAL);
1660
1661         if (le32_to_cpu(lmv1->lmv_stripe_count) < 1)
1662                 RETURN(0);
1663
1664         LASSERT(lo->ldo_stripe == NULL);
1665         OBD_ALLOC(stripe, sizeof(stripe[0]) *
1666                   (le32_to_cpu(lmv1->lmv_stripe_count)));
1667         if (stripe == NULL)
1668                 RETURN(-ENOMEM);
1669
1670         for (i = 0; i < le32_to_cpu(lmv1->lmv_stripe_count); i++) {
1671                 struct dt_device        *tgt_dt;
1672                 struct dt_object        *dto;
1673                 int                     type = LU_SEQ_RANGE_ANY;
1674                 __u32                   idx;
1675
1676                 fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[i]);
1677                 if (!fid_is_sane(fid))
1678                         GOTO(out, rc = -ESTALE);
1679
1680                 rc = lod_fld_lookup(env, lod, fid, &idx, &type);
1681                 if (rc != 0)
1682                         GOTO(out, rc);
1683
1684                 if (idx == lod2lu_dev(lod)->ld_site->ld_seq_site->ss_node_id) {
1685                         tgt_dt = lod->lod_child;
1686                 } else {
1687                         struct lod_tgt_desc     *tgt;
1688
1689                         tgt = LTD_TGT(ltd, idx);
1690                         if (tgt == NULL)
1691                                 GOTO(out, rc = -ESTALE);
1692                         tgt_dt = tgt->ltd_tgt;
1693                 }
1694
1695                 dto = dt_locate_at(env, tgt_dt, fid,
1696                                   lo->ldo_obj.do_lu.lo_dev->ld_site->ls_top_dev,
1697                                   NULL);
1698                 if (IS_ERR(dto))
1699                         GOTO(out, rc = PTR_ERR(dto));
1700
1701                 stripe[i] = dto;
1702         }
1703 out:
1704         lo->ldo_stripe = stripe;
1705         lo->ldo_dir_stripe_count = le32_to_cpu(lmv1->lmv_stripe_count);
1706         lo->ldo_dir_stripes_allocated = le32_to_cpu(lmv1->lmv_stripe_count);
1707         if (rc != 0)
1708                 lod_object_free_striping(env, lo);
1709
1710         RETURN(rc);
1711 }
1712
1713 /**
1714  * Declare create a striped directory.
1715  *
1716  * Declare creating a striped directory with a given stripe pattern on the
1717  * specified MDTs. A striped directory is represented as a regular directory
1718  * - an index listing all the stripes. The stripes point back to the master
1719  * object with ".." and LinkEA. The master object gets LMV EA which
1720  * identifies it as a striped directory. The function allocates FIDs
1721  * for all stripes.
1722  *
1723  * \param[in] env       execution environment
1724  * \param[in] dt        object
1725  * \param[in] attr      attributes to initialize the objects with
1726  * \param[in] dof       type of objects to be created
1727  * \param[in] th        transaction handle
1728  *
1729  * \retval              0 on success
1730  * \retval              negative if failed
1731  */
1732 static int lod_dir_declare_create_stripes(const struct lu_env *env,
1733                                           struct dt_object *dt,
1734                                           struct lu_attr *attr,
1735                                           struct dt_object_format *dof,
1736                                           struct thandle *th)
1737 {
1738         struct lod_thread_info  *info = lod_env_info(env);
1739         struct lu_buf           lmv_buf;
1740         struct lu_buf           slave_lmv_buf;
1741         struct lmv_mds_md_v1    *lmm;
1742         struct lmv_mds_md_v1    *slave_lmm = NULL;
1743         struct dt_insert_rec    *rec = &info->lti_dt_rec;
1744         struct lod_object       *lo = lod_dt_obj(dt);
1745         int                     rc;
1746         __u32                   i;
1747         ENTRY;
1748
1749         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
1750         if (rc != 0)
1751                 GOTO(out, rc);
1752         lmm = lmv_buf.lb_buf;
1753
1754         OBD_ALLOC_PTR(slave_lmm);
1755         if (slave_lmm == NULL)
1756                 GOTO(out, rc = -ENOMEM);
1757
1758         lod_prep_slave_lmv_md(slave_lmm, lmm);
1759         slave_lmv_buf.lb_buf = slave_lmm;
1760         slave_lmv_buf.lb_len = sizeof(*slave_lmm);
1761
1762         if (!dt_try_as_dir(env, dt_object_child(dt)))
1763                 GOTO(out, rc = -EINVAL);
1764
1765         rec->rec_type = S_IFDIR;
1766         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
1767                 struct dt_object        *dto = lo->ldo_stripe[i];
1768                 char                    *stripe_name = info->lti_key;
1769                 struct lu_name          *sname;
1770                 struct linkea_data       ldata          = { NULL };
1771                 struct lu_buf           linkea_buf;
1772
1773                 rc = lod_sub_declare_create(env, dto, attr, NULL, dof, th);
1774                 if (rc != 0)
1775                         GOTO(out, rc);
1776
1777                 if (!dt_try_as_dir(env, dto))
1778                         GOTO(out, rc = -EINVAL);
1779
1780                 rc = lod_sub_declare_ref_add(env, dto, th);
1781                 if (rc != 0)
1782                         GOTO(out, rc);
1783
1784                 rec->rec_fid = lu_object_fid(&dto->do_lu);
1785                 rc = lod_sub_declare_insert(env, dto,
1786                                             (const struct dt_rec *)rec,
1787                                             (const struct dt_key *)dot, th);
1788                 if (rc != 0)
1789                         GOTO(out, rc);
1790
1791                 /* master stripe FID will be put to .. */
1792                 rec->rec_fid = lu_object_fid(&dt->do_lu);
1793                 rc = lod_sub_declare_insert(env, dto,
1794                                             (const struct dt_rec *)rec,
1795                                             (const struct dt_key *)dotdot, th);
1796                 if (rc != 0)
1797                         GOTO(out, rc);
1798
1799                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
1800                     cfs_fail_val != i) {
1801                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
1802                             cfs_fail_val == i)
1803                                 slave_lmm->lmv_master_mdt_index =
1804                                                         cpu_to_le32(i + 1);
1805                         else
1806                                 slave_lmm->lmv_master_mdt_index =
1807                                                         cpu_to_le32(i);
1808                         rc = lod_sub_declare_xattr_set(env, dto, &slave_lmv_buf,
1809                                                        XATTR_NAME_LMV, 0, th);
1810                         if (rc != 0)
1811                                 GOTO(out, rc);
1812                 }
1813
1814                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
1815                     cfs_fail_val == i)
1816                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1817                                 PFID(lu_object_fid(&dto->do_lu)), i + 1);
1818                 else
1819                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%u",
1820                                 PFID(lu_object_fid(&dto->do_lu)), i);
1821
1822                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
1823                 rc = linkea_links_new(&ldata, &info->lti_linkea_buf,
1824                                       sname, lu_object_fid(&dt->do_lu));
1825                 if (rc != 0)
1826                         GOTO(out, rc);
1827
1828                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
1829                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
1830                 rc = lod_sub_declare_xattr_set(env, dto, &linkea_buf,
1831                                                XATTR_NAME_LINK, 0, th);
1832                 if (rc != 0)
1833                         GOTO(out, rc);
1834
1835                 rec->rec_fid = lu_object_fid(&dto->do_lu);
1836                 rc = lod_sub_declare_insert(env, dt_object_child(dt),
1837                                             (const struct dt_rec *)rec,
1838                                             (const struct dt_key *)stripe_name,
1839                                             th);
1840                 if (rc != 0)
1841                         GOTO(out, rc);
1842
1843                 rc = lod_sub_declare_ref_add(env, dt_object_child(dt), th);
1844                 if (rc != 0)
1845                         GOTO(out, rc);
1846         }
1847
1848         rc = lod_sub_declare_xattr_set(env, dt_object_child(dt),
1849                                        &lmv_buf, XATTR_NAME_LMV, 0, th);
1850         if (rc != 0)
1851                 GOTO(out, rc);
1852 out:
1853         if (slave_lmm != NULL)
1854                 OBD_FREE_PTR(slave_lmm);
1855
1856         RETURN(rc);
1857 }
1858
1859 static int lod_prep_md_striped_create(const struct lu_env *env,
1860                                       struct dt_object *dt,
1861                                       struct lu_attr *attr,
1862                                       const struct lmv_user_md_v1 *lum,
1863                                       struct dt_object_format *dof,
1864                                       struct thandle *th)
1865 {
1866         struct lod_device       *lod = lu2lod_dev(dt->do_lu.lo_dev);
1867         struct lod_tgt_descs    *ltd = &lod->lod_mdt_descs;
1868         struct lod_object       *lo = lod_dt_obj(dt);
1869         struct dt_object        **stripe;
1870         __u32                   stripe_count;
1871         int                     *idx_array;
1872         __u32                   master_index;
1873         int                     rc = 0;
1874         __u32                   i;
1875         __u32                   j;
1876         bool                    is_specific = false;
1877         ENTRY;
1878
1879         /* The lum has been verifed in lod_verify_md_striping */
1880         LASSERT(le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC ||
1881                 le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC_SPECIFIC);
1882         LASSERT(le32_to_cpu(lum->lum_stripe_count) > 0);
1883
1884         stripe_count = le32_to_cpu(lum->lum_stripe_count);
1885
1886         OBD_ALLOC(idx_array, sizeof(idx_array[0]) * stripe_count);
1887         if (idx_array == NULL)
1888                 RETURN(-ENOMEM);
1889
1890         OBD_ALLOC(stripe, sizeof(stripe[0]) * stripe_count);
1891         if (stripe == NULL)
1892                 GOTO(out_free, rc = -ENOMEM);
1893
1894         /* Start index must be the master MDT */
1895         master_index = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id;
1896         idx_array[0] = master_index;
1897         if (le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC_SPECIFIC) {
1898                 is_specific = true;
1899                 for (i = 1; i < stripe_count; i++)
1900                         idx_array[i] = le32_to_cpu(lum->lum_objects[i].lum_mds);
1901         }
1902
1903         for (i = 0; i < stripe_count; i++) {
1904                 struct lod_tgt_desc     *tgt = NULL;
1905                 struct dt_object        *dto;
1906                 struct lu_fid           fid = { 0 };
1907                 int                     idx;
1908                 struct lu_object_conf   conf = { 0 };
1909                 struct dt_device        *tgt_dt = NULL;
1910
1911                 /* Try to find next avaible target */
1912                 idx = idx_array[i];
1913                 for (j = 0; j < lod->lod_remote_mdt_count;
1914                      j++, idx = (idx + 1) % (lod->lod_remote_mdt_count + 1)) {
1915                         bool already_allocated = false;
1916                         __u32 k;
1917
1918                         CDEBUG(D_INFO, "try idx %d, mdt cnt %u, allocated %u\n",
1919                                idx, lod->lod_remote_mdt_count + 1, i);
1920
1921                         if (likely(!is_specific &&
1922                                    !OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE))) {
1923                                 /* check whether the idx already exists
1924                                  * in current allocated array */
1925                                 for (k = 0; k < i; k++) {
1926                                         if (idx_array[k] == idx) {
1927                                                 already_allocated = true;
1928                                                 break;
1929                                         }
1930                                 }
1931
1932                                 if (already_allocated)
1933                                         continue;
1934                         }
1935
1936                         /* Sigh, this index is not in the bitmap, let's check
1937                          * next available target */
1938                         if (!cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx) &&
1939                             idx != master_index)
1940                                 continue;
1941
1942                         if (idx == master_index) {
1943                                 /* Allocate the FID locally */
1944                                 rc = obd_fid_alloc(env, lod->lod_child_exp,
1945                                                    &fid, NULL);
1946                                 if (rc < 0)
1947                                         GOTO(out_put, rc);
1948                                 tgt_dt = lod->lod_child;
1949                                 break;
1950                         }
1951
1952                         /* check the status of the OSP */
1953                         tgt = LTD_TGT(ltd, idx);
1954                         if (tgt == NULL)
1955                                 continue;
1956
1957                         tgt_dt = tgt->ltd_tgt;
1958                         rc = dt_statfs(env, tgt_dt, NULL);
1959                         if (rc) {
1960                                 /* this OSP doesn't feel well */
1961                                 rc = 0;
1962                                 continue;
1963                         }
1964
1965                         rc = obd_fid_alloc(env, tgt->ltd_exp, &fid, NULL);
1966                         if (rc < 0) {
1967                                 rc = 0;
1968                                 continue;
1969                         }
1970
1971                         break;
1972                 }
1973
1974                 /* Can not allocate more stripes */
1975                 if (j == lod->lod_remote_mdt_count) {
1976                         CDEBUG(D_INFO, "%s: require stripes %u only get %d\n",
1977                                lod2obd(lod)->obd_name, stripe_count, i);
1978                         break;
1979                 }
1980
1981                 CDEBUG(D_INFO, "Get idx %d, for stripe %d "DFID"\n",
1982                        idx, i, PFID(&fid));
1983                 idx_array[i] = idx;
1984                 /* Set the start index for next stripe allocation */
1985                 if (!is_specific && i < stripe_count - 1) {
1986                         /*
1987                          * for large dir test, put all other slaves on one
1988                          * remote MDT, otherwise we may save too many local
1989                          * slave locks which will exceed RS_MAX_LOCKS.
1990                          */
1991                         if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE)))
1992                                 idx = master_index;
1993                         idx_array[i + 1] = (idx + 1) %
1994                                            (lod->lod_remote_mdt_count + 1);
1995                 }
1996                 /* tgt_dt and fid must be ready after search avaible OSP
1997                  * in the above loop */
1998                 LASSERT(tgt_dt != NULL);
1999                 LASSERT(fid_is_sane(&fid));
2000                 conf.loc_flags = LOC_F_NEW;
2001                 dto = dt_locate_at(env, tgt_dt, &fid,
2002                                    dt->do_lu.lo_dev->ld_site->ls_top_dev,
2003                                    &conf);
2004                 if (IS_ERR(dto))
2005                         GOTO(out_put, rc = PTR_ERR(dto));
2006                 stripe[i] = dto;
2007         }
2008
2009         lo->ldo_dir_stripe_loaded = 1;
2010         lo->ldo_dir_striped = 1;
2011         lo->ldo_stripe = stripe;
2012         lo->ldo_dir_stripe_count = i;
2013         lo->ldo_dir_stripes_allocated = stripe_count;
2014
2015         if (lo->ldo_dir_stripe_count == 0)
2016                 GOTO(out_put, rc = -ENOSPC);
2017
2018         rc = lod_dir_declare_create_stripes(env, dt, attr, dof, th);
2019         if (rc != 0)
2020                 GOTO(out_put, rc);
2021
2022 out_put:
2023         if (rc < 0) {
2024                 for (i = 0; i < stripe_count; i++)
2025                         if (stripe[i] != NULL)
2026                                 dt_object_put(env, stripe[i]);
2027                 OBD_FREE(stripe, sizeof(stripe[0]) * stripe_count);
2028                 lo->ldo_dir_stripe_count = 0;
2029                 lo->ldo_dir_stripes_allocated = 0;
2030                 lo->ldo_stripe = NULL;
2031         }
2032
2033 out_free:
2034         OBD_FREE(idx_array, sizeof(idx_array[0]) * stripe_count);
2035
2036         RETURN(rc);
2037 }
2038
2039 /**
2040  * Declare create striped md object.
2041  *
2042  * The function declares intention to create a striped directory. This is a
2043  * wrapper for lod_prep_md_striped_create(). The only additional functionality
2044  * is to verify pattern \a lum_buf is good. Check that function for the details.
2045  *
2046  * \param[in] env       execution environment
2047  * \param[in] dt        object
2048  * \param[in] attr      attributes to initialize the objects with
2049  * \param[in] lum_buf   a pattern specifying the number of stripes and
2050  *                      MDT to start from
2051  * \param[in] dof       type of objects to be created
2052  * \param[in] th        transaction handle
2053  *
2054  * \retval              0 on success
2055  * \retval              negative if failed
2056  *
2057  */
2058 static int lod_declare_xattr_set_lmv(const struct lu_env *env,
2059                                      struct dt_object *dt,
2060                                      struct lu_attr *attr,
2061                                      const struct lu_buf *lum_buf,
2062                                      struct dt_object_format *dof,
2063                                      struct thandle *th)
2064 {
2065         struct lod_object       *lo = lod_dt_obj(dt);
2066         struct lmv_user_md_v1   *lum = lum_buf->lb_buf;
2067         int                     rc;
2068         ENTRY;
2069
2070         LASSERT(lum != NULL);
2071
2072         CDEBUG(D_INFO, "lum magic = %x count = %u offset = %d\n",
2073                le32_to_cpu(lum->lum_magic), le32_to_cpu(lum->lum_stripe_count),
2074                (int)le32_to_cpu(lum->lum_stripe_offset));
2075
2076         if (lo->ldo_dir_stripe_count == 0)
2077                 GOTO(out, rc = 0);
2078
2079         /* prepare dir striped objects */
2080         rc = lod_prep_md_striped_create(env, dt, attr, lum, dof, th);
2081         if (rc != 0) {
2082                 /* failed to create striping, let's reset
2083                  * config so that others don't get confused */
2084                 lod_object_free_striping(env, lo);
2085                 GOTO(out, rc);
2086         }
2087 out:
2088         RETURN(rc);
2089 }
2090
2091 /**
2092  * Implementation of dt_object_operations::do_declare_xattr_set.
2093  *
2094  * Used with regular (non-striped) objects. Basically it
2095  * initializes the striping information and applies the
2096  * change to all the stripes.
2097  *
2098  * \see dt_object_operations::do_declare_xattr_set() in the API description
2099  * for details.
2100  */
2101 static int lod_dir_declare_xattr_set(const struct lu_env *env,
2102                                      struct dt_object *dt,
2103                                      const struct lu_buf *buf,
2104                                      const char *name, int fl,
2105                                      struct thandle *th)
2106 {
2107         struct dt_object        *next = dt_object_child(dt);
2108         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2109         struct lod_object       *lo = lod_dt_obj(dt);
2110         int                     i;
2111         int                     rc;
2112         ENTRY;
2113
2114         if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
2115                 struct lmv_user_md_v1 *lum;
2116
2117                 LASSERT(buf != NULL && buf->lb_buf != NULL);
2118                 lum = buf->lb_buf;
2119                 rc = lod_verify_md_striping(d, lum);
2120                 if (rc != 0)
2121                         RETURN(rc);
2122         } else if (strcmp(name, XATTR_NAME_LOV) == 0) {
2123                 rc = lod_verify_striping(d, lo, buf, false);
2124                 if (rc != 0)
2125                         RETURN(rc);
2126         }
2127
2128         rc = lod_sub_declare_xattr_set(env, next, buf, name, fl, th);
2129         if (rc != 0)
2130                 RETURN(rc);
2131
2132         /* Note: Do not set LinkEA on sub-stripes, otherwise
2133          * it will confuse the fid2path process(see mdt_path_current()).
2134          * The linkEA between master and sub-stripes is set in
2135          * lod_xattr_set_lmv(). */
2136         if (strcmp(name, XATTR_NAME_LINK) == 0)
2137                 RETURN(0);
2138
2139         /* set xattr to each stripes, if needed */
2140         rc = lod_load_striping(env, lo);
2141         if (rc != 0)
2142                 RETURN(rc);
2143
2144         if (lo->ldo_dir_stripe_count == 0)
2145                 RETURN(0);
2146
2147         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
2148                 LASSERT(lo->ldo_stripe[i]);
2149
2150                 rc = lod_sub_declare_xattr_set(env, lo->ldo_stripe[i],
2151                                                buf, name, fl, th);
2152                 if (rc != 0)
2153                         break;
2154         }
2155
2156         RETURN(rc);
2157 }
2158
2159 static int
2160 lod_obj_stripe_replace_parent_fid_cb(const struct lu_env *env,
2161                                      struct lod_object *lo,
2162                                      struct dt_object *dt, struct thandle *th,
2163                                      int comp_idx, int stripe_idx,
2164                                      struct lod_obj_stripe_cb_data *data)
2165 {
2166         struct lod_thread_info *info = lod_env_info(env);
2167         struct lod_layout_component *comp = &lo->ldo_comp_entries[comp_idx];
2168         struct filter_fid *ff = &info->lti_ff;
2169         struct lu_buf *buf = &info->lti_buf;
2170         int rc;
2171
2172         buf->lb_buf = ff;
2173         buf->lb_len = sizeof(*ff);
2174         rc = dt_xattr_get(env, dt, buf, XATTR_NAME_FID);
2175         if (rc < 0) {
2176                 if (rc == -ENODATA)
2177                         return 0;
2178                 return rc;
2179         }
2180
2181         filter_fid_le_to_cpu(ff, ff, sizeof(*ff));
2182         if (lu_fid_eq(lu_object_fid(&lo->ldo_obj.do_lu), &ff->ff_parent) &&
2183             ff->ff_layout.ol_comp_id == comp->llc_id)
2184                 return 0;
2185
2186         /* rewrite filter_fid */
2187         memset(ff, 0, sizeof(*ff));
2188         ff->ff_parent = *lu_object_fid(&lo->ldo_obj.do_lu);
2189         ff->ff_parent.f_ver = stripe_idx;
2190         ff->ff_layout.ol_stripe_size = comp->llc_stripe_size;
2191         ff->ff_layout.ol_stripe_count = comp->llc_stripe_count;
2192         ff->ff_layout.ol_comp_id = comp->llc_id;
2193         ff->ff_layout.ol_comp_start = comp->llc_extent.e_start;
2194         ff->ff_layout.ol_comp_end = comp->llc_extent.e_end;
2195         filter_fid_cpu_to_le(ff, ff, sizeof(*ff));
2196
2197         if (data->locd_declare)
2198                 rc = lod_sub_declare_xattr_set(env, dt, buf, XATTR_NAME_FID,
2199                                                LU_XATTR_REPLACE, th);
2200         else
2201                 rc = lod_sub_xattr_set(env, dt, buf, XATTR_NAME_FID,
2202                                        LU_XATTR_REPLACE, th);
2203
2204         return rc;
2205 }
2206
2207 /**
2208  * Reset parent FID on OST object
2209  *
2210  * Replace parent FID with @dt object FID, which is only called during migration
2211  * to reset the parent FID after the MDT object is migrated to the new MDT, i.e.
2212  * the FID is changed.
2213  *
2214  * \param[in] env execution environment
2215  * \param[in] dt dt_object whose stripes's parent FID will be reset
2216  * \parem[in] th thandle
2217  * \param[in] declare if it is declare
2218  *
2219  * \retval      0 if reset succeeds
2220  * \retval      negative errno if reset fails
2221  */
2222 static int lod_replace_parent_fid(const struct lu_env *env,
2223                                   struct dt_object *dt,
2224                                   struct thandle *th, bool declare)
2225 {
2226         struct lod_object *lo = lod_dt_obj(dt);
2227         struct lod_thread_info  *info = lod_env_info(env);
2228         struct lu_buf *buf = &info->lti_buf;
2229         struct filter_fid *ff;
2230         struct lod_obj_stripe_cb_data data = { { 0 } };
2231         int rc;
2232         ENTRY;
2233
2234         LASSERT(S_ISREG(dt->do_lu.lo_header->loh_attr));
2235
2236         /* set xattr to each stripes, if needed */
2237         rc = lod_load_striping(env, lo);
2238         if (rc != 0)
2239                 RETURN(rc);
2240
2241         if (!lod_obj_is_striped(dt))
2242                 RETURN(0);
2243
2244         if (info->lti_ea_store_size < sizeof(*ff)) {
2245                 rc = lod_ea_store_resize(info, sizeof(*ff));
2246                 if (rc != 0)
2247                         RETURN(rc);
2248         }
2249
2250         buf->lb_buf = info->lti_ea_store;
2251         buf->lb_len = info->lti_ea_store_size;
2252
2253         data.locd_declare = declare;
2254         data.locd_stripe_cb = lod_obj_stripe_replace_parent_fid_cb;
2255         rc = lod_obj_for_each_stripe(env, lo, th, &data);
2256
2257         RETURN(rc);
2258 }
2259
2260 inline __u16 lod_comp_entry_stripe_count(struct lod_object *lo,
2261                                          struct lod_layout_component *entry,
2262                                          bool is_dir)
2263 {
2264         struct lod_device *lod = lu2lod_dev(lod2lu_obj(lo)->lo_dev);
2265
2266         if (is_dir)
2267                 return  0;
2268         else if (lod_comp_inited(entry))
2269                 return entry->llc_stripe_count;
2270         else if ((__u16)-1 == entry->llc_stripe_count)
2271                 return lod->lod_desc.ld_tgt_count;
2272         else
2273                 return lod_get_stripe_count(lod, lo, entry->llc_stripe_count);
2274 }
2275
2276 static int lod_comp_md_size(struct lod_object *lo, bool is_dir)
2277 {
2278         int magic, size = 0, i;
2279         struct lod_layout_component *comp_entries;
2280         __u16 comp_cnt;
2281         bool is_composite;
2282
2283         if (is_dir) {
2284                 comp_cnt = lo->ldo_def_striping->lds_def_comp_cnt;
2285                 comp_entries = lo->ldo_def_striping->lds_def_comp_entries;
2286                 is_composite =
2287                         lo->ldo_def_striping->lds_def_striping_is_composite;
2288         } else {
2289                 comp_cnt = lo->ldo_comp_cnt;
2290                 comp_entries = lo->ldo_comp_entries;
2291                 is_composite = lo->ldo_is_composite;
2292         }
2293
2294
2295         LASSERT(comp_cnt != 0 && comp_entries != NULL);
2296         if (is_composite) {
2297                 size = sizeof(struct lov_comp_md_v1) +
2298                        sizeof(struct lov_comp_md_entry_v1) * comp_cnt;
2299                 LASSERT(size % sizeof(__u64) == 0);
2300         }
2301
2302         for (i = 0; i < comp_cnt; i++) {
2303                 __u16 stripe_count;
2304
2305                 magic = comp_entries[i].llc_pool ? LOV_MAGIC_V3 : LOV_MAGIC_V1;
2306                 stripe_count = lod_comp_entry_stripe_count(lo, &comp_entries[i],
2307                                                            is_dir);
2308                 if (!is_dir && is_composite)
2309                         lod_comp_shrink_stripe_count(&comp_entries[i],
2310                                                      &stripe_count);
2311
2312                 size += lov_user_md_size(stripe_count, magic);
2313                 LASSERT(size % sizeof(__u64) == 0);
2314         }
2315         return size;
2316 }
2317
2318 /**
2319  * Declare component add. The xattr name is XATTR_LUSTRE_LOV.add, and
2320  * the xattr value is binary lov_comp_md_v1 which contains component(s)
2321  * to be added.
2322   *
2323  * \param[in] env       execution environment
2324  * \param[in] dt        dt_object to add components on
2325  * \param[in] buf       buffer contains components to be added
2326  * \parem[in] th        thandle
2327  *
2328  * \retval      0 on success
2329  * \retval      negative errno on failure
2330  */
2331 static int lod_declare_layout_add(const struct lu_env *env,
2332                                   struct dt_object *dt,
2333                                   const struct lu_buf *buf,
2334                                   struct thandle *th)
2335 {
2336         struct lod_thread_info  *info = lod_env_info(env);
2337         struct lod_layout_component *comp_array, *lod_comp, *old_array;
2338         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2339         struct dt_object *next = dt_object_child(dt);
2340         struct lov_desc         *desc = &d->lod_desc;
2341         struct lod_object       *lo = lod_dt_obj(dt);
2342         struct lov_user_md_v3   *v3;
2343         struct lov_comp_md_v1   *comp_v1 = buf->lb_buf;
2344         __u32   magic;
2345         int     i, rc, array_cnt, old_array_cnt;
2346         ENTRY;
2347
2348         LASSERT(lo->ldo_is_composite);
2349
2350         if (lo->ldo_flr_state != LCM_FL_NONE)
2351                 RETURN(-EBUSY);
2352
2353         rc = lod_verify_striping(d, lo, buf, false);
2354         if (rc != 0)
2355                 RETURN(rc);
2356
2357         magic = comp_v1->lcm_magic;
2358         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
2359                 lustre_swab_lov_comp_md_v1(comp_v1);
2360                 magic = comp_v1->lcm_magic;
2361         }
2362
2363         if (magic != LOV_USER_MAGIC_COMP_V1)
2364                 RETURN(-EINVAL);
2365
2366         array_cnt = lo->ldo_comp_cnt + comp_v1->lcm_entry_count;
2367         OBD_ALLOC(comp_array, sizeof(*comp_array) * array_cnt);
2368         if (comp_array == NULL)
2369                 RETURN(-ENOMEM);
2370
2371         memcpy(comp_array, lo->ldo_comp_entries,
2372                sizeof(*comp_array) * lo->ldo_comp_cnt);
2373
2374         for (i = 0; i < comp_v1->lcm_entry_count; i++) {
2375                 struct lov_user_md_v1 *v1;
2376                 struct lu_extent *ext;
2377
2378                 v1 = (struct lov_user_md *)((char *)comp_v1 +
2379                                 comp_v1->lcm_entries[i].lcme_offset);
2380                 ext = &comp_v1->lcm_entries[i].lcme_extent;
2381
2382                 lod_comp = &comp_array[lo->ldo_comp_cnt + i];
2383                 lod_comp->llc_extent.e_start = ext->e_start;
2384                 lod_comp->llc_extent.e_end = ext->e_end;
2385                 lod_comp->llc_stripe_offset = v1->lmm_stripe_offset;
2386                 lod_comp->llc_flags = comp_v1->lcm_entries[i].lcme_flags;
2387
2388                 lod_comp->llc_stripe_count = v1->lmm_stripe_count;
2389                 lod_comp->llc_stripe_size = v1->lmm_stripe_size;
2390                 lod_adjust_stripe_info(lod_comp, desc);
2391
2392                 if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
2393                         v3 = (struct lov_user_md_v3 *) v1;
2394                         if (v3->lmm_pool_name[0] != '\0') {
2395                                 rc = lod_set_pool(&lod_comp->llc_pool,
2396                                                   v3->lmm_pool_name);
2397                                 if (rc)
2398                                         GOTO(error, rc);
2399                         }
2400                 }
2401         }
2402
2403         old_array = lo->ldo_comp_entries;
2404         old_array_cnt = lo->ldo_comp_cnt;
2405
2406         lo->ldo_comp_entries = comp_array;
2407         lo->ldo_comp_cnt = array_cnt;
2408
2409         /* No need to increase layout generation here, it will be increased
2410          * later when generating component ID for the new components */
2411
2412         info->lti_buf.lb_len = lod_comp_md_size(lo, false);
2413         rc = lod_sub_declare_xattr_set(env, next, &info->lti_buf,
2414                                               XATTR_NAME_LOV, 0, th);
2415         if (rc) {
2416                 lo->ldo_comp_entries = old_array;
2417                 lo->ldo_comp_cnt = old_array_cnt;
2418                 GOTO(error, rc);
2419         }
2420
2421         OBD_FREE(old_array, sizeof(*lod_comp) * old_array_cnt);
2422
2423         LASSERT(lo->ldo_mirror_count == 1);
2424         lo->ldo_mirrors[0].lme_end = array_cnt - 1;
2425
2426         RETURN(0);
2427
2428 error:
2429         for (i = lo->ldo_comp_cnt; i < array_cnt; i++) {
2430                 lod_comp = &comp_array[i];
2431                 if (lod_comp->llc_pool != NULL) {
2432                         OBD_FREE(lod_comp->llc_pool,
2433                                  strlen(lod_comp->llc_pool) + 1);
2434                         lod_comp->llc_pool = NULL;
2435                 }
2436         }
2437         OBD_FREE(comp_array, sizeof(*comp_array) * array_cnt);
2438         RETURN(rc);
2439 }
2440
2441 /**
2442  * Declare component set. The xattr is name XATTR_LUSTRE_LOV.set.$field,
2443  * the '$field' can only be 'flags' now. The xattr value is binary
2444  * lov_comp_md_v1 which contains the component ID(s) and the value of
2445  * the field to be modified.
2446  *
2447  * \param[in] env       execution environment
2448  * \param[in] dt        dt_object to be modified
2449  * \param[in] op        operation string, like "set.flags"
2450  * \param[in] buf       buffer contains components to be set
2451  * \parem[in] th        thandle
2452  *
2453  * \retval      0 on success
2454  * \retval      negative errno on failure
2455  */
2456 static int lod_declare_layout_set(const struct lu_env *env,
2457                                   struct dt_object *dt,
2458                                   char *op, const struct lu_buf *buf,
2459                                   struct thandle *th)
2460 {
2461         struct lod_layout_component     *lod_comp;
2462         struct lod_thread_info  *info = lod_env_info(env);
2463         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
2464         struct lod_object       *lo = lod_dt_obj(dt);
2465         struct lov_comp_md_v1   *comp_v1 = buf->lb_buf;
2466         __u32   magic;
2467         int     i, j, rc;
2468         bool    changed = false;
2469         ENTRY;
2470
2471         if (strcmp(op, "set.flags") != 0) {
2472                 CDEBUG(D_LAYOUT, "%s: operation (%s) not supported.\n",
2473                        lod2obd(d)->obd_name, op);
2474                 RETURN(-ENOTSUPP);
2475         }
2476
2477         magic = comp_v1->lcm_magic;
2478         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
2479                 lustre_swab_lov_comp_md_v1(comp_v1);
2480                 magic = comp_v1->lcm_magic;
2481         }
2482
2483         if (magic != LOV_USER_MAGIC_COMP_V1)
2484                 RETURN(-EINVAL);
2485
2486         if (comp_v1->lcm_entry_count == 0) {
2487                 CDEBUG(D_LAYOUT, "%s: entry count is zero.\n",
2488                        lod2obd(d)->obd_name);
2489                 RETURN(-EINVAL);
2490         }
2491
2492         for (i = 0; i < comp_v1->lcm_entry_count; i++) {
2493                 __u32 id = comp_v1->lcm_entries[i].lcme_id;
2494                 __u32 flags = comp_v1->lcm_entries[i].lcme_flags;
2495
2496                 if (flags & LCME_FL_INIT) {
2497                         if (changed)
2498                                 lod_object_free_striping(env, lo);
2499                         RETURN(-EINVAL);
2500                 }
2501
2502                 for (j = 0; j < lo->ldo_comp_cnt; j++) {
2503                         lod_comp = &lo->ldo_comp_entries[j];
2504                         if (id != lod_comp->llc_id)
2505                                 continue;
2506
2507                         if (flags & LCME_FL_NEG) {
2508                                 flags &= ~LCME_FL_NEG;
2509                                 lod_comp->llc_flags &= ~flags;
2510                         } else {
2511                                 lod_comp->llc_flags |= flags;
2512                         }
2513                         changed = true;
2514                 }
2515         }
2516
2517         if (!changed) {
2518                 CDEBUG(D_LAYOUT, "%s: requested component(s) not found.\n",
2519                        lod2obd(d)->obd_name);
2520                 RETURN(-EINVAL);
2521         }
2522
2523         lod_obj_inc_layout_gen(lo);
2524
2525         info->lti_buf.lb_len = lod_comp_md_size(lo, false);
2526         rc = lod_sub_declare_xattr_set(env, dt_object_child(dt), &info->lti_buf,
2527                                        XATTR_NAME_LOV, LU_XATTR_REPLACE, th);
2528         RETURN(rc);
2529 }
2530
2531 /**
2532  * Declare component deletion. The xattr name is XATTR_LUSTRE_LOV.del,
2533  * and the xattr value is a unique component ID or a special lcme_id.
2534  *
2535  * \param[in] env       execution environment
2536  * \param[in] dt        dt_object to be operated on
2537  * \param[in] buf       buffer contains component ID or lcme_id
2538  * \parem[in] th        thandle
2539  *
2540  * \retval      0 on success
2541  * \retval      negative errno on failure
2542  */
2543 static int lod_declare_layout_del(const struct lu_env *env,
2544                                   struct dt_object *dt,
2545                                   const struct lu_buf *buf,
2546                                   struct thandle *th)
2547 {
2548         struct lod_thread_info  *info = lod_env_info(env);
2549         struct dt_object *next = dt_object_child(dt);
2550         struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
2551         struct lod_object *lo = lod_dt_obj(dt);
2552         struct lu_attr *attr = &lod_env_info(env)->lti_attr;
2553         struct lov_comp_md_v1 *comp_v1 = buf->lb_buf;
2554         __u32 magic, id, flags, neg_flags = 0;
2555         int rc, i, j, left;
2556         ENTRY;
2557
2558         LASSERT(lo->ldo_is_composite);
2559
2560         if (lo->ldo_flr_state != LCM_FL_NONE)
2561                 RETURN(-EBUSY);
2562
2563         magic = comp_v1->lcm_magic;
2564         if (magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
2565                 lustre_swab_lov_comp_md_v1(comp_v1);
2566                 magic = comp_v1->lcm_magic;
2567         }
2568
2569         if (magic != LOV_USER_MAGIC_COMP_V1)
2570                 RETURN(-EINVAL);
2571
2572         id = comp_v1->lcm_entries[0].lcme_id;
2573         flags = comp_v1->lcm_entries[0].lcme_flags;
2574
2575         if (id > LCME_ID_MAX || (flags & ~LCME_KNOWN_FLAGS)) {
2576                 CDEBUG(D_LAYOUT, "%s: invalid component id %#x, flags %#x\n",
2577                        lod2obd(d)->obd_name, id, flags);
2578                 RETURN(-EINVAL);
2579         }
2580
2581         if (id != LCME_ID_INVAL && flags != 0) {
2582                 CDEBUG(D_LAYOUT, "%s: specified both id and flags.\n",
2583                        lod2obd(d)->obd_name);
2584                 RETURN(-EINVAL);
2585         }
2586
2587         if (id == LCME_ID_INVAL && !flags) {
2588                 CDEBUG(D_LAYOUT, "%s: no id or flags specified.\n",
2589                        lod2obd(d)->obd_name);
2590                 RETURN(-EINVAL);
2591         }
2592
2593         if (flags & LCME_FL_NEG) {
2594                 neg_flags = flags & ~LCME_FL_NEG;
2595                 flags = 0;
2596         }
2597
2598         left = lo->ldo_comp_cnt;
2599         if (left <= 0)
2600                 RETURN(-EINVAL);
2601
2602         for (i = (lo->ldo_comp_cnt - 1); i >= 0; i--) {
2603                 struct lod_layout_component *lod_comp;
2604
2605                 lod_comp = &lo->ldo_comp_entries[i];
2606
2607                 if (id != LCME_ID_INVAL && id != lod_comp->llc_id)
2608                         continue;
2609                 else if (flags && !(flags & lod_comp->llc_flags))
2610                         continue;
2611                 else if (neg_flags && (neg_flags & lod_comp->llc_flags))
2612                         continue;
2613
2614                 if (left != (i + 1)) {
2615                         CDEBUG(D_LAYOUT, "%s: this deletion will create "
2616                                "a hole.\n", lod2obd(d)->obd_name);
2617                         RETURN(-EINVAL);
2618                 }
2619                 left--;
2620
2621                 /* Mark the component as deleted */
2622                 lod_comp->llc_id = LCME_ID_INVAL;
2623
2624                 /* Not instantiated component */
2625                 if (lod_comp->llc_stripe == NULL)
2626                         continue;
2627
2628                 LASSERT(lod_comp->llc_stripe_count > 0);
2629                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
2630                         struct dt_object *obj = lod_comp->llc_stripe[j];
2631
2632                         if (obj == NULL)
2633                                 continue;
2634                         rc = lod_sub_declare_destroy(env, obj, th);
2635                         if (rc)
2636                                 RETURN(rc);
2637                 }
2638         }
2639
2640         LASSERTF(left >= 0, "left = %d\n", left);
2641         if (left == lo->ldo_comp_cnt) {
2642                 CDEBUG(D_LAYOUT, "%s: requested component id:%#x not found\n",
2643                        lod2obd(d)->obd_name, id);
2644                 RETURN(-EINVAL);
2645         }
2646
2647         memset(attr, 0, sizeof(*attr));
2648         attr->la_valid = LA_SIZE;
2649         rc = lod_sub_declare_attr_set(env, next, attr, th);
2650         if (rc)
2651                 RETURN(rc);
2652
2653         if (left > 0) {
2654                 info->lti_buf.lb_len = lod_comp_md_size(lo, false);
2655                 rc = lod_sub_declare_xattr_set(env, next, &info->lti_buf,
2656                                                XATTR_NAME_LOV, 0, th);
2657         } else {
2658                 rc = lod_sub_declare_xattr_del(env, next, XATTR_NAME_LOV, th);
2659         }
2660
2661         RETURN(rc);
2662 }
2663
2664 /**
2665  * Declare layout add/set/del operations issued by special xattr names:
2666  *
2667  * XATTR_LUSTRE_LOV.add         add component(s) to existing file
2668  * XATTR_LUSTRE_LOV.del         delete component(s) from existing file
2669  * XATTR_LUSTRE_LOV.set.$field  set specified field of certain component(s)
2670  *
2671  * \param[in] env       execution environment
2672  * \param[in] dt        object
2673  * \param[in] name      name of xattr
2674  * \param[in] buf       lu_buf contains xattr value
2675  * \param[in] th        transaction handle
2676  *
2677  * \retval              0 on success
2678  * \retval              negative if failed
2679  */
2680 static int lod_declare_modify_layout(const struct lu_env *env,
2681                                      struct dt_object *dt,
2682                                      const char *name,
2683                                      const struct lu_buf *buf,
2684                                      struct thandle *th)
2685 {
2686         struct lod_device *d = lu2lod_dev(dt->do_lu.lo_dev);
2687         struct lod_object *lo = lod_dt_obj(dt);
2688         struct dt_object *next = dt_object_child(&lo->ldo_obj);
2689         char *op;
2690         int rc, len = strlen(XATTR_LUSTRE_LOV);
2691         ENTRY;
2692
2693         LASSERT(dt_object_exists(dt));
2694
2695         if (strlen(name) <= len || name[len] != '.') {
2696                 CDEBUG(D_LAYOUT, "%s: invalid xattr name: %s\n",
2697                        lod2obd(d)->obd_name, name);
2698                 RETURN(-EINVAL);
2699         }
2700         len++;
2701
2702         dt_write_lock(env, next, 0);
2703         rc = lod_load_striping_locked(env, lo);
2704         if (rc)
2705                 GOTO(unlock, rc);
2706
2707         /* the layout to be modified must be a composite layout */
2708         if (!lo->ldo_is_composite) {
2709                 CDEBUG(D_LAYOUT, "%s: object "DFID" isn't a composite file.\n",
2710                        lod2obd(d)->obd_name, PFID(lu_object_fid(&dt->do_lu)));
2711                 GOTO(unlock, rc = -EINVAL);
2712         }
2713
2714         op = (char *)name + len;
2715         if (strcmp(op, "add") == 0) {
2716                 rc = lod_declare_layout_add(env, dt, buf, th);
2717         } else if (strcmp(op, "del") == 0) {
2718                 rc = lod_declare_layout_del(env, dt, buf, th);
2719         } else if (strncmp(op, "set", strlen("set")) == 0) {
2720                 rc = lod_declare_layout_set(env, dt, op, buf, th);
2721         } else  {
2722                 CDEBUG(D_LAYOUT, "%s: unsupported xattr name:%s\n",
2723                        lod2obd(d)->obd_name, name);
2724                 GOTO(unlock, rc = -ENOTSUPP);
2725         }
2726 unlock:
2727         if (rc)
2728                 lod_object_free_striping(env, lo);
2729         dt_write_unlock(env, next);
2730
2731         RETURN(rc);
2732 }
2733
2734 /**
2735  * Convert a plain file lov_mds_md to a composite layout.
2736  *
2737  * \param[in,out] info  the thread info::lti_ea_store buffer contains little
2738  *                      endian plain file layout
2739  *
2740  * \retval              0 on success, <0 on failure
2741  */
2742 static int lod_layout_convert(struct lod_thread_info *info)
2743 {
2744         struct lov_mds_md *lmm = info->lti_ea_store;
2745         struct lov_mds_md *lmm_save;
2746         struct lov_comp_md_v1 *lcm;
2747         struct lov_comp_md_entry_v1 *lcme;
2748         size_t size;
2749         __u32 blob_size;
2750         int rc = 0;
2751         ENTRY;
2752
2753         /* realloc buffer to a composite layout which contains one component */
2754         blob_size = lov_mds_md_size(le16_to_cpu(lmm->lmm_stripe_count),
2755                                     le32_to_cpu(lmm->lmm_magic));
2756         size = sizeof(*lcm) + sizeof(*lcme) + blob_size;
2757
2758         OBD_ALLOC_LARGE(lmm_save, blob_size);
2759         if (!lmm_save)
2760                 GOTO(out, rc = -ENOMEM);
2761
2762         memcpy(lmm_save, lmm, blob_size);
2763
2764         if (info->lti_ea_store_size < size) {
2765                 rc = lod_ea_store_resize(info, size);
2766                 if (rc)
2767                         GOTO(out, rc);
2768         }
2769
2770         lcm = info->lti_ea_store;
2771         lcm->lcm_magic = cpu_to_le32(LOV_MAGIC_COMP_V1);
2772         lcm->lcm_size = cpu_to_le32(size);
2773         lcm->lcm_layout_gen = cpu_to_le32(le16_to_cpu(
2774                                                 lmm_save->lmm_layout_gen));
2775         lcm->lcm_flags = cpu_to_le16(LCM_FL_NONE);
2776         lcm->lcm_entry_count = cpu_to_le16(1);
2777         lcm->lcm_mirror_count = 0;
2778
2779         lcme = &lcm->lcm_entries[0];
2780         lcme->lcme_flags = cpu_to_le32(LCME_FL_INIT);
2781         lcme->lcme_extent.e_start = 0;
2782         lcme->lcme_extent.e_end = cpu_to_le64(OBD_OBJECT_EOF);
2783         lcme->lcme_offset = cpu_to_le32(sizeof(*lcm) + sizeof(*lcme));
2784         lcme->lcme_size = cpu_to_le32(blob_size);
2785
2786         memcpy((char *)lcm + lcme->lcme_offset, (char *)lmm_save, blob_size);
2787
2788         EXIT;
2789 out:
2790         if (lmm_save)
2791                 OBD_FREE_LARGE(lmm_save, blob_size);
2792         return rc;
2793 }
2794
2795 /**
2796  * Merge layouts to form a mirrored file.
2797  */
2798 static int lod_declare_layout_merge(const struct lu_env *env,
2799                 struct dt_object *dt, const struct lu_buf *mbuf,
2800                 struct thandle *th)
2801 {
2802         struct lod_thread_info  *info = lod_env_info(env);
2803         struct lu_buf           *buf = &info->lti_buf;
2804         struct lod_object       *lo = lod_dt_obj(dt);
2805         struct lov_comp_md_v1   *lcm;
2806         struct lov_comp_md_v1   *cur_lcm;
2807         struct lov_comp_md_v1   *merge_lcm;
2808         struct lov_comp_md_entry_v1     *lcme;
2809         size_t size = 0;
2810         size_t offset;
2811         __u16 cur_entry_count;
2812         __u16 merge_entry_count;
2813         __u32 id = 0;
2814         __u16 mirror_id = 0;
2815         __u32 mirror_count;
2816         int     rc, i;
2817         ENTRY;
2818
2819         merge_lcm = mbuf->lb_buf;
2820         if (mbuf->lb_len < sizeof(*merge_lcm))
2821                 RETURN(-EINVAL);
2822
2823         /* must be an existing layout from disk */
2824         if (le32_to_cpu(merge_lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
2825                 RETURN(-EINVAL);
2826
2827         merge_entry_count = le16_to_cpu(merge_lcm->lcm_entry_count);
2828
2829         /* do not allow to merge two mirrored files */
2830         if (le16_to_cpu(merge_lcm->lcm_mirror_count))
2831                 RETURN(-EBUSY);
2832
2833         /* verify the target buffer */
2834         rc = lod_get_lov_ea(env, lo);
2835         if (rc <= 0)
2836                 RETURN(rc ? : -ENODATA);
2837
2838         cur_lcm = info->lti_ea_store;
2839         switch (le32_to_cpu(cur_lcm->lcm_magic)) {
2840         case LOV_MAGIC_V1:
2841         case LOV_MAGIC_V3:
2842                 rc = lod_layout_convert(info);
2843                 break;
2844         case LOV_MAGIC_COMP_V1:
2845                 rc = 0;
2846                 break;
2847         default:
2848                 rc = -EINVAL;
2849         }
2850         if (rc)
2851                 RETURN(rc);
2852
2853         /* info->lti_ea_store could be reallocated in lod_layout_convert() */
2854         cur_lcm = info->lti_ea_store;
2855         cur_entry_count = le16_to_cpu(cur_lcm->lcm_entry_count);
2856
2857         /* 'lcm_mirror_count + 1' is the current # of mirrors the file has */
2858         mirror_count = le16_to_cpu(cur_lcm->lcm_mirror_count) + 1;
2859         if (mirror_count + 1 > LUSTRE_MIRROR_COUNT_MAX)
2860                 RETURN(-ERANGE);
2861
2862         /* size of new layout */
2863         size = le32_to_cpu(cur_lcm->lcm_size) +
2864                le32_to_cpu(merge_lcm->lcm_size) - sizeof(*cur_lcm);
2865
2866         memset(buf, 0, sizeof(*buf));
2867         lu_buf_alloc(buf, size);
2868         if (buf->lb_buf == NULL)
2869                 RETURN(-ENOMEM);
2870
2871         lcm = buf->lb_buf;
2872         memcpy(lcm, cur_lcm, sizeof(*lcm) + cur_entry_count * sizeof(*lcme));
2873
2874         offset = sizeof(*lcm) +
2875                  sizeof(*lcme) * (cur_entry_count + merge_entry_count);
2876         for (i = 0; i < cur_entry_count; i++) {
2877                 struct lov_comp_md_entry_v1 *cur_lcme;
2878
2879                 lcme = &lcm->lcm_entries[i];
2880                 cur_lcme = &cur_lcm->lcm_entries[i];
2881
2882                 lcme->lcme_offset = cpu_to_le32(offset);
2883                 memcpy((char *)lcm + offset,
2884                        (char *)cur_lcm + le32_to_cpu(cur_lcme->lcme_offset),
2885                        le32_to_cpu(lcme->lcme_size));
2886
2887                 offset += le32_to_cpu(lcme->lcme_size);
2888
2889                 if (mirror_count == 1) {
2890                         /* new mirrored file, create new mirror ID */
2891                         id = pflr_id(1, i + 1);
2892                         lcme->lcme_id = cpu_to_le32(id);
2893                 }
2894
2895                 id = MAX(le32_to_cpu(lcme->lcme_id), id);
2896         }
2897
2898         mirror_id = mirror_id_of(id) + 1;
2899         for (i = 0; i < merge_entry_count; i++) {
2900                 struct lov_comp_md_entry_v1 *merge_lcme;
2901
2902                 merge_lcme = &merge_lcm->lcm_entries[i];
2903                 lcme = &lcm->lcm_entries[cur_entry_count + i];
2904
2905                 *lcme = *merge_lcme;
2906                 lcme->lcme_offset = cpu_to_le32(offset);
2907
2908                 id = pflr_id(mirror_id, i + 1);
2909                 lcme->lcme_id = cpu_to_le32(id);
2910
2911                 memcpy((char *)lcm + offset,
2912                        (char *)merge_lcm + le32_to_cpu(merge_lcme->lcme_offset),
2913                        le32_to_cpu(lcme->lcme_size));
2914
2915                 offset += le32_to_cpu(lcme->lcme_size);
2916         }
2917
2918         /* fixup layout information */
2919         lod_obj_inc_layout_gen(lo);
2920         lcm->lcm_layout_gen = cpu_to_le32(lo->ldo_layout_gen);
2921         lcm->lcm_size = cpu_to_le32(size);
2922         lcm->lcm_entry_count = cpu_to_le16(cur_entry_count + merge_entry_count);
2923         lcm->lcm_mirror_count = cpu_to_le16(mirror_count);
2924         if ((le16_to_cpu(lcm->lcm_flags) & LCM_FL_FLR_MASK) == LCM_FL_NONE)
2925                 lcm->lcm_flags = cpu_to_le32(LCM_FL_RDONLY);
2926
2927         LASSERT(dt_write_locked(env, dt_object_child(dt)));
2928         lod_object_free_striping(env, lo);
2929         rc = lod_parse_striping(env, lo, buf);
2930         if (rc)
2931                 GOTO(out, rc);
2932
2933         rc = lod_sub_declare_xattr_set(env, dt_object_child(dt), buf,
2934                                         XATTR_NAME_LOV, LU_XATTR_REPLACE, th);
2935
2936 out:
2937         lu_buf_free(buf);
2938         RETURN(rc);
2939 }
2940
2941 /**
2942  * Split layouts, just set the LOVEA with the layout from mbuf.
2943  */
2944 static int lod_declare_layout_split(const struct lu_env *env,
2945                 struct dt_object *dt, const struct lu_buf *mbuf,
2946                 struct thandle *th)
2947 {
2948         struct lod_object *lo = lod_dt_obj(dt);
2949         struct lov_comp_md_v1 *lcm = mbuf->lb_buf;
2950         int rc;
2951         ENTRY;
2952
2953         lod_obj_inc_layout_gen(lo);
2954         lcm->lcm_layout_gen = cpu_to_le32(lo->ldo_layout_gen);
2955
2956         lod_object_free_striping(env, lo);
2957         rc = lod_parse_striping(env, lo, mbuf);
2958         if (rc)
2959                 RETURN(rc);
2960
2961         rc = lod_sub_declare_xattr_set(env, dt_object_child(dt), mbuf,
2962                                        XATTR_NAME_LOV, LU_XATTR_REPLACE, th);
2963         RETURN(rc);
2964 }
2965
2966 /**
2967  * Implementation of dt_object_operations::do_declare_xattr_set.
2968  *
2969  * \see dt_object_operations::do_declare_xattr_set() in the API description
2970  * for details.
2971  *
2972  * the extension to the API:
2973  *   - declaring LOVEA requests striping creation
2974  *   - LU_XATTR_REPLACE means layout swap
2975  */
2976 static int lod_declare_xattr_set(const struct lu_env *env,
2977                                  struct dt_object *dt,
2978                                  const struct lu_buf *buf,
2979                                  const char *name, int fl,
2980                                  struct thandle *th)
2981 {
2982         struct dt_object *next = dt_object_child(dt);
2983         struct lu_attr   *attr = &lod_env_info(env)->lti_attr;
2984         __u32             mode;
2985         int               rc;
2986         ENTRY;
2987
2988         mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
2989         if ((S_ISREG(mode) || mode == 0) &&
2990             !(fl & (LU_XATTR_REPLACE | LU_XATTR_MERGE | LU_XATTR_SPLIT)) &&
2991             (strcmp(name, XATTR_NAME_LOV) == 0 ||
2992              strcmp(name, XATTR_LUSTRE_LOV) == 0)) {
2993                 /*
2994                  * this is a request to create object's striping.
2995                  *
2996                  * allow to declare predefined striping on a new (!mode) object
2997                  * which is supposed to be replay of regular file creation
2998                  * (when LOV setting is declared)
2999                  *
3000                  * LU_XATTR_REPLACE is set to indicate a layout swap
3001                  */
3002                 if (dt_object_exists(dt)) {
3003                         rc = dt_attr_get(env, next, attr);
3004                         if (rc)
3005                                 RETURN(rc);
3006                 } else {
3007                         memset(attr, 0, sizeof(*attr));
3008                         attr->la_valid = LA_TYPE | LA_MODE;
3009                         attr->la_mode = S_IFREG;
3010                 }
3011                 rc = lod_declare_striped_create(env, dt, attr, buf, th);
3012         } else if (fl & LU_XATTR_MERGE) {
3013                 LASSERT(strcmp(name, XATTR_NAME_LOV) == 0 ||
3014                         strcmp(name, XATTR_LUSTRE_LOV) == 0);
3015                 rc = lod_declare_layout_merge(env, dt, buf, th);
3016         } else if (fl & LU_XATTR_SPLIT) {
3017                 LASSERT(strcmp(name, XATTR_NAME_LOV) == 0 ||
3018                         strcmp(name, XATTR_LUSTRE_LOV) == 0);
3019                 rc = lod_declare_layout_split(env, dt, buf, th);
3020         } else if (S_ISREG(mode) &&
3021                    strlen(name) > strlen(XATTR_LUSTRE_LOV) + 1 &&
3022                    strncmp(name, XATTR_LUSTRE_LOV,
3023                            strlen(XATTR_LUSTRE_LOV)) == 0) {
3024                 /*
3025                  * this is a request to modify object's striping.
3026                  * add/set/del component(s).
3027                  */
3028                 if (!dt_object_exists(dt))
3029                         RETURN(-ENOENT);
3030
3031                 rc = lod_declare_modify_layout(env, dt, name, buf, th);
3032         } else if (S_ISDIR(mode)) {
3033                 rc = lod_dir_declare_xattr_set(env, dt, buf, name, fl, th);
3034         } else if (strcmp(name, XATTR_NAME_FID) == 0) {
3035                 rc = lod_replace_parent_fid(env, dt, th, true);
3036         } else {
3037                 rc = lod_sub_declare_xattr_set(env, next, buf, name, fl, th);
3038         }
3039
3040         RETURN(rc);
3041 }
3042
3043 /**
3044  * Apply xattr changes to the object.
3045  *
3046  * Applies xattr changes to the object and the stripes if the latter exist.
3047  *
3048  * \param[in] env       execution environment
3049  * \param[in] dt        object
3050  * \param[in] buf       buffer pointing to the new value of xattr
3051  * \param[in] name      name of xattr
3052  * \param[in] fl        flags
3053  * \param[in] th        transaction handle
3054  *
3055  * \retval              0 on success
3056  * \retval              negative if failed
3057  */
3058 static int lod_xattr_set_internal(const struct lu_env *env,
3059                                   struct dt_object *dt,
3060                                   const struct lu_buf *buf,
3061                                   const char *name, int fl,
3062                                   struct thandle *th)
3063 {
3064         struct dt_object        *next = dt_object_child(dt);
3065         struct lod_object       *lo = lod_dt_obj(dt);
3066         int                     rc;
3067         int                     i;
3068         ENTRY;
3069
3070         rc = lod_sub_xattr_set(env, next, buf, name, fl, th);
3071         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
3072                 RETURN(rc);
3073
3074         /* Note: Do not set LinkEA on sub-stripes, otherwise
3075          * it will confuse the fid2path process(see mdt_path_current()).
3076          * The linkEA between master and sub-stripes is set in
3077          * lod_xattr_set_lmv(). */
3078         if (lo->ldo_dir_stripe_count == 0 || strcmp(name, XATTR_NAME_LINK) == 0)
3079                 RETURN(0);
3080
3081         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
3082                 LASSERT(lo->ldo_stripe[i]);
3083
3084                 rc = lod_sub_xattr_set(env, lo->ldo_stripe[i], buf, name,
3085                                        fl, th);
3086                 if (rc != 0)
3087                         break;
3088         }
3089
3090         RETURN(rc);
3091 }
3092
3093 /**
3094  * Delete an extended attribute.
3095  *
3096  * Deletes specified xattr from the object and the stripes if the latter exist.
3097  *
3098  * \param[in] env       execution environment
3099  * \param[in] dt        object
3100  * \param[in] name      name of xattr
3101  * \param[in] th        transaction handle
3102  *
3103  * \retval              0 on success
3104  * \retval              negative if failed
3105  */
3106 static int lod_xattr_del_internal(const struct lu_env *env,
3107                                   struct dt_object *dt,
3108                                   const char *name, struct thandle *th)
3109 {
3110         struct dt_object        *next = dt_object_child(dt);
3111         struct lod_object       *lo = lod_dt_obj(dt);
3112         int                     rc;
3113         int                     i;
3114         ENTRY;
3115
3116         rc = lod_sub_xattr_del(env, next, name, th);
3117         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
3118                 RETURN(rc);
3119
3120         if (lo->ldo_dir_stripe_count == 0)
3121                 RETURN(rc);
3122
3123         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
3124                 LASSERT(lo->ldo_stripe[i]);
3125
3126                 rc = lod_sub_xattr_del(env, lo->ldo_stripe[i], name, th);
3127                 if (rc != 0)
3128                         break;
3129         }
3130
3131         RETURN(rc);
3132 }
3133
3134 /**
3135  * Set default striping on a directory.
3136  *
3137  * Sets specified striping on a directory object unless it matches the default
3138  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
3139  * EA. This striping will be used when regular file is being created in this
3140  * directory.
3141  *
3142  * \param[in] env       execution environment
3143  * \param[in] dt        the striped object
3144  * \param[in] buf       buffer with the striping
3145  * \param[in] name      name of EA
3146  * \param[in] fl        xattr flag (see OSD API description)
3147  * \param[in] th        transaction handle
3148  *
3149  * \retval              0 on success
3150  * \retval              negative if failed
3151  */
3152 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
3153                                     struct dt_object *dt,
3154                                     const struct lu_buf *buf,
3155                                     const char *name, int fl,
3156                                     struct thandle *th)
3157 {
3158         struct lov_user_md_v1   *lum;
3159         struct lov_user_md_v3   *v3 = NULL;
3160         const char              *pool_name = NULL;
3161         int                      rc;
3162         bool                     is_del;
3163         ENTRY;
3164
3165         LASSERT(buf != NULL && buf->lb_buf != NULL);
3166         lum = buf->lb_buf;
3167
3168         switch (lum->lmm_magic) {
3169         case LOV_USER_MAGIC_V3:
3170                 v3 = buf->lb_buf;
3171                 if (v3->lmm_pool_name[0] != '\0')
3172                         pool_name = v3->lmm_pool_name;
3173                 /* fall through */
3174         case LOV_USER_MAGIC_V1:
3175                 /* if { size, offset, count } = { 0, -1, 0 } and no pool
3176                  * (i.e. all default values specified) then delete default
3177                  * striping from dir. */
3178                 CDEBUG(D_LAYOUT,
3179                        "set default striping: sz %u # %u offset %d %s %s\n",
3180                        (unsigned)lum->lmm_stripe_size,
3181                        (unsigned)lum->lmm_stripe_count,
3182                        (int)lum->lmm_stripe_offset,
3183                        v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
3184
3185                 is_del = LOVEA_DELETE_VALUES(lum->lmm_stripe_size,
3186                                              lum->lmm_stripe_count,
3187                                              lum->lmm_stripe_offset,
3188                                              pool_name);
3189                 break;
3190         case LOV_USER_MAGIC_COMP_V1:
3191                 is_del = false;
3192                 break;
3193         default:
3194                 CERROR("Invalid magic %x\n", lum->lmm_magic);
3195                 RETURN(-EINVAL);
3196         }
3197
3198         if (is_del) {
3199                 rc = lod_xattr_del_internal(env, dt, name, th);
3200                 if (rc == -ENODATA)
3201                         rc = 0;
3202         } else {
3203                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th);
3204         }
3205
3206         RETURN(rc);
3207 }
3208
3209 /**
3210  * Set default striping on a directory object.
3211  *
3212  * Sets specified striping on a directory object unless it matches the default
3213  * striping (LOVEA_DELETE_VALUES() macro). In the latter case remove existing
3214  * EA. This striping will be used when a new directory is being created in the
3215  * directory.
3216  *
3217  * \param[in] env       execution environment
3218  * \param[in] dt        the striped object
3219  * \param[in] buf       buffer with the striping
3220  * \param[in] name      name of EA
3221  * \param[in] fl        xattr flag (see OSD API description)
3222  * \param[in] th        transaction handle
3223  *
3224  * \retval              0 on success
3225  * \retval              negative if failed
3226  */
3227 static int lod_xattr_set_default_lmv_on_dir(const struct lu_env *env,
3228                                             struct dt_object *dt,
3229                                             const struct lu_buf *buf,
3230                                             const char *name, int fl,
3231                                             struct thandle *th)
3232 {
3233         struct lmv_user_md_v1   *lum;
3234         int                      rc;
3235         ENTRY;
3236
3237         LASSERT(buf != NULL && buf->lb_buf != NULL);
3238         lum = buf->lb_buf;
3239
3240         CDEBUG(D_OTHER, "set default stripe_count # %u stripe_offset %d\n",
3241               le32_to_cpu(lum->lum_stripe_count),
3242               (int)le32_to_cpu(lum->lum_stripe_offset));
3243
3244         if (LMVEA_DELETE_VALUES((le32_to_cpu(lum->lum_stripe_count)),
3245                                  le32_to_cpu(lum->lum_stripe_offset)) &&
3246                                 le32_to_cpu(lum->lum_magic) == LMV_USER_MAGIC) {
3247                 rc = lod_xattr_del_internal(env, dt, name, th);
3248                 if (rc == -ENODATA)
3249                         rc = 0;
3250         } else {
3251                 rc = lod_xattr_set_internal(env, dt, buf, name, fl, th);
3252                 if (rc != 0)
3253                         RETURN(rc);
3254         }
3255
3256         RETURN(rc);
3257 }
3258
3259 /**
3260  * Turn directory into a striped directory.
3261  *
3262  * During replay the client sends the striping created before MDT
3263  * failure, then the layer above LOD sends this defined striping
3264  * using ->do_xattr_set(), so LOD uses this method to replay creation
3265  * of the stripes. Notice the original information for the striping
3266  * (#stripes, FIDs, etc) was transferred in declare path.
3267  *
3268  * \param[in] env       execution environment
3269  * \param[in] dt        the striped object
3270  * \param[in] buf       not used currently
3271  * \param[in] name      not used currently
3272  * \param[in] fl        xattr flag (see OSD API description)
3273  * \param[in] th        transaction handle
3274  *
3275  * \retval              0 on success
3276  * \retval              negative if failed
3277  */
3278 static int lod_xattr_set_lmv(const struct lu_env *env, struct dt_object *dt,
3279                              const struct lu_buf *buf, const char *name,
3280                              int fl, struct thandle *th)
3281 {
3282         struct lod_object       *lo = lod_dt_obj(dt);
3283         struct lod_thread_info  *info = lod_env_info(env);
3284         struct lu_attr          *attr = &info->lti_attr;
3285         struct dt_object_format *dof = &info->lti_format;
3286         struct lu_buf           lmv_buf;
3287         struct lu_buf           slave_lmv_buf;
3288         struct lmv_mds_md_v1    *lmm;
3289         struct lmv_mds_md_v1    *slave_lmm = NULL;
3290         struct dt_insert_rec    *rec = &info->lti_dt_rec;
3291         int                     i;
3292         int                     rc;
3293         ENTRY;
3294
3295         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
3296                 RETURN(-ENOTDIR);
3297
3298         /* The stripes are supposed to be allocated in declare phase,
3299          * if there are no stripes being allocated, it will skip */
3300         if (lo->ldo_dir_stripe_count == 0)
3301                 RETURN(0);
3302
3303         rc = dt_attr_get(env, dt_object_child(dt), attr);
3304         if (rc != 0)
3305                 RETURN(rc);
3306
3307         attr->la_valid = LA_ATIME | LA_MTIME | LA_CTIME |
3308                          LA_MODE | LA_UID | LA_GID | LA_TYPE | LA_PROJID;
3309         dof->dof_type = DFT_DIR;
3310
3311         rc = lod_prep_lmv_md(env, dt, &lmv_buf);
3312         if (rc != 0)
3313                 RETURN(rc);
3314         lmm = lmv_buf.lb_buf;
3315
3316         OBD_ALLOC_PTR(slave_lmm);
3317         if (slave_lmm == NULL)
3318                 RETURN(-ENOMEM);
3319
3320         lod_prep_slave_lmv_md(slave_lmm, lmm);
3321         slave_lmv_buf.lb_buf = slave_lmm;
3322         slave_lmv_buf.lb_len = sizeof(*slave_lmm);
3323
3324         rec->rec_type = S_IFDIR;
3325         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
3326                 struct dt_object *dto;
3327                 char             *stripe_name = info->lti_key;
3328                 struct lu_name          *sname;
3329                 struct linkea_data       ldata          = { NULL };
3330                 struct lu_buf            linkea_buf;
3331
3332                 dto = lo->ldo_stripe[i];
3333
3334                 dt_write_lock(env, dto, MOR_TGT_CHILD);
3335                 rc = lod_sub_create(env, dto, attr, NULL, dof, th);
3336                 if (rc != 0) {
3337                         dt_write_unlock(env, dto);
3338                         GOTO(out, rc);
3339                 }
3340
3341                 rc = lod_sub_ref_add(env, dto, th);
3342                 dt_write_unlock(env, dto);
3343                 if (rc != 0)
3344                         GOTO(out, rc);
3345
3346                 rec->rec_fid = lu_object_fid(&dto->do_lu);
3347                 rc = lod_sub_insert(env, dto, (const struct dt_rec *)rec,
3348                                     (const struct dt_key *)dot, th, 0);
3349                 if (rc != 0)
3350                         GOTO(out, rc);
3351
3352                 rec->rec_fid = lu_object_fid(&dt->do_lu);
3353                 rc = lod_sub_insert(env, dto, (struct dt_rec *)rec,
3354                                     (const struct dt_key *)dotdot, th, 0);
3355                 if (rc != 0)
3356                         GOTO(out, rc);
3357
3358                 if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SLAVE_LMV) ||
3359                     cfs_fail_val != i) {
3360                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_LMV) &&
3361                             cfs_fail_val == i)
3362                                 slave_lmm->lmv_master_mdt_index =
3363                                                         cpu_to_le32(i + 1);
3364                         else
3365                                 slave_lmm->lmv_master_mdt_index =
3366                                                         cpu_to_le32(i);
3367
3368                         rc = lod_sub_xattr_set(env, dto, &slave_lmv_buf,
3369                                                XATTR_NAME_LMV, fl, th);
3370                         if (rc != 0)
3371                                 GOTO(out, rc);
3372                 }
3373
3374                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_SLAVE_NAME) &&
3375                     cfs_fail_val == i)
3376                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
3377                                  PFID(lu_object_fid(&dto->do_lu)), i + 1);
3378                 else
3379                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
3380                                  PFID(lu_object_fid(&dto->do_lu)), i);
3381
3382                 sname = lod_name_get(env, stripe_name, strlen(stripe_name));
3383                 rc = linkea_links_new(&ldata, &info->lti_linkea_buf,
3384                                       sname, lu_object_fid(&dt->do_lu));
3385                 if (rc != 0)
3386                         GOTO(out, rc);
3387
3388                 linkea_buf.lb_buf = ldata.ld_buf->lb_buf;
3389                 linkea_buf.lb_len = ldata.ld_leh->leh_len;
3390                 rc = lod_sub_xattr_set(env, dto, &linkea_buf,
3391                                        XATTR_NAME_LINK, 0, th);
3392                 if (rc != 0)
3393                         GOTO(out, rc);
3394
3395                 rec->rec_fid = lu_object_fid(&dto->do_lu);
3396                 rc = lod_sub_insert(env, dt_object_child(dt),
3397                                     (const struct dt_rec *)rec,
3398                                     (const struct dt_key *)stripe_name, th, 0);
3399                 if (rc != 0)
3400                         GOTO(out, rc);
3401
3402                 rc = lod_sub_ref_add(env, dt_object_child(dt), th);
3403                 if (rc != 0)
3404                         GOTO(out, rc);
3405         }
3406
3407         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MASTER_LMV))
3408                 rc = lod_sub_xattr_set(env, dt_object_child(dt),
3409                                        &lmv_buf, XATTR_NAME_LMV, fl, th);
3410 out:
3411         if (slave_lmm != NULL)
3412                 OBD_FREE_PTR(slave_lmm);
3413
3414         RETURN(rc);
3415 }
3416
3417 /**
3418  * Helper function to declare/execute creation of a striped directory
3419  *
3420  * Called in declare/create object path, prepare striping for a directory
3421  * and prepare defaults data striping for the objects to be created in
3422  * that directory. Notice the function calls "declaration" or "execution"
3423  * methods depending on \a declare param. This is a consequence of the
3424  * current approach while we don't have natural distributed transactions:
3425  * we basically execute non-local updates in the declare phase. So, the
3426  * arguments for the both phases are the same and this is the reason for
3427  * this function to exist.
3428  *
3429  * \param[in] env       execution environment
3430  * \param[in] dt        object
3431  * \param[in] attr      attributes the stripes will be created with
3432  * \param[in] lmu       lmv_user_md if MDT indices are specified
3433  * \param[in] dof       format of stripes (see OSD API description)
3434  * \param[in] th        transaction handle
3435  * \param[in] declare   where to call "declare" or "execute" methods
3436  *
3437  * \retval              0 on success
3438  * \retval              negative if failed
3439  */
3440 static int lod_dir_striping_create_internal(const struct lu_env *env,
3441                                             struct dt_object *dt,
3442                                             struct lu_attr *attr,
3443                                             const struct lu_buf *lmu,
3444                                             struct dt_object_format *dof,
3445                                             struct thandle *th,
3446                                             bool declare)
3447 {
3448         struct lod_thread_info *info = lod_env_info(env);
3449         struct lod_object *lo = lod_dt_obj(dt);
3450         const struct lod_default_striping *lds = lo->ldo_def_striping;
3451         int rc;
3452         ENTRY;
3453
3454         LASSERT(ergo(lds != NULL,
3455                      lds->lds_def_striping_set ||
3456                      lds->lds_dir_def_striping_set));
3457
3458         if (!LMVEA_DELETE_VALUES(lo->ldo_dir_stripe_count,
3459                                  lo->ldo_dir_stripe_offset)) {
3460                 if (!lmu) {
3461                         struct lmv_user_md_v1 *v1 = info->lti_ea_store;
3462                         int stripe_count = lo->ldo_dir_stripe_count;
3463
3464                         if (info->lti_ea_store_size < sizeof(*v1)) {
3465                                 rc = lod_ea_store_resize(info, sizeof(*v1));
3466                                 if (rc != 0)
3467                                         RETURN(rc);
3468                                 v1 = info->lti_ea_store;
3469                         }
3470
3471                         memset(v1, 0, sizeof(*v1));
3472                         v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
3473                         v1->lum_stripe_count = cpu_to_le32(stripe_count);
3474                         v1->lum_stripe_offset =
3475                                         cpu_to_le32(lo->ldo_dir_stripe_offset);
3476
3477                         info->lti_buf.lb_buf = v1;
3478                         info->lti_buf.lb_len = sizeof(*v1);
3479                         lmu = &info->lti_buf;
3480                 }
3481
3482                 if (declare)
3483                         rc = lod_declare_xattr_set_lmv(env, dt, attr, lmu, dof,
3484                                                        th);
3485                 else
3486                         rc = lod_xattr_set_lmv(env, dt, lmu, XATTR_NAME_LMV, 0,
3487                                                th);
3488                 if (rc != 0)
3489                         RETURN(rc);
3490         }
3491
3492         /* Transfer default LMV striping from the parent */
3493         if (lds != NULL && lds->lds_dir_def_striping_set &&
3494             !LMVEA_DELETE_VALUES(lds->lds_dir_def_stripe_count,
3495                                  lds->lds_dir_def_stripe_offset)) {
3496                 struct lmv_user_md_v1 *v1 = info->lti_ea_store;
3497
3498                 if (info->lti_ea_store_size < sizeof(*v1)) {
3499                         rc = lod_ea_store_resize(info, sizeof(*v1));
3500                         if (rc != 0)
3501                                 RETURN(rc);
3502                         v1 = info->lti_ea_store;
3503                 }
3504
3505                 memset(v1, 0, sizeof(*v1));
3506                 v1->lum_magic = cpu_to_le32(LMV_USER_MAGIC);
3507                 v1->lum_stripe_count =
3508                         cpu_to_le32(lds->lds_dir_def_stripe_count);
3509                 v1->lum_stripe_offset =
3510                         cpu_to_le32(lds->lds_dir_def_stripe_offset);
3511                 v1->lum_hash_type =
3512                         cpu_to_le32(lds->lds_dir_def_hash_type);
3513
3514                 info->lti_buf.lb_buf = v1;
3515                 info->lti_buf.lb_len = sizeof(*v1);
3516                 if (declare)
3517                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
3518                                                        XATTR_NAME_DEFAULT_LMV,
3519                                                        0, th);
3520                 else
3521                         rc = lod_xattr_set_default_lmv_on_dir(env, dt,
3522                                                   &info->lti_buf,
3523                                                   XATTR_NAME_DEFAULT_LMV, 0,
3524                                                   th);
3525                 if (rc != 0)
3526                         RETURN(rc);
3527         }
3528
3529         /* Transfer default LOV striping from the parent */
3530         if (lds != NULL && lds->lds_def_striping_set &&
3531             lds->lds_def_comp_cnt != 0) {
3532                 struct lov_mds_md *lmm;
3533                 int lmm_size = lod_comp_md_size(lo, true);
3534
3535                 if (info->lti_ea_store_size < lmm_size) {
3536                         rc = lod_ea_store_resize(info, lmm_size);
3537                         if (rc != 0)
3538                                 RETURN(rc);
3539                 }
3540                 lmm = info->lti_ea_store;
3541
3542                 rc = lod_generate_lovea(env, lo, lmm, &lmm_size, true);
3543                 if (rc != 0)
3544                         RETURN(rc);
3545
3546                 info->lti_buf.lb_buf = lmm;
3547                 info->lti_buf.lb_len = lmm_size;
3548
3549                 if (declare)
3550                         rc = lod_dir_declare_xattr_set(env, dt, &info->lti_buf,
3551                                                        XATTR_NAME_LOV, 0, th);
3552                 else
3553                         rc = lod_xattr_set_lov_on_dir(env, dt, &info->lti_buf,
3554                                                       XATTR_NAME_LOV, 0, th);
3555                 if (rc != 0)
3556                         RETURN(rc);
3557         }
3558
3559         RETURN(0);
3560 }
3561
3562 static int lod_declare_dir_striping_create(const struct lu_env *env,
3563                                            struct dt_object *dt,
3564                                            struct lu_attr *attr,
3565                                            struct lu_buf *lmu,
3566                                            struct dt_object_format *dof,
3567                                            struct thandle *th)
3568 {
3569         return lod_dir_striping_create_internal(env, dt, attr, lmu, dof, th,
3570                                                 true);
3571 }
3572
3573 static int lod_dir_striping_create(const struct lu_env *env,
3574                                    struct dt_object *dt,
3575                                    struct lu_attr *attr,
3576                                    struct dt_object_format *dof,
3577                                    struct thandle *th)
3578 {
3579         return lod_dir_striping_create_internal(env, dt, attr, NULL, dof, th,
3580                                                 false);
3581 }
3582
3583 /**
3584  * Make LOV EA for striped object.
3585  *
3586  * Generate striping information and store it in the LOV EA of the given
3587  * object. The caller must ensure nobody else is calling the function
3588  * against the object concurrently. The transaction must be started.
3589  * FLDB service must be running as well; it's used to map FID to the target,
3590  * which is stored in LOV EA.
3591  *
3592  * \param[in] env               execution environment for this thread
3593  * \param[in] lo                LOD object
3594  * \param[in] th                transaction handle
3595  *
3596  * \retval                      0 if LOV EA is stored successfully
3597  * \retval                      negative error number on failure
3598  */
3599 static int lod_generate_and_set_lovea(const struct lu_env *env,
3600                                       struct lod_object *lo,
3601                                       struct thandle *th)
3602 {
3603         struct lod_thread_info  *info = lod_env_info(env);
3604         struct dt_object        *next = dt_object_child(&lo->ldo_obj);
3605         struct lov_mds_md_v1    *lmm;
3606         int                      rc, lmm_size;
3607         ENTRY;
3608
3609         LASSERT(lo);
3610
3611         if (lo->ldo_comp_cnt == 0) {
3612                 lod_object_free_striping(env, lo);
3613                 rc = lod_sub_xattr_del(env, next, XATTR_NAME_LOV, th);
3614                 RETURN(rc);
3615         }
3616
3617         lmm_size = lod_comp_md_size(lo, false);
3618         if (info->lti_ea_store_size < lmm_size) {
3619                 rc = lod_ea_store_resize(info, lmm_size);
3620                 if (rc)
3621                         RETURN(rc);
3622         }
3623         lmm = info->lti_ea_store;
3624
3625         rc = lod_generate_lovea(env, lo, lmm, &lmm_size, false);
3626         if (rc)
3627                 RETURN(rc);
3628
3629         info->lti_buf.lb_buf = lmm;
3630         info->lti_buf.lb_len = lmm_size;
3631         rc = lod_sub_xattr_set(env, next, &info->lti_buf,
3632                                XATTR_NAME_LOV, 0, th);
3633         RETURN(rc);
3634 }
3635
3636 /**
3637  * Delete layout component(s)
3638  *
3639  * \param[in] env       execution environment for this thread
3640  * \param[in] dt        object
3641  * \param[in] th        transaction handle
3642  *
3643  * \retval      0 on success
3644  * \retval      negative error number on failure
3645  */
3646 static int lod_layout_del(const struct lu_env *env, struct dt_object *dt,
3647                           struct thandle *th)
3648 {
3649         struct lod_layout_component     *lod_comp;
3650         struct lod_object       *lo = lod_dt_obj(dt);
3651         struct dt_object        *next = dt_object_child(dt);
3652         struct lu_attr  *attr = &lod_env_info(env)->lti_attr;
3653         int     rc, i, j, left;
3654
3655         LASSERT(lo->ldo_is_composite);
3656         LASSERT(lo->ldo_comp_cnt > 0 && lo->ldo_comp_entries != NULL);
3657
3658         left = lo->ldo_comp_cnt;
3659         for (i = (lo->ldo_comp_cnt - 1); i >= 0; i--) {
3660                 lod_comp = &lo->ldo_comp_entries[i];
3661
3662                 if (lod_comp->llc_id != LCME_ID_INVAL)
3663                         break;
3664                 left--;
3665
3666                 /* Not instantiated component */
3667                 if (lod_comp->llc_stripe == NULL)
3668                         continue;
3669
3670                 LASSERT(lod_comp->llc_stripe_count > 0);
3671                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
3672                         struct dt_object *obj = lod_comp->llc_stripe[j];
3673
3674                         if (obj == NULL)
3675                                 continue;
3676                         rc = lod_sub_destroy(env, obj, th);
3677                         if (rc)
3678                                 GOTO(out, rc);
3679
3680                         lu_object_put(env, &obj->do_lu);
3681                         lod_comp->llc_stripe[j] = NULL;
3682                 }
3683                 OBD_FREE(lod_comp->llc_stripe, sizeof(struct dt_object *) *
3684                                         lod_comp->llc_stripes_allocated);
3685                 lod_comp->llc_stripe = NULL;
3686                 lod_comp->llc_stripes_allocated = 0;
3687                 lod_obj_set_pool(lo, i, NULL);
3688                 if (lod_comp->llc_ostlist.op_array) {
3689                         OBD_FREE(lod_comp->llc_ostlist.op_array,
3690                                  lod_comp->llc_ostlist.op_size);
3691                         lod_comp->llc_ostlist.op_array = NULL;
3692                         lod_comp->llc_ostlist.op_size = 0;
3693                 }
3694         }
3695
3696         LASSERTF(left >= 0 && left < lo->ldo_comp_cnt, "left = %d\n", left);
3697         if (left > 0) {
3698                 struct lod_layout_component     *comp_array;
3699
3700                 OBD_ALLOC(comp_array, sizeof(*comp_array) * left);
3701                 if (comp_array == NULL)
3702                         GOTO(out, rc = -ENOMEM);
3703
3704                 memcpy(&comp_array[0], &lo->ldo_comp_entries[0],
3705                        sizeof(*comp_array) * left);
3706
3707                 OBD_FREE(lo->ldo_comp_entries,
3708                        &nb