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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/mdd/mdd_object.c
  37  *
  38  * Lustre Metadata Server (mdd) routines
  39  *
  40  * Author: Wang Di <wangdi@clusterfs.com>
  41  */
  42
  43 #define DEBUG_SUBSYSTEM S_MDS
  44
  45 #include <linux/module.h>
  46 #include <obd.h>
  47 #include <obd_class.h>
  48 #include <obd_support.h>
  49 #include <lprocfs_status.h>
  50 /* fid_be_cpu(), fid_cpu_to_be(). */
  51 #include <lustre_fid.h>
  52 #include <obd_lov.h>
  53 #include <lustre_idmap.h>
  54 #include <lustre_param.h>
  55 #include <lustre_mds.h>
  56 #include <lustre/lustre_idl.h>
  57
  58 #include "mdd_internal.h"
  59
  60 static const struct lu_object_operations mdd_lu_obj_ops;
  61 extern struct kmem_cache *mdd_object_kmem;
  62
  63 static int mdd_xattr_get(const struct lu_env *env,
  64                          struct md_object *obj, struct lu_buf *buf,
  65                          const char *name);
  66
  67 int mdd_data_get(const struct lu_env *env, struct mdd_object *obj,
  68                  void **data)
  69 {
  70         if (mdd_object_exists(obj) == 0) {
  71                 CERROR("%s: object "DFID" not found: rc = -2\n",
  72                        mdd_obj_dev_name(obj), PFID(mdd_object_fid(obj)));
  73                 return -ENOENT;
  74         }
  75         mdo_data_get(env, obj, data);
  76         return 0;
  77 }
  78
  79 int mdd_la_get(const struct lu_env *env, struct mdd_object *obj,
  80                struct lu_attr *la, struct lustre_capa *capa)
  81 {
  82         if (mdd_object_exists(obj) == 0) {
  83                 CERROR("%s: object "DFID" not found: rc = -2\n",
  84                        mdd_obj_dev_name(obj), PFID(mdd_object_fid(obj)));
  85                 return -ENOENT;
  86         }
  87         return mdo_attr_get(env, obj, la, capa);
  88 }
  89
  90 static void mdd_flags_xlate(struct mdd_object *obj, __u32 flags)
  91 {
  92         obj->mod_flags &= ~(APPEND_OBJ|IMMUTE_OBJ);
  93
  94         if (flags & LUSTRE_APPEND_FL)
  95                 obj->mod_flags |= APPEND_OBJ;
  96
  97         if (flags & LUSTRE_IMMUTABLE_FL)
  98                 obj->mod_flags |= IMMUTE_OBJ;
  99 }
 100
 101 struct mdd_thread_info *mdd_env_info(const struct lu_env *env)
 102 {
 103         struct mdd_thread_info *info;
 104
 105         info = lu_context_key_get(&env->le_ctx, &mdd_thread_key);
 106         LASSERT(info != NULL);
 107         return info;
 108 }
 109
 110 const struct lu_name *mdd_name_get_const(const struct lu_env *env,
 111                                          const void *area, ssize_t len)
 112 {
 113         struct lu_name *lname;
 114
 115         lname = &mdd_env_info(env)->mti_name;
 116         lname->ln_name = area;
 117         lname->ln_namelen = len;
 118         return lname;
 119 }
 120
 121 struct lu_buf *mdd_buf_get(const struct lu_env *env, void *area, ssize_t len)
 122 {
 123         struct lu_buf *buf;
 124
 125         buf = &mdd_env_info(env)->mti_buf[0];
 126         buf->lb_buf = area;
 127         buf->lb_len = len;
 128         return buf;
 129 }
 130
 131 const struct lu_buf *mdd_buf_get_const(const struct lu_env *env,
 132                                        const void *area, ssize_t len)
 133 {
 134         struct lu_buf *buf;
 135
 136         buf = &mdd_env_info(env)->mti_buf[0];
 137         buf->lb_buf = (void *)area;
 138         buf->lb_len = len;
 139         return buf;
 140 }
 141
 142 struct lu_object *mdd_object_alloc(const struct lu_env *env,
 143                                    const struct lu_object_header *hdr,
 144                                    struct lu_device *d)
 145 {
 146         struct mdd_object *mdd_obj;
 147
 148         OBD_SLAB_ALLOC_PTR_GFP(mdd_obj, mdd_object_kmem, GFP_NOFS);
 149         if (mdd_obj != NULL) {
 150                 struct lu_object *o;
 151
 152                 o = mdd2lu_obj(mdd_obj);
 153                 lu_object_init(o, NULL, d);
 154                 mdd_obj->mod_obj.mo_ops = &mdd_obj_ops;
 155                 mdd_obj->mod_obj.mo_dir_ops = &mdd_dir_ops;
 156                 mdd_obj->mod_count = 0;
 157                 o->lo_ops = &mdd_lu_obj_ops;
 158                 return o;
 159         } else {
 160                 return NULL;
 161         }
 162 }
 163
 164 static int mdd_object_init(const struct lu_env *env, struct lu_object *o,
 165                            const struct lu_object_conf *unused)
 166 {
 167         struct mdd_device *d = lu2mdd_dev(o->lo_dev);
 168         struct mdd_object *mdd_obj = lu2mdd_obj(o);
 169         struct lu_object  *below;
 170         struct lu_device  *under;
 171         ENTRY;
 172
 173         mdd_obj->mod_cltime = 0;
 174         under = &d->mdd_child->dd_lu_dev;
 175         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
 176         if (IS_ERR(below))
 177                 RETURN(PTR_ERR(below));
 178
 179         lu_object_add(o, below);
 180
 181         RETURN(0);
 182 }
 183
 184 static int mdd_object_start(const struct lu_env *env, struct lu_object *o)
 185 {
 186         if (lu_object_exists(o))
 187                 return mdd_get_flags(env, lu2mdd_obj(o));
 188         else
 189                 return 0;
 190 }
 191
 192 static void mdd_object_free(const struct lu_env *env, struct lu_object *o)
 193 {
 194         struct mdd_object *mdd = lu2mdd_obj(o);
 195
 196         lu_object_fini(o);
 197         OBD_SLAB_FREE_PTR(mdd, mdd_object_kmem);
 198 }
 199
 200 static int mdd_object_print(const struct lu_env *env, void *cookie,
 201                             lu_printer_t p, const struct lu_object *o)
 202 {
 203         struct mdd_object *mdd = lu2mdd_obj((struct lu_object *)o);
 204         return (*p)(env, cookie, LUSTRE_MDD_NAME"-object@%p(open_count=%d, "
 205                     "valid=%x, cltime="LPU64", flags=%lx)",
 206                     mdd, mdd->mod_count, mdd->mod_valid,
 207                     mdd->mod_cltime, mdd->mod_flags);
 208 }
 209
 210 static const struct lu_object_operations mdd_lu_obj_ops = {
 211         .loo_object_init    = mdd_object_init,
 212         .loo_object_start   = mdd_object_start,
 213         .loo_object_free    = mdd_object_free,
 214         .loo_object_print   = mdd_object_print,
 215 };
 216
 217 struct mdd_object *mdd_object_find(const struct lu_env *env,
 218                                    struct mdd_device *d,
 219                                    const struct lu_fid *f)
 220 {
 221         return md2mdd_obj(md_object_find_slice(env, &d->mdd_md_dev, f));
 222 }
 223
 224 /* Returns the full path to this fid, as of changelog record recno. */
 225 int mdd_get_flags(const struct lu_env *env, struct mdd_object *obj)
 226 {
 227         struct lu_attr *la = &mdd_env_info(env)->mti_la;
 228         int rc;
 229
 230         ENTRY;
 231         rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
 232         if (rc == 0) {
 233                 mdd_flags_xlate(obj, la->la_flags);
 234         }
 235         RETURN(rc);
 236 }
 237
 238 /*
 239  * No permission check is needed.
 240  */
 241 int mdd_attr_get(const struct lu_env *env, struct md_object *obj,
 242                  struct md_attr *ma)
 243 {
 244         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 245         int               rc;
 246
 247         ENTRY;
 248
 249         rc = mdd_la_get(env, mdd_obj, &ma->ma_attr,
 250                         mdd_object_capa(env, md2mdd_obj(obj)));
 251         if ((ma->ma_need & MA_INODE) != 0 && mdd_is_dead_obj(mdd_obj))
 252                 ma->ma_attr.la_nlink = 0;
 253
 254         RETURN(rc);
 255 }
 256
 257 /*
 258  * No permission check is needed.
 259  */
 260 static int mdd_xattr_get(const struct lu_env *env,
 261                          struct md_object *obj, struct lu_buf *buf,
 262                          const char *name)
 263 {
 264         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 265         int rc;
 266
 267         ENTRY;
 268
 269         if (mdd_object_exists(mdd_obj) == 0) {
 270                 CERROR("%s: object "DFID" not found: rc = -2\n",
 271                        mdd_obj_dev_name(mdd_obj),PFID(mdd_object_fid(mdd_obj)));
 272                 return -ENOENT;
 273         }
 274
 275         /* If the object has been delete from the namespace, then
 276          * get linkEA should return -ENOENT as well */
 277         if (unlikely((mdd_obj->mod_flags & (DEAD_OBJ | ORPHAN_OBJ)) &&
 278                       strcmp(name, XATTR_NAME_LINK) == 0))
 279                 RETURN(-ENOENT);
 280
 281         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 282         rc = mdo_xattr_get(env, mdd_obj, buf, name,
 283                            mdd_object_capa(env, mdd_obj));
 284         mdd_read_unlock(env, mdd_obj);
 285
 286         RETURN(rc);
 287 }
 288
 289 /*
 290  * Permission check is done when open,
 291  * no need check again.
 292  */
 293 static int mdd_readlink(const struct lu_env *env, struct md_object *obj,
 294                         struct lu_buf *buf)
 295 {
 296         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 297         struct dt_object  *next;
 298         loff_t             pos = 0;
 299         int                rc;
 300         ENTRY;
 301
 302         if (mdd_object_exists(mdd_obj) == 0) {
 303                 CERROR("%s: object "DFID" not found: rc = -2\n",
 304                        mdd_obj_dev_name(mdd_obj),PFID(mdd_object_fid(mdd_obj)));
 305                 return -ENOENT;
 306         }
 307
 308         next = mdd_object_child(mdd_obj);
 309         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 310         rc = next->do_body_ops->dbo_read(env, next, buf, &pos,
 311                                          mdd_object_capa(env, mdd_obj));
 312         mdd_read_unlock(env, mdd_obj);
 313         RETURN(rc);
 314 }
 315
 316 /*
 317  * No permission check is needed.
 318  */
 319 static int mdd_xattr_list(const struct lu_env *env, struct md_object *obj,
 320                           struct lu_buf *buf)
 321 {
 322         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 323         int rc;
 324
 325         ENTRY;
 326
 327         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 328         rc = mdo_xattr_list(env, mdd_obj, buf, mdd_object_capa(env, mdd_obj));
 329         mdd_read_unlock(env, mdd_obj);
 330
 331         RETURN(rc);
 332 }
 333
 334 int mdd_declare_object_create_internal(const struct lu_env *env,
 335                                        struct mdd_object *p,
 336                                        struct mdd_object *c,
 337                                        struct lu_attr *attr,
 338                                        struct thandle *handle,
 339                                        const struct md_op_spec *spec)
 340 {
 341         struct dt_object_format *dof = &mdd_env_info(env)->mti_dof;
 342         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
 343         const struct dt_index_features *feat = spec->sp_feat;
 344         int rc;
 345         ENTRY;
 346
 347         if (feat != &dt_directory_features && feat != NULL) {
 348                 dof->dof_type = DFT_INDEX;
 349                 dof->u.dof_idx.di_feat = feat;
 350
 351         } else {
 352                 dof->dof_type = dt_mode_to_dft(attr->la_mode);
 353                 if (dof->dof_type == DFT_REGULAR) {
 354                         dof->u.dof_reg.striped =
 355                                 md_should_create(spec->sp_cr_flags);
 356                         if (spec->sp_cr_flags & MDS_OPEN_HAS_EA)
 357                                 dof->u.dof_reg.striped = 0;
 358                         /* is this replay? */
 359                         if (spec->no_create)
 360                                 dof->u.dof_reg.striped = 0;
 361                 }
 362         }
 363
 364         rc = mdo_declare_create_obj(env, c, attr, hint, dof, handle);
 365
 366         RETURN(rc);
 367 }
 368
 369 int mdd_object_create_internal(const struct lu_env *env, struct mdd_object *p,
 370                                struct mdd_object *c, struct lu_attr *attr,
 371                                struct thandle *handle,
 372                                const struct md_op_spec *spec)
 373 {
 374         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
 375         struct dt_object_format *dof = &mdd_env_info(env)->mti_dof;
 376         int rc;
 377         ENTRY;
 378
 379         LASSERT(!mdd_object_exists(c));
 380
 381         rc = mdo_create_obj(env, c, attr, hint, dof, handle);
 382
 383         LASSERT(ergo(rc == 0, mdd_object_exists(c)));
 384
 385         RETURN(rc);
 386 }
 387
 388 /**
 389  * Make sure the ctime is increased only.
 390  */
 391 static inline int mdd_attr_check(const struct lu_env *env,
 392                                  struct mdd_object *obj,
 393                                  struct lu_attr *attr)
 394 {
 395         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
 396         int rc;
 397         ENTRY;
 398
 399         if (attr->la_valid & LA_CTIME) {
 400                 rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 401                 if (rc)
 402                         RETURN(rc);
 403
 404                 if (attr->la_ctime < tmp_la->la_ctime)
 405                         attr->la_valid &= ~(LA_MTIME | LA_CTIME);
 406                 else if (attr->la_valid == LA_CTIME &&
 407                          attr->la_ctime == tmp_la->la_ctime)
 408                         attr->la_valid &= ~LA_CTIME;
 409         }
 410         RETURN(0);
 411 }
 412
 413 int mdd_attr_set_internal(const struct lu_env *env, struct mdd_object *obj,
 414                           const struct lu_attr *attr, struct thandle *handle,
 415                           int needacl)
 416 {
 417         int rc;
 418         ENTRY;
 419
 420         rc = mdo_attr_set(env, obj, attr, handle, mdd_object_capa(env, obj));
 421 #ifdef CONFIG_FS_POSIX_ACL
 422         if (!rc && (attr->la_valid & LA_MODE) && needacl)
 423                 rc = mdd_acl_chmod(env, obj, attr->la_mode, handle);
 424 #endif
 425         RETURN(rc);
 426 }
 427
 428 int mdd_attr_check_set_internal(const struct lu_env *env,
 429                                 struct mdd_object *obj, struct lu_attr *attr,
 430                                 struct thandle *handle, int needacl)
 431 {
 432         int rc;
 433         ENTRY;
 434
 435         rc = mdd_attr_check(env, obj, attr);
 436         if (rc)
 437                 RETURN(rc);
 438
 439         if (attr->la_valid)
 440                 rc = mdd_attr_set_internal(env, obj, attr, handle, needacl);
 441         RETURN(rc);
 442 }
 443
 444 /*
 445  * This gives the same functionality as the code between
 446  * sys_chmod and inode_setattr
 447  * chown_common and inode_setattr
 448  * utimes and inode_setattr
 449  * This API is ported from mds_fix_attr but remove some unnecesssary stuff.
 450  */
 451 static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
 452                         struct lu_attr *la, const unsigned long flags)
 453 {
 454         struct lu_attr   *tmp_la     = &mdd_env_info(env)->mti_la;
 455         struct lu_ucred  *uc;
 456         int               rc;
 457         ENTRY;
 458
 459         if (!la->la_valid)
 460                 RETURN(0);
 461
 462         /* Do not permit change file type */
 463         if (la->la_valid & LA_TYPE)
 464                 RETURN(-EPERM);
 465
 466         /* They should not be processed by setattr */
 467         if (la->la_valid & (LA_NLINK | LA_RDEV | LA_BLKSIZE))
 468                 RETURN(-EPERM);
 469
 470         /* export destroy does not have ->le_ses, but we may want
 471          * to drop LUSTRE_SOM_FL. */
 472         uc = lu_ucred_check(env);
 473         if (uc == NULL)
 474                 RETURN(0);
 475
 476         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 477         if (rc)
 478                 RETURN(rc);
 479
 480         if (la->la_valid == LA_CTIME) {
 481                 if (!(flags & MDS_PERM_BYPASS))
 482                         /* This is only for set ctime when rename's source is
 483                          * on remote MDS. */
 484                         rc = mdd_may_delete(env, NULL, obj, tmp_la, NULL, 1, 0);
 485                 if (rc == 0 && la->la_ctime <= tmp_la->la_ctime)
 486                         la->la_valid &= ~LA_CTIME;
 487                 RETURN(rc);
 488         }
 489
 490         if (la->la_valid == LA_ATIME) {
 491                 /* This is atime only set for read atime update on close. */
 492                 if (la->la_atime >= tmp_la->la_atime &&
 493                     la->la_atime < (tmp_la->la_atime +
 494                                     mdd_obj2mdd_dev(obj)->mdd_atime_diff))
 495                         la->la_valid &= ~LA_ATIME;
 496                 RETURN(0);
 497         }
 498
 499         /* Check if flags change. */
 500         if (la->la_valid & LA_FLAGS) {
 501                 unsigned int oldflags = 0;
 502                 unsigned int newflags = la->la_flags &
 503                                 (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL);
 504
 505                 if ((uc->uc_fsuid != tmp_la->la_uid) &&
 506                     !md_capable(uc, CFS_CAP_FOWNER))
 507                         RETURN(-EPERM);
 508
 509                 /* XXX: the IMMUTABLE and APPEND_ONLY flags can
 510                  * only be changed by the relevant capability. */
 511                 if (mdd_is_immutable(obj))
 512                         oldflags |= LUSTRE_IMMUTABLE_FL;
 513                 if (mdd_is_append(obj))
 514                         oldflags |= LUSTRE_APPEND_FL;
 515                 if ((oldflags ^ newflags) &&
 516                     !md_capable(uc, CFS_CAP_LINUX_IMMUTABLE))
 517                         RETURN(-EPERM);
 518
 519                 if (!S_ISDIR(tmp_la->la_mode))
 520                         la->la_flags &= ~LUSTRE_DIRSYNC_FL;
 521         }
 522
 523         if ((mdd_is_immutable(obj) || mdd_is_append(obj)) &&
 524             (la->la_valid & ~LA_FLAGS) &&
 525             !(flags & MDS_PERM_BYPASS))
 526                 RETURN(-EPERM);
 527
 528         /* Check for setting the obj time. */
 529         if ((la->la_valid & (LA_MTIME | LA_ATIME | LA_CTIME)) &&
 530             !(la->la_valid & ~(LA_MTIME | LA_ATIME | LA_CTIME))) {
 531                 if ((uc->uc_fsuid != tmp_la->la_uid) &&
 532                     !md_capable(uc, CFS_CAP_FOWNER)) {
 533                         rc = mdd_permission_internal(env, obj, tmp_la,
 534                                                      MAY_WRITE);
 535                         if (rc)
 536                                 RETURN(rc);
 537                 }
 538         }
 539
 540         if (la->la_valid & LA_KILL_SUID) {
 541                 la->la_valid &= ~LA_KILL_SUID;
 542                 if ((tmp_la->la_mode & S_ISUID) &&
 543                     !(la->la_valid & LA_MODE)) {
 544                         la->la_mode = tmp_la->la_mode;
 545                         la->la_valid |= LA_MODE;
 546                 }
 547                 la->la_mode &= ~S_ISUID;
 548         }
 549
 550         if (la->la_valid & LA_KILL_SGID) {
 551                 la->la_valid &= ~LA_KILL_SGID;
 552                 if (((tmp_la->la_mode & (S_ISGID | S_IXGRP)) ==
 553                                         (S_ISGID | S_IXGRP)) &&
 554                     !(la->la_valid & LA_MODE)) {
 555                         la->la_mode = tmp_la->la_mode;
 556                         la->la_valid |= LA_MODE;
 557                 }
 558                 la->la_mode &= ~S_ISGID;
 559         }
 560
 561         /* Make sure a caller can chmod. */
 562         if (la->la_valid & LA_MODE) {
 563                 if (!(flags & MDS_PERM_BYPASS) &&
 564                     (uc->uc_fsuid != tmp_la->la_uid) &&
 565                     !md_capable(uc, CFS_CAP_FOWNER))
 566                         RETURN(-EPERM);
 567
 568                 if (la->la_mode == (umode_t) -1)
 569                         la->la_mode = tmp_la->la_mode;
 570                 else
 571                         la->la_mode = (la->la_mode & S_IALLUGO) |
 572                                       (tmp_la->la_mode & ~S_IALLUGO);
 573
 574                 /* Also check the setgid bit! */
 575                 if (!lustre_in_group_p(uc, (la->la_valid & LA_GID) ?
 576                                        la->la_gid : tmp_la->la_gid) &&
 577                     !md_capable(uc, CFS_CAP_FSETID))
 578                         la->la_mode &= ~S_ISGID;
 579         } else {
 580                la->la_mode = tmp_la->la_mode;
 581         }
 582
 583         /* Make sure a caller can chown. */
 584         if (la->la_valid & LA_UID) {
 585                 if (la->la_uid == (uid_t) -1)
 586                         la->la_uid = tmp_la->la_uid;
 587                 if (((uc->uc_fsuid != tmp_la->la_uid) ||
 588                      (la->la_uid != tmp_la->la_uid)) &&
 589                     !md_capable(uc, CFS_CAP_CHOWN))
 590                         RETURN(-EPERM);
 591
 592                 /* If the user or group of a non-directory has been
 593                  * changed by a non-root user, remove the setuid bit.
 594                  * 19981026 David C Niemi <niemi@tux.org>
 595                  *
 596                  * Changed this to apply to all users, including root,
 597                  * to avoid some races. This is the behavior we had in
 598                  * 2.0. The check for non-root was definitely wrong
 599                  * for 2.2 anyway, as it should have been using
 600                  * CAP_FSETID rather than fsuid -- 19990830 SD. */
 601                 if (((tmp_la->la_mode & S_ISUID) == S_ISUID) &&
 602                     !S_ISDIR(tmp_la->la_mode)) {
 603                         la->la_mode &= ~S_ISUID;
 604                         la->la_valid |= LA_MODE;
 605                 }
 606         }
 607
 608         /* Make sure caller can chgrp. */
 609         if (la->la_valid & LA_GID) {
 610                 if (la->la_gid == (gid_t) -1)
 611                         la->la_gid = tmp_la->la_gid;
 612                 if (((uc->uc_fsuid != tmp_la->la_uid) ||
 613                      ((la->la_gid != tmp_la->la_gid) &&
 614                       !lustre_in_group_p(uc, la->la_gid))) &&
 615                     !md_capable(uc, CFS_CAP_CHOWN))
 616                         RETURN(-EPERM);
 617
 618                 /* Likewise, if the user or group of a non-directory
 619                  * has been changed by a non-root user, remove the
 620                  * setgid bit UNLESS there is no group execute bit
 621                  * (this would be a file marked for mandatory
 622                  * locking).  19981026 David C Niemi <niemi@tux.org>
 623                  *
 624                  * Removed the fsuid check (see the comment above) --
 625                  * 19990830 SD. */
 626                 if (((tmp_la->la_mode & (S_ISGID | S_IXGRP)) ==
 627                      (S_ISGID | S_IXGRP)) && !S_ISDIR(tmp_la->la_mode)) {
 628                         la->la_mode &= ~S_ISGID;
 629                         la->la_valid |= LA_MODE;
 630                 }
 631         }
 632
 633         /* For both Size-on-MDS case and truncate case,
 634          * "la->la_valid & (LA_SIZE | LA_BLOCKS)" are ture.
 635          * We distinguish them by "flags & MDS_SOM".
 636          * For SOM case, it is true, the MAY_WRITE perm has been checked
 637          * when open, no need check again. For truncate case, it is false,
 638          * the MAY_WRITE perm should be checked here. */
 639         if (flags & MDS_SOM) {
 640                 /* For the "Size-on-MDS" setattr update, merge coming
 641                  * attributes with the set in the inode. BUG 10641 */
 642                 if ((la->la_valid & LA_ATIME) &&
 643                     (la->la_atime <= tmp_la->la_atime))
 644                         la->la_valid &= ~LA_ATIME;
 645
 646                 /* OST attributes do not have a priority over MDS attributes,
 647                  * so drop times if ctime is equal. */
 648                 if ((la->la_valid & LA_CTIME) &&
 649                     (la->la_ctime <= tmp_la->la_ctime))
 650                         la->la_valid &= ~(LA_MTIME | LA_CTIME);
 651         } else {
 652                 if (la->la_valid & (LA_SIZE | LA_BLOCKS)) {
 653                         if (!((flags & MDS_OWNEROVERRIDE) &&
 654                               (uc->uc_fsuid == tmp_la->la_uid)) &&
 655                             !(flags & MDS_PERM_BYPASS)) {
 656                                 rc = mdd_permission_internal(env, obj,
 657                                                              tmp_la, MAY_WRITE);
 658                                 if (rc != 0)
 659                                         RETURN(rc);
 660                         }
 661                 }
 662                 if (la->la_valid & LA_CTIME) {
 663                         /* The pure setattr, it has the priority over what is
 664                          * already set, do not drop it if ctime is equal. */
 665                         if (la->la_ctime < tmp_la->la_ctime)
 666                                 la->la_valid &= ~(LA_ATIME | LA_MTIME |
 667                                                   LA_CTIME);
 668                 }
 669         }
 670
 671         RETURN(0);
 672 }
 673
 674 /** Store a data change changelog record
 675  * If this fails, we must fail the whole transaction; we don't
 676  * want the change to commit without the log entry.
 677  * \param mdd_obj - mdd_object of change
 678  * \param handle - transacion handle
 679  */
 680 int mdd_changelog_data_store(const struct lu_env *env, struct mdd_device *mdd,
 681                              enum changelog_rec_type type, int flags,
 682                              struct mdd_object *mdd_obj, struct thandle *handle)
 683 {
 684         const struct lu_fid             *tfid;
 685         struct llog_changelog_rec       *rec;
 686         struct lu_buf                   *buf;
 687         int                              reclen;
 688         int                              rc;
 689
 690         /* Not recording */
 691         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
 692                 RETURN(0);
 693         if ((mdd->mdd_cl.mc_mask & (1 << type)) == 0)
 694                 RETURN(0);
 695
 696         LASSERT(mdd_obj != NULL);
 697         LASSERT(handle != NULL);
 698
 699         tfid = mdo2fid(mdd_obj);
 700
 701         if ((type >= CL_MTIME) && (type <= CL_ATIME) &&
 702             cfs_time_before_64(mdd->mdd_cl.mc_starttime, mdd_obj->mod_cltime)) {
 703                 /* Don't need multiple updates in this log */
 704                 /* Don't check under lock - no big deal if we get an extra
 705                    entry */
 706                 RETURN(0);
 707         }
 708
 709         reclen = llog_data_len(sizeof(*rec));
 710         buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf, reclen);
 711         if (buf->lb_buf == NULL)
 712                 RETURN(-ENOMEM);
 713         rec = buf->lb_buf;
 714
 715         rec->cr.cr_flags = CLF_VERSION | (CLF_FLAGMASK & flags);
 716         rec->cr.cr_type = (__u32)type;
 717         rec->cr.cr_tfid = *tfid;
 718         rec->cr.cr_namelen = 0;
 719         mdd_obj->mod_cltime = cfs_time_current_64();
 720
 721         rc = mdd_changelog_store(env, mdd, rec, handle);
 722
 723         RETURN(rc);
 724 }
 725
 726 int mdd_changelog(const struct lu_env *env, enum changelog_rec_type type,
 727                   int flags, struct md_object *obj)
 728 {
 729         struct thandle *handle;
 730         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 731         struct mdd_device *mdd = mdo2mdd(obj);
 732         int rc;
 733         ENTRY;
 734
 735         handle = mdd_trans_create(env, mdd);
 736         if (IS_ERR(handle))
 737                 RETURN(PTR_ERR(handle));
 738
 739         rc = mdd_declare_changelog_store(env, mdd, NULL, handle);
 740         if (rc)
 741                 GOTO(stop, rc);
 742
 743         rc = mdd_trans_start(env, mdd, handle);
 744         if (rc)
 745                 GOTO(stop, rc);
 746
 747         rc = mdd_changelog_data_store(env, mdd, type, flags, mdd_obj,
 748                                       handle);
 749
 750 stop:
 751         mdd_trans_stop(env, mdd, rc, handle);
 752
 753         RETURN(rc);
 754 }
 755
 756 /**
 757  * Save LMA extended attributes with data from \a ma.
 758  *
 759  * HSM and Size-On-MDS data will be extracted from \ma if they are valid, if
 760  * not, LMA EA will be first read from disk, modified and write back.
 761  *
 762  */
 763 /* Precedence for choosing record type when multiple
 764  * attributes change: setattr > mtime > ctime > atime
 765  * (ctime changes when mtime does, plus chmod/chown.
 766  * atime and ctime are independent.) */
 767 static int mdd_attr_set_changelog(const struct lu_env *env,
 768                                   struct md_object *obj, struct thandle *handle,
 769                                   __u64 valid)
 770 {
 771         struct mdd_device *mdd = mdo2mdd(obj);
 772         int bits, type = 0;
 773
 774         bits =  (valid & LA_SIZE)  ? 1 << CL_TRUNC : 0;
 775         bits |= (valid & ~(LA_CTIME|LA_MTIME|LA_ATIME)) ? 1 << CL_SETATTR : 0;
 776         bits |= (valid & LA_MTIME) ? 1 << CL_MTIME : 0;
 777         bits |= (valid & LA_CTIME) ? 1 << CL_CTIME : 0;
 778         bits |= (valid & LA_ATIME) ? 1 << CL_ATIME : 0;
 779         bits = bits & mdd->mdd_cl.mc_mask;
 780         /* This is an implementation limit rather than a protocol limit */
 781         CLASSERT(CL_LAST <= sizeof(int) * 8);
 782         if (bits == 0)
 783                 return 0;
 784
 785         /* The record type is the lowest non-masked set bit */
 786         type = __ffs(bits);
 787
 788         /* FYI we only store the first CLF_FLAGMASK bits of la_valid */
 789         return mdd_changelog_data_store(env, mdd, type, (int)valid,
 790                                         md2mdd_obj(obj), handle);
 791 }
 792
 793 static int mdd_declare_attr_set(const struct lu_env *env,
 794                                 struct mdd_device *mdd,
 795                                 struct mdd_object *obj,
 796                                 const struct lu_attr *attr,
 797                                 struct thandle *handle)
 798 {
 799         int rc;
 800
 801         rc = mdo_declare_attr_set(env, obj, attr, handle);
 802         if (rc)
 803                 return rc;
 804
 805 #ifdef CONFIG_FS_POSIX_ACL
 806         if (attr->la_valid & LA_MODE) {
 807                 mdd_read_lock(env, obj, MOR_TGT_CHILD);
 808                 rc = mdo_xattr_get(env, obj, &LU_BUF_NULL,
 809                                    XATTR_NAME_ACL_ACCESS, BYPASS_CAPA);
 810                 mdd_read_unlock(env, obj);
 811                 if (rc == -EOPNOTSUPP || rc == -ENODATA)
 812                         rc = 0;
 813                 else if (rc < 0)
 814                         return rc;
 815
 816                 if (rc != 0) {
 817                         struct lu_buf *buf = mdd_buf_get(env, NULL, rc);
 818                         rc = mdo_declare_xattr_set(env, obj, buf,
 819                                                    XATTR_NAME_ACL_ACCESS, 0,
 820                                                    handle);
 821                         if (rc)
 822                                 return rc;
 823                 }
 824         }
 825 #endif
 826
 827         rc = mdd_declare_changelog_store(env, mdd, NULL, handle);
 828         return rc;
 829 }
 830
 831 static inline bool permission_is_reduced(const struct lu_attr *old,
 832                                          const struct lu_attr *new)
 833 {
 834         return ((new->la_mode & old->la_mode) & S_IRWXUGO) !=
 835                (old->la_mode & S_IRWXUGO);
 836 }
 837
 838 /* set attr and LOV EA at once, return updated attr */
 839 int mdd_attr_set(const struct lu_env *env, struct md_object *obj,
 840                  const struct md_attr *ma)
 841 {
 842         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 843         struct mdd_device *mdd = mdo2mdd(obj);
 844         struct thandle *handle;
 845         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
 846         const struct lu_attr *la = &ma->ma_attr;
 847         bool sync_perm = false;
 848         int rc;
 849         ENTRY;
 850
 851         /* we do not use ->attr_set() for LOV/SOM/HSM EA any more */
 852         LASSERT((ma->ma_valid & MA_LOV) == 0);
 853         LASSERT((ma->ma_valid & MA_HSM) == 0);
 854         LASSERT((ma->ma_valid & MA_SOM) == 0);
 855
 856         *la_copy = ma->ma_attr;
 857         rc = mdd_fix_attr(env, mdd_obj, la_copy, ma->ma_attr_flags);
 858         if (rc)
 859                 RETURN(rc);
 860
 861         /* setattr on "close" only change atime, or do nothing */
 862         if (la->la_valid == LA_ATIME && la_copy->la_valid == 0)
 863                 RETURN(0);
 864
 865         handle = mdd_trans_create(env, mdd);
 866         if (IS_ERR(handle))
 867                 RETURN(PTR_ERR(handle));
 868
 869         rc = mdd_declare_attr_set(env, mdd, mdd_obj, la, handle);
 870         if (rc)
 871                 GOTO(stop, rc);
 872
 873         rc = mdd_trans_start(env, mdd, handle);
 874         if (rc)
 875                 GOTO(stop, rc);
 876
 877         /*
 878          * LU-3671
 879          *
 880          * permission changes may require sync operation, to mitigate
 881          * performance impact, only do this for dir and when permission is
 882          * reduced.
 883          *
 884          * For regular files, version is updated with permission change
 885          * (see VBR), async permission won't cause any issue, while missing
 886          * permission change on directory may affect accessibility of other
 887          * objects.
 888          */
 889         if (S_ISDIR(mdd_object_type(mdd_obj))) {
 890                 if (la->la_valid & (LA_UID | LA_GID)) {
 891                         sync_perm = true;
 892                 } else if (la->la_valid & LA_MODE &&
 893                            la->la_mode & (S_ISUID | S_ISGID | S_ISVTX)) {
 894                         sync_perm = true;
 895                 } else if (la->la_valid & LA_MODE) {
 896                         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
 897
 898                         rc = mdd_la_get(env, mdd_obj, tmp_la, BYPASS_CAPA);
 899                         if (rc)
 900                                 GOTO(stop, rc);
 901
 902                         if (permission_is_reduced(tmp_la, la))
 903                                 sync_perm = true;
 904                 }
 905         }
 906
 907         if (sync_perm)
 908                 handle->th_sync |= !!mdd->mdd_sync_permission;
 909
 910         if (la->la_valid & (LA_MTIME | LA_CTIME))
 911                 CDEBUG(D_INODE, "setting mtime "LPU64", ctime "LPU64"\n",
 912                        la->la_mtime, la->la_ctime);
 913
 914         if (la_copy->la_valid & LA_FLAGS) {
 915                 rc = mdd_attr_set_internal(env, mdd_obj, la_copy, handle, 1);
 916                 if (rc == 0)
 917                         mdd_flags_xlate(mdd_obj, la_copy->la_flags);
 918         } else if (la_copy->la_valid) {            /* setattr */
 919                 rc = mdd_attr_set_internal(env, mdd_obj, la_copy, handle, 1);
 920         }
 921
 922         if (rc == 0)
 923                 rc = mdd_attr_set_changelog(env, obj, handle,
 924                                             la->la_valid);
 925 stop:
 926         mdd_trans_stop(env, mdd, rc, handle);
 927         RETURN(rc);
 928 }
 929
 930 static int mdd_xattr_sanity_check(const struct lu_env *env,
 931                                   struct mdd_object *obj)
 932 {
 933         struct lu_attr  *tmp_la = &mdd_env_info(env)->mti_la;
 934         struct lu_ucred *uc     = lu_ucred_assert(env);
 935         int rc;
 936         ENTRY;
 937
 938         if (mdd_is_immutable(obj) || mdd_is_append(obj))
 939                 RETURN(-EPERM);
 940
 941         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 942         if (rc)
 943                 RETURN(rc);
 944
 945         if ((uc->uc_fsuid != tmp_la->la_uid) &&
 946             !md_capable(uc, CFS_CAP_FOWNER))
 947                 RETURN(-EPERM);
 948
 949         RETURN(rc);
 950 }
 951
 952 static int mdd_declare_xattr_set(const struct lu_env *env,
 953                                  struct mdd_device *mdd,
 954                                  struct mdd_object *obj,
 955                                  const struct lu_buf *buf,
 956                                  const char *name,
 957                                  int fl, struct thandle *handle)
 958 {
 959         int     rc;
 960
 961         rc = mdo_declare_xattr_set(env, obj, buf, name, fl, handle);
 962         if (rc)
 963                 return rc;
 964
 965         /* Only record user and layout xattr changes */
 966         if (strncmp(XATTR_USER_PREFIX, name,
 967                     sizeof(XATTR_USER_PREFIX) - 1) == 0 ||
 968             strncmp(XATTR_NAME_LOV, name,
 969                     sizeof(XATTR_NAME_LOV) - 1) == 0) {
 970                 rc = mdd_declare_changelog_store(env, mdd, NULL, handle);
 971                 if (rc)
 972                         return rc;
 973         }
 974
 975         /* If HSM data is modified, this could add a changelog */
 976         if (strncmp(XATTR_NAME_HSM, name, sizeof(XATTR_NAME_HSM) - 1) == 0) {
 977                 rc = mdd_declare_changelog_store(env, mdd, NULL, handle);
 978                 if (rc)
 979                         return rc;
 980         }
 981
 982         rc = mdd_declare_changelog_store(env, mdd, NULL, handle);
 983         return rc;
 984 }
 985
 986 /*
 987  * Compare current and future data of HSM EA and add a changelog if needed.
 988  *
 989  * Caller should have write-locked \param obj.
 990  *
 991  * \param buf - Future HSM EA content.
 992  * \retval 0 if no changelog is needed or changelog was added properly.
 993  * \retval -ve errno if there was a problem
 994  */
 995 static int mdd_hsm_update_locked(const struct lu_env *env,
 996                                  struct md_object *obj,
 997                                  const struct lu_buf *buf,
 998                                  struct thandle *handle)
 999 {
1000         struct mdd_thread_info *info = mdd_env_info(env);
1001         struct mdd_device      *mdd = mdo2mdd(obj);
1002         struct mdd_object      *mdd_obj = md2mdd_obj(obj);
1003         struct lu_buf          *current_buf;
1004         struct md_hsm          *current_mh;
1005         struct md_hsm          *new_mh;
1006         int                     rc;
1007         ENTRY;
1008
1009         OBD_ALLOC_PTR(current_mh);
1010         if (current_mh == NULL)
1011                 RETURN(-ENOMEM);
1012
1013         /* Read HSM attrs from disk */
1014         CLASSERT(sizeof(struct hsm_attrs) <= sizeof(info->mti_xattr_buf));
1015         current_buf = mdd_buf_get(env, info->mti_xattr_buf,
1016                                   sizeof(info->mti_xattr_buf));
1017         rc = mdo_xattr_get(env, mdd_obj, current_buf, XATTR_NAME_HSM,
1018                            mdd_object_capa(env, mdd_obj));
1019         rc = lustre_buf2hsm(current_buf->lb_buf, rc, current_mh);
1020         if (rc < 0 && rc != -ENODATA)
1021                 GOTO(free, rc);
1022         else if (rc == -ENODATA)
1023                 current_mh->mh_flags = 0;
1024
1025         /* Map future HSM xattr */
1026         OBD_ALLOC_PTR(new_mh);
1027         if (new_mh == NULL)
1028                 GOTO(free, rc = -ENOMEM);
1029         lustre_buf2hsm(buf->lb_buf, buf->lb_len, new_mh);
1030
1031         /* If HSM flags are different, add a changelog */
1032         rc = 0;
1033         if (current_mh->mh_flags != new_mh->mh_flags) {
1034                 int flags = 0;
1035                 hsm_set_cl_event(&flags, HE_STATE);
1036                 if (new_mh->mh_flags & HS_DIRTY)
1037                         hsm_set_cl_flags(&flags, CLF_HSM_DIRTY);
1038
1039                 rc = mdd_changelog_data_store(env, mdd, CL_HSM, flags, mdd_obj,
1040                                               handle);
1041         }
1042
1043         OBD_FREE_PTR(new_mh);
1044 free:
1045         OBD_FREE_PTR(current_mh);
1046         return(rc);
1047 }
1048
1049 static int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
1050                          const char *name);
1051
1052 /**
1053  * The caller should guarantee to update the object ctime
1054  * after xattr_set if needed.
1055  */
1056 static int mdd_xattr_set(const struct lu_env *env, struct md_object *obj,
1057                          const struct lu_buf *buf, const char *name,
1058                          int fl)
1059 {
1060         struct mdd_object       *mdd_obj = md2mdd_obj(obj);
1061         struct mdd_device       *mdd = mdo2mdd(obj);
1062         struct thandle          *handle;
1063         int                      rc;
1064         ENTRY;
1065
1066         rc = mdd_xattr_sanity_check(env, mdd_obj);
1067         if (rc)
1068                 RETURN(rc);
1069
1070         if (strcmp(name, XATTR_NAME_ACL_ACCESS) == 0 ||
1071             strcmp(name, XATTR_NAME_ACL_DEFAULT) == 0) {
1072                 struct posix_acl *acl;
1073
1074                 /* user may set empty ACL, which should be treated as removing
1075                  * ACL. */
1076                 acl = posix_acl_from_xattr(&init_user_ns, buf->lb_buf,
1077                                            buf->lb_len);
1078                 if (acl == NULL) {
1079                         rc = mdd_xattr_del(env, obj, name);
1080                         RETURN(rc);
1081                 }
1082                 posix_acl_release(acl);
1083         }
1084
1085         if (!strcmp(name, XATTR_NAME_ACL_ACCESS)) {
1086                 rc = mdd_acl_set(env, mdd_obj, buf, fl);
1087                 RETURN(rc);
1088         }
1089
1090         handle = mdd_trans_create(env, mdd);
1091         if (IS_ERR(handle))
1092                 RETURN(PTR_ERR(handle));
1093
1094         rc = mdd_declare_xattr_set(env, mdd, mdd_obj, buf, name, fl, handle);
1095         if (rc)
1096                 GOTO(stop, rc);
1097
1098         rc = mdd_trans_start(env, mdd, handle);
1099         if (rc)
1100                 GOTO(stop, rc);
1101
1102         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1103
1104         if (strncmp(XATTR_NAME_HSM, name, sizeof(XATTR_NAME_HSM) - 1) == 0) {
1105                 rc = mdd_hsm_update_locked(env, obj, buf, handle);
1106                 if (rc) {
1107                         mdd_write_unlock(env, mdd_obj);
1108                         GOTO(stop, rc);
1109                 }
1110         }
1111
1112         rc = mdo_xattr_set(env, mdd_obj, buf, name, fl, handle,
1113                            mdd_object_capa(env, mdd_obj));
1114         mdd_write_unlock(env, mdd_obj);
1115         if (rc)
1116                 GOTO(stop, rc);
1117
1118         if (strncmp(XATTR_NAME_LOV, name, sizeof(XATTR_NAME_LOV) - 1) == 0)
1119                 rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, mdd_obj,
1120                                               handle);
1121         else if (strncmp(XATTR_USER_PREFIX, name,
1122                         sizeof(XATTR_USER_PREFIX) - 1) == 0 ||
1123             strncmp(POSIX_ACL_XATTR_ACCESS, name,
1124                         sizeof(POSIX_ACL_XATTR_ACCESS) - 1) == 0 ||
1125             strncmp(POSIX_ACL_XATTR_DEFAULT, name,
1126                         sizeof(POSIX_ACL_XATTR_DEFAULT) - 1) == 0)
1127                 rc = mdd_changelog_data_store(env, mdd, CL_XATTR, 0, mdd_obj,
1128                                               handle);
1129
1130 stop:
1131         mdd_trans_stop(env, mdd, rc, handle);
1132
1133         RETURN(rc);
1134 }
1135
1136 static int mdd_declare_xattr_del(const struct lu_env *env,
1137                                  struct mdd_device *mdd,
1138                                  struct mdd_object *obj,
1139                                  const char *name,
1140                                  struct thandle *handle)
1141 {
1142         int rc;
1143
1144         rc = mdo_declare_xattr_del(env, obj, name, handle);
1145         if (rc)
1146                 return rc;
1147
1148         /* Only record user xattr changes */
1149         if ((strncmp("user.", name, 5) == 0))
1150                 rc = mdd_declare_changelog_store(env, mdd, NULL, handle);
1151
1152         return rc;
1153 }
1154
1155 /**
1156  * The caller should guarantee to update the object ctime
1157  * after xattr_set if needed.
1158  */
1159 int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
1160                   const char *name)
1161 {
1162         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1163         struct mdd_device *mdd = mdo2mdd(obj);
1164         struct thandle *handle;
1165         int  rc;
1166         ENTRY;
1167
1168         rc = mdd_xattr_sanity_check(env, mdd_obj);
1169         if (rc)
1170                 RETURN(rc);
1171
1172         handle = mdd_trans_create(env, mdd);
1173         if (IS_ERR(handle))
1174                 RETURN(PTR_ERR(handle));
1175
1176         rc = mdd_declare_xattr_del(env, mdd, mdd_obj, name, handle);
1177         if (rc)
1178                 GOTO(stop, rc);
1179
1180         rc = mdd_trans_start(env, mdd, handle);
1181         if (rc)
1182                 GOTO(stop, rc);
1183
1184         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1185         rc = mdo_xattr_del(env, mdd_obj, name, handle,
1186                            mdd_object_capa(env, mdd_obj));
1187         mdd_write_unlock(env, mdd_obj);
1188         if (rc)
1189                 GOTO(stop, rc);
1190
1191         /* Only record system & user xattr changes */
1192         if (strncmp(XATTR_USER_PREFIX, name,
1193                                   sizeof(XATTR_USER_PREFIX) - 1) == 0 ||
1194                           strncmp(POSIX_ACL_XATTR_ACCESS, name,
1195                                   sizeof(POSIX_ACL_XATTR_ACCESS) - 1) == 0 ||
1196                           strncmp(POSIX_ACL_XATTR_DEFAULT, name,
1197                                   sizeof(POSIX_ACL_XATTR_DEFAULT) - 1) == 0)
1198                 rc = mdd_changelog_data_store(env, mdd, CL_XATTR, 0, mdd_obj,
1199                                               handle);
1200
1201 stop:
1202         mdd_trans_stop(env, mdd, rc, handle);
1203
1204         RETURN(rc);
1205 }
1206
1207 /*
1208  * read lov EA of an object
1209  * return the lov EA in an allocated lu_buf
1210  */
1211 static int mdd_get_lov_ea(const struct lu_env *env,
1212                           struct mdd_object *obj,
1213                           struct lu_buf *lmm_buf)
1214 {
1215         struct lu_buf   *buf = &mdd_env_info(env)->mti_big_buf;
1216         int              rc, sz;
1217         ENTRY;
1218
1219 repeat:
1220         rc = mdo_xattr_get(env, obj, buf, XATTR_NAME_LOV,
1221                            mdd_object_capa(env, obj));
1222
1223         if (rc == -ERANGE) {
1224                 /* mti_big_buf is allocated but is too small
1225                  * we need to increase it */
1226                 buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf,
1227                                              buf->lb_len * 2);
1228                 if (buf->lb_buf == NULL)
1229                         GOTO(out, rc = -ENOMEM);
1230                 goto repeat;
1231         }
1232
1233         if (rc < 0)
1234                 GOTO(out, rc);
1235
1236         if (rc == 0)
1237                 GOTO(out, rc = -ENODATA);
1238
1239         sz = rc;
1240         if (memcmp(buf, &LU_BUF_NULL, sizeof(*buf)) == 0) {
1241                 /* mti_big_buf was not allocated, so we have to
1242                  * allocate it based on the ea size */
1243                 buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf,
1244                                              sz);
1245                 if (buf->lb_buf == NULL)
1246                         GOTO(out, rc = -ENOMEM);
1247                 goto repeat;
1248         }
1249
1250         lu_buf_alloc(lmm_buf, sz);
1251         if (lmm_buf->lb_buf == NULL)
1252                 GOTO(out, rc = -ENOMEM);
1253
1254         memcpy(lmm_buf->lb_buf, buf->lb_buf, sz);
1255         rc = 0;
1256         EXIT;
1257
1258 out:
1259         if (rc < 0)
1260                 lu_buf_free(lmm_buf);
1261         return rc;
1262 }
1263
1264 static int mdd_xattr_hsm_replace(const struct lu_env *env,
1265                                  struct mdd_object *o, struct lu_buf *buf,
1266                                  struct thandle *handle)
1267 {
1268         struct hsm_attrs *attrs;
1269         __u32 hsm_flags;
1270         int flags = 0;
1271         int rc;
1272         ENTRY;
1273
1274         rc = mdo_xattr_set(env, o, buf, XATTR_NAME_HSM, LU_XATTR_REPLACE,
1275                            handle, mdd_object_capa(env, o));
1276         if (rc != 0)
1277                 RETURN(rc);
1278
1279         attrs = buf->lb_buf;
1280         hsm_flags = le32_to_cpu(attrs->hsm_flags);
1281         if (!(hsm_flags & HS_RELEASED) || mdd_is_dead_obj(o))
1282                 RETURN(0);
1283
1284         /* Add a changelog record for release. */
1285         hsm_set_cl_event(&flags, HE_RELEASE);
1286         rc = mdd_changelog_data_store(env, mdo2mdd(&o->mod_obj), CL_HSM,
1287                                       flags, o, handle);
1288         RETURN(rc);
1289 }
1290
1291 /*
1292  *  check if layout swapping between 2 objects is allowed
1293  *  the rules are:
1294  *  - only normal FIDs or non-system IGIFs
1295  *  - same type of objects
1296  *  - same owner/group (so quotas are still valid)
1297  */
1298 static int mdd_layout_swap_allowed(const struct lu_env *env,
1299                                    struct mdd_object *o1,
1300                                    struct mdd_object *o2)
1301 {
1302         const struct lu_fid     *fid1, *fid2;
1303         __u32                    uid, gid;
1304         struct lu_attr          *tmp_la = &mdd_env_info(env)->mti_la;
1305         int                      rc;
1306         ENTRY;
1307
1308         fid1 = mdo2fid(o1);
1309         fid2 = mdo2fid(o2);
1310
1311         if (!fid_is_norm(fid1) &&
1312             (!fid_is_igif(fid1) || IS_ERR(mdd_links_get(env, o1))))
1313                 RETURN(-EBADF);
1314
1315         if (!fid_is_norm(fid2) &&
1316             (!fid_is_igif(fid2) || IS_ERR(mdd_links_get(env, o2))))
1317                 RETURN(-EBADF);
1318
1319         if (mdd_object_type(o1) != mdd_object_type(o2)) {
1320                 if (S_ISDIR(mdd_object_type(o1)))
1321                         RETURN(-ENOTDIR);
1322                 if (S_ISREG(mdd_object_type(o1)))
1323                         RETURN(-EISDIR);
1324                 RETURN(-EBADF);
1325         }
1326
1327         tmp_la->la_valid = 0;
1328         rc = mdd_la_get(env, o1, tmp_la, BYPASS_CAPA);
1329         if (rc)
1330                 RETURN(rc);
1331         uid = tmp_la->la_uid;
1332         gid = tmp_la->la_gid;
1333
1334         tmp_la->la_valid = 0;
1335         rc = mdd_la_get(env, o2, tmp_la, BYPASS_CAPA);
1336         if (rc)
1337                 RETURN(rc);
1338
1339         if ((uid != tmp_la->la_uid) || (gid != tmp_la->la_gid))
1340                 RETURN(-EPERM);
1341
1342         RETURN(0);
1343 }
1344
1345 /**
1346  * swap layouts between 2 lustre objects
1347  */
1348 static int mdd_swap_layouts(const struct lu_env *env, struct md_object *obj1,
1349                             struct md_object *obj2, __u64 flags)
1350 {
1351         struct mdd_thread_info  *info = mdd_env_info(env);
1352         struct mdd_object       *fst_o = md2mdd_obj(obj1);
1353         struct mdd_object       *snd_o = md2mdd_obj(obj2);
1354         struct mdd_device       *mdd = mdo2mdd(obj1);
1355         struct lov_mds_md       *fst_lmm, *snd_lmm;
1356         struct lu_buf           *fst_buf = &info->mti_buf[0];
1357         struct lu_buf           *snd_buf = &info->mti_buf[1];
1358         struct lu_buf           *fst_hsm_buf = &info->mti_buf[2];
1359         struct lu_buf           *snd_hsm_buf = &info->mti_buf[3];
1360         struct ost_id           *saved_oi = NULL;
1361         struct thandle          *handle;
1362         __u16                    fst_gen, snd_gen;
1363         int                      fst_fl;
1364         int                      rc;
1365         int                      rc2;
1366         ENTRY;
1367
1368         CLASSERT(ARRAY_SIZE(info->mti_buf) >= 4);
1369         memset(info->mti_buf, 0, sizeof(info->mti_buf));
1370
1371         /* we have to sort the 2 obj, so locking will always
1372          * be in the same order, even in case of 2 concurrent swaps */
1373         rc = lu_fid_cmp(mdo2fid(fst_o), mdo2fid(snd_o));
1374         if (rc == 0) /* same fid ? */
1375                 RETURN(-EPERM);
1376
1377         if (rc < 0)
1378                 swap(fst_o, snd_o);
1379
1380         /* check if layout swapping is allowed */
1381         rc = mdd_layout_swap_allowed(env, fst_o, snd_o);
1382         if (rc != 0)
1383                 RETURN(rc);
1384
1385         handle = mdd_trans_create(env, mdd);
1386         if (IS_ERR(handle))
1387                 RETURN(PTR_ERR(handle));
1388
1389         /* objects are already sorted */
1390         mdd_write_lock(env, fst_o, MOR_TGT_CHILD);
1391         mdd_write_lock(env, snd_o, MOR_TGT_CHILD);
1392
1393         rc = mdd_get_lov_ea(env, fst_o, fst_buf);
1394         if (rc < 0 && rc != -ENODATA)
1395                 GOTO(stop, rc);
1396
1397         rc = mdd_get_lov_ea(env, snd_o, snd_buf);
1398         if (rc < 0 && rc != -ENODATA)
1399                 GOTO(stop, rc);
1400
1401         /* swapping 2 non existant layouts is a success */
1402         if (fst_buf->lb_buf == NULL && snd_buf->lb_buf == NULL)
1403                 GOTO(stop, rc = 0);
1404
1405         /* to help inode migration between MDT, it is better to
1406          * start by the no layout file (if one), so we order the swap */
1407         if (snd_buf->lb_buf == NULL) {
1408                 swap(fst_o, snd_o);
1409                 swap(fst_buf, snd_buf);
1410         }
1411
1412         /* lmm and generation layout initialization */
1413         if (fst_buf->lb_buf != NULL) {
1414                 fst_lmm = fst_buf->lb_buf;
1415                 fst_gen = le16_to_cpu(fst_lmm->lmm_layout_gen);
1416                 fst_fl  = LU_XATTR_REPLACE;
1417         } else {
1418                 fst_lmm = NULL;
1419                 fst_gen = 0;
1420                 fst_fl  = LU_XATTR_CREATE;
1421         }
1422
1423         snd_lmm = snd_buf->lb_buf;
1424         snd_gen = le16_to_cpu(snd_lmm->lmm_layout_gen);
1425
1426         /* increase the generation layout numbers */
1427         snd_gen++;
1428         fst_gen++;
1429
1430         /* set the file specific informations in lmm */
1431         if (fst_lmm != NULL) {
1432                 saved_oi = &info->mti_oa.o_oi;
1433
1434                 *saved_oi = fst_lmm->lmm_oi;
1435                 fst_lmm->lmm_layout_gen = cpu_to_le16(snd_gen);
1436                 fst_lmm->lmm_oi = snd_lmm->lmm_oi;
1437                 snd_lmm->lmm_oi = *saved_oi;
1438         } else {
1439                 if (snd_lmm->lmm_magic == cpu_to_le32(LOV_MAGIC_V1))
1440                         snd_lmm->lmm_magic = cpu_to_le32(LOV_MAGIC_V1_DEF);
1441                 else if (snd_lmm->lmm_magic == cpu_to_le32(LOV_MAGIC_V3))
1442                         snd_lmm->lmm_magic = cpu_to_le32(LOV_MAGIC_V3_DEF);
1443                 else
1444                         GOTO(stop, rc = -EPROTO);
1445         }
1446         snd_lmm->lmm_layout_gen = cpu_to_le16(fst_gen);
1447
1448         /* Prepare HSM attribute if it's required */
1449         if (flags & SWAP_LAYOUTS_MDS_HSM) {
1450                 const int buflen = sizeof(struct hsm_attrs);
1451
1452                 lu_buf_alloc(fst_hsm_buf, buflen);
1453                 lu_buf_alloc(snd_hsm_buf, buflen);
1454                 if (fst_hsm_buf->lb_buf == NULL || snd_hsm_buf->lb_buf == NULL)
1455                         GOTO(stop, rc = -ENOMEM);
1456
1457                 /* Read HSM attribute */
1458                 rc = mdo_xattr_get(env, fst_o, fst_hsm_buf, XATTR_NAME_HSM,
1459                                    BYPASS_CAPA);
1460                 if (rc < 0)
1461                         GOTO(stop, rc);
1462
1463                 rc = mdo_xattr_get(env, snd_o, snd_hsm_buf, XATTR_NAME_HSM,
1464                                    BYPASS_CAPA);
1465                 if (rc < 0)
1466                         GOTO(stop, rc);
1467
1468                 rc = mdd_declare_xattr_set(env, mdd, fst_o, snd_hsm_buf,
1469                                            XATTR_NAME_HSM, LU_XATTR_REPLACE,
1470                                            handle);
1471                 if (rc < 0)
1472                         GOTO(stop, rc);
1473
1474                 rc = mdd_declare_xattr_set(env, mdd, snd_o, fst_hsm_buf,
1475                                            XATTR_NAME_HSM, LU_XATTR_REPLACE,
1476                                            handle);
1477                 if (rc < 0)
1478                         GOTO(stop, rc);
1479         }
1480
1481         /* prepare transaction */
1482         rc = mdd_declare_xattr_set(env, mdd, fst_o, snd_buf, XATTR_NAME_LOV,
1483                                    fst_fl, handle);
1484         if (rc != 0)
1485                 GOTO(stop, rc);
1486
1487         if (fst_buf->lb_buf != NULL)
1488                 rc = mdd_declare_xattr_set(env, mdd, snd_o, fst_buf,
1489                                            XATTR_NAME_LOV, LU_XATTR_REPLACE,
1490                                            handle);
1491         else
1492                 rc = mdd_declare_xattr_del(env, mdd, snd_o, XATTR_NAME_LOV,
1493                                            handle);
1494         if (rc != 0)
1495                 GOTO(stop, rc);
1496
1497         rc = mdd_trans_start(env, mdd, handle);
1498         if (rc != 0)
1499                 GOTO(stop, rc);
1500
1501         if (flags & SWAP_LAYOUTS_MDS_HSM) {
1502                 rc = mdd_xattr_hsm_replace(env, fst_o, snd_hsm_buf, handle);
1503                 if (rc < 0)
1504                         GOTO(stop, rc);
1505
1506                 rc = mdd_xattr_hsm_replace(env, snd_o, fst_hsm_buf, handle);
1507                 if (rc < 0) {
1508                         rc2 = mdd_xattr_hsm_replace(env, fst_o, fst_hsm_buf,
1509                                                     handle);
1510                         if (rc2 < 0)
1511                                 CERROR("%s: restore "DFID" HSM error: %d/%d\n",
1512                                        mdd_obj_dev_name(fst_o),
1513                                        PFID(mdo2fid(fst_o)), rc, rc2);
1514                         GOTO(stop, rc);
1515                 }
1516         }
1517
1518         rc = mdo_xattr_set(env, fst_o, snd_buf, XATTR_NAME_LOV, fst_fl, handle,
1519                            mdd_object_capa(env, fst_o));
1520         if (rc != 0)
1521                 GOTO(stop, rc);
1522
1523         if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_MDS_HSM_SWAP_LAYOUTS))) {
1524                 rc = -EOPNOTSUPP;
1525         } else {
1526                 if (fst_buf->lb_buf != NULL)
1527                         rc = mdo_xattr_set(env, snd_o, fst_buf, XATTR_NAME_LOV,
1528                                            LU_XATTR_REPLACE, handle,
1529                                            mdd_object_capa(env, snd_o));
1530                 else
1531                         rc = mdo_xattr_del(env, snd_o, XATTR_NAME_LOV, handle,
1532                                            mdd_object_capa(env, snd_o));
1533         }
1534
1535         if (rc != 0) {
1536                 int steps = 0;
1537
1538                 /* failure on second file, but first was done, so we have
1539                  * to roll back first. */
1540                 if (fst_buf->lb_buf != NULL) {
1541                         fst_lmm->lmm_oi = *saved_oi;
1542                         fst_lmm->lmm_layout_gen = cpu_to_le16(fst_gen - 1);
1543                         rc2 = mdo_xattr_set(env, fst_o, fst_buf, XATTR_NAME_LOV,
1544                                             LU_XATTR_REPLACE, handle,
1545                                             mdd_object_capa(env, fst_o));
1546                 } else {
1547                         rc2 = mdo_xattr_del(env, fst_o, XATTR_NAME_LOV, handle,
1548                                             mdd_object_capa(env, fst_o));
1549                 }
1550                 if (rc2 < 0)
1551                         goto do_lbug;
1552
1553                 ++steps;
1554                 rc2 = mdd_xattr_hsm_replace(env, fst_o, fst_hsm_buf, handle);
1555                 if (rc2 < 0)
1556                         goto do_lbug;
1557
1558                 ++steps;
1559                 rc2 = mdd_xattr_hsm_replace(env, snd_o, snd_hsm_buf, handle);
1560
1561         do_lbug:
1562                 if (rc2 < 0) {
1563                         /* very bad day */
1564                         CERROR("%s: unable to roll back layout swap. FIDs: "
1565                                DFID" and "DFID "error: %d/%d, steps: %d\n",
1566                                mdd_obj_dev_name(fst_o),
1567                                PFID(mdo2fid(snd_o)), PFID(mdo2fid(fst_o)),
1568                                rc, rc2, steps);
1569                         /* a solution to avoid journal commit is to panic,
1570                          * but it has strong consequences so we use LBUG to
1571                          * allow sysdamin to choose to panic or not
1572                          */
1573                         LBUG();
1574                 }
1575                 GOTO(stop, rc);
1576         }
1577
1578         /* Issue one changelog record per file */
1579         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, fst_o, handle);
1580         if (rc)
1581                 GOTO(stop, rc);
1582
1583         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, snd_o, handle);
1584         if (rc)
1585                 GOTO(stop, rc);
1586         EXIT;
1587
1588 stop:
1589         mdd_trans_stop(env, mdd, rc, handle);
1590         mdd_write_unlock(env, snd_o);
1591         mdd_write_unlock(env, fst_o);
1592
1593         lu_buf_free(fst_buf);
1594         lu_buf_free(snd_buf);
1595         lu_buf_free(fst_hsm_buf);
1596         lu_buf_free(snd_hsm_buf);
1597         return rc;
1598 }
1599
1600 void mdd_object_make_hint(const struct lu_env *env, struct mdd_object *parent,
1601                 struct mdd_object *child, struct lu_attr *attr)
1602 {
1603         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
1604         struct dt_object *np = parent ? mdd_object_child(parent) : NULL;
1605         struct dt_object *nc = mdd_object_child(child);
1606
1607         /* @hint will be initialized by underlying device. */
1608         nc->do_ops->do_ah_init(env, hint, np, nc, attr->la_mode & S_IFMT);
1609 }
1610
1611 /*
1612  * do NOT or the MAY_*'s, you'll get the weakest
1613  */
1614 int accmode(const struct lu_env *env, struct lu_attr *la, int flags)
1615 {
1616         int res = 0;
1617
1618         /* Sadly, NFSD reopens a file repeatedly during operation, so the
1619          * "acc_mode = 0" allowance for newly-created files isn't honoured.
1620          * NFSD uses the MDS_OPEN_OWNEROVERRIDE flag to say that a file
1621          * owner can write to a file even if it is marked readonly to hide
1622          * its brokenness. (bug 5781) */
1623         if (flags & MDS_OPEN_OWNEROVERRIDE) {
1624                 struct lu_ucred *uc = lu_ucred_check(env);
1625
1626                 if ((uc == NULL) || (la->la_uid == uc->uc_fsuid))
1627                         return 0;
1628         }
1629
1630         if (flags & FMODE_READ)
1631                 res |= MAY_READ;
1632         if (flags & (FMODE_WRITE | MDS_OPEN_TRUNC | MDS_OPEN_APPEND))
1633                 res |= MAY_WRITE;
1634         if (flags & MDS_FMODE_EXEC)
1635                 res = MAY_EXEC;
1636         return res;
1637 }
1638
1639 static int mdd_open_sanity_check(const struct lu_env *env,
1640                                  struct mdd_object *obj, int flag)
1641 {
1642         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
1643         int mode, rc;
1644         ENTRY;
1645
1646         /* EEXIST check */
1647         if (mdd_is_dead_obj(obj))
1648                 RETURN(-ENOENT);
1649
1650         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
1651         if (rc)
1652                RETURN(rc);
1653
1654         if (S_ISLNK(tmp_la->la_mode))
1655                 RETURN(-ELOOP);
1656
1657         mode = accmode(env, tmp_la, flag);
1658
1659         if (S_ISDIR(tmp_la->la_mode) && (mode & MAY_WRITE))
1660                 RETURN(-EISDIR);
1661
1662         if (!(flag & MDS_OPEN_CREATED)) {
1663                 rc = mdd_permission_internal(env, obj, tmp_la, mode);
1664                 if (rc)
1665                         RETURN(rc);
1666         }
1667
1668         if (S_ISFIFO(tmp_la->la_mode) || S_ISSOCK(tmp_la->la_mode) ||
1669             S_ISBLK(tmp_la->la_mode) || S_ISCHR(tmp_la->la_mode))
1670                 flag &= ~MDS_OPEN_TRUNC;
1671
1672         /* For writing append-only file must open it with append mode. */
1673         if (mdd_is_append(obj)) {
1674                 if ((flag & FMODE_WRITE) && !(flag & MDS_OPEN_APPEND))
1675                         RETURN(-EPERM);
1676                 if (flag & MDS_OPEN_TRUNC)
1677                         RETURN(-EPERM);
1678         }
1679
1680 #if 0
1681         /*
1682          * Now, flag -- O_NOATIME does not be packed by client.
1683          */
1684         if (flag & O_NOATIME) {
1685                 struct lu_ucred *uc = lu_ucred(env);
1686
1687                 if (uc && ((uc->uc_valid == UCRED_OLD) ||
1688                            (uc->uc_valid == UCRED_NEW)) &&
1689                     (uc->uc_fsuid != tmp_la->la_uid) &&
1690                     !md_capable(uc, CFS_CAP_FOWNER))
1691                         RETURN(-EPERM);
1692         }
1693 #endif
1694
1695         RETURN(0);
1696 }
1697
1698 static int mdd_open(const struct lu_env *env, struct md_object *obj,
1699                     int flags)
1700 {
1701         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1702         int rc = 0;
1703
1704         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1705
1706         rc = mdd_open_sanity_check(env, mdd_obj, flags);
1707         if (rc == 0)
1708                 mdd_obj->mod_count++;
1709
1710         mdd_write_unlock(env, mdd_obj);
1711         return rc;
1712 }
1713
1714 int mdd_declare_object_kill(const struct lu_env *env, struct mdd_object *obj,
1715                             struct md_attr *ma, struct thandle *handle)
1716 {
1717         return mdo_declare_destroy(env, obj, handle);
1718 }
1719
1720 /* return md_attr back,
1721  * if it is last unlink then return lov ea + llog cookie*/
1722 int mdd_object_kill(const struct lu_env *env, struct mdd_object *obj,
1723                     struct md_attr *ma, struct thandle *handle)
1724 {
1725         int rc;
1726         ENTRY;
1727
1728         rc = mdo_destroy(env, obj, handle);
1729
1730         RETURN(rc);
1731 }
1732
1733 static int mdd_declare_close(const struct lu_env *env,
1734                              struct mdd_object *obj,
1735                              struct md_attr *ma,
1736                              struct thandle *handle)
1737 {
1738         int rc;
1739
1740         rc = orph_declare_index_delete(env, obj, handle);
1741         if (rc)
1742                 return rc;
1743
1744         return mdo_declare_destroy(env, obj, handle);
1745 }
1746
1747 /*
1748  * No permission check is needed.
1749  */
1750 static int mdd_close(const struct lu_env *env, struct md_object *obj,
1751                      struct md_attr *ma, int mode)
1752 {
1753         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1754         struct mdd_device *mdd = mdo2mdd(obj);
1755         struct thandle    *handle = NULL;
1756         int rc, is_orphan = 0;
1757         ENTRY;
1758
1759         if (ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_KEEP_ORPHAN) {
1760                 mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1761                 mdd_obj->mod_count--;
1762                 mdd_write_unlock(env, mdd_obj);
1763
1764                 if (mdd_obj->mod_flags & ORPHAN_OBJ && !mdd_obj->mod_count)
1765                         CDEBUG(D_HA, "Object "DFID" is retained in orphan "
1766                                "list\n", PFID(mdd_object_fid(mdd_obj)));
1767                 RETURN(0);
1768         }
1769
1770         /* mdd_finish_unlink() will always set orphan object as DEAD_OBJ, but
1771          * it might fail to add the object to orphan list (w/o ORPHAN_OBJ). */
1772         /* check without any lock */
1773         is_orphan = mdd_obj->mod_count == 1 &&
1774                     (mdd_obj->mod_flags & (ORPHAN_OBJ | DEAD_OBJ)) != 0;
1775
1776 again:
1777         if (is_orphan) {
1778                 handle = mdd_trans_create(env, mdo2mdd(obj));
1779                 if (IS_ERR(handle))
1780                         RETURN(PTR_ERR(handle));
1781
1782                 rc = mdd_declare_close(env, mdd_obj, ma, handle);
1783                 if (rc)
1784                         GOTO(stop, rc);
1785
1786                 rc = mdd_declare_changelog_store(env, mdd, NULL, handle);
1787                 if (rc)
1788                         GOTO(stop, rc);
1789
1790                 rc = mdd_trans_start(env, mdo2mdd(obj), handle);
1791                 if (rc)
1792                         GOTO(stop, rc);
1793         }
1794
1795         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1796         rc = mdd_la_get(env, mdd_obj, &ma->ma_attr,
1797                         mdd_object_capa(env, mdd_obj));
1798         if (rc != 0) {
1799                 CERROR("Failed to get lu_attr of "DFID": %d\n",
1800                        PFID(mdd_object_fid(mdd_obj)), rc);
1801                 GOTO(out, rc);
1802         }
1803
1804         /* check again with lock */
1805         is_orphan = (mdd_obj->mod_count == 1) &&
1806                     ((mdd_obj->mod_flags & (ORPHAN_OBJ | DEAD_OBJ)) != 0 ||
1807                      ma->ma_attr.la_nlink == 0);
1808
1809         if (is_orphan && handle == NULL) {
1810                 mdd_write_unlock(env, mdd_obj);
1811                 goto again;
1812         }
1813
1814         mdd_obj->mod_count--; /*release open count */
1815
1816         if (!is_orphan)
1817                 GOTO(out, rc = 0);
1818
1819         /* Orphan object */
1820         /* NB: Object maybe not in orphan list originally, it is rare case for
1821          * mdd_finish_unlink() failure, in that case, the object doesn't have
1822          * ORPHAN_OBJ flag */
1823         if ((mdd_obj->mod_flags & ORPHAN_OBJ) != 0) {
1824                 /* remove link to object from orphan index */
1825                 LASSERT(handle != NULL);
1826                 rc = __mdd_orphan_del(env, mdd_obj, handle);
1827                 if (rc != 0) {
1828                         CERROR("%s: unable to delete "DFID" from orphan list: "
1829                                "rc = %d\n", lu_dev_name(mdd2lu_dev(mdd)),
1830                                PFID(mdd_object_fid(mdd_obj)), rc);
1831                         /* If object was not deleted from orphan list, do not
1832                          * destroy OSS objects, which will be done when next
1833                          * recovery. */
1834                         GOTO(out, rc);
1835                 }
1836
1837                 CDEBUG(D_HA, "Object "DFID" is deleted from orphan "
1838                        "list, OSS objects to be destroyed.\n",
1839                        PFID(mdd_object_fid(mdd_obj)));
1840         }
1841
1842         rc = mdo_destroy(env, mdd_obj, handle);
1843
1844         if (rc != 0) {
1845                 CERROR("%s: unable to delete "DFID" from orphan list: "
1846                        "rc = %d\n", lu_dev_name(mdd2lu_dev(mdd)),
1847                        PFID(mdd_object_fid(mdd_obj)), rc);
1848         }
1849         EXIT;
1850
1851 out:
1852         mdd_write_unlock(env, mdd_obj);
1853
1854         if (rc == 0 &&
1855             (mode & (FMODE_WRITE | MDS_OPEN_APPEND | MDS_OPEN_TRUNC)) &&
1856             !(ma->ma_valid & MA_FLAGS && ma->ma_attr_flags & MDS_RECOV_OPEN)) {
1857                 if (handle == NULL) {
1858                         handle = mdd_trans_create(env, mdo2mdd(obj));
1859                         if (IS_ERR(handle))
1860                                 GOTO(stop, rc = PTR_ERR(handle));
1861
1862                         rc = mdd_declare_changelog_store(env, mdd, NULL,
1863                                                          handle);
1864                         if (rc)
1865                                 GOTO(stop, rc);
1866
1867                         rc = mdd_trans_start(env, mdo2mdd(obj), handle);
1868                         if (rc)
1869                                 GOTO(stop, rc);
1870                 }
1871
1872                 mdd_changelog_data_store(env, mdd, CL_CLOSE, mode,
1873                                          mdd_obj, handle);
1874         }
1875
1876 stop:
1877         if (handle != NULL && !IS_ERR(handle))
1878                 mdd_trans_stop(env, mdd, rc, handle);
1879
1880         return rc;
1881 }
1882
1883 /*
1884  * Permission check is done when open,
1885  * no need check again.
1886  */
1887 static int mdd_readpage_sanity_check(const struct lu_env *env,
1888                                      struct mdd_object *obj)
1889 {
1890         struct dt_object *next = mdd_object_child(obj);
1891         int rc;
1892         ENTRY;
1893
1894         if (S_ISDIR(mdd_object_type(obj)) && dt_try_as_dir(env, next))
1895                 rc = 0;
1896         else
1897                 rc = -ENOTDIR;
1898
1899         RETURN(rc);
1900 }
1901
1902 static int mdd_dir_page_build(const struct lu_env *env, union lu_page *lp,
1903                               int nob, const struct dt_it_ops *iops,
1904                               struct dt_it *it, __u32 attr, void *arg)
1905 {
1906         struct lu_dirpage       *dp = &lp->lp_dir;
1907         void                    *area = dp;
1908         int                      result;
1909         __u64                    hash = 0;
1910         struct lu_dirent        *ent;
1911         struct lu_dirent        *last = NULL;
1912         int                      first = 1;
1913
1914         memset(area, 0, sizeof (*dp));
1915         area += sizeof (*dp);
1916         nob  -= sizeof (*dp);
1917
1918         ent  = area;
1919         do {
1920                 int    len;
1921                 int    recsize;
1922
1923                 len = iops->key_size(env, it);
1924
1925                 /* IAM iterator can return record with zero len. */
1926                 if (len == 0)
1927                         goto next;
1928
1929                 hash = iops->store(env, it);
1930                 if (unlikely(first)) {
1931                         first = 0;
1932                         dp->ldp_hash_start = cpu_to_le64(hash);
1933                 }
1934
1935                 /* calculate max space required for lu_dirent */
1936                 recsize = lu_dirent_calc_size(len, attr);
1937
1938                 if (nob >= recsize) {
1939                         result = iops->rec(env, it, (struct dt_rec *)ent, attr);
1940                         if (result == -ESTALE)
1941                                 goto next;
1942                         if (result != 0)
1943                                 goto out;
1944
1945                         /* osd might not able to pack all attributes,
1946                          * so recheck rec length */
1947                         recsize = le16_to_cpu(ent->lde_reclen);
1948                 } else {
1949                         result = (last != NULL) ? 0 :-EINVAL;
1950                         goto out;
1951                 }
1952                 last = ent;
1953                 ent = (void *)ent + recsize;
1954                 nob -= recsize;
1955
1956 next:
1957                 result = iops->next(env, it);
1958                 if (result == -ESTALE)
1959                         goto next;
1960         } while (result == 0);
1961
1962 out:
1963         dp->ldp_hash_end = cpu_to_le64(hash);
1964         if (last != NULL) {
1965                 if (last->lde_hash == dp->ldp_hash_end)
1966                         dp->ldp_flags |= cpu_to_le32(LDF_COLLIDE);
1967                 last->lde_reclen = 0; /* end mark */
1968         }
1969         if (result > 0)
1970                 /* end of directory */
1971                 dp->ldp_hash_end = cpu_to_le64(MDS_DIR_END_OFF);
1972         else if (result < 0)
1973                 CWARN("build page failed: %d!\n", result);
1974         return result;
1975 }
1976
1977 int mdd_readpage(const struct lu_env *env, struct md_object *obj,
1978                  const struct lu_rdpg *rdpg)
1979 {
1980         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1981         int rc;
1982         ENTRY;
1983
1984         if (mdd_object_exists(mdd_obj) == 0) {
1985                 CERROR("%s: object "DFID" not found: rc = -2\n",
1986                        mdd_obj_dev_name(mdd_obj),PFID(mdd_object_fid(mdd_obj)));
1987                 return -ENOENT;
1988         }
1989
1990         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
1991         rc = mdd_readpage_sanity_check(env, mdd_obj);
1992         if (rc)
1993                 GOTO(out_unlock, rc);
1994
1995         if (mdd_is_dead_obj(mdd_obj)) {
1996                 struct page *pg;
1997                 struct lu_dirpage *dp;
1998
1999                 /*
2000                  * According to POSIX, please do not return any entry to client:
2001                  * even dot and dotdot should not be returned.
2002                  */
2003                 CDEBUG(D_INODE, "readdir from dead object: "DFID"\n",
2004                        PFID(mdd_object_fid(mdd_obj)));
2005
2006                 if (rdpg->rp_count <= 0)
2007                         GOTO(out_unlock, rc = -EFAULT);
2008                 LASSERT(rdpg->rp_pages != NULL);
2009
2010                 pg = rdpg->rp_pages[0];
2011                 dp = (struct lu_dirpage *)kmap(pg);
2012                 memset(dp, 0 , sizeof(struct lu_dirpage));
2013                 dp->ldp_hash_start = cpu_to_le64(rdpg->rp_hash);
2014                 dp->ldp_hash_end   = cpu_to_le64(MDS_DIR_END_OFF);
2015                 dp->ldp_flags = cpu_to_le32(LDF_EMPTY);
2016                 kunmap(pg);
2017                 GOTO(out_unlock, rc = LU_PAGE_SIZE);
2018         }
2019
2020         rc = dt_index_walk(env, mdd_object_child(mdd_obj), rdpg,
2021                            mdd_dir_page_build, NULL);
2022         if (rc >= 0) {
2023                 struct lu_dirpage       *dp;
2024
2025                 dp = kmap(rdpg->rp_pages[0]);
2026                 dp->ldp_hash_start = cpu_to_le64(rdpg->rp_hash);
2027                 if (rc == 0) {
2028                         /*
2029                          * No pages were processed, mark this for first page
2030                          * and send back.
2031                          */
2032                         dp->ldp_flags = cpu_to_le32(LDF_EMPTY);
2033                         rc = min_t(unsigned int, LU_PAGE_SIZE, rdpg->rp_count);
2034                 }
2035                 kunmap(rdpg->rp_pages[0]);
2036         }
2037
2038         GOTO(out_unlock, rc);
2039 out_unlock:
2040         mdd_read_unlock(env, mdd_obj);
2041         return rc;
2042 }
2043
2044 static int mdd_object_sync(const struct lu_env *env, struct md_object *obj)
2045 {
2046         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2047
2048         if (mdd_object_exists(mdd_obj) == 0) {
2049                 CERROR("%s: object "DFID" not found: rc = -2\n",
2050                        mdd_obj_dev_name(mdd_obj),PFID(mdd_object_fid(mdd_obj)));
2051                 return -ENOENT;
2052         }
2053         return dt_object_sync(env, mdd_object_child(mdd_obj));
2054 }
2055
2056 static int mdd_object_lock(const struct lu_env *env,
2057                            struct md_object *obj,
2058                            struct lustre_handle *lh,
2059                            struct ldlm_enqueue_info *einfo,
2060                            void *policy)
2061 {
2062         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2063         return dt_object_lock(env, mdd_object_child(mdd_obj), lh,
2064                               einfo, policy);
2065 }
2066
2067 const struct md_object_operations mdd_obj_ops = {
2068         .moo_permission         = mdd_permission,
2069         .moo_attr_get           = mdd_attr_get,
2070         .moo_attr_set           = mdd_attr_set,
2071         .moo_xattr_get          = mdd_xattr_get,
2072         .moo_xattr_set          = mdd_xattr_set,
2073         .moo_xattr_list         = mdd_xattr_list,
2074         .moo_xattr_del          = mdd_xattr_del,
2075         .moo_swap_layouts       = mdd_swap_layouts,
2076         .moo_open               = mdd_open,
2077         .moo_close              = mdd_close,
2078         .moo_readpage           = mdd_readpage,
2079         .moo_readlink           = mdd_readlink,
2080         .moo_changelog          = mdd_changelog,
2081         .moo_capa_get           = mdd_capa_get,
2082         .moo_object_sync        = mdd_object_sync,
2083         .moo_object_lock        = mdd_object_lock,
2084 };