Whamcloud - gitweb
land clio.
[fs/lustre-release.git] / lustre / mdd / mdd_object.c
index ec6f727..5cf0a15 100644 (file)
@@ -1,30 +1,45 @@
-/* -*- MODE: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
+/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
  * vim:expandtab:shiftwidth=8:tabstop=8:
  *
- *  mdd/mdd_handler.c
- *  Lustre Metadata Server (mdd) routines
+ * GPL HEADER START
  *
- *  Copyright (C) 2006 Cluster File Systems, Inc.
- *   Author: Wang Di <wangdi@clusterfs.com>
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
- *   This file is part of the Lustre file system, http://www.lustre.org
- *   Lustre is a trademark of Cluster File Systems, Inc.
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 only,
+ * as published by the Free Software Foundation.
  *
- *   You may have signed or agreed to another license before downloading
- *   this software.  If so, you are bound by the terms and conditions
- *   of that agreement, and the following does not apply to you.  See the
- *   LICENSE file included with this distribution for more information.
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License version 2 for more details (a copy is included
+ * in the LICENSE file that accompanied this code).
  *
- *   If you did not agree to a different license, then this copy of Lustre
- *   is open source software; you can redistribute it and/or modify it
- *   under the terms of version 2 of the GNU General Public License as
- *   published by the Free Software Foundation.
+ * You should have received a copy of the GNU General Public License
+ * version 2 along with this program; If not, see
+ * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  *
- *   In either case, Lustre is distributed in the hope that it will be
- *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
- *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *   license text for more details.
+ * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
+ * CA 95054 USA or visit www.sun.com if you need additional information or
+ * have any questions.
+ *
+ * GPL HEADER END
+ */
+/*
+ * Copyright  2008 Sun Microsystems, Inc. All rights reserved
+ * Use is subject to license terms.
+ */
+/*
+ * This file is part of Lustre, http://www.lustre.org/
+ * Lustre is a trademark of Sun Microsystems, Inc.
+ *
+ * lustre/mdd/mdd_object.c
+ *
+ * Lustre Metadata Server (mdd) routines
+ *
+ * Author: Wang Di <wangdi@clusterfs.com>
  */
+
 #ifndef EXPORT_SYMTAB
 # define EXPORT_SYMTAB
 #endif
@@ -45,7 +60,7 @@
 
 #include "mdd_internal.h"
 
-static struct lu_object_operations mdd_lu_obj_ops;
+static const struct lu_object_operations mdd_lu_obj_ops;
 
 int mdd_la_get(const struct lu_env *env, struct mdd_object *obj,
                struct lu_attr *la, struct lustre_capa *capa)
@@ -91,9 +106,11 @@ struct llog_cookie *mdd_max_cookie_get(const struct lu_env *env,
         }
         if (unlikely(mti->mti_max_cookie == NULL)) {
                 OBD_ALLOC(mti->mti_max_cookie, max_cookie_size);
-                if (unlikely(mti->mti_max_cookie != NULL))
+                if (likely(mti->mti_max_cookie != NULL))
                         mti->mti_max_cookie_size = max_cookie_size;
         }
+        if (likely(mti->mti_max_cookie != NULL))
+                memset(mti->mti_max_cookie, 0, mti->mti_max_cookie_size);
         return mti->mti_max_cookie;
 }
 
@@ -160,18 +177,19 @@ struct lu_object *mdd_object_alloc(const struct lu_env *env,
         }
 }
 
-static int mdd_object_init(const struct lu_env *env, struct lu_object *o)
+static int mdd_object_init(const struct lu_env *env, struct lu_object *o,
+                           const struct lu_object_conf *_)
 {
-       struct mdd_device *d = lu2mdd_dev(o->lo_dev);
-       struct lu_object  *below;
+        struct mdd_device *d = lu2mdd_dev(o->lo_dev);
+        struct lu_object  *below;
         struct lu_device  *under;
         ENTRY;
 
-       under = &d->mdd_child->dd_lu_dev;
-       below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
+        under = &d->mdd_child->dd_lu_dev;
+        below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
         mdd_pdlock_init(lu2mdd_obj(o));
         if (below == NULL)
-               RETURN(-ENOMEM);
+                RETURN(-ENOMEM);
 
         lu_object_add(o, below);
         RETURN(0);
@@ -188,20 +206,13 @@ static int mdd_object_start(const struct lu_env *env, struct lu_object *o)
 static void mdd_object_free(const struct lu_env *env, struct lu_object *o)
 {
         struct mdd_object *mdd = lu2mdd_obj(o);
-       
+
         lu_object_fini(o);
         OBD_FREE_PTR(mdd);
 }
 
-static int mdd_object_print(const struct lu_env *env, void *cookie,
-                            lu_printer_t p, const struct lu_object *o)
-{
-        return (*p)(env, cookie, LUSTRE_MDD_NAME"-object@%p", o);
-}
-
 /* orphan handling is here */
-static void mdd_object_delete(const struct lu_env *env,
-                               struct lu_object *o)
+static void mdd_object_delete(const struct lu_env *env, struct lu_object *o)
 {
         struct mdd_object *mdd_obj = lu2mdd_obj(o);
         struct thandle *handle = NULL;
@@ -217,7 +228,7 @@ static void mdd_object_delete(const struct lu_env *env,
                 if (IS_ERR(handle))
                         CERROR("Cannot get thandle\n");
                 else {
-                        mdd_write_lock(env, mdd_obj);
+                        mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
                         /* let's remove obj from the orphan list */
                         __mdd_orphan_del(env, mdd_obj, handle);
                         mdd_write_unlock(env, mdd_obj);
@@ -227,11 +238,10 @@ static void mdd_object_delete(const struct lu_env *env,
         }
 }
 
-static struct lu_object_operations mdd_lu_obj_ops = {
-       .loo_object_init    = mdd_object_init,
-       .loo_object_start   = mdd_object_start,
-       .loo_object_free    = mdd_object_free,
-       .loo_object_print   = mdd_object_print,
+static const struct lu_object_operations mdd_lu_obj_ops = {
+        .loo_object_init    = mdd_object_init,
+        .loo_object_start   = mdd_object_start,
+        .loo_object_free    = mdd_object_free,
         .loo_object_delete  = mdd_object_delete
 };
 
@@ -239,21 +249,7 @@ struct mdd_object *mdd_object_find(const struct lu_env *env,
                                    struct mdd_device *d,
                                    const struct lu_fid *f)
 {
-        struct lu_object *o, *lo;
-        struct mdd_object *m;
-        ENTRY;
-
-        o = lu_object_find(env, mdd2lu_dev(d)->ld_site, f);
-        if (IS_ERR(o))
-                m = (struct mdd_object *)o;
-        else {
-                lo = lu_object_locate(o->lo_header, mdd2lu_dev(d)->ld_type);
-                /* remote object can't be located and should be put then */
-                if (lo == NULL)
-                        lu_object_put(env, o);
-                m = lu2mdd_obj(lo);
-        }
-        RETURN(m);
+        return md2mdd_obj(md_object_find_slice(env, &d->mdd_md_dev, f));
 }
 
 int mdd_get_flags(const struct lu_env *env, struct mdd_object *obj)
@@ -342,7 +338,7 @@ int mdd_lmm_get_locked(const struct lu_env *env, struct mdd_object *mdd_obj,
         int rc;
         ENTRY;
 
-        mdd_read_lock(env, mdd_obj);
+        mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
         rc = __mdd_lmm_get(env, mdd_obj, ma);
         mdd_read_unlock(env, mdd_obj);
         RETURN(rc);
@@ -389,7 +385,7 @@ static int mdd_attr_get_internal(const struct lu_env *env,
 #ifdef CONFIG_FS_POSIX_ACL
         if (rc == 0 && ma->ma_need & MA_ACL_DEF) {
                 if (S_ISDIR(mdd_object_type(mdd_obj)))
-                        rc = mdd_acl_def_get(env, mdd_obj, ma);
+                        rc = mdd_def_acl_get(env, mdd_obj, ma);
         }
 #endif
         CDEBUG(D_INODE, "after getattr rc = %d, ma_valid = "LPX64"\n",
@@ -404,7 +400,7 @@ int mdd_attr_get_internal_locked(const struct lu_env *env,
         int needlock = ma->ma_need & (MA_LOV | MA_LMV | MA_ACL_DEF);
 
         if (needlock)
-                mdd_read_lock(env, mdd_obj);
+                mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
         rc = mdd_attr_get_internal(env, mdd_obj, ma);
         if (needlock)
                 mdd_read_unlock(env, mdd_obj);
@@ -439,7 +435,7 @@ static int mdd_xattr_get(const struct lu_env *env,
 
         LASSERT(mdd_object_exists(mdd_obj));
 
-        mdd_read_lock(env, mdd_obj);
+        mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
         rc = mdo_xattr_get(env, mdd_obj, buf, name,
                            mdd_object_capa(env, mdd_obj));
         mdd_read_unlock(env, mdd_obj);
@@ -463,7 +459,7 @@ static int mdd_readlink(const struct lu_env *env, struct md_object *obj,
         LASSERT(mdd_object_exists(mdd_obj));
 
         next = mdd_object_child(mdd_obj);
-        mdd_read_lock(env, mdd_obj);
+        mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
         rc = next->do_body_ops->dbo_read(env, next, buf, &pos,
                                          mdd_object_capa(env, mdd_obj));
         mdd_read_unlock(env, mdd_obj);
@@ -481,7 +477,7 @@ static int mdd_xattr_list(const struct lu_env *env, struct md_object *obj,
 
         ENTRY;
 
-        mdd_read_lock(env, mdd_obj);
+        mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
         rc = mdo_xattr_list(env, mdd_obj, buf, mdd_object_capa(env, mdd_obj));
         mdd_read_unlock(env, mdd_obj);
 
@@ -513,6 +509,9 @@ int mdd_object_create_internal(const struct lu_env *env, struct mdd_object *p,
         RETURN(rc);
 }
 
+/**
+ * Make sure the ctime is increased only.
+ */
 static inline int mdd_attr_check(const struct lu_env *env,
                                  struct mdd_object *obj,
                                  struct lu_attr *attr)
@@ -580,12 +579,9 @@ static int mdd_attr_set_internal_locked(const struct lu_env *env,
         ENTRY;
 
         needacl = needacl && (attr->la_valid & LA_MODE);
-
         if (needacl)
-                mdd_write_lock(env, obj);
-
+                mdd_write_lock(env, obj, MOR_TGT_CHILD);
         rc = mdd_attr_set_internal(env, obj, attr, handle, needacl);
-
         if (needacl)
                 mdd_write_unlock(env, obj);
         RETURN(rc);
@@ -600,13 +596,12 @@ int mdd_attr_check_set_internal_locked(const struct lu_env *env,
         int rc;
         ENTRY;
 
-        rc = mdd_attr_check(env, obj, attr);
-        if (rc)
-                RETURN(rc);
-
-        if (attr->la_valid)
-                rc = mdd_attr_set_internal_locked(env, obj, attr, handle,
-                                                  needacl);
+        needacl = needacl && (attr->la_valid & LA_MODE);
+        if (needacl)
+                mdd_write_lock(env, obj, MOR_TGT_CHILD);
+        rc = mdd_attr_check_set_internal(env, obj, attr, handle, needacl);
+        if (needacl)
+                mdd_write_unlock(env, obj);
         RETURN(rc);
 }
 
@@ -656,9 +651,12 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
         if (rc)
                 RETURN(rc);
 
-        /* This is only for set ctime when rename's source is on remote MDS. */
-        if (unlikely(la->la_valid == LA_CTIME)) {
-                rc = mdd_may_delete(env, NULL, obj, (struct md_attr *)ma, 1, 0);
+        if (la->la_valid == LA_CTIME) {
+                if (!(ma->ma_attr_flags & MDS_PERM_BYPASS))
+                        /* This is only for set ctime when rename's source is
+                         * on remote MDS. */
+                        rc = mdd_may_delete(env, NULL, obj,
+                                            (struct md_attr *)ma, 1, 0);
                 if (rc == 0 && la->la_ctime <= tmp_la->la_ctime)
                         la->la_valid &= ~LA_CTIME;
                 RETURN(rc);
@@ -671,7 +669,7 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
                         la->la_valid &= ~LA_ATIME;
                 RETURN(0);
         }
+
         /* Check if flags change. */
         if (la->la_valid & LA_FLAGS) {
                 unsigned int oldflags = 0;
@@ -679,7 +677,7 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
                                 (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL);
 
                 if ((uc->mu_fsuid != tmp_la->la_uid) &&
-                    !mdd_capable(uc, CAP_FOWNER))
+                    !mdd_capable(uc, CFS_CAP_FOWNER))
                         RETURN(-EPERM);
 
                 /* XXX: the IMMUTABLE and APPEND_ONLY flags can
@@ -687,9 +685,9 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
                 if (mdd_is_immutable(obj))
                         oldflags |= LUSTRE_IMMUTABLE_FL;
                 if (mdd_is_append(obj))
-                        oldflags |= LUSTRE_APPEND_FL; 
+                        oldflags |= LUSTRE_APPEND_FL;
                 if ((oldflags ^ newflags) &&
-                    !mdd_capable(uc, CAP_LINUX_IMMUTABLE))
+                    !mdd_capable(uc, CFS_CAP_LINUX_IMMUTABLE))
                         RETURN(-EPERM);
 
                 if (!S_ISDIR(tmp_la->la_mode))
@@ -705,9 +703,10 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
         if ((la->la_valid & (LA_MTIME | LA_ATIME | LA_CTIME)) &&
             !(la->la_valid & ~(LA_MTIME | LA_ATIME | LA_CTIME))) {
                 if ((uc->mu_fsuid != tmp_la->la_uid) &&
-                    !mdd_capable(uc, CAP_FOWNER)) {
+                    !mdd_capable(uc, CFS_CAP_FOWNER)) {
                         rc = mdd_permission_internal_locked(env, obj, tmp_la,
-                                                            MAY_WRITE);
+                                                            MAY_WRITE,
+                                                            MOR_TGT_CHILD);
                         if (rc)
                                 RETURN(rc);
                 }
@@ -715,13 +714,12 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
 
         /* Make sure a caller can chmod. */
         if (la->la_valid & LA_MODE) {
-                /*
-                 * Bypass la_vaild == LA_MODE,
-                 * this is for changing file with SUID or SGID.
-                 */
+                /* Bypass la_vaild == LA_MODE,
+                 * this is for changing file with SUID or SGID. */
                 if ((la->la_valid & ~LA_MODE) &&
+                    !(ma->ma_attr_flags & MDS_PERM_BYPASS) &&
                     (uc->mu_fsuid != tmp_la->la_uid) &&
-                    !mdd_capable(uc, CAP_FOWNER))
+                    !mdd_capable(uc, CFS_CAP_FOWNER))
                         RETURN(-EPERM);
 
                 if (la->la_mode == (umode_t) -1)
@@ -731,8 +729,9 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
                                       (tmp_la->la_mode & ~S_IALLUGO);
 
                 /* Also check the setgid bit! */
-                if (!lustre_in_group_p(uc, (la->la_valid & LA_GID) ? la->la_gid :
-                                tmp_la->la_gid) && !mdd_capable(uc, CAP_FSETID))
+                if (!lustre_in_group_p(uc, (la->la_valid & LA_GID) ?
+                                       la->la_gid : tmp_la->la_gid) &&
+                    !mdd_capable(uc, CFS_CAP_FSETID))
                         la->la_mode &= ~S_ISGID;
         } else {
                la->la_mode = tmp_la->la_mode;
@@ -744,11 +743,10 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
                         la->la_uid = tmp_la->la_uid;
                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
                     (la->la_uid != tmp_la->la_uid)) &&
-                    !mdd_capable(uc, CAP_CHOWN))
+                    !mdd_capable(uc, CFS_CAP_CHOWN))
                         RETURN(-EPERM);
 
-                /*
-                 * If the user or group of a non-directory has been
+                /* If the user or group of a non-directory has been
                  * changed by a non-root user, remove the setuid bit.
                  * 19981026 David C Niemi <niemi@tux.org>
                  *
@@ -756,8 +754,7 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
                  * to avoid some races. This is the behavior we had in
                  * 2.0. The check for non-root was definitely wrong
                  * for 2.2 anyway, as it should have been using
-                 * CAP_FSETID rather than fsuid -- 19990830 SD.
-                 */
+                 * CAP_FSETID rather than fsuid -- 19990830 SD. */
                 if (((tmp_la->la_mode & S_ISUID) == S_ISUID) &&
                     !S_ISDIR(tmp_la->la_mode)) {
                         la->la_mode &= ~S_ISUID;
@@ -772,19 +769,17 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
                     ((la->la_gid != tmp_la->la_gid) &&
                     !lustre_in_group_p(uc, la->la_gid))) &&
-                    !mdd_capable(uc, CAP_CHOWN))
+                    !mdd_capable(uc, CFS_CAP_CHOWN))
                         RETURN(-EPERM);
 
-                /*
-                 * Likewise, if the user or group of a non-directory
+                /* Likewise, if the user or group of a non-directory
                  * has been changed by a non-root user, remove the
                  * setgid bit UNLESS there is no group execute bit
                  * (this would be a file marked for mandatory
                  * locking).  19981026 David C Niemi <niemi@tux.org>
                  *
                  * Removed the fsuid check (see the comment above) --
-                 * 19990830 SD.
-                 */
+                 * 19990830 SD. */
                 if (((tmp_la->la_mode & (S_ISGID | S_IXGRP)) ==
                      (S_ISGID | S_IXGRP)) && !S_ISDIR(tmp_la->la_mode)) {
                         la->la_mode &= ~S_ISGID;
@@ -792,11 +787,13 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
                 }
         }
 
-        if (la->la_valid & (LA_SIZE | LA_BLOCKS) &&
-            !(la->la_valid & LA_TRUNC)) {
-                /* For "Size-on-MDS" case, the MAY_WRITE perm
-                 * has been checked when file open. */
-
+        /* For both Size-on-MDS case and truncate case,
+         * "la->la_valid & (LA_SIZE | LA_BLOCKS)" are ture.
+         * We distinguish them by "ma->ma_attr_flags & MDS_SOM".
+         * For SOM case, it is true, the MAY_WRITE perm has been checked
+         * when open, no need check again. For truncate case, it is false,
+         * the MAY_WRITE perm should be checked here. */
+        if (ma->ma_attr_flags & MDS_SOM) {
                 /* For the "Size-on-MDS" setattr update, merge coming
                  * attributes with the set in the inode. BUG 10641 */
                 if ((la->la_valid & LA_ATIME) &&
@@ -809,13 +806,13 @@ static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
                     (la->la_ctime <= tmp_la->la_ctime))
                         la->la_valid &= ~(LA_MTIME | LA_CTIME);
         } else {
-                if (la->la_valid & LA_TRUNC) {
-                        /* For truncate, we should have MAY_WRITE perm. */
-                        if (!((la->la_valid & MDS_OPEN_OWNEROVERRIDE) &&
+                if (la->la_valid & (LA_SIZE | LA_BLOCKS)) {
+                        if (!((ma->ma_attr_flags & MDS_OPEN_OWNEROVERRIDE) &&
                               (uc->mu_fsuid == tmp_la->la_uid)) &&
                             !(ma->ma_attr_flags & MDS_PERM_BYPASS)) {
                                 rc = mdd_permission_internal_locked(env, obj,
-                                                            tmp_la, MAY_WRITE);
+                                                            tmp_la, MAY_WRITE,
+                                                            MOR_TGT_CHILD);
                                 if (rc)
                                         RETURN(rc);
                         }
@@ -926,7 +923,7 @@ int mdd_xattr_set_txn(const struct lu_env *env, struct mdd_object *obj,
         int  rc;
         ENTRY;
 
-        mdd_write_lock(env, obj);
+        mdd_write_lock(env, obj, MOR_TGT_CHILD);
         rc = __mdd_xattr_set(env, obj, buf, name, fl, handle);
         mdd_write_unlock(env, obj);
 
@@ -948,17 +945,21 @@ static int mdd_xattr_sanity_check(const struct lu_env *env,
         if (rc)
                 RETURN(rc);
 
-        if ((uc->mu_fsuid != tmp_la->la_uid) && !mdd_capable(uc, CAP_FOWNER))
+        if ((uc->mu_fsuid != tmp_la->la_uid) &&
+            !mdd_capable(uc, CFS_CAP_FOWNER))
                 RETURN(-EPERM);
 
         RETURN(rc);
 }
 
+/**
+ * The caller should guarantee to update the object ctime
+ * after xattr_set if needed.
+ */
 static int mdd_xattr_set(const struct lu_env *env, struct md_object *obj,
                          const struct lu_buf *buf, const char *name,
-                         int fl, const struct lu_attr *la)
+                         int fl)
 {
-        struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
         struct mdd_object *mdd_obj = md2mdd_obj(obj);
         struct mdd_device *mdd = mdo2mdd(obj);
         struct thandle *handle;
@@ -975,22 +976,18 @@ static int mdd_xattr_set(const struct lu_env *env, struct md_object *obj,
                 RETURN(PTR_ERR(handle));
 
         rc = mdd_xattr_set_txn(env, mdd_obj, buf, name, fl, handle);
-        if (rc == 0 && likely(la != NULL)) {
-                LASSERT(la->la_valid & LA_CTIME);
-                la_copy->la_ctime = la->la_ctime;
-                la_copy->la_valid = LA_CTIME;
-                rc = mdd_attr_check_set_internal_locked(env, mdd_obj, la_copy,
-                                                        handle, 0);
-        }
         mdd_trans_stop(env, mdd, rc, handle);
 
         RETURN(rc);
 }
 
+/**
+ * The caller should guarantee to update the object ctime
+ * after xattr_set if needed.
+ */
 int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
-                  const char *name, const struct lu_attr *la)
+                  const char *name)
 {
-        struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
         struct mdd_object *mdd_obj = md2mdd_obj(obj);
         struct mdd_device *mdd = mdo2mdd(obj);
         struct thandle *handle;
@@ -1006,18 +1003,10 @@ int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
         if (IS_ERR(handle))
                 RETURN(PTR_ERR(handle));
 
-        mdd_write_lock(env, mdd_obj);
+        mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
         rc = mdo_xattr_del(env, mdd_obj, name, handle,
                            mdd_object_capa(env, mdd_obj));
         mdd_write_unlock(env, mdd_obj);
-        if (rc == 0 && likely(la != NULL)) {
-                LASSERT(la->la_valid & LA_CTIME);
-                la_copy->la_ctime = la->la_ctime;
-                la_copy->la_valid = LA_CTIME;
-                rc = mdd_attr_check_set_internal_locked(env, mdd_obj, la_copy,
-                                                        handle, 0);
-        }
-
         mdd_trans_stop(env, mdd, rc, handle);
 
         RETURN(rc);
@@ -1051,7 +1040,7 @@ static int mdd_ref_del(const struct lu_env *env, struct md_object *obj,
         if (IS_ERR(handle))
                 RETURN(-ENOMEM);
 
-        mdd_write_lock(env, mdd_obj);
+        mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
 
         rc = mdd_unlink_sanity_check(env, NULL, mdd_obj, ma);
         if (rc)
@@ -1124,7 +1113,7 @@ static int mdd_object_create(const struct lu_env *env,
         if (IS_ERR(handle))
                 RETURN(PTR_ERR(handle));
 
-        mdd_write_lock(env, mdd_obj);
+        mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
         rc = mdd_oc_sanity_check(env, mdd_obj, ma);
         if (rc)
                 GOTO(unlock, rc);
@@ -1173,9 +1162,9 @@ static int mdd_object_create(const struct lu_env *env,
         }
         EXIT;
 unlock:
-        mdd_write_unlock(env, mdd_obj);
         if (rc == 0)
-                rc = mdd_attr_get_internal_locked(env, mdd_obj, ma);
+                rc = mdd_attr_get_internal(env, mdd_obj, ma);
+        mdd_write_unlock(env, mdd_obj);
 
         mdd_trans_stop(env, mdd, rc, handle);
         return rc;
@@ -1197,7 +1186,7 @@ static int mdd_ref_add(const struct lu_env *env, struct md_object *obj,
         if (IS_ERR(handle))
                 RETURN(-ENOMEM);
 
-        mdd_write_lock(env, mdd_obj);
+        mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
         rc = mdd_link_sanity_check(env, NULL, NULL, mdd_obj);
         if (rc == 0)
                 __mdd_ref_add(env, mdd_obj, handle);
@@ -1295,7 +1284,7 @@ static int mdd_open_sanity_check(const struct lu_env *env,
                 if (uc && ((uc->mu_valid == UCRED_OLD) ||
                     (uc->mu_valid == UCRED_NEW)) &&
                     (uc->mu_fsuid != tmp_la->la_uid) &&
-                    !mdd_capable(uc, CAP_FOWNER))
+                    !mdd_capable(uc, CFS_CAP_FOWNER))
                         RETURN(-EPERM);
         }
 #endif
@@ -1309,7 +1298,7 @@ static int mdd_open(const struct lu_env *env, struct md_object *obj,
         struct mdd_object *mdd_obj = md2mdd_obj(obj);
         int rc = 0;
 
-        mdd_write_lock(env, mdd_obj);
+        mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
 
         rc = mdd_open_sanity_check(env, mdd_obj, flags);
         if (rc == 0)
@@ -1356,7 +1345,7 @@ static int mdd_close(const struct lu_env *env, struct md_object *obj,
         if (IS_ERR(handle))
                 RETURN(PTR_ERR(handle));
 
-        mdd_write_lock(env, mdd_obj);
+        mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
         /* release open count */
         mdd_obj->mod_count --;
 
@@ -1365,7 +1354,7 @@ static int mdd_close(const struct lu_env *env, struct md_object *obj,
                 rc = mdd_object_kill(env, mdd_obj, ma);
         else
                 ma->ma_valid &= ~(MA_LOV | MA_COOKIE);
-        
+
         mdd_write_unlock(env, mdd_obj);
         mdd_trans_stop(env, mdo2mdd(obj), rc, handle);
         RETURN(rc);
@@ -1391,7 +1380,7 @@ static int mdd_readpage_sanity_check(const struct lu_env *env,
 }
 
 static int mdd_dir_page_build(const struct lu_env *env, int first,
-                              void *area, int nob, struct dt_it_ops *iops,
+                              void *area, int nob, const struct dt_it_ops *iops,
                               struct dt_it *it, __u64 *start, __u64 *end,
                               struct lu_dirent **last)
 {
@@ -1465,14 +1454,14 @@ static int __mdd_readpage(const struct lu_env *env, struct mdd_object *obj,
 {
         struct dt_it      *it;
         struct dt_object  *next = mdd_object_child(obj);
-        struct dt_it_ops  *iops;
+        const struct dt_it_ops  *iops;
         struct page       *pg;
-        struct lu_dirent  *last;
+        struct lu_dirent  *last = NULL;
         int i;
         int rc;
         int nob;
         __u64 hash_start;
-        __u64 hash_end;
+        __u64 hash_end = 0;
 
         LASSERT(rdpg->rp_pages != NULL);
         LASSERT(next->do_index_ops != NULL);
@@ -1560,7 +1549,7 @@ static int mdd_readpage(const struct lu_env *env, struct md_object *obj,
 
         LASSERT(mdd_object_exists(mdd_obj));
 
-        mdd_read_lock(env, mdd_obj);
+        mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
         rc = mdd_readpage_sanity_check(env, mdd_obj);
         if (rc)
                 GOTO(out_unlock, rc);
@@ -1599,7 +1588,17 @@ out_unlock:
         return rc;
 }
 
-struct md_object_operations mdd_obj_ops = {
+static int mdd_object_sync(const struct lu_env *env, struct md_object *obj)
+{
+        struct mdd_object *mdd_obj = md2mdd_obj(obj);
+        struct dt_object *next;
+
+        LASSERT(mdd_object_exists(mdd_obj));
+        next = mdd_object_child(mdd_obj);
+        return next->do_ops->do_object_sync(env, next);
+}
+
+const struct md_object_operations mdd_obj_ops = {
         .moo_permission    = mdd_permission,
         .moo_attr_get      = mdd_attr_get,
         .moo_attr_set      = mdd_attr_set,
@@ -1614,5 +1613,6 @@ struct md_object_operations mdd_obj_ops = {
         .moo_close         = mdd_close,
         .moo_readpage      = mdd_readpage,
         .moo_readlink      = mdd_readlink,
-        .moo_capa_get      = mdd_capa_get
+        .moo_capa_get      = mdd_capa_get,
+        .moo_object_sync   = mdd_object_sync,
 };