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