lustre/mdd/mdd_object.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * lustre/mdd/mdd_object.c
  33  *
  34  * Lustre Metadata Server (mdd) routines
  35  *
  36  * Author: Wang Di <wangdi@clusterfs.com>
  37  */
  38
  39 #define DEBUG_SUBSYSTEM S_MDS
  40
  41 #include <linux/module.h>
  42 #include <obd.h>
  43 #include <obd_class.h>
  44 #include <obd_support.h>
  45 #include <lprocfs_status.h>
  46 /* fid_be_cpu(), fid_cpu_to_be(). */
  47 #include <lustre_fid.h>
  48 #include <lustre_idmap.h>
  49 #include <uapi/linux/lustre/lustre_param.h>
  50 #include <lustre_mds.h>
  51
  52 #include "mdd_internal.h"
  53
  54 static const struct lu_object_operations mdd_lu_obj_ops;
  55
  56 static int mdd_xattr_get(const struct lu_env *env,
  57                          struct md_object *obj, struct lu_buf *buf,
  58                          const char *name);
  59
  60 static int mdd_changelog_data_store_by_fid(const struct lu_env *env,
  61                                            struct mdd_device *mdd,
  62                                            enum changelog_rec_type type,
  63                                            int flags, const struct lu_fid *fid,
  64                                            const char *xattr_name,
  65                                            struct thandle *handle);
  66
  67 static inline bool has_prefix(const char *str, const char *prefix);
  68
  69 int mdd_la_get(const struct lu_env *env, struct mdd_object *obj,
  70                struct lu_attr *la)
  71 {
  72         int rc;
  73
  74         if (mdd_object_exists(obj) == 0)
  75                 return -ENOENT;
  76
  77         rc = mdo_attr_get(env, obj, la);
  78         if (unlikely(rc != 0)) {
  79                 if (rc == -ENOENT)
  80                         obj->mod_flags |= DEAD_OBJ;
  81                 return rc;
  82         }
  83
  84         if (la->la_valid & LA_FLAGS &&
  85             la->la_flags & LUSTRE_ORPHAN_FL)
  86                 obj->mod_flags |= ORPHAN_OBJ | DEAD_OBJ;
  87
  88         return 0;
  89 }
  90
  91 struct mdd_thread_info *mdd_env_info(const struct lu_env *env)
  92 {
  93         struct mdd_thread_info *info;
  94
  95         lu_env_refill((struct lu_env *)env);
  96         info = lu_context_key_get(&env->le_ctx, &mdd_thread_key);
  97         LASSERT(info != NULL);
  98         return info;
  99 }
 100
 101 struct lu_buf *mdd_buf_get(const struct lu_env *env, void *area, ssize_t len)
 102 {
 103         struct lu_buf *buf;
 104
 105         buf = &mdd_env_info(env)->mti_buf[0];
 106         buf->lb_buf = area;
 107         buf->lb_len = len;
 108         return buf;
 109 }
 110
 111 const struct lu_buf *mdd_buf_get_const(const struct lu_env *env,
 112                                        const void *area, ssize_t len)
 113 {
 114         struct lu_buf *buf;
 115
 116         buf = &mdd_env_info(env)->mti_buf[0];
 117         buf->lb_buf = (void *)area;
 118         buf->lb_len = len;
 119         return buf;
 120 }
 121
 122 struct lu_object *mdd_object_alloc(const struct lu_env *env,
 123                                    const struct lu_object_header *hdr,
 124                                    struct lu_device *d)
 125 {
 126         struct mdd_object *mdd_obj;
 127
 128         OBD_SLAB_ALLOC_PTR_GFP(mdd_obj, mdd_object_kmem, GFP_NOFS);
 129         if (mdd_obj != NULL) {
 130                 struct lu_object *o;
 131
 132                 o = mdd2lu_obj(mdd_obj);
 133                 lu_object_init(o, NULL, d);
 134                 mdd_obj->mod_obj.mo_ops = &mdd_obj_ops;
 135                 mdd_obj->mod_obj.mo_dir_ops = &mdd_dir_ops;
 136                 mdd_obj->mod_count = 0;
 137                 o->lo_ops = &mdd_lu_obj_ops;
 138                 return o;
 139         } else {
 140                 return NULL;
 141         }
 142 }
 143
 144 static int mdd_object_init(const struct lu_env *env, struct lu_object *o,
 145                            const struct lu_object_conf *unused)
 146 {
 147         struct mdd_device *d = lu2mdd_dev(o->lo_dev);
 148         struct mdd_object *mdd_obj = lu2mdd_obj(o);
 149         struct lu_object  *below;
 150         struct lu_device  *under;
 151         ENTRY;
 152
 153         mdd_obj->mod_cltime = ktime_set(0, 0);
 154         under = &d->mdd_child->dd_lu_dev;
 155         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
 156         if (IS_ERR(below))
 157                 RETURN(PTR_ERR(below));
 158
 159         lu_object_add(o, below);
 160
 161         RETURN(0);
 162 }
 163
 164 static int mdd_object_start(const struct lu_env *env, struct lu_object *o)
 165 {
 166         int rc = 0;
 167
 168         if (lu_object_exists(o)) {
 169                 struct mdd_object *mdd_obj = lu2mdd_obj(o);
 170                 struct lu_attr *attr = MDD_ENV_VAR(env, la_for_start);
 171
 172                 rc = mdd_la_get(env, mdd_obj, attr);
 173         }
 174
 175         return rc;
 176 }
 177
 178 static void mdd_object_free(const struct lu_env *env, struct lu_object *o)
 179 {
 180         struct mdd_object *mdd = lu2mdd_obj(o);
 181
 182         lu_object_fini(o);
 183         OBD_SLAB_FREE_PTR(mdd, mdd_object_kmem);
 184 }
 185
 186 static int mdd_object_print(const struct lu_env *env, void *cookie,
 187                             lu_printer_t p, const struct lu_object *o)
 188 {
 189         struct mdd_object *mdd = lu2mdd_obj((struct lu_object *)o);
 190
 191         return (*p)(env, cookie,
 192                     LUSTRE_MDD_NAME"-object@%p(open_count=%d, valid=%x, cltime=%lldns, flags=%lx)",
 193                     mdd, mdd->mod_count, mdd->mod_valid,
 194                     ktime_to_ns(mdd->mod_cltime), mdd->mod_flags);
 195 }
 196
 197 static const struct lu_object_operations mdd_lu_obj_ops = {
 198         .loo_object_init    = mdd_object_init,
 199         .loo_object_start   = mdd_object_start,
 200         .loo_object_free    = mdd_object_free,
 201         .loo_object_print   = mdd_object_print,
 202 };
 203
 204 struct mdd_object *mdd_object_find(const struct lu_env *env,
 205                                    struct mdd_device *d,
 206                                    const struct lu_fid *f)
 207 {
 208         return md2mdd_obj(md_object_find_slice(env, &d->mdd_md_dev, f));
 209 }
 210
 211 /*
 212  * No permission check is needed.
 213  */
 214 int mdd_attr_get(const struct lu_env *env, struct md_object *obj,
 215                  struct md_attr *ma)
 216 {
 217         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 218         int               rc;
 219
 220         ENTRY;
 221
 222         rc = mdd_la_get(env, mdd_obj, &ma->ma_attr);
 223         if ((ma->ma_need & MA_INODE) != 0 && mdd_is_dead_obj(mdd_obj))
 224                 ma->ma_attr.la_nlink = 0;
 225
 226         RETURN(rc);
 227 }
 228
 229 /*
 230  * No permission check is needed.
 231  */
 232 static int mdd_xattr_get(const struct lu_env *env,
 233                          struct md_object *obj, struct lu_buf *buf,
 234                          const char *name)
 235 {
 236         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 237         struct md_device *md_dev = lu2md_dev(mdd2lu_dev(mdo2mdd(obj)));
 238         int rc;
 239
 240         ENTRY;
 241
 242         if (mdd_object_exists(mdd_obj) == 0) {
 243                 CERROR("%s: object "DFID" not found: rc = -2\n",
 244                        mdd_obj_dev_name(mdd_obj),
 245                        PFID(mdd_object_fid(mdd_obj)));
 246                 return -ENOENT;
 247         }
 248
 249         /* If the object has been destroyed, then do not get LMVEA, because
 250          * it needs to load stripes from the iteration of the master object,
 251          * and it will cause problem if master object has been destroyed, see
 252          * LU-6427 */
 253         if (unlikely((mdd_obj->mod_flags & DEAD_OBJ) &&
 254                      !(mdd_obj->mod_flags & ORPHAN_OBJ) &&
 255                       strcmp(name, XATTR_NAME_LMV) == 0))
 256                 RETURN(-ENOENT);
 257
 258         /* If the object has been delete from the namespace, then
 259          * get linkEA should return -ENOENT as well */
 260         if (unlikely((mdd_obj->mod_flags & (DEAD_OBJ | ORPHAN_OBJ)) &&
 261                       strcmp(name, XATTR_NAME_LINK) == 0))
 262                 RETURN(-ENOENT);
 263
 264         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 265         rc = mdo_xattr_get(env, mdd_obj, buf, name);
 266         mdd_read_unlock(env, mdd_obj);
 267
 268         /* record only getting user xattrs and acls */
 269         if (rc >= 0 &&
 270             (has_prefix(name, XATTR_USER_PREFIX) ||
 271              has_prefix(name, XATTR_NAME_POSIX_ACL_ACCESS) ||
 272              has_prefix(name, XATTR_NAME_POSIX_ACL_DEFAULT))) {
 273                 struct thandle *handle;
 274                 struct mdd_device *mdd = lu2mdd_dev(&md_dev->md_lu_dev);
 275                 int rc2;
 276
 277                 /* Not recording */
 278                 if (!(mdd->mdd_cl.mc_flags & CLM_ON))
 279                         RETURN(rc);
 280                 if (!(mdd->mdd_cl.mc_mask & (1 << CL_GETXATTR)))
 281                         RETURN(rc);
 282
 283                 LASSERT(mdo2fid(mdd_obj) != NULL);
 284
 285                 handle = mdd_trans_create(env, mdd);
 286                 if (IS_ERR(handle))
 287                         RETURN(PTR_ERR(handle));
 288
 289                 rc2 = mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
 290                 if (rc2)
 291                         GOTO(stop, rc2);
 292
 293                 rc2 = mdd_trans_start(env, mdd, handle);
 294                 if (rc2)
 295                         GOTO(stop, rc2);
 296
 297                 rc2 = mdd_changelog_data_store_by_fid(env, mdd, CL_GETXATTR, 0,
 298                                                       mdo2fid(mdd_obj), name,
 299                                                       handle);
 300
 301 stop:
 302                 rc2 = mdd_trans_stop(env, mdd, rc2, handle);
 303                 if (rc2)
 304                         rc = rc2;
 305         }
 306
 307         RETURN(rc);
 308 }
 309
 310 /*
 311  * Permission check is done when open,
 312  * no need check again.
 313  */
 314 int mdd_readlink(const struct lu_env *env, struct md_object *obj,
 315                  struct lu_buf *buf)
 316 {
 317         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 318         struct dt_object  *next;
 319         loff_t             pos = 0;
 320         int                rc;
 321         ENTRY;
 322
 323         if (mdd_object_exists(mdd_obj) == 0) {
 324                 CERROR("%s: object "DFID" not found: rc = -2\n",
 325                        mdd_obj_dev_name(mdd_obj),PFID(mdd_object_fid(mdd_obj)));
 326                 return -ENOENT;
 327         }
 328
 329         next = mdd_object_child(mdd_obj);
 330         LASSERT(next != NULL);
 331         LASSERT(next->do_body_ops != NULL);
 332         LASSERT(next->do_body_ops->dbo_read != NULL);
 333         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 334         rc = dt_read(env, next, buf, &pos);
 335         mdd_read_unlock(env, mdd_obj);
 336         RETURN(rc);
 337 }
 338
 339 /*
 340  * No permission check is needed.
 341  */
 342 static int mdd_xattr_list(const struct lu_env *env, struct md_object *obj,
 343                           struct lu_buf *buf)
 344 {
 345         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 346         int rc;
 347
 348         ENTRY;
 349
 350         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 351         rc = mdo_xattr_list(env, mdd_obj, buf);
 352         mdd_read_unlock(env, mdd_obj);
 353
 354         /* If the buffer is NULL then we are only here to get the
 355          * length of the xattr name list. */
 356         if (rc < 0 || buf->lb_buf == NULL)
 357                 RETURN(rc);
 358
 359         /*
 360          * Filter out XATTR_NAME_LINK if this is an orphan object.  See
 361          * mdd_xattr_get().
 362          */
 363         if (unlikely(mdd_obj->mod_flags & (DEAD_OBJ | ORPHAN_OBJ))) {
 364                 char   *end = (char *)buf->lb_buf + rc;
 365                 char   *p = buf->lb_buf;
 366
 367                 while (p < end) {
 368                         char   *next = p + strlen(p) + 1;
 369
 370                         if (strcmp(p, XATTR_NAME_LINK) == 0) {
 371                                 if (end - next > 0)
 372                                         memmove(p, next, end - next);
 373                                 rc -= next - p;
 374                                 CDEBUG(D_INFO, "Filtered out "XATTR_NAME_LINK
 375                                        " of orphan "DFID"\n",
 376                                        PFID(mdd_object_fid(mdd_obj)));
 377                                 break;
 378                         }
 379
 380                         p = next;
 381                 }
 382         }
 383
 384         RETURN(rc);
 385 }
 386
 387 int mdd_invalidate(const struct lu_env *env, struct md_object *obj)
 388 {
 389         return mdo_invalidate(env, md2mdd_obj(obj));
 390 }
 391
 392 int mdd_declare_create_object_internal(const struct lu_env *env,
 393                                        struct mdd_object *p,
 394                                        struct mdd_object *c,
 395                                        struct lu_attr *attr,
 396                                        struct thandle *handle,
 397                                        const struct md_op_spec *spec,
 398                                        struct dt_allocation_hint *hint)
 399 {
 400         struct dt_object_format *dof = &mdd_env_info(env)->mti_dof;
 401         const struct dt_index_features *feat = spec->sp_feat;
 402         int rc;
 403         ENTRY;
 404
 405         if (feat != &dt_directory_features && feat != NULL) {
 406                 dof->dof_type = DFT_INDEX;
 407                 dof->u.dof_idx.di_feat = feat;
 408         } else {
 409                 dof->dof_type = dt_mode_to_dft(attr->la_mode);
 410                 if (dof->dof_type == DFT_REGULAR) {
 411                         dof->u.dof_reg.striped =
 412                                 md_should_create(spec->sp_cr_flags);
 413                         if (spec->sp_cr_flags & MDS_OPEN_HAS_EA)
 414                                 dof->u.dof_reg.striped = 0;
 415                         /* is this replay? */
 416                         if (spec->no_create)
 417                                 dof->u.dof_reg.striped = 0;
 418                 }
 419         }
 420
 421         rc = mdo_declare_create_object(env, c, attr, hint, dof, handle);
 422
 423         RETURN(rc);
 424 }
 425
 426 int mdd_create_object_internal(const struct lu_env *env, struct mdd_object *p,
 427                                struct mdd_object *c, struct lu_attr *attr,
 428                                struct thandle *handle,
 429                                const struct md_op_spec *spec,
 430                                struct dt_allocation_hint *hint)
 431 {
 432         struct dt_object_format *dof = &mdd_env_info(env)->mti_dof;
 433         int rc;
 434         ENTRY;
 435
 436         LASSERT(!mdd_object_exists(c));
 437
 438         rc = mdo_create_object(env, c, attr, hint, dof, handle);
 439
 440         RETURN(rc);
 441 }
 442
 443 int mdd_attr_set_internal(const struct lu_env *env, struct mdd_object *obj,
 444                           const struct lu_attr *attr, struct thandle *handle,
 445                           int needacl)
 446 {
 447         int rc;
 448         ENTRY;
 449
 450         rc = mdo_attr_set(env, obj, attr, handle);
 451 #ifdef CONFIG_FS_POSIX_ACL
 452         if (!rc && (attr->la_valid & LA_MODE) && needacl)
 453                 rc = mdd_acl_chmod(env, obj, attr->la_mode, handle);
 454 #endif
 455         RETURN(rc);
 456 }
 457
 458 int mdd_update_time(const struct lu_env *env, struct mdd_object *obj,
 459                     const struct lu_attr *oattr, struct lu_attr *attr,
 460                     struct thandle *handle)
 461 {
 462         int rc = 0;
 463         ENTRY;
 464
 465         LASSERT(attr->la_valid & LA_CTIME);
 466         LASSERT(oattr != NULL);
 467
 468         /* Make sure the ctime is increased only, however, it's not strictly
 469          * reliable at here because there is not guarantee to hold lock on
 470          * object, so we just bypass some unnecessary cmtime setting first
 471          * and OSD has to check it again. */
 472         if (attr->la_ctime < oattr->la_ctime)
 473                 attr->la_valid &= ~(LA_MTIME | LA_CTIME);
 474         else if (attr->la_valid == LA_CTIME &&
 475                  attr->la_ctime == oattr->la_ctime)
 476                 attr->la_valid &= ~LA_CTIME;
 477
 478         if (attr->la_valid != 0)
 479                 rc = mdd_attr_set_internal(env, obj, attr, handle, 0);
 480         RETURN(rc);
 481 }
 482
 483 /*
 484  * This gives the same functionality as the code between
 485  * sys_chmod and inode_setattr
 486  * chown_common and inode_setattr
 487  * utimes and inode_setattr
 488  * This API is ported from mds_fix_attr but remove some unnecesssary stuff.
 489  */
 490 static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
 491                         const struct lu_attr *oattr, struct lu_attr *la,
 492                         const unsigned long flags)
 493 {
 494         struct lu_ucred  *uc;
 495         int               rc = 0;
 496         ENTRY;
 497
 498         if (!la->la_valid)
 499                 RETURN(0);
 500
 501         /* Do not permit change file type */
 502         if (la->la_valid & LA_TYPE)
 503                 RETURN(-EPERM);
 504
 505         /* They should not be processed by setattr */
 506         if (la->la_valid & (LA_NLINK | LA_RDEV | LA_BLKSIZE))
 507                 RETURN(-EPERM);
 508
 509         LASSERT(oattr != NULL);
 510
 511         uc = lu_ucred_check(env);
 512         if (uc == NULL)
 513                 RETURN(0);
 514
 515         if (la->la_valid == LA_CTIME) {
 516                 if (!(flags & MDS_PERM_BYPASS))
 517                         /* This is only for set ctime when rename's source is
 518                          * on remote MDS. */
 519                         rc = mdd_may_delete(env, NULL, NULL, obj, oattr, NULL,
 520                                             1, 0);
 521                 if (rc == 0 && la->la_ctime <= oattr->la_ctime)
 522                         la->la_valid &= ~LA_CTIME;
 523                 RETURN(rc);
 524         }
 525
 526         if (la->la_valid == LA_ATIME) {
 527                 /* This is an atime-only attribute update for close RPCs. */
 528                 if (la->la_atime < (oattr->la_atime +
 529                                 mdd_obj2mdd_dev(obj)->mdd_atime_diff))
 530                         la->la_valid &= ~LA_ATIME;
 531                 RETURN(0);
 532         }
 533
 534         /* Check if flags change. */
 535         if (la->la_valid & LA_FLAGS) {
 536                 unsigned int oldflags = oattr->la_flags &
 537                                 (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL);
 538                 unsigned int newflags = la->la_flags &
 539                                 (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL);
 540
 541                 if ((uc->uc_fsuid != oattr->la_uid) &&
 542                     !md_capable(uc, CFS_CAP_FOWNER))
 543                         RETURN(-EPERM);
 544
 545                 /* The IMMUTABLE and APPEND_ONLY flags can
 546                  * only be changed by the relevant capability. */
 547                 if ((oldflags ^ newflags) &&
 548                     !md_capable(uc, CFS_CAP_LINUX_IMMUTABLE))
 549                         RETURN(-EPERM);
 550
 551                 if (!S_ISDIR(oattr->la_mode))
 552                         la->la_flags &= ~(LUSTRE_DIRSYNC_FL | LUSTRE_TOPDIR_FL);
 553         }
 554
 555         if (oattr->la_flags & (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL) &&
 556             (la->la_valid & ~LA_FLAGS) &&
 557             !(flags & MDS_PERM_BYPASS))
 558                 RETURN(-EPERM);
 559
 560         /* Check for setting the obj time. */
 561         if ((la->la_valid & (LA_MTIME | LA_ATIME | LA_CTIME)) &&
 562             !(la->la_valid & ~(LA_MTIME | LA_ATIME | LA_CTIME))) {
 563                 if ((uc->uc_fsuid != oattr->la_uid) &&
 564                     !md_capable(uc, CFS_CAP_FOWNER)) {
 565                         rc = mdd_permission_internal(env, obj, oattr,
 566                                                      MAY_WRITE);
 567                         if (rc)
 568                                 RETURN(rc);
 569                 }
 570         }
 571
 572         if (la->la_valid & LA_KILL_SUID) {
 573                 la->la_valid &= ~LA_KILL_SUID;
 574                 if ((oattr->la_mode & S_ISUID) &&
 575                     !(la->la_valid & LA_MODE)) {
 576                         la->la_mode = oattr->la_mode;
 577                         la->la_valid |= LA_MODE;
 578                 }
 579                 la->la_mode &= ~S_ISUID;
 580         }
 581
 582         if (la->la_valid & LA_KILL_SGID) {
 583                 la->la_valid &= ~LA_KILL_SGID;
 584                 if (((oattr->la_mode & (S_ISGID | S_IXGRP)) ==
 585                         (S_ISGID | S_IXGRP)) &&
 586                     !(la->la_valid & LA_MODE)) {
 587                         la->la_mode = oattr->la_mode;
 588                         la->la_valid |= LA_MODE;
 589                 }
 590                 la->la_mode &= ~S_ISGID;
 591         }
 592
 593         /* Make sure a caller can chmod. */
 594         if (la->la_valid & LA_MODE) {
 595                 if (!(flags & MDS_PERM_BYPASS) &&
 596                     (uc->uc_fsuid != oattr->la_uid) &&
 597                     !md_capable(uc, CFS_CAP_FOWNER))
 598                         RETURN(-EPERM);
 599
 600                 if (la->la_mode == (umode_t) -1)
 601                         la->la_mode = oattr->la_mode;
 602                 else
 603                         la->la_mode = (la->la_mode & S_IALLUGO) |
 604                                         (oattr->la_mode & ~S_IALLUGO);
 605
 606                 /* Also check the setgid bit! */
 607                 if (!lustre_in_group_p(uc, (la->la_valid & LA_GID) ?
 608                                        la->la_gid : oattr->la_gid) &&
 609                     !md_capable(uc, CFS_CAP_FSETID))
 610                         la->la_mode &= ~S_ISGID;
 611         } else {
 612                la->la_mode = oattr->la_mode;
 613         }
 614
 615         /* Make sure a caller can chown. */
 616         if (la->la_valid & LA_UID) {
 617                 if (la->la_uid == (uid_t) -1)
 618                         la->la_uid = oattr->la_uid;
 619                 if (((uc->uc_fsuid != oattr->la_uid) ||
 620                      (la->la_uid != oattr->la_uid)) &&
 621                     !md_capable(uc, CFS_CAP_CHOWN))
 622                         RETURN(-EPERM);
 623
 624                 /* If the user or group of a non-directory has been
 625                  * changed by a non-root user, remove the setuid bit.
 626                  * 19981026 David C Niemi <niemi@tux.org>
 627                  *
 628                  * Changed this to apply to all users, including root,
 629                  * to avoid some races. This is the behavior we had in
 630                  * 2.0. The check for non-root was definitely wrong
 631                  * for 2.2 anyway, as it should have been using
 632                  * CAP_FSETID rather than fsuid -- 19990830 SD. */
 633                 if (((oattr->la_mode & S_ISUID) == S_ISUID) &&
 634                 !S_ISDIR(oattr->la_mode)) {
 635                         la->la_mode &= ~S_ISUID;
 636                         la->la_valid |= LA_MODE;
 637                 }
 638         }
 639
 640         /* Make sure caller can chgrp. */
 641         if (la->la_valid & LA_GID) {
 642                 if (la->la_gid == (gid_t) -1)
 643                         la->la_gid = oattr->la_gid;
 644                 if (((uc->uc_fsuid != oattr->la_uid) ||
 645                      ((la->la_gid != oattr->la_gid) &&
 646                       !lustre_in_group_p(uc, la->la_gid))) &&
 647                     !md_capable(uc, CFS_CAP_CHOWN))
 648                         RETURN(-EPERM);
 649
 650                 /* Likewise, if the user or group of a non-directory
 651                  * has been changed by a non-root user, remove the
 652                  * setgid bit UNLESS there is no group execute bit
 653                  * (this would be a file marked for mandatory
 654                  * locking).  19981026 David C Niemi <niemi@tux.org>
 655                  *
 656                  * Removed the fsuid check (see the comment above) --
 657                  * 19990830 SD. */
 658                 if (((oattr->la_mode & (S_ISGID | S_IXGRP)) ==
 659                     (S_ISGID | S_IXGRP)) && !S_ISDIR(oattr->la_mode)) {
 660                         la->la_mode &= ~S_ISGID;
 661                         la->la_valid |= LA_MODE;
 662                 }
 663         }
 664
 665         if (la->la_valid & (LA_SIZE | LA_BLOCKS)) {
 666                 if (!((flags & MDS_OWNEROVERRIDE) &&
 667                       (uc->uc_fsuid == oattr->la_uid)) &&
 668                     !(flags & MDS_PERM_BYPASS)) {
 669                         rc = mdd_permission_internal(env, obj, oattr,
 670                                                      MAY_WRITE);
 671                         if (rc != 0)
 672                                 RETURN(rc);
 673                 }
 674         }
 675
 676         if (la->la_valid & LA_CTIME) {
 677                 /* The pure setattr, it has the priority over what is
 678                  * already set, do not drop it if ctime is equal. */
 679                 if (la->la_ctime < oattr->la_ctime)
 680                         la->la_valid &= ~(LA_ATIME | LA_MTIME | LA_CTIME);
 681         }
 682
 683         RETURN(0);
 684 }
 685
 686 static int mdd_changelog_data_store_by_fid(const struct lu_env *env,
 687                                            struct mdd_device *mdd,
 688                                            enum changelog_rec_type type,
 689                                            int flags, const struct lu_fid *fid,
 690                                            const char *xattr_name,
 691                                            struct thandle *handle)
 692 {
 693         const struct lu_ucred *uc = lu_ucred(env);
 694         struct llog_changelog_rec *rec;
 695         struct lu_buf *buf;
 696         int reclen;
 697         int xflags = CLFE_INVALID;
 698         int rc;
 699
 700         flags = (flags & CLF_FLAGMASK) | CLF_VERSION | CLF_EXTRA_FLAGS;
 701
 702         if (uc) {
 703                 if (uc->uc_jobid[0] != '\0')
 704                         flags |= CLF_JOBID;
 705                 xflags |= CLFE_UIDGID;
 706                 xflags |= CLFE_NID;
 707         }
 708         if (type == CL_OPEN)
 709                 xflags |= CLFE_OPEN;
 710         if (type == CL_SETXATTR || type == CL_GETXATTR)
 711                 xflags |= CLFE_XATTR;
 712
 713         reclen = llog_data_len(LLOG_CHANGELOG_HDR_SZ +
 714                                changelog_rec_offset(flags & CLF_SUPPORTED,
 715                                                     xflags & CLFE_SUPPORTED));
 716         buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf, reclen);
 717         if (buf->lb_buf == NULL)
 718                 RETURN(-ENOMEM);
 719         rec = buf->lb_buf;
 720
 721         rec->cr_hdr.lrh_len = reclen;
 722         rec->cr.cr_flags = flags;
 723         rec->cr.cr_type = (__u32)type;
 724         rec->cr.cr_tfid = *fid;
 725         rec->cr.cr_namelen = 0;
 726
 727         if (flags & CLF_JOBID)
 728                 mdd_changelog_rec_ext_jobid(&rec->cr, uc->uc_jobid);
 729
 730         if (flags & CLF_EXTRA_FLAGS) {
 731                 mdd_changelog_rec_ext_extra_flags(&rec->cr, xflags);
 732                 if (xflags & CLFE_UIDGID)
 733                         mdd_changelog_rec_extra_uidgid(&rec->cr,
 734                                                        uc->uc_uid, uc->uc_gid);
 735                 if (xflags & CLFE_NID)
 736                         mdd_changelog_rec_extra_nid(&rec->cr, uc->uc_nid);
 737                 if (xflags & CLFE_OPEN)
 738                         mdd_changelog_rec_extra_omode(&rec->cr, flags);
 739                 if (xflags & CLFE_XATTR) {
 740                         if (xattr_name == NULL)
 741                                 RETURN(-EINVAL);
 742                         mdd_changelog_rec_extra_xattr(&rec->cr, xattr_name);
 743                 }
 744         }
 745
 746         rc = mdd_changelog_store(env, mdd, rec, handle);
 747         RETURN(rc);
 748 }
 749
 750
 751 /** Store a data change changelog record
 752  * If this fails, we must fail the whole transaction; we don't
 753  * want the change to commit without the log entry.
 754  * \param mdd_obj - mdd_object of change
 755  * \param handle - transaction handle
 756  */
 757 int mdd_changelog_data_store(const struct lu_env *env, struct mdd_device *mdd,
 758                              enum changelog_rec_type type, int flags,
 759                              struct mdd_object *mdd_obj, struct thandle *handle)
 760 {
 761         int                              rc;
 762
 763         LASSERT(mdd_obj != NULL);
 764         LASSERT(handle != NULL);
 765
 766         /* Not recording */
 767         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
 768                 RETURN(0);
 769         if ((mdd->mdd_cl.mc_mask & (1 << type)) == 0)
 770                 RETURN(0);
 771
 772         if (mdd_is_volatile_obj(mdd_obj))
 773                 RETURN(0);
 774
 775         if ((type >= CL_MTIME) && (type <= CL_ATIME) &&
 776             ktime_before(mdd->mdd_cl.mc_starttime, mdd_obj->mod_cltime)) {
 777                 /* Don't need multiple updates in this log */
 778                 /* Don't check under lock - no big deal if we get an extra
 779                    entry */
 780                 RETURN(0);
 781         }
 782
 783         rc = mdd_changelog_data_store_by_fid(env, mdd, type, flags,
 784                                              mdo2fid(mdd_obj), NULL, handle);
 785         if (rc == 0)
 786                 mdd_obj->mod_cltime = ktime_get();
 787
 788         RETURN(rc);
 789 }
 790
 791 static int mdd_changelog_data_store_xattr(const struct lu_env *env,
 792                                           struct mdd_device *mdd,
 793                                           enum changelog_rec_type type,
 794                                           int flags, struct mdd_object *mdd_obj,
 795                                           const char *xattr_name,
 796                                           struct thandle *handle)
 797 {
 798         int                              rc;
 799
 800         LASSERT(mdd_obj != NULL);
 801         LASSERT(handle != NULL);
 802
 803         /* Not recording */
 804         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
 805                 RETURN(0);
 806         if ((mdd->mdd_cl.mc_mask & (1 << type)) == 0)
 807                 RETURN(0);
 808
 809         if (mdd_is_volatile_obj(mdd_obj))
 810                 RETURN(0);
 811
 812         if ((type >= CL_MTIME) && (type <= CL_ATIME) &&
 813             ktime_before(mdd->mdd_cl.mc_starttime, mdd_obj->mod_cltime)) {
 814                 /* Don't need multiple updates in this log */
 815                 /* Don't check under lock - no big deal if we get an extra
 816                  * entry
 817                  */
 818                 RETURN(0);
 819         }
 820
 821         rc = mdd_changelog_data_store_by_fid(env, mdd, type, flags,
 822                                              mdo2fid(mdd_obj), xattr_name,
 823                                              handle);
 824         if (rc == 0)
 825                 mdd_obj->mod_cltime = ktime_get();
 826
 827         RETURN(rc);
 828 }
 829
 830 static int mdd_changelog(const struct lu_env *env, enum changelog_rec_type type,
 831                   int flags, struct md_device *m, const struct lu_fid *fid)
 832 {
 833         struct thandle *handle;
 834         struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
 835         int rc;
 836         ENTRY;
 837
 838         /* Not recording */
 839         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
 840                 RETURN(0);
 841         if (!(mdd->mdd_cl.mc_mask & (1 << type)))
 842                 RETURN(0);
 843
 844         LASSERT(fid != NULL);
 845
 846         handle = mdd_trans_create(env, mdd);
 847         if (IS_ERR(handle))
 848                 RETURN(PTR_ERR(handle));
 849
 850         rc = mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
 851         if (rc)
 852                 GOTO(stop, rc);
 853
 854         rc = mdd_trans_start(env, mdd, handle);
 855         if (rc)
 856                 GOTO(stop, rc);
 857
 858         rc = mdd_changelog_data_store_by_fid(env, mdd, type, flags,
 859                                              fid, NULL, handle);
 860
 861 stop:
 862         rc = mdd_trans_stop(env, mdd, rc, handle);
 863
 864         RETURN(rc);
 865 }
 866
 867 /**
 868  * Save LMA extended attributes with data from \a ma.
 869  *
 870  * HSM and Size-On-MDS data will be extracted from \ma if they are valid, if
 871  * not, LMA EA will be first read from disk, modified and write back.
 872  *
 873  */
 874 /* Precedence for choosing record type when multiple
 875  * attributes change: setattr > mtime > ctime > atime
 876  * (ctime changes when mtime does, plus chmod/chown.
 877  * atime and ctime are independent.) */
 878 static int mdd_attr_set_changelog(const struct lu_env *env,
 879                                   struct md_object *obj, struct thandle *handle,
 880                                   __u64 valid)
 881 {
 882         struct mdd_device *mdd = mdo2mdd(obj);
 883         int bits, type = 0;
 884
 885         bits =  (valid & LA_SIZE)  ? 1 << CL_TRUNC : 0;
 886         bits |= (valid & ~(LA_CTIME|LA_MTIME|LA_ATIME)) ? 1 << CL_SETATTR : 0;
 887         bits |= (valid & LA_MTIME) ? 1 << CL_MTIME : 0;
 888         bits |= (valid & LA_CTIME) ? 1 << CL_CTIME : 0;
 889         bits |= (valid & LA_ATIME) ? 1 << CL_ATIME : 0;
 890         bits = bits & mdd->mdd_cl.mc_mask;
 891         /* This is an implementation limit rather than a protocol limit */
 892         CLASSERT(CL_LAST <= sizeof(int) * 8);
 893         if (bits == 0)
 894                 return 0;
 895
 896         /* The record type is the lowest non-masked set bit */
 897         type = __ffs(bits);
 898
 899         /* FYI we only store the first CLF_FLAGMASK bits of la_valid */
 900         return mdd_changelog_data_store(env, mdd, type, (int)valid,
 901                                         md2mdd_obj(obj), handle);
 902 }
 903
 904 static int mdd_declare_attr_set(const struct lu_env *env,
 905                                 struct mdd_device *mdd,
 906                                 struct mdd_object *obj,
 907                                 const struct lu_attr *attr,
 908                                 struct thandle *handle)
 909 {
 910         int rc;
 911
 912         rc = mdo_declare_attr_set(env, obj, attr, handle);
 913         if (rc)
 914                 return rc;
 915
 916 #ifdef CONFIG_FS_POSIX_ACL
 917         if (attr->la_valid & LA_MODE) {
 918                 mdd_read_lock(env, obj, MOR_TGT_CHILD);
 919                 rc = mdo_xattr_get(env, obj, &LU_BUF_NULL,
 920                                    XATTR_NAME_ACL_ACCESS);
 921                 mdd_read_unlock(env, obj);
 922                 if (rc == -EOPNOTSUPP || rc == -ENODATA)
 923                         rc = 0;
 924                 else if (rc < 0)
 925                         return rc;
 926
 927                 if (rc != 0) {
 928                         struct lu_buf *buf = mdd_buf_get(env, NULL, rc);
 929                         rc = mdo_declare_xattr_set(env, obj, buf,
 930                                                    XATTR_NAME_ACL_ACCESS, 0,
 931                                                    handle);
 932                         if (rc)
 933                                 return rc;
 934                 }
 935         }
 936 #endif
 937
 938         rc = mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
 939         return rc;
 940 }
 941
 942 /*
 943  * LU-3671
 944  *
 945  * permission changes may require sync operation, to mitigate performance
 946  * impact, only do this for dir and when permission is reduced.
 947  *
 948  * For regular files, version is updated with permission change (see VBR), async
 949  * permission won't cause any issue, while missing permission change on
 950  * directory may affect accessibility of other objects after recovery.
 951  */
 952 static inline bool permission_needs_sync(const struct lu_attr *old,
 953                                          const struct lu_attr *new)
 954 {
 955         if (!S_ISDIR(old->la_mode))
 956                 return false;
 957
 958         if (new->la_valid & (LA_UID | LA_GID))
 959                 return true;
 960
 961         if (new->la_valid & LA_MODE &&
 962             new->la_mode & (S_ISUID | S_ISGID | S_ISVTX))
 963                 return true;
 964
 965         if ((new->la_valid & LA_MODE) &&
 966             ((new->la_mode & old->la_mode) & S_IRWXUGO) !=
 967              (old->la_mode & S_IRWXUGO))
 968                 return true;
 969
 970         return false;
 971 }
 972
 973 /* set attr and LOV EA at once, return updated attr */
 974 int mdd_attr_set(const struct lu_env *env, struct md_object *obj,
 975                  const struct md_attr *ma)
 976 {
 977         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 978         struct mdd_device *mdd = mdo2mdd(obj);
 979         struct thandle *handle = NULL;
 980         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
 981         struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
 982         const struct lu_attr *la = &ma->ma_attr;
 983         struct lu_ucred  *uc;
 984         int rc;
 985         ENTRY;
 986
 987         /* we do not use ->attr_set() for LOV/HSM EA any more */
 988         LASSERT((ma->ma_valid & MA_LOV) == 0);
 989         LASSERT((ma->ma_valid & MA_HSM) == 0);
 990
 991         rc = mdd_la_get(env, mdd_obj, attr);
 992         if (rc)
 993                 RETURN(rc);
 994
 995         *la_copy = ma->ma_attr;
 996         rc = mdd_fix_attr(env, mdd_obj, attr, la_copy, ma->ma_attr_flags);
 997         if (rc)
 998                 RETURN(rc);
 999
1000         /* no need to setattr anymore */
1001         if (la_copy->la_valid == 0) {
1002                 CDEBUG(D_INODE, "%s: no valid attribute on "DFID", previous"
1003                        "valid is %#llx\n", mdd2obd_dev(mdd)->obd_name,
1004                        PFID(mdo2fid(mdd_obj)), la->la_valid);
1005
1006                 RETURN(0);
1007         }
1008
1009         /* If an unprivileged user changes group of some file,
1010          * the setattr operation will be processed synchronously to
1011          * honor the quota limit of the corresponding group. see LU-5152 */
1012         uc = lu_ucred_check(env);
1013         if (S_ISREG(attr->la_mode) && la->la_valid & LA_GID &&
1014             la->la_gid != attr->la_gid && uc != NULL && uc->uc_fsuid != 0) {
1015                 la_copy->la_valid |= LA_FLAGS;
1016                 la_copy->la_flags |= LUSTRE_SET_SYNC_FL;
1017
1018                 /* Flush the possible existing sync requests to OSTs to
1019                  * keep the order of sync for the current setattr operation
1020                  * will be sent directly to OSTs. see LU-5152 */
1021                 rc = dt_sync(env, mdd->mdd_child);
1022                 if (rc)
1023                         GOTO(out, rc);
1024         }
1025
1026         handle = mdd_trans_create(env, mdd);
1027         if (IS_ERR(handle)) {
1028                 rc = PTR_ERR(handle);
1029                 handle = NULL;
1030
1031                 GOTO(out, rc);
1032         }
1033
1034         rc = mdd_declare_attr_set(env, mdd, mdd_obj, la_copy, handle);
1035         if (rc)
1036                 GOTO(out, rc);
1037
1038         rc = mdd_trans_start(env, mdd, handle);
1039         if (rc)
1040                 GOTO(out, rc);
1041
1042         if (mdd->mdd_sync_permission && permission_needs_sync(attr, la))
1043                 handle->th_sync = 1;
1044
1045         if (la->la_valid & (LA_MTIME | LA_CTIME))
1046                 CDEBUG(D_INODE, "setting mtime %llu, ctime %llu\n",
1047                        la->la_mtime, la->la_ctime);
1048
1049         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1050         if (la_copy->la_valid) {
1051                 rc = mdd_attr_set_internal(env, mdd_obj, la_copy, handle, 1);
1052
1053                 if (rc == -EDQUOT && la_copy->la_flags & LUSTRE_SET_SYNC_FL) {
1054                         /* rollback to the original gid */
1055                         la_copy->la_flags &= ~LUSTRE_SET_SYNC_FL;
1056                         la_copy->la_gid = attr->la_gid;
1057                         mdd_attr_set_internal(env, mdd_obj, la_copy, handle, 1);
1058                 }
1059         }
1060         mdd_write_unlock(env, mdd_obj);
1061
1062 out:
1063         if (rc == 0)
1064                 rc = mdd_attr_set_changelog(env, obj, handle,
1065                                             la_copy->la_valid);
1066
1067         if (handle != NULL)
1068                 rc = mdd_trans_stop(env, mdd, rc, handle);
1069
1070         return rc;
1071 }
1072
1073 static int mdd_xattr_sanity_check(const struct lu_env *env,
1074                                   struct mdd_object *obj,
1075                                   const struct lu_attr *attr,
1076                                   const char *name)
1077 {
1078         struct lu_ucred *uc     = lu_ucred_assert(env);
1079         ENTRY;
1080
1081         if (attr->la_flags & (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL))
1082                 RETURN(-EPERM);
1083
1084         if (strncmp(XATTR_USER_PREFIX, name,
1085                     sizeof(XATTR_USER_PREFIX) - 1) == 0) {
1086                 /* For sticky directories, only the owner and privileged user
1087                  * can write attributes. */
1088                 if (S_ISDIR(attr->la_mode) && (attr->la_mode & S_ISVTX) &&
1089                     (uc->uc_fsuid != attr->la_uid) &&
1090                     !md_capable(uc, CFS_CAP_FOWNER))
1091                         RETURN(-EPERM);
1092         } else {
1093                 if ((uc->uc_fsuid != attr->la_uid) &&
1094                     !md_capable(uc, CFS_CAP_FOWNER))
1095                         RETURN(-EPERM);
1096         }
1097
1098         RETURN(0);
1099 }
1100
1101 /**
1102  * Check if a string begins with a given prefix.
1103  *
1104  * \param str     String to check
1105  * \param prefix  Substring to check at the beginning of \a str
1106  * \return true/false whether the condition is verified.
1107  */
1108 static inline bool has_prefix(const char *str, const char *prefix)
1109 {
1110         return strncmp(prefix, str, strlen(prefix)) == 0;
1111 }
1112
1113 /**
1114  * Indicate the kind of changelog to store (if any) for a xattr set/del.
1115  *
1116  * \param[in]  xattr_name  Full extended attribute name.
1117  *
1118  * \return The type of changelog to use, or -1 if no changelog is to be emitted.
1119  */
1120 static enum changelog_rec_type
1121 mdd_xattr_changelog_type(const struct lu_env *env, struct mdd_device *mdd,
1122                          const char *xattr_name)
1123 {
1124         /* Layout changes systematically recorded */
1125         if (strcmp(XATTR_NAME_LOV, xattr_name) == 0 ||
1126             strncmp(XATTR_LUSTRE_LOV, xattr_name,
1127                     strlen(XATTR_LUSTRE_LOV)) == 0)
1128                 return CL_LAYOUT;
1129
1130         /* HSM information changes systematically recorded */
1131         if (strcmp(XATTR_NAME_HSM, xattr_name) == 0)
1132                 return CL_HSM;
1133
1134         if (has_prefix(xattr_name, XATTR_USER_PREFIX) ||
1135             has_prefix(xattr_name, XATTR_NAME_POSIX_ACL_ACCESS) ||
1136             has_prefix(xattr_name, XATTR_NAME_POSIX_ACL_DEFAULT) ||
1137             has_prefix(xattr_name, XATTR_TRUSTED_PREFIX) ||
1138             has_prefix(xattr_name, XATTR_SECURITY_PREFIX))
1139                 return CL_SETXATTR;
1140
1141         return -1;
1142 }
1143
1144 static int mdd_declare_xattr_set(const struct lu_env *env,
1145                                  struct mdd_device *mdd,
1146                                  struct mdd_object *obj,
1147                                  const struct lu_buf *buf,
1148                                  const char *name,
1149                                  int fl, struct thandle *handle)
1150 {
1151         int     rc;
1152
1153         rc = mdo_declare_xattr_set(env, obj, buf, name, fl, handle);
1154         if (rc)
1155                 return rc;
1156
1157         if (mdd_xattr_changelog_type(env, mdd, name) < 0)
1158                 return 0; /* no changelog to store */
1159
1160         return mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
1161 }
1162
1163 /*
1164  * Compare current and future data of HSM EA and add a changelog if needed.
1165  *
1166  * Caller should have write-locked \param obj.
1167  *
1168  * \param buf - Future HSM EA content.
1169  * \retval 0 if no changelog is needed or changelog was added properly.
1170  * \retval -ve errno if there was a problem
1171  */
1172 static int mdd_hsm_update_locked(const struct lu_env *env,
1173                                  struct md_object *obj,
1174                                  const struct lu_buf *buf,
1175                                  struct thandle *handle, int *cl_flags)
1176 {
1177         struct mdd_thread_info *info = mdd_env_info(env);
1178         struct mdd_object      *mdd_obj = md2mdd_obj(obj);
1179         struct lu_buf          *current_buf;
1180         struct md_hsm          *current_mh;
1181         struct md_hsm          *new_mh;
1182         int                     rc;
1183         ENTRY;
1184
1185         OBD_ALLOC_PTR(current_mh);
1186         if (current_mh == NULL)
1187                 RETURN(-ENOMEM);
1188
1189         /* Read HSM attrs from disk */
1190         current_buf = lu_buf_check_and_alloc(&info->mti_xattr_buf,
1191                                 mdo2mdd(obj)->mdd_dt_conf.ddp_max_ea_size);
1192         rc = mdo_xattr_get(env, mdd_obj, current_buf, XATTR_NAME_HSM);
1193         rc = lustre_buf2hsm(current_buf->lb_buf, rc, current_mh);
1194         if (rc < 0 && rc != -ENODATA)
1195                 GOTO(free, rc);
1196         else if (rc == -ENODATA)
1197                 current_mh->mh_flags = 0;
1198
1199         /* Map future HSM xattr */
1200         OBD_ALLOC_PTR(new_mh);
1201         if (new_mh == NULL)
1202                 GOTO(free, rc = -ENOMEM);
1203         lustre_buf2hsm(buf->lb_buf, buf->lb_len, new_mh);
1204
1205         rc = 0;
1206
1207         /* Flags differ, set flags for the changelog that will be added */
1208         if (current_mh->mh_flags != new_mh->mh_flags) {
1209                 hsm_set_cl_event(cl_flags, HE_STATE);
1210                 if (new_mh->mh_flags & HS_DIRTY)
1211                         hsm_set_cl_flags(cl_flags, CLF_HSM_DIRTY);
1212         }
1213
1214         OBD_FREE_PTR(new_mh);
1215         EXIT;
1216 free:
1217         OBD_FREE_PTR(current_mh);
1218         return rc;
1219 }
1220
1221 static int mdd_object_pfid_replace(const struct lu_env *env,
1222                                    struct mdd_object *o)
1223 {
1224         struct mdd_device *mdd = mdo2mdd(&o->mod_obj);
1225         struct thandle *handle;
1226         int rc;
1227
1228         handle = mdd_trans_create(env, mdd);
1229         if (IS_ERR(handle))
1230                 RETURN(PTR_ERR(handle));
1231
1232         handle->th_complex = 1;
1233
1234         /* it doesn't need to track the PFID update via llog, because LFSCK
1235          * will repair it even it goes wrong */
1236         rc = mdd_declare_xattr_set(env, mdd, o, NULL, XATTR_NAME_FID,
1237                                    0, handle);
1238         if (rc)
1239                 GOTO(out, rc);
1240
1241         rc = mdd_trans_start(env, mdd, handle);
1242         if (rc != 0)
1243                 GOTO(out, rc);
1244
1245         rc = mdo_xattr_set(env, o, NULL, XATTR_NAME_FID, 0, handle);
1246         if (rc)
1247                 GOTO(out, rc);
1248
1249 out:
1250         mdd_trans_stop(env, mdd, rc, handle);
1251         return rc;
1252 }
1253
1254
1255 static int mdd_declare_xattr_del(const struct lu_env *env,
1256                                  struct mdd_device *mdd,
1257                                  struct mdd_object *obj,
1258                                  const char *name,
1259                                  struct thandle *handle);
1260
1261 static int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
1262                          const char *name);
1263
1264 static int mdd_xattr_merge(const struct lu_env *env, struct md_object *md_obj,
1265                            struct md_object *md_vic)
1266 {
1267         struct mdd_device *mdd = mdo2mdd(md_obj);
1268         struct mdd_object *obj = md2mdd_obj(md_obj);
1269         struct mdd_object *vic = md2mdd_obj(md_vic);
1270         struct lu_buf *buf = &mdd_env_info(env)->mti_buf[0];
1271         struct lu_buf *buf_vic = &mdd_env_info(env)->mti_buf[1];
1272         struct lov_mds_md *lmm;
1273         struct thandle *handle;
1274         int rc;
1275         ENTRY;
1276
1277         rc = lu_fid_cmp(mdo2fid(obj), mdo2fid(vic));
1278         if (rc == 0) /* same fid */
1279                 RETURN(-EPERM);
1280
1281         handle = mdd_trans_create(env, mdd);
1282         if (IS_ERR(handle))
1283                 RETURN(PTR_ERR(handle));
1284
1285         if (rc > 0) {
1286                 mdd_write_lock(env, obj, MOR_TGT_CHILD);
1287                 mdd_write_lock(env, vic, MOR_TGT_CHILD);
1288         } else {
1289                 mdd_write_lock(env, vic, MOR_TGT_CHILD);
1290                 mdd_write_lock(env, obj, MOR_TGT_CHILD);
1291         }
1292
1293         /* get EA of victim file */
1294         memset(buf_vic, 0, sizeof(*buf_vic));
1295         rc = mdd_get_lov_ea(env, vic, buf_vic);
1296         if (rc < 0) {
1297                 if (rc == -ENODATA)
1298                         rc = 0;
1299                 GOTO(out, rc);
1300         }
1301
1302         /* parse the layout of victim file */
1303         lmm = buf_vic->lb_buf;
1304         if (le32_to_cpu(lmm->lmm_magic) != LOV_MAGIC_COMP_V1)
1305                 GOTO(out, rc = -EINVAL);
1306
1307         /* save EA of target file for restore */
1308         memset(buf, 0, sizeof(*buf));
1309         rc = mdd_get_lov_ea(env, obj, buf);
1310         if (rc < 0)
1311                 GOTO(out, rc);
1312
1313         /* Get rid of the layout from victim object */
1314         rc = mdd_declare_xattr_del(env, mdd, vic, XATTR_NAME_LOV, handle);
1315         if (rc)
1316                 GOTO(out, rc);
1317
1318         rc = mdd_declare_xattr_set(env, mdd, obj, buf_vic, XATTR_LUSTRE_LOV,
1319                                    LU_XATTR_MERGE, handle);
1320         if (rc)
1321                 GOTO(out, rc);
1322
1323         rc = mdd_trans_start(env, mdd, handle);
1324         if (rc != 0)
1325                 GOTO(out, rc);
1326
1327         rc = mdo_xattr_set(env, obj, buf_vic, XATTR_LUSTRE_LOV, LU_XATTR_MERGE,
1328                            handle);
1329         if (rc)
1330                 GOTO(out, rc);
1331
1332         rc = mdo_xattr_del(env, vic, XATTR_NAME_LOV, handle);
1333         if (rc) /* wtf? */
1334                 GOTO(out_restore, rc);
1335
1336         (void)mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, obj, handle);
1337         (void)mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, vic, handle);
1338         EXIT;
1339
1340 out_restore:
1341         if (rc) {
1342                 int rc2 = mdo_xattr_set(env, obj, buf, XATTR_NAME_LOV,
1343                                         LU_XATTR_REPLACE, handle);
1344                 if (rc2)
1345                         CERROR("%s: failed to rollback of layout of: "DFID
1346                                ": %d, file state unknown\n",
1347                                mdd_obj_dev_name(obj), PFID(mdo2fid(obj)), rc2);
1348         }
1349
1350 out:
1351         mdd_trans_stop(env, mdd, rc, handle);
1352         mdd_write_unlock(env, obj);
1353         mdd_write_unlock(env, vic);
1354         lu_buf_free(buf);
1355         lu_buf_free(buf_vic);
1356
1357         if (!rc)
1358                 (void) mdd_object_pfid_replace(env, obj);
1359
1360         return rc;
1361 }
1362
1363 /**
1364  * Extract the mirror with specified mirror id, and store the splitted
1365  * mirror layout to @buf.
1366  *
1367  * \param[in] comp_v1   mirrored layout
1368  * \param[in] mirror_id the mirror with mirror_id to be extracted
1369  * \param[out] buf      store the layout excluding the extracted mirror,
1370  *                      caller free the buffer we allocated in this function
1371  * \param[out] buf_vic  store the extracted layout, caller free the buffer
1372  *                      we allocated in this function
1373  *
1374  * \retval      0 on success; < 0 if error happens
1375  */
1376 static int mdd_split_ea(struct lov_comp_md_v1 *comp_v1, __u16 mirror_id,
1377                         struct lu_buf *buf, struct lu_buf *buf_vic)
1378 {
1379         struct lov_comp_md_v1 *comp_rem;
1380         struct lov_comp_md_v1 *comp_vic;
1381         struct lov_comp_md_entry_v1 *entry;
1382         struct lov_comp_md_entry_v1 *entry_rem;
1383         struct lov_comp_md_entry_v1 *entry_vic;
1384         __u16 mirror_cnt;
1385         __u16 comp_cnt, count = 0;
1386         int lmm_size, lmm_size_vic = 0;
1387         int i, j, k;
1388         int offset, offset_rem, offset_vic;
1389
1390         mirror_cnt = le16_to_cpu(comp_v1->lcm_mirror_count) + 1;
1391         /* comp_v1 should contains more than 1 mirror */
1392         if (mirror_cnt <= 1)
1393                 return -EINVAL;
1394         comp_cnt = le16_to_cpu(comp_v1->lcm_entry_count);
1395         lmm_size = le32_to_cpu(comp_v1->lcm_size);
1396
1397         for (i = 0; i < comp_cnt; i++) {
1398                 entry = &comp_v1->lcm_entries[i];
1399                 if (mirror_id_of(le32_to_cpu(entry->lcme_id)) == mirror_id) {
1400                         count++;
1401                         lmm_size_vic += sizeof(*entry);
1402                         lmm_size_vic += le32_to_cpu(entry->lcme_size);
1403                 } else if (count > 0) {
1404                         /* find the specified mirror */
1405                         break;
1406                 }
1407         }
1408
1409         if (count == 0)
1410                 return -EINVAL;
1411
1412         lu_buf_alloc(buf, lmm_size - lmm_size_vic);
1413         if (!buf->lb_buf)
1414                 return -ENOMEM;
1415
1416         lu_buf_alloc(buf_vic, sizeof(*comp_vic) + lmm_size_vic);
1417         if (!buf_vic->lb_buf) {
1418                 lu_buf_free(buf);
1419                 return -ENOMEM;
1420         }
1421
1422         comp_rem = (struct lov_comp_md_v1 *)buf->lb_buf;
1423         comp_vic = (struct lov_comp_md_v1 *)buf_vic->lb_buf;
1424
1425         memcpy(comp_rem, comp_v1, sizeof(*comp_v1));
1426         comp_rem->lcm_mirror_count = cpu_to_le16(mirror_cnt - 2);
1427         comp_rem->lcm_entry_count = cpu_to_le32(comp_cnt - count);
1428         comp_rem->lcm_size = cpu_to_le32(lmm_size - lmm_size_vic);
1429         if (!comp_rem->lcm_mirror_count)
1430                 comp_rem->lcm_flags = cpu_to_le16(LCM_FL_NONE);
1431
1432         memset(comp_vic, 0, sizeof(*comp_v1));
1433         comp_vic->lcm_magic = cpu_to_le32(LOV_MAGIC_COMP_V1);
1434         comp_vic->lcm_mirror_count = 0;
1435         comp_vic->lcm_entry_count = cpu_to_le32(count);
1436         comp_vic->lcm_size = cpu_to_le32(lmm_size_vic + sizeof(*comp_vic));
1437         comp_vic->lcm_flags = cpu_to_le16(LCM_FL_NONE);
1438         comp_vic->lcm_layout_gen = 0;
1439
1440         offset = sizeof(*comp_v1) + sizeof(*entry) * comp_cnt;
1441         offset_rem = sizeof(*comp_rem) +
1442                      sizeof(*entry_rem) * (comp_cnt - count);
1443         offset_vic = sizeof(*comp_vic) + sizeof(*entry_vic) * count;
1444         for (i = j = k = 0; i < comp_cnt; i++) {
1445                 struct lov_mds_md *lmm, *lmm_dst;
1446                 bool vic = false;
1447
1448                 entry = &comp_v1->lcm_entries[i];
1449                 entry_vic = &comp_vic->lcm_entries[j];
1450                 entry_rem = &comp_rem->lcm_entries[k];
1451
1452                 if (mirror_id_of(le32_to_cpu(entry->lcme_id)) == mirror_id)
1453                         vic = true;
1454
1455                 /* copy component entry */
1456                 if (vic) {
1457                         memcpy(entry_vic, entry, sizeof(*entry));
1458                         entry_vic->lcme_flags &= cpu_to_le32(LCME_FL_INIT);
1459                         entry_vic->lcme_offset = cpu_to_le32(offset_vic);
1460                         j++;
1461                 } else {
1462                         memcpy(entry_rem, entry, sizeof(*entry));
1463                         entry_rem->lcme_offset = cpu_to_le32(offset_rem);
1464                         k++;
1465                 }
1466
1467                 lmm = (struct lov_mds_md *)((char *)comp_v1 + offset);
1468                 if (vic)
1469                         lmm_dst = (struct lov_mds_md *)
1470                                         ((char *)comp_vic + offset_vic);
1471                 else
1472                         lmm_dst = (struct lov_mds_md *)
1473                                         ((char *)comp_rem + offset_rem);
1474
1475                 /* copy component entry blob */
1476                 memcpy(lmm_dst, lmm, le32_to_cpu(entry->lcme_size));
1477
1478                 /* blob offset advance */
1479                 offset += le32_to_cpu(entry->lcme_size);
1480                 if (vic)
1481                         offset_vic += le32_to_cpu(entry->lcme_size);
1482                 else
1483                         offset_rem += le32_to_cpu(entry->lcme_size);
1484         }
1485
1486         return 0;
1487 }
1488
1489 static int mdd_xattr_split(const struct lu_env *env, struct md_object *md_obj,
1490                            struct md_rejig_data *mrd)
1491 {
1492         struct mdd_device *mdd = mdo2mdd(md_obj);
1493         struct mdd_object *obj = md2mdd_obj(md_obj);
1494         struct mdd_object *vic = md2mdd_obj(mrd->mrd_obj);
1495         struct lu_buf *buf = &mdd_env_info(env)->mti_buf[0];
1496         struct lu_buf *buf_save = &mdd_env_info(env)->mti_buf[1];
1497         struct lu_buf *buf_vic = &mdd_env_info(env)->mti_buf[2];
1498         struct lov_comp_md_v1 *lcm;
1499         struct thandle *handle;
1500         int rc;
1501         ENTRY;
1502
1503         rc = lu_fid_cmp(mdo2fid(obj), mdo2fid(vic));
1504         if (rc == 0) /* same fid */
1505                 RETURN(-EPERM);
1506
1507         handle = mdd_trans_create(env, mdd);
1508         if (IS_ERR(handle))
1509                 RETURN(PTR_ERR(handle));
1510
1511         if (rc > 0) {
1512                 mdd_write_lock(env, obj, MOR_TGT_CHILD);
1513                 mdd_write_lock(env, vic, MOR_TGT_CHILD);
1514         } else {
1515                 mdd_write_lock(env, vic, MOR_TGT_CHILD);
1516                 mdd_write_lock(env, obj, MOR_TGT_CHILD);
1517         }
1518
1519         /* get EA of mirrored file */
1520         memset(buf_save, 0, sizeof(*buf));
1521         rc = mdd_get_lov_ea(env, obj, buf_save);
1522         if (rc < 0)
1523                 GOTO(out, rc);
1524
1525         lcm = buf_save->lb_buf;
1526         if (le32_to_cpu(lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
1527                 GOTO(out, rc = -EINVAL);
1528
1529         /**
1530          * Extract the mirror with specified mirror id, and store the splitted
1531          * mirror layout to the victim file.
1532          */
1533         memset(buf, 0, sizeof(*buf));
1534         memset(buf_vic, 0, sizeof(*buf_vic));
1535         rc = mdd_split_ea(lcm, mrd->mrd_mirror_id, buf, buf_vic);
1536         if (rc < 0)
1537                 GOTO(out, rc);
1538
1539         rc = mdd_declare_xattr_set(env, mdd, obj, buf, XATTR_NAME_LOV,
1540                                    LU_XATTR_SPLIT, handle);
1541         if (rc)
1542                 GOTO(out, rc);
1543         rc = mdd_declare_xattr_set(env, mdd, vic, buf_vic, XATTR_NAME_LOV,
1544                                    LU_XATTR_SPLIT, handle);
1545         if (rc)
1546                 GOTO(out, rc);
1547
1548         rc = mdd_trans_start(env, mdd, handle);
1549         if (rc)
1550                 GOTO(out, rc);
1551
1552         rc = mdo_xattr_set(env, obj, buf, XATTR_NAME_LOV, LU_XATTR_REPLACE,
1553                            handle);
1554         if (rc)
1555                 GOTO(out, rc);
1556
1557         rc = mdo_xattr_set(env, vic, buf_vic, XATTR_NAME_LOV, LU_XATTR_CREATE,
1558                            handle);
1559         if (rc)
1560                 GOTO(out_restore, rc);
1561
1562         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, obj, handle);
1563         if (rc)
1564                 GOTO(out, rc);
1565
1566         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, vic, handle);
1567         if (rc)
1568                 GOTO(out, rc);
1569         EXIT;
1570
1571 out_restore:
1572         if (rc) {
1573                 /* restore obj's layout */
1574                 int rc2 = mdo_xattr_set(env, obj, buf_save, XATTR_NAME_LOV,
1575                                         LU_XATTR_REPLACE, handle);
1576                 if (rc2)
1577                         CERROR("%s: failed to rollback of layout of: "DFID
1578                                ": %d, file state unkonwn.\n",
1579                                mdd_obj_dev_name(obj), PFID(mdo2fid(obj)), rc2);
1580         }
1581 out:
1582         mdd_trans_stop(env, mdd, rc, handle);
1583         mdd_write_unlock(env, obj);
1584         mdd_write_unlock(env, vic);
1585         lu_buf_free(buf_save);
1586         lu_buf_free(buf);
1587         lu_buf_free(buf_vic);
1588
1589         if (!rc)
1590                 (void) mdd_object_pfid_replace(env, obj);
1591
1592         return rc;
1593 }
1594
1595 static int mdd_layout_merge_allowed(const struct lu_env *env,
1596                                     struct md_object *target,
1597                                     struct md_object *victim)
1598 {
1599         struct mdd_object *o1 = md2mdd_obj(target);
1600
1601         /* cannot extend directory's LOVEA */
1602         if (S_ISDIR(mdd_object_type(o1))) {
1603                 CERROR("%s: Don't extend directory's LOVEA, just set it.\n",
1604                        mdd_obj_dev_name(o1));
1605                 RETURN(-EISDIR);
1606         }
1607
1608         RETURN(0);
1609 }
1610
1611 /**
1612  * The caller should guarantee to update the object ctime
1613  * after xattr_set if needed.
1614  */
1615 static int mdd_xattr_set(const struct lu_env *env, struct md_object *obj,
1616                          const struct lu_buf *buf, const char *name,
1617                          int fl)
1618 {
1619         struct mdd_object       *mdd_obj = md2mdd_obj(obj);
1620         struct lu_attr          *attr = MDD_ENV_VAR(env, cattr);
1621         struct mdd_device       *mdd = mdo2mdd(obj);
1622         struct thandle          *handle;
1623         enum changelog_rec_type  cl_type;
1624         int                      cl_flags = 0;
1625         int                      rc;
1626         ENTRY;
1627
1628         rc = mdd_la_get(env, mdd_obj, attr);
1629         if (rc)
1630                 RETURN(rc);
1631
1632         rc = mdd_xattr_sanity_check(env, mdd_obj, attr, name);
1633         if (rc)
1634                 RETURN(rc);
1635
1636         if (strcmp(name, XATTR_LUSTRE_LOV) == 0 &&
1637             (fl == LU_XATTR_MERGE || fl == LU_XATTR_SPLIT)) {
1638                 struct md_rejig_data *mrd = buf->lb_buf;
1639                 struct md_object *victim = mrd->mrd_obj;
1640
1641                 if (buf->lb_len != sizeof(*mrd))
1642                         RETURN(-EINVAL);
1643
1644                 rc = mdd_layout_merge_allowed(env, obj, victim);
1645                 if (rc)
1646                         RETURN(rc);
1647
1648                 if (fl == LU_XATTR_MERGE)
1649                         /* merge layout of victim as a mirror of obj's. */
1650                         rc = mdd_xattr_merge(env, obj, victim);
1651                 else
1652                         rc = mdd_xattr_split(env, obj, mrd);
1653                 RETURN(rc);
1654         }
1655
1656         if (strcmp(name, XATTR_NAME_ACL_ACCESS) == 0 ||
1657             strcmp(name, XATTR_NAME_ACL_DEFAULT) == 0) {
1658                 struct posix_acl *acl;
1659
1660                 /* user may set empty ACL, which should be treated as removing
1661                  * ACL. */
1662                 acl = posix_acl_from_xattr(&init_user_ns, buf->lb_buf,
1663                                            buf->lb_len);
1664                 if (IS_ERR(acl))
1665                         RETURN(PTR_ERR(acl));
1666                 if (acl == NULL) {
1667                         rc = mdd_xattr_del(env, obj, name);
1668                         RETURN(rc);
1669                 }
1670                 posix_acl_release(acl);
1671         }
1672
1673         if (!strcmp(name, XATTR_NAME_ACL_ACCESS)) {
1674                 rc = mdd_acl_set(env, mdd_obj, attr, buf, fl);
1675                 RETURN(rc);
1676         }
1677
1678         handle = mdd_trans_create(env, mdd);
1679         if (IS_ERR(handle))
1680                 RETURN(PTR_ERR(handle));
1681
1682         rc = mdd_declare_xattr_set(env, mdd, mdd_obj, buf, name, fl, handle);
1683         if (rc)
1684                 GOTO(stop, rc);
1685
1686         rc = mdd_trans_start(env, mdd, handle);
1687         if (rc)
1688                 GOTO(stop, rc);
1689
1690         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1691
1692         if (strcmp(XATTR_NAME_HSM, name) == 0) {
1693                 rc = mdd_hsm_update_locked(env, obj, buf, handle, &cl_flags);
1694                 if (rc) {
1695                         mdd_write_unlock(env, mdd_obj);
1696                         GOTO(stop, rc);
1697                 }
1698         }
1699
1700         rc = mdo_xattr_set(env, mdd_obj, buf, name, fl, handle);
1701         mdd_write_unlock(env, mdd_obj);
1702         if (rc)
1703                 GOTO(stop, rc);
1704
1705         cl_type = mdd_xattr_changelog_type(env, mdd, name);
1706         if (cl_type < 0)
1707                 GOTO(stop, rc = 0);
1708
1709         rc = mdd_changelog_data_store_xattr(env, mdd, cl_type, cl_flags,
1710                                             mdd_obj, name, handle);
1711
1712         EXIT;
1713 stop:
1714         return mdd_trans_stop(env, mdd, rc, handle);
1715 }
1716
1717 static int mdd_declare_xattr_del(const struct lu_env *env,
1718                                  struct mdd_device *mdd,
1719                                  struct mdd_object *obj,
1720                                  const char *name,
1721                                  struct thandle *handle)
1722 {
1723         int rc;
1724
1725         rc = mdo_declare_xattr_del(env, obj, name, handle);
1726         if (rc)
1727                 return rc;
1728
1729         if (mdd_xattr_changelog_type(env, mdd, name) < 0)
1730                 return 0; /* no changelog to store */
1731
1732         return mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
1733 }
1734
1735 /**
1736  * The caller should guarantee to update the object ctime
1737  * after xattr_set if needed.
1738  */
1739 static int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
1740                          const char *name)
1741 {
1742         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1743         struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
1744         struct mdd_device *mdd = mdo2mdd(obj);
1745         struct thandle *handle;
1746         int rc;
1747         ENTRY;
1748
1749         rc = mdd_la_get(env, mdd_obj, attr);
1750         if (rc)
1751                 RETURN(rc);
1752
1753         rc = mdd_xattr_sanity_check(env, mdd_obj, attr, name);
1754         if (rc)
1755                 RETURN(rc);
1756
1757         handle = mdd_trans_create(env, mdd);
1758         if (IS_ERR(handle))
1759                 RETURN(PTR_ERR(handle));
1760
1761         rc = mdd_declare_xattr_del(env, mdd, mdd_obj, name, handle);
1762         if (rc)
1763                 GOTO(stop, rc);
1764
1765         rc = mdd_trans_start(env, mdd, handle);
1766         if (rc)
1767                 GOTO(stop, rc);
1768
1769         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1770         rc = mdo_xattr_del(env, mdd_obj, name, handle);
1771         mdd_write_unlock(env, mdd_obj);
1772         if (rc)
1773                 GOTO(stop, rc);
1774
1775         if (mdd_xattr_changelog_type(env, mdd, name) < 0)
1776                 GOTO(stop, rc = 0);
1777
1778         rc = mdd_changelog_data_store_xattr(env, mdd, CL_SETXATTR, 0, mdd_obj,
1779                                             name, handle);
1780
1781         EXIT;
1782 stop:
1783         return mdd_trans_stop(env, mdd, rc, handle);
1784 }
1785
1786 /*
1787  * read lov EA of an object
1788  * return the lov EA in an allocated lu_buf
1789  */
1790 int mdd_get_lov_ea(const struct lu_env *env, struct mdd_object *obj,
1791                    struct lu_buf *lmm_buf)
1792 {
1793         struct lu_buf   *buf = &mdd_env_info(env)->mti_big_buf;
1794         int              rc, bufsize;
1795         ENTRY;
1796
1797 repeat:
1798         rc = mdo_xattr_get(env, obj, buf, XATTR_NAME_LOV);
1799
1800         if (rc == -ERANGE) {
1801                 /* mti_big_buf is allocated but is too small
1802                  * we need to increase it */
1803                 buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf,
1804                                              buf->lb_len * 2);
1805                 if (buf->lb_buf == NULL)
1806                         GOTO(out, rc = -ENOMEM);
1807                 goto repeat;
1808         }
1809
1810         if (rc < 0)
1811                 RETURN(rc);
1812
1813         if (rc == 0)
1814                 RETURN(-ENODATA);
1815
1816         bufsize = rc;
1817         if (memcmp(buf, &LU_BUF_NULL, sizeof(*buf)) == 0) {
1818                 /* mti_big_buf was not allocated, so we have to
1819                  * allocate it based on the ea size */
1820                 buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf,
1821                                              bufsize);
1822                 if (buf->lb_buf == NULL)
1823                         GOTO(out, rc = -ENOMEM);
1824                 goto repeat;
1825         }
1826
1827         lu_buf_alloc(lmm_buf, bufsize);
1828         if (lmm_buf->lb_buf == NULL)
1829                 GOTO(out, rc = -ENOMEM);
1830
1831         memcpy(lmm_buf->lb_buf, buf->lb_buf, bufsize);
1832         rc = 0;
1833         EXIT;
1834
1835 out:
1836         if (rc < 0)
1837                 lu_buf_free(lmm_buf);
1838         return rc;
1839 }
1840
1841 static int mdd_xattr_hsm_replace(const struct lu_env *env,
1842                                  struct mdd_object *o, struct lu_buf *buf,
1843                                  struct thandle *handle)
1844 {
1845         struct hsm_attrs *attrs;
1846         __u32 hsm_flags;
1847         int flags = 0;
1848         int rc;
1849         ENTRY;
1850
1851         rc = mdo_xattr_set(env, o, buf, XATTR_NAME_HSM, LU_XATTR_REPLACE,
1852                            handle);
1853         if (rc != 0)
1854                 RETURN(rc);
1855
1856         attrs = buf->lb_buf;
1857         hsm_flags = le32_to_cpu(attrs->hsm_flags);
1858         if (!(hsm_flags & HS_RELEASED) || mdd_is_dead_obj(o))
1859                 RETURN(0);
1860
1861         /* Add a changelog record for release. */
1862         hsm_set_cl_event(&flags, HE_RELEASE);
1863         rc = mdd_changelog_data_store(env, mdo2mdd(&o->mod_obj), CL_HSM,
1864                                       flags, o, handle);
1865         RETURN(rc);
1866 }
1867
1868 /*
1869  *  check if layout swapping between 2 objects is allowed
1870  *  the rules are:
1871  *  - only normal FIDs or non-system IGIFs
1872  *  - same type of objects
1873  *  - same owner/group (so quotas are still valid) unless this is from HSM
1874  *    release.
1875  */
1876 static int mdd_layout_swap_allowed(const struct lu_env *env,
1877                                    struct mdd_object *o1,
1878                                    const struct lu_attr *attr1,
1879                                    struct mdd_object *o2,
1880                                    const struct lu_attr *attr2,
1881                                    __u64 flags)
1882 {
1883         const struct lu_fid     *fid1, *fid2;
1884         ENTRY;
1885
1886         fid1 = mdo2fid(o1);
1887         fid2 = mdo2fid(o2);
1888
1889         if (!fid_is_norm(fid1) &&
1890             (!fid_is_igif(fid1) || IS_ERR(mdd_links_get(env, o1))))
1891                 RETURN(-EBADF);
1892
1893         if (!fid_is_norm(fid2) &&
1894             (!fid_is_igif(fid2) || IS_ERR(mdd_links_get(env, o2))))
1895                 RETURN(-EBADF);
1896
1897         if (mdd_object_type(o1) != mdd_object_type(o2)) {
1898                 if (S_ISDIR(mdd_object_type(o1)))
1899                         RETURN(-ENOTDIR);
1900                 if (S_ISREG(mdd_object_type(o1)))
1901                         RETURN(-EISDIR);
1902                 RETURN(-EBADF);
1903         }
1904
1905         if (flags & SWAP_LAYOUTS_MDS_HSM)
1906                 RETURN(0);
1907
1908         if ((attr1->la_uid != attr2->la_uid) ||
1909             (attr1->la_gid != attr2->la_gid))
1910                 RETURN(-EPERM);
1911
1912         RETURN(0);
1913 }
1914
1915 /* XXX To set the proper lmm_oi & lmm_layout_gen when swap layouts, we have to
1916  *     look into the layout in MDD layer. */
1917 static int mdd_lmm_oi(struct lov_mds_md *lmm, struct ost_id *oi, bool get)
1918 {
1919         struct lov_comp_md_v1   *comp_v1;
1920         struct lov_mds_md       *v1;
1921         int                      i, ent_count;
1922         __u32                    off;
1923
1924         if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_COMP_V1) {
1925                 comp_v1 = (struct lov_comp_md_v1 *)lmm;
1926                 ent_count = le16_to_cpu(comp_v1->lcm_entry_count);
1927
1928                 if (ent_count == 0)
1929                         return -EINVAL;
1930
1931                 if (get) {
1932                         off = le32_to_cpu(comp_v1->lcm_entries[0].lcme_offset);
1933                         v1 = (struct lov_mds_md *)((char *)comp_v1 + off);
1934                         *oi = v1->lmm_oi;
1935                 } else {
1936                         for (i = 0; i < le32_to_cpu(ent_count); i++) {
1937                                 off = le32_to_cpu(comp_v1->lcm_entries[i].
1938                                                 lcme_offset);
1939                                 v1 = (struct lov_mds_md *)((char *)comp_v1 +
1940                                                 off);
1941                                 v1->lmm_oi = *oi;
1942                         }
1943                 }
1944         } else if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_V1 ||
1945                    le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_V3) {
1946                 if (get)
1947                         *oi = lmm->lmm_oi;
1948                 else
1949                         lmm->lmm_oi = *oi;
1950         } else {
1951                 return -EINVAL;
1952         }
1953         return 0;
1954 }
1955
1956 static inline int mdd_get_lmm_oi(struct lov_mds_md *lmm, struct ost_id *oi)
1957 {
1958         return mdd_lmm_oi(lmm, oi, true);
1959 }
1960
1961 static inline int mdd_set_lmm_oi(struct lov_mds_md *lmm, struct ost_id *oi)
1962 {
1963         return mdd_lmm_oi(lmm, oi, false);
1964 }
1965
1966 static int mdd_lmm_gen(struct lov_mds_md *lmm, __u32 *gen, bool get)
1967 {
1968         struct lov_comp_md_v1 *comp_v1;
1969
1970         if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_COMP_V1) {
1971                 comp_v1 = (struct lov_comp_md_v1 *)lmm;
1972                 if (get)
1973                         *gen = le32_to_cpu(comp_v1->lcm_layout_gen);
1974                 else
1975                         comp_v1->lcm_layout_gen = cpu_to_le32(*gen);
1976         } else if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_V1 ||
1977                    le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_V3) {
1978                 __u16 tmp_gen = *gen;
1979                 if (get)
1980                         *gen = le16_to_cpu(lmm->lmm_layout_gen);
1981                 else
1982                         lmm->lmm_layout_gen = cpu_to_le16(tmp_gen);
1983         } else {
1984                 return -EINVAL;
1985         }
1986         return 0;
1987 }
1988
1989 static inline int mdd_get_lmm_gen(struct lov_mds_md *lmm, __u32 *gen)
1990 {
1991         return mdd_lmm_gen(lmm, gen, true);
1992 }
1993
1994 static inline int mdd_set_lmm_gen(struct lov_mds_md *lmm, __u32 *gen)
1995 {
1996         return mdd_lmm_gen(lmm, gen, false);
1997 }
1998
1999 /**
2000  * swap layouts between 2 lustre objects
2001  */
2002 static int mdd_swap_layouts(const struct lu_env *env, struct md_object *obj1,
2003                             struct md_object *obj2, __u64 flags)
2004 {
2005         struct mdd_thread_info  *info = mdd_env_info(env);
2006         struct mdd_object       *fst_o = md2mdd_obj(obj1);
2007         struct mdd_object       *snd_o = md2mdd_obj(obj2);
2008         struct lu_attr          *fst_la = MDD_ENV_VAR(env, cattr);
2009         struct lu_attr          *snd_la = MDD_ENV_VAR(env, tattr);
2010         struct mdd_device       *mdd = mdo2mdd(obj1);
2011         struct lov_mds_md       *fst_lmm, *snd_lmm;
2012         struct lu_buf           *fst_buf = &info->mti_buf[0];
2013         struct lu_buf           *snd_buf = &info->mti_buf[1];
2014         struct lu_buf           *fst_hsm_buf = &info->mti_buf[2];
2015         struct lu_buf           *snd_hsm_buf = &info->mti_buf[3];
2016         struct ost_id           *saved_oi = NULL;
2017         struct thandle          *handle;
2018         __u32                    fst_gen, snd_gen, saved_gen;
2019         int                      fst_fl;
2020         int                      rc;
2021         int                      rc2;
2022         ENTRY;
2023
2024         CLASSERT(ARRAY_SIZE(info->mti_buf) >= 4);
2025         memset(info->mti_buf, 0, sizeof(info->mti_buf));
2026
2027         /* we have to sort the 2 obj, so locking will always
2028          * be in the same order, even in case of 2 concurrent swaps */
2029         rc = lu_fid_cmp(mdo2fid(fst_o), mdo2fid(snd_o));
2030         if (rc == 0) /* same fid ? */
2031                 RETURN(-EPERM);
2032
2033         if (rc < 0)
2034                 swap(fst_o, snd_o);
2035
2036         rc = mdd_la_get(env, fst_o, fst_la);
2037         if (rc != 0)
2038                 RETURN(rc);
2039
2040         rc = mdd_la_get(env, snd_o, snd_la);
2041         if (rc != 0)
2042                 RETURN(rc);
2043
2044         /* check if layout swapping is allowed */
2045         rc = mdd_layout_swap_allowed(env, fst_o, fst_la, snd_o, snd_la, flags);
2046         if (rc != 0)
2047                 RETURN(rc);
2048
2049         handle = mdd_trans_create(env, mdd);
2050         if (IS_ERR(handle))
2051                 RETURN(PTR_ERR(handle));
2052
2053         /* objects are already sorted */
2054         mdd_write_lock(env, fst_o, MOR_TGT_CHILD);
2055         mdd_write_lock(env, snd_o, MOR_TGT_CHILD);
2056
2057         rc = mdd_get_lov_ea(env, fst_o, fst_buf);
2058         if (rc < 0 && rc != -ENODATA)
2059                 GOTO(stop, rc);
2060
2061         rc = mdd_get_lov_ea(env, snd_o, snd_buf);
2062         if (rc < 0 && rc != -ENODATA)
2063                 GOTO(stop, rc);
2064
2065         /* swapping 2 non existant layouts is a success */
2066         if (fst_buf->lb_buf == NULL && snd_buf->lb_buf == NULL)
2067                 GOTO(stop, rc = 0);
2068
2069         /* to help inode migration between MDT, it is better to
2070          * start by the no layout file (if one), so we order the swap */
2071         if (snd_buf->lb_buf == NULL) {
2072                 swap(fst_o, snd_o);
2073                 swap(fst_buf, snd_buf);
2074         }
2075
2076         fst_gen = snd_gen = 0;
2077         /* lmm and generation layout initialization */
2078         if (fst_buf->lb_buf != NULL) {
2079                 fst_lmm = fst_buf->lb_buf;
2080                 mdd_get_lmm_gen(fst_lmm, &fst_gen);
2081                 fst_fl  = LU_XATTR_REPLACE;
2082         } else {
2083                 fst_lmm = NULL;
2084                 fst_gen = 0;
2085                 fst_fl  = LU_XATTR_CREATE;
2086         }
2087
2088         snd_lmm = snd_buf->lb_buf;
2089         mdd_get_lmm_gen(snd_lmm, &snd_gen);
2090
2091         saved_gen = fst_gen;
2092         /* increase the generation layout numbers */
2093         snd_gen++;
2094         fst_gen++;
2095
2096         /*
2097          * XXX The layout generation is used to generate component IDs for
2098          *     the composite file, we have to do some special tweaks to make
2099          *     sure the layout generation is always adequate for that job.
2100          */
2101
2102         /* Skip invalid generation number for composite layout */
2103         if ((snd_gen & LCME_ID_MASK) == 0)
2104                 snd_gen++;
2105         if ((fst_gen & LCME_ID_MASK) == 0)
2106                 fst_gen++;
2107         /* Make sure the generation is greater than all the component IDs */
2108         if (fst_gen < snd_gen)
2109                 fst_gen = snd_gen;
2110         else if (fst_gen > snd_gen)
2111                 snd_gen = fst_gen;
2112
2113         /* set the file specific informations in lmm */
2114         if (fst_lmm != NULL) {
2115                 saved_oi = &info->mti_oa.o_oi;
2116                 mdd_get_lmm_oi(fst_lmm, saved_oi);
2117                 mdd_set_lmm_gen(fst_lmm, &snd_gen);
2118                 mdd_set_lmm_oi(fst_lmm, &snd_lmm->lmm_oi);
2119                 mdd_set_lmm_oi(snd_lmm, saved_oi);
2120         } else {
2121                 if ((snd_lmm->lmm_magic & cpu_to_le32(LOV_MAGIC_MASK)) ==
2122                     cpu_to_le32(LOV_MAGIC_MAGIC))
2123                         snd_lmm->lmm_magic |= cpu_to_le32(LOV_MAGIC_DEFINED);
2124                 else
2125                         GOTO(stop, rc = -EPROTO);
2126         }
2127         mdd_set_lmm_gen(snd_lmm, &fst_gen);
2128
2129         /* Prepare HSM attribute if it's required */
2130         if (flags & SWAP_LAYOUTS_MDS_HSM) {
2131                 const int buflen = sizeof(struct hsm_attrs);
2132
2133                 lu_buf_alloc(fst_hsm_buf, buflen);
2134                 lu_buf_alloc(snd_hsm_buf, buflen);
2135                 if (fst_hsm_buf->lb_buf == NULL || snd_hsm_buf->lb_buf == NULL)
2136                         GOTO(stop, rc = -ENOMEM);
2137
2138                 /* Read HSM attribute */
2139                 rc = mdo_xattr_get(env, fst_o, fst_hsm_buf, XATTR_NAME_HSM);
2140                 if (rc < 0)
2141                         GOTO(stop, rc);
2142
2143                 rc = mdo_xattr_get(env, snd_o, snd_hsm_buf, XATTR_NAME_HSM);
2144                 if (rc < 0)
2145                         GOTO(stop, rc);
2146
2147                 rc = mdd_declare_xattr_set(env, mdd, fst_o, snd_hsm_buf,
2148                                            XATTR_NAME_HSM, LU_XATTR_REPLACE,
2149                                            handle);
2150                 if (rc < 0)
2151                         GOTO(stop, rc);
2152
2153                 rc = mdd_declare_xattr_set(env, mdd, snd_o, fst_hsm_buf,
2154                                            XATTR_NAME_HSM, LU_XATTR_REPLACE,
2155                                            handle);
2156                 if (rc < 0)
2157                         GOTO(stop, rc);
2158         }
2159
2160         /* prepare transaction */
2161         rc = mdd_declare_xattr_set(env, mdd, fst_o, snd_buf, XATTR_NAME_LOV,
2162                                    fst_fl, handle);
2163         if (rc != 0)
2164                 GOTO(stop, rc);
2165
2166         if (fst_buf->lb_buf != NULL)
2167                 rc = mdd_declare_xattr_set(env, mdd, snd_o, fst_buf,
2168                                            XATTR_NAME_LOV, LU_XATTR_REPLACE,
2169                                            handle);
2170         else
2171                 rc = mdd_declare_xattr_del(env, mdd, snd_o, XATTR_NAME_LOV,
2172                                            handle);
2173         if (rc != 0)
2174                 GOTO(stop, rc);
2175
2176         rc = mdd_trans_start(env, mdd, handle);
2177         if (rc != 0)
2178                 GOTO(stop, rc);
2179
2180         if (flags & SWAP_LAYOUTS_MDS_HSM) {
2181                 rc = mdd_xattr_hsm_replace(env, fst_o, snd_hsm_buf, handle);
2182                 if (rc < 0)
2183                         GOTO(stop, rc);
2184
2185                 rc = mdd_xattr_hsm_replace(env, snd_o, fst_hsm_buf, handle);
2186                 if (rc < 0) {
2187                         rc2 = mdd_xattr_hsm_replace(env, fst_o, fst_hsm_buf,
2188                                                     handle);
2189                         if (rc2 < 0)
2190                                 CERROR("%s: restore "DFID" HSM error: %d/%d\n",
2191                                        mdd_obj_dev_name(fst_o),
2192                                        PFID(mdo2fid(fst_o)), rc, rc2);
2193                         GOTO(stop, rc);
2194                 }
2195         }
2196
2197         rc = mdo_xattr_set(env, fst_o, snd_buf, XATTR_NAME_LOV, fst_fl, handle);
2198         if (rc != 0)
2199                 GOTO(stop, rc);
2200
2201         if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_MDS_HSM_SWAP_LAYOUTS))) {
2202                 rc = -EOPNOTSUPP;
2203         } else {
2204                 if (fst_buf->lb_buf != NULL)
2205                         rc = mdo_xattr_set(env, snd_o, fst_buf, XATTR_NAME_LOV,
2206                                            LU_XATTR_REPLACE, handle);
2207                 else
2208                         rc = mdo_xattr_del(env, snd_o, XATTR_NAME_LOV, handle);
2209         }
2210         if (rc != 0)
2211                 GOTO(out_restore, rc);
2212
2213         /* Issue one changelog record per file */
2214         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, fst_o, handle);
2215         if (rc)
2216                 GOTO(stop, rc);
2217
2218         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, snd_o, handle);
2219         if (rc)
2220                 GOTO(stop, rc);
2221         EXIT;
2222
2223 out_restore:
2224         if (rc != 0) {
2225                 int steps = 0;
2226
2227                 /* failure on second file, but first was done, so we have
2228                  * to roll back first. */
2229                 if (fst_buf->lb_buf != NULL) {
2230                         mdd_set_lmm_oi(fst_lmm, saved_oi);
2231                         mdd_set_lmm_gen(fst_lmm, &saved_gen);
2232                         rc2 = mdo_xattr_set(env, fst_o, fst_buf, XATTR_NAME_LOV,
2233                                             LU_XATTR_REPLACE, handle);
2234                 } else {
2235                         rc2 = mdo_xattr_del(env, fst_o, XATTR_NAME_LOV, handle);
2236                 }
2237                 if (rc2 < 0)
2238                         goto do_lbug;
2239
2240                 ++steps;
2241                 rc2 = mdd_xattr_hsm_replace(env, fst_o, fst_hsm_buf, handle);
2242                 if (rc2 < 0)
2243                         goto do_lbug;
2244
2245                 ++steps;
2246                 rc2 = mdd_xattr_hsm_replace(env, snd_o, snd_hsm_buf, handle);
2247
2248         do_lbug:
2249                 if (rc2 < 0) {
2250                         /* very bad day */
2251                         CERROR("%s: unable to roll back layout swap. FIDs: "
2252                                DFID" and "DFID "error: %d/%d, steps: %d\n",
2253                                mdd_obj_dev_name(fst_o),
2254                                PFID(mdo2fid(snd_o)), PFID(mdo2fid(fst_o)),
2255                                rc, rc2, steps);
2256                         /* a solution to avoid journal commit is to panic,
2257                          * but it has strong consequences so we use LBUG to
2258                          * allow sysdamin to choose to panic or not
2259                          */
2260                         LBUG();
2261                 }
2262         }
2263
2264 stop:
2265         rc = mdd_trans_stop(env, mdd, rc, handle);
2266
2267         mdd_write_unlock(env, snd_o);
2268         mdd_write_unlock(env, fst_o);
2269
2270         lu_buf_free(fst_buf);
2271         lu_buf_free(snd_buf);
2272         lu_buf_free(fst_hsm_buf);
2273         lu_buf_free(snd_hsm_buf);
2274
2275         if (!rc) {
2276                 (void) mdd_object_pfid_replace(env, fst_o);
2277                 (void) mdd_object_pfid_replace(env, snd_o);
2278         }
2279         return rc;
2280 }
2281
2282 static int mdd_declare_layout_change(const struct lu_env *env,
2283                                      struct mdd_device *mdd,
2284                                      struct mdd_object *obj,
2285                                      struct md_layout_change *mlc,
2286                                      struct thandle *handle)
2287 {
2288         int rc;
2289
2290         rc = mdo_declare_layout_change(env, obj, mlc, handle);
2291         if (rc)
2292                 return rc;
2293
2294         return mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
2295 }
2296
2297 /* For PFL, this is used to instantiate necessary component objects. */
2298 static int
2299 mdd_layout_instantiate_component(const struct lu_env *env,
2300                 struct mdd_object *obj, struct md_layout_change *mlc,
2301                 struct thandle *handle)
2302 {
2303         struct mdd_device *mdd = mdd_obj2mdd_dev(obj);
2304         int rc;
2305         ENTRY;
2306
2307         if (mlc->mlc_opc != MD_LAYOUT_WRITE)
2308                 RETURN(-ENOTSUPP);
2309
2310         rc = mdd_declare_layout_change(env, mdd, obj, mlc, handle);
2311         /**
2312          * It's possible that another layout write intent has already
2313          * instantiated our objects, so a -EALREADY returned, and we need to
2314          * do nothing.
2315          */
2316         if (rc)
2317                 RETURN(rc == -EALREADY ? 0 : rc);
2318
2319         rc = mdd_trans_start(env, mdd, handle);
2320         if (rc)
2321                 RETURN(rc);
2322
2323         mdd_write_lock(env, obj, MOR_TGT_CHILD);
2324         rc = mdo_layout_change(env, obj, mlc, handle);
2325         mdd_write_unlock(env, obj);
2326         if (rc)
2327                 RETURN(rc);
2328
2329         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, obj, handle);
2330         RETURN(rc);
2331 }
2332
2333 /**
2334  * Change the FLR layout from RDONLY to WRITE_PENDING.
2335  *
2336  * It picks the primary mirror, and bumps the layout version, and set
2337  * layout version xattr to OST objects in a sync tx. In order to facilitate
2338  * the handling of phantom writers from evicted clients, the clients carry
2339  * layout version of the file with write RPC, so that the OSTs can verify
2340  * if the write RPCs are legitimate, meaning not from evicted clients.
2341  */
2342 static int
2343 mdd_layout_update_rdonly(const struct lu_env *env, struct mdd_object *obj,
2344                          struct md_layout_change *mlc, struct thandle *handle)
2345 {
2346         struct mdd_device *mdd = mdd_obj2mdd_dev(obj);
2347         int rc;
2348         ENTRY;
2349
2350         /* Verify acceptable operations */
2351         switch (mlc->mlc_opc) {
2352         case MD_LAYOUT_WRITE:
2353                 break;
2354         case MD_LAYOUT_RESYNC:
2355                 /* these are legal operations - this represents the case that
2356                  * a few mirrors were missed in the last resync.
2357                  * XXX: it will be supported later */
2358         case MD_LAYOUT_RESYNC_DONE:
2359         default:
2360                 RETURN(0);
2361         }
2362
2363         rc = mdd_declare_layout_change(env, mdd, obj, mlc, handle);
2364         if (rc)
2365                 GOTO(out, rc);
2366
2367         rc = mdd_declare_xattr_del(env, mdd, obj, XATTR_NAME_SOM, handle);
2368         if (rc)
2369                 GOTO(out, rc);
2370
2371         /* record a changelog for data mover to consume */
2372         rc = mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
2373         if (rc)
2374                 GOTO(out, rc);
2375
2376         rc = mdd_trans_start(env, mdd, handle);
2377         if (rc)
2378                 GOTO(out, rc);
2379
2380         /* it needs a sync tx to make FLR to work properly */
2381         handle->th_sync = 1;
2382
2383         mdd_write_lock(env, obj, MOR_TGT_CHILD);
2384         rc = mdo_layout_change(env, obj, mlc, handle);
2385         if (!rc) {
2386                 rc = mdo_xattr_del(env, obj, XATTR_NAME_SOM, handle);
2387                 if (rc == -ENODATA)
2388                         rc = 0;
2389         }
2390         mdd_write_unlock(env, obj);
2391         if (rc)
2392                 GOTO(out, rc);
2393
2394         rc = mdd_changelog_data_store(env, mdd, CL_FLRW, 0, obj, handle);
2395         if (rc)
2396                 GOTO(out, rc);
2397
2398         EXIT;
2399
2400 out:
2401         return rc;
2402 }
2403
2404 /**
2405  * Handle mirrored file state transition when it's in WRITE_PENDING.
2406  *
2407  * Only MD_LAYOUT_RESYNC, which represents start of resync, is allowed when
2408  * the file is in WRITE_PENDING state. If everything goes fine, the file's
2409  * layout version will be increased, and the file's state will be changed to
2410  * SYNC_PENDING.
2411  */
2412 static int
2413 mdd_layout_update_write_pending(const struct lu_env *env,
2414                 struct mdd_object *obj, struct md_layout_change *mlc,
2415                 struct thandle *handle)
2416 {
2417         struct mdd_device *mdd = mdd_obj2mdd_dev(obj);
2418         int rc;
2419         ENTRY;
2420
2421         switch (mlc->mlc_opc) {
2422         case MD_LAYOUT_RESYNC:
2423                 /* Upon receiving the resync request, it should
2424                  * instantiate all stale components right away to get ready
2425                  * for mirror copy. In order to avoid layout version change,
2426                  * client should avoid sending LAYOUT_WRITE request at the
2427                  * resync state. */
2428                 break;
2429         case MD_LAYOUT_WRITE:
2430                 /* legal race for concurrent write, the file state has been
2431                  * changed by another client. */
2432                 break;
2433         default:
2434                 RETURN(-EBUSY);
2435         }
2436
2437         rc = mdd_declare_layout_change(env, mdd, obj, mlc, handle);
2438         if (rc)
2439                 GOTO(out, rc);
2440
2441         rc = mdd_trans_start(env, mdd, handle);
2442         if (rc)
2443                 GOTO(out, rc);
2444
2445         /* it needs a sync tx to make FLR to work properly */
2446         handle->th_sync = 1;
2447
2448         mdd_write_lock(env, obj, MOR_TGT_CHILD);
2449         rc = mdo_layout_change(env, obj, mlc, handle);
2450         mdd_write_unlock(env, obj);
2451         if (rc)
2452                 GOTO(out, rc);
2453
2454         EXIT;
2455
2456 out:
2457         return rc;
2458 }
2459
2460 /**
2461  * Handle the requests when a FLR file's state is in SYNC_PENDING.
2462  *
2463  * Only concurrent write and sync complete requests are possible when the
2464  * file is in SYNC_PENDING. For the latter request, it will pass in the
2465  * mirrors that have been synchronized, then the stale bit will be cleared
2466  * to make the file's state turn into RDONLY.
2467  * For concurrent write reqeust, it just needs to change the file's state
2468  * to WRITE_PENDING in a sync tx. It doesn't have to change the layout
2469  * version because the version will be increased in the transition to
2470  * SYNC_PENDING later so that it can deny the write request from potential
2471  * evicted SYNC clients. */
2472 static int
2473 mdd_object_update_sync_pending(const struct lu_env *env, struct mdd_object *obj,
2474                 struct md_layout_change *mlc, struct thandle *handle)
2475 {
2476         struct mdd_device *mdd = mdd_obj2mdd_dev(obj);
2477         struct lu_buf *som_buf = &mdd_env_info(env)->mti_buf[1];
2478         int fl = 0;
2479         int rc;
2480         ENTRY;
2481
2482         /* operation validation */
2483         switch (mlc->mlc_opc) {
2484         case MD_LAYOUT_RESYNC_DONE:
2485                 /* resync complete. */
2486         case MD_LAYOUT_WRITE:
2487                 /* concurrent write. */
2488                 break;
2489         case MD_LAYOUT_RESYNC:
2490                 /* resync again, most likely the previous run failed.
2491                  * no-op if it's already in SYNC_PENDING state */
2492                 RETURN(0);
2493         default:
2494                 RETURN(-EBUSY);
2495         }
2496
2497         if (mlc->mlc_som.lsa_valid & LSOM_FL_VALID) {
2498                 rc = mdo_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_SOM);
2499                 if (rc && rc != -ENODATA)
2500                         RETURN(rc);
2501
2502                 fl = rc == -ENODATA ? LU_XATTR_CREATE : LU_XATTR_REPLACE;
2503                 som_buf->lb_buf = &mlc->mlc_som;
2504                 som_buf->lb_len = sizeof(mlc->mlc_som);
2505         }
2506
2507         rc = mdd_declare_layout_change(env, mdd, obj, mlc, handle);
2508         if (rc)
2509                 GOTO(out, rc);
2510
2511         /* record a changelog for the completion of resync */
2512         rc = mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
2513         if (rc)
2514                 GOTO(out, rc);
2515
2516         /* RESYNC_DONE has piggybacked size and blocks */
2517         if (fl) {
2518                 rc = mdd_declare_xattr_set(env, mdd, obj, som_buf,
2519                                            XATTR_NAME_SOM, fl, handle);
2520                 if (rc)
2521                         GOTO(out, rc);
2522         }
2523
2524         rc = mdd_trans_start(env, mdd, handle);
2525         if (rc)
2526                 GOTO(out, rc);
2527
2528         /* it needs a sync tx to make FLR to work properly */
2529         handle->th_sync = 1;
2530
2531         rc = mdo_layout_change(env, obj, mlc, handle);
2532         if (rc)
2533                 GOTO(out, rc);
2534
2535         if (fl) {
2536                 rc = mdo_xattr_set(env, obj, som_buf, XATTR_NAME_SOM,
2537                                    fl, handle);
2538                 if (rc)
2539                         GOTO(out, rc);
2540         }
2541
2542         rc = mdd_changelog_data_store(env, mdd, CL_RESYNC, 0, obj, handle);
2543         if (rc)
2544                 GOTO(out, rc);
2545         EXIT;
2546 out:
2547         return rc;
2548 }
2549
2550 /**
2551  * Layout change callback for object.
2552  *
2553  * This is only used by FLR for now. In the future, it can be exteneded to
2554  * handle all layout change.
2555  */
2556 static int
2557 mdd_layout_change(const struct lu_env *env, struct md_object *o,
2558                   struct md_layout_change *mlc)
2559 {
2560         struct mdd_object       *obj = md2mdd_obj(o);
2561         struct mdd_device       *mdd = mdd_obj2mdd_dev(obj);
2562         struct lu_buf           *buf = mdd_buf_get(env, NULL, 0);
2563         struct lov_comp_md_v1   *lcm;
2564         struct thandle          *handle;
2565         int flr_state;
2566         int rc;
2567         ENTRY;
2568
2569         /* Verify acceptable operations */
2570         switch (mlc->mlc_opc) {
2571         case MD_LAYOUT_WRITE:
2572         case MD_LAYOUT_RESYNC:
2573         case MD_LAYOUT_RESYNC_DONE:
2574                 break;
2575         default:
2576                 RETURN(-ENOTSUPP);
2577         }
2578
2579         handle = mdd_trans_create(env, mdd);
2580         if (IS_ERR(handle))
2581                 RETURN(PTR_ERR(handle));
2582
2583         rc = mdd_get_lov_ea(env, obj, buf);
2584         if (rc < 0) {
2585                 if (rc == -ENODATA)
2586                         rc = -EINVAL;
2587                 GOTO(out, rc);
2588         }
2589
2590         /* analyze the layout to make sure it's a FLR file */
2591         lcm = buf->lb_buf;
2592         if (le32_to_cpu(lcm->lcm_magic) != LOV_MAGIC_COMP_V1)
2593                 GOTO(out, rc = -EINVAL);
2594
2595         flr_state = le16_to_cpu(lcm->lcm_flags) & LCM_FL_FLR_MASK;
2596
2597         /* please refer to HLD of FLR for state transition */
2598         switch (flr_state) {
2599         case LCM_FL_NONE:
2600                 rc = mdd_layout_instantiate_component(env, obj, mlc, handle);
2601                 break;
2602         case LCM_FL_WRITE_PENDING:
2603                 rc = mdd_layout_update_write_pending(env, obj, mlc, handle);
2604                 break;
2605         case LCM_FL_RDONLY:
2606                 rc = mdd_layout_update_rdonly(env, obj, mlc, handle);
2607                 break;
2608         case LCM_FL_SYNC_PENDING:
2609                 rc = mdd_object_update_sync_pending(env, obj, mlc, handle);
2610                 break;
2611         default:
2612                 rc = 0;
2613                 break;
2614         }
2615         EXIT;
2616
2617 out:
2618         mdd_trans_stop(env, mdd, rc, handle);
2619         lu_buf_free(buf);
2620         return rc;
2621 }
2622
2623 void mdd_object_make_hint(const struct lu_env *env, struct mdd_object *parent,
2624                           struct mdd_object *child, const struct lu_attr *attr,
2625                           const struct md_op_spec *spec,
2626                           struct dt_allocation_hint *hint)
2627 {
2628         struct dt_object *np = parent ?  mdd_object_child(parent) : NULL;
2629         struct dt_object *nc = mdd_object_child(child);
2630
2631         memset(hint, 0, sizeof(*hint));
2632
2633         /* For striped directory, give striping EA to lod_ah_init, which will
2634          * decide the stripe_offset and stripe count by it. */
2635         if (S_ISDIR(attr->la_mode) &&
2636             unlikely(spec != NULL && spec->sp_cr_flags & MDS_OPEN_HAS_EA)) {
2637                 hint->dah_eadata = spec->u.sp_ea.eadata;
2638                 hint->dah_eadata_len = spec->u.sp_ea.eadatalen;
2639         } else {
2640                 hint->dah_eadata = NULL;
2641                 hint->dah_eadata_len = 0;
2642         }
2643
2644         CDEBUG(D_INFO, DFID" eadata %p len %d\n", PFID(mdd_object_fid(child)),
2645                hint->dah_eadata, hint->dah_eadata_len);
2646         /* @hint will be initialized by underlying device. */
2647         nc->do_ops->do_ah_init(env, hint, np, nc, attr->la_mode & S_IFMT);
2648 }
2649
2650 /*
2651  * do NOT or the MAY_*'s, you'll get the weakest
2652  */
2653 int accmode(const struct lu_env *env, const struct lu_attr *la, int flags)
2654 {
2655         int res = 0;
2656
2657         /* Sadly, NFSD reopens a file repeatedly during operation, so the
2658          * "acc_mode = 0" allowance for newly-created files isn't honoured.
2659          * NFSD uses the MDS_OPEN_OWNEROVERRIDE flag to say that a file
2660          * owner can write to a file even if it is marked readonly to hide
2661          * its brokenness. (bug 5781) */
2662         if (flags & MDS_OPEN_OWNEROVERRIDE) {
2663                 struct lu_ucred *uc = lu_ucred_check(env);
2664
2665                 if ((uc == NULL) || (la->la_uid == uc->uc_fsuid))
2666                         return 0;
2667         }
2668
2669         if (flags & FMODE_READ)
2670                 res |= MAY_READ;
2671         if (flags & (FMODE_WRITE | MDS_OPEN_TRUNC | MDS_OPEN_APPEND))
2672                 res |= MAY_WRITE;
2673         if (flags & MDS_FMODE_EXEC)
2674                 res = MAY_EXEC;
2675         return res;
2676 }
2677
2678 static int mdd_open_sanity_check(const struct lu_env *env,
2679                                 struct mdd_object *obj,
2680                                 const struct lu_attr *attr, int flag)
2681 {
2682         int mode, rc;
2683         ENTRY;
2684
2685         /* EEXIST check */
2686         if (mdd_is_dead_obj(obj))
2687                 RETURN(-ENOENT);
2688
2689         if (S_ISLNK(attr->la_mode))
2690                 RETURN(-ELOOP);
2691
2692         mode = accmode(env, attr, flag);
2693
2694         if (S_ISDIR(attr->la_mode) && (mode & MAY_WRITE))
2695                 RETURN(-EISDIR);
2696
2697         if (!(flag & MDS_OPEN_CREATED)) {
2698                 rc = mdd_permission_internal(env, obj, attr, mode);
2699                 if (rc)
2700                         RETURN(rc);
2701         }
2702
2703         if (S_ISFIFO(attr->la_mode) || S_ISSOCK(attr->la_mode) ||
2704             S_ISBLK(attr->la_mode) || S_ISCHR(attr->la_mode))
2705                 flag &= ~MDS_OPEN_TRUNC;
2706
2707         /* For writing append-only file must open it with append mode. */
2708         if (attr->la_flags & LUSTRE_APPEND_FL) {
2709                 if ((flag & FMODE_WRITE) && !(flag & MDS_OPEN_APPEND))
2710                         RETURN(-EPERM);
2711                 if (flag & MDS_OPEN_TRUNC)
2712                         RETURN(-EPERM);
2713         }
2714
2715         RETURN(0);
2716 }
2717
2718 static int mdd_open(const struct lu_env *env, struct md_object *obj,
2719                     int flags)
2720 {
2721         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2722         struct md_device *md_dev = lu2md_dev(mdd2lu_dev(mdo2mdd(obj)));
2723         struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
2724         int rc = 0;
2725
2726         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
2727
2728         rc = mdd_la_get(env, mdd_obj, attr);
2729         if (rc != 0)
2730                 GOTO(out, rc);
2731
2732         rc = mdd_open_sanity_check(env, mdd_obj, attr, flags);
2733         if (rc != 0)
2734                 GOTO(out, rc);
2735
2736         mdd_obj->mod_count++;
2737
2738         mdd_changelog(env, CL_OPEN, flags, md_dev, mdo2fid(mdd_obj));
2739
2740         EXIT;
2741 out:
2742         mdd_write_unlock(env, mdd_obj);
2743         return rc;
2744 }
2745
2746 static int mdd_declare_close(const struct lu_env *env,
2747                              struct mdd_object *obj,
2748                              struct md_attr *ma,
2749                              struct thandle *handle)
2750 {
2751         int rc;
2752
2753         rc = orph_declare_index_delete(env, obj, handle);
2754         if (rc)
2755                 return rc;
2756
2757         return mdo_declare_destroy(env, obj, handle);
2758 }
2759
2760 /*
2761  * No permission check is needed.
2762  */
2763 static int mdd_close(const struct lu_env *env, struct md_object *obj,
2764                      struct md_attr *ma, int mode)
2765 {
2766         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2767         struct mdd_device *mdd = mdo2mdd(obj);
2768         struct thandle *handle = NULL;
2769         int is_orphan = 0;
2770         int rc;
2771         bool blocked = false;
2772         ENTRY;
2773
2774         if (ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_KEEP_ORPHAN) {
2775                 mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
2776                 mdd_obj->mod_count--;
2777                 mdd_write_unlock(env, mdd_obj);
2778
2779                 if (mdd_obj->mod_flags & ORPHAN_OBJ && !mdd_obj->mod_count)
2780                         CDEBUG(D_HA, "Object "DFID" is retained in orphan "
2781                                 "list\n", PFID(mdd_object_fid(mdd_obj)));
2782                 RETURN(0);
2783         }
2784
2785         /* mdd_finish_unlink() will always set orphan object as DEAD_OBJ, but
2786          * it might fail to add the object to orphan list (w/o ORPHAN_OBJ). */
2787         /* check without any lock */
2788         is_orphan = mdd_obj->mod_count == 1 &&
2789                     (mdd_obj->mod_flags & (ORPHAN_OBJ | DEAD_OBJ)) != 0;
2790
2791 again:
2792         if (is_orphan) {
2793                 /* mdd_trans_create() maybe failed because of barrier_entry(),
2794                  * under such case, the orphan MDT-object will be left in the
2795                  * orphan list, and when the MDT remount next time, the unused
2796                  * orphans will be destroyed automatically.
2797                  *
2798                  * One exception: the former mdd_finish_unlink may failed to
2799                  * add the orphan MDT-object to the orphan list, then if the
2800                  * mdd_trans_create() failed because of barrier_entry(), the
2801                  * MDT-object will become real orphan that is neither in the
2802                  * namespace nor in the orphan list. Such bad case should be
2803                  * very rare and will be handled by e2fsck/lfsck. */
2804                 handle = mdd_trans_create(env, mdo2mdd(obj));
2805                 if (IS_ERR(handle)) {
2806                         rc = PTR_ERR(handle);
2807                         if (rc != -EINPROGRESS)
2808                                 GOTO(stop, rc);
2809
2810                         handle = NULL;
2811                         blocked = true;
2812                         goto cont;
2813                 }
2814
2815                 rc = mdd_declare_close(env, mdd_obj, ma, handle);
2816                 if (rc)
2817                         GOTO(stop, rc);
2818
2819                 rc = mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
2820                 if (rc)
2821                         GOTO(stop, rc);
2822
2823                 rc = mdd_trans_start(env, mdo2mdd(obj), handle);
2824                 if (rc)
2825                         GOTO(stop, rc);
2826         }
2827
2828 cont:
2829         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
2830         rc = mdd_la_get(env, mdd_obj, &ma->ma_attr);
2831         if (rc != 0) {
2832                 CERROR("Failed to get lu_attr of "DFID": %d\n",
2833                        PFID(mdd_object_fid(mdd_obj)), rc);
2834                 GOTO(out, rc);
2835         }
2836
2837         /* check again with lock */
2838         is_orphan = (mdd_obj->mod_count == 1) &&
2839                     ((mdd_obj->mod_flags & (ORPHAN_OBJ | DEAD_OBJ)) != 0 ||
2840                      ma->ma_attr.la_nlink == 0);
2841
2842         if (is_orphan && !handle && !blocked) {
2843                 mdd_write_unlock(env, mdd_obj);
2844                 goto again;
2845         }
2846
2847         mdd_obj->mod_count--; /*release open count */
2848
2849         if (!is_orphan || blocked)
2850                 GOTO(out, rc = 0);
2851
2852         /* Orphan object */
2853         /* NB: Object maybe not in orphan list originally, it is rare case for
2854          * mdd_finish_unlink() failure, in that case, the object doesn't have
2855          * ORPHAN_OBJ flag */
2856         if ((mdd_obj->mod_flags & ORPHAN_OBJ) != 0) {
2857                 /* remove link to object from orphan index */
2858                 LASSERT(handle != NULL);
2859                 rc = __mdd_orphan_del(env, mdd_obj, handle);
2860                 if (rc != 0) {
2861                         CERROR("%s: unable to delete "DFID" from orphan list: "
2862                                "rc = %d\n", lu_dev_name(mdd2lu_dev(mdd)),
2863                                PFID(mdd_object_fid(mdd_obj)), rc);
2864                         /* If object was not deleted from orphan list, do not
2865                          * destroy OSS objects, which will be done when next
2866                          * recovery. */
2867                         GOTO(out, rc);
2868                 }
2869
2870                 CDEBUG(D_HA, "Object "DFID" is deleted from orphan "
2871                        "list, OSS objects to be destroyed.\n",
2872                        PFID(mdd_object_fid(mdd_obj)));
2873         }
2874
2875         rc = mdo_destroy(env, mdd_obj, handle);
2876
2877         if (rc != 0) {
2878                 CERROR("%s: unable to delete "DFID" from orphan list: "
2879                        "rc = %d\n", lu_dev_name(mdd2lu_dev(mdd)),
2880                        PFID(mdd_object_fid(mdd_obj)), rc);
2881         }
2882         EXIT;
2883
2884 out:
2885         mdd_write_unlock(env, mdd_obj);
2886
2887         /* Record CL_CLOSE in changelog only if file was opened in write mode,
2888          * or if CL_OPEN was recorded.
2889          * Changelogs mask may change between open and close operations, but
2890          * this is not a big deal if we have a CL_CLOSE entry with no matching
2891          * CL_OPEN. Plus Changelogs mask may not change often.
2892          */
2893         if (!rc && !blocked &&
2894             ((mode & (FMODE_WRITE | MDS_OPEN_APPEND | MDS_OPEN_TRUNC)) ||
2895              (mdd->mdd_cl.mc_mask & (1 << CL_OPEN))) &&
2896             !(ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_RECOV_OPEN)) {
2897                 if (handle == NULL) {
2898                         handle = mdd_trans_create(env, mdo2mdd(obj));
2899                         if (IS_ERR(handle))
2900                                 GOTO(stop, rc = PTR_ERR(handle));
2901
2902                         rc = mdd_declare_changelog_store(env, mdd, NULL, NULL,
2903                                                          handle);
2904                         if (rc)
2905                                 GOTO(stop, rc);
2906
2907                         rc = mdd_trans_start(env, mdo2mdd(obj), handle);
2908                         if (rc)
2909                                 GOTO(stop, rc);
2910                 }
2911
2912                 mdd_changelog_data_store(env, mdd, CL_CLOSE, mode,
2913                                          mdd_obj, handle);
2914         }
2915
2916 stop:
2917         if (handle != NULL && !IS_ERR(handle))
2918                 rc = mdd_trans_stop(env, mdd, rc, handle);
2919
2920         return rc;
2921 }
2922
2923 /*
2924  * Permission check is done when open,
2925  * no need check again.
2926  */
2927 static int mdd_readpage_sanity_check(const struct lu_env *env,
2928                                      struct mdd_object *obj)
2929 {
2930         struct dt_object *next = mdd_object_child(obj);
2931         int rc;
2932         ENTRY;
2933
2934         if (S_ISDIR(mdd_object_type(obj)) && dt_try_as_dir(env, next))
2935                 rc = 0;
2936         else
2937                 rc = -ENOTDIR;
2938
2939         RETURN(rc);
2940 }
2941
2942 static int mdd_dir_page_build(const struct lu_env *env, union lu_page *lp,
2943                               size_t nob, const struct dt_it_ops *iops,
2944                               struct dt_it *it, __u32 attr, void *arg)
2945 {
2946         struct lu_dirpage       *dp = &lp->lp_dir;
2947         void                    *area = dp;
2948         int                      result;
2949         __u64                    hash = 0;
2950         struct lu_dirent        *ent;
2951         struct lu_dirent        *last = NULL;
2952         struct lu_fid            fid;
2953         int                      first = 1;
2954
2955         if (nob < sizeof(*dp))
2956                 return -EINVAL;
2957
2958         memset(area, 0, sizeof (*dp));
2959         area += sizeof (*dp);
2960         nob  -= sizeof (*dp);
2961
2962         ent  = area;
2963         do {
2964                 int    len;
2965                 size_t recsize;
2966
2967                 len = iops->key_size(env, it);
2968
2969                 /* IAM iterator can return record with zero len. */
2970                 if (len == 0)
2971                         goto next;
2972
2973                 hash = iops->store(env, it);
2974                 if (unlikely(first)) {
2975                         first = 0;
2976                         dp->ldp_hash_start = cpu_to_le64(hash);
2977                 }
2978
2979                 /* calculate max space required for lu_dirent */
2980                 recsize = lu_dirent_calc_size(len, attr);
2981
2982                 if (nob >= recsize) {
2983                         result = iops->rec(env, it, (struct dt_rec *)ent, attr);
2984                         if (result == -ESTALE)
2985                                 goto next;
2986                         if (result != 0)
2987                                 goto out;
2988
2989                         /* osd might not able to pack all attributes,
2990                          * so recheck rec length */
2991                         recsize = le16_to_cpu(ent->lde_reclen);
2992
2993                         if (le32_to_cpu(ent->lde_attrs) & LUDA_FID) {
2994                                 fid_le_to_cpu(&fid, &ent->lde_fid);
2995                                 if (fid_is_dot_lustre(&fid))
2996                                         goto next;
2997                         }
2998                 } else {
2999                         result = (last != NULL) ? 0 :-EINVAL;
3000                         goto out;
3001                 }
3002                 last = ent;
3003                 ent = (void *)ent + recsize;
3004                 nob -= recsize;
3005
3006 next:
3007                 result = iops->next(env, it);
3008                 if (result == -ESTALE)
3009                         goto next;
3010         } while (result == 0);
3011
3012 out:
3013         dp->ldp_hash_end = cpu_to_le64(hash);
3014         if (last != NULL) {
3015                 if (last->lde_hash == dp->ldp_hash_end)
3016                         dp->ldp_flags |= cpu_to_le32(LDF_COLLIDE);
3017                 last->lde_reclen = 0; /* end mark */
3018         }
3019         if (result > 0)
3020                 /* end of directory */
3021                 dp->ldp_hash_end = cpu_to_le64(MDS_DIR_END_OFF);
3022         else if (result < 0)
3023                 CWARN("build page failed: %d!\n", result);
3024         return result;
3025 }
3026
3027 int mdd_readpage(const struct lu_env *env, struct md_object *obj,
3028                  const struct lu_rdpg *rdpg)
3029 {
3030         struct mdd_object *mdd_obj = md2mdd_obj(obj);
3031         int rc;
3032         ENTRY;
3033
3034         if (mdd_object_exists(mdd_obj) == 0) {
3035                 CERROR("%s: object "DFID" not found: rc = -2\n",
3036                        mdd_obj_dev_name(mdd_obj),PFID(mdd_object_fid(mdd_obj)));
3037                 return -ENOENT;
3038         }
3039
3040         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
3041         rc = mdd_readpage_sanity_check(env, mdd_obj);
3042         if (rc)
3043                 GOTO(out_unlock, rc);
3044
3045         if (mdd_is_dead_obj(mdd_obj)) {
3046                 struct page *pg;
3047                 struct lu_dirpage *dp;
3048
3049                 /*
3050                  * According to POSIX, please do not return any entry to client:
3051                  * even dot and dotdot should not be returned.
3052                  */
3053                 CDEBUG(D_INODE, "readdir from dead object: "DFID"\n",
3054                        PFID(mdd_object_fid(mdd_obj)));
3055
3056                 if (rdpg->rp_count <= 0)
3057                         GOTO(out_unlock, rc = -EFAULT);
3058                 LASSERT(rdpg->rp_pages != NULL);
3059
3060                 pg = rdpg->rp_pages[0];
3061                 dp = (struct lu_dirpage *)kmap(pg);
3062                 memset(dp, 0 , sizeof(struct lu_dirpage));
3063                 dp->ldp_hash_start = cpu_to_le64(rdpg->rp_hash);
3064                 dp->ldp_hash_end   = cpu_to_le64(MDS_DIR_END_OFF);
3065                 dp->ldp_flags = cpu_to_le32(LDF_EMPTY);
3066                 kunmap(pg);
3067                 GOTO(out_unlock, rc = LU_PAGE_SIZE);
3068         }
3069
3070         rc = dt_index_walk(env, mdd_object_child(mdd_obj), rdpg,
3071                            mdd_dir_page_build, NULL);
3072         if (rc >= 0) {
3073                 struct lu_dirpage       *dp;
3074
3075                 dp = kmap(rdpg->rp_pages[0]);
3076                 dp->ldp_hash_start = cpu_to_le64(rdpg->rp_hash);
3077                 if (rc == 0) {
3078                         /*
3079                          * No pages were processed, mark this for first page
3080                          * and send back.
3081                          */
3082                         dp->ldp_hash_end = cpu_to_le64(MDS_DIR_END_OFF);
3083                         dp->ldp_flags = cpu_to_le32(LDF_EMPTY);
3084                         rc = min_t(unsigned int, LU_PAGE_SIZE, rdpg->rp_count);
3085                 }
3086                 kunmap(rdpg->rp_pages[0]);
3087         }
3088
3089         GOTO(out_unlock, rc);
3090 out_unlock:
3091         mdd_read_unlock(env, mdd_obj);
3092         return rc;
3093 }
3094
3095 static int mdd_object_sync(const struct lu_env *env, struct md_object *obj)
3096 {
3097         struct mdd_object *mdd_obj = md2mdd_obj(obj);
3098
3099         if (mdd_object_exists(mdd_obj) == 0) {
3100                 int rc = -ENOENT;
3101
3102                 CERROR("%s: object "DFID" not found: rc = %d\n",
3103                        mdd_obj_dev_name(mdd_obj),
3104                        PFID(mdd_object_fid(mdd_obj)), rc);
3105                 return rc;
3106         }
3107         return dt_object_sync(env, mdd_object_child(mdd_obj),
3108                               0, OBD_OBJECT_EOF);
3109 }
3110
3111 static int mdd_object_lock(const struct lu_env *env,
3112                            struct md_object *obj,
3113                            struct lustre_handle *lh,
3114                            struct ldlm_enqueue_info *einfo,
3115                            union ldlm_policy_data *policy)
3116 {
3117         struct mdd_object *mdd_obj = md2mdd_obj(obj);
3118         return dt_object_lock(env, mdd_object_child(mdd_obj), lh,
3119                               einfo, policy);
3120 }
3121
3122 static int mdd_object_unlock(const struct lu_env *env,
3123                              struct md_object *obj,
3124                              struct ldlm_enqueue_info *einfo,
3125                              union ldlm_policy_data *policy)
3126 {
3127         struct mdd_object *mdd_obj = md2mdd_obj(obj);
3128         return dt_object_unlock(env, mdd_object_child(mdd_obj), einfo, policy);
3129 }
3130
3131 const struct md_object_operations mdd_obj_ops = {
3132         .moo_permission         = mdd_permission,
3133         .moo_attr_get           = mdd_attr_get,
3134         .moo_attr_set           = mdd_attr_set,
3135         .moo_xattr_get          = mdd_xattr_get,
3136         .moo_xattr_set          = mdd_xattr_set,
3137         .moo_xattr_list         = mdd_xattr_list,
3138         .moo_invalidate         = mdd_invalidate,
3139         .moo_xattr_del          = mdd_xattr_del,
3140         .moo_swap_layouts       = mdd_swap_layouts,
3141         .moo_open               = mdd_open,
3142         .moo_close              = mdd_close,
3143         .moo_readpage           = mdd_readpage,
3144         .moo_readlink           = mdd_readlink,
3145         .moo_changelog          = mdd_changelog,
3146         .moo_object_sync        = mdd_object_sync,
3147         .moo_object_lock        = mdd_object_lock,
3148         .moo_object_unlock      = mdd_object_unlock,
3149         .moo_layout_change      = mdd_layout_change,
3150 };