lustre/mdd/mdd_object.c

   1 // SPDX-License-Identifier: GPL-2.0
   2
   3 /*
   4  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
   5  * Use is subject to license terms.
   6  *
   7  * Copyright (c) 2011, 2017, Intel Corporation.
   8  */
   9
  10 /*
  11  * This file is part of Lustre, http://www.lustre.org/
  12  *
  13  * Lustre Metadata Server (mdd) routines
  14  *
  15  * Author: Wang Di <wangdi@clusterfs.com>
  16  */
  17
  18 #define DEBUG_SUBSYSTEM S_MDS
  19
  20 #include <linux/module.h>
  21 #include <obd.h>
  22 #include <obd_class.h>
  23 #include <obd_support.h>
  24 #include <lprocfs_status.h>
  25 /* fid_be_cpu(), fid_cpu_to_be(). */
  26 #include <lustre_fid.h>
  27 #include <lustre_idmap.h>
  28 #include <uapi/linux/lustre/lustre_param.h>
  29 #include <lustre_mds.h>
  30
  31 #include "mdd_internal.h"
  32
  33 static const struct lu_object_operations mdd_lu_obj_ops;
  34
  35 struct mdd_object_user {
  36         struct list_head        mou_list;       /* linked off mod_users */
  37         enum mds_open_flags     mou_open_flags; /* open mode by client */
  38         __u64                   mou_uidgid;     /* uid_gid on client */
  39         int                     mou_opencount;  /* # opened */
  40         /* time of next access denied notificaiton */
  41         ktime_t                 mou_deniednext;
  42 };
  43
  44 static int mdd_xattr_get(const struct lu_env *env,
  45                          struct md_object *obj, struct lu_buf *buf,
  46                          const char *name);
  47
  48 static int mdd_changelog_data_store_by_fid(const struct lu_env *env,
  49                                            struct mdd_device *mdd,
  50                                            enum changelog_rec_type type,
  51                                            enum changelog_rec_flags clf_flags,
  52                                            const struct lu_fid *fid,
  53                                            const struct lu_fid *pfid,
  54                                            const char *xattr_name,
  55                                            struct thandle *handle);
  56
  57 static inline bool has_prefix(const char *str, const char *prefix);
  58
  59 static u32 mdd_open_flags_to_mode(enum mds_open_flags open_flags)
  60 {
  61         u32 open_mode = 0;
  62
  63         if (open_flags & MDS_FMODE_EXEC) {
  64                 open_mode = MDS_FMODE_EXEC;
  65         } else {
  66                 if (open_flags & MDS_FMODE_READ)
  67                         open_mode = MDS_FMODE_READ;
  68                 if (open_flags &
  69                     (MDS_FMODE_WRITE | MDS_OPEN_TRUNC | MDS_OPEN_APPEND))
  70                         open_mode |= MDS_FMODE_WRITE;
  71         }
  72
  73         return open_mode;
  74 }
  75
  76 /**
  77  * mdd_obj_user_alloc() - Allocate/init a user and its sub-structures.
  78  *
  79  * @flags: open flags passed from client
  80  * @uid: client uid
  81  * @gid: client gid
  82  *
  83  * Return:
  84  * * populate mdd_object_user with valid struct on success
  85  */
  86 static struct mdd_object_user *mdd_obj_user_alloc(enum mds_open_flags o_flags,
  87                                                   uid_t uid, gid_t gid)
  88 {
  89         struct mdd_object_user *mou;
  90         enum mds_open_flags open_flags = o_flags;
  91
  92         ENTRY;
  93
  94         OBD_SLAB_ALLOC_PTR(mou, mdd_object_kmem);
  95         if (mou == NULL)
  96                 RETURN(ERR_PTR(-ENOMEM));
  97
  98         mou->mou_open_flags = open_flags;
  99         mou->mou_uidgid = ((__u64)uid << 32) | gid;
 100         mou->mou_opencount = 0;
 101         mou->mou_deniednext = ktime_set(0, 0);
 102
 103         RETURN(mou);
 104 }
 105
 106 /**
 107  * Free a user and its sub-structures.
 108  *
 109  * \param mou [IN]  user to be freed.
 110  */
 111 static void mdd_obj_user_free(struct mdd_object_user *mou)
 112 {
 113         OBD_SLAB_FREE_PTR(mou, mdd_object_kmem);
 114 }
 115
 116 /**
 117  * mdd_obj_user_find() - Find if UID/GID already has this file open
 118  *
 119  * @mdd_obj: Metadata server side object
 120  * @uid: Client UID
 121  * @gid: Client GID
 122  * @open_flags: MDS flags passed from client
 123  *
 124  * Caller should have write-locked on param @mdd_obj.
 125  *
 126  * Return: mdd_object_user pointer or NULL if not found
 127  */
 128 static
 129 struct mdd_object_user *mdd_obj_user_find(struct mdd_object *mdd_obj,
 130                                           uid_t uid, gid_t gid,
 131                                           enum mds_open_flags open_flags)
 132 {
 133         struct mdd_object_user *mou;
 134         __u64 uidgid;
 135
 136         ENTRY;
 137
 138         uidgid = ((__u64)uid << 32) | gid;
 139         list_for_each_entry(mou, &mdd_obj->mod_users, mou_list) {
 140                 if (mou->mou_uidgid == uidgid &&
 141                     mdd_open_flags_to_mode(mou->mou_open_flags) ==
 142                     mdd_open_flags_to_mode(open_flags))
 143                         RETURN(mou);
 144         }
 145         RETURN(NULL);
 146 }
 147
 148 /**
 149  * Add a user to the list of openers for this file
 150  *
 151  * Caller should have write-locked \param mdd_obj.
 152  * \param mdd_obj [IN] mdd_obj
 153  * \param mou [IN] user
 154  * \retval 0 success
 155  * \retval -ve failure
 156  */
 157 static int mdd_obj_user_add(struct mdd_object *mdd_obj,
 158                             struct mdd_object_user *mou,
 159                             bool denied)
 160 {
 161         struct mdd_device *mdd = mdd_obj2mdd_dev(mdd_obj);
 162         struct mdd_object_user *tmp;
 163         __u32 uid = mou->mou_uidgid >> 32;
 164         __u32 gid = mou->mou_uidgid & ((1UL << 32) - 1);
 165
 166         ENTRY;
 167         tmp = mdd_obj_user_find(mdd_obj, uid, gid, mou->mou_open_flags);
 168         if (tmp != NULL)
 169                 RETURN(-EEXIST);
 170
 171         list_add_tail(&mou->mou_list, &mdd_obj->mod_users);
 172
 173         if (denied)
 174                 /* next 'access denied' notification cannot happen before
 175                  * mou_deniednext
 176                  */
 177                 mou->mou_deniednext =
 178                         ktime_add(ktime_get(),
 179                                   ktime_set(mdd->mdd_cl.mc_deniednext, 0));
 180         else
 181                 mou->mou_opencount++;
 182
 183         RETURN(0);
 184 }
 185
 186 /**
 187  * Remove UID from the list
 188  *
 189  * Caller should have write-locked \param mdd_obj.
 190  * \param mdd_obj [IN] mdd_obj
 191  * \param uid [IN] user
 192  * \retval -ve failure
 193  */
 194 static int mdd_obj_user_remove(struct mdd_object *mdd_obj,
 195                                struct mdd_object_user *mou)
 196 {
 197         ENTRY;
 198
 199         if (mou == NULL)
 200                 RETURN(-ENOENT);
 201
 202         list_del_init(&mou->mou_list);
 203
 204         mdd_obj_user_free(mou);
 205
 206         RETURN(0);
 207 }
 208
 209 int mdd_la_get(const struct lu_env *env, struct mdd_object *obj,
 210                struct lu_attr *la)
 211 {
 212         int rc;
 213
 214         if (mdd_object_exists(obj) == 0)
 215                 return -ENOENT;
 216
 217         rc = mdo_attr_get(env, obj, la);
 218         if (unlikely(rc != 0)) {
 219                 if (rc == -ENOENT)
 220                         obj->mod_flags |= DEAD_OBJ;
 221                 return rc;
 222         }
 223
 224         if (la->la_valid & LA_FLAGS && la->la_flags & LUSTRE_ORPHAN_FL)
 225                 obj->mod_flags |= ORPHAN_OBJ | DEAD_OBJ;
 226
 227         return 0;
 228 }
 229
 230 struct mdd_thread_info *mdd_env_info(const struct lu_env *env)
 231 {
 232         return lu_env_info(env, &mdd_thread_key);
 233 }
 234
 235 struct lu_buf *mdd_buf_get(const struct lu_env *env, void *area, ssize_t len)
 236 {
 237         struct lu_buf *buf;
 238
 239         buf = &mdd_env_info(env)->mdi_buf[0];
 240         buf->lb_buf = area;
 241         buf->lb_len = len;
 242         return buf;
 243 }
 244
 245 const struct lu_buf *mdd_buf_get_const(const struct lu_env *env,
 246                                        const void *area, ssize_t len)
 247 {
 248         struct lu_buf *buf;
 249
 250         buf = &mdd_env_info(env)->mdi_buf[0];
 251         buf->lb_buf = (void *)area;
 252         buf->lb_len = len;
 253         return buf;
 254 }
 255
 256 struct lu_object *mdd_object_alloc(const struct lu_env *env,
 257                                    const struct lu_object_header *hdr,
 258                                    struct lu_device *d)
 259 {
 260         struct mdd_device *mdd = lu2mdd_dev(d);
 261         struct mdd_object *mdd_obj;
 262         struct lu_object *o;
 263
 264         OBD_SLAB_ALLOC_PTR_GFP(mdd_obj, mdd_object_kmem, GFP_NOFS);
 265         if (!mdd_obj)
 266                 return NULL;
 267
 268         o = mdd2lu_obj(mdd_obj);
 269         lu_object_init(o, NULL, d);
 270         mdd_obj->mod_obj.mo_ops = &mdd_obj_ops;
 271         mdd_obj->mod_obj.mo_dir_ops = &mdd_dir_ops;
 272         mdd_obj->mod_count = 0;
 273         obd_obt_init(mdd2obd_dev(mdd));
 274         o->lo_ops = &mdd_lu_obj_ops;
 275         INIT_LIST_HEAD(&mdd_obj->mod_users);
 276
 277         return o;
 278 }
 279
 280 static int mdd_object_init(const struct lu_env *env, struct lu_object *o,
 281                            const struct lu_object_conf *unused)
 282 {
 283         struct mdd_device *d = lu2mdd_dev(o->lo_dev);
 284         struct mdd_object *mdd_obj = lu2mdd_obj(o);
 285         struct lu_object  *below;
 286         struct lu_device  *under;
 287
 288         ENTRY;
 289
 290         mdd_obj->mod_cltime = ktime_set(0, 0);
 291         under = &d->mdd_child->dd_lu_dev;
 292         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
 293         if (IS_ERR(below))
 294                 RETURN(PTR_ERR(below));
 295
 296         lu_object_add(o, below);
 297
 298         RETURN(0);
 299 }
 300
 301 static int mdd_object_start(const struct lu_env *env, struct lu_object *o)
 302 {
 303         int rc = 0;
 304
 305         if (lu_object_exists(o)) {
 306                 struct mdd_object *mdd_obj = lu2mdd_obj(o);
 307                 struct lu_attr *attr = MDD_ENV_VAR(env, la_for_start);
 308
 309                 rc = mdd_la_get(env, mdd_obj, attr);
 310         }
 311
 312         return rc;
 313 }
 314
 315 static void mdd_object_free(const struct lu_env *env, struct lu_object *o)
 316 {
 317         struct mdd_object *mdd = lu2mdd_obj(o);
 318         struct mdd_object_user *mou, *tmp2;
 319
 320         /* free user list */
 321         list_for_each_entry_safe(mou, tmp2, &mdd->mod_users, mou_list) {
 322                 list_del(&mou->mou_list);
 323                 mdd_obj_user_free(mou);
 324         }
 325
 326         lu_object_fini(o);
 327         /* mdd doesn't contain an lu_object_header, so don't need call_rcu */
 328         OBD_SLAB_FREE_PTR(mdd, mdd_object_kmem);
 329 }
 330
 331 static int mdd_object_print(const struct lu_env *env, void *cookie,
 332                             lu_printer_t p, const struct lu_object *o)
 333 {
 334         struct mdd_object *mdd = lu2mdd_obj((struct lu_object *)o);
 335
 336         return (*p)(env, cookie,
 337                     LUSTRE_MDD_NAME"-object@%p(open_count=%d, valid=%x, cltime=%lldns, flags=%lx)",
 338                     mdd, mdd->mod_count, mdd->mod_valid,
 339                     ktime_to_ns(mdd->mod_cltime), mdd->mod_flags);
 340 }
 341
 342 static const struct lu_object_operations mdd_lu_obj_ops = {
 343         .loo_object_init    = mdd_object_init,
 344         .loo_object_start   = mdd_object_start,
 345         .loo_object_free    = mdd_object_free,
 346         .loo_object_print   = mdd_object_print,
 347 };
 348
 349 struct mdd_object *mdd_object_find(const struct lu_env *env,
 350                                    struct mdd_device *d,
 351                                    const struct lu_fid *f)
 352 {
 353         return md2mdd_obj(md_object_find_slice(env, &d->mdd_md_dev, f));
 354 }
 355
 356 /*
 357  * No permission check is needed.
 358  */
 359 int mdd_attr_get(const struct lu_env *env, struct md_object *obj,
 360                  struct md_attr *ma)
 361 {
 362         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 363         int               rc;
 364
 365         ENTRY;
 366
 367         rc = mdd_la_get(env, mdd_obj, &ma->ma_attr);
 368         if ((ma->ma_need & MA_INODE) != 0 && mdd_is_dead_obj(mdd_obj))
 369                 ma->ma_attr.la_nlink = 0;
 370
 371         RETURN(rc);
 372 }
 373
 374 /*
 375  * No permission check is needed.
 376  */
 377 static int mdd_xattr_get(const struct lu_env *env,
 378                          struct md_object *obj, struct lu_buf *buf,
 379                          const char *name)
 380 {
 381         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 382         struct mdd_device *mdd;
 383         int rc;
 384
 385         ENTRY;
 386
 387         if (mdd_object_exists(mdd_obj) == 0) {
 388                 rc = -ENOENT;
 389                 CERROR("%s: object "DFID" not found: rc = %d\n",
 390                        mdd_obj_dev_name(mdd_obj),
 391                        PFID(mdd_object_fid(mdd_obj)), rc);
 392                 return rc;
 393         }
 394
 395         /* If the object has been destroyed, then do not get LMVEA, because
 396          * it needs to load stripes from the iteration of the master object,
 397          * and it will cause problem if master object has been destroyed, see
 398          * LU-6427
 399          */
 400         if (unlikely((mdd_obj->mod_flags & DEAD_OBJ) &&
 401                      !(mdd_obj->mod_flags & ORPHAN_OBJ) &&
 402                       strcmp(name, XATTR_NAME_LMV) == 0))
 403                 RETURN(-ENOENT);
 404
 405         /* If the object has been delete from the namespace, then
 406          * get linkEA should return -ENOENT as well
 407          */
 408         if (unlikely((mdd_obj->mod_flags & (DEAD_OBJ | ORPHAN_OBJ)) &&
 409                       strcmp(name, XATTR_NAME_LINK) == 0))
 410                 RETURN(-ENOENT);
 411
 412         mdd_read_lock(env, mdd_obj, DT_TGT_CHILD);
 413         rc = mdo_xattr_get(env, mdd_obj, buf, name);
 414         mdd_read_unlock(env, mdd_obj);
 415
 416         mdd = mdo2mdd(obj);
 417
 418         /* record only getting user xattrs and acls */
 419         if (rc >= 0 && buf->lb_buf &&
 420             mdd_changelog_enabled(env, mdd, CL_GETXATTR) &&
 421             (has_prefix(name, XATTR_USER_PREFIX) ||
 422              has_prefix(name, XATTR_NAME_POSIX_ACL_ACCESS) ||
 423              has_prefix(name, XATTR_NAME_POSIX_ACL_DEFAULT))) {
 424                 struct thandle *handle;
 425                 int rc2;
 426
 427                 LASSERT(mdd_object_fid(mdd_obj) != NULL);
 428
 429                 handle = mdd_trans_create(env, mdd);
 430                 if (IS_ERR(handle))
 431                         RETURN(PTR_ERR(handle));
 432
 433                 rc2 = mdd_declare_changelog_store(env, mdd, CL_GETXATTR, NULL,
 434                                                   NULL, handle);
 435                 if (rc2)
 436                         GOTO(stop, rc2);
 437
 438                 rc2 = mdd_trans_start(env, mdd, handle);
 439                 if (rc2)
 440                         GOTO(stop, rc2);
 441
 442                 rc2 = mdd_changelog_data_store_by_fid(env, mdd, CL_GETXATTR, 0,
 443                                                       mdd_object_fid(mdd_obj),
 444                                                       NULL, name, handle);
 445
 446 stop:
 447                 rc2 = mdd_trans_stop(env, mdd, rc2, handle);
 448                 if (rc2)
 449                         rc = rc2;
 450         }
 451
 452         RETURN(rc);
 453 }
 454
 455 /* Permission check is done when open, no need check again. */
 456 int mdd_readlink(const struct lu_env *env, struct md_object *obj,
 457                  struct lu_buf *buf)
 458 {
 459         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 460         struct dt_object *next;
 461         loff_t pos = 0;
 462         int rc;
 463
 464         ENTRY;
 465
 466         if (mdd_object_exists(mdd_obj) == 0) {
 467                 rc = -ENOENT;
 468                 CERROR("%s: object "DFID" not found: rc = %d\n",
 469                        mdd_obj_dev_name(mdd_obj),
 470                        PFID(mdd_object_fid(mdd_obj)), rc);
 471                 return rc;
 472         }
 473
 474         next = mdd_object_child(mdd_obj);
 475         LASSERT(next != NULL);
 476         LASSERT(next->do_body_ops != NULL);
 477         LASSERT(next->do_body_ops->dbo_read != NULL);
 478         mdd_read_lock(env, mdd_obj, DT_TGT_CHILD);
 479         rc = dt_read(env, next, buf, &pos);
 480         mdd_read_unlock(env, mdd_obj);
 481         RETURN(rc);
 482 }
 483
 484 /* No permission check is needed.  */
 485 static int mdd_xattr_list(const struct lu_env *env, struct md_object *obj,
 486                           struct lu_buf *buf)
 487 {
 488         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 489         int rc;
 490
 491         ENTRY;
 492
 493         mdd_read_lock(env, mdd_obj, DT_TGT_CHILD);
 494         rc = mdo_xattr_list(env, mdd_obj, buf);
 495         mdd_read_unlock(env, mdd_obj);
 496
 497         /* If the buffer is NULL then we are only here to get the
 498          * length of the xattr name list.
 499          */
 500         if (rc < 0 || buf->lb_buf == NULL)
 501                 RETURN(rc);
 502
 503         /*
 504          * Filter out XATTR_NAME_LINK if this is an orphan object.  See
 505          * mdd_xattr_get().
 506          */
 507         if (unlikely(mdd_obj->mod_flags & (DEAD_OBJ | ORPHAN_OBJ))) {
 508                 char   *end = (char *)buf->lb_buf + rc;
 509                 char   *p = buf->lb_buf;
 510
 511                 while (p < end) {
 512                         char   *next = p + strlen(p) + 1;
 513
 514                         if (strcmp(p, XATTR_NAME_LINK) == 0) {
 515                                 if (end - next > 0)
 516                                         memmove(p, next, end - next);
 517                                 rc -= next - p;
 518                                 CDEBUG(D_INFO, "Filtered out "XATTR_NAME_LINK
 519                                        " of orphan "DFID"\n",
 520                                        PFID(mdd_object_fid(mdd_obj)));
 521                                 break;
 522                         }
 523
 524                         p = next;
 525                 }
 526         }
 527
 528         RETURN(rc);
 529 }
 530
 531 int mdd_invalidate(const struct lu_env *env, struct md_object *obj)
 532 {
 533         return mdo_invalidate(env, md2mdd_obj(obj));
 534 }
 535
 536 int mdd_declare_create_object_internal(const struct lu_env *env,
 537                                        struct mdd_object *p,
 538                                        struct mdd_object *c,
 539                                        struct lu_attr *attr,
 540                                        struct thandle *handle,
 541                                        const struct md_op_spec *spec,
 542                                        struct dt_allocation_hint *hint)
 543 {
 544         struct dt_object_format *dof = &mdd_env_info(env)->mdi_dof;
 545         const struct dt_index_features *feat = spec->sp_feat;
 546         int rc;
 547
 548         ENTRY;
 549
 550         if (feat != &dt_directory_features && feat != NULL) {
 551                 dof->dof_type = DFT_INDEX;
 552                 dof->u.dof_idx.di_feat = feat;
 553         } else {
 554                 dof->dof_type = dt_mode_to_dft(attr->la_mode);
 555                 if (dof->dof_type == DFT_REGULAR) {
 556                         dof->u.dof_reg.striped =
 557                                 md_should_create(spec->sp_cr_flags);
 558                         if (spec->sp_cr_flags & MDS_OPEN_HAS_EA)
 559                                 dof->u.dof_reg.striped = 0;
 560                         /* is this replay? */
 561                         if (spec->no_create)
 562                                 dof->u.dof_reg.striped = 0;
 563                 }
 564         }
 565
 566         rc = mdo_declare_create_object(env, c, attr, hint, dof, handle);
 567
 568         RETURN(rc);
 569 }
 570
 571 int mdd_create_object_internal(const struct lu_env *env, struct mdd_object *p,
 572                                struct mdd_object *c, struct lu_attr *attr,
 573                                struct thandle *handle,
 574                                const struct md_op_spec *spec,
 575                                struct dt_allocation_hint *hint)
 576 {
 577         struct dt_object_format *dof = &mdd_env_info(env)->mdi_dof;
 578         int rc;
 579
 580         ENTRY;
 581
 582         LASSERT(!mdd_object_exists(c));
 583
 584         rc = mdo_create_object(env, c, attr, hint, dof, handle);
 585
 586         RETURN(rc);
 587 }
 588
 589 int mdd_attr_set_internal(const struct lu_env *env, struct mdd_object *obj,
 590                           const struct lu_attr *attr, struct thandle *handle,
 591                           int needacl)
 592 {
 593         int rc;
 594
 595         ENTRY;
 596
 597         rc = mdo_attr_set(env, obj, attr, handle);
 598 #ifdef CONFIG_LUSTRE_FS_POSIX_ACL
 599         if (!rc && (attr->la_valid & LA_MODE) && needacl)
 600                 rc = mdd_acl_chmod(env, obj, attr->la_mode, handle);
 601 #endif
 602         RETURN(rc);
 603 }
 604
 605 int mdd_update_time(const struct lu_env *env, struct mdd_object *obj,
 606                     const struct lu_attr *oattr, struct lu_attr *attr,
 607                     struct thandle *handle)
 608 {
 609         int rc = 0;
 610
 611         ENTRY;
 612
 613         LASSERT(attr->la_valid & LA_CTIME);
 614         LASSERT(oattr != NULL);
 615
 616         /* Make sure the ctime is increased only, however, it's not strictly
 617          * reliable at here because there is not guarantee to hold lock on
 618          * object, so we just bypass some unnecessary cmtime setting first
 619          * and OSD has to check it again.
 620          */
 621         if (attr->la_ctime < oattr->la_ctime)
 622                 attr->la_valid &= ~(LA_MTIME | LA_CTIME);
 623         else if (attr->la_valid == LA_CTIME &&
 624                  attr->la_ctime == oattr->la_ctime)
 625                 attr->la_valid &= ~LA_CTIME;
 626
 627         if (attr->la_valid != 0)
 628                 rc = mdd_attr_set_internal(env, obj, attr, handle, 0);
 629         RETURN(rc);
 630 }
 631
 632
 633 static bool is_project_state_change(const struct lu_attr *oattr,
 634                                     struct lu_attr *la)
 635 {
 636         if (la->la_valid & LA_PROJID &&
 637             oattr->la_projid != la->la_projid)
 638                 return true;
 639
 640         if ((la->la_valid & LA_FLAGS) &&
 641             (la->la_flags & LUSTRE_PROJINHERIT_FL) !=
 642             (oattr->la_flags & LUSTRE_PROJINHERIT_FL))
 643                 return true;
 644
 645         return false;
 646 }
 647
 648 /**
 649  * This gives the same functionality as the code between
 650  * sys_chmod and inode_setattr
 651  * chown_common and inode_setattr
 652  * utimes and inode_setattr
 653  * This API is ported from mds_fix_attr but remove some unnecesssary stuff.
 654  *
 655  * @param[in]     oattr  Original, current object attributes.
 656  * @param[in,out] la     New attributes to be evaluated and potentially
 657  *                       modified depending on oattr content and other rules.
 658  * @param[in]     ma     md_attr to be evaluated for that object.
 659  */
 660 static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
 661                         const struct lu_attr *oattr, struct lu_attr *la,
 662                         const struct md_attr *ma)
 663 {
 664         struct lu_ucred  *uc;
 665         int               rc = 0;
 666         const unsigned long flags = ma->ma_attr_flags;
 667
 668         ENTRY;
 669
 670         if (!la->la_valid)
 671                 RETURN(0);
 672
 673         /* Do not permit change file type */
 674         if (la->la_valid & LA_TYPE)
 675                 RETURN(-EPERM);
 676
 677         /* They should not be processed by setattr */
 678         if (la->la_valid & (LA_NLINK | LA_RDEV | LA_BLKSIZE))
 679                 RETURN(-EPERM);
 680
 681         LASSERT(oattr != NULL);
 682
 683         uc = lu_ucred_check(env);
 684         if (uc == NULL)
 685                 RETURN(0);
 686
 687         if (is_project_state_change(oattr, la)) {
 688                 /* we want rbac roles to have precedence over any other
 689                  * permission or capability checks
 690                  */
 691                 if (!uc->uc_rbac_file_perms ||
 692                     (!cap_raised(uc->uc_cap, CAP_SYS_RESOURCE) &&
 693                      !lustre_in_group_p(uc, ma->ma_enable_chprojid_gid) &&
 694                      !(ma->ma_enable_chprojid_gid == -1 &&
 695                        mdd_permission_internal(env, obj, oattr, MAY_WRITE))))
 696                         RETURN(-EPERM);
 697         }
 698
 699         if (la->la_valid == LA_CTIME) {
 700                 if (!(flags & MDS_PERM_BYPASS))
 701                         /* This is only for set ctime when rename's source is
 702                          * on remote MDS.
 703                          */
 704                         rc = mdd_may_delete(env, NULL, NULL, obj, oattr, NULL,
 705                                             1, 0);
 706                 if (rc == 0 && la->la_ctime <= oattr->la_ctime)
 707                         la->la_valid &= ~LA_CTIME;
 708                 RETURN(rc);
 709         }
 710
 711         if (flags & MDS_CLOSE_UPDATE_TIMES) {
 712                 /* This is an atime/mtime/ctime update for close RPCs. */
 713                 if (!(la->la_valid & LA_CTIME) ||
 714                     (la->la_ctime <= oattr->la_ctime))
 715                         la->la_valid &= ~(LA_MTIME | LA_CTIME);
 716
 717                 if (la->la_valid & LA_ATIME &&
 718                     (la->la_atime <= obj->mod_atime_set ||
 719                      la->la_atime <= (oattr->la_atime +
 720                                       mdd_obj2mdd_dev(obj)->mdd_atime_diff)))
 721                         la->la_valid &= ~LA_ATIME;
 722                 else
 723                         obj->mod_atime_set = la->la_atime;
 724                 if (la->la_valid & LA_MTIME && la->la_mtime <= oattr->la_mtime)
 725                         la->la_valid &= ~LA_MTIME;
 726                 RETURN(0);
 727         } else if ((la->la_valid & LA_ATIME) && (la->la_valid & LA_CTIME)) {
 728                 /* save the time when atime was changed, in case this is
 729                  * set-in-past, to not lose it later on close. */
 730                 obj->mod_atime_set = la->la_ctime;
 731         }
 732
 733         /* Check if flags change. */
 734         if (la->la_valid & LA_FLAGS) {
 735                 unsigned int oldflags = oattr->la_flags &
 736                                 (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL);
 737                 unsigned int newflags = la->la_flags &
 738                                 (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL);
 739
 740                 if ((uc->uc_fsuid != oattr->la_uid) &&
 741                     !cap_raised(uc->uc_cap, CAP_FOWNER))
 742                         RETURN(-EPERM);
 743
 744                 /* The IMMUTABLE and APPEND_ONLY flags can
 745                  * only be changed by the relevant capability.
 746                  */
 747                 if ((oldflags ^ newflags) &&
 748                     !cap_raised(uc->uc_cap, CAP_LINUX_IMMUTABLE))
 749                         RETURN(-EPERM);
 750
 751                 if (!S_ISDIR(oattr->la_mode)) {
 752                         la->la_flags &= ~(LUSTRE_DIRSYNC_FL | LUSTRE_TOPDIR_FL);
 753                 } else if (la->la_flags & LUSTRE_ENCRYPT_FL) {
 754                         /* when trying to add encryption flag on dir,
 755                          * make sure it is empty
 756                          */
 757                         rc = mdd_dir_is_empty(env, obj);
 758                         if (rc)
 759                                 RETURN(rc);
 760                         rc = 0;
 761                 }
 762         }
 763
 764         if (oattr->la_flags & (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL) &&
 765             (la->la_valid & ~LA_FLAGS) &&
 766             !(flags & MDS_PERM_BYPASS))
 767                 RETURN(-EPERM);
 768
 769         /* Check for setting the obj time. */
 770         if ((la->la_valid & (LA_MTIME | LA_ATIME | LA_CTIME)) &&
 771             !(la->la_valid & ~(LA_MTIME | LA_ATIME | LA_CTIME))) {
 772                 if ((uc->uc_fsuid != oattr->la_uid) &&
 773                     !cap_raised(uc->uc_cap, CAP_FOWNER)) {
 774                         rc = mdd_permission_internal(env, obj, oattr,
 775                                                      MAY_WRITE);
 776                         if (rc)
 777                                 RETURN(rc);
 778                 }
 779         }
 780
 781         if (la->la_valid & LA_KILL_SUID) {
 782                 la->la_valid &= ~LA_KILL_SUID;
 783                 if ((oattr->la_mode & S_ISUID) &&
 784                     !(la->la_valid & LA_MODE)) {
 785                         la->la_mode = oattr->la_mode;
 786                         la->la_valid |= LA_MODE;
 787                 }
 788                 la->la_mode &= ~S_ISUID;
 789         }
 790
 791         if (la->la_valid & LA_KILL_SGID) {
 792                 la->la_valid &= ~LA_KILL_SGID;
 793                 if (((oattr->la_mode & (S_ISGID | 0010)) ==
 794                         (S_ISGID | 0010)) &&
 795                     !(la->la_valid & LA_MODE)) {
 796                         la->la_mode = oattr->la_mode;
 797                         la->la_valid |= LA_MODE;
 798                 }
 799                 la->la_mode &= ~S_ISGID;
 800         }
 801
 802         /* Make sure a caller can chmod. */
 803         if (la->la_valid & LA_MODE) {
 804                 /* we want rbac roles to have precedence over any other
 805                  * permission or capability checks
 806                  */
 807                 if (!uc->uc_rbac_file_perms ||
 808                     (!(flags & MDS_PERM_BYPASS) &&
 809                      (uc->uc_fsuid != oattr->la_uid) &&
 810                      !cap_raised(uc->uc_cap, CAP_FOWNER)))
 811                         RETURN(-EPERM);
 812
 813                 if (la->la_mode == (umode_t) -1)
 814                         la->la_mode = oattr->la_mode;
 815                 else
 816                         la->la_mode = (la->la_mode & S_IALLUGO) |
 817                                         (oattr->la_mode & ~S_IALLUGO);
 818
 819                 /* Also check the setgid bit! */
 820                 if (!lustre_in_group_p(uc, (la->la_valid & LA_GID) ?
 821                                        la->la_gid : oattr->la_gid) &&
 822                     !cap_raised(uc->uc_cap, CAP_FSETID))
 823                         la->la_mode &= ~S_ISGID;
 824         } else {
 825                 la->la_mode = oattr->la_mode;
 826         }
 827
 828         /* Make sure a caller can chown. */
 829         if (la->la_valid & LA_UID) {
 830                 if (la->la_uid == (uid_t) -1)
 831                         la->la_uid = oattr->la_uid;
 832                 /* we want rbac roles to have precedence over any other
 833                  * permission or capability checks
 834                  */
 835                 if (!uc->uc_rbac_file_perms ||
 836                     ((uc->uc_fsuid != oattr->la_uid ||
 837                       la->la_uid != oattr->la_uid) &&
 838                      !cap_raised(uc->uc_cap, CAP_CHOWN)))
 839                         RETURN(-EPERM);
 840
 841                 /* If the user or group of a non-directory has been
 842                  * changed by a non-root user, remove the setuid bit.
 843                  * 19981026 David C Niemi <niemi@tux.org>
 844                  *
 845                  * Changed this to apply to all users, including root,
 846                  * to avoid some races. This is the behavior we had in
 847                  * 2.0. The check for non-root was definitely wrong
 848                  * for 2.2 anyway, as it should have been using
 849                  * CAP_FSETID rather than fsuid -- 19990830 SD.
 850                  */
 851                 if (((oattr->la_mode & S_ISUID) == S_ISUID) &&
 852                 !S_ISDIR(oattr->la_mode)) {
 853                         la->la_mode &= ~S_ISUID;
 854                         la->la_valid |= LA_MODE;
 855                 }
 856         }
 857
 858         /* Make sure caller can chgrp. */
 859         if (la->la_valid & LA_GID) {
 860                 if (la->la_gid == (gid_t) -1)
 861                         la->la_gid = oattr->la_gid;
 862                 /* we want rbac roles to have precedence over any other
 863                  * permission or capability checks
 864                  */
 865                 if (!uc->uc_rbac_file_perms ||
 866                     ((uc->uc_fsuid != oattr->la_uid ||
 867                       (la->la_gid != oattr->la_gid &&
 868                        !lustre_in_group_p(uc, la->la_gid))) &&
 869                      !cap_raised(uc->uc_cap, CAP_CHOWN)))
 870                         RETURN(-EPERM);
 871
 872                 /* Likewise, if the user or group of a non-directory
 873                  * has been changed by a non-root user, remove the
 874                  * setgid bit UNLESS there is no group execute bit
 875                  * (this would be a file marked for mandatory
 876                  * locking).  19981026 David C Niemi <niemi@tux.org>
 877                  *
 878                  * Removed the fsuid check (see the comment above) --
 879                  * 19990830 SD.
 880                  */
 881                 if (((oattr->la_mode & (S_ISGID | 0010)) ==
 882                     (S_ISGID | 0010)) && !S_ISDIR(oattr->la_mode)) {
 883                         la->la_mode &= ~S_ISGID;
 884                         la->la_valid |= LA_MODE;
 885                 }
 886         }
 887
 888         if (la->la_valid & (LA_SIZE | LA_BLOCKS)) {
 889                 if (!((flags & MDS_OWNEROVERRIDE) &&
 890                       (uc->uc_fsuid == oattr->la_uid)) &&
 891                     !(flags & MDS_PERM_BYPASS)) {
 892                         int mask = MAY_WRITE;
 893
 894                         /* for chgrp, allow the update with
 895                          * read-only permissions
 896                          */
 897                         if (S_ISREG(oattr->la_mode) && la->la_valid & LA_GID &&
 898                             la->la_gid != oattr->la_gid)
 899                                 mask = MAY_READ;
 900                         rc = mdd_permission_internal(env, obj, oattr,
 901                                                      mask);
 902                         if (rc != 0)
 903                                 RETURN(rc);
 904                 }
 905         }
 906
 907         if (la->la_valid & LA_CTIME) {
 908                 /**
 909                  * The pure setattr, it has the priority over what is
 910                  * already set, do not drop it if ctime is equal.
 911                  */
 912                 if (la->la_ctime < oattr->la_ctime)
 913                         la->la_valid &= ~(LA_ATIME | LA_MTIME | LA_CTIME);
 914         }
 915
 916         RETURN(0);
 917 }
 918
 919 static int mdd_changelog_data_store_by_fid(const struct lu_env *env,
 920                                            struct mdd_device *mdd,
 921                                            enum changelog_rec_type type,
 922                                            enum changelog_rec_flags clf_flags,
 923                                            const struct lu_fid *fid,
 924                                            const struct lu_fid *pfid,
 925                                            const char *xattr_name,
 926                                            struct thandle *handle)
 927 {
 928         const struct lu_ucred *uc = lu_ucred(env);
 929         enum changelog_rec_extra_flags xflags = CLFE_INVALID;
 930         struct llog_changelog_rec *rec;
 931         struct lu_buf *buf;
 932         int reclen;
 933         int rc;
 934
 935         clf_flags = (clf_flags & CLF_FLAGMASK) | CLF_VERSION | CLF_EXTRA_FLAGS;
 936
 937         if (uc) {
 938                 if (uc->uc_jobid[0] != '\0')
 939                         clf_flags |= CLF_JOBID;
 940                 xflags |= CLFE_UIDGID;
 941                 xflags |= CLFE_NID;
 942                 xflags |= CLFE_NID_BE;
 943         }
 944         if (type == CL_OPEN || type == CL_DN_OPEN)
 945                 xflags |= CLFE_OPEN;
 946         if (type == CL_SETXATTR || type == CL_GETXATTR)
 947                 xflags |= CLFE_XATTR;
 948
 949         reclen = llog_data_len(LLOG_CHANGELOG_HDR_SZ +
 950                                changelog_rec_offset(clf_flags & CLF_SUPPORTED,
 951                                                     xflags & CLFE_SUPPORTED));
 952         buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mdi_chlg_buf, reclen);
 953         if (buf->lb_buf == NULL)
 954                 RETURN(-ENOMEM);
 955         rec = buf->lb_buf;
 956
 957         rec->cr_hdr.lrh_len = reclen;
 958         rec->cr.cr_flags = clf_flags;
 959         rec->cr.cr_type = (__u32)type;
 960         rec->cr.cr_tfid = *fid;
 961         if (pfid)
 962                 rec->cr.cr_pfid = *pfid;
 963         rec->cr.cr_namelen = 0;
 964
 965         if (clf_flags & CLF_JOBID)
 966                 mdd_changelog_rec_ext_jobid(&rec->cr, uc->uc_jobid);
 967
 968         if (clf_flags & CLF_EXTRA_FLAGS) {
 969                 mdd_changelog_rec_ext_extra_flags(&rec->cr, xflags);
 970                 if (xflags & CLFE_UIDGID)
 971                         mdd_changelog_rec_extra_uidgid(&rec->cr,
 972                                                        uc->uc_uid, uc->uc_gid);
 973                 if (xflags & CLFE_NID)
 974                         mdd_changelog_rec_extra_nid(&rec->cr, &uc->uc_nid);
 975                 if (xflags & CLFE_OPEN)
 976                         mdd_changelog_rec_extra_omode(&rec->cr, clf_flags);
 977                 if (xflags & CLFE_XATTR) {
 978                         if (xattr_name == NULL)
 979                                 RETURN(-EINVAL);
 980                         mdd_changelog_rec_extra_xattr(&rec->cr, xattr_name);
 981                 }
 982         }
 983
 984         rc = mdd_changelog_store(env, mdd, rec, handle);
 985         RETURN(rc);
 986 }
 987
 988
 989 /** Store a data change changelog record
 990  * If this fails, we must fail the whole transaction; we don't
 991  * want the change to commit without the log entry.
 992  * \param mdd_obj - mdd_object of change
 993  * \param handle - transaction handle
 994  * \param pfid - parent FID for CL_MTIME changelogs
 995  */
 996 int mdd_changelog_data_store(const struct lu_env *env, struct mdd_device *mdd,
 997                              enum changelog_rec_type type,
 998                              enum changelog_rec_flags clf_flags,
 999                              struct mdd_object *mdd_obj, struct thandle *handle,
1000                              const struct lu_fid *pfid)
1001 {
1002         int                              rc;
1003
1004         LASSERT(mdd_obj != NULL);
1005         LASSERT(handle != NULL);
1006
1007         if (!mdd_changelog_enabled(env, mdd, type))
1008                 RETURN(0);
1009
1010         if (mdd_is_volatile_obj(mdd_obj))
1011                 RETURN(0);
1012
1013         if ((type >= CL_MTIME) && (type <= CL_ATIME) &&
1014             ktime_before(mdd->mdd_cl.mc_starttime, mdd_obj->mod_cltime)) {
1015                 /* Don't need multiple updates in this log */
1016                 /* Don't check under lock - no big deal if we get extra entry */
1017                 RETURN(0);
1018         }
1019
1020         rc = mdd_changelog_data_store_by_fid(env, mdd, type, clf_flags,
1021                                              mdd_object_fid(mdd_obj), pfid,
1022                                              NULL, handle);
1023         if (rc == 0)
1024                 mdd_obj->mod_cltime = ktime_get();
1025
1026         RETURN(rc);
1027 }
1028
1029 int mdd_changelog_data_store_xattr(const struct lu_env *env,
1030                                    struct mdd_device *mdd,
1031                                    enum changelog_rec_type type,
1032                                    enum changelog_rec_flags clf_flags,
1033                                    struct mdd_object *mdd_obj,
1034                                    const char *xattr_name,
1035                                    struct thandle *handle)
1036 {
1037         int rc;
1038
1039         LASSERT(mdd_obj != NULL);
1040         LASSERT(handle != NULL);
1041
1042         if (!mdd_changelog_enabled(env, mdd, type))
1043                 RETURN(0);
1044
1045         if (mdd_is_volatile_obj(mdd_obj))
1046                 RETURN(0);
1047
1048         if ((type >= CL_MTIME) && (type <= CL_ATIME) &&
1049             ktime_before(mdd->mdd_cl.mc_starttime, mdd_obj->mod_cltime)) {
1050                 /* Don't need multiple updates in this log */
1051                 /* Don't check under lock - no big deal if we get an extra
1052                  * entry
1053                  */
1054                 RETURN(0);
1055         }
1056
1057         rc = mdd_changelog_data_store_by_fid(env, mdd, type, clf_flags,
1058                                              mdd_object_fid(mdd_obj), NULL,
1059                                              xattr_name, handle);
1060         if (rc == 0)
1061                 mdd_obj->mod_cltime = ktime_get();
1062
1063         RETURN(rc);
1064 }
1065
1066 /* Only the bottom CLF_FLAGMASK bits of @flags are stored in the record,
1067  * except for open flags have a dedicated record to store 32 bits of flags.
1068  */
1069 static int mdd_changelog(const struct lu_env *env, enum changelog_rec_type type,
1070                          enum changelog_rec_flags clf_flags,
1071                          struct md_device *m, const struct lu_fid *fid)
1072 {
1073         struct thandle *handle;
1074         struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
1075         int rc;
1076
1077         ENTRY;
1078
1079         LASSERT(fid != NULL);
1080
1081         /* We'll check this again below, but we check now before we
1082          * start a transaction.
1083          */
1084         if (!mdd_changelog_enabled(env, mdd, type))
1085                 RETURN(0);
1086
1087         handle = mdd_trans_create(env, mdd);
1088         if (IS_ERR(handle))
1089                 RETURN(PTR_ERR(handle));
1090
1091         rc = mdd_declare_changelog_store(env, mdd, type, NULL, NULL, handle);
1092         if (rc)
1093                 GOTO(stop, rc);
1094
1095         rc = mdd_trans_start(env, mdd, handle);
1096         if (rc)
1097                 GOTO(stop, rc);
1098
1099         /* note we only store the low CLF_FLAGMASK bits of la_valid */
1100         rc = mdd_changelog_data_store_by_fid(env, mdd, type,
1101                                              clf_flags & CLF_FLAGMASK,
1102                                              fid, NULL, NULL, handle);
1103
1104 stop:
1105         rc = mdd_trans_stop(env, mdd, rc, handle);
1106
1107         RETURN(rc);
1108 }
1109
1110 /**
1111  * Save LMA extended attributes with data from \a ma.
1112  *
1113  * HSM and Size-On-MDS data will be extracted from \ma if they are valid, if
1114  * not, LMA EA will be first read from disk, modified and write back.
1115  *
1116  */
1117 /* Precedence for choosing record type when multiple
1118  * attributes change: setattr > mtime > ctime > atime
1119  * (ctime changes when mtime does, plus chmod/chown.
1120  * atime and ctime are independent.)
1121  */
1122 static int mdd_attr_set_changelog(const struct lu_env *env,
1123                                   struct md_object *obj, struct thandle *handle,
1124                                   const struct lu_fid *pfid, __u64 valid)
1125 {
1126         struct mdd_device *mdd = mdo2mdd(obj);
1127         int bits, type = 0;
1128
1129         bits =  (valid & LA_SIZE)  ? BIT(CL_TRUNC) : 0;
1130         bits |= (valid & ~(LA_CTIME|LA_MTIME|LA_ATIME)) ? BIT(CL_SETATTR) : 0;
1131         bits |= (valid & LA_MTIME) ? BIT(CL_MTIME) : 0;
1132         bits |= (valid & LA_CTIME) ? BIT(CL_CTIME) : 0;
1133         bits |= (valid & LA_ATIME) ? BIT(CL_ATIME) : 0;
1134         bits = bits & mdd->mdd_cl.mc_current_mask;
1135         /* This is an implementation limit rather than a protocol limit */
1136         BUILD_BUG_ON(CL_LAST > sizeof(int) * 8);
1137         if (bits == 0)
1138                 return 0;
1139
1140         /* The record type is the lowest non-masked set bit */
1141         type = __ffs(bits);
1142
1143         /* only the low CLF_FLAGMASK bits of la_valid are stored */
1144         return mdd_changelog_data_store(env, mdd, type, valid & CLF_FLAGMASK,
1145                                         md2mdd_obj(obj), handle, pfid);
1146 }
1147
1148 static int mdd_declare_attr_set(const struct lu_env *env,
1149                                 struct mdd_device *mdd,
1150                                 struct mdd_object *obj,
1151                                 const struct lu_attr *attr,
1152                                 struct thandle *handle)
1153 {
1154         int rc;
1155
1156         rc = mdo_declare_attr_set(env, obj, attr, handle);
1157         if (rc)
1158                 return rc;
1159
1160 #ifdef CONFIG_LUSTRE_FS_POSIX_ACL
1161         if (attr->la_valid & LA_MODE) {
1162                 mdd_read_lock(env, obj, DT_TGT_CHILD);
1163                 rc = mdo_xattr_get(env, obj, &LU_BUF_NULL,
1164                                    XATTR_NAME_ACL_ACCESS);
1165                 mdd_read_unlock(env, obj);
1166                 if (rc == -EOPNOTSUPP || rc == -ENODATA)
1167                         rc = 0;
1168                 else if (rc < 0)
1169                         return rc;
1170
1171                 if (rc != 0) {
1172                         struct lu_buf *buf = mdd_buf_get(env, NULL, rc);
1173
1174                         rc = mdo_declare_xattr_set(env, obj, buf,
1175                                                    XATTR_NAME_ACL_ACCESS, 0,
1176                                                    handle);
1177                         if (rc)
1178                                 return rc;
1179                 }
1180         }
1181 #endif
1182
1183         rc = mdd_declare_changelog_store(env, mdd, CL_SETXATTR, NULL, NULL,
1184                                          handle);
1185         return rc;
1186 }
1187
1188 /*
1189  * LU-3671
1190  * LU-7239
1191  *
1192  * permission changes may require sync operation, to mitigate performance
1193  * impact, only do this for dir and when permission is reduced.
1194  *
1195  * For regular files, version is updated with permission change (see VBR), async
1196  * permission won't cause any issue, while missing permission change on
1197  * directory may affect accessibility of other objects after recovery.
1198  */
1199 static inline bool permission_needs_sync(const struct lu_attr *old,
1200                                          const struct lu_attr *new)
1201 {
1202         if (!S_ISDIR(old->la_mode))
1203                 return false;
1204
1205         if (new->la_valid & LA_UID && old->la_uid != new->la_uid)
1206                 return true;
1207
1208         if (new->la_valid & LA_GID && old->la_gid != new->la_gid)
1209                 return true;
1210
1211         if (new->la_valid & LA_MODE) {
1212                 /* turned on sticky bit */
1213                 if (!(old->la_mode & S_ISVTX) && (new->la_mode & S_ISVTX))
1214                         return true;
1215
1216                 /* set-GID has no impact on what is allowed, not checked */
1217
1218                 /* turned off setuid bit, or one of rwx for someone */
1219                 if (((new->la_mode & old->la_mode) & (0777 | S_ISUID)) !=
1220                      (old->la_mode & (0777 | S_ISUID)))
1221                         return true;
1222         }
1223
1224         return false;
1225 }
1226
1227 static inline __u64 mdd_lmm_dom_size(void *buf)
1228 {
1229         struct lov_mds_md *lmm = buf;
1230         struct lov_comp_md_v1 *comp_v1;
1231         struct lov_mds_md *v1;
1232         __u32 off;
1233
1234         if (lmm == NULL)
1235                 return 0;
1236
1237         if (le32_to_cpu(lmm->lmm_magic) != LOV_MAGIC_COMP_V1)
1238                 return 0;
1239
1240         comp_v1 = (struct lov_comp_md_v1 *)lmm;
1241         off = le32_to_cpu(comp_v1->lcm_entries[0].lcme_offset);
1242         v1 = (struct lov_mds_md *)((char *)comp_v1 + off);
1243
1244         /* DoM entry is the first entry always */
1245         if (lov_pattern(le32_to_cpu(v1->lmm_pattern)) & LOV_PATTERN_MDT)
1246                 return le64_to_cpu(comp_v1->lcm_entries[0].lcme_extent.e_end);
1247
1248         return 0;
1249 }
1250
1251 /* set attr and LOV EA at once, return updated attr */
1252 int mdd_attr_set(const struct lu_env *env, struct md_object *obj,
1253                  const struct md_attr *ma)
1254 {
1255         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1256         struct mdd_device *mdd = mdo2mdd(obj);
1257         struct thandle *handle = NULL;
1258         struct lu_attr *la_copy = &mdd_env_info(env)->mdi_la_for_fix;
1259         struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
1260         const struct lu_attr *la = &ma->ma_attr;
1261         struct lu_ucred  *uc;
1262         struct lquota_id_info qi;
1263         bool quota_reserved = false;
1264         bool chrgrp_by_unprivileged_user = false;
1265         int rc;
1266
1267         ENTRY;
1268
1269         /* we do not use ->attr_set() for LOV/HSM EA any more */
1270         LASSERT((ma->ma_valid & MA_LOV) == 0);
1271         LASSERT((ma->ma_valid & MA_HSM) == 0);
1272
1273         rc = mdd_la_get(env, mdd_obj, attr);
1274         if (rc)
1275                 RETURN(rc);
1276
1277         *la_copy = ma->ma_attr;
1278         mdd_write_lock(env, mdd_obj, DT_TGT_CHILD);
1279         rc = mdd_fix_attr(env, mdd_obj, attr, la_copy, ma);
1280         mdd_write_unlock(env, mdd_obj);
1281         if (rc)
1282                 RETURN(rc);
1283
1284         /* no need to setattr anymore */
1285         if (la_copy->la_valid == 0) {
1286                 CDEBUG(D_INODE,
1287                        "%s: no valid attribute on "DFID", previous was %#llx\n",
1288                        mdd_obj_dev_name(mdd_obj),
1289                        PFID(mdd_object_fid(mdd_obj)), la->la_valid);
1290
1291                 RETURN(0);
1292         }
1293
1294         /* If an unprivileged user changes group of some file,
1295          * the setattr operation will be processed synchronously to
1296          * honor the quota limit of the corresponding group. see LU-5152
1297          */
1298         uc = lu_ucred_check(env);
1299         memset(&qi, 0, sizeof(qi));
1300         if (S_ISREG(attr->la_mode) && la->la_valid & LA_GID &&
1301             la->la_gid != attr->la_gid && uc != NULL && uc->uc_fsuid != 0) {
1302                 CDEBUG(D_QUOTA, "%s: reserve quota for changing group: gid=%u size=%llu\n",
1303                        mdd2obd_dev(mdd)->obd_name, la->la_gid, la->la_size);
1304
1305                 if (la->la_valid & LA_BLOCKS)
1306                         qi.lqi_space = la->la_blocks << 9;
1307                 else if (la->la_valid & LA_SIZE)
1308                         qi.lqi_space = la->la_size;
1309                 /* use local attr gotten above */
1310                 else if (attr->la_valid & LA_BLOCKS)
1311                         qi.lqi_space = attr->la_blocks << 9;
1312                 else if (attr->la_valid & LA_SIZE)
1313                         qi.lqi_space = attr->la_size;
1314
1315                 if (qi.lqi_space > 0) {
1316                         qi.lqi_id.qid_gid = la->la_gid;
1317                         qi.lqi_type = GRPQUOTA;
1318                         qi.lqi_space = toqb(qi.lqi_space);
1319                         qi.lqi_is_blk = true;
1320                         rc = dt_reserve_or_free_quota(env, mdd->mdd_bottom,
1321                                                       &qi);
1322
1323                         if (rc) {
1324                                 CDEBUG(D_QUOTA, "%s: failed to reserve quota for gid %d size %llu\n",
1325                                        mdd2obd_dev(mdd)->obd_name,
1326                                        la->la_gid, qi.lqi_space);
1327
1328                                 GOTO(out, rc);
1329                         }
1330
1331                         quota_reserved = true;
1332                         la_copy->la_valid |= LA_FLAGS;
1333                 }
1334
1335                 chrgrp_by_unprivileged_user = true;
1336
1337                 /* Flush the possible existing client setattr requests to OSTs
1338                  * to keep the order with the current setattr operation that
1339                  * will be sent directly to OSTs. see LU-5152
1340                  *
1341                  * LU-11303 disable sync as this is too heavyweight.
1342                  * This should be replaced with a sync only for the object
1343                  * being modified here, not the whole filesystem.
1344                    rc = dt_sync(env, mdd->mdd_child);
1345                    if (rc)
1346                         GOTO(out, rc);
1347                  */
1348         }
1349
1350         handle = mdd_trans_create(env, mdd);
1351         if (IS_ERR(handle)) {
1352                 rc = PTR_ERR(handle);
1353                 handle = NULL;
1354
1355                 GOTO(out, rc);
1356         }
1357
1358         rc = mdd_declare_attr_set(env, mdd, mdd_obj, la_copy, handle);
1359         if (rc)
1360                 GOTO(out, rc);
1361
1362         rc = mdd_trans_start(env, mdd, handle);
1363         if (rc)
1364                 GOTO(out, rc);
1365
1366         if (!chrgrp_by_unprivileged_user && mdd->mdd_sync_permission &&
1367             permission_needs_sync(attr, la))
1368                 handle->th_sync = 1;
1369
1370         if (la->la_valid & (LA_MTIME | LA_CTIME))
1371                 CDEBUG(D_INODE, "setting mtime %llu, ctime %llu\n",
1372                        la->la_mtime, la->la_ctime);
1373
1374         mdd_write_lock(env, mdd_obj, DT_TGT_CHILD);
1375
1376         /* LU-10509: setattr of LA_SIZE should be skipped case of DOM,
1377          * otherwise following truncate will do nothing and truncated
1378          * data may be read again. This is a quick fix until LU-11033
1379          * will be resolved.
1380          */
1381         if (la_copy->la_valid & LA_SIZE) {
1382                 struct lu_buf *lov_buf = mdd_buf_get(env, NULL, 0);
1383
1384                 rc = mdd_stripe_get(env, mdd_obj, lov_buf, XATTR_NAME_LOV);
1385                 if (rc) {
1386                         rc = 0;
1387                 } else {
1388                         if (mdd_lmm_dom_size(lov_buf->lb_buf) > 0)
1389                                 la_copy->la_valid &= ~LA_SIZE;
1390                         lu_buf_free(lov_buf);
1391                 }
1392         }
1393
1394         if (la_copy->la_valid) {
1395                 rc = mdd_attr_set_internal(env, mdd_obj, la_copy, handle, 1);
1396
1397                 if (rc == -EDQUOT && la_copy->la_flags & LUSTRE_SET_SYNC_FL) {
1398                         /* rollback to the original gid */
1399                         la_copy->la_flags &= ~LUSTRE_SET_SYNC_FL;
1400                         la_copy->la_gid = attr->la_gid;
1401                         mdd_attr_set_internal(env, mdd_obj, la_copy, handle, 1);
1402                 }
1403         }
1404         mdd_write_unlock(env, mdd_obj);
1405
1406 out:
1407         if (rc == 0)
1408                 rc = mdd_attr_set_changelog(env, obj, handle, &ma->ma_pfid,
1409                                             la_copy->la_valid);
1410
1411         if (rc == 0 && quota_reserved) {
1412                 struct thandle *sub_th;
1413
1414                 sub_th = thandle_get_sub_by_dt(env, handle, mdd->mdd_bottom);
1415                 if (unlikely(IS_ERR(sub_th))) {
1416                         qi.lqi_space *= -1;
1417                         dt_reserve_or_free_quota(env, mdd->mdd_bottom, &qi);
1418                 } else {
1419                         sub_th->th_reserved_quota = qi;
1420                 }
1421         } else if (quota_reserved) {
1422                 qi.lqi_space *= -1;
1423                 dt_reserve_or_free_quota(env, mdd->mdd_bottom, &qi);
1424         }
1425
1426         if (handle != NULL)
1427                 rc = mdd_trans_stop(env, mdd, rc, handle);
1428
1429         return rc;
1430 }
1431
1432 static int mdd_xattr_sanity_check(const struct lu_env *env,
1433                                   struct mdd_object *obj,
1434                                   const struct lu_attr *attr,
1435                                   const char *name)
1436 {
1437         struct lu_ucred *uc     = lu_ucred_assert(env);
1438
1439         ENTRY;
1440
1441         if (attr->la_flags & (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL))
1442                 RETURN(-EPERM);
1443
1444         if (strncmp(XATTR_USER_PREFIX, name,
1445                     sizeof(XATTR_USER_PREFIX) - 1) == 0) {
1446                 /* For sticky directories, only the owner and privileged user
1447                  * can write attributes.
1448                  */
1449                 if (S_ISDIR(attr->la_mode) && (attr->la_mode & S_ISVTX) &&
1450                     (uc->uc_fsuid != attr->la_uid) &&
1451                     !cap_raised(uc->uc_cap, CAP_FOWNER))
1452                         RETURN(-EPERM);
1453         } else if (strcmp(name, XATTR_NAME_SOM) != 0 &&
1454                    (uc->uc_fsuid != attr->la_uid) &&
1455                    !cap_raised(uc->uc_cap, CAP_FOWNER)) {
1456                 RETURN(-EPERM);
1457         }
1458
1459         RETURN(0);
1460 }
1461
1462 /**
1463  * Check if a string begins with a given prefix.
1464  *
1465  * \param str     String to check
1466  * \param prefix  Substring to check at the beginning of \a str
1467  * \return true/false whether the condition is verified.
1468  */
1469 static inline bool has_prefix(const char *str, const char *prefix)
1470 {
1471         return strncmp(prefix, str, strlen(prefix)) == 0;
1472 }
1473
1474 /**
1475  * Indicate the kind of changelog to store (if any) for a xattr set/del.
1476  *
1477  * \param[in]  xattr_name  Full extended attribute name.
1478  *
1479  * \return type of changelog to use, or CL_NONE if no changelog is to be emitted
1480  */
1481 static enum changelog_rec_type
1482 mdd_xattr_changelog_type(const struct lu_env *env, struct mdd_device *mdd,
1483                          const char *xattr_name)
1484 {
1485         /* Layout changes systematically recorded */
1486         if (strcmp(XATTR_NAME_LOV, xattr_name) == 0 ||
1487             strcmp(XATTR_LUSTRE_LOV, xattr_name) == 0 ||
1488             allowed_lustre_lov(xattr_name))
1489                 return CL_LAYOUT;
1490
1491         /* HSM information changes systematically recorded */
1492         if (strcmp(XATTR_NAME_HSM, xattr_name) == 0)
1493                 return CL_HSM;
1494
1495         /* Avoid logging SOM xattr for every file */
1496         if (strcmp(XATTR_NAME_SOM, xattr_name) == 0)
1497                 return CL_NONE;
1498
1499         if (has_prefix(xattr_name, XATTR_USER_PREFIX) ||
1500             has_prefix(xattr_name, XATTR_NAME_POSIX_ACL_ACCESS) ||
1501             has_prefix(xattr_name, XATTR_NAME_POSIX_ACL_DEFAULT) ||
1502             has_prefix(xattr_name, XATTR_TRUSTED_PREFIX) ||
1503             has_prefix(xattr_name, XATTR_SECURITY_PREFIX))
1504                 return CL_SETXATTR;
1505
1506         return CL_NONE;
1507 }
1508
1509 static int mdd_declare_xattr_set(const struct lu_env *env,
1510                                  struct mdd_device *mdd,
1511                                  struct mdd_object *obj,
1512                                  const struct lu_buf *buf,
1513                                  const char *name,
1514                                  int fl, struct thandle *handle)
1515 {
1516         enum changelog_rec_type type;
1517         int rc;
1518
1519         rc = mdo_declare_xattr_set(env, obj, buf, name, fl, handle);
1520         if (rc)
1521                 return rc;
1522
1523         type = mdd_xattr_changelog_type(env, mdd, name);
1524         if (type < 0)
1525                 return 0; /* no changelog to store */
1526
1527         return mdd_declare_changelog_store(env, mdd, type, NULL, NULL, handle);
1528 }
1529
1530 /*
1531  * Compare current and future data of HSM EA and add a changelog if needed.
1532  *
1533  * Caller should have write-locked \param obj.
1534  *
1535  * \param buf - Future HSM EA content.
1536  * \retval 0 if no changelog is needed or changelog was added properly.
1537  * \retval -ve errno if there was a problem
1538  */
1539 static int mdd_hsm_update_locked(const struct lu_env *env,
1540                                  struct md_object *obj,
1541                                  const struct lu_buf *buf,
1542                                  struct thandle *handle,
1543                                  enum changelog_rec_flags *clf_flags)
1544 {
1545         struct mdd_thread_info *info = mdd_env_info(env);
1546         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1547         struct lu_buf *current_buf;
1548         struct md_hsm *current_mh;
1549         struct md_hsm *new_mh;
1550         int rc;
1551
1552         ENTRY;
1553         OBD_ALLOC_PTR(current_mh);
1554         if (current_mh == NULL)
1555                 RETURN(-ENOMEM);
1556
1557         /* Read HSM attrs from disk */
1558         current_buf = lu_buf_check_and_alloc(&info->mdi_xattr_buf,
1559                         min_t(unsigned int,
1560                               mdd_obj2mdd_dev(mdd_obj)->mdd_dt_conf.ddp_max_ea_size,
1561                               XATTR_SIZE_MAX));
1562         rc = mdo_xattr_get(env, mdd_obj, current_buf, XATTR_NAME_HSM);
1563         rc = lustre_buf2hsm(current_buf->lb_buf, rc, current_mh);
1564         if (rc < 0 && rc != -ENODATA)
1565                 GOTO(free, rc);
1566         else if (rc == -ENODATA)
1567                 current_mh->mh_flags = 0;
1568
1569         /* Map future HSM xattr */
1570         OBD_ALLOC_PTR(new_mh);
1571         if (new_mh == NULL)
1572                 GOTO(free, rc = -ENOMEM);
1573         lustre_buf2hsm(buf->lb_buf, buf->lb_len, new_mh);
1574
1575         rc = 0;
1576
1577         /* Flags differ, set flags for the changelog that will be added */
1578         if (current_mh->mh_flags != new_mh->mh_flags) {
1579                 hsm_set_cl_event(clf_flags, HE_STATE);
1580                 if (new_mh->mh_flags & HS_DIRTY)
1581                         hsm_set_cl_flags(clf_flags, CLF_HSM_DIRTY);
1582         }
1583
1584         OBD_FREE_PTR(new_mh);
1585         EXIT;
1586 free:
1587         OBD_FREE_PTR(current_mh);
1588         return rc;
1589 }
1590
1591 static int mdd_object_pfid_replace(const struct lu_env *env,
1592                                    struct mdd_object *o)
1593 {
1594         struct mdd_device *mdd = mdo2mdd(&o->mod_obj);
1595         struct thandle *handle;
1596         int rc;
1597
1598         handle = mdd_trans_create(env, mdd);
1599         if (IS_ERR(handle))
1600                 RETURN(PTR_ERR(handle));
1601
1602         handle->th_complex = 1;
1603
1604         /* it doesn't need to track the PFID update via llog, because LFSCK
1605          * will repair it even it goes wrong
1606          */
1607         rc = mdd_declare_xattr_set(env, mdd, o, NULL, XATTR_NAME_FID,
1608                                    0, handle);
1609         if (rc)
1610                 GOTO(out, rc);
1611
1612         rc = mdd_trans_start(env, mdd, handle);
1613         if (rc != 0)
1614                 GOTO(out, rc);
1615
1616         rc = mdo_xattr_set(env, o, NULL, XATTR_NAME_FID, 0, handle);
1617         if (rc)
1618                 GOTO(out, rc);
1619
1620 out:
1621         mdd_trans_stop(env, mdd, rc, handle);
1622         return rc;
1623 }
1624
1625
1626 static int mdd_declare_xattr_del(const struct lu_env *env,
1627                                  struct mdd_device *mdd,
1628                                  struct mdd_object *obj,
1629                                  const char *name,
1630                                  struct thandle *handle);
1631
1632 static int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
1633                          const char *name);
1634
1635 static int mdd_xattr_merge(const struct lu_env *env, struct md_object *md_obj,
1636                            struct md_object *md_vic)
1637 {
1638         struct mdd_device *mdd = mdo2mdd(md_obj);
1639         struct mdd_object *obj = md2mdd_obj(md_obj);
1640         struct mdd_object *vic = md2mdd_obj(md_vic);
1641         struct lu_buf *buf;
1642         struct lu_buf *buf_vic;
1643         struct lov_mds_md *lmm;
1644         struct thandle *handle;
1645         struct lu_attr *cattr = MDD_ENV_VAR(env, cattr);
1646         struct lu_attr *tattr = MDD_ENV_VAR(env, tattr);
1647         bool is_same_projid;
1648         int rc;
1649         int retried = 0;
1650
1651         ENTRY;
1652
1653         rc = mdd_la_get(env, obj, cattr);
1654         if (rc)
1655                 RETURN(rc);
1656
1657         rc = mdd_la_get(env, vic, tattr);
1658         if (rc)
1659                 RETURN(rc);
1660
1661         is_same_projid = cattr->la_projid == tattr->la_projid;
1662
1663         if (lu_fid_cmp(mdd_object_fid(obj), mdd_object_fid(vic)) == 0)
1664                 RETURN(-EPERM);
1665 retry:
1666         handle = mdd_trans_create(env, mdd);
1667         if (IS_ERR(handle))
1668                 RETURN(PTR_ERR(handle));
1669
1670         buf = &mdd_env_info(env)->mdi_buf[0];
1671         buf_vic = &mdd_env_info(env)->mdi_buf[1];
1672
1673         if (!is_same_projid) {
1674                 rc = mdd_declare_attr_set(env, mdd, vic, cattr, handle);
1675                 if (rc)
1676                         GOTO(stop, rc);
1677         }
1678
1679         /* get EA of victim file */
1680         memset(buf_vic, 0, sizeof(*buf_vic));
1681         rc = mdd_stripe_get(env, vic, buf_vic, XATTR_NAME_LOV);
1682         if (rc < 0) {
1683                 if (rc == -ENODATA)
1684                         rc = 0;
1685                 GOTO(stop, rc);
1686         }
1687
1688         /* parse the layout of victim file */
1689         lmm = buf_vic->lb_buf;
1690         if (le32_to_cpu(lmm->lmm_magic) != LOV_MAGIC_COMP_V1)
1691                 GOTO(stop, rc = -EINVAL);
1692
1693         /* save EA of target file for restore */
1694         memset(buf, 0, sizeof(*buf));
1695         rc = mdd_stripe_get(env, obj, buf, XATTR_NAME_LOV);
1696         if (rc < 0)
1697                 GOTO(stop, rc);
1698
1699         /* Get rid of the layout from victim object */
1700         rc = mdd_declare_xattr_del(env, mdd, vic, XATTR_NAME_LOV, handle);
1701         if (rc)
1702                 GOTO(stop, rc);
1703
1704         rc = mdd_declare_xattr_set(env, mdd, obj, buf_vic, XATTR_NAME_LOV,
1705                                    LU_XATTR_MERGE, handle);
1706         if (rc)
1707                 GOTO(stop, rc);
1708
1709         rc = mdd_trans_start(env, mdd, handle);
1710         if (rc != 0)
1711                 GOTO(stop, rc);
1712
1713         rc = mdd_write_lock_two_objects(env, obj, vic);
1714         if (rc)
1715                 GOTO(stop, rc);
1716
1717         if (!is_same_projid) {
1718                 cattr->la_valid = LA_PROJID;
1719                 rc = mdd_attr_set_internal(env, vic, cattr, handle, 0);
1720                 if (rc)
1721                         GOTO(out, rc);
1722         }
1723
1724         rc = mdo_xattr_set(env, obj, buf_vic, XATTR_NAME_LOV, LU_XATTR_MERGE,
1725                            handle);
1726         if (rc)
1727                 GOTO(out, rc);
1728
1729         rc = mdo_xattr_del(env, vic, XATTR_NAME_LOV, handle);
1730         if (rc)
1731                 GOTO(out_restore, rc);
1732
1733         (void)mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, obj, handle,
1734                                        NULL);
1735         (void)mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, vic, handle,
1736                                        NULL);
1737
1738 out_restore:
1739         if (rc) {
1740                 int rc2 = mdo_xattr_set(env, obj, buf, XATTR_NAME_LOV,
1741                                         LU_XATTR_REPLACE, handle);
1742                 if (rc2)
1743                         CERROR("%s: failed rollback of "DFID" layout: file state unknown: rc = %d\n",
1744                                mdd_obj_dev_name(obj),
1745                                PFID(mdd_object_fid(obj)), rc2);
1746         }
1747
1748 out:
1749         mdd_write_unlock_two_objects(env, obj, vic);
1750 stop:
1751         mdd_trans_stop(env, mdd, rc, handle);
1752         lu_buf_free(buf);
1753         lu_buf_free(buf_vic);
1754
1755         /**
1756          * -EAGAIN means transaction execution phase detect the layout
1757          * has been changed by others.
1758          */
1759         if (!rc)
1760                 (void) mdd_object_pfid_replace(env, obj);
1761         else if (rc == -EAGAIN && retried++ < MAX_TRANS_RETRIED)
1762                 GOTO(retry, retried);
1763
1764         RETURN(rc);
1765 }
1766
1767 /**
1768  * Extract the mirror with specified mirror id, and store the splitted
1769  * mirror layout to @buf.
1770  *
1771  * \param[in] comp_v1   mirrored layout
1772  * \param[in] mirror_id the mirror with mirror_id to be extracted
1773  * \param[out] buf      store the layout excluding the extracted mirror,
1774  *                      caller free the buffer we allocated in this function
1775  * \param[out] buf_vic  store the extracted layout, caller free the buffer
1776  *                      we allocated in this function
1777  *
1778  * \retval      0 on success; < 0 if error happens
1779  */
1780 static int mdd_split_ea(struct lov_comp_md_v1 *comp_v1, __u16 mirror_id,
1781                         struct lu_buf *buf, struct lu_buf *buf_vic)
1782 {
1783         struct lov_comp_md_v1 *comp_rem;
1784         struct lov_comp_md_v1 *comp_vic;
1785         struct lov_comp_md_entry_v1 *entry;
1786         struct lov_comp_md_entry_v1 *entry_rem;
1787         struct lov_comp_md_entry_v1 *entry_vic;
1788         __u16 mirror_cnt;
1789         __u16 comp_cnt, count = 0;
1790         int lmm_size, lmm_size_vic = 0;
1791         int i, j, k;
1792         int offset, offset_rem, offset_vic;
1793
1794         mirror_cnt = le16_to_cpu(comp_v1->lcm_mirror_count) + 1;
1795         /* comp_v1 should contains more than 1 mirror */
1796         if (mirror_cnt <= 1)
1797                 return -EINVAL;
1798         comp_cnt = le16_to_cpu(comp_v1->lcm_entry_count);
1799         lmm_size = le32_to_cpu(comp_v1->lcm_size);
1800
1801         for (i = 0; i < comp_cnt; i++) {
1802                 entry = &comp_v1->lcm_entries[i];
1803                 if (mirror_id_of(le32_to_cpu(entry->lcme_id)) == mirror_id) {
1804                         count++;
1805                         lmm_size_vic += sizeof(*entry);
1806                         lmm_size_vic += le32_to_cpu(entry->lcme_size);
1807                 } else if (count > 0) {
1808                         /* find the specified mirror */
1809                         break;
1810                 }
1811         }
1812
1813         if (count == 0)
1814                 return -EINVAL;
1815
1816         lu_buf_alloc(buf, lmm_size - lmm_size_vic);
1817         if (!buf->lb_buf)
1818                 return -ENOMEM;
1819
1820         lu_buf_alloc(buf_vic, sizeof(*comp_vic) + lmm_size_vic);
1821         if (!buf_vic->lb_buf) {
1822                 lu_buf_free(buf);
1823                 return -ENOMEM;
1824         }
1825
1826         comp_rem = (struct lov_comp_md_v1 *)buf->lb_buf;
1827         comp_vic = (struct lov_comp_md_v1 *)buf_vic->lb_buf;
1828
1829         memcpy(comp_rem, comp_v1, sizeof(*comp_v1));
1830         comp_rem->lcm_mirror_count = cpu_to_le16(mirror_cnt - 2);
1831         comp_rem->lcm_entry_count = cpu_to_le32(comp_cnt - count);
1832         comp_rem->lcm_size = cpu_to_le32(lmm_size - lmm_size_vic);
1833         if (!comp_rem->lcm_mirror_count)
1834                 comp_rem->lcm_flags = cpu_to_le16(comp_rem->lcm_flags &
1835                                                   ~LCM_FL_FLR_MASK);
1836
1837         memset(comp_vic, 0, sizeof(*comp_v1));
1838         comp_vic->lcm_magic = cpu_to_le32(LOV_MAGIC_COMP_V1);
1839         comp_vic->lcm_mirror_count = 0;
1840         comp_vic->lcm_entry_count = cpu_to_le32(count);
1841         comp_vic->lcm_size = cpu_to_le32(lmm_size_vic + sizeof(*comp_vic));
1842         comp_vic->lcm_flags = cpu_to_le16(comp_vic->lcm_flags &
1843                                           ~LCM_FL_FLR_MASK);
1844         comp_vic->lcm_layout_gen = 0;
1845
1846         offset = sizeof(*comp_v1) + sizeof(*entry) * comp_cnt;
1847         offset_rem = sizeof(*comp_rem) +
1848                      sizeof(*entry_rem) * (comp_cnt - count);
1849         offset_vic = sizeof(*comp_vic) + sizeof(*entry_vic) * count;
1850         for (i = j = k = 0; i < comp_cnt; i++) {
1851                 struct lov_mds_md *lmm, *lmm_dst;
1852                 bool vic = false;
1853
1854                 entry = &comp_v1->lcm_entries[i];
1855                 entry_vic = &comp_vic->lcm_entries[j];
1856                 entry_rem = &comp_rem->lcm_entries[k];
1857
1858                 if (mirror_id_of(le32_to_cpu(entry->lcme_id)) == mirror_id)
1859                         vic = true;
1860
1861                 /* copy component entry */
1862                 if (vic) {
1863                         memcpy(entry_vic, entry, sizeof(*entry));
1864                         entry_vic->lcme_flags &= cpu_to_le32(LCME_FL_INIT);
1865                         entry_vic->lcme_offset = cpu_to_le32(offset_vic);
1866                         j++;
1867                 } else {
1868                         memcpy(entry_rem, entry, sizeof(*entry));
1869                         entry_rem->lcme_offset = cpu_to_le32(offset_rem);
1870                         k++;
1871                 }
1872
1873                 lmm = (struct lov_mds_md *)((char *)comp_v1 + offset);
1874                 if (vic)
1875                         lmm_dst = (struct lov_mds_md *)
1876                                         ((char *)comp_vic + offset_vic);
1877                 else
1878                         lmm_dst = (struct lov_mds_md *)
1879                                         ((char *)comp_rem + offset_rem);
1880
1881                 /* copy component entry blob */
1882                 memcpy(lmm_dst, lmm, le32_to_cpu(entry->lcme_size));
1883
1884                 /* blob offset advance */
1885                 offset += le32_to_cpu(entry->lcme_size);
1886                 if (vic)
1887                         offset_vic += le32_to_cpu(entry->lcme_size);
1888                 else
1889                         offset_rem += le32_to_cpu(entry->lcme_size);
1890         }
1891
1892         return 0;
1893 }
1894
1895 static int mdd_xattr_split(const struct lu_env *env, struct md_object *md_obj,
1896                            struct md_rejig_data *mrd)
1897 {
1898         struct mdd_device *mdd = mdo2mdd(md_obj);
1899         struct mdd_object *obj = md2mdd_obj(md_obj);
1900         struct mdd_object *vic = NULL;
1901         struct lu_buf *buf;
1902         struct lu_buf *buf_save;
1903         struct lu_buf *buf_vic;
1904         struct lov_comp_md_v1 *lcm;
1905         struct thandle *handle;
1906         int rc;
1907         bool dom_stripe = false;
1908         int retried = 0;
1909
1910         ENTRY;
1911
1912         /**
1913          * NULL @mrd_obj means mirror deleting, and use NULL vic to indicate
1914          * mirror deleting
1915          */
1916         if (mrd->mrd_obj)
1917                 vic = md2mdd_obj(mrd->mrd_obj);
1918
1919 retry:
1920         buf = &mdd_env_info(env)->mdi_buf[0];
1921         buf_save = &mdd_env_info(env)->mdi_buf[1];
1922         buf_vic = &mdd_env_info(env)->mdi_buf[2];
1923
1924         handle = mdd_trans_create(env, mdd);
1925         if (IS_ERR(handle))
1926                 RETURN(PTR_ERR(handle));
1927
1928         /* get EA of mirrored file */
1929         memset(buf_save, 0, sizeof(*buf));
1930         rc = mdd_stripe_get(env, obj, buf_save, XATTR_NAME_LOV);
1931         if (rc < 0)
1932                 GOTO(stop, rc);
1933
1934         lcm = buf_save->lb_buf;
1935         if (le32_to_cpu(lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
1936                 GOTO(stop, rc = -EINVAL);
1937
1938         /**
1939          * Extract the mirror with specified mirror id, and store the splitted
1940          * mirror layout to the victim buffer.
1941          */
1942         memset(buf, 0, sizeof(*buf));
1943         memset(buf_vic, 0, sizeof(*buf_vic));
1944         rc = mdd_split_ea(lcm, mrd->mrd_mirror_id, buf, buf_vic);
1945         if (rc < 0)
1946                 GOTO(stop, rc);
1947         /**
1948          * @buf stores layout w/o the specified mirror, @buf_vic stores the
1949          * splitted mirror
1950          */
1951
1952         dom_stripe = mdd_lmm_dom_size(buf_vic->lb_buf) > 0;
1953
1954         if (vic) {
1955                 /**
1956                  * non delete mirror split
1957                  *
1958                  * declare obj set remaining layout in @buf, will set obj's
1959                  * in-memory layout
1960                  */
1961                 rc = mdd_declare_xattr_set(env, mdd, obj, buf, XATTR_NAME_LOV,
1962                                            LU_XATTR_SPLIT, handle);
1963                 if (rc)
1964                         GOTO(stop, rc);
1965
1966                 /* declare vic set splitted layout in @buf_vic */
1967                 rc = mdd_declare_xattr_set(env, mdd, vic, buf_vic,
1968                                            XATTR_NAME_LOV, LU_XATTR_SPLIT,
1969                                            handle);
1970                 if (rc)
1971                         GOTO(stop, rc);
1972         } else {
1973                 /**
1974                  * declare delete mirror objects in @buf_vic, will change obj's
1975                  * in-memory layout
1976                  */
1977                 rc = mdd_declare_xattr_set(env, mdd, obj, buf_vic,
1978                                            XATTR_NAME_LOV, LU_XATTR_PURGE,
1979                                            handle);
1980                 if (rc)
1981                         GOTO(stop, rc);
1982
1983                 /* declare obj set remaining layout in @buf */
1984                 rc = mdd_declare_xattr_set(env, mdd, obj, buf,
1985                                            XATTR_NAME_LOV, LU_XATTR_SPLIT,
1986                                            handle);
1987                 if (rc)
1988                         GOTO(stop, rc);
1989         }
1990
1991         rc = mdd_trans_start(env, mdd, handle);
1992         if (rc)
1993                 GOTO(stop, rc);
1994
1995         rc = mdd_write_lock_two_objects(env, obj, vic);
1996         if (rc)
1997                 GOTO(stop, rc);
1998
1999         /* set obj's layout in @buf */
2000         rc = mdo_xattr_set(env, obj, buf, XATTR_NAME_LOV, LU_XATTR_SPLIT,
2001                            handle);
2002         if (rc)
2003                 GOTO(unlock, rc);
2004
2005         if (vic) {
2006                 /* set vic's layout in @buf_vic */
2007                 rc = mdo_xattr_set(env, vic, buf_vic, XATTR_NAME_LOV,
2008                                    LU_XATTR_CREATE, handle);
2009                 if (rc)
2010                         GOTO(out_restore, rc);
2011         } else {
2012                 /* delete mirror objects */
2013                 rc = mdo_xattr_set(env, obj, buf_vic, XATTR_NAME_LOV,
2014                                    LU_XATTR_PURGE, handle);
2015                 if (rc)
2016                         GOTO(out_restore, rc);
2017         }
2018
2019         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, obj, handle,
2020                                       NULL);
2021         if (rc)
2022                 GOTO(out_restore, rc);
2023
2024         if (vic) {
2025                 rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, vic,
2026                                               handle, NULL);
2027                 if (rc)
2028                         GOTO(out_restore, rc);
2029         }
2030
2031 out_restore:
2032         if (rc) {
2033                 /* restore obj's in-memory and on-disk layout */
2034                 int rc2 = mdo_xattr_set(env, obj, buf_save, XATTR_NAME_LOV,
2035                                         LU_XATTR_REPLACE, handle);
2036                 if (rc2)
2037                         CERROR("%s: failed rollback "DFID" layout: file state unknown: rc = %d\n",
2038                                mdd_obj_dev_name(obj),
2039                                PFID(mdd_object_fid(obj)), rc);
2040         }
2041
2042 unlock:
2043         mdd_write_unlock_two_objects(env, obj, vic);
2044 stop:
2045         rc = mdd_trans_stop(env, mdd, rc, handle);
2046
2047         /* Truncate local DOM data if all went well */
2048         if (!rc && dom_stripe)
2049                 mdd_dom_fixup(env, mdd, obj, NULL);
2050
2051         lu_buf_free(buf_save);
2052         lu_buf_free(buf);
2053         lu_buf_free(buf_vic);
2054
2055         /**
2056          * -EAGAIN means transaction execution phase detect the layout
2057          * has been changed by others.
2058          */
2059         if (!rc)
2060                 (void) mdd_object_pfid_replace(env, obj);
2061         else if (rc == -EAGAIN && retried++ < MAX_TRANS_RETRIED)
2062                 GOTO(retry, retried);
2063
2064         RETURN(rc);
2065 }
2066
2067 static int mdd_layout_merge_allowed(const struct lu_env *env,
2068                                     struct md_object *target,
2069                                     struct md_object *victim)
2070 {
2071         struct mdd_object *o1 = md2mdd_obj(target);
2072         int rc = 0;
2073
2074         /* cannot extend directory's LOVEA */
2075         if (S_ISDIR(mdd_object_type(o1))) {
2076                 rc = -EISDIR;
2077                 CERROR("%s: Don't extend directory's LOVEA, just set it: rc = %d\n",
2078                        mdd_obj_dev_name(o1), rc);
2079                 RETURN(rc);
2080         }
2081
2082         RETURN(rc);
2083 }
2084
2085 /**
2086  * The caller should guarantee to update the object ctime
2087  * after xattr_set if needed.
2088  */
2089 static int mdd_xattr_set(const struct lu_env *env, struct md_object *obj,
2090                          const struct lu_buf *buf, const char *name,
2091                          int fl)
2092 {
2093         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2094         struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
2095         struct mdd_device *mdd = mdo2mdd(obj);
2096         struct thandle *handle;
2097         enum changelog_rec_type  cl_type;
2098         enum changelog_rec_flags clf_flags = 0;
2099         int retried = 0;
2100         int rc;
2101
2102         ENTRY;
2103
2104 retry:
2105         rc = mdd_la_get(env, mdd_obj, attr);
2106         if (rc)
2107                 RETURN(rc);
2108
2109         rc = mdd_xattr_sanity_check(env, mdd_obj, attr, name);
2110         if (rc)
2111                 RETURN(rc);
2112
2113         if (strcmp(name, XATTR_LUSTRE_LOV) == 0 &&
2114             (fl == LU_XATTR_MERGE || fl == LU_XATTR_SPLIT)) {
2115                 struct md_rejig_data *mrd = buf->lb_buf;
2116                 struct md_object *victim = mrd->mrd_obj;
2117
2118                 if (buf->lb_len != sizeof(*mrd))
2119                         RETURN(-EINVAL);
2120
2121
2122                 if (fl == LU_XATTR_MERGE) {
2123                         rc = mdd_layout_merge_allowed(env, obj, victim);
2124                         if (rc)
2125                                 RETURN(rc);
2126                         /* merge layout of victim as a mirror of obj's. */
2127                         rc = mdd_xattr_merge(env, obj, victim);
2128                 } else {
2129                         rc = mdd_xattr_split(env, obj, mrd);
2130                 }
2131                 RETURN(rc);
2132         }
2133
2134         if (strcmp(name, XATTR_NAME_ACL_ACCESS) == 0 ||
2135             strcmp(name, XATTR_NAME_ACL_DEFAULT) == 0) {
2136                 struct posix_acl *acl;
2137
2138                 /* user set empty ACL, this should be treated as removing ACL */
2139                 acl = posix_acl_from_xattr(&init_user_ns, buf->lb_buf,
2140                                            buf->lb_len);
2141                 if (IS_ERR(acl))
2142                         RETURN(PTR_ERR(acl));
2143                 if (acl == NULL) {
2144                         rc = mdd_xattr_del(env, obj, name);
2145                         RETURN(rc);
2146                 }
2147                 posix_acl_release(acl);
2148         }
2149
2150         if (!strcmp(name, XATTR_NAME_ACL_ACCESS)) {
2151                 rc = mdd_acl_set(env, mdd_obj, attr, buf, fl);
2152                 RETURN(rc);
2153         }
2154
2155         handle = mdd_trans_create(env, mdd);
2156         if (IS_ERR(handle))
2157                 RETURN(PTR_ERR(handle));
2158
2159         rc = mdd_declare_xattr_set(env, mdd, mdd_obj, buf, name, fl, handle);
2160         if (rc)
2161                 GOTO(stop, rc);
2162
2163         rc = mdd_trans_start(env, mdd, handle);
2164         if (rc)
2165                 GOTO(stop, rc);
2166
2167         mdd_write_lock(env, mdd_obj, DT_TGT_CHILD);
2168
2169         if (strcmp(XATTR_NAME_HSM, name) == 0) {
2170                 rc = mdd_hsm_update_locked(env, obj, buf, handle, &clf_flags);
2171                 if (rc)
2172                         GOTO(out_unlock, rc);
2173         }
2174
2175         rc = mdo_xattr_set(env, mdd_obj, buf, name, fl, handle);
2176 out_unlock:
2177         mdd_write_unlock(env, mdd_obj);
2178         if (rc)
2179                 GOTO(stop, rc);
2180
2181         cl_type = mdd_xattr_changelog_type(env, mdd, name);
2182         if (cl_type != CL_NONE)
2183                 rc = mdd_changelog_data_store_xattr(env, mdd, cl_type,
2184                                                     clf_flags, mdd_obj, name,
2185                                                     handle);
2186
2187 stop:
2188         rc = mdd_trans_stop(env, mdd, rc, handle);
2189
2190         /**
2191          * -EAGAIN means transaction execution phase detect the layout
2192          * has been changed by others.
2193          */
2194         if (rc == -EAGAIN && retried++ < MAX_TRANS_RETRIED)
2195                 GOTO(retry, retried);
2196
2197         RETURN(rc);
2198 }
2199
2200 static int mdd_declare_xattr_del(const struct lu_env *env,
2201                                  struct mdd_device *mdd,
2202                                  struct mdd_object *obj,
2203                                  const char *name,
2204                                  struct thandle *handle)
2205 {
2206         enum changelog_rec_type type;
2207         int rc;
2208
2209         rc = mdo_declare_xattr_del(env, obj, name, handle);
2210         if (rc)
2211                 return rc;
2212
2213         type = mdd_xattr_changelog_type(env, mdd, name);
2214         if (type < 0)
2215                 return 0; /* no changelog to store */
2216
2217         return mdd_declare_changelog_store(env, mdd, type, NULL, NULL, handle);
2218 }
2219
2220 /**
2221  * The caller should guarantee to update the object ctime
2222  * after xattr_set if needed.
2223  */
2224 static int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
2225                          const char *name)
2226 {
2227         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2228         struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
2229         struct mdd_device *mdd = mdo2mdd(obj);
2230         struct thandle *handle;
2231         int rc;
2232
2233         ENTRY;
2234
2235         rc = mdd_la_get(env, mdd_obj, attr);
2236         if (rc)
2237                 RETURN(rc);
2238
2239         rc = mdd_xattr_sanity_check(env, mdd_obj, attr, name);
2240         if (rc)
2241                 RETURN(rc);
2242
2243         handle = mdd_trans_create(env, mdd);
2244         if (IS_ERR(handle))
2245                 RETURN(PTR_ERR(handle));
2246
2247         rc = mdd_declare_xattr_del(env, mdd, mdd_obj, name, handle);
2248         if (rc)
2249                 GOTO(stop, rc);
2250
2251         rc = mdd_trans_start(env, mdd, handle);
2252         if (rc)
2253                 GOTO(stop, rc);
2254
2255         mdd_write_lock(env, mdd_obj, DT_TGT_CHILD);
2256         rc = mdo_xattr_del(env, mdd_obj, name, handle);
2257         mdd_write_unlock(env, mdd_obj);
2258         if (rc)
2259                 GOTO(stop, rc);
2260
2261         if (mdd_xattr_changelog_type(env, mdd, name) < 0)
2262                 GOTO(stop, rc = 0);
2263
2264         rc = mdd_changelog_data_store_xattr(env, mdd, CL_SETXATTR, 0, mdd_obj,
2265                                             name, handle);
2266
2267         EXIT;
2268 stop:
2269         return mdd_trans_stop(env, mdd, rc, handle);
2270 }
2271
2272 /*
2273  * read lov/lmv EA of an object
2274  * return the lov/lmv EA in an allocated lu_buf
2275  */
2276 int mdd_stripe_get(const struct lu_env *env, struct mdd_object *obj,
2277                    struct lu_buf *lmm_buf, const char *name)
2278 {
2279         struct lu_buf *buf = &mdd_env_info(env)->mdi_big_buf;
2280         int rc;
2281
2282         ENTRY;
2283
2284         if (buf->lb_buf == NULL) {
2285                 buf = lu_buf_check_and_alloc(buf, 4096);
2286                 if (buf->lb_buf == NULL)
2287                         RETURN(-ENOMEM);
2288         }
2289
2290 repeat:
2291         rc = mdo_xattr_get(env, obj, buf, name);
2292         if (rc == -ERANGE) {
2293                 /* mdi_big_buf is allocated but too small need to increase it */
2294                 buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mdi_big_buf,
2295                                              buf->lb_len * 2);
2296                 if (buf->lb_buf == NULL)
2297                         RETURN(-ENOMEM);
2298                 goto repeat;
2299         } else if (rc < 0) {
2300                 RETURN(rc);
2301         } else if (rc == 0) {
2302                 RETURN(-ENODATA);
2303         }
2304
2305         lu_buf_alloc(lmm_buf, rc);
2306         if (lmm_buf->lb_buf == NULL)
2307                 RETURN(-ENOMEM);
2308
2309         /*
2310          * we don't use lmm_buf directly, because we don't know xattr size, so
2311          * by using mdi_big_buf we can avoid calling mdo_xattr_get() twice.
2312          */
2313         memcpy(lmm_buf->lb_buf, buf->lb_buf, rc);
2314
2315         RETURN(0);
2316 }
2317
2318 static int emit_changelog_after_swap_layout(const struct lu_env *env,
2319                                             struct thandle *handle,
2320                                             struct mdd_object *o,
2321                                             struct lu_buf *hsm_buf)
2322 {
2323         enum changelog_rec_flags flags = 0;
2324         enum changelog_rec_type type;
2325         enum hsm_states hsm_flags;
2326         struct hsm_attrs *attrs;
2327
2328         ENTRY;
2329         attrs = hsm_buf->lb_buf;
2330         hsm_flags = le32_to_cpu(attrs->hsm_flags);
2331
2332         if ((hsm_flags & HS_RELEASED) && !mdd_is_dead_obj(o)) {
2333                 hsm_set_cl_event(&flags, HE_RELEASE);
2334                 type = CL_HSM;
2335         } else {
2336                 type = CL_LAYOUT;
2337         }
2338
2339         return mdd_changelog_data_store(env, mdo2mdd(&o->mod_obj), type,
2340                                         flags, o, handle, NULL);
2341 }
2342
2343 /*
2344  *  check if layout swapping between 2 objects is allowed
2345  *  the rules are:
2346  *  - only normal FIDs or non-system IGIFs
2347  *  - same type of objects
2348  *  - same owner/group (so quotas are still valid) unless this is from HSM
2349  *    release.
2350  */
2351 static int mdd_layout_swap_allowed(const struct lu_env *env,
2352                                    struct mdd_object *o1,
2353                                    const struct lu_attr *attr1,
2354                                    struct mdd_object *o2,
2355                                    const struct lu_attr *attr2,
2356                                    __u64 flags)
2357 {
2358         const struct lu_fid *fid1, *fid2;
2359
2360         ENTRY;
2361
2362         fid1 = mdd_object_fid(o1);
2363         fid2 = mdd_object_fid(o2);
2364
2365         if (!fid_is_norm(fid1) &&
2366             (!fid_is_igif(fid1) || IS_ERR(mdd_links_get(env, o1))))
2367                 RETURN(-EBADF);
2368
2369         if (!fid_is_norm(fid2) &&
2370             (!fid_is_igif(fid2) || IS_ERR(mdd_links_get(env, o2))))
2371                 RETURN(-EBADF);
2372
2373         if (mdd_object_type(o1) != mdd_object_type(o2)) {
2374                 if (S_ISDIR(mdd_object_type(o1)))
2375                         RETURN(-ENOTDIR);
2376                 if (S_ISREG(mdd_object_type(o1)))
2377                         RETURN(-EISDIR);
2378                 RETURN(-EBADF);
2379         }
2380
2381         /* Do not check uid/gid for release since the orphan is created as root
2382          */
2383         if (flags & SWAP_LAYOUTS_MDS_RELEASE)
2384                 RETURN(0);
2385
2386         if ((attr1->la_uid != attr2->la_uid) ||
2387             (attr1->la_gid != attr2->la_gid))
2388                 RETURN(-EPERM);
2389
2390         RETURN(0);
2391 }
2392
2393 /* XXX To set the proper lmm_oi & lmm_layout_gen when swap layouts, we have to
2394  *     look into the layout in MDD layer.
2395  */
2396 static int mdd_lmm_oi(struct lov_mds_md *lmm, struct ost_id *oi, bool get)
2397 {
2398         struct lov_comp_md_v1   *comp_v1;
2399         struct lov_mds_md       *v1;
2400         int                      i, ent_count;
2401         __u32                    off;
2402
2403         if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_COMP_V1) {
2404                 comp_v1 = (struct lov_comp_md_v1 *)lmm;
2405                 ent_count = le16_to_cpu(comp_v1->lcm_entry_count);
2406
2407                 if (ent_count == 0)
2408                         return -EINVAL;
2409
2410                 if (get) {
2411                         int i = 0;
2412
2413                         off = le32_to_cpu(comp_v1->lcm_entries[i].lcme_offset);
2414                         v1 = (struct lov_mds_md *)((char *)comp_v1 + off);
2415                         if (le32_to_cpu(v1->lmm_magic) != LOV_MAGIC_FOREIGN) {
2416                                 *oi = v1->lmm_oi;
2417                         } else {
2418                                 if (ent_count == 1)
2419                                         return -EINVAL;
2420
2421                                 i = 1;
2422                                 off = le32_to_cpu(
2423                                         comp_v1->lcm_entries[i].lcme_offset);
2424                                 v1 = (struct lov_mds_md *)((char *)comp_v1 +
2425                                                            off);
2426                                 if (le32_to_cpu(v1->lmm_magic) ==
2427                                                         LOV_MAGIC_FOREIGN)
2428                                         return -EINVAL;
2429
2430                                 *oi = v1->lmm_oi;
2431                         }
2432                 } else {
2433                         for (i = 0; i < le32_to_cpu(ent_count); i++) {
2434                                 off = le32_to_cpu(comp_v1->lcm_entries[i].
2435                                                 lcme_offset);
2436                                 v1 = (struct lov_mds_md *)((char *)comp_v1 +
2437                                                 off);
2438                                 if (le32_to_cpu(v1->lmm_magic) !=
2439                                                         LOV_MAGIC_FOREIGN)
2440                                         v1->lmm_oi = *oi;
2441                         }
2442                 }
2443         } else if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_V1 ||
2444                    le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_V3) {
2445                 if (get)
2446                         *oi = lmm->lmm_oi;
2447                 else
2448                         lmm->lmm_oi = *oi;
2449         } else {
2450                 return -EINVAL;
2451         }
2452         return 0;
2453 }
2454
2455 static inline int mdd_get_lmm_oi(struct lov_mds_md *lmm, struct ost_id *oi)
2456 {
2457         return mdd_lmm_oi(lmm, oi, true);
2458 }
2459
2460 static inline int mdd_set_lmm_oi(struct lov_mds_md *lmm, struct ost_id *oi)
2461 {
2462         return mdd_lmm_oi(lmm, oi, false);
2463 }
2464
2465 static int mdd_lmm_gen(struct lov_mds_md *lmm, __u32 *gen, bool get)
2466 {
2467         struct lov_comp_md_v1 *comp_v1;
2468
2469         if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_COMP_V1) {
2470                 comp_v1 = (struct lov_comp_md_v1 *)lmm;
2471                 if (get)
2472                         *gen = le32_to_cpu(comp_v1->lcm_layout_gen);
2473                 else
2474                         comp_v1->lcm_layout_gen = cpu_to_le32(*gen);
2475         } else if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_V1 ||
2476                    le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_V3) {
2477                 __u16 tmp_gen = *gen;
2478
2479                 if (get)
2480                         *gen = le16_to_cpu(lmm->lmm_layout_gen);
2481                 else
2482                         lmm->lmm_layout_gen = cpu_to_le16(tmp_gen);
2483         } else {
2484                 return -EINVAL;
2485         }
2486         return 0;
2487 }
2488
2489 static inline int mdd_get_lmm_gen(struct lov_mds_md *lmm, __u32 *gen)
2490 {
2491         return mdd_lmm_gen(lmm, gen, true);
2492 }
2493
2494 static inline int mdd_set_lmm_gen(struct lov_mds_md *lmm, __u32 *gen)
2495 {
2496         return mdd_lmm_gen(lmm, gen, false);
2497 }
2498
2499 int mdd_dom_fixup(const struct lu_env *env, struct mdd_device *mdd,
2500                   struct mdd_object *mo, struct mdd_object *vo)
2501 {
2502         struct dt_object *dom, *vlt;
2503         dt_obj_version_t dv = 0;
2504         struct thandle *th;
2505         int rc;
2506
2507         ENTRY;
2508
2509         if (vo) {
2510                 vlt = dt_object_locate(mdd_object_child(vo), mdd->mdd_bottom);
2511                 if (!vlt)
2512                         GOTO(out, rc = -ENOENT);
2513                 dv = dt_data_version_get(env, vlt);
2514                 if (!dv)
2515                         GOTO(out, rc = -ENODATA);
2516         }
2517
2518         dom = dt_object_locate(mdd_object_child(mo), mdd->mdd_bottom);
2519
2520         th = dt_trans_create(env, mdd->mdd_bottom);
2521         if (IS_ERR(th))
2522                 GOTO(out, rc = PTR_ERR(th));
2523
2524         if (vo) {
2525                 rc = dt_declare_data_version_set(env, dom, th);
2526                 if (rc)
2527                         GOTO(stop, rc);
2528         } else {
2529                 rc = dt_declare_data_version_del(env, dom, th);
2530                 if (rc)
2531                         GOTO(stop, rc);
2532                 rc = dt_declare_punch(env, dom, 0, OBD_OBJECT_EOF, th);
2533                 if (rc)
2534                         GOTO(stop, rc);
2535         }
2536
2537         rc = dt_trans_start_local(env, mdd->mdd_bottom, th);
2538         if (rc != 0)
2539                 GOTO(stop, rc);
2540
2541         if (vo) {
2542                 dt_data_version_set(env, dom, dv, th);
2543         } else {
2544                 dt_data_version_del(env, dom, th);
2545                 rc = dt_punch(env, dom, 0, OBD_OBJECT_EOF, th);
2546         }
2547
2548 stop:
2549         dt_trans_stop(env, mdd->mdd_bottom, th);
2550 out:
2551         /* Ignore failure but report the error */
2552         if (rc)
2553                 CERROR("%s: can't manage DOM file "DFID" data: rc = %d\n",
2554                        mdd_obj_dev_name(mo), PFID(mdd_object_fid(mo)), rc);
2555         return rc;
2556 }
2557
2558 /* Swap Layout HSM object information: every information needed to decide how
2559  * to update the HSM xattr flag during a layout swap.
2560  */
2561 struct sl_hsm_object_info {
2562         struct mdd_object *o;   /* the object whose HSM attribute to check */
2563         struct lu_buf *hsm_buf; /* buffer initialized with the content of the
2564                                  * HSM xattr.
2565                                  */
2566         int xattr_flags;        /* 0: do not update HSM xattr
2567                                  * LU_XATTR_REPLACE or LU_XATTR_CREATE: update
2568                                  * the xattr with this flag
2569                                  */
2570         __u64 dv;               /* dataversion of the object */
2571 };
2572
2573 /* Read the HSM xattr and store its content into \p hsm_buf. */
2574 static int fetch_hsm_xattr(const struct lu_env *env, struct mdd_object *o,
2575                            struct lu_buf *hsm_buf)
2576 {
2577         int rc;
2578
2579         lu_buf_alloc(hsm_buf, sizeof(struct hsm_attrs));
2580         if (hsm_buf->lb_buf == NULL)
2581                 return -ENOMEM;
2582
2583         rc = mdo_xattr_get(env, o, hsm_buf, XATTR_NAME_HSM);
2584         if (rc < 0)
2585                 return rc;
2586
2587         return 0;
2588 }
2589
2590 /* return true if HSM xattr needs update */
2591 static bool swap_hsm_set_dirty(struct lu_buf *out_buf, struct lu_buf *src_buf)
2592 {
2593         struct md_hsm src_hsm;
2594
2595         lustre_buf2hsm(src_buf->lb_buf, src_buf->lb_len, &src_hsm);
2596         if (!(src_hsm.mh_flags & (HS_ARCHIVED | HS_EXISTS)) ||
2597             src_hsm.mh_flags & HS_DIRTY)
2598                 /* do not update HSM attr of non archived or dirty files */
2599                 return false;
2600
2601         src_hsm.mh_flags |= HS_DIRTY;
2602         lustre_hsm2buf(out_buf->lb_buf, &src_hsm);
2603
2604         return true;
2605 }
2606
2607 /* return 1 HSM xattr needs update
2608  *        0 do not update HSM xattr
2609  *   -EPERM data version mismatch, no swap
2610  */
2611 static int swap_hsm_update_version(struct lu_buf *out_buf, __u64 out_dv,
2612                                    struct lu_buf *src_buf, __u64 src_dv)
2613 {
2614         struct md_hsm src_hsm;
2615
2616         /* Put lb_len as a second argument since we know that src_buf is a valid
2617          * HSM xattr buffer.
2618          */
2619         lustre_buf2hsm(src_buf->lb_buf, src_buf->lb_len, &src_hsm);
2620
2621         if (!(src_hsm.mh_flags & (HS_ARCHIVED | HS_EXISTS)) ||
2622             src_hsm.mh_flags & HS_DIRTY)
2623                 return 0;
2624
2625         /* migration with old client -> set the dirty flag */
2626         if (!src_dv || !out_dv) {
2627                 src_hsm.mh_flags |= HS_DIRTY;
2628                 goto hsm2buf;
2629         }
2630
2631         if (src_hsm.mh_arch_ver != src_dv) {
2632                 CDEBUG(D_HSM,
2633                        "HSM archive version and previous data version mismatch (arch_ver=%llu, prev_ver=%llu, new_ver=%llu)\n",
2634                        src_hsm.mh_arch_ver, src_dv, out_dv);
2635                 return -EPERM;
2636         }
2637
2638         src_hsm.mh_arch_ver = out_dv;
2639 hsm2buf:
2640         lustre_hsm2buf(out_buf->lb_buf, &src_hsm);
2641
2642         return 1;
2643 }
2644
2645 /* Allow HSM xattr swap only with volatile/orphan files.
2646  * This is used by HSM release/restore.
2647  */
2648 static inline bool swap_hsm_xattr_allowed(bool fst_has_hsm, bool snd_has_hsm,
2649                                           unsigned long o_fst_flag,
2650                                           unsigned long o_snd_flag)
2651 {
2652         return (fst_has_hsm && snd_has_hsm &&
2653                 ((o_fst_flag | o_snd_flag) & (ORPHAN_OBJ | VOLATILE_OBJ)));
2654 }
2655
2656 /* Read and update the data version of both objects if necessary. If they have
2657  * to be updated, declare the update.
2658  * \p xattr_flags will be updated if necessary to be used by mdo_xattr_set
2659  *
2660  * \p dv contains the data version of the file before and after the migration.
2661  *       It can be NULL when swapping layouts in other contexts (HSM or
2662  *       lfs swap_layout).
2663  */
2664 static int swap_layouts_prepare_hsm_attr(const struct lu_env *env,
2665                                          struct mdd_device *mdd,
2666                                          struct thandle *handle,
2667                                          struct sl_hsm_object_info *fst,
2668                                          struct sl_hsm_object_info *snd)
2669 {
2670         unsigned long o_fst_fl;
2671         unsigned long o_snd_fl;
2672         int rc2;
2673         int rc;
2674
2675         fst->xattr_flags = 0;
2676         snd->xattr_flags = 0;
2677
2678         rc = fetch_hsm_xattr(env, fst->o, fst->hsm_buf);
2679         if (rc != -ENODATA && rc != 0)
2680                 return rc;
2681
2682         rc2 = fetch_hsm_xattr(env, snd->o, snd->hsm_buf);
2683         if (rc2 != -ENODATA && rc2 != 0)
2684                 return rc2;
2685
2686         /* if nothing to swap, not an error */
2687         if (rc && rc2)
2688                 return 0;
2689
2690         /* swap if the first object have no HSM xattr */
2691         if (rc && !rc2) {
2692                 swap(fst, snd);
2693                 swap(rc, rc2);
2694         }
2695
2696         o_fst_fl = fst->o->mod_flags;
2697         o_snd_fl = snd->o->mod_flags;
2698
2699         if ((o_snd_fl & VOLATILE_OBJ) && rc2) {
2700                 /* migration of fst */
2701                 rc = swap_hsm_update_version(snd->hsm_buf, snd->dv,
2702                                              fst->hsm_buf, fst->dv);
2703                 if (rc == 0)
2704                         return 0;
2705                 if (rc < 0)
2706                         return rc;
2707
2708                 fst->xattr_flags = LU_XATTR_REPLACE;
2709
2710         } else if (swap_hsm_xattr_allowed(!rc, !rc2, o_fst_fl, o_snd_fl)) {
2711                 /* HSM release/restore -> HSM xattr swap */
2712                 fst->xattr_flags = LU_XATTR_REPLACE;
2713                 snd->xattr_flags = LU_XATTR_REPLACE;
2714
2715         } else if (!rc && !rc2) {
2716                 /* swap on 2 archived files is not supported (no rollback) */
2717                 return -EPERM;
2718
2719         } else if (rc2) {
2720                 /* swap layout with HSM fst and non-HSM snd */
2721                 if (!swap_hsm_set_dirty(snd->hsm_buf, fst->hsm_buf))
2722                         return 0;
2723
2724                 fst->xattr_flags = LU_XATTR_REPLACE;
2725         }
2726
2727         if (fst->xattr_flags) {
2728                 rc = mdd_declare_xattr_set(env, mdd, fst->o, snd->hsm_buf,
2729                                            XATTR_NAME_HSM, fst->xattr_flags,
2730                                            handle);
2731                 if (rc < 0)
2732                         return rc;
2733         }
2734
2735         if (snd->xattr_flags) {
2736                 rc = mdd_declare_xattr_set(env, mdd, snd->o, fst->hsm_buf,
2737                                            XATTR_NAME_HSM, snd->xattr_flags,
2738                                            handle);
2739                 if (rc < 0)
2740                         return rc;
2741         }
2742
2743         return 0;
2744 }
2745
2746 /* swap layouts between 2 lustre objects */
2747 static int mdd_swap_layouts(const struct lu_env *env,
2748                             struct md_object *obj1, struct md_object *obj2,
2749                             __u64 dv1, __u64 dv2, __u64 flags)
2750 {
2751         struct mdd_thread_info *info = mdd_env_info(env);
2752         struct mdd_object *fst_o = md2mdd_obj(obj1);
2753         struct mdd_object *snd_o = md2mdd_obj(obj2);
2754         struct lu_attr *fst_la = MDD_ENV_VAR(env, cattr);
2755         struct lu_attr *snd_la = MDD_ENV_VAR(env, tattr);
2756         struct mdd_device *mdd = mdo2mdd(obj1);
2757         struct lov_mds_md *fst_lmm, *snd_lmm;
2758         struct sl_hsm_object_info fst_info;
2759         struct sl_hsm_object_info snd_info;
2760         struct mdd_object *vlt_o = NULL;
2761         struct mdd_object *dom_o = NULL;
2762         struct ost_id *saved_oi = NULL;
2763         struct lu_buf *fst_hsm_buf;
2764         struct lu_buf *snd_hsm_buf;
2765         struct lu_buf *fst_buf;
2766         struct lu_buf *snd_buf;
2767         struct thandle *handle;
2768         __u64 domsize_dom;
2769         __u64 domsize_vlt;
2770         __u32 saved_gen;
2771         int retried = 0;
2772         __u32 fst_gen;
2773         __u32 snd_gen;
2774         int steps = 0;
2775         int fst_fl;
2776         int rc2 = 0;
2777         int rc;
2778
2779         ENTRY;
2780
2781         BUILD_BUG_ON(ARRAY_SIZE(info->mdi_buf) < 4);
2782 retry:
2783         fst_buf = &info->mdi_buf[0];
2784         snd_buf = &info->mdi_buf[1];
2785         fst_hsm_buf = &info->mdi_buf[2];
2786         snd_hsm_buf = &info->mdi_buf[3];
2787
2788         memset(info->mdi_buf, 0, sizeof(info->mdi_buf));
2789
2790         /* we have to sort the 2 obj, so locking will always
2791          * be in the same order, even in case of 2 concurrent swaps
2792          */
2793         rc = lu_fid_cmp(mdd_object_fid(fst_o), mdd_object_fid(snd_o));
2794         if (rc == 0) /* same fid ? */
2795                 RETURN(-EPERM);
2796
2797         if (rc < 0) {
2798                 swap(fst_o, snd_o);
2799                 swap(dv1, dv2);
2800         }
2801
2802         rc = mdd_la_get(env, fst_o, fst_la);
2803         if (rc != 0)
2804                 RETURN(rc);
2805
2806         rc = mdd_la_get(env, snd_o, snd_la);
2807         if (rc != 0)
2808                 RETURN(rc);
2809
2810         /* check if layout swapping is allowed */
2811         rc = mdd_layout_swap_allowed(env, fst_o, fst_la, snd_o, snd_la, flags);
2812         if (rc != 0)
2813                 RETURN(rc);
2814
2815         handle = mdd_trans_create(env, mdd);
2816         if (IS_ERR(handle))
2817                 RETURN(PTR_ERR(handle));
2818
2819         rc = mdd_stripe_get(env, fst_o, fst_buf, XATTR_NAME_LOV);
2820         if (rc < 0 && rc != -ENODATA)
2821                 GOTO(stop, rc);
2822
2823         rc = mdd_stripe_get(env, snd_o, snd_buf, XATTR_NAME_LOV);
2824         if (rc < 0 && rc != -ENODATA)
2825                 GOTO(stop, rc);
2826
2827         /* check if file has DoM. DoM file can be migrated only to another
2828          * DoM layout with the same DoM component size or to an non-DOM
2829          * layout. After migration to OSTs layout, local MDT inode data
2830          * should be truncated.
2831          * Objects are sorted by FIDs, considering that original file's FID
2832          * is always smaller the snd_o is always original file we are migrating
2833          * from.
2834          */
2835         domsize_dom = mdd_lmm_dom_size(snd_buf->lb_buf);
2836         domsize_vlt = mdd_lmm_dom_size(fst_buf->lb_buf);
2837
2838         /* Only migration is supported for DoM files, not 'swap_layouts' so
2839          * target file must be volatile and orphan.
2840          */
2841         if (fst_o->mod_flags & (ORPHAN_OBJ | VOLATILE_OBJ)) {
2842                 vlt_o = domsize_vlt ? fst_o : NULL;
2843                 dom_o = domsize_dom ? snd_o : NULL;
2844         } else if (snd_o->mod_flags & (ORPHAN_OBJ | VOLATILE_OBJ)) {
2845                 swap(domsize_dom, domsize_vlt);
2846                 vlt_o = domsize_vlt ? snd_o : NULL;
2847                 dom_o = domsize_dom ? fst_o : NULL;
2848         } else if (domsize_dom > 0 || domsize_vlt > 0) {
2849                 /* 'lfs swap_layouts' case, neither file should have DoM */
2850                 rc = -EOPNOTSUPP;
2851                 CDEBUG(D_LAYOUT, "cannot swap layouts with DOM component, use migration instead: rc = %d\n",
2852                        rc);
2853                 GOTO(stop, rc);
2854         }
2855
2856         if (domsize_vlt > 0 && domsize_dom == 0) {
2857                 rc = -EOPNOTSUPP;
2858                 CDEBUG(D_LAYOUT,
2859                        "%s: cannot swap "DFID" layout: OST to DOM migration not supported: rc = %d\n",
2860                        mdd_obj_dev_name(snd_o),
2861                        PFID(mdd_object_fid(snd_o)), rc);
2862                 GOTO(stop, rc);
2863         } else if (domsize_vlt > 0 && domsize_dom != domsize_vlt) {
2864                 rc = -EOPNOTSUPP;
2865                 CDEBUG(D_LAYOUT,
2866                        "%s: cannot swap "DFID" layout: new layout must have same DoM component size: rc = %d\n",
2867                        mdd_obj_dev_name(fst_o),
2868                        PFID(mdd_object_fid(fst_o)), rc);
2869                 GOTO(stop, rc);
2870         }
2871
2872         /* swapping 2 non existant layouts is a success */
2873         if (fst_buf->lb_buf == NULL && snd_buf->lb_buf == NULL)
2874                 GOTO(stop, rc = 0);
2875
2876         /* to help inode migration between MDT, it is better to
2877          * start by the no layout file (if one), so we order the swap
2878          */
2879         if (snd_buf->lb_buf == NULL) {
2880                 swap(fst_o, snd_o);
2881                 swap(fst_buf, snd_buf);
2882                 swap(dv1, dv2);
2883         }
2884
2885         fst_gen = snd_gen = 0;
2886         /* lmm and generation layout initialization */
2887         if (fst_buf->lb_buf != NULL) {
2888                 fst_lmm = fst_buf->lb_buf;
2889                 mdd_get_lmm_gen(fst_lmm, &fst_gen);
2890                 fst_fl  = LU_XATTR_REPLACE;
2891         } else {
2892                 fst_lmm = NULL;
2893                 fst_gen = 0;
2894                 fst_fl  = LU_XATTR_CREATE;
2895         }
2896
2897         snd_lmm = snd_buf->lb_buf;
2898         mdd_get_lmm_gen(snd_lmm, &snd_gen);
2899
2900         saved_gen = fst_gen;
2901         /* increase the generation layout numbers */
2902         snd_gen++;
2903         fst_gen++;
2904
2905         /*
2906          * XXX The layout generation is used to generate component IDs for
2907          *     the composite file, we have to do some special tweaks to make
2908          *     sure the layout generation is always adequate for that job.
2909          */
2910
2911         /* Skip invalid generation number for composite layout */
2912         if ((snd_gen & LCME_ID_MASK) == 0)
2913                 snd_gen++;
2914         if ((fst_gen & LCME_ID_MASK) == 0)
2915                 fst_gen++;
2916         /* Make sure the generation is greater than all the component IDs */
2917         if (fst_gen < snd_gen)
2918                 fst_gen = snd_gen;
2919         else if (fst_gen > snd_gen)
2920                 snd_gen = fst_gen;
2921
2922         /* set the file specific informations in lmm */
2923         if (fst_lmm != NULL) {
2924                 struct ost_id temp_oi;
2925
2926                 saved_oi = &info->mdi_oa.o_oi;
2927                 mdd_get_lmm_oi(fst_lmm, saved_oi);
2928                 mdd_get_lmm_oi(snd_lmm, &temp_oi);
2929                 mdd_set_lmm_gen(fst_lmm, &snd_gen);
2930                 mdd_set_lmm_oi(fst_lmm, &temp_oi);
2931                 mdd_set_lmm_oi(snd_lmm, saved_oi);
2932         } else {
2933                 if ((snd_lmm->lmm_magic & cpu_to_le32(LOV_MAGIC_MASK)) ==
2934                     cpu_to_le32(LOV_MAGIC_MAGIC))
2935                         snd_lmm->lmm_magic |= cpu_to_le32(LOV_MAGIC_DEFINED);
2936                 else
2937                         GOTO(stop, rc = -EPROTO);
2938         }
2939         mdd_set_lmm_gen(snd_lmm, &fst_gen);
2940
2941         fst_info.o = fst_o;
2942         fst_info.hsm_buf = fst_hsm_buf;
2943         fst_info.dv = dv1;
2944
2945         snd_info.o = snd_o;
2946         snd_info.hsm_buf = snd_hsm_buf;
2947         snd_info.dv = dv2;
2948         rc = swap_layouts_prepare_hsm_attr(env, mdd, handle, &fst_info,
2949                                            &snd_info);
2950         if (rc)
2951                 GOTO(stop, rc);
2952
2953         /* prepare transaction */
2954         rc = mdd_declare_xattr_set(env, mdd, fst_o, snd_buf, XATTR_NAME_LOV,
2955                                    fst_fl, handle);
2956         if (rc != 0)
2957                 GOTO(stop, rc);
2958
2959         if (fst_buf->lb_buf != NULL)
2960                 rc = mdd_declare_xattr_set(env, mdd, snd_o, fst_buf,
2961                                            XATTR_NAME_LOV, LU_XATTR_REPLACE,
2962                                            handle);
2963         else
2964                 rc = mdd_declare_xattr_del(env, mdd, snd_o, XATTR_NAME_LOV,
2965                                            handle);
2966         if (rc != 0)
2967                 GOTO(stop, rc);
2968
2969         rc = mdd_trans_start(env, mdd, handle);
2970         if (rc != 0)
2971                 GOTO(stop, rc);
2972
2973         rc = mdd_write_lock_two_objects(env, fst_o, snd_o);
2974         if (rc < 0)
2975                 GOTO(stop, rc);
2976
2977         if (!mdd_object_exists(fst_o))
2978                 GOTO(unlock, rc = -ENOENT);
2979         if (!mdd_object_exists(snd_o))
2980                 GOTO(unlock, rc = -ENOENT);
2981
2982         if (fst_info.xattr_flags) {
2983                 rc = mdo_xattr_set(env, fst_o, snd_info.hsm_buf,
2984                                    XATTR_NAME_HSM, fst_info.xattr_flags,
2985                                    handle);
2986                 if (rc < 0)
2987                         GOTO(unlock, rc);
2988         }
2989
2990         if (snd_info.xattr_flags) {
2991                 rc = mdo_xattr_set(env, snd_o, fst_info.hsm_buf,
2992                                    XATTR_NAME_HSM, snd_info.xattr_flags,
2993                                    handle);
2994                 if (rc < 0)
2995                         GOTO(out_restore_hsm_fst, rc);
2996         }
2997
2998         rc = mdo_xattr_set(env, fst_o, snd_buf, XATTR_NAME_LOV, fst_fl, handle);
2999         if (rc != 0)
3000                 GOTO(out_restore_hsm, rc);
3001
3002         if (unlikely(CFS_FAIL_CHECK(OBD_FAIL_MDS_HSM_SWAP_LAYOUTS))) {
3003                 rc = -EOPNOTSUPP;
3004         } else {
3005                 if (fst_buf->lb_buf != NULL)
3006                         rc = mdo_xattr_set(env, snd_o, fst_buf, XATTR_NAME_LOV,
3007                                            LU_XATTR_REPLACE, handle);
3008                 else
3009                         rc = mdo_xattr_del(env, snd_o, XATTR_NAME_LOV, handle);
3010         }
3011         if (rc != 0)
3012                 GOTO(out_restore, rc);
3013
3014         /* Issue one changelog record per file */
3015         rc = emit_changelog_after_swap_layout(env, handle, fst_o, fst_hsm_buf);
3016         if (rc)
3017                 GOTO(unlock, rc);
3018
3019         rc = emit_changelog_after_swap_layout(env, handle, snd_o, snd_hsm_buf);
3020         if (rc)
3021                 GOTO(unlock, rc);
3022         EXIT;
3023
3024 out_restore:
3025         if (rc != 0) {
3026                 /* failure on second file, but first was done, so we have
3027                  * to roll back first.
3028                  */
3029                 if (fst_buf->lb_buf != NULL) {
3030                         mdd_set_lmm_oi(fst_lmm, saved_oi);
3031                         mdd_set_lmm_gen(fst_lmm, &saved_gen);
3032                         rc2 = mdo_xattr_set(env, fst_o, fst_buf, XATTR_NAME_LOV,
3033                                             LU_XATTR_REPLACE, handle);
3034                 } else {
3035                         rc2 = mdo_xattr_del(env, fst_o, XATTR_NAME_LOV, handle);
3036                 }
3037                 if (rc2 < 0)
3038                         goto do_lbug;
3039
3040 out_restore_hsm:
3041                 if (snd_info.xattr_flags) {
3042                         /* roll back swap HSM */
3043                         steps = 1;
3044                         rc2 = mdo_xattr_set(env, snd_o, snd_hsm_buf,
3045                                             XATTR_NAME_HSM, LU_XATTR_REPLACE,
3046                                             handle);
3047                         if (rc2 < 0)
3048                                 goto do_lbug;
3049                 }
3050
3051 out_restore_hsm_fst:
3052                 if (fst_info.xattr_flags) {
3053                         steps = 2;
3054                         rc2 = mdo_xattr_set(env, fst_o, fst_hsm_buf,
3055                                             XATTR_NAME_HSM, LU_XATTR_REPLACE,
3056                                             handle);
3057                 }
3058
3059 do_lbug:
3060                 /* very bad day - a solution to avoid journal commit
3061                  * is to panic, but it has strong consequences so we
3062                  * use LASSERT to allow sysdamin to choose to panic
3063                  * or not
3064                  */
3065                 LASSERTF(rc2 >= 0,
3066                          "%s: unable to roll back layout swap of " DFID " and " DFID ", steps: %d: rc = %d/%d\n",
3067                          mdd_obj_dev_name(fst_o), PFID(mdd_object_fid(snd_o)),
3068                          PFID(mdd_object_fid(fst_o)), rc, rc2, steps);
3069         }
3070
3071 unlock:
3072         mdd_write_unlock_two_objects(env, fst_o, snd_o);
3073
3074 stop:
3075         rc = mdd_trans_stop(env, mdd, rc, handle);
3076
3077         /* Truncate local DOM data if all went well, except migration case
3078          * with the same DOM component size. In that case a local data is
3079          * still in use and shouldn't be deleted.
3080          */
3081         if (!rc && dom_o)
3082                 mdd_dom_fixup(env, mdd, dom_o, vlt_o);
3083
3084         lu_buf_free(fst_buf);
3085         lu_buf_free(snd_buf);
3086         lu_buf_free(fst_hsm_buf);
3087         lu_buf_free(snd_hsm_buf);
3088
3089         if (!rc) {
3090                 (void) mdd_object_pfid_replace(env, fst_o);
3091                 (void) mdd_object_pfid_replace(env, snd_o);
3092         } else if (rc == -EAGAIN && retried++ < MAX_TRANS_RETRIED) {
3093                 /**
3094                  * -EAGAIN means transaction execution phase detect the layout
3095                  * has been changed by others.
3096                  */
3097                 GOTO(retry, retried);
3098         }
3099
3100         RETURN(rc);
3101 }
3102
3103 static int mdd_declare_layout_change(const struct lu_env *env,
3104                                      struct mdd_device *mdd,
3105                                      struct mdd_object *obj,
3106                                      struct md_layout_change *mlc,
3107                                      struct thandle *handle)
3108 {
3109         int rc;
3110
3111         rc = mdo_declare_layout_change(env, obj, mlc, handle);
3112         if (rc)
3113                 return rc;
3114
3115         return mdd_declare_changelog_store(env, mdd, CL_LAYOUT, NULL, NULL,
3116                                            handle);
3117 }
3118
3119 /* For PFL, this is used to instantiate necessary component objects. */
3120 static int
3121 mdd_layout_instantiate_component(const struct lu_env *env,
3122                 struct mdd_object *obj, struct md_layout_change *mlc,
3123                 struct thandle *handle)
3124 {
3125         struct mdd_device *mdd = mdd_obj2mdd_dev(obj);
3126         int rc;
3127
3128         ENTRY;
3129
3130         if (mlc->mlc_opc != MD_LAYOUT_WRITE)
3131                 RETURN(-ENOTSUPP);
3132
3133         rc = mdd_declare_layout_change(env, mdd, obj, mlc, handle);
3134         /**
3135          * It's possible that another layout write intent has already
3136          * instantiated our objects, so a -EALREADY returned, and we need to
3137          * do nothing.
3138          */
3139         if (rc)
3140                 RETURN(rc == -EALREADY ? 0 : rc);
3141
3142         rc = mdd_trans_start(env, mdd, handle);
3143         if (rc)
3144                 RETURN(rc);
3145
3146         mdd_write_lock(env, obj, DT_TGT_CHILD);
3147         rc = mdo_layout_change(env, obj, mlc, handle);
3148         mdd_write_unlock(env, obj);
3149         if (rc)
3150                 RETURN(rc);
3151
3152         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, obj, handle,
3153                                       NULL);
3154         RETURN(rc);
3155 }
3156
3157 /**
3158  * Change the FLR layout from RDONLY to WRITE_PENDING.
3159  *
3160  * It picks the primary mirror, and bumps the layout version, and set
3161  * layout version xattr to OST objects in a sync tx. In order to facilitate
3162  * the handling of phantom writers from evicted clients, the clients carry
3163  * layout version of the file with write RPC, so that the OSTs can verify
3164  * if the write RPCs are legitimate, meaning not from evicted clients.
3165  */
3166 static int
3167 mdd_layout_update_rdonly(const struct lu_env *env, struct mdd_object *obj,
3168                          struct md_layout_change *mlc, struct thandle *handle)
3169 {
3170         struct mdd_device *mdd = mdd_obj2mdd_dev(obj);
3171         struct lu_buf *som_buf = &mdd_env_info(env)->mdi_buf[1];
3172         struct lustre_som_attrs *som = &mlc->mlc_som;
3173         int fl = 0;
3174         int rc;
3175
3176         ENTRY;
3177
3178         /* Verify acceptable operations */
3179         switch (mlc->mlc_opc) {
3180         case MD_LAYOUT_WRITE:
3181         case MD_LAYOUT_RESYNC:
3182                 /* these are legal operations - this represents the case that
3183                  * a few mirrors were missed in the last resync.
3184                  */
3185                 break;
3186         case MD_LAYOUT_RESYNC_DONE:
3187         default:
3188                 RETURN(0);
3189         }
3190
3191         som_buf->lb_buf = som;
3192         som_buf->lb_len = sizeof(*som);
3193         rc = mdo_xattr_get(env, obj, som_buf, XATTR_NAME_SOM);
3194         if (rc < 0 && rc != -ENODATA)
3195                 RETURN(rc);
3196
3197         if (rc > 0) {
3198                 lustre_som_swab(som);
3199                 if (som->lsa_valid & SOM_FL_STRICT)
3200                         fl = LU_XATTR_REPLACE;
3201
3202                 if (mlc->mlc_opc == MD_LAYOUT_WRITE &&
3203                     mlc->mlc_intent->lai_extent.e_end > som->lsa_size) {
3204                         som->lsa_size = mlc->mlc_intent->lai_extent.e_end + 1;
3205                         fl = LU_XATTR_REPLACE;
3206                 }
3207         }
3208
3209         rc = mdd_declare_layout_change(env, mdd, obj, mlc, handle);
3210         if (rc)
3211                 GOTO(out, rc);
3212
3213         if (fl) {
3214                 rc = mdd_declare_xattr_set(env, mdd, obj, som_buf,
3215                                            XATTR_NAME_SOM, fl, handle);
3216                 if (rc)
3217                         GOTO(out, rc);
3218         }
3219
3220         /* record a changelog for data mover to consume */
3221         rc = mdd_declare_changelog_store(env, mdd, CL_FLRW, NULL, NULL, handle);
3222         if (rc)
3223                 GOTO(out, rc);
3224
3225         rc = mdd_trans_start(env, mdd, handle);
3226         if (rc)
3227                 GOTO(out, rc);
3228
3229         /* it needs a sync tx to make FLR to work properly */
3230         handle->th_sync = 1;
3231
3232         mdd_write_lock(env, obj, DT_TGT_CHILD);
3233         rc = mdo_layout_change(env, obj, mlc, handle);
3234         if (!rc && fl) {
3235                 /* SOM state transition from STRICT to STALE */
3236                 som->lsa_valid = SOM_FL_STALE;
3237                 lustre_som_swab(som);
3238                 rc = mdo_xattr_set(env, obj, som_buf, XATTR_NAME_SOM,
3239                                    fl, handle);
3240         }
3241         mdd_write_unlock(env, obj);
3242         if (rc)
3243                 GOTO(out, rc);
3244
3245         rc = mdd_changelog_data_store(env, mdd, CL_FLRW, 0, obj, handle, NULL);
3246
3247 out:
3248         RETURN(rc);
3249 }
3250
3251 /**
3252  * Handle mirrored file state transition when it's in WRITE_PENDING.
3253  *
3254  * Only MD_LAYOUT_RESYNC, which represents start of resync, is allowed when
3255  * the file is in WRITE_PENDING state. If everything goes fine, the file's
3256  * layout version will be increased, and the file's state will be changed to
3257  * SYNC_PENDING.
3258  */
3259 static int
3260 mdd_layout_update_write_pending(const struct lu_env *env,
3261                 struct mdd_object *obj, struct md_layout_change *mlc,
3262                 struct thandle *handle)
3263 {
3264         struct mdd_device *mdd = mdd_obj2mdd_dev(obj);
3265         struct lu_buf *som_buf = &mdd_env_info(env)->mdi_buf[1];
3266         struct lustre_som_attrs *som = &mlc->mlc_som;
3267         int fl = 0;
3268         int rc;
3269
3270         ENTRY;
3271
3272         switch (mlc->mlc_opc) {
3273         case MD_LAYOUT_RESYNC:
3274                 /* Upon receiving the resync request, it should
3275                  * instantiate all stale components right away to get ready
3276                  * for mirror copy. In order to avoid layout version change,
3277                  * client should avoid sending LAYOUT_WRITE request at the
3278                  * resync state.
3279                  */
3280                 break;
3281         case MD_LAYOUT_WRITE:
3282                 /**
3283                  * legal race for concurrent write, the file state has been
3284                  * changed by another client. Or a jump over file size and
3285                  * write.
3286                  */
3287                 som_buf->lb_buf = som;
3288                 som_buf->lb_len = sizeof(*som);
3289                 rc = mdo_xattr_get(env, obj, som_buf, XATTR_NAME_SOM);
3290                 if (rc < 0 && rc != -ENODATA)
3291                         RETURN(rc);
3292
3293                 if (rc > 0) {
3294                         lustre_som_swab(som);
3295                         if (mlc->mlc_intent->lai_extent.e_end > som->lsa_size) {
3296                                 som->lsa_size =
3297                                         mlc->mlc_intent->lai_extent.e_end + 1;
3298                                 fl = LU_XATTR_REPLACE;
3299                         }
3300                 }
3301                 break;
3302         default:
3303                 RETURN(-EBUSY);
3304         }
3305
3306         rc = mdd_declare_layout_change(env, mdd, obj, mlc, handle);
3307         if (rc)
3308                 GOTO(out, rc);
3309
3310         if (fl) {
3311                 rc = mdd_declare_xattr_set(env, mdd, obj, som_buf,
3312                                            XATTR_NAME_SOM, fl, handle);
3313                 if (rc)
3314                         GOTO(out, rc);
3315         }
3316
3317         rc = mdd_trans_start(env, mdd, handle);
3318         if (rc)
3319                 GOTO(out, rc);
3320
3321         /* it needs a sync tx to make FLR to work properly */
3322         handle->th_sync = 1;
3323
3324         mdd_write_lock(env, obj, DT_TGT_CHILD);
3325         rc = mdo_layout_change(env, obj, mlc, handle);
3326         if (!rc && fl) {
3327                 som->lsa_valid = SOM_FL_STALE;
3328                 lustre_som_swab(som);
3329                 rc = mdo_xattr_set(env, obj, som_buf, XATTR_NAME_SOM,
3330                                    fl, handle);
3331         }
3332         mdd_write_unlock(env, obj);
3333
3334 out:
3335         RETURN(rc);
3336 }
3337
3338 /**
3339  * Handle the requests when a FLR file's state is in SYNC_PENDING.
3340  *
3341  * Only concurrent write and sync complete requests are possible when the
3342  * file is in SYNC_PENDING. For the latter request, it will pass in the
3343  * mirrors that have been synchronized, then the stale bit will be cleared
3344  * to make the file's state turn into RDONLY.
3345  * For concurrent write reqeust, it just needs to change the file's state
3346  * to WRITE_PENDING in a sync tx. It doesn't have to change the layout
3347  * version because the version will be increased in the transition to
3348  * SYNC_PENDING later so that it can deny the write request from potential
3349  * evicted SYNC clients.
3350  */
3351 static int
3352 mdd_object_update_sync_pending(const struct lu_env *env, struct mdd_object *obj,
3353                 struct md_layout_change *mlc, struct thandle *handle)
3354 {
3355         struct mdd_device *mdd = mdd_obj2mdd_dev(obj);
3356         struct lu_buf *som_buf = &mdd_env_info(env)->mdi_buf[1];
3357         int fl = 0;
3358         int rc;
3359
3360         ENTRY;
3361
3362         /* operation validation */
3363         switch (mlc->mlc_opc) {
3364         case MD_LAYOUT_RESYNC_DONE:
3365                 /* resync complete. */
3366         case MD_LAYOUT_WRITE:
3367                 /* concurrent write. */
3368                 break;
3369         case MD_LAYOUT_RESYNC:
3370                 /* resync again, most likely the previous run failed.
3371                  * no-op if it's already in SYNC_PENDING state
3372                  */
3373                 RETURN(0);
3374         default:
3375                 RETURN(-EBUSY);
3376         }
3377
3378         if (mlc->mlc_som.lsa_valid & SOM_FL_STRICT) {
3379                 rc = mdo_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_SOM);
3380                 if (rc < 0 && rc != -ENODATA)
3381                         RETURN(rc);
3382
3383                 fl = rc == -ENODATA ? LU_XATTR_CREATE : LU_XATTR_REPLACE;
3384                 lustre_som_swab(&mlc->mlc_som);
3385                 som_buf->lb_buf = &mlc->mlc_som;
3386                 som_buf->lb_len = sizeof(mlc->mlc_som);
3387         }
3388
3389         rc = mdd_declare_layout_change(env, mdd, obj, mlc, handle);
3390         if (rc)
3391                 GOTO(out, rc);
3392
3393         /* record a changelog for the completion of resync */
3394         rc = mdd_declare_changelog_store(env, mdd, CL_RESYNC, NULL, NULL,
3395                                          handle);
3396         if (rc)
3397                 GOTO(out, rc);
3398
3399         /* RESYNC_DONE has piggybacked size and blocks */
3400         if (fl) {
3401                 rc = mdd_declare_xattr_set(env, mdd, obj, som_buf,
3402                                            XATTR_NAME_SOM, fl, handle);
3403                 if (rc)
3404                         GOTO(out, rc);
3405         }
3406
3407         rc = mdd_trans_start(env, mdd, handle);
3408         if (rc)
3409                 GOTO(out, rc);
3410
3411         /* it needs a sync tx to make FLR to work properly */
3412         handle->th_sync = 1;
3413
3414         mdd_write_lock(env, obj, DT_TGT_CHILD);
3415         rc = mdo_layout_change(env, obj, mlc, handle);
3416         mdd_write_unlock(env, obj);
3417         if (rc)
3418                 GOTO(out, rc);
3419
3420         if (fl) {
3421                 rc = mdo_xattr_set(env, obj, som_buf, XATTR_NAME_SOM,
3422                                    fl, handle);
3423                 if (rc)
3424                         GOTO(out, rc);
3425         }
3426
3427         rc = mdd_changelog_data_store(env, mdd, CL_RESYNC, 0, obj, handle,
3428                                       NULL);
3429 out:
3430         RETURN(rc);
3431 }
3432
3433 static int
3434 mdd_layout_check(const struct lu_env *env, struct md_object *o,
3435                        struct md_layout_change *mlc)
3436 {
3437         return mdo_layout_check(env, md2mdd_obj(o), mlc);
3438 }
3439
3440 /**
3441  *  Update the layout for PCC-RO.
3442  */
3443 static int
3444 mdd_layout_update_pccro(const struct lu_env *env, struct md_object *o,
3445                         struct md_layout_change *mlc)
3446 {
3447         struct mdd_object *obj = md2mdd_obj(o);
3448         struct mdd_device *mdd = mdd_obj2mdd_dev(obj);
3449         struct thandle *handle;
3450         int rc;
3451
3452         ENTRY;
3453
3454         handle = mdd_trans_create(env, mdd);
3455         if (IS_ERR(handle))
3456                 RETURN(PTR_ERR(handle));
3457
3458         /* TODO: Set SOM strict correct when the file is PCC-RO cached. */
3459         rc = mdd_declare_layout_change(env, mdd, obj, mlc, handle);
3460         /**
3461          * It is possible that another layout write intent has already
3462          * set/cleared read-only flag on the object, so as to return
3463          * -EALREADY, and we need to do nothing in this case.
3464          */
3465         if (rc)
3466                 GOTO(out, rc == -EALREADY ? rc = 0 : rc);
3467
3468         rc = mdd_trans_start(env, mdd, handle);
3469         if (rc != 0)
3470                 GOTO(out, rc);
3471
3472         mdd_write_lock(env, obj, DT_TGT_CHILD);
3473         rc = mdo_layout_change(env, obj, mlc, handle);
3474         mdd_write_unlock(env, obj);
3475         if (rc != 0)
3476                 GOTO(out, rc);
3477
3478         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, obj, handle,
3479                                       NULL);
3480 out:
3481         mdd_trans_stop(env, mdd, rc, handle);
3482
3483         RETURN(rc);
3484 }
3485
3486 /**
3487  * Layout change callback for object.
3488  *
3489  * This is used by FLR and PCC-RO as well as dir migration
3490  * and restriping.
3491  */
3492 static int
3493 mdd_layout_change(const struct lu_env *env, struct md_object *o,
3494                   struct md_layout_change *mlc)
3495 {
3496         struct mdd_object       *obj = md2mdd_obj(o);
3497         struct mdd_device       *mdd = mdd_obj2mdd_dev(obj);
3498         struct lu_buf           *buf = mdd_buf_get(env, NULL, 0);
3499         struct lov_comp_md_v1   *lcm;
3500         struct thandle          *handle;
3501         int flr_state;
3502         int retried = 0;
3503         int rc;
3504
3505         ENTRY;
3506
3507         if (S_ISDIR(mdd_object_type(obj))) {
3508                 switch (mlc->mlc_opc) {
3509                 case MD_LAYOUT_SHRINK:
3510                         rc = mdd_dir_layout_shrink(env, o, mlc);
3511                         break;
3512                 case MD_LAYOUT_SPLIT:
3513                         rc = mdd_dir_layout_split(env, o, mlc);
3514                         break;
3515                 default:
3516                         LBUG();
3517                 }
3518
3519                 RETURN(rc);
3520         }
3521
3522         /* Verify acceptable operations */
3523         switch (mlc->mlc_opc) {
3524         case MD_LAYOUT_WRITE: {
3525                 struct layout_intent *intent = mlc->mlc_intent;
3526
3527                 if (intent->lai_opc == LAYOUT_INTENT_PCCRO_SET ||
3528                     intent->lai_opc == LAYOUT_INTENT_PCCRO_CLEAR)
3529                         RETURN(mdd_layout_update_pccro(env, o, mlc));
3530                 break;
3531         }
3532         case MD_LAYOUT_RESYNC:
3533         case MD_LAYOUT_RESYNC_DONE:
3534                 break;
3535         default:
3536                 RETURN(-ENOTSUPP);
3537         }
3538
3539 retry:
3540         handle = mdd_trans_create(env, mdd);
3541         if (IS_ERR(handle))
3542                 RETURN(PTR_ERR(handle));
3543
3544         rc = mdd_stripe_get(env, obj, buf, XATTR_NAME_LOV);
3545         if (rc < 0) {
3546                 if (rc == -ENODATA)
3547                         rc = -EINVAL;
3548                 GOTO(out, rc);
3549         }
3550
3551         /* analyze the layout to make sure it's a FLR file */
3552         lcm = buf->lb_buf;
3553         if (le32_to_cpu(lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
3554                 GOTO(out, rc = -EINVAL);
3555
3556         flr_state = le16_to_cpu(lcm->lcm_flags) & LCM_FL_FLR_MASK;
3557
3558         /* please refer to HLD of FLR for state transition */
3559         switch (flr_state) {
3560         case LCM_FL_NONE:
3561                 rc = mdd_layout_instantiate_component(env, obj, mlc, handle);
3562                 break;
3563         case LCM_FL_WRITE_PENDING:
3564                 rc = mdd_layout_update_write_pending(env, obj, mlc, handle);
3565                 break;
3566         case LCM_FL_RDONLY:
3567                 rc = mdd_layout_update_rdonly(env, obj, mlc, handle);
3568                 break;
3569         case LCM_FL_SYNC_PENDING:
3570                 rc = mdd_object_update_sync_pending(env, obj, mlc, handle);
3571                 break;
3572         default:
3573                 rc = 0;
3574                 break;
3575         }
3576         EXIT;
3577
3578 out:
3579         mdd_trans_stop(env, mdd, rc, handle);
3580         lu_buf_free(buf);
3581
3582         /**
3583          * -EAGAIN means transaction execution phase detect the layout
3584          * has been changed by others.
3585          */
3586         if (rc == -EAGAIN && retried++ < MAX_TRANS_RETRIED)
3587                 GOTO(retry, retried);
3588
3589         return rc;
3590 }
3591
3592 void mdd_object_make_hint(const struct lu_env *env, struct mdd_object *parent,
3593                           struct mdd_object *child, const struct lu_attr *attr,
3594                           const struct md_op_spec *spec,
3595                           struct dt_allocation_hint *hint)
3596 {
3597         struct dt_object *np = parent ?  mdd_object_child(parent) : NULL;
3598         struct mdd_device *mdd = mdd_obj2mdd_dev(child);
3599         struct dt_object *nc = mdd_object_child(child);
3600
3601         memset(hint, 0, sizeof(*hint));
3602
3603         /* For striped directory, give striping EA to lod_ah_init, which will
3604          * decide the stripe_offset and stripe count by it.
3605          */
3606         if (S_ISDIR(attr->la_mode) && spec) {
3607                 if (unlikely(spec->sp_cr_flags & MDS_OPEN_HAS_EA)) {
3608                         hint->dah_eadata = spec->u.sp_ea.eadata;
3609                         hint->dah_eadata_len = spec->u.sp_ea.eadatalen;
3610                         if (spec->sp_cr_flags & MDS_OPEN_DEFAULT_LMV)
3611                                 hint->dah_eadata_is_dmv = 1;
3612                 }
3613                 hint->dah_dmv_imp_inherit = spec->sp_dmv_imp_inherit;
3614         } else if (S_ISREG(attr->la_mode) &&
3615                    spec->sp_cr_flags & MDS_OPEN_APPEND) {
3616                 hint->dah_append_stripe_count = mdd->mdd_append_stripe_count;
3617                 hint->dah_append_pool = mdd->mdd_append_pool;
3618         }
3619
3620         CDEBUG(D_INFO, DFID" eadata %p len %d\n", PFID(mdd_object_fid(child)),
3621                hint->dah_eadata, hint->dah_eadata_len);
3622         /* @hint will be initialized by underlying device. */
3623         nc->do_ops->do_ah_init(env, hint, np, nc, attr->la_mode & S_IFMT);
3624 }
3625
3626 static int mdd_accmode(const struct lu_env *env, const struct lu_attr *la,
3627                        enum mds_open_flags open_flags)
3628 {
3629         /* Sadly, NFSD reopens a file repeatedly during operation, so the
3630          * "acc_mode = 0" allowance for newly-created files isn't honoured.
3631          * NFSD uses the MDS_OPEN_OWNEROVERRIDE flag to say that a file
3632          * owner can write to a file even if it is marked readonly to hide
3633          * its brokenness. (bug 5781)
3634          */
3635         if (open_flags & MDS_OPEN_OWNEROVERRIDE) {
3636                 struct lu_ucred *uc = lu_ucred_check(env);
3637
3638                 if ((uc == NULL) || (la->la_uid == uc->uc_fsuid))
3639                         return 0;
3640         }
3641
3642         return mds_accmode(open_flags);
3643 }
3644
3645 /**
3646  * mdd_open_sanity_check() - Check if mdd object is valid(not unlinked)
3647  *
3648  * @env: execution environment for this thread
3649  * @obj: metadata object
3650  * @attr: common attribute of the object
3651  * @open_flags: MDS flags passed from client
3652  * @is_replay: Additional params passed to open/create
3653  *
3654  * Return:
3655  * * O on success
3656  * * negative errno on failure
3657  */
3658 static int mdd_open_sanity_check(const struct lu_env *env,
3659                                  struct mdd_object *obj,
3660                                  const struct lu_attr *attr,
3661                                  enum mds_open_flags open_flags, int is_replay)
3662 {
3663         unsigned int may_mask;
3664         int rc;
3665
3666         ENTRY;
3667
3668         /* EEXIST check, also opening of *open* orphans is allowed so we can
3669          * open-by-handle unlinked files
3670          */
3671         if (mdd_is_dead_obj(obj) && !is_replay &&
3672             likely(!(mdd_is_orphan_obj(obj) && obj->mod_count > 0)))
3673                 RETURN(-ENOENT);
3674
3675         if (S_ISLNK(attr->la_mode))
3676                 RETURN(-ELOOP);
3677
3678         may_mask = mdd_accmode(env, attr, open_flags);
3679
3680         if (S_ISDIR(attr->la_mode) && (may_mask & MAY_WRITE))
3681                 RETURN(-EISDIR);
3682
3683         if (!(open_flags & MDS_OPEN_CREATED)) {
3684                 rc = mdd_permission_internal(env, obj, attr, may_mask);
3685                 if (rc)
3686                         RETURN(rc);
3687         }
3688
3689         if (S_ISFIFO(attr->la_mode) || S_ISSOCK(attr->la_mode) ||
3690             S_ISBLK(attr->la_mode) || S_ISCHR(attr->la_mode))
3691                 open_flags &= ~MDS_OPEN_TRUNC;
3692
3693         /* For writing append-only file must open it with append mode. */
3694         if (attr->la_flags & LUSTRE_APPEND_FL) {
3695                 if ((open_flags & MDS_FMODE_WRITE) &&
3696                     !(open_flags & MDS_OPEN_APPEND))
3697                         RETURN(-EPERM);
3698                 if (open_flags & MDS_OPEN_TRUNC)
3699                         RETURN(-EPERM);
3700         }
3701
3702         RETURN(0);
3703 }
3704
3705 /**
3706  * mdd_open() - Called when object under metadata is opened
3707  *
3708  * @env: execution environment for this thread
3709  * @obj: metadata object
3710  * @open_flags: MDS flags passed from client
3711  * @spec: Additional params passed to open/create
3712  *
3713  * Return:
3714  * * 0 on Success
3715  * * <0 on Failure
3716  */
3717 static int mdd_open(const struct lu_env *env, struct md_object *obj,
3718                     enum mds_open_flags open_flags, struct md_op_spec *spec)
3719 {
3720         struct mdd_object *mdd_obj = md2mdd_obj(obj);
3721         struct md_device *md_dev = lu2md_dev(mdd2lu_dev(mdo2mdd(obj)));
3722         struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
3723         struct mdd_object_user *mou = NULL;
3724         const struct lu_ucred *uc = lu_ucred(env);
3725         struct mdd_device *mdd = mdo2mdd(obj);
3726         enum changelog_rec_type type = CL_OPEN;
3727         int rc = 0;
3728
3729         ENTRY;
3730
3731         mdd_write_lock(env, mdd_obj, DT_TGT_CHILD);
3732
3733         rc = mdd_la_get(env, mdd_obj, attr);
3734         if (rc != 0)
3735                 GOTO(out, rc);
3736
3737         rc = mdd_open_sanity_check(env, mdd_obj, attr, open_flags,
3738                                    spec->no_create);
3739         if ((rc == -EACCES) && (mdd->mdd_cl.mc_current_mask & BIT(CL_DN_OPEN)))
3740                 type = CL_DN_OPEN;
3741         else if (rc != 0)
3742                 GOTO(out, rc);
3743         else
3744                 mdd_obj->mod_count++;
3745
3746         if (!mdd_changelog_enabled(env, mdd, type))
3747                 GOTO(out, rc);
3748
3749 find:
3750         /* look for existing opener in list under mdd_write_lock */
3751         mou = mdd_obj_user_find(mdd_obj, uc->uc_uid, uc->uc_gid, open_flags);
3752
3753         if (!mou) {
3754                 int rc2;
3755
3756                 /* add user to list */
3757                 mou = mdd_obj_user_alloc(open_flags, uc->uc_uid, uc->uc_gid);
3758                 if (IS_ERR(mou)) {
3759                         if (rc == 0)
3760                                 rc = PTR_ERR(mou);
3761                         GOTO(out, rc);
3762                 }
3763                 rc2 = mdd_obj_user_add(mdd_obj, mou, type == CL_DN_OPEN);
3764                 if (rc2 != 0) {
3765                         mdd_obj_user_free(mou);
3766                         if (rc2 == -EEXIST)
3767                                 GOTO(find, rc2);
3768                 }
3769         } else {
3770                 if (type == CL_DN_OPEN) {
3771                         if (ktime_before(ktime_get(), mou->mou_deniednext))
3772                                 /* same user denied again same access within
3773                                  * time interval: do not record
3774                                  */
3775                                 GOTO(out, rc);
3776
3777                         /* this user already denied, but some time ago:
3778                          * update denied time
3779                          */
3780                         mou->mou_deniednext =
3781                                 ktime_add(ktime_get(),
3782                                           ktime_set(mdd->mdd_cl.mc_deniednext,
3783                                                     0));
3784                 } else {
3785                         mou->mou_opencount++;
3786                         /* same user opening file again with same flags:
3787                          * don't record
3788                          */
3789                         GOTO(out, rc);
3790                 }
3791         }
3792
3793         /* we can't hold object lock over transaction start
3794          * and we don't actually need the object to be locked
3795          */
3796         mdd_write_unlock(env, mdd_obj);
3797
3798         /* FYI, only the bottom 32 bits of open_flags are recorded */
3799         /* only the low CLF_FLAGMASK bits of la_valid are stored*/
3800         mdd_changelog(env, type, open_flags & CLF_FLAGMASK, md_dev,
3801                       mdd_object_fid(mdd_obj));
3802
3803         GOTO(unlocked, rc);
3804
3805 out:
3806         mdd_write_unlock(env, mdd_obj);
3807 unlocked:
3808         return rc;
3809 }
3810
3811 static int mdd_declare_close(const struct lu_env *env, struct mdd_object *obj,
3812                              struct md_attr *ma, struct thandle *handle)
3813 {
3814         int rc;
3815
3816         rc = mdd_orphan_declare_delete(env, obj, handle);
3817         if (rc)
3818                 return rc;
3819
3820         return mdo_declare_destroy(env, obj, handle);
3821 }
3822
3823 /* No permission check is needed.  */
3824 static int mdd_close(const struct lu_env *env, struct md_object *obj,
3825                      struct md_attr *ma, u64 open_flags)
3826 {
3827         struct mdd_object *mdd_obj = md2mdd_obj(obj);
3828         struct mdd_device *mdd = mdo2mdd(obj);
3829         struct thandle *handle = NULL;
3830         int is_orphan = 0;
3831         int rc;
3832         bool blocked = false;
3833         bool last_close_by_uid = false;
3834         const struct lu_ucred *uc = lu_ucred(env);
3835
3836         ENTRY;
3837
3838         if (ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_KEEP_ORPHAN) {
3839                 mdd_write_lock(env, mdd_obj, DT_TGT_CHILD);
3840                 mdd_obj->mod_count--;
3841                 mdd_write_unlock(env, mdd_obj);
3842
3843                 if (mdd_obj->mod_flags & ORPHAN_OBJ && !mdd_obj->mod_count)
3844                         CDEBUG(D_HA, "Object "DFID" is retained in orphan list\n",
3845                                 PFID(mdd_object_fid(mdd_obj)));
3846                 RETURN(0);
3847         }
3848
3849         /* mdd_finish_unlink() will always set orphan object as DEAD_OBJ, but
3850          * it might fail to add the object to orphan list (w/o ORPHAN_OBJ).
3851          */
3852         /* check without any lock */
3853         is_orphan = mdd_obj->mod_count == 1 &&
3854                     (mdd_obj->mod_flags & (ORPHAN_OBJ | DEAD_OBJ)) != 0;
3855
3856 again:
3857         if (is_orphan) {
3858                 /* mdd_trans_create() maybe failed because of barrier_entry(),
3859                  * under such case, the orphan MDT-object will be left in the
3860                  * orphan list, and when the MDT remount next time, the unused
3861                  * orphans will be destroyed automatically.
3862                  *
3863                  * One exception: the former mdd_finish_unlink may failed to
3864                  * add the orphan MDT-object to the orphan list, then if the
3865                  * mdd_trans_create() failed because of barrier_entry(), the
3866                  * MDT-object will become real orphan that is neither in the
3867                  * namespace nor in the orphan list. Such bad case should be
3868                  * very rare and will be handled by e2fsck/lfsck.
3869                  */
3870                 handle = mdd_trans_create(env, mdo2mdd(obj));
3871                 if (IS_ERR(handle)) {
3872                         rc = PTR_ERR(handle);
3873                         if (rc != -EINPROGRESS)
3874                                 GOTO(stop, rc);
3875
3876                         handle = NULL;
3877                         blocked = true;
3878                         goto cont;
3879                 }
3880
3881                 rc = mdd_declare_close(env, mdd_obj, ma, handle);
3882                 if (rc)
3883                         GOTO(stop, rc);
3884
3885                 rc = mdd_declare_changelog_store(env, mdd, CL_CLOSE, NULL, NULL,
3886                                                  handle);
3887                 if (rc)
3888                         GOTO(stop, rc);
3889
3890                 rc = mdd_trans_start(env, mdo2mdd(obj), handle);
3891                 if (rc)
3892                         GOTO(stop, rc);
3893         }
3894
3895 cont:
3896         mdd_write_lock(env, mdd_obj, DT_TGT_CHILD);
3897         rc = mdd_la_get(env, mdd_obj, &ma->ma_attr);
3898         if (rc != 0) {
3899                 CERROR("%s: failed to get lu_attr of "DFID": rc = %d\n",
3900                        lu_dev_name(mdd2lu_dev(mdd)),
3901                        PFID(mdd_object_fid(mdd_obj)), rc);
3902                 GOTO(out, rc);
3903         }
3904
3905         /* check again with lock */
3906         is_orphan = (mdd_obj->mod_count == 1) &&
3907                     ((mdd_obj->mod_flags & (ORPHAN_OBJ | DEAD_OBJ)) != 0 ||
3908                      ma->ma_attr.la_nlink == 0);
3909
3910         if (is_orphan && !handle && !blocked) {
3911                 mdd_write_unlock(env, mdd_obj);
3912                 goto again;
3913         }
3914
3915         mdd_obj->mod_count--; /*release open count */
3916
3917         /* under mdd write lock */
3918         /* If recording, see if we need to remove UID from list. uc is not
3919          * initialized if the client has been evicted.
3920          */
3921         if (mdd_changelog_enabled(env, mdd, CL_OPEN) && uc) {
3922                 struct mdd_object_user *mou;
3923
3924                 /* look for UID in list */
3925                 /* If mou is NULL, it probably means logging was enabled after
3926                  * the user had the file open. So the corresponding close
3927                  * will not be logged.
3928                  */
3929                 mou = mdd_obj_user_find(mdd_obj, uc->uc_uid, uc->uc_gid,
3930                                         open_flags);
3931                 if (mou) {
3932                         mou->mou_opencount--;
3933                         if (mou->mou_opencount == 0) {
3934                                 mdd_obj_user_remove(mdd_obj, mou);
3935                                 last_close_by_uid = true;
3936                         }
3937                 }
3938         }
3939
3940         if (!is_orphan || blocked)
3941                 GOTO(out, rc = 0);
3942
3943         /* Orphan object */
3944         /* NB: Object maybe not in orphan list originally, it is rare case for
3945          * mdd_finish_unlink() failure, in that case, the object doesn't have
3946          * ORPHAN_OBJ flag
3947          */
3948         if ((mdd_obj->mod_flags & ORPHAN_OBJ) != 0) {
3949                 /* remove link to object from orphan index */
3950                 LASSERT(handle != NULL);
3951                 rc = mdd_orphan_delete(env, mdd_obj, handle);
3952                 if (rc != 0) {
3953                         CERROR("%s: unable to delete "DFID" from orphan list: rc = %d\n",
3954                                lu_dev_name(mdd2lu_dev(mdd)),
3955                                PFID(mdd_object_fid(mdd_obj)), rc);
3956                         /* If object was not deleted from orphan list, do not
3957                          * destroy OSS objects, which will be done when next
3958                          * recovery.
3959                          */
3960                         GOTO(out, rc);
3961                 }
3962
3963                 CDEBUG(D_HA, "Object "DFID" is deleted from orphan list, OSS objects to be destroyed.\n",
3964                        PFID(mdd_object_fid(mdd_obj)));
3965         }
3966
3967         rc = mdo_destroy(env, mdd_obj, handle);
3968
3969         if (rc != 0) {
3970                 CERROR("%s: unable to delete "DFID" from orphan list: rc = %d\n",
3971                        lu_dev_name(mdd2lu_dev(mdd)),
3972                        PFID(mdd_object_fid(mdd_obj)), rc);
3973         }
3974         EXIT;
3975
3976 out:
3977         mdd_write_unlock(env, mdd_obj);
3978
3979         if (rc != 0 || blocked ||
3980             !mdd_changelog_enabled(env, mdd, CL_CLOSE))
3981                 GOTO(stop, rc);
3982
3983         /* Record CL_CLOSE in changelog only if file was opened in write mode,
3984          * or if CL_OPEN was recorded and it's last close by user.
3985          * Changelogs mask may change between open and close operations, but
3986          * this is not a big deal if we have a CL_CLOSE entry with no matching
3987          * CL_OPEN. Plus Changelogs mask may not change often.
3988          */
3989         if (((!(mdd->mdd_cl.mc_current_mask & BIT(CL_OPEN)) &&
3990               (open_flags & (MDS_FMODE_WRITE | MDS_OPEN_APPEND |
3991                              MDS_OPEN_TRUNC))) ||
3992              ((mdd->mdd_cl.mc_current_mask & BIT(CL_OPEN)) &&
3993               last_close_by_uid)) &&
3994             !(ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_RECOV_OPEN)) {
3995                 if (handle == NULL) {
3996                         handle = mdd_trans_create(env, mdo2mdd(obj));
3997                         if (IS_ERR(handle))
3998                                 GOTO(stop, rc = PTR_ERR(handle));
3999
4000                         rc = mdd_declare_changelog_store(env, mdd, CL_CLOSE,
4001                                                          NULL, NULL, handle);
4002                         if (rc)
4003                                 GOTO(stop, rc);
4004
4005                         rc = mdd_trans_start(env, mdo2mdd(obj), handle);
4006                         if (rc)
4007                                 GOTO(stop, rc);
4008                 }
4009
4010                 /* FYI, only the bottom 32 bits of open_flags are recorded */
4011                 mdd_changelog_data_store(env, mdd, CL_CLOSE, open_flags,
4012                                          mdd_obj, handle, NULL);
4013         }
4014
4015 stop:
4016         if (handle != NULL && !IS_ERR(handle))
4017                 rc = mdd_trans_stop(env, mdd, rc, handle);
4018
4019         return rc;
4020 }
4021
4022 /* Permission check is done when open, no need check again.  */
4023 static int mdd_readpage_sanity_check(const struct lu_env *env,
4024                                      struct mdd_object *obj)
4025 {
4026         if (!dt_try_as_dir(env, mdd_object_child(obj), true))
4027                 return -ENOTDIR;
4028
4029         return 0;
4030 }
4031
4032 static int mdd_dir_page_build(const struct lu_env *env, struct dt_object *obj,
4033                               union lu_page *lp, size_t bytes,
4034                               const struct dt_it_ops *iops,
4035                               struct dt_it *it, __u32 attr, void *arg)
4036 {
4037         struct lu_dirpage *dp = &lp->lp_dir;
4038         void *area = dp;
4039         __u64 hash = 0;
4040         struct lu_dirent *ent;
4041         struct lu_dirent *last = NULL;
4042         struct lu_fid fid;
4043         int first = 1;
4044         int result;
4045
4046         if (bytes < sizeof(*dp))
4047                 GOTO(out_err, result = -EOVERFLOW);
4048
4049         memset(area, 0, sizeof(*dp));
4050         area += sizeof(*dp);
4051         bytes -= sizeof(*dp);
4052
4053         ent = area;
4054         do {
4055                 int len;
4056                 size_t recsize;
4057
4058                 len = iops->key_size(env, it);
4059
4060                 /* IAM iterator can return record with zero len. */
4061                 if (len == 0)
4062                         GOTO(next, 0);
4063
4064                 hash = iops->store(env, it);
4065                 if (unlikely(first)) {
4066                         first = 0;
4067                         dp->ldp_hash_start = cpu_to_le64(hash);
4068                 }
4069
4070                 /* calculate max space required for lu_dirent */
4071                 recsize = lu_dirent_calc_size(len, attr);
4072
4073                 if (bytes >= recsize &&
4074                     !CFS_FAIL_CHECK(OBD_FAIL_MDS_DIR_PAGE_WALK)) {
4075                         result = iops->rec(env, it, (struct dt_rec *)ent, attr);
4076                         if (result == -ESTALE)
4077                                 GOTO(next, result);
4078                         if (result != 0)
4079                                 GOTO(out, result);
4080
4081                         /* OSD might not able to pack all attributes, so
4082                          * recheck record length had room to store FID
4083                          */
4084                         recsize = le16_to_cpu(ent->lde_reclen);
4085
4086                         if (le32_to_cpu(ent->lde_attrs) & LUDA_FID) {
4087                                 fid_le_to_cpu(&fid, &ent->lde_fid);
4088                                 if (fid_is_dot_lustre(&fid))
4089                                         GOTO(next, recsize);
4090                         }
4091                 } else {
4092                         result = (last != NULL) ? 0 : -EBADSLT;
4093                         GOTO(out, result);
4094                 }
4095                 last = ent;
4096                 ent = (void *)ent + recsize;
4097                 bytes -= recsize;
4098
4099 next:
4100                 result = iops->next(env, it);
4101                 if (result == -ESTALE)
4102                         GOTO(next, result);
4103         } while (result == 0);
4104
4105 out:
4106         dp->ldp_hash_end = cpu_to_le64(hash);
4107         if (last != NULL) {
4108                 if (last->lde_hash == dp->ldp_hash_end)
4109                         dp->ldp_flags |= cpu_to_le32(LDF_COLLIDE);
4110                 last->lde_reclen = 0; /* end mark */
4111         }
4112 out_err:
4113         if (result > 0) {
4114                 /* end of directory */
4115                 dp->ldp_hash_end = cpu_to_le64(MDS_DIR_END_OFF);
4116                 if (last == NULL && fid_is_dot_lustre(&fid))
4117                         /*
4118                          * .lustre is last in directory and alone in
4119                          * last directory block
4120                          */
4121                         dp->ldp_flags = cpu_to_le32(LDF_EMPTY);
4122         } else if (result < 0) {
4123                 CWARN("%s: build page failed for "DFID": rc = %d\n",
4124                       lu_dev_name(obj->do_lu.lo_dev),
4125                       PFID(lu_object_fid(&obj->do_lu)), result);
4126         }
4127         return result;
4128 }
4129
4130 int mdd_readpage(const struct lu_env *env, struct md_object *obj,
4131                  const struct lu_rdpg *rdpg)
4132 {
4133         struct mdd_object *mdd_obj = md2mdd_obj(obj);
4134         int rc;
4135
4136         ENTRY;
4137
4138         if (mdd_object_exists(mdd_obj) == 0) {
4139                 rc = -ENOENT;
4140                 CERROR("%s: object "DFID" not found: rc = %d\n",
4141                        mdd_obj_dev_name(mdd_obj),
4142                        PFID(mdd_object_fid(mdd_obj)), rc);
4143                 return rc;
4144         }
4145
4146         mdd_read_lock(env, mdd_obj, DT_TGT_CHILD);
4147         rc = mdd_readpage_sanity_check(env, mdd_obj);
4148         if (rc)
4149                 GOTO(out_unlock, rc);
4150
4151         if (mdd_is_dead_obj(mdd_obj)) {
4152                 struct lu_dirpage *dp;
4153
4154                 /*
4155                  * According to POSIX, please do not return any entry to client:
4156                  * even dot and dotdot should not be returned.
4157                  */
4158                 CDEBUG(D_INODE, "readdir from dead object: "DFID"\n",
4159                        PFID(mdd_object_fid(mdd_obj)));
4160
4161                 if (rdpg->rp_count <= 0)
4162                         GOTO(out_unlock, rc = -EFAULT);
4163                 LASSERT(rdpg->rp_pages != NULL);
4164
4165                 dp = (struct lu_dirpage *)rdpg_page_get(rdpg, 0);
4166                 memset(dp, 0, sizeof(struct lu_dirpage));
4167                 dp->ldp_hash_start = cpu_to_le64(rdpg->rp_hash);
4168                 dp->ldp_hash_end   = cpu_to_le64(MDS_DIR_END_OFF);
4169                 dp->ldp_flags = cpu_to_le32(LDF_EMPTY);
4170                 rdpg_page_put(rdpg, 0);
4171                 GOTO(out_unlock, rc = LU_PAGE_SIZE);
4172         }
4173
4174         rc = dt_index_walk(env, mdd_object_child(mdd_obj), rdpg,
4175                            mdd_dir_page_build, NULL);
4176         if (rc >= 0) {
4177                 struct lu_dirpage       *dp;
4178
4179                 dp = (struct lu_dirpage *)rdpg_page_get(rdpg, 0);
4180                 dp->ldp_hash_start = cpu_to_le64(rdpg->rp_hash);
4181                 if (rc == 0) {
4182                         /*
4183                          * No pages were processed, mark this for first page
4184                          * and send back.
4185                          */
4186                         dp->ldp_hash_end = cpu_to_le64(MDS_DIR_END_OFF);
4187                         dp->ldp_flags = cpu_to_le32(LDF_EMPTY);
4188                         rc = min_t(unsigned int, LU_PAGE_SIZE, rdpg->rp_count);
4189                 }
4190                 rdpg_page_put(rdpg, 0);
4191         }
4192
4193         GOTO(out_unlock, rc);
4194 out_unlock:
4195         mdd_read_unlock(env, mdd_obj);
4196         return rc;
4197 }
4198
4199 static int mdd_object_sync(const struct lu_env *env, struct md_object *obj)
4200 {
4201         struct mdd_object *mdd_obj = md2mdd_obj(obj);
4202
4203         if (mdd_object_exists(mdd_obj) == 0) {
4204                 int rc = -ENOENT;
4205
4206                 CERROR("%s: object "DFID" not found: rc = %d\n",
4207                        mdd_obj_dev_name(mdd_obj),
4208                        PFID(mdd_object_fid(mdd_obj)), rc);
4209                 return rc;
4210         }
4211         return dt_object_sync(env, mdd_object_child(mdd_obj),
4212                               0, OBD_OBJECT_EOF);
4213 }
4214
4215 static int mdd_object_lock(const struct lu_env *env,
4216                            struct md_object *obj,
4217                            struct lustre_handle *lh,
4218                            struct ldlm_enqueue_info *einfo,
4219                            union ldlm_policy_data *policy)
4220 {
4221         struct mdd_object *mdd_obj = md2mdd_obj(obj);
4222
4223         return dt_object_lock(env, mdd_object_child(mdd_obj), lh,
4224                               einfo, policy);
4225 }
4226
4227 static int mdd_object_unlock(const struct lu_env *env,
4228                              struct md_object *obj,
4229                              struct ldlm_enqueue_info *einfo,
4230                              union ldlm_policy_data *policy)
4231 {
4232         struct mdd_object *mdd_obj = md2mdd_obj(obj);
4233
4234         return dt_object_unlock(env, mdd_object_child(mdd_obj), einfo, policy);
4235 }
4236
4237 const struct md_object_operations mdd_obj_ops = {
4238         .moo_permission         = mdd_permission,
4239         .moo_attr_get           = mdd_attr_get,
4240         .moo_attr_set           = mdd_attr_set,
4241         .moo_xattr_get          = mdd_xattr_get,
4242         .moo_xattr_set          = mdd_xattr_set,
4243         .moo_xattr_list         = mdd_xattr_list,
4244         .moo_invalidate         = mdd_invalidate,
4245         .moo_xattr_del          = mdd_xattr_del,
4246         .moo_swap_layouts       = mdd_swap_layouts,
4247         .moo_open               = mdd_open,
4248         .moo_close              = mdd_close,
4249         .moo_readpage           = mdd_readpage,
4250         .moo_readlink           = mdd_readlink,
4251         .moo_changelog          = mdd_changelog,
4252         .moo_object_sync        = mdd_object_sync,
4253         .moo_object_lock        = mdd_object_lock,
4254         .moo_object_unlock      = mdd_object_unlock,
4255         .moo_layout_change      = mdd_layout_change,
4256         .moo_layout_check       = mdd_layout_check,
4257 };