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                          sizeof(*comp_array) * lo->ldo_comp_cnt);
3709                 lo->ldo_comp_entries = comp_array;
3710                 lo->ldo_comp_cnt = left;
3711
3712                 LASSERT(lo->ldo_mirror_count == 1);
3713                 lo->ldo_mirrors[0].lme_end = left - 1;
3714                 lod_obj_inc_layout_gen(lo);
3715         } else {
3716                 lod_free_comp_entries(lo);
3717         }
3718
3719         LASSERT(dt_object_exists(dt));
3720         rc = dt_attr_get(env, next, attr);
3721         if (rc)
3722                 GOTO(out, rc);
3723
3724         if (attr->la_size > 0) {
3725                 attr->la_size = 0;
3726                 attr->la_valid = LA_SIZE;
3727                 rc = lod_sub_attr_set(env, next, attr, th);
3728                 if (rc)
3729                         GOTO(out, rc);
3730         }
3731
3732         rc = lod_generate_and_set_lovea(env, lo, th);
3733         EXIT;
3734 out:
3735         if (rc)
3736                 lod_object_free_striping(env, lo);
3737         return rc;
3738 }
3739
3740
3741 static int lod_get_default_lov_striping(const struct lu_env *env,
3742                                         struct lod_object *lo,
3743                                         struct lod_default_striping *lds);
3744 /**
3745  * Implementation of dt_object_operations::do_xattr_set.
3746  *
3747  * Sets specified extended attribute on the object. Three types of EAs are
3748  * special:
3749  *   LOV EA - stores striping for a regular file or default striping (when set
3750  *            on a directory)
3751  *   LMV EA - stores a marker for the striped directories
3752  *   DMV EA - stores default directory striping
3753  *
3754  * When striping is applied to a non-striped existing object (this is called
3755  * late striping), then LOD notices the caller wants to turn the object into a
3756  * striped one. The stripe objects are created and appropriate EA is set:
3757  * LOV EA storing all the stripes directly or LMV EA storing just a small header
3758  * with striping configuration.
3759  *
3760  * \see dt_object_operations::do_xattr_set() in the API description for details.
3761  */
3762 static int lod_xattr_set(const struct lu_env *env,
3763                          struct dt_object *dt, const struct lu_buf *buf,
3764                          const char *name, int fl, struct thandle *th)
3765 {
3766         struct dt_object        *next = dt_object_child(dt);
3767         int                      rc;
3768         ENTRY;
3769
3770         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
3771             strcmp(name, XATTR_NAME_LMV) == 0) {
3772                 struct lmv_mds_md_v1 *lmm = buf->lb_buf;
3773
3774                 if (lmm != NULL && le32_to_cpu(lmm->lmv_hash_type) &
3775                                                 LMV_HASH_FLAG_MIGRATION)
3776                         rc = lod_sub_xattr_set(env, next, buf, name, fl, th);
3777                 else
3778                         rc = lod_dir_striping_create(env, dt, NULL, NULL, th);
3779
3780                 RETURN(rc);
3781         }
3782
3783         if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
3784             strcmp(name, XATTR_NAME_LOV) == 0) {
3785                 struct lod_thread_info *info = lod_env_info(env);
3786                 struct lod_default_striping *lds = &info->lti_def_striping;
3787                 struct lov_user_md_v1 *v1 = buf->lb_buf;
3788                 char pool[LOV_MAXPOOLNAME + 1];
3789                 bool is_del;
3790
3791                 /* get existing striping config */
3792                 rc = lod_get_default_lov_striping(env, lod_dt_obj(dt), lds);
3793                 if (rc)
3794                         RETURN(rc);
3795
3796                 memset(pool, 0, sizeof(pool));
3797                 if (lds->lds_def_striping_set == 1)
3798                         lod_layout_get_pool(lds->lds_def_comp_entries,
3799                                             lds->lds_def_comp_cnt, pool,
3800                                             sizeof(pool));
3801
3802                 is_del = LOVEA_DELETE_VALUES(v1->lmm_stripe_size,
3803                                              v1->lmm_stripe_count,
3804                                              v1->lmm_stripe_offset,
3805                                              NULL);
3806
3807                 /* Retain the pool name if it is not given */
3808                 if (v1->lmm_magic == LOV_USER_MAGIC_V1 && pool[0] != '\0' &&
3809                         !is_del) {
3810                         struct lod_thread_info *info = lod_env_info(env);
3811                         struct lov_user_md_v3 *v3  = info->lti_ea_store;
3812
3813                         memset(v3, 0, sizeof(*v3));
3814                         v3->lmm_magic = cpu_to_le32(LOV_USER_MAGIC_V3);
3815                         v3->lmm_pattern = cpu_to_le32(v1->lmm_pattern);
3816                         v3->lmm_stripe_count =
3817                                         cpu_to_le32(v1->lmm_stripe_count);
3818                         v3->lmm_stripe_offset =
3819                                         cpu_to_le32(v1->lmm_stripe_offset);
3820                         v3->lmm_stripe_size = cpu_to_le32(v1->lmm_stripe_size);
3821
3822                         strlcpy(v3->lmm_pool_name, pool,
3823                                 sizeof(v3->lmm_pool_name));
3824
3825                         info->lti_buf.lb_buf = v3;
3826                         info->lti_buf.lb_len = sizeof(*v3);
3827                         rc = lod_xattr_set_lov_on_dir(env, dt, &info->lti_buf,
3828                                                       name, fl, th);
3829                 } else {
3830                         rc = lod_xattr_set_lov_on_dir(env, dt, buf, name,
3831                                                       fl, th);
3832                 }
3833
3834                 if (lds->lds_def_striping_set == 1 &&
3835                     lds->lds_def_comp_entries != NULL)
3836                         lod_free_def_comp_entries(lds);
3837
3838                 RETURN(rc);
3839         } else if (S_ISDIR(dt->do_lu.lo_header->loh_attr) &&
3840                    strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
3841                 /* default LMVEA */
3842                 rc = lod_xattr_set_default_lmv_on_dir(env, dt, buf, name, fl,
3843                                                       th);
3844                 RETURN(rc);
3845         } else if (S_ISREG(dt->do_lu.lo_header->loh_attr) &&
3846                    (!strcmp(name, XATTR_NAME_LOV) ||
3847                     !strncmp(name, XATTR_LUSTRE_LOV,
3848                              strlen(XATTR_LUSTRE_LOV)))) {
3849                 /* in case of lov EA swap, just set it
3850                  * if not, it is a replay so check striping match what we
3851                  * already have during req replay, declare_xattr_set()
3852                  * defines striping, then create() does the work */
3853                 if (fl & LU_XATTR_REPLACE) {
3854                         /* free stripes, then update disk */
3855                         lod_object_free_striping(env, lod_dt_obj(dt));
3856
3857                         rc = lod_sub_xattr_set(env, next, buf, name, fl, th);
3858                 } else if (dt_object_remote(dt)) {
3859                         /* This only happens during migration, see
3860                          * mdd_migrate_create(), in which Master MDT will
3861                          * create a remote target object, and only set
3862                          * (migrating) stripe EA on the remote object,
3863                          * and does not need creating each stripes. */
3864                         rc = lod_sub_xattr_set(env, next, buf, name,
3865                                                       fl, th);
3866                 } else if (strcmp(name, XATTR_LUSTRE_LOV".del") == 0) {
3867                         /* delete component(s) */
3868                         LASSERT(lod_dt_obj(dt)->ldo_comp_cached);
3869                         rc = lod_layout_del(env, dt, th);
3870                 } else {
3871                         /*
3872                          * When 'name' is XATTR_LUSTRE_LOV or XATTR_NAME_LOV,
3873                          * it's going to create create file with specified
3874                          * component(s), the striping must have not being
3875                          * cached in this case;
3876                          *
3877                          * Otherwise, it's going to add/change component(s) to
3878                          * an existing file, the striping must have been cached
3879                          * in this case.
3880                          */
3881                         LASSERT(equi(!strcmp(name, XATTR_LUSTRE_LOV) ||
3882                                      !strcmp(name, XATTR_NAME_LOV),
3883                                 !lod_dt_obj(dt)->ldo_comp_cached));
3884
3885                         rc = lod_striped_create(env, dt, NULL, NULL, th);
3886                 }
3887                 RETURN(rc);
3888         } else if (strcmp(name, XATTR_NAME_FID) == 0) {
3889                 rc = lod_replace_parent_fid(env, dt, th, false);
3890
3891                 RETURN(rc);
3892         }
3893
3894         /* then all other xattr */
3895         rc = lod_xattr_set_internal(env, dt, buf, name, fl, th);
3896
3897         RETURN(rc);
3898 }
3899
3900 /**
3901  * Implementation of dt_object_operations::do_declare_xattr_del.
3902  *
3903  * \see dt_object_operations::do_declare_xattr_del() in the API description
3904  * for details.
3905  */
3906 static int lod_declare_xattr_del(const struct lu_env *env,
3907                                  struct dt_object *dt, const char *name,
3908                                  struct thandle *th)
3909 {
3910         struct lod_object       *lo = lod_dt_obj(dt);
3911         int                     rc;
3912         int                     i;
3913         ENTRY;
3914
3915         rc = lod_sub_declare_xattr_del(env, dt_object_child(dt), name, th);
3916         if (rc != 0)
3917                 RETURN(rc);
3918
3919         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
3920                 RETURN(0);
3921
3922         /* set xattr to each stripes, if needed */
3923         rc = lod_load_striping(env, lo);
3924         if (rc != 0)
3925                 RETURN(rc);
3926
3927         if (lo->ldo_dir_stripe_count == 0)
3928                 RETURN(0);
3929
3930         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
3931                 LASSERT(lo->ldo_stripe[i]);
3932                 rc = lod_sub_declare_xattr_del(env, lo->ldo_stripe[i],
3933                                                name, th);
3934                 if (rc != 0)
3935                         break;
3936         }
3937
3938         RETURN(rc);
3939 }
3940
3941 /**
3942  * Implementation of dt_object_operations::do_xattr_del.
3943  *
3944  * If EA storing a regular striping is being deleted, then release
3945  * all the references to the stripe objects in core.
3946  *
3947  * \see dt_object_operations::do_xattr_del() in the API description for details.
3948  */
3949 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
3950                          const char *name, struct thandle *th)
3951 {
3952         struct dt_object        *next = dt_object_child(dt);
3953         struct lod_object       *lo = lod_dt_obj(dt);
3954         int                     rc;
3955         int                     i;
3956         ENTRY;
3957
3958         if (!strcmp(name, XATTR_NAME_LOV))
3959                 lod_object_free_striping(env, lod_dt_obj(dt));
3960
3961         rc = lod_sub_xattr_del(env, next, name, th);
3962         if (rc != 0 || !S_ISDIR(dt->do_lu.lo_header->loh_attr))
3963                 RETURN(rc);
3964
3965         if (lo->ldo_dir_stripe_count == 0)
3966                 RETURN(0);
3967
3968         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
3969                 LASSERT(lo->ldo_stripe[i]);
3970
3971                 rc = lod_sub_xattr_del(env, lo->ldo_stripe[i], name, th);
3972                 if (rc != 0)
3973                         break;
3974         }
3975
3976         RETURN(rc);
3977 }
3978
3979 /**
3980  * Implementation of dt_object_operations::do_xattr_list.
3981  *
3982  * \see dt_object_operations::do_xattr_list() in the API description
3983  * for details.
3984  */
3985 static int lod_xattr_list(const struct lu_env *env,
3986                           struct dt_object *dt, const struct lu_buf *buf)
3987 {
3988         return dt_xattr_list(env, dt_object_child(dt), buf);
3989 }
3990
3991 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
3992 {
3993         return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
3994 }
3995
3996
3997 /**
3998  * Get default striping.
3999  *
4000  * \param[in] env               execution environment
4001  * \param[in] lo                object
4002  * \param[out] lds              default striping
4003  *
4004  * \retval              0 on success
4005  * \retval              negative if failed
4006  */
4007 static int lod_get_default_lov_striping(const struct lu_env *env,
4008                                         struct lod_object *lo,
4009                                         struct lod_default_striping *lds)
4010 {
4011         struct lod_thread_info *info = lod_env_info(env);
4012         struct lov_user_md_v1 *v1 = NULL;
4013         struct lov_user_md_v3 *v3 = NULL;
4014         struct lov_comp_md_v1 *comp_v1 = NULL;
4015         __u16   comp_cnt;
4016         __u16   mirror_cnt;
4017         bool    composite;
4018         int     rc, i;
4019         ENTRY;
4020
4021         lds->lds_def_striping_set = 0;
4022
4023         rc = lod_get_lov_ea(env, lo);
4024         if (rc < 0)
4025                 RETURN(rc);
4026
4027         if (rc < (typeof(rc))sizeof(struct lov_user_md))
4028                 RETURN(0);
4029
4030         v1 = info->lti_ea_store;
4031         if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1)) {
4032                 lustre_swab_lov_user_md_v1(v1);
4033         } else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3)) {
4034                 v3 = (struct lov_user_md_v3 *)v1;
4035                 lustre_swab_lov_user_md_v3(v3);
4036         } else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_COMP_V1)) {
4037                 comp_v1 = (struct lov_comp_md_v1 *)v1;
4038                 lustre_swab_lov_comp_md_v1(comp_v1);
4039         }
4040
4041         if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1 &&
4042             v1->lmm_magic != LOV_MAGIC_COMP_V1)
4043                 RETURN(-ENOTSUPP);
4044
4045         if (v1->lmm_magic == LOV_MAGIC_COMP_V1) {
4046                 comp_v1 = (struct lov_comp_md_v1 *)v1;
4047                 comp_cnt = comp_v1->lcm_entry_count;
4048                 if (comp_cnt == 0)
4049                         RETURN(-EINVAL);
4050                 mirror_cnt = comp_v1->lcm_mirror_count + 1;
4051                 composite = true;
4052         } else {
4053                 comp_cnt = 1;
4054                 mirror_cnt = 0;
4055                 composite = false;
4056         }
4057
4058         /* realloc default comp entries if necessary */
4059         rc = lod_def_striping_comp_resize(lds, comp_cnt);
4060         if (rc < 0)
4061                 RETURN(rc);
4062
4063         lds->lds_def_comp_cnt = comp_cnt;
4064         lds->lds_def_striping_is_composite = composite;
4065         lds->lds_def_mirror_cnt = mirror_cnt;
4066
4067         for (i = 0; i < comp_cnt; i++) {
4068                 struct lod_layout_component *lod_comp;
4069                 struct lu_extent *ext;
4070                 char *pool;
4071
4072                 lod_comp = &lds->lds_def_comp_entries[i];
4073                 /*
4074                  * reset lod_comp values, llc_stripes is always NULL in
4075                  * the default striping template, llc_pool will be reset
4076                  * later below.
4077                  */
4078                 memset(lod_comp, 0, offsetof(typeof(*lod_comp), llc_pool));
4079
4080                 if (composite) {
4081                         v1 = (struct lov_user_md *)((char *)comp_v1 +
4082                                         comp_v1->lcm_entries[i].lcme_offset);
4083                         ext = &comp_v1->lcm_entries[i].lcme_extent;
4084                         lod_comp->llc_extent = *ext;
4085                 }
4086
4087                 if (v1->lmm_pattern != LOV_PATTERN_RAID0 &&
4088                     v1->lmm_pattern != LOV_PATTERN_MDT &&
4089                     v1->lmm_pattern != 0) {
4090                         lod_free_def_comp_entries(lds);
4091                         RETURN(-EINVAL);
4092                 }
4093
4094                 CDEBUG(D_LAYOUT, DFID" stripe_count=%d stripe_size=%d "
4095                        "stripe_offset=%d\n",
4096                        PFID(lu_object_fid(&lo->ldo_obj.do_lu)),
4097                        (int)v1->lmm_stripe_count, (int)v1->lmm_stripe_size,
4098                        (int)v1->lmm_stripe_offset);
4099
4100                 lod_comp->llc_stripe_count = v1->lmm_stripe_count;
4101                 lod_comp->llc_stripe_size = v1->lmm_stripe_size;
4102                 lod_comp->llc_stripe_offset = v1->lmm_stripe_offset;
4103                 lod_comp->llc_pattern = v1->lmm_pattern;
4104
4105                 pool = NULL;
4106                 if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
4107                         /* XXX: sanity check here */
4108                         v3 = (struct lov_user_md_v3 *) v1;
4109                         if (v3->lmm_pool_name[0] != '\0')
4110                                 pool = v3->lmm_pool_name;
4111                 }
4112                 lod_set_def_pool(lds, i, pool);
4113         }
4114
4115         lds->lds_def_striping_set = 1;
4116         RETURN(rc);
4117 }
4118
4119 /**
4120  * Get default directory striping.
4121  *
4122  * \param[in] env               execution environment
4123  * \param[in] lo                object
4124  * \param[out] lds              default striping
4125  *
4126  * \retval              0 on success
4127  * \retval              negative if failed
4128  */
4129 static int lod_get_default_lmv_striping(const struct lu_env *env,
4130                                         struct lod_object *lo,
4131                                         struct lod_default_striping *lds)
4132 {
4133         struct lod_thread_info  *info = lod_env_info(env);
4134         struct lmv_user_md_v1   *v1 = NULL;
4135         int                      rc;
4136         ENTRY;
4137
4138         lds->lds_dir_def_striping_set = 0;
4139         rc = lod_get_default_lmv_ea(env, lo);
4140         if (rc < 0)
4141                 RETURN(rc);
4142
4143         if (rc < (typeof(rc))sizeof(struct lmv_user_md))
4144                 RETURN(0);
4145
4146         v1 = info->lti_ea_store;
4147
4148         lds->lds_dir_def_stripe_count = le32_to_cpu(v1->lum_stripe_count);
4149         lds->lds_dir_def_stripe_offset = le32_to_cpu(v1->lum_stripe_offset);
4150         lds->lds_dir_def_hash_type = le32_to_cpu(v1->lum_hash_type);
4151         lds->lds_dir_def_striping_set = 1;
4152
4153         RETURN(0);
4154 }
4155
4156 /**
4157  * Get default striping in the object.
4158  *
4159  * Get object default striping and default directory striping.
4160  *
4161  * \param[in] env               execution environment
4162  * \param[in] lo                object
4163  * \param[out] lds              default striping
4164  *
4165  * \retval              0 on success
4166  * \retval              negative if failed
4167  */
4168 static int lod_get_default_striping(const struct lu_env *env,
4169                                     struct lod_object *lo,
4170                                     struct lod_default_striping *lds)
4171 {
4172         int rc, rc1;
4173
4174         rc = lod_get_default_lov_striping(env, lo, lds);
4175         rc1 = lod_get_default_lmv_striping(env, lo, lds);
4176         if (rc == 0 && rc1 < 0)
4177                 rc = rc1;
4178
4179         return rc;
4180 }
4181
4182 /**
4183  * Apply default striping on object.
4184  *
4185  * If object striping pattern is not set, set to the one in default striping.
4186  * The default striping is from parent or fs.
4187  *
4188  * \param[in] lo                new object
4189  * \param[in] lds               default striping
4190  * \param[in] mode              new object's mode
4191  */
4192 static void lod_striping_from_default(struct lod_object *lo,
4193                                       const struct lod_default_striping *lds,
4194                                       umode_t mode)
4195 {
4196         struct lod_device *d = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
4197         struct lov_desc *desc = &d->lod_desc;
4198         int i, rc;
4199
4200         if (lds->lds_def_striping_set && S_ISREG(mode)) {
4201                 rc = lod_alloc_comp_entries(lo, lds->lds_def_mirror_cnt,
4202                                             lds->lds_def_comp_cnt);
4203                 if (rc != 0)
4204                         return;
4205
4206                 lo->ldo_is_composite = lds->lds_def_striping_is_composite;
4207                 if (lds->lds_def_mirror_cnt > 1)
4208                         lo->ldo_flr_state = LCM_FL_RDONLY;
4209
4210                 for (i = 0; i < lo->ldo_comp_cnt; i++) {
4211                         struct lod_layout_component *obj_comp =
4212                                                 &lo->ldo_comp_entries[i];
4213                         struct lod_layout_component *def_comp =
4214                                                 &lds->lds_def_comp_entries[i];
4215
4216                         CDEBUG(D_LAYOUT, "Inherite from default: size:%hu "
4217                                "nr:%u offset:%u pattern %#x %s\n",
4218                                def_comp->llc_stripe_size,
4219                                def_comp->llc_stripe_count,
4220                                def_comp->llc_stripe_offset,
4221                                def_comp->llc_pattern,
4222                                def_comp->llc_pool ?: "");
4223
4224                         *obj_comp = *def_comp;
4225                         if (def_comp->llc_pool != NULL) {
4226                                 /* pointer was copied from def_comp */
4227                                 obj_comp->llc_pool = NULL;
4228                                 lod_obj_set_pool(lo, i, def_comp->llc_pool);
4229                         }
4230
4231                         /*
4232                          * Don't initialize these fields for plain layout
4233                          * (v1/v3) here, they are inherited in the order of
4234                          * 'parent' -> 'fs default (root)' -> 'global default
4235                          * values for stripe_count & stripe_size'.
4236                          *
4237                          * see lod_ah_init().
4238                          */
4239                         if (!lo->ldo_is_composite)
4240                                 continue;
4241
4242                         lod_adjust_stripe_info(obj_comp, desc);
4243                 }
4244         } else if (lds->lds_dir_def_striping_set && S_ISDIR(mode)) {
4245                 if (lo->ldo_dir_stripe_count == 0)
4246                         lo->ldo_dir_stripe_count =
4247                                 lds->lds_dir_def_stripe_count;
4248                 if (lo->ldo_dir_stripe_offset == -1)
4249                         lo->ldo_dir_stripe_offset =
4250                                 lds->lds_dir_def_stripe_offset;
4251                 if (lo->ldo_dir_hash_type == 0)
4252                         lo->ldo_dir_hash_type = lds->lds_dir_def_hash_type;
4253
4254                 CDEBUG(D_LAYOUT, "striping from default dir: count:%hu, "
4255                        "offset:%u, hash_type:%u\n",
4256                        lo->ldo_dir_stripe_count, lo->ldo_dir_stripe_offset,
4257                        lo->ldo_dir_hash_type);
4258         }
4259 }
4260
4261 static inline bool lod_need_inherit_more(struct lod_object *lo, bool from_root)
4262 {
4263         struct lod_layout_component *lod_comp;
4264
4265         if (lo->ldo_comp_cnt == 0)
4266                 return true;
4267
4268         if (lo->ldo_is_composite)
4269                 return false;
4270
4271         lod_comp = &lo->ldo_comp_entries[0];
4272
4273         if (lod_comp->llc_stripe_count <= 0 ||
4274             lod_comp->llc_stripe_size <= 0)
4275                 return true;
4276
4277         if (from_root && (lod_comp->llc_pool == NULL ||
4278                           lod_comp->llc_stripe_offset == LOV_OFFSET_DEFAULT))
4279                 return true;
4280
4281         return false;
4282 }
4283
4284 /**
4285  * Implementation of dt_object_operations::do_ah_init.
4286  *
4287  * This method is used to make a decision on the striping configuration for the
4288  * object being created. It can be taken from the \a parent object if it exists,
4289  * or filesystem's default. The resulting configuration (number of stripes,
4290  * stripe size/offset, pool name, etc) is stored in the object itself and will
4291  * be used by the methods like ->doo_declare_create().
4292  *
4293  * \see dt_object_operations::do_ah_init() in the API description for details.
4294  */
4295 static void lod_ah_init(const struct lu_env *env,
4296                         struct dt_allocation_hint *ah,
4297                         struct dt_object *parent,
4298                         struct dt_object *child,
4299                         umode_t child_mode)
4300 {
4301         struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
4302         struct lod_thread_info *info = lod_env_info(env);
4303         struct lod_default_striping *lds = &info->lti_def_striping;
4304         struct dt_object *nextp = NULL;
4305         struct dt_object *nextc;
4306         struct lod_object *lp = NULL;
4307         struct lod_object *lc;
4308         struct lov_desc *desc;
4309         struct lod_layout_component *lod_comp;
4310         int rc;
4311         ENTRY;
4312
4313         LASSERT(child);
4314
4315         if (likely(parent)) {
4316                 nextp = dt_object_child(parent);
4317                 lp = lod_dt_obj(parent);
4318         }
4319
4320         nextc = dt_object_child(child);
4321         lc = lod_dt_obj(child);
4322
4323         LASSERT(!lod_obj_is_striped(child));
4324         /* default layout template may have been set on the regular file
4325          * when this is called from mdd_create_data() */
4326         if (S_ISREG(child_mode))
4327                 lod_free_comp_entries(lc);
4328
4329         if (!dt_object_exists(nextc))
4330                 nextc->do_ops->do_ah_init(env, ah, nextp, nextc, child_mode);
4331
4332         if (S_ISDIR(child_mode)) {
4333                 const struct lmv_user_md_v1 *lum1 = ah->dah_eadata;
4334
4335                 /* other default values are 0 */
4336                 lc->ldo_dir_stripe_offset = -1;
4337
4338                 /* get default striping from parent object */
4339                 if (likely(lp != NULL))
4340                         lod_get_default_striping(env, lp, lds);
4341
4342                 /* set child default striping info, default value is NULL */
4343                 if (lds->lds_def_striping_set || lds->lds_dir_def_striping_set)
4344                         lc->ldo_def_striping = lds;
4345
4346                 /* It should always honour the specified stripes */
4347                 /* Note: old client (< 2.7)might also do lfs mkdir, whose EA
4348                  * will have old magic. In this case, we should ignore the
4349                  * stripe count and try to create dir by default stripe.
4350                  */
4351                 if (ah->dah_eadata != NULL && ah->dah_eadata_len != 0 &&
4352                     (le32_to_cpu(lum1->lum_magic) == LMV_USER_MAGIC ||
4353                      le32_to_cpu(lum1->lum_magic) == LMV_USER_MAGIC_SPECIFIC)) {
4354                         lc->ldo_dir_stripe_count =
4355                                 le32_to_cpu(lum1->lum_stripe_count);
4356                         lc->ldo_dir_stripe_offset =
4357                                 le32_to_cpu(lum1->lum_stripe_offset);
4358                         lc->ldo_dir_hash_type =
4359                                 le32_to_cpu(lum1->lum_hash_type);
4360                         CDEBUG(D_INFO,
4361                                "set dirstripe: count %hu, offset %d, hash %u\n",
4362                                 lc->ldo_dir_stripe_count,
4363                                 (int)lc->ldo_dir_stripe_offset,
4364                                 lc->ldo_dir_hash_type);
4365                 } else {
4366                         /* transfer defaults LMV to new directory */
4367                         lod_striping_from_default(lc, lds, child_mode);
4368                 }
4369
4370                 /* shrink the stripe_count to the avaible MDT count */
4371                 if (lc->ldo_dir_stripe_count > d->lod_remote_mdt_count + 1 &&
4372                     !OBD_FAIL_CHECK(OBD_FAIL_LARGE_STRIPE))
4373                         lc->ldo_dir_stripe_count = d->lod_remote_mdt_count + 1;
4374
4375                 /* Directory will be striped only if stripe_count > 1, if
4376                  * stripe_count == 1, let's reset stripe_count = 0 to avoid
4377                  * create single master stripe and also help to unify the
4378                  * stripe handling of directories and files */
4379                 if (lc->ldo_dir_stripe_count == 1)
4380                         lc->ldo_dir_stripe_count = 0;
4381
4382                 CDEBUG(D_INFO, "final dir stripe [%hu %d %u]\n",
4383                        lc->ldo_dir_stripe_count,
4384                        (int)lc->ldo_dir_stripe_offset, lc->ldo_dir_hash_type);
4385
4386                 RETURN_EXIT;
4387         }
4388
4389         /* child object regular file*/
4390
4391         if (!lod_object_will_be_striped(S_ISREG(child_mode),
4392                                         lu_object_fid(&child->do_lu)))
4393                 RETURN_EXIT;
4394
4395         /* If object is going to be striped over OSTs, transfer default
4396          * striping information to the child, so that we can use it
4397          * during declaration and creation.
4398          *
4399          * Try from the parent first.
4400          */
4401         if (likely(lp != NULL)) {
4402                 rc = lod_get_default_lov_striping(env, lp, lds);
4403                 if (rc == 0)
4404                         lod_striping_from_default(lc, lds, child_mode);
4405         }
4406
4407         /* Initialize lod_device::lod_md_root object reference */
4408         if (d->lod_md_root == NULL) {
4409                 struct dt_object *root;
4410                 struct lod_object *lroot;
4411
4412                 lu_root_fid(&info->lti_fid);
4413                 root = dt_locate(env, &d->lod_dt_dev, &info->lti_fid);
4414                 if (!IS_ERR(root)) {
4415                         lroot = lod_dt_obj(root);
4416
4417                         spin_lock(&d->lod_lock);
4418                         if (d->lod_md_root != NULL)
4419                                 dt_object_put(env, &d->lod_md_root->ldo_obj);
4420                         d->lod_md_root = lroot;
4421                         spin_unlock(&d->lod_lock);
4422                 }
4423         }
4424
4425         /* try inherit layout from the root object (fs default) when:
4426          *  - parent does not have default layout; or
4427          *  - parent has plain(v1/v3) default layout, and some attributes
4428          *    are not specified in the default layout;
4429          */
4430         if (d->lod_md_root != NULL && lod_need_inherit_more(lc, true)) {
4431                 rc = lod_get_default_lov_striping(env, d->lod_md_root, lds);
4432                 if (rc)
4433                         goto out;
4434                 if (lc->ldo_comp_cnt == 0) {
4435                         lod_striping_from_default(lc, lds, child_mode);
4436                 } else if (!lds->lds_def_striping_is_composite) {
4437                         struct lod_layout_component *def_comp;
4438
4439                         LASSERT(!lc->ldo_is_composite);
4440                         lod_comp = &lc->ldo_comp_entries[0];
4441                         def_comp = &lds->lds_def_comp_entries[0];
4442
4443                         if (lod_comp->llc_stripe_count <= 0)
4444                                 lod_comp->llc_stripe_count =
4445                                         def_comp->llc_stripe_count;
4446                         if (lod_comp->llc_stripe_size <= 0)
4447                                 lod_comp->llc_stripe_size =
4448                                         def_comp->llc_stripe_size;
4449                         if (lod_comp->llc_stripe_offset == LOV_OFFSET_DEFAULT)
4450                                 lod_comp->llc_stripe_offset =
4451                                         def_comp->llc_stripe_offset;
4452                         if (lod_comp->llc_pool == NULL)
4453                                 lod_obj_set_pool(lc, 0, def_comp->llc_pool);
4454                 }
4455         }
4456 out:
4457         /*
4458          * fs default striping may not be explicitly set, or historically set
4459          * in config log, use them.
4460          */
4461         if (lod_need_inherit_more(lc, false)) {
4462                 if (lc->ldo_comp_cnt == 0) {
4463                         rc = lod_alloc_comp_entries(lc, 0, 1);
4464                         if (rc)
4465                                 /* fail to allocate memory, will create a
4466                                  * non-striped file. */
4467                                 RETURN_EXIT;
4468                         lc->ldo_is_composite = 0;
4469                         lod_comp = &lc->ldo_comp_entries[0];
4470                         lod_comp->llc_stripe_offset = LOV_OFFSET_DEFAULT;
4471                 }
4472                 LASSERT(!lc->ldo_is_composite);
4473                 lod_comp = &lc->ldo_comp_entries[0];
4474                 desc = &d->lod_desc;
4475                 lod_adjust_stripe_info(lod_comp, desc);
4476         }
4477
4478         EXIT;
4479 }
4480
4481 #define ll_do_div64(aaa,bbb)    do_div((aaa), (bbb))
4482 /**
4483  * Size initialization on late striping.
4484  *
4485  * Propagate the size of a truncated object to a deferred striping.
4486  * This function handles a special case when truncate was done on a
4487  * non-striped object and now while the striping is being created
4488  * we can't lose that size, so we have to propagate it to the stripes
4489  * being created.
4490  *
4491  * \param[in] env       execution environment
4492  * \param[in] dt        object
4493  * \param[in] th        transaction handle
4494  *
4495  * \retval              0 on success
4496  * \retval              negative if failed
4497  */
4498 static int lod_declare_init_size(const struct lu_env *env,
4499                                  struct dt_object *dt, struct thandle *th)
4500 {
4501         struct dt_object        *next = dt_object_child(dt);
4502         struct lod_object       *lo = lod_dt_obj(dt);
4503         struct dt_object        **objects = NULL;
4504         struct lu_attr  *attr = &lod_env_info(env)->lti_attr;
4505         uint64_t        size, offs;
4506         int     i, rc, stripe, stripe_count = 0, stripe_size = 0;
4507         struct lu_extent size_ext;
4508         ENTRY;
4509
4510         if (!lod_obj_is_striped(dt))
4511                 RETURN(0);
4512
4513         rc = dt_attr_get(env, next, attr);
4514         LASSERT(attr->la_valid & LA_SIZE);
4515         if (rc)
4516                 RETURN(rc);
4517
4518         size = attr->la_size;
4519         if (size == 0)
4520                 RETURN(0);
4521
4522         size_ext = (typeof(size_ext)){ .e_start = size - 1, .e_end = size };
4523         for (i = 0; i < lo->ldo_comp_cnt; i++) {
4524                 struct lod_layout_component *lod_comp;
4525                 struct lu_extent *extent;
4526
4527                 lod_comp = &lo->ldo_comp_entries[i];
4528
4529                 if (lod_comp->llc_stripe == NULL)
4530                         continue;
4531
4532                 extent = &lod_comp->llc_extent;
4533                 CDEBUG(D_INFO, "%lld "DEXT"\n", size, PEXT(extent));
4534                 if (!lo->ldo_is_composite ||
4535                     lu_extent_is_overlapped(extent, &size_ext)) {
4536                         objects = lod_comp->llc_stripe;
4537                         stripe_count = lod_comp->llc_stripe_count;
4538                         stripe_size = lod_comp->llc_stripe_size;
4539
4540                         /* next mirror */
4541                         if (stripe_count == 0)
4542                                 continue;
4543
4544                         LASSERT(objects != NULL && stripe_size != 0);
4545                         /* ll_do_div64(a, b) returns a % b, and a = a / b */
4546                         ll_do_div64(size, (__u64)stripe_size);
4547                         stripe = ll_do_div64(size, (__u64)stripe_count);
4548                         LASSERT(objects[stripe] != NULL);
4549
4550                         size = size * stripe_size;
4551                         offs = attr->la_size;
4552                         size += ll_do_div64(offs, stripe_size);
4553
4554                         attr->la_valid = LA_SIZE;
4555                         attr->la_size = size;
4556
4557                         rc = lod_sub_declare_attr_set(env, objects[stripe],
4558                                                       attr, th);
4559                 }
4560         }
4561
4562         RETURN(rc);
4563 }
4564
4565 /**
4566  * Declare creation of striped object.
4567  *
4568  * The function declares creation stripes for a regular object. The function
4569  * also declares whether the stripes will be created with non-zero size if
4570  * previously size was set non-zero on the master object. If object \a dt is
4571  * not local, then only fully defined striping can be applied in \a lovea.
4572  * Otherwise \a lovea can be in the form of pattern, see lod_qos_parse_config()
4573  * for the details.
4574  *
4575  * \param[in] env       execution environment
4576  * \param[in] dt        object
4577  * \param[in] attr      attributes the stripes will be created with
4578  * \param[in] lovea     a buffer containing striping description
4579  * \param[in] th        transaction handle
4580  *
4581  * \retval              0 on success
4582  * \retval              negative if failed
4583  */
4584 int lod_declare_striped_create(const struct lu_env *env, struct dt_object *dt,
4585                                struct lu_attr *attr,
4586                                const struct lu_buf *lovea, struct thandle *th)
4587 {
4588         struct lod_thread_info  *info = lod_env_info(env);
4589         struct dt_object        *next = dt_object_child(dt);
4590         struct lod_object       *lo = lod_dt_obj(dt);
4591         int                      rc;
4592         ENTRY;
4593
4594         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO))
4595                 GOTO(out, rc = -ENOMEM);
4596
4597         if (!dt_object_remote(next)) {
4598                 /* choose OST and generate appropriate objects */
4599                 rc = lod_prepare_create(env, lo, attr, lovea, th);
4600                 if (rc)
4601                         GOTO(out, rc);
4602
4603                 /*
4604                  * declare storage for striping data
4605                  */
4606                 info->lti_buf.lb_len = lod_comp_md_size(lo, false);
4607         } else {
4608                 /* LOD can not choose OST objects for remote objects, i.e.
4609                  * stripes must be ready before that. Right now, it can only
4610                  * happen during migrate, i.e. migrate process needs to create
4611                  * remote regular file (mdd_migrate_create), then the migrate
4612                  * process will provide stripeEA. */
4613                 LASSERT(lovea != NULL);
4614                 info->lti_buf = *lovea;
4615         }
4616
4617         rc = lod_sub_declare_xattr_set(env, next, &info->lti_buf,
4618                                        XATTR_NAME_LOV, 0, th);
4619         if (rc)
4620                 GOTO(out, rc);
4621
4622         /*
4623          * if striping is created with local object's size > 0,
4624          * we have to propagate this size to specific object
4625          * the case is possible only when local object was created previously
4626          */
4627         if (dt_object_exists(next))
4628                 rc = lod_declare_init_size(env, dt, th);
4629
4630 out:
4631         /* failed to create striping or to set initial size, let's reset
4632          * config so that others don't get confused */
4633         if (rc)
4634                 lod_object_free_striping(env, lo);
4635
4636         RETURN(rc);
4637 }
4638
4639 /**
4640  * Implementation of dt_object_operations::do_declare_create.
4641  *
4642  * The method declares creation of a new object. If the object will be striped,
4643  * then helper functions are called to find FIDs for the stripes, declare
4644  * creation of the stripes and declare initialization of the striping
4645  * information to be stored in the master object.
4646  *
4647  * \see dt_object_operations::do_declare_create() in the API description
4648  * for details.
4649  */
4650 static int lod_declare_create(const struct lu_env *env, struct dt_object *dt,
4651                               struct lu_attr *attr,
4652                               struct dt_allocation_hint *hint,
4653                               struct dt_object_format *dof, struct thandle *th)
4654 {
4655         struct dt_object   *next = dt_object_child(dt);
4656         struct lod_object  *lo = lod_dt_obj(dt);
4657         int                 rc;
4658         ENTRY;
4659
4660         LASSERT(dof);
4661         LASSERT(attr);
4662         LASSERT(th);
4663
4664         /*
4665          * first of all, we declare creation of local object
4666          */
4667         rc = lod_sub_declare_create(env, next, attr, hint, dof, th);
4668         if (rc != 0)
4669                 GOTO(out, rc);
4670
4671         /*
4672          * it's lod_ah_init() that has decided the object will be striped
4673          */
4674         if (dof->dof_type == DFT_REGULAR) {
4675                 /* callers don't want stripes */
4676                 /* XXX: all tricky interactions with ->ah_make_hint() decided
4677                  * to use striping, then ->declare_create() behaving differently
4678                  * should be cleaned */
4679                 if (dof->u.dof_reg.striped != 0)
4680                         rc = lod_declare_striped_create(env, dt, attr,
4681                                                         NULL, th);
4682         } else if (dof->dof_type == DFT_DIR) {
4683                 struct seq_server_site *ss;
4684                 struct lu_buf buf = { NULL };
4685                 struct lu_buf *lmu = NULL;
4686
4687                 ss = lu_site2seq(dt->do_lu.lo_dev->ld_site);
4688
4689                 /* If the parent has default stripeEA, and client
4690                  * did not find it before sending create request,
4691                  * then MDT will return -EREMOTE, and client will
4692                  * retrieve the default stripeEA and re-create the
4693                  * sub directory.
4694                  *
4695                  * Note: if dah_eadata != NULL, it means creating the
4696                  * striped directory with specified stripeEA, then it
4697                  * should ignore the default stripeEA */
4698                 if (hint != NULL && hint->dah_eadata == NULL) {
4699                         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_STALE_DIR_LAYOUT))
4700                                 GOTO(out, rc = -EREMOTE);
4701
4702                         if (lo->ldo_dir_stripe_offset == -1) {
4703                                 /* child and parent should be in the same MDT */
4704                                 if (hint->dah_parent != NULL &&
4705                                     dt_object_remote(hint->dah_parent))
4706                                         GOTO(out, rc = -EREMOTE);
4707                         } else if (lo->ldo_dir_stripe_offset !=
4708                                    ss->ss_node_id) {
4709                                 struct lod_device *lod;
4710                                 struct lod_tgt_descs *ltd;
4711                                 struct lod_tgt_desc *tgt = NULL;
4712                                 bool found_mdt = false;
4713                                 int i;
4714
4715                                 lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
4716                                 ltd = &lod->lod_mdt_descs;
4717                                 cfs_foreach_bit(ltd->ltd_tgt_bitmap, i) {
4718                                         tgt = LTD_TGT(ltd, i);
4719                                         if (tgt->ltd_index ==
4720                                                 lo->ldo_dir_stripe_offset) {
4721                                                 found_mdt = true;
4722                                                 break;
4723                                         }
4724                                 }
4725
4726                                 /* If the MDT indicated by stripe_offset can be
4727                                  * found, then tell client to resend the create
4728                                  * request to the correct MDT, otherwise return
4729                                  * error to client */
4730                                 if (found_mdt)
4731                                         GOTO(out, rc = -EREMOTE);
4732                                 else
4733                                         GOTO(out, rc = -EINVAL);
4734                         }
4735                 } else if (hint && hint->dah_eadata) {
4736                         lmu = &buf;
4737                         lmu->lb_buf = (void *)hint->dah_eadata;
4738                         lmu->lb_len = hint->dah_eadata_len;
4739                 }
4740
4741                 rc = lod_declare_dir_striping_create(env, dt, attr, lmu, dof,
4742                                                      th);
4743         }
4744 out:
4745         /* failed to create striping or to set initial size, let's reset
4746          * config so that others don't get confused */
4747         if (rc)
4748                 lod_object_free_striping(env, lo);
4749         RETURN(rc);
4750 }
4751
4752 /**
4753  * Generate component ID for new created component.
4754  *
4755  * \param[in] lo                LOD object
4756  * \param[in] comp_idx          index of ldo_comp_entries
4757  *
4758  * \retval                      component ID on success
4759  * \retval                      LCME_ID_INVAL on failure
4760  */
4761 static __u32 lod_gen_component_id(struct lod_object *lo,
4762                                   int mirror_id, int comp_idx)
4763 {
4764         struct lod_layout_component *lod_comp;
4765         __u32   id, start, end;
4766         int     i;
4767
4768         LASSERT(lo->ldo_comp_entries[comp_idx].llc_id == LCME_ID_INVAL);
4769
4770         lod_obj_inc_layout_gen(lo);
4771         id = lo->ldo_layout_gen;
4772         if (likely(id <= SEQ_ID_MAX))
4773                 RETURN(pflr_id(mirror_id, id & SEQ_ID_MASK));
4774
4775         /* Layout generation wraps, need to check collisions. */
4776         start = id & SEQ_ID_MASK;
4777         end = SEQ_ID_MAX;
4778 again:
4779         for (id = start; id <= end; id++) {
4780                 for (i = 0; i < lo->ldo_comp_cnt; i++) {
4781                         lod_comp = &lo->ldo_comp_entries[i];
4782                         if (pflr_id(mirror_id, id) == lod_comp->llc_id)
4783                                 break;
4784                 }
4785                 /* Found the ununsed ID */
4786                 if (i == lo->ldo_comp_cnt)
4787                         RETURN(pflr_id(mirror_id, id));
4788         }
4789         if (end == LCME_ID_MAX) {
4790                 start = 1;
4791                 end = min(lo->ldo_layout_gen & LCME_ID_MASK,
4792                           (__u32)(LCME_ID_MAX - 1));
4793                 goto again;
4794         }
4795
4796         RETURN(LCME_ID_INVAL);
4797 }
4798
4799 /**
4800  * Creation of a striped regular object.
4801  *
4802  * The function is called to create the stripe objects for a regular
4803  * striped file. This can happen at the initial object creation or
4804  * when the caller asks LOD to do so using ->do_xattr_set() method
4805  * (so called late striping). Notice all the information are already
4806  * prepared in the form of the list of objects (ldo_stripe field).
4807  * This is done during declare phase.
4808  *
4809  * \param[in] env       execution environment
4810  * \param[in] dt        object
4811  * \param[in] attr      attributes the stripes will be created with
4812  * \param[in] dof       format of stripes (see OSD API description)
4813  * \param[in] th        transaction handle
4814  *
4815  * \retval              0 on success
4816  * \retval              negative if failed
4817  */
4818 int lod_striped_create(const struct lu_env *env, struct dt_object *dt,
4819                        struct lu_attr *attr, struct dt_object_format *dof,
4820                        struct thandle *th)
4821 {
4822         struct lod_layout_component     *lod_comp;
4823         struct lod_object       *lo = lod_dt_obj(dt);
4824         __u16   mirror_id;
4825         int     rc = 0, i, j;
4826         ENTRY;
4827
4828         LASSERT(lo->ldo_comp_cnt != 0 && lo->ldo_comp_entries != NULL);
4829
4830         mirror_id = lo->ldo_mirror_count > 1 ? 1 : 0;
4831
4832         /* create all underlying objects */
4833         for (i = 0; i < lo->ldo_comp_cnt; i++) {
4834                 lod_comp = &lo->ldo_comp_entries[i];
4835
4836                 if (lod_comp->llc_extent.e_start == 0 && i > 0) /* new mirror */
4837                         ++mirror_id;
4838
4839                 if (lod_comp->llc_id == LCME_ID_INVAL) {
4840                         lod_comp->llc_id = lod_gen_component_id(lo,
4841                                                                 mirror_id, i);
4842                         if (lod_comp->llc_id == LCME_ID_INVAL)
4843                                 GOTO(out, rc = -ERANGE);
4844                 }
4845
4846                 if (lod_comp_inited(lod_comp))
4847                         continue;
4848
4849                 if (lod_comp->llc_pattern & LOV_PATTERN_F_RELEASED)
4850                         lod_comp_set_init(lod_comp);
4851
4852                 if (lov_pattern(lod_comp->llc_pattern) == LOV_PATTERN_MDT)
4853                         lod_comp_set_init(lod_comp);
4854
4855                 if (lod_comp->llc_stripe == NULL)
4856                         continue;
4857
4858                 LASSERT(lod_comp->llc_stripe_count);
4859                 for (j = 0; j < lod_comp->llc_stripe_count; j++) {
4860                         struct dt_object *object = lod_comp->llc_stripe[j];
4861                         LASSERT(object != NULL);
4862                         rc = lod_sub_create(env, object, attr, NULL, dof, th);
4863                         if (rc)
4864                                 GOTO(out, rc);
4865                 }
4866                 lod_comp_set_init(lod_comp);
4867         }
4868
4869         rc = lod_fill_mirrors(lo);
4870         if (rc)
4871                 GOTO(out, rc);
4872
4873         rc = lod_generate_and_set_lovea(env, lo, th);
4874         if (rc)
4875                 GOTO(out, rc);
4876
4877         lo->ldo_comp_cached = 1;
4878         RETURN(0);
4879
4880 out:
4881         lod_object_free_striping(env, lo);
4882         RETURN(rc);
4883 }
4884
4885 /**
4886  * Implementation of dt_object_operations::do_create.
4887  *
4888  * If any of preceeding methods (like ->do_declare_create(),
4889  * ->do_ah_init(), etc) chose to create a striped object,
4890  * then this method will create the master and the stripes.
4891  *
4892  * \see dt_object_operations::do_create() in the API description for details.
4893  */
4894 static int lod_create(const struct lu_env *env, struct dt_object *dt,
4895                       struct lu_attr *attr, struct dt_allocation_hint *hint,
4896                       struct dt_object_format *dof, struct thandle *th)
4897 {
4898         int                 rc;
4899         ENTRY;
4900
4901         /* create local object */
4902         rc = lod_sub_create(env, dt_object_child(dt), attr, hint, dof, th);
4903         if (rc != 0)
4904                 RETURN(rc);
4905
4906         if (S_ISREG(dt->do_lu.lo_header->loh_attr) &&
4907             lod_obj_is_striped(dt) && dof->u.dof_reg.striped != 0) {
4908                 LASSERT(lod_dt_obj(dt)->ldo_comp_cached == 0);
4909                 rc = lod_striped_create(env, dt, attr, dof, th);
4910         }
4911
4912         RETURN(rc);
4913 }
4914
4915 static inline int
4916 lod_obj_stripe_destroy_cb(const struct lu_env *env, struct lod_object *lo,
4917                           struct dt_object *dt, struct thandle *th,
4918                           int comp_idx, int stripe_idx,
4919                           struct lod_obj_stripe_cb_data *data)
4920 {
4921         if (data->locd_declare)
4922                 return lod_sub_declare_destroy(env, dt, th);
4923         else if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SPEOBJ) ||
4924                  stripe_idx == cfs_fail_val)
4925                 return lod_sub_destroy(env, dt, th);
4926         else
4927                 return 0;
4928 }
4929
4930 /**
4931  * Implementation of dt_object_operations::do_declare_destroy.
4932  *
4933  * If the object is a striped directory, then the function declares reference
4934  * removal from the master object (this is an index) to the stripes and declares
4935  * destroy of all the stripes. In all the cases, it declares an intention to
4936  * destroy the object itself.
4937  *
4938  * \see dt_object_operations::do_declare_destroy() in the API description
4939  * for details.
4940  */
4941 static int lod_declare_destroy(const struct lu_env *env, struct dt_object *dt,
4942                                struct thandle *th)
4943 {
4944         struct dt_object   *next = dt_object_child(dt);
4945         struct lod_object  *lo = lod_dt_obj(dt);
4946         struct lod_thread_info *info = lod_env_info(env);
4947         char               *stripe_name = info->lti_key;
4948         int                 rc, i;
4949         ENTRY;
4950
4951         /*
4952          * load striping information, notice we don't do this when object
4953          * is being initialized as we don't need this information till
4954          * few specific cases like destroy, chown
4955          */
4956         rc = lod_load_striping(env, lo);
4957         if (rc)
4958                 RETURN(rc);
4959
4960         /* declare destroy for all underlying objects */
4961         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
4962                 rc = next->do_ops->do_index_try(env, next,
4963                                                 &dt_directory_features);
4964                 if (rc != 0)
4965                         RETURN(rc);
4966
4967                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
4968                         rc = lod_sub_declare_ref_del(env, next, th);
4969                         if (rc != 0)
4970                                 RETURN(rc);
4971
4972                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
4973                                 PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)),
4974                                 i);
4975                         rc = lod_sub_declare_delete(env, next,
4976                                         (const struct dt_key *)stripe_name, th);
4977                         if (rc != 0)
4978                                 RETURN(rc);
4979                 }
4980         }
4981
4982         /*
4983          * we declare destroy for the local object
4984          */
4985         rc = lod_sub_declare_destroy(env, next, th);
4986         if (rc)
4987                 RETURN(rc);
4988
4989         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ) ||
4990             OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ2))
4991                 RETURN(0);
4992
4993         if (!lod_obj_is_striped(dt))
4994                 RETURN(0);
4995
4996         /* declare destroy all striped objects */
4997         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
4998                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
4999                         if (lo->ldo_stripe[i] == NULL)
5000                                 continue;
5001
5002                         rc = lod_sub_declare_ref_del(env, lo->ldo_stripe[i],
5003                                                      th);
5004
5005                         rc = lod_sub_declare_destroy(env, lo->ldo_stripe[i],
5006                                                      th);
5007                         if (rc != 0)
5008                                 break;
5009                 }
5010         } else {
5011                 struct lod_obj_stripe_cb_data data = { { 0 } };
5012
5013                 data.locd_declare = true;
5014                 data.locd_stripe_cb = lod_obj_stripe_destroy_cb;
5015                 rc = lod_obj_for_each_stripe(env, lo, th, &data);
5016         }
5017
5018         RETURN(rc);
5019 }
5020
5021 /**
5022  * Implementation of dt_object_operations::do_destroy.
5023  *
5024  * If the object is a striped directory, then the function removes references
5025  * from the master object (this is an index) to the stripes and destroys all
5026  * the stripes. In all the cases, the function destroys the object itself.
5027  *
5028  * \see dt_object_operations::do_destroy() in the API description for details.
5029  */
5030 static int lod_destroy(const struct lu_env *env, struct dt_object *dt,
5031                        struct thandle *th)
5032 {
5033         struct dt_object  *next = dt_object_child(dt);
5034         struct lod_object *lo = lod_dt_obj(dt);
5035         struct lod_thread_info *info = lod_env_info(env);
5036         char               *stripe_name = info->lti_key;
5037         unsigned int       i;
5038         int                rc;
5039         ENTRY;
5040
5041         /* destroy sub-stripe of master object */
5042         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
5043                 rc = next->do_ops->do_index_try(env, next,
5044                                                 &dt_directory_features);
5045                 if (rc != 0)
5046                         RETURN(rc);
5047
5048                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
5049                         rc = lod_sub_ref_del(env, next, th);
5050                         if (rc != 0)
5051                                 RETURN(rc);
5052
5053                         snprintf(stripe_name, sizeof(info->lti_key), DFID":%d",
5054                                 PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)),
5055                                 i);
5056
5057                         CDEBUG(D_INFO, DFID" delete stripe %s "DFID"\n",
5058                                PFID(lu_object_fid(&dt->do_lu)), stripe_name,
5059                                PFID(lu_object_fid(&lo->ldo_stripe[i]->do_lu)));
5060
5061                         rc = lod_sub_delete(env, next,
5062                                        (const struct dt_key *)stripe_name, th);
5063                         if (rc != 0)
5064                                 RETURN(rc);
5065                 }
5066         }
5067
5068         rc = lod_sub_destroy(env, next, th);
5069         if (rc != 0)
5070                 RETURN(rc);
5071
5072         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ) ||
5073             OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_MDTOBJ2))
5074                 RETURN(0);
5075
5076         if (!lod_obj_is_striped(dt))
5077                 RETURN(0);
5078
5079         /* destroy all striped objects */
5080         if (S_ISDIR(dt->do_lu.lo_header->loh_attr)) {
5081                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
5082                         if (lo->ldo_stripe[i] == NULL)
5083                                 continue;
5084                         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LOST_SPEOBJ) ||
5085                             i == cfs_fail_val) {
5086                                 dt_write_lock(env, lo->ldo_stripe[i],
5087                                               MOR_TGT_CHILD);
5088                                 rc = lod_sub_ref_del(env, lo->ldo_stripe[i],
5089                                                      th);
5090                                 dt_write_unlock(env, lo->ldo_stripe[i]);
5091                                 if (rc != 0)
5092                                         break;
5093
5094                                 rc = lod_sub_destroy(env, lo->ldo_stripe[i],
5095                                                      th);
5096                                 if (rc != 0)
5097                                         break;
5098                         }
5099                 }
5100         } else {
5101                 struct lod_obj_stripe_cb_data data = { { 0 } };
5102
5103                 data.locd_declare = false;
5104                 data.locd_stripe_cb = lod_obj_stripe_destroy_cb;
5105                 rc = lod_obj_for_each_stripe(env, lo, th, &data);
5106         }
5107
5108         RETURN(rc);
5109 }
5110
5111 /**
5112  * Implementation of dt_object_operations::do_declare_ref_add.
5113  *
5114  * \see dt_object_operations::do_declare_ref_add() in the API description
5115  * for details.
5116  */
5117 static int lod_declare_ref_add(const struct lu_env *env,
5118                                struct dt_object *dt, struct thandle *th)
5119 {
5120         return lod_sub_declare_ref_add(env, dt_object_child(dt), th);
5121 }
5122
5123 /**
5124  * Implementation of dt_object_operations::do_ref_add.
5125  *
5126  * \see dt_object_operations::do_ref_add() in the API description for details.
5127  */
5128 static int lod_ref_add(const struct lu_env *env,
5129                        struct dt_object *dt, struct thandle *th)
5130 {
5131         return lod_sub_ref_add(env, dt_object_child(dt), th);
5132 }
5133
5134 /**
5135  * Implementation of dt_object_operations::do_declare_ref_del.
5136  *
5137  * \see dt_object_operations::do_declare_ref_del() in the API description
5138  * for details.
5139  */
5140 static int lod_declare_ref_del(const struct lu_env *env,
5141                                struct dt_object *dt, struct thandle *th)
5142 {
5143         return lod_sub_declare_ref_del(env, dt_object_child(dt), th);
5144 }
5145
5146 /**
5147  * Implementation of dt_object_operations::do_ref_del
5148  *
5149  * \see dt_object_operations::do_ref_del() in the API description for details.
5150  */
5151 static int lod_ref_del(const struct lu_env *env,
5152                        struct dt_object *dt, struct thandle *th)
5153 {
5154         return lod_sub_ref_del(env, dt_object_child(dt), th);
5155 }
5156
5157 /**
5158  * Implementation of dt_object_operations::do_object_sync.
5159  *
5160  * \see dt_object_operations::do_object_sync() in the API description
5161  * for details.
5162  */
5163 static int lod_object_sync(const struct lu_env *env, struct dt_object *dt,
5164                            __u64 start, __u64 end)
5165 {
5166         return dt_object_sync(env, dt_object_child(dt), start, end);
5167 }
5168
5169 /**
5170  * Implementation of dt_object_operations::do_object_unlock.
5171  *
5172  * Used to release LDLM lock(s).
5173  *
5174  * \see dt_object_operations::do_object_unlock() in the API description
5175  * for details.
5176  */
5177 static int lod_object_unlock(const struct lu_env *env, struct dt_object *dt,
5178                              struct ldlm_enqueue_info *einfo,
5179                              union ldlm_policy_data *policy)
5180 {
5181         struct lod_object *lo = lod_dt_obj(dt);
5182         struct lustre_handle_array *slave_locks = einfo->ei_cbdata;
5183         int slave_locks_size;
5184         int i;
5185         ENTRY;
5186
5187         if (slave_locks == NULL)
5188                 RETURN(0);
5189
5190         LASSERT(S_ISDIR(dt->do_lu.lo_header->loh_attr));
5191         LASSERT(lo->ldo_dir_stripe_count > 1);
5192         /* Note: for remote lock for single stripe dir, MDT will cancel
5193          * the lock by lockh directly */
5194         LASSERT(!dt_object_remote(dt_object_child(dt)));
5195
5196         /* locks were unlocked in MDT layer */
5197         for (i = 0; i < slave_locks->ha_count; i++)
5198                 LASSERT(!lustre_handle_is_used(&slave_locks->ha_handles[i]));
5199
5200         /*
5201          * NB, ha_count may not equal to ldo_dir_stripe_count, because dir
5202          * layout may change, e.g., shrink dir layout after migration.
5203          */
5204         for (i = 0; i < lo->ldo_dir_stripe_count; i++)
5205                 dt_invalidate(env, lo->ldo_stripe[i]);
5206
5207         slave_locks_size = offsetof(typeof(*slave_locks),
5208                                     ha_handles[slave_locks->ha_count]);
5209         OBD_FREE(slave_locks, slave_locks_size);
5210         einfo->ei_cbdata = NULL;
5211
5212         RETURN(0);
5213 }
5214
5215 /**
5216  * Implementation of dt_object_operations::do_object_lock.
5217  *
5218  * Used to get LDLM lock on the non-striped and striped objects.
5219  *
5220  * \see dt_object_operations::do_object_lock() in the API description
5221  * for details.
5222  */
5223 static int lod_object_lock(const struct lu_env *env,
5224                            struct dt_object *dt,
5225                            struct lustre_handle *lh,
5226                            struct ldlm_enqueue_info *einfo,
5227                            union ldlm_policy_data *policy)
5228 {
5229         struct lod_object *lo = lod_dt_obj(dt);
5230         int slave_locks_size;
5231         struct lustre_handle_array *slave_locks = NULL;
5232         int i;
5233         int rc;
5234         ENTRY;
5235
5236         /* remote object lock */
5237         if (!einfo->ei_enq_slave) {
5238                 LASSERT(dt_object_remote(dt));
5239                 return dt_object_lock(env, dt_object_child(dt), lh, einfo,
5240                                       policy);
5241         }
5242
5243         if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
5244                 RETURN(-ENOTDIR);
5245
5246         rc = lod_load_striping(env, lo);
5247         if (rc != 0)
5248                 RETURN(rc);
5249
5250         /* No stripes */
5251         if (lo->ldo_dir_stripe_count <= 1)
5252                 RETURN(0);
5253
5254         slave_locks_size = offsetof(typeof(*slave_locks),
5255                                     ha_handles[lo->ldo_dir_stripe_count]);
5256         /* Freed in lod_object_unlock */
5257         OBD_ALLOC(slave_locks, slave_locks_size);
5258         if (!slave_locks)
5259                 RETURN(-ENOMEM);
5260         slave_locks->ha_count = lo->ldo_dir_stripe_count;
5261
5262         /* striped directory lock */
5263         for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
5264                 struct lustre_handle lockh;
5265                 struct ldlm_res_id *res_id;
5266
5267                 res_id = &lod_env_info(env)->lti_res_id;
5268                 fid_build_reg_res_name(lu_object_fid(&lo->ldo_stripe[i]->do_lu),
5269                                        res_id);
5270                 einfo->ei_res_id = res_id;
5271
5272                 LASSERT(lo->ldo_stripe[i] != NULL);
5273                 if (dt_object_remote(lo->ldo_stripe[i])) {
5274                         set_bit(i, (void *)slave_locks->ha_map);
5275                         rc = dt_object_lock(env, lo->ldo_stripe[i], &lockh,
5276                                             einfo, policy);
5277                 } else {
5278                         struct ldlm_namespace *ns = einfo->ei_namespace;
5279                         ldlm_blocking_callback blocking = einfo->ei_cb_local_bl;
5280                         ldlm_completion_callback completion = einfo->ei_cb_cp;
5281                         __u64 dlmflags = LDLM_FL_ATOMIC_CB;
5282
5283                         if (einfo->ei_mode == LCK_PW ||
5284                             einfo->ei_mode == LCK_EX)
5285                                 dlmflags |= LDLM_FL_COS_INCOMPAT;
5286
5287                         LASSERT(ns != NULL);
5288                         rc = ldlm_cli_enqueue_local(ns, res_id, LDLM_IBITS,
5289                                                     policy, einfo->ei_mode,
5290                                                     &dlmflags, blocking,
5291                                                     completion, NULL,
5292                                                     NULL, 0, LVB_T_NONE,
5293                                                     NULL, &lockh);
5294                 }
5295                 if (rc) {
5296                         while (i--)
5297                                 ldlm_lock_decref_and_cancel(
5298                                                 &slave_locks->ha_handles[i],
5299                                                 einfo->ei_mode);
5300                         OBD_FREE(slave_locks, slave_locks_size);
5301                         RETURN(rc);
5302                 }
5303                 slave_locks->ha_handles[i] = lockh;
5304         }
5305         einfo->ei_cbdata = slave_locks;
5306
5307         RETURN(0);
5308 }
5309
5310 /**
5311  * Implementation of dt_object_operations::do_invalidate.
5312  *
5313  * \see dt_object_operations::do_invalidate() in the API description for details
5314  */
5315 static int lod_invalidate(const struct lu_env *env, struct dt_object *dt)
5316 {
5317         return dt_invalidate(env, dt_object_child(dt));
5318 }
5319
5320 static int lod_layout_data_init(struct lod_thread_info *info, __u32 comp_cnt)
5321 {
5322         ENTRY;
5323
5324         /* clear memory region that will be used for layout change */
5325         memset(&info->lti_layout_attr, 0, sizeof(struct lu_attr));
5326         info->lti_count = 0;
5327
5328         if (info->lti_comp_size >= comp_cnt)
5329                 RETURN(0);
5330
5331         if (info->lti_comp_size > 0) {
5332                 OBD_FREE(info->lti_comp_idx,
5333                          info->lti_comp_size * sizeof(__u32));
5334                 info->lti_comp_size = 0;
5335         }
5336
5337         OBD_ALLOC(info->lti_comp_idx, comp_cnt * sizeof(__u32));
5338         if (!info->lti_comp_idx)
5339                 RETURN(-ENOMEM);
5340
5341         info->lti_comp_size = comp_cnt;
5342         RETURN(0);
5343 }
5344
5345 static int lod_declare_instantiate_components(const struct lu_env *env,
5346                 struct lod_object *lo, struct thandle *th)
5347 {
5348         struct lod_thread_info *info = lod_env_info(env);
5349         struct ost_pool *inuse = &info->lti_inuse_osts;
5350         int i;
5351         int rc = 0;
5352         ENTRY;
5353
5354         LASSERT(info->lti_count < lo->ldo_comp_cnt);
5355         if (info->lti_count > 0) {
5356                 /* Prepare inuse array for composite file */
5357                 rc = lod_prepare_inuse(env, lo);
5358                 if (rc)
5359                         RETURN(rc);
5360         }
5361
5362         for (i = 0; i < info->lti_count; i++) {
5363                 rc = lod_qos_prep_create(env, lo, NULL, th,
5364                                          info->lti_comp_idx[i], inuse);
5365                 if (rc)
5366                         break;
5367         }
5368
5369         if (!rc) {
5370                 info->lti_buf.lb_len = lod_comp_md_size(lo, false);
5371                 rc = lod_sub_declare_xattr_set(env, lod_object_child(lo),
5372                                 &info->lti_buf, XATTR_NAME_LOV, 0, th);
5373         }
5374
5375         RETURN(rc);
5376 }
5377
5378 static int lod_declare_update_plain(const struct lu_env *env,
5379                 struct lod_object *lo, struct layout_intent *layout,
5380                 const struct lu_buf *buf, struct thandle *th)
5381 {
5382         struct lod_thread_info *info = lod_env_info(env);
5383         struct lod_device *d = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
5384         struct lod_layout_component *lod_comp;
5385         struct lov_comp_md_v1 *comp_v1 = NULL;
5386         bool replay = false;
5387         int i, rc;
5388         ENTRY;
5389
5390         LASSERT(lo->ldo_flr_state == LCM_FL_NONE);
5391
5392         /*
5393          * In case the client is passing lovea, which only happens during
5394          * the replay of layout intent write RPC for now, we may need to
5395          * parse the lovea and apply new layout configuration.
5396          */
5397         if (buf && buf->lb_len)  {
5398                 struct lov_user_md_v1 *v1 = buf->lb_buf;
5399
5400                 if (v1->lmm_magic != (LOV_MAGIC_DEFINED | LOV_MAGIC_COMP_V1) &&
5401                     v1->lmm_magic != __swab32(LOV_MAGIC_DEFINED |
5402                                               LOV_MAGIC_COMP_V1)) {
5403                         CERROR("%s: the replay buffer of layout extend "
5404                                "(magic %#x) does not contain expected "
5405                                "composite layout.\n",
5406                                lod2obd(d)->obd_name, v1->lmm_magic);
5407                         GOTO(out, rc = -EINVAL);
5408                 }
5409
5410                 lod_object_free_striping(env, lo);
5411                 rc = lod_use_defined_striping(env, lo, buf);
5412                 if (rc)
5413                         GOTO(out, rc);
5414
5415                 rc = lod_get_lov_ea(env, lo);
5416                 if (rc <= 0)
5417                         GOTO(out, rc);
5418                 /* old on-disk EA is stored in info->lti_buf */
5419                 comp_v1 = (struct lov_comp_md_v1 *)info->lti_buf.lb_buf;
5420                 replay = true;
5421         } else {
5422                 /* non replay path */
5423                 rc = lod_load_striping_locked(env, lo);
5424                 if (rc)
5425                         GOTO(out, rc);
5426         }
5427
5428         /* Make sure defined layout covers the requested write range. */
5429         lod_comp = &lo->ldo_comp_entries[lo->ldo_comp_cnt - 1];
5430         if (lo->ldo_comp_cnt > 1 &&
5431             lod_comp->llc_extent.e_end != OBD_OBJECT_EOF &&
5432             lod_comp->llc_extent.e_end < layout->li_extent.e_end) {
5433                 CDEBUG(replay ? D_ERROR : D_LAYOUT,
5434                        "%s: the defined layout [0, %#llx) does not covers "
5435                        "the write range "DEXT"\n",
5436                        lod2obd(d)->obd_name, lod_comp->llc_extent.e_end,
5437                        PEXT(&layout->li_extent));
5438                 GOTO(out, rc = -EINVAL);
5439         }
5440
5441         CDEBUG(D_LAYOUT, "%s: "DFID": instantiate components "DEXT"\n",
5442                lod2obd(d)->obd_name, PFID(lod_object_fid(lo)),
5443                PEXT(&layout->li_extent));
5444
5445         /*
5446          * Iterate ld->ldo_comp_entries, find the component whose extent under
5447          * the write range and not instantianted.
5448          */
5449         for (i = 0; i < lo->ldo_comp_cnt; i++) {
5450                 lod_comp = &lo->ldo_comp_entries[i];
5451
5452                 if (lod_comp->llc_extent.e_start >= layout->li_extent.e_end)
5453                         break;
5454
5455                 if (!replay) {
5456                         if (lod_comp_inited(lod_comp))
5457                                 continue;
5458                 } else {
5459                         /**
5460                          * In replay path, lod_comp is the EA passed by
5461                          * client replay buffer,  comp_v1 is the pre-recovery
5462                          * on-disk EA, we'd sift out those components which
5463                          * were init-ed in the on-disk EA.
5464                          */
5465                         if (le32_to_cpu(comp_v1->lcm_entries[i].lcme_flags) &
5466                             LCME_FL_INIT)
5467                                 continue;
5468                 }
5469                 /*
5470                  * this component hasn't instantiated in normal path, or during
5471                  * replay it needs replay the instantiation.
5472                  */
5473
5474                 /* A released component is being extended */
5475                 if (lod_comp->llc_pattern & LOV_PATTERN_F_RELEASED)
5476                         GOTO(out, rc = -EINVAL);
5477
5478                 LASSERT(info->lti_comp_idx != NULL);
5479                 info->lti_comp_idx[info->lti_count++] = i;
5480         }
5481
5482         if (info->lti_count == 0)
5483                 RETURN(-EALREADY);
5484
5485         lod_obj_inc_layout_gen(lo);
5486         rc = lod_declare_instantiate_components(env, lo, th);
5487 out:
5488         if (rc)
5489                 lod_object_free_striping(env, lo);
5490         RETURN(rc);
5491 }
5492
5493 #define lod_foreach_mirror_comp(comp, lo, mirror_idx)                      \
5494 for (comp = &lo->ldo_comp_entries[lo->ldo_mirrors[mirror_idx].lme_start];  \
5495      comp <= &lo->ldo_comp_entries[lo->ldo_mirrors[mirror_idx].lme_end];   \
5496      comp++)
5497
5498 static inline int lod_comp_index(struct lod_object *lo,
5499                                  struct lod_layout_component *lod_comp)
5500 {
5501         LASSERT(lod_comp >= lo->ldo_comp_entries &&
5502                 lod_comp <= &lo->ldo_comp_entries[lo->ldo_comp_cnt - 1]);
5503
5504         return lod_comp - lo->ldo_comp_entries;
5505 }
5506
5507 /**
5508  * Stale other mirrors by writing extent.
5509  */
5510 static void lod_stale_components(struct lod_object *lo, int primary,
5511                                  struct lu_extent *extent)
5512 {
5513         struct lod_layout_component *pri_comp, *lod_comp;
5514         int i;
5515
5516         /* The writing extent decides which components in the primary
5517          * are affected... */
5518         CDEBUG(D_LAYOUT, "primary mirror %d, "DEXT"\n", primary, PEXT(extent));
5519         lod_foreach_mirror_comp(pri_comp, lo, primary) {
5520                 if (!lu_extent_is_overlapped(extent, &pri_comp->llc_extent))
5521                         continue;
5522
5523                 CDEBUG(D_LAYOUT, "primary comp %u "DEXT"\n",
5524                        lod_comp_index(lo, pri_comp),
5525                        PEXT(&pri_comp->llc_extent));
5526
5527                 for (i = 0; i < lo->ldo_mirror_count; i++) {
5528                         if (i == primary)
5529                                 continue;
5530
5531                         /* ... and then stale other components that are
5532                          * overlapping with primary components */
5533                         lod_foreach_mirror_comp(lod_comp, lo, i) {
5534                                 if (!lu_extent_is_overlapped(
5535                                                         &pri_comp->llc_extent,
5536                                                         &lod_comp->llc_extent))
5537                                         continue;
5538
5539                                 CDEBUG(D_LAYOUT, "stale: %u / %u\n",
5540                                       i, lod_comp_index(lo, lod_comp));
5541
5542                                 lod_comp->llc_flags |= LCME_FL_STALE;
5543                                 lo->ldo_mirrors[i].lme_stale = 1;
5544                         }
5545                 }
5546         }
5547 }
5548
5549 /**
5550  * check an OST's availability
5551  * \param[in] env       execution environment
5552  * \param[in] lo        lod object
5553  * \param[in] dt        dt object
5554  * \param[in] index     mirror index
5555  *
5556  * \retval      negative if failed
5557  * \retval      1 if \a dt is available
5558  * \retval      0 if \a dt is not available
5559  */
5560 static inline int lod_check_ost_avail(const struct lu_env *env,
5561                                       struct lod_object *lo,
5562                                       struct dt_object *dt, int index)
5563 {
5564         struct lod_device *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
5565         struct lod_tgt_desc *ost;
5566         __u32 idx;
5567         int type = LU_SEQ_RANGE_OST;
5568         int rc;
5569
5570         rc = lod_fld_lookup(env, lod, lu_object_fid(&dt->do_lu), &idx, &type);
5571         if (rc < 0) {
5572                 CERROR("%s: can't locate "DFID":rc = %d\n",
5573                        lod2obd(lod)->obd_name, PFID(lu_object_fid(&dt->do_lu)),
5574                        rc);
5575                 return rc;
5576         }
5577
5578         ost = OST_TGT(lod, idx);
5579         if (ost->ltd_statfs.os_state &
5580                 (OS_STATE_READONLY | OS_STATE_ENOSPC | OS_STATE_ENOINO) ||
5581             ost->ltd_active == 0) {
5582                 CDEBUG(D_LAYOUT, DFID ": mirror %d OST%d unavail, rc = %d\n",
5583                        PFID(lod_object_fid(lo)), index, idx, rc);
5584                 return 0;
5585         }
5586
5587         return 1;
5588 }
5589
5590 /**
5591  * Pick primary mirror for write
5592  * \param[in] env       execution environment
5593  * \param[in] lo        object
5594  * \param[in] extent    write range
5595  */
5596 static int lod_primary_pick(const struct lu_env *env, struct lod_object *lo,
5597                             struct lu_extent *extent)
5598 {
5599         struct lod_device *lod = lu2lod_dev(lo->ldo_obj.do_lu.lo_dev);
5600         unsigned int seq = 0;
5601         struct lod_layout_component *lod_comp;
5602         int i, j, rc;
5603         int picked = -1, second_pick = -1, third_pick = -1;
5604         ENTRY;
5605
5606         if (OBD_FAIL_CHECK(OBD_FAIL_FLR_RANDOM_PICK_MIRROR)) {
5607                 get_random_bytes(&seq, sizeof(seq));
5608                 seq %= lo->ldo_mirror_count;
5609         }
5610
5611         /**
5612          * Pick a mirror as the primary, and check the availability of OSTs.
5613          *
5614          * This algo can be revised later after knowing the topology of
5615          * cluster.
5616          */
5617         lod_qos_statfs_update(env, lod);
5618         for (i = 0; i < lo->ldo_mirror_count; i++) {
5619                 bool ost_avail = true;
5620                 int index = (i + seq) % lo->ldo_mirror_count;
5621
5622                 if (lo->ldo_mirrors[index].lme_stale) {
5623                         CDEBUG(D_LAYOUT, DFID": mirror %d stale\n",
5624                                PFID(lod_object_fid(lo)), index);
5625                         continue;
5626                 }
5627
5628                 /* 2nd pick is for the primary mirror containing unavail OST */
5629                 if (lo->ldo_mirrors[index].lme_primary && second_pick < 0)
5630                         second_pick = index;
5631
5632                 /* 3rd pick is for non-primary mirror containing unavail OST */
5633                 if (second_pick < 0 && third_pick < 0)
5634                         third_pick = index;
5635
5636                 /**
5637                  * we found a non-primary 1st pick, we'd like to find a
5638                  * potential pirmary mirror.
5639                  */
5640                 if (picked >= 0 && !lo->ldo_mirrors[index].lme_primary)
5641                         continue;
5642
5643                 /* check the availability of OSTs */
5644                 lod_foreach_mirror_comp(lod_comp, lo, index) {
5645                         if (!lod_comp_inited(lod_comp) || !lod_comp->llc_stripe)
5646                                 continue;
5647
5648                         for (j = 0; j < lod_comp->llc_stripe_count; j++) {
5649                                 struct dt_object *dt = lod_comp->llc_stripe[j];
5650
5651                                 rc = lod_check_ost_avail(env, lo, dt, index);
5652                                 if (rc < 0)
5653                                         RETURN(rc);
5654
5655                                 ost_avail = !!rc;
5656                                 if (!ost_avail)
5657                                         break;
5658                         } /* for all dt object in one component */
5659                         if (!ost_avail)
5660                                 break;
5661                 } /* for all components in a mirror */
5662
5663                 /**
5664                  * the OSTs where allocated objects locates in the components
5665                  * of the mirror are available.
5666                  */
5667                 if (!ost_avail)
5668                         continue;
5669
5670                 /* this mirror has all OSTs available */
5671                 picked = index;
5672
5673                 /**
5674                  * primary with all OSTs are available, this is the perfect
5675                  * 1st pick.
5676                  */
5677                 if (lo->ldo_mirrors[index].lme_primary)
5678                         break;
5679         } /* for all mirrors */
5680
5681         /* failed to pick a sound mirror, lower our expectation */
5682         if (picked < 0)
5683                 picked = second_pick;
5684         if (picked < 0)
5685                 picked = third_pick;
5686         if (picked < 0)
5687                 RETURN(-ENODATA);
5688
5689         RETURN(picked);
5690 }
5691
5692 /**
5693  * figure out the components should be instantiated for resync.
5694  */
5695 static int lod_prepare_resync(const struct lu_env *env, struct lod_object *lo,
5696                               struct lu_extent *extent)
5697 {
5698         struct lod_thread_info *info = lod_env_info(env);
5699         struct lod_layout_component *lod_comp;
5700         unsigned int need_sync = 0;
5701         int i;
5702
5703         CDEBUG(D_LAYOUT,
5704                DFID": instantiate all stale components in "DEXT"\n",
5705                PFID(lod_object_fid(lo)), PEXT(extent));
5706
5707         /**
5708          * instantiate all components within this extent, even non-stale
5709          * components.
5710          */
5711         for (i = 0; i < lo->ldo_mirror_count; i++) {
5712                 if (!lo->ldo_mirrors[i].lme_stale)
5713                         continue;
5714
5715                 lod_foreach_mirror_comp(lod_comp, lo, i) {
5716                         if (!lu_extent_is_overlapped(extent,
5717                                                 &lod_comp->llc_extent))
5718                                 break;
5719
5720                         need_sync++;
5721
5722                         if (lod_comp_inited(lod_comp))
5723                                 continue;
5724
5725                         CDEBUG(D_LAYOUT, "resync instantiate %d / %d\n",
5726                                i, lod_comp_index(lo, lod_comp));
5727                         info->lti_comp_idx[info->lti_count++] =
5728                                         lod_comp_index(lo, lod_comp);
5729                 }
5730         }
5731
5732         return need_sync ? 0 : -EALREADY;
5733 }
5734
5735 static int lod_declare_update_rdonly(const struct lu_env *env,
5736                 struct lod_object *lo, struct md_layout_change *mlc,
5737                 struct thandle *th)
5738 {
5739         struct lod_thread_info *info = lod_env_info(env);
5740         struct lu_attr *layout_attr = &info->lti_layout_attr;
5741         struct lod_layout_component *lod_comp;
5742         struct lu_extent extent = { 0 };
5743         int rc;
5744         ENTRY;
5745
5746         LASSERT(lo->ldo_flr_state == LCM_FL_RDONLY);
5747         LASSERT(mlc->mlc_opc == MD_LAYOUT_WRITE ||
5748                 mlc->mlc_opc == MD_LAYOUT_RESYNC);
5749         LASSERT(lo->ldo_mirror_count > 0);
5750
5751         if (mlc->mlc_opc == MD_LAYOUT_WRITE) {
5752                 struct layout_intent *layout = mlc->mlc_intent;
5753                 int picked;
5754
5755                 extent = layout->li_extent;
5756                 CDEBUG(D_LAYOUT, DFID": trying to write :"DEXT"\n",
5757                        PFID(lod_object_fid(lo)), PEXT(&extent));
5758
5759                 picked = lod_primary_pick(env, lo, &extent);
5760                 if (picked < 0)
5761                         RETURN(picked);
5762
5763                 CDEBUG(D_LAYOUT, DFID": picked mirror id %u as primary\n",
5764                        PFID(lod_object_fid(lo)),
5765                        lo->ldo_mirrors[picked].lme_id);
5766
5767                 if (layout->li_opc == LAYOUT_INTENT_TRUNC) {
5768                         /**
5769                          * trunc transfers [0, size) in the intent extent, we'd
5770                          * stale components overlapping [size, eof).
5771                          */
5772                         extent.e_start = extent.e_end;
5773                         extent.e_end = OBD_OBJECT_EOF;
5774                 }
5775
5776                 /* stale overlapping components from other mirrors */
5777                 lod_stale_components(lo, picked, &extent);
5778
5779                 /* restore truncate intent extent */
5780                 if (layout->li_opc == LAYOUT_INTENT_TRUNC)
5781                         extent.e_end = extent.e_start;
5782
5783                 /* instantiate components for the picked mirror, start from 0 */
5784                 extent.e_start = 0;
5785
5786                 lod_foreach_mirror_comp(lod_comp, lo, picked) {
5787                         if (!lu_extent_is_overlapped(&extent,
5788                                                      &lod_comp->llc_extent))
5789                                 break;
5790
5791                         if (lod_comp_inited(lod_comp))
5792                                 continue;
5793
5794                         info->lti_comp_idx[info->lti_count++] =
5795                                                 lod_comp_index(lo, lod_comp);
5796                 }
5797
5798                 lo->ldo_flr_state = LCM_FL_WRITE_PENDING;
5799         } else { /* MD_LAYOUT_RESYNC */
5800                 int i;
5801
5802                 /**
5803                  * could contain multiple non-stale mirrors, so we need to
5804                  * prep uninited all components assuming any non-stale mirror
5805                  * could be picked as the primary mirror.
5806                  */
5807                 for (i = 0; i < lo->ldo_mirror_count; i++) {
5808                         if (lo->ldo_mirrors[i].lme_stale)
5809                                 continue;
5810
5811                         lod_foreach_mirror_comp(lod_comp, lo, i) {
5812                                 if (!lod_comp_inited(lod_comp))
5813                                         break;
5814
5815                                 if (extent.e_end < lod_comp->llc_extent.e_end)
5816                                         extent.e_end =
5817                                                 lod_comp->llc_extent.e_end;
5818                         }
5819                 }
5820
5821                 rc = lod_prepare_resync(env, lo, &extent);
5822                 if (rc)
5823                         GOTO(out, rc);
5824                 /* change the file state to SYNC_PENDING */
5825                 lo->ldo_flr_state = LCM_FL_SYNC_PENDING;
5826         }
5827
5828         /* Reset the layout version once it's becoming too large.
5829          * This way it can make sure that the layout version is
5830          * monotonously increased in this writing era. */
5831         lod_obj_inc_layout_gen(lo);
5832         if (lo->ldo_layout_gen > (LCME_ID_MAX >> 1)) {
5833                 __u32 layout_version;
5834
5835                 cfs_get_random_bytes(&layout_version, sizeof(layout_version));
5836                 lo->ldo_layout_gen = layout_version & 0xffff;
5837         }
5838
5839         rc = lod_declare_instantiate_components(env, lo, th);
5840         if (rc)
5841                 GOTO(out, rc);
5842
5843         layout_attr->la_valid = LA_LAYOUT_VERSION;
5844         layout_attr->la_layout_version = 0; /* set current version */
5845         if (mlc->mlc_opc == MD_LAYOUT_RESYNC)
5846                 layout_attr->la_layout_version = LU_LAYOUT_RESYNC;
5847         rc = lod_declare_attr_set(env, &lo->ldo_obj, layout_attr, th);
5848         if (rc)
5849                 GOTO(out, rc);
5850
5851 out:
5852         if (rc)
5853                 lod_object_free_striping(env, lo);
5854         RETURN(rc);
5855 }
5856
5857 static int lod_declare_update_write_pending(const struct lu_env *env,
5858                 struct lod_object *lo, struct md_layout_change *mlc,
5859                 struct thandle *th)
5860 {
5861         struct lod_thread_info *info = lod_env_info(env);
5862         struct lu_attr *layout_attr = &info->lti_layout_attr;
5863         struct lod_layout_component *lod_comp;
5864         struct lu_extent extent = { 0 };
5865         int primary = -1;
5866         int i;
5867         int rc;
5868         ENTRY;
5869
5870         LASSERT(lo->ldo_flr_state == LCM_FL_WRITE_PENDING);
5871         LASSERT(mlc->mlc_opc == MD_LAYOUT_WRITE ||
5872                 mlc->mlc_opc == MD_LAYOUT_RESYNC);
5873
5874         /* look for the primary mirror */
5875         for (i = 0; i < lo->ldo_mirror_count; i++) {
5876                 if (lo->ldo_mirrors[i].lme_stale)
5877                         continue;
5878
5879                 LASSERTF(primary < 0, DFID " has multiple primary: %u / %u",
5880                          PFID(lod_object_fid(lo)),
5881                          lo->ldo_mirrors[i].lme_id,
5882                          lo->ldo_mirrors[primary].lme_id);
5883
5884                 primary = i;
5885         }
5886         if (primary < 0) {
5887                 CERROR(DFID ": doesn't have a primary mirror\n",
5888                        PFID(lod_object_fid(lo)));
5889                 GOTO(out, rc = -ENODATA);
5890         }
5891
5892         CDEBUG(D_LAYOUT, DFID": found primary %u\n",
5893                PFID(lod_object_fid(lo)), lo->ldo_mirrors[primary].lme_id);
5894
5895         LASSERT(!lo->ldo_mirrors[primary].lme_stale);
5896
5897         /* for LAYOUT_WRITE opc, it has to do the following operations:
5898          * 1. stale overlapping componets from stale mirrors;
5899          * 2. instantiate components of the primary mirror;
5900          * 3. transfter layout version to all objects of the primary;
5901          *
5902          * for LAYOUT_RESYNC opc, it will do:
5903          * 1. instantiate components of all stale mirrors;
5904          * 2. transfer layout version to all objects to close write era. */
5905
5906         if (mlc->mlc_opc == MD_LAYOUT_WRITE) {
5907                 LASSERT(mlc->mlc_intent != NULL);
5908
5909                 extent = mlc->mlc_intent->li_extent;
5910
5911                 CDEBUG(D_LAYOUT, DFID": intent to write: "DEXT"\n",
5912                        PFID(lod_object_fid(lo)), PEXT(&extent));
5913
5914                 if (mlc->mlc_intent->li_opc == LAYOUT_INTENT_TRUNC) {
5915                         /**
5916                          * trunc transfers [0, size) in the intent extent, we'd
5917                          * stale components overlapping [size, eof).
5918                          */
5919                         extent.e_start = extent.e_end;
5920                         extent.e_end = OBD_OBJECT_EOF;
5921                 }
5922                 /* 1. stale overlapping components */
5923                 lod_stale_components(lo, primary, &extent);
5924
5925                 /* 2. find out the components need instantiating.
5926                  * instantiate [0, mlc->mlc_intent->e_end) */
5927
5928                 /* restore truncate intent extent */
5929                 if (mlc->mlc_intent->li_opc == LAYOUT_INTENT_TRUNC)
5930                         extent.e_end = extent.e_start;
5931                 extent.e_start = 0;
5932
5933                 lod_foreach_mirror_comp(lod_comp, lo, primary) {
5934                         if (!lu_extent_is_overlapped(&extent,
5935                                                      &lod_comp->llc_extent))
5936                                 break;
5937
5938                         if (lod_comp_inited(lod_comp))
5939                                 continue;
5940
5941                         CDEBUG(D_LAYOUT, "write instantiate %d / %d\n",
5942                                primary, lod_comp_index(lo, lod_comp));
5943                         info->lti_comp_idx[info->lti_count++] =
5944                                                 lod_comp_index(lo, lod_comp);
5945                 }
5946         } else { /* MD_LAYOUT_RESYNC */
5947                 lod_foreach_mirror_comp(lod_comp, lo, primary) {
5948                         if (!lod_comp_inited(lod_comp))
5949                                 break;
5950
5951                         extent.e_end = lod_comp->llc_extent.e_end;
5952                 }
5953
5954                 rc = lod_prepare_resync(env, lo, &extent);
5955                 if (rc)
5956                         GOTO(out, rc);
5957                 /* change the file state to SYNC_PENDING */
5958                 lo->ldo_flr_state = LCM_FL_SYNC_PENDING;
5959         }
5960
5961         rc = lod_declare_instantiate_components(env, lo, th);
5962         if (rc)
5963                 GOTO(out, rc);
5964
5965         /* 3. transfer layout version to OST objects.
5966          * transfer new layout version to OST objects so that stale writes
5967          * can be denied. It also ends an era of writing by setting
5968          * LU_LAYOUT_RESYNC. Normal client can never use this bit to
5969          * send write RPC; only resync RPCs could do it. */
5970         layout_attr->la_valid = LA_LAYOUT_VERSION;
5971         layout_attr->la_layout_version = 0; /* set current version */
5972         if (mlc->mlc_opc == MD_LAYOUT_RESYNC)
5973                 layout_attr->la_layout_version = LU_LAYOUT_RESYNC;
5974         rc = lod_declare_attr_set(env, &lo->ldo_obj, layout_attr, th);
5975         if (rc)
5976                 GOTO(out, rc);
5977
5978         lod_obj_inc_layout_gen(lo);
5979 out:
5980         if (rc)
5981                 lod_object_free_striping(env, lo);
5982         RETURN(rc);
5983 }
5984
5985 static int lod_declare_update_sync_pending(const struct lu_env *env,
5986                 struct lod_object *lo, struct md_layout_change *mlc,
5987                 struct thandle *th)
5988 {
5989         struct lod_thread_info  *info = lod_env_info(env);
5990         unsigned sync_components = 0;
5991         unsigned resync_components = 0;
5992         int i;
5993         int rc;
5994         ENTRY;
5995
5996         LASSERT(lo->ldo_flr_state == LCM_FL_SYNC_PENDING);
5997         LASSERT(mlc->mlc_opc == MD_LAYOUT_RESYNC_DONE ||
5998                 mlc->mlc_opc == MD_LAYOUT_WRITE);
5999
6000         CDEBUG(D_LAYOUT, DFID ": received op %d in sync pending\n",
6001                PFID(lod_object_fid(lo)), mlc->mlc_opc);
6002
6003         if (mlc->mlc_opc == MD_LAYOUT_WRITE) {
6004                 CDEBUG(D_LAYOUT, DFID": cocurrent write to sync pending\n",
6005                        PFID(lod_object_fid(lo)));
6006
6007                 lo->ldo_flr_state = LCM_FL_WRITE_PENDING;
6008                 return lod_declare_update_write_pending(env, lo, mlc, th);
6009         }
6010
6011         /* MD_LAYOUT_RESYNC_DONE */
6012
6013         for (i = 0; i < lo->ldo_comp_cnt; i++) {
6014                 struct lod_layout_component *lod_comp;
6015                 int j;
6016
6017                 lod_comp = &lo->ldo_comp_entries[i];
6018
6019                 if (!(lod_comp->llc_flags & LCME_FL_STALE)) {
6020                         sync_components++;
6021                         continue;
6022                 }
6023
6024                 for (j = 0; j < mlc->mlc_resync_count; j++) {
6025                         if (lod_comp->llc_id != mlc->mlc_resync_ids[j])
6026                                 continue;
6027
6028                         mlc->mlc_resync_ids[j] = LCME_ID_INVAL;
6029                         lod_comp->llc_flags &= ~LCME_FL_STALE;
6030                         resync_components++;
6031                         break;
6032                 }
6033         }
6034
6035         /* valid check */
6036         for (i = 0; i < mlc->mlc_resync_count; i++) {
6037                 if (mlc->mlc_resync_ids[i] == LCME_ID_INVAL)
6038                         continue;
6039
6040                 CDEBUG(D_LAYOUT, DFID": lcme id %u (%d / %zd) not exist "
6041                        "or already synced\n", PFID(lod_object_fid(lo)),
6042                        mlc->mlc_resync_ids[i], i, mlc->mlc_resync_count);
6043                 GOTO(out, rc = -EINVAL);
6044         }
6045
6046         if (!sync_components || (mlc->mlc_resync_count && !resync_components)) {
6047                 CDEBUG(D_LAYOUT, DFID": no mirror in sync\n",
6048                        PFID(lod_object_fid(lo)));
6049
6050                 /* tend to return an error code here to prevent
6051                  * the MDT from setting SoM attribute */
6052                 GOTO(out, rc = -EINVAL);
6053         }
6054
6055         CDEBUG(D_LAYOUT, DFID": resynced %u/%zu components\n",
6056                PFID(lod_object_fid(lo)),
6057                resync_components, mlc->mlc_resync_count);
6058
6059         lo->ldo_flr_state = LCM_FL_RDONLY;
6060         lod_obj_inc_layout_gen(lo);
6061
6062         info->lti_buf.lb_len = lod_comp_md_size(lo, false);
6063         rc = lod_sub_declare_xattr_set(env, lod_object_child(lo),
6064                                        &info->lti_buf, XATTR_NAME_LOV, 0, th);
6065         EXIT;
6066
6067 out:
6068         if (rc)
6069                 lod_object_free_striping(env, lo);
6070         RETURN(rc);
6071 }
6072
6073 static int lod_declare_layout_change(const struct lu_env *env,
6074                 struct dt_object *dt, struct md_layout_change *mlc,
6075                 struct thandle *th)
6076 {
6077         struct lod_thread_info  *info = lod_env_info(env);
6078         struct lod_object *lo = lod_dt_obj(dt);
6079         int rc;
6080         ENTRY;
6081
6082         if (!S_ISREG(dt->do_lu.lo_header->loh_attr) || !dt_object_exists(dt) ||
6083             dt_object_remote(dt_object_child(dt)))
6084                 RETURN(-EINVAL);
6085
6086         lod_write_lock(env, dt, 0);
6087         rc = lod_load_striping_locked(env, lo);
6088         if (rc)
6089                 GOTO(out, rc);
6090
6091         LASSERT(lo->ldo_comp_cnt > 0);
6092
6093         rc = lod_layout_data_init(info, lo->ldo_comp_cnt);
6094         if (rc)
6095                 GOTO(out, rc);
6096
6097         switch (lo->ldo_flr_state) {
6098         case LCM_FL_NONE:
6099                 rc = lod_declare_update_plain(env, lo, mlc->mlc_intent,
6100                                               &mlc->mlc_buf, th);
6101                 break;
6102         case LCM_FL_RDONLY:
6103                 rc = lod_declare_update_rdonly(env, lo, mlc, th);
6104                 break;
6105         case LCM_FL_WRITE_PENDING:
6106                 rc = lod_declare_update_write_pending(env, lo, mlc, th);
6107                 break;
6108         case LCM_FL_SYNC_PENDING:
6109                 rc = lod_declare_update_sync_pending(env, lo, mlc, th);
6110                 break;
6111         default:
6112                 rc = -ENOTSUPP;
6113                 break;
6114         }
6115 out:
6116         dt_write_unlock(env, dt);
6117         RETURN(rc);
6118 }
6119
6120 /**
6121  * Instantiate layout component objects which covers the intent write offset.
6122  */
6123 static int lod_layout_change(const struct lu_env *env, struct dt_object *dt,
6124                              struct md_layout_change *mlc, struct thandle *th)
6125 {
6126         struct lu_attr *attr = &lod_env_info(env)->lti_attr;
6127         struct lu_attr *layout_attr = &lod_env_info(env)->lti_layout_attr;
6128         struct lod_object *lo = lod_dt_obj(dt);
6129         int rc;
6130
6131         rc = lod_striped_create(env, dt, attr, NULL, th);
6132         if (!rc && layout_attr->la_valid & LA_LAYOUT_VERSION) {
6133                 layout_attr->la_layout_version |= lo->ldo_layout_gen;
6134                 rc = lod_attr_set(env, dt, layout_attr, th);
6135         }
6136
6137         return rc;
6138 }
6139
6140 struct dt_object_operations lod_obj_ops = {
6141         .do_read_lock           = lod_read_lock,
6142         .do_write_lock          = lod_write_lock,
6143         .do_read_unlock         = lod_read_unlock,
6144         .do_write_unlock        = lod_write_unlock,
6145         .do_write_locked        = lod_write_locked,
6146         .do_attr_get            = lod_attr_get,
6147         .do_declare_attr_set    = lod_declare_attr_set,
6148         .do_attr_set            = lod_attr_set,
6149         .do_xattr_get           = lod_xattr_get,
6150         .do_declare_xattr_set   = lod_declare_xattr_set,
6151         .do_xattr_set           = lod_xattr_set,
6152         .do_declare_xattr_del   = lod_declare_xattr_del,
6153         .do_xattr_del           = lod_xattr_del,
6154         .do_xattr_list          = lod_xattr_list,
6155         .do_ah_init             = lod_ah_init,
6156         .do_declare_create      = lod_declare_create,
6157         .do_create              = lod_create,
6158         .do_declare_destroy     = lod_declare_destroy,
6159         .do_destroy             = lod_destroy,
6160         .do_index_try           = lod_index_try,
6161         .do_declare_ref_add     = lod_declare_ref_add,
6162         .do_ref_add             = lod_ref_add,
6163         .do_declare_ref_del     = lod_declare_ref_del,
6164         .do_ref_del             = lod_ref_del,
6165         .do_object_sync         = lod_object_sync,
6166         .do_object_lock         = lod_object_lock,
6167         .do_object_unlock       = lod_object_unlock,
6168         .do_invalidate          = lod_invalidate,
6169         .do_declare_layout_change = lod_declare_layout_change,
6170         .do_layout_change       = lod_layout_change,
6171 };
6172
6173 /**
6174  * Implementation of dt_body_operations::dbo_read.
6175  *
6176  * \see dt_body_operations::dbo_read() in the API description for details.
6177  */
6178 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
6179                         struct lu_buf *buf, loff_t *pos)
6180 {
6181         struct dt_object *next = dt_object_child(dt);
6182
6183         LASSERT(S_ISREG(dt->do_lu.lo_header->loh_attr) ||
6184                 S_ISLNK(dt->do_lu.lo_header->loh_attr));
6185         return next->do_body_ops->dbo_read(env, next, buf, pos);
6186 }
6187
6188 /**
6189  * Implementation of dt_body_operations::dbo_declare_write.
6190  *
6191  * \see dt_body_operations::dbo_declare_write() in the API description
6192  * for details.
6193  */
6194 static ssize_t lod_declare_write(const struct lu_env *env,
6195                                  struct dt_object *dt,
6196                                  const struct lu_buf *buf, loff_t pos,
6197                                  struct thandle *th)
6198 {
6199         return lod_sub_declare_write(env, dt_object_child(dt), buf, pos, th);
6200 }
6201
6202 /**
6203  * Implementation of dt_body_operations::dbo_write.
6204  *
6205  * \see dt_body_operations::dbo_write() in the API description for details.
6206  */
6207 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,
6208                          const struct lu_buf *buf, loff_t *pos,
6209                          struct thandle *th, int iq)
6210 {
6211         LASSERT(S_ISREG(dt->do_lu.lo_header->loh_attr) ||
6212                 S_ISLNK(dt->do_lu.lo_header->loh_attr));
6213         return lod_sub_write(env, dt_object_child(dt), buf, pos, th, iq);
6214 }
6215
6216 static int lod_declare_punch(const struct lu_env *env, struct dt_object *dt,
6217                              __u64 start, __u64 end, struct thandle *th)
6218 {
6219         if (dt_object_remote(dt))
6220                 return -ENOTSUPP;
6221
6222         return lod_sub_declare_punch(env, dt_object_child(dt), start, end, th);
6223 }
6224
6225 static int lod_punch(const struct lu_env *env, struct dt_object *dt,
6226                      __u64 start, __u64 end, struct thandle *th)
6227 {
6228         if (dt_object_remote(dt))
6229                 return -ENOTSUPP;
6230
6231         LASSERT(S_ISREG(dt->do_lu.lo_header->loh_attr));
6232         return lod_sub_punch(env, dt_object_child(dt), start, end, th);
6233 }
6234
6235 /*
6236  * different type of files use the same body_ops because object may be created
6237  * in OUT, where there is no chance to set correct body_ops for each type, so
6238  * body_ops themselves will check file type inside, see lod_read/write/punch for
6239  * details.
6240  */
6241 const struct dt_body_operations lod_body_ops = {
6242         .dbo_read               = lod_read,
6243         .dbo_declare_write      = lod_declare_write,
6244         .dbo_write              = lod_write,
6245         .dbo_declare_punch      = lod_declare_punch,
6246         .dbo_punch              = lod_punch,
6247 };
6248
6249 /**
6250  * Implementation of lu_object_operations::loo_object_init.
6251  *
6252  * The function determines the type and the index of the target device using
6253  * sequence of the object's FID. Then passes control down to the
6254  * corresponding device:
6255  *  OSD for the local objects, OSP for remote
6256  *
6257  * \see lu_object_operations::loo_object_init() in the API description
6258  * for details.
6259  */
6260 static int lod_object_init(const struct lu_env *env, struct lu_object *lo,
6261                            const struct lu_object_conf *conf)
6262 {
6263         struct lod_device       *lod    = lu2lod_dev(lo->lo_dev);
6264         struct lu_device        *cdev   = NULL;
6265         struct lu_object        *cobj;
6266         struct lod_tgt_descs    *ltd    = NULL;
6267         struct lod_tgt_desc     *tgt;
6268         u32                      idx    = 0;
6269         int                      type   = LU_SEQ_RANGE_ANY;
6270         int                      rc;
6271         ENTRY;
6272
6273         rc = lod_fld_lookup(env, lod, lu_object_fid(lo), &idx, &type);
6274         if (rc != 0) {
6275                 /* Note: Sometimes, it will Return EAGAIN here, see
6276                  * ptrlpc_import_delay_req(), which might confuse
6277                  * lu_object_find_at() and make it wait there incorrectly.
6278                  * so we convert it to EIO here.*/
6279                 if (rc == -EAGAIN)
6280                         rc = -EIO;
6281
6282                 RETURN(rc);
6283         }
6284
6285         if (type == LU_SEQ_RANGE_MDT &&
6286             idx == lu_site2seq(lo->lo_dev->ld_site)->ss_node_id) {
6287                 cdev = &lod->lod_child->dd_lu_dev;
6288         } else if (type == LU_SEQ_RANGE_MDT) {
6289                 ltd = &lod->lod_mdt_descs;
6290                 lod_getref(ltd);
6291         } else if (type == LU_SEQ_RANGE_OST) {
6292                 ltd = &lod->lod_ost_descs;
6293                 lod_getref(ltd);
6294         } else {
6295                 LBUG();
6296         }
6297
6298         if (ltd != NULL) {
6299                 if (ltd->ltd_tgts_size > idx &&
6300                     cfs_bitmap_check(ltd->ltd_tgt_bitmap, idx)) {
6301                         tgt = LTD_TGT(ltd, idx);
6302
6303                         LASSERT(tgt != NULL);
6304                         LASSERT(tgt->ltd_tgt != NULL);
6305
6306                         cdev = &(tgt->ltd_tgt->dd_lu_dev);
6307                 }
6308                 lod_putref(lod, ltd);
6309         }
6310
6311         if (unlikely(cdev == NULL))
6312                 RETURN(-ENOENT);
6313
6314         cobj = cdev->ld_ops->ldo_object_alloc(env, lo->lo_header, cdev);
6315         if (unlikely(cobj == NULL))
6316                 RETURN(-ENOMEM);
6317
6318         lu2lod_obj(lo)->ldo_obj.do_body_ops = &lod_body_ops;
6319
6320         lu_object_add(lo, cobj);
6321
6322         RETURN(0);
6323 }
6324
6325 /**
6326  *
6327  * Release resources associated with striping.
6328  *
6329  * If the object is striped (regular or directory), then release
6330  * the stripe objects references and free the ldo_stripe array.
6331  *
6332  * \param[in] env       execution environment
6333  * \param[in] lo        object
6334  */
6335 void lod_object_free_striping(const struct lu_env *env, struct lod_object *lo)
6336 {
6337         struct lod_layout_component *lod_comp;
6338         int i, j;
6339
6340         if (lo->ldo_stripe != NULL) {
6341                 LASSERT(lo->ldo_comp_entries == NULL);
6342                 LASSERT(lo->ldo_dir_stripes_allocated > 0);
6343
6344                 for (i = 0; i < lo->ldo_dir_stripe_count; i++) {
6345                         if (lo->ldo_stripe[i])
6346                                 dt_object_put(env, lo->ldo_stripe[i]);
6347                 }
6348
6349                 j = sizeof(struct dt_object *) * lo->ldo_dir_stripes_allocated;
6350                 OBD_FREE(lo->ldo_stripe, j);
6351                 lo->ldo_stripe = NULL;
6352                 lo->ldo_dir_stripes_allocated = 0;
6353                 lo->ldo_dir_stripe_loaded = 0;
6354                 lo->ldo_dir_stripe_count = 0;
6355         } else if (lo->ldo_comp_entries != NULL) {
6356                 for (i = 0; i < lo->ldo_comp_cnt; i++) {
6357                         /* free lod_layout_component::llc_stripe array */
6358                         lod_comp = &lo->ldo_comp_entries[i];
6359
6360                         if (lod_comp->llc_stripe == NULL)
6361                                 continue;
6362                         LASSERT(lod_comp->llc_stripes_allocated != 0);
6363                         for (j = 0; j < lod_comp->llc_stripes_allocated; j++) {
6364                                 if (lod_comp->llc_stripe[j] != NULL)
6365                                         lu_object_put(env,
6366                                                &lod_comp->llc_stripe[j]->do_lu);
6367                         }
6368                         OBD_FREE(lod_comp->llc_stripe,
6369                                  sizeof(struct dt_object *) *
6370                                  lod_comp->llc_stripes_allocated);
6371                         lod_comp->llc_stripe = NULL;
6372                         lod_comp->llc_stripes_allocated = 0;
6373                 }
6374                 lod_free_comp_entries(lo);
6375                 lo->ldo_comp_cached = 0;
6376         }
6377 }
6378
6379 /**
6380  * Implementation of lu_object_operations::loo_object_free.
6381  *
6382  * \see lu_object_operations::loo_object_free() in the API description
6383  * for details.
6384  */
6385 static void lod_object_free(const struct lu_env *env, struct lu_object *o)
6386 {
6387         struct lod_object *lo = lu2lod_obj(o);
6388
6389         /* release all underlying object pinned */
6390         lod_object_free_striping(env, lo);
6391         lu_object_fini(o);
6392         OBD_SLAB_FREE_PTR(lo, lod_object_kmem);
6393 }
6394
6395 /**
6396  * Implementation of lu_object_operations::loo_object_release.
6397  *
6398  * \see lu_object_operations::loo_object_release() in the API description
6399  * for details.
6400  */
6401 static void lod_object_release(const struct lu_env *env, struct lu_object *o)
6402 {
6403         /* XXX: shouldn't we release everything here in case if object
6404          * creation failed before? */
6405 }
6406
6407 /**
6408  * Implementation of lu_object_operations::loo_object_print.
6409  *
6410  * \see lu_object_operations::loo_object_print() in the API description
6411  * for details.
6412  */
6413 static int lod_object_print(const struct lu_env *env, void *cookie,
6414                             lu_printer_t p, const struct lu_object *l)
6415 {
6416         struct lod_object *o = lu2lod_obj((struct lu_object *) l);
6417
6418         return (*p)(env, cookie, LUSTRE_LOD_NAME"-object@%p", o);
6419 }
6420
6421 struct lu_object_operations lod_lu_obj_ops = {
6422         .loo_object_init        = lod_object_init,
6423         .loo_object_free        = lod_object_free,
6424         .loo_object_release     = lod_object_release,
6425         .loo_object_print       = lod_object_print,
6426 };