lustre/mdd/mdd_object.c

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