lustre/mdd/mdd_object.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  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 #ifndef EXPORT_SYMTAB
  44 # define EXPORT_SYMTAB
  45 #endif
  46 #define DEBUG_SUBSYSTEM S_MDS
  47
  48 #include <linux/module.h>
  49 #ifdef HAVE_EXT4_LDISKFS
  50 #include <ldiskfs/ldiskfs_jbd2.h>
  51 #else
  52 #include <linux/jbd.h>
  53 #endif
  54 #include <obd.h>
  55 #include <obd_class.h>
  56 #include <obd_support.h>
  57 #include <lprocfs_status.h>
  58 /* fid_be_cpu(), fid_cpu_to_be(). */
  59 #include <lustre_fid.h>
  60
  61 #include <lustre_param.h>
  62 #ifdef HAVE_EXT4_LDISKFS
  63 #include <ldiskfs/ldiskfs.h>
  64 #else
  65 #include <linux/ldiskfs_fs.h>
  66 #endif
  67 #include <lustre_mds.h>
  68 #include <lustre/lustre_idl.h>
  69
  70 #include "mdd_internal.h"
  71
  72 static const struct lu_object_operations mdd_lu_obj_ops;
  73
  74 static int mdd_xattr_get(const struct lu_env *env,
  75                          struct md_object *obj, struct lu_buf *buf,
  76                          const char *name);
  77
  78 int mdd_data_get(const struct lu_env *env, struct mdd_object *obj,
  79                  void **data)
  80 {
  81         LASSERTF(mdd_object_exists(obj), "FID is "DFID"\n",
  82                  PFID(mdd_object_fid(obj)));
  83         mdo_data_get(env, obj, data);
  84         return 0;
  85 }
  86
  87 int mdd_la_get(const struct lu_env *env, struct mdd_object *obj,
  88                struct lu_attr *la, struct lustre_capa *capa)
  89 {
  90         LASSERTF(mdd_object_exists(obj), "FID is "DFID"\n",
  91                  PFID(mdd_object_fid(obj)));
  92         return mdo_attr_get(env, obj, la, capa);
  93 }
  94
  95 static void mdd_flags_xlate(struct mdd_object *obj, __u32 flags)
  96 {
  97         obj->mod_flags &= ~(APPEND_OBJ|IMMUTE_OBJ);
  98
  99         if (flags & LUSTRE_APPEND_FL)
 100                 obj->mod_flags |= APPEND_OBJ;
 101
 102         if (flags & LUSTRE_IMMUTABLE_FL)
 103                 obj->mod_flags |= IMMUTE_OBJ;
 104 }
 105
 106 struct mdd_thread_info *mdd_env_info(const struct lu_env *env)
 107 {
 108         struct mdd_thread_info *info;
 109
 110         info = lu_context_key_get(&env->le_ctx, &mdd_thread_key);
 111         LASSERT(info != NULL);
 112         return info;
 113 }
 114
 115 struct lu_buf *mdd_buf_get(const struct lu_env *env, void *area, ssize_t len)
 116 {
 117         struct lu_buf *buf;
 118
 119         buf = &mdd_env_info(env)->mti_buf;
 120         buf->lb_buf = area;
 121         buf->lb_len = len;
 122         return buf;
 123 }
 124
 125 void mdd_buf_put(struct lu_buf *buf)
 126 {
 127         if (buf == NULL || buf->lb_buf == NULL)
 128                 return;
 129         if (buf->lb_vmalloc)
 130                 OBD_VFREE(buf->lb_buf, buf->lb_len);
 131         else
 132                 OBD_FREE(buf->lb_buf, buf->lb_len);
 133         buf->lb_buf = NULL;
 134         buf->lb_len = 0;
 135 }
 136
 137 const struct lu_buf *mdd_buf_get_const(const struct lu_env *env,
 138                                        const void *area, ssize_t len)
 139 {
 140         struct lu_buf *buf;
 141
 142         buf = &mdd_env_info(env)->mti_buf;
 143         buf->lb_buf = (void *)area;
 144         buf->lb_len = len;
 145         return buf;
 146 }
 147
 148 #define BUF_VMALLOC_SIZE (CFS_PAGE_SIZE<<2) /* 16k */
 149 struct lu_buf *mdd_buf_alloc(const struct lu_env *env, ssize_t len)
 150 {
 151         struct lu_buf *buf = &mdd_env_info(env)->mti_big_buf;
 152
 153         if ((len > buf->lb_len) && (buf->lb_buf != NULL)) {
 154                 if (buf->lb_vmalloc)
 155                         OBD_VFREE(buf->lb_buf, buf->lb_len);
 156                 else
 157                         OBD_FREE(buf->lb_buf, buf->lb_len);
 158                 buf->lb_buf = NULL;
 159         }
 160         if (buf->lb_buf == NULL) {
 161                 buf->lb_len = len;
 162                 if (buf->lb_len <= BUF_VMALLOC_SIZE) {
 163                         OBD_ALLOC(buf->lb_buf, buf->lb_len);
 164                         buf->lb_vmalloc = 0;
 165                 }
 166                 if (buf->lb_buf == NULL) {
 167                         OBD_VMALLOC(buf->lb_buf, buf->lb_len);
 168                         buf->lb_vmalloc = 1;
 169                 }
 170                 if (buf->lb_buf == NULL)
 171                         buf->lb_len = 0;
 172         }
 173         return buf;
 174 }
 175
 176 /** Increase the size of the \a mti_big_buf.
 177  * preserves old data in buffer
 178  * old buffer remains unchanged on error
 179  * \retval 0 or -ENOMEM
 180  */
 181 int mdd_buf_grow(const struct lu_env *env, ssize_t len)
 182 {
 183         struct lu_buf *oldbuf = &mdd_env_info(env)->mti_big_buf;
 184         struct lu_buf buf;
 185
 186         LASSERT(len >= oldbuf->lb_len);
 187         if (len > BUF_VMALLOC_SIZE) {
 188                 OBD_VMALLOC(buf.lb_buf, len);
 189                 buf.lb_vmalloc = 1;
 190         } else {
 191                 OBD_ALLOC(buf.lb_buf, len);
 192                 buf.lb_vmalloc = 0;
 193         }
 194         if (buf.lb_buf == NULL)
 195                 return -ENOMEM;
 196
 197         buf.lb_len = len;
 198         memcpy(buf.lb_buf, oldbuf->lb_buf, oldbuf->lb_len);
 199
 200         if (oldbuf->lb_vmalloc)
 201                 OBD_VFREE(oldbuf->lb_buf, oldbuf->lb_len);
 202         else
 203                 OBD_FREE(oldbuf->lb_buf, oldbuf->lb_len);
 204
 205         memcpy(oldbuf, &buf, sizeof(buf));
 206
 207         return 0;
 208 }
 209
 210 struct llog_cookie *mdd_max_cookie_get(const struct lu_env *env,
 211                                        struct mdd_device *mdd)
 212 {
 213         struct mdd_thread_info *mti = mdd_env_info(env);
 214         int                     max_cookie_size;
 215
 216         max_cookie_size = mdd_lov_cookiesize(env, mdd);
 217         if (unlikely(mti->mti_max_cookie_size < max_cookie_size)) {
 218                 if (mti->mti_max_cookie)
 219                         OBD_FREE(mti->mti_max_cookie, mti->mti_max_cookie_size);
 220                 mti->mti_max_cookie = NULL;
 221                 mti->mti_max_cookie_size = 0;
 222         }
 223         if (unlikely(mti->mti_max_cookie == NULL)) {
 224                 OBD_ALLOC(mti->mti_max_cookie, max_cookie_size);
 225                 if (likely(mti->mti_max_cookie != NULL))
 226                         mti->mti_max_cookie_size = max_cookie_size;
 227         }
 228         if (likely(mti->mti_max_cookie != NULL))
 229                 memset(mti->mti_max_cookie, 0, mti->mti_max_cookie_size);
 230         return mti->mti_max_cookie;
 231 }
 232
 233 struct lov_mds_md *mdd_max_lmm_get(const struct lu_env *env,
 234                                    struct mdd_device *mdd)
 235 {
 236         struct mdd_thread_info *mti = mdd_env_info(env);
 237         int                     max_lmm_size;
 238
 239         max_lmm_size = mdd_lov_mdsize(env, mdd);
 240         if (unlikely(mti->mti_max_lmm_size < max_lmm_size)) {
 241                 if (mti->mti_max_lmm)
 242                         OBD_FREE(mti->mti_max_lmm, mti->mti_max_lmm_size);
 243                 mti->mti_max_lmm = NULL;
 244                 mti->mti_max_lmm_size = 0;
 245         }
 246         if (unlikely(mti->mti_max_lmm == NULL)) {
 247                 OBD_ALLOC(mti->mti_max_lmm, max_lmm_size);
 248                 if (unlikely(mti->mti_max_lmm != NULL))
 249                         mti->mti_max_lmm_size = max_lmm_size;
 250         }
 251         return mti->mti_max_lmm;
 252 }
 253
 254 struct lu_object *mdd_object_alloc(const struct lu_env *env,
 255                                    const struct lu_object_header *hdr,
 256                                    struct lu_device *d)
 257 {
 258         struct mdd_object *mdd_obj;
 259
 260         OBD_ALLOC_PTR(mdd_obj);
 261         if (mdd_obj != NULL) {
 262                 struct lu_object *o;
 263
 264                 o = mdd2lu_obj(mdd_obj);
 265                 lu_object_init(o, NULL, d);
 266                 mdd_obj->mod_obj.mo_ops = &mdd_obj_ops;
 267                 mdd_obj->mod_obj.mo_dir_ops = &mdd_dir_ops;
 268                 mdd_obj->mod_count = 0;
 269                 o->lo_ops = &mdd_lu_obj_ops;
 270                 return o;
 271         } else {
 272                 return NULL;
 273         }
 274 }
 275
 276 static int mdd_object_init(const struct lu_env *env, struct lu_object *o,
 277                            const struct lu_object_conf *unused)
 278 {
 279         struct mdd_device *d = lu2mdd_dev(o->lo_dev);
 280         struct mdd_object *mdd_obj = lu2mdd_obj(o);
 281         struct lu_object  *below;
 282         struct lu_device  *under;
 283         ENTRY;
 284
 285         mdd_obj->mod_cltime = 0;
 286         under = &d->mdd_child->dd_lu_dev;
 287         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
 288         mdd_pdlock_init(mdd_obj);
 289         if (below == NULL)
 290                 RETURN(-ENOMEM);
 291
 292         lu_object_add(o, below);
 293
 294         RETURN(0);
 295 }
 296
 297 static int mdd_object_start(const struct lu_env *env, struct lu_object *o)
 298 {
 299         if (lu_object_exists(o))
 300                 return mdd_get_flags(env, lu2mdd_obj(o));
 301         else
 302                 return 0;
 303 }
 304
 305 static void mdd_object_free(const struct lu_env *env, struct lu_object *o)
 306 {
 307         struct mdd_object *mdd = lu2mdd_obj(o);
 308
 309         lu_object_fini(o);
 310         OBD_FREE_PTR(mdd);
 311 }
 312
 313 static int mdd_object_print(const struct lu_env *env, void *cookie,
 314                             lu_printer_t p, const struct lu_object *o)
 315 {
 316         struct mdd_object *mdd = lu2mdd_obj((struct lu_object *)o);
 317         return (*p)(env, cookie, LUSTRE_MDD_NAME"-object@%p(open_count=%d, "
 318                     "valid=%x, cltime="LPU64", flags=%lx)",
 319                     mdd, mdd->mod_count, mdd->mod_valid,
 320                     mdd->mod_cltime, mdd->mod_flags);
 321 }
 322
 323 static const struct lu_object_operations mdd_lu_obj_ops = {
 324         .loo_object_init    = mdd_object_init,
 325         .loo_object_start   = mdd_object_start,
 326         .loo_object_free    = mdd_object_free,
 327         .loo_object_print   = mdd_object_print,
 328 };
 329
 330 struct mdd_object *mdd_object_find(const struct lu_env *env,
 331                                    struct mdd_device *d,
 332                                    const struct lu_fid *f)
 333 {
 334         return md2mdd_obj(md_object_find_slice(env, &d->mdd_md_dev, f));
 335 }
 336
 337 static int mdd_path2fid(const struct lu_env *env, struct mdd_device *mdd,
 338                         const char *path, struct lu_fid *fid)
 339 {
 340         struct lu_buf *buf;
 341         struct lu_fid *f = &mdd_env_info(env)->mti_fid;
 342         struct mdd_object *obj;
 343         struct lu_name *lname = &mdd_env_info(env)->mti_name;
 344         char *name;
 345         int rc = 0;
 346         ENTRY;
 347
 348         /* temp buffer for path element */
 349         buf = mdd_buf_alloc(env, PATH_MAX);
 350         if (buf->lb_buf == NULL)
 351                 RETURN(-ENOMEM);
 352
 353         lname->ln_name = name = buf->lb_buf;
 354         lname->ln_namelen = 0;
 355         *f = mdd->mdd_root_fid;
 356
 357         while(1) {
 358                 while (*path == '/')
 359                         path++;
 360                 if (*path == '\0')
 361                         break;
 362                 while (*path != '/' && *path != '\0') {
 363                         *name = *path;
 364                         path++;
 365                         name++;
 366                         lname->ln_namelen++;
 367                 }
 368
 369                 *name = '\0';
 370                 /* find obj corresponding to fid */
 371                 obj = mdd_object_find(env, mdd, f);
 372                 if (obj == NULL)
 373                         GOTO(out, rc = -EREMOTE);
 374                 if (IS_ERR(obj))
 375                         GOTO(out, rc = -PTR_ERR(obj));
 376                 /* get child fid from parent and name */
 377                 rc = mdd_lookup(env, &obj->mod_obj, lname, f, NULL);
 378                 mdd_object_put(env, obj);
 379                 if (rc)
 380                         break;
 381
 382                 name = buf->lb_buf;
 383                 lname->ln_namelen = 0;
 384         }
 385
 386         if (!rc)
 387                 *fid = *f;
 388 out:
 389         RETURN(rc);
 390 }
 391
 392 /** The maximum depth that fid2path() will search.
 393  * This is limited only because we want to store the fids for
 394  * historical path lookup purposes.
 395  */
 396 #define MAX_PATH_DEPTH 100
 397
 398 /** mdd_path() lookup structure. */
 399 struct path_lookup_info {
 400         __u64                pli_recno;        /**< history point */
 401         __u64                pli_currec;       /**< current record */
 402         struct lu_fid        pli_fid;
 403         struct lu_fid        pli_fids[MAX_PATH_DEPTH]; /**< path, in fids */
 404         struct mdd_object   *pli_mdd_obj;
 405         char                *pli_path;         /**< full path */
 406         int                  pli_pathlen;
 407         int                  pli_linkno;       /**< which hardlink to follow */
 408         int                  pli_fidcount;     /**< number of \a pli_fids */
 409 };
 410
 411 static int mdd_path_current(const struct lu_env *env,
 412                             struct path_lookup_info *pli)
 413 {
 414         struct mdd_device *mdd = mdo2mdd(&pli->pli_mdd_obj->mod_obj);
 415         struct mdd_object *mdd_obj;
 416         struct lu_buf     *buf = NULL;
 417         struct link_ea_header *leh;
 418         struct link_ea_entry  *lee;
 419         struct lu_name *tmpname = &mdd_env_info(env)->mti_name;
 420         struct lu_fid  *tmpfid = &mdd_env_info(env)->mti_fid;
 421         char *ptr;
 422         int reclen;
 423         int rc;
 424         ENTRY;
 425
 426         ptr = pli->pli_path + pli->pli_pathlen - 1;
 427         *ptr = 0;
 428         --ptr;
 429         pli->pli_fidcount = 0;
 430         pli->pli_fids[0] = *(struct lu_fid *)mdd_object_fid(pli->pli_mdd_obj);
 431
 432         while (!mdd_is_root(mdd, &pli->pli_fids[pli->pli_fidcount])) {
 433                 mdd_obj = mdd_object_find(env, mdd,
 434                                           &pli->pli_fids[pli->pli_fidcount]);
 435                 if (mdd_obj == NULL)
 436                         GOTO(out, rc = -EREMOTE);
 437                 if (IS_ERR(mdd_obj))
 438                         GOTO(out, rc = -PTR_ERR(mdd_obj));
 439                 rc = lu_object_exists(&mdd_obj->mod_obj.mo_lu);
 440                 if (rc <= 0) {
 441                         mdd_object_put(env, mdd_obj);
 442                         if (rc == -1)
 443                                 rc = -EREMOTE;
 444                         else if (rc == 0)
 445                                 /* Do I need to error out here? */
 446                                 rc = -ENOENT;
 447                         GOTO(out, rc);
 448                 }
 449
 450                 /* Get parent fid and object name */
 451                 mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 452                 buf = mdd_links_get(env, mdd_obj);
 453                 mdd_read_unlock(env, mdd_obj);
 454                 mdd_object_put(env, mdd_obj);
 455                 if (IS_ERR(buf))
 456                         GOTO(out, rc = PTR_ERR(buf));
 457
 458                 leh = buf->lb_buf;
 459                 lee = (struct link_ea_entry *)(leh + 1); /* link #0 */
 460                 mdd_lee_unpack(lee, &reclen, tmpname, tmpfid);
 461
 462                 /* If set, use link #linkno for path lookup, otherwise use
 463                    link #0.  Only do this for the final path element. */
 464                 if ((pli->pli_fidcount == 0) &&
 465                     (pli->pli_linkno < leh->leh_reccount)) {
 466                         int count;
 467                         for (count = 0; count < pli->pli_linkno; count++) {
 468                                 lee = (struct link_ea_entry *)
 469                                      ((char *)lee + reclen);
 470                                 mdd_lee_unpack(lee, &reclen, tmpname, tmpfid);
 471                         }
 472                         if (pli->pli_linkno < leh->leh_reccount - 1)
 473                                 /* indicate to user there are more links */
 474                                 pli->pli_linkno++;
 475                 }
 476
 477                 /* Pack the name in the end of the buffer */
 478                 ptr -= tmpname->ln_namelen;
 479                 if (ptr - 1 <= pli->pli_path)
 480                         GOTO(out, rc = -EOVERFLOW);
 481                 strncpy(ptr, tmpname->ln_name, tmpname->ln_namelen);
 482                 *(--ptr) = '/';
 483
 484                 /* Store the parent fid for historic lookup */
 485                 if (++pli->pli_fidcount >= MAX_PATH_DEPTH)
 486                         GOTO(out, rc = -EOVERFLOW);
 487                 pli->pli_fids[pli->pli_fidcount] = *tmpfid;
 488         }
 489
 490         /* Verify that our path hasn't changed since we started the lookup.
 491            Record the current index, and verify the path resolves to the
 492            same fid. If it does, then the path is correct as of this index. */
 493         cfs_spin_lock(&mdd->mdd_cl.mc_lock);
 494         pli->pli_currec = mdd->mdd_cl.mc_index;
 495         cfs_spin_unlock(&mdd->mdd_cl.mc_lock);
 496         rc = mdd_path2fid(env, mdd, ptr, &pli->pli_fid);
 497         if (rc) {
 498                 CDEBUG(D_INFO, "mdd_path2fid(%s) failed %d\n", ptr, rc);
 499                 GOTO (out, rc = -EAGAIN);
 500         }
 501         if (!lu_fid_eq(&pli->pli_fids[0], &pli->pli_fid)) {
 502                 CDEBUG(D_INFO, "mdd_path2fid(%s) found another FID o="DFID
 503                        " n="DFID"\n", ptr, PFID(&pli->pli_fids[0]),
 504                        PFID(&pli->pli_fid));
 505                 GOTO(out, rc = -EAGAIN);
 506         }
 507         ptr++; /* skip leading / */
 508         memmove(pli->pli_path, ptr, pli->pli_path + pli->pli_pathlen - ptr);
 509
 510         EXIT;
 511 out:
 512         if (buf && !IS_ERR(buf) && buf->lb_vmalloc)
 513                 /* if we vmalloced a large buffer drop it */
 514                 mdd_buf_put(buf);
 515
 516         return rc;
 517 }
 518
 519 static int mdd_path_historic(const struct lu_env *env,
 520                              struct path_lookup_info *pli)
 521 {
 522         return 0;
 523 }
 524
 525 /* Returns the full path to this fid, as of changelog record recno. */
 526 static int mdd_path(const struct lu_env *env, struct md_object *obj,
 527                     char *path, int pathlen, __u64 *recno, int *linkno)
 528 {
 529         struct path_lookup_info *pli;
 530         int tries = 3;
 531         int rc = -EAGAIN;
 532         ENTRY;
 533
 534         if (pathlen < 3)
 535                 RETURN(-EOVERFLOW);
 536
 537         if (mdd_is_root(mdo2mdd(obj), mdd_object_fid(md2mdd_obj(obj)))) {
 538                 path[0] = '\0';
 539                 RETURN(0);
 540         }
 541
 542         OBD_ALLOC_PTR(pli);
 543         if (pli == NULL)
 544                 RETURN(-ENOMEM);
 545
 546         pli->pli_mdd_obj = md2mdd_obj(obj);
 547         pli->pli_recno = *recno;
 548         pli->pli_path = path;
 549         pli->pli_pathlen = pathlen;
 550         pli->pli_linkno = *linkno;
 551
 552         /* Retry multiple times in case file is being moved */
 553         while (tries-- && rc == -EAGAIN)
 554                 rc = mdd_path_current(env, pli);
 555
 556         /* For historical path lookup, the current links may not have existed
 557          * at "recno" time.  We must switch over to earlier links/parents
 558          * by using the changelog records.  If the earlier parent doesn't
 559          * exist, we must search back through the changelog to reconstruct
 560          * its parents, then check if it exists, etc.
 561          * We may ignore this problem for the initial implementation and
 562          * state that an "original" hardlink must still exist for us to find
 563          * historic path name. */
 564         if (pli->pli_recno != -1) {
 565                 rc = mdd_path_historic(env, pli);
 566         } else {
 567                 *recno = pli->pli_currec;
 568                 /* Return next link index to caller */
 569                 *linkno = pli->pli_linkno;
 570         }
 571
 572         OBD_FREE_PTR(pli);
 573
 574         RETURN (rc);
 575 }
 576
 577 int mdd_get_flags(const struct lu_env *env, struct mdd_object *obj)
 578 {
 579         struct lu_attr *la = &mdd_env_info(env)->mti_la;
 580         int rc;
 581
 582         ENTRY;
 583         rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
 584         if (rc == 0) {
 585                 mdd_flags_xlate(obj, la->la_flags);
 586                 if (S_ISDIR(la->la_mode) && la->la_nlink == 1)
 587                         obj->mod_flags |= MNLINK_OBJ;
 588         }
 589         RETURN(rc);
 590 }
 591
 592 /* get only inode attributes */
 593 int mdd_iattr_get(const struct lu_env *env, struct mdd_object *mdd_obj,
 594                   struct md_attr *ma)
 595 {
 596         int rc = 0;
 597         ENTRY;
 598
 599         if (ma->ma_valid & MA_INODE)
 600                 RETURN(0);
 601
 602         rc = mdd_la_get(env, mdd_obj, &ma->ma_attr,
 603                           mdd_object_capa(env, mdd_obj));
 604         if (rc == 0)
 605                 ma->ma_valid |= MA_INODE;
 606         RETURN(rc);
 607 }
 608
 609 int mdd_get_default_md(struct mdd_object *mdd_obj, struct lov_mds_md *lmm)
 610 {
 611         struct lov_desc *ldesc;
 612         struct mdd_device *mdd = mdo2mdd(&mdd_obj->mod_obj);
 613         struct lov_user_md *lum = (struct lov_user_md*)lmm;
 614         ENTRY;
 615
 616         if (!lum)
 617                 RETURN(0);
 618
 619         ldesc = &mdd->mdd_obd_dev->u.mds.mds_lov_desc;
 620         LASSERT(ldesc != NULL);
 621
 622         lum->lmm_magic = LOV_MAGIC_V1;
 623         lum->lmm_object_seq = LOV_OBJECT_GROUP_DEFAULT;
 624         lum->lmm_pattern = ldesc->ld_pattern;
 625         lum->lmm_stripe_size = ldesc->ld_default_stripe_size;
 626         lum->lmm_stripe_count = ldesc->ld_default_stripe_count;
 627         lum->lmm_stripe_offset = ldesc->ld_default_stripe_offset;
 628
 629         RETURN(sizeof(*lum));
 630 }
 631
 632 /* get lov EA only */
 633 static int __mdd_lmm_get(const struct lu_env *env,
 634                          struct mdd_object *mdd_obj, struct md_attr *ma)
 635 {
 636         int rc;
 637         ENTRY;
 638
 639         if (ma->ma_valid & MA_LOV)
 640                 RETURN(0);
 641
 642         rc = mdd_get_md(env, mdd_obj, ma->ma_lmm, &ma->ma_lmm_size,
 643                         XATTR_NAME_LOV);
 644         if (rc == 0 && (ma->ma_need & MA_LOV_DEF))
 645                 rc = mdd_get_default_md(mdd_obj, ma->ma_lmm);
 646         if (rc > 0) {
 647                 ma->ma_lmm_size = rc;
 648                 ma->ma_valid |= MA_LOV;
 649                 rc = 0;
 650         }
 651         RETURN(rc);
 652 }
 653
 654 int mdd_lmm_get_locked(const struct lu_env *env, struct mdd_object *mdd_obj,
 655                        struct md_attr *ma)
 656 {
 657         int rc;
 658         ENTRY;
 659
 660         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 661         rc = __mdd_lmm_get(env, mdd_obj, ma);
 662         mdd_read_unlock(env, mdd_obj);
 663         RETURN(rc);
 664 }
 665
 666 /* get lmv EA only*/
 667 static int __mdd_lmv_get(const struct lu_env *env,
 668                          struct mdd_object *mdd_obj, struct md_attr *ma)
 669 {
 670         int rc;
 671         ENTRY;
 672
 673         if (ma->ma_valid & MA_LMV)
 674                 RETURN(0);
 675
 676         rc = mdd_get_md(env, mdd_obj, ma->ma_lmv, &ma->ma_lmv_size,
 677                         XATTR_NAME_LMV);
 678         if (rc > 0) {
 679                 ma->ma_valid |= MA_LMV;
 680                 rc = 0;
 681         }
 682         RETURN(rc);
 683 }
 684
 685 static int __mdd_lma_get(const struct lu_env *env, struct mdd_object *mdd_obj,
 686                          struct md_attr *ma)
 687 {
 688         struct mdd_thread_info *info = mdd_env_info(env);
 689         struct lustre_mdt_attrs *lma =
 690                                  (struct lustre_mdt_attrs *)info->mti_xattr_buf;
 691         int lma_size;
 692         int rc;
 693         ENTRY;
 694
 695         /* If all needed data are already valid, nothing to do */
 696         if ((ma->ma_valid & (MA_HSM | MA_SOM)) ==
 697             (ma->ma_need & (MA_HSM | MA_SOM)))
 698                 RETURN(0);
 699
 700         /* Read LMA from disk EA */
 701         lma_size = sizeof(info->mti_xattr_buf);
 702         rc = mdd_get_md(env, mdd_obj, lma, &lma_size, XATTR_NAME_LMA);
 703         if (rc <= 0)
 704                 RETURN(rc);
 705
 706         /* Useless to check LMA incompatibility because this is already done in
 707          * osd_ea_fid_get(), and this will fail long before this code is
 708          * called.
 709          * So, if we are here, LMA is compatible.
 710          */
 711
 712         lustre_lma_swab(lma);
 713
 714         /* Swab and copy LMA */
 715         if (ma->ma_need & MA_HSM) {
 716                 if (lma->lma_compat & LMAC_HSM)
 717                         ma->ma_hsm_flags = lma->lma_flags & HSM_FLAGS_MASK;
 718                 else
 719                         ma->ma_hsm_flags = 0;
 720                 ma->ma_valid |= MA_HSM;
 721         }
 722
 723         /* Copy SOM */
 724         if (ma->ma_need & MA_SOM && lma->lma_compat & LMAC_SOM) {
 725                 LASSERT(ma->ma_som != NULL);
 726                 ma->ma_som->msd_ioepoch = lma->lma_ioepoch;
 727                 ma->ma_som->msd_size    = lma->lma_som_size;
 728                 ma->ma_som->msd_blocks  = lma->lma_som_blocks;
 729                 ma->ma_som->msd_mountid = lma->lma_som_mountid;
 730                 ma->ma_valid |= MA_SOM;
 731         }
 732
 733         RETURN(0);
 734 }
 735
 736 static int mdd_attr_get_internal(const struct lu_env *env,
 737                                  struct mdd_object *mdd_obj,
 738                                  struct md_attr *ma)
 739 {
 740         int rc = 0;
 741         ENTRY;
 742
 743         if (ma->ma_need & MA_INODE)
 744                 rc = mdd_iattr_get(env, mdd_obj, ma);
 745
 746         if (rc == 0 && ma->ma_need & MA_LOV) {
 747                 if (S_ISREG(mdd_object_type(mdd_obj)) ||
 748                     S_ISDIR(mdd_object_type(mdd_obj)))
 749                         rc = __mdd_lmm_get(env, mdd_obj, ma);
 750         }
 751         if (rc == 0 && ma->ma_need & MA_LMV) {
 752                 if (S_ISDIR(mdd_object_type(mdd_obj)))
 753                         rc = __mdd_lmv_get(env, mdd_obj, ma);
 754         }
 755         if (rc == 0 && ma->ma_need & (MA_HSM | MA_SOM)) {
 756                 if (S_ISREG(mdd_object_type(mdd_obj)))
 757                         rc = __mdd_lma_get(env, mdd_obj, ma);
 758         }
 759 #ifdef CONFIG_FS_POSIX_ACL
 760         if (rc == 0 && ma->ma_need & MA_ACL_DEF) {
 761                 if (S_ISDIR(mdd_object_type(mdd_obj)))
 762                         rc = mdd_def_acl_get(env, mdd_obj, ma);
 763         }
 764 #endif
 765         CDEBUG(D_INODE, "after getattr rc = %d, ma_valid = "LPX64"\n",
 766                rc, ma->ma_valid);
 767         RETURN(rc);
 768 }
 769
 770 int mdd_attr_get_internal_locked(const struct lu_env *env,
 771                                  struct mdd_object *mdd_obj, struct md_attr *ma)
 772 {
 773         int rc;
 774         int needlock = ma->ma_need &
 775                        (MA_LOV | MA_LMV | MA_ACL_DEF | MA_HSM | MA_SOM);
 776
 777         if (needlock)
 778                 mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 779         rc = mdd_attr_get_internal(env, mdd_obj, ma);
 780         if (needlock)
 781                 mdd_read_unlock(env, mdd_obj);
 782         return rc;
 783 }
 784
 785 /*
 786  * No permission check is needed.
 787  */
 788 static int mdd_attr_get(const struct lu_env *env, struct md_object *obj,
 789                         struct md_attr *ma)
 790 {
 791         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 792         int                rc;
 793
 794         ENTRY;
 795         rc = mdd_attr_get_internal_locked(env, mdd_obj, ma);
 796         RETURN(rc);
 797 }
 798
 799 /*
 800  * No permission check is needed.
 801  */
 802 static int mdd_xattr_get(const struct lu_env *env,
 803                          struct md_object *obj, struct lu_buf *buf,
 804                          const char *name)
 805 {
 806         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 807         int rc;
 808
 809         ENTRY;
 810
 811         LASSERT(mdd_object_exists(mdd_obj));
 812
 813         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 814         rc = mdo_xattr_get(env, mdd_obj, buf, name,
 815                            mdd_object_capa(env, mdd_obj));
 816         mdd_read_unlock(env, mdd_obj);
 817
 818         RETURN(rc);
 819 }
 820
 821 /*
 822  * Permission check is done when open,
 823  * no need check again.
 824  */
 825 static int mdd_readlink(const struct lu_env *env, struct md_object *obj,
 826                         struct lu_buf *buf)
 827 {
 828         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 829         struct dt_object  *next;
 830         loff_t             pos = 0;
 831         int                rc;
 832         ENTRY;
 833
 834         LASSERT(mdd_object_exists(mdd_obj));
 835
 836         next = mdd_object_child(mdd_obj);
 837         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 838         rc = next->do_body_ops->dbo_read(env, next, buf, &pos,
 839                                          mdd_object_capa(env, mdd_obj));
 840         mdd_read_unlock(env, mdd_obj);
 841         RETURN(rc);
 842 }
 843
 844 /*
 845  * No permission check is needed.
 846  */
 847 static int mdd_xattr_list(const struct lu_env *env, struct md_object *obj,
 848                           struct lu_buf *buf)
 849 {
 850         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 851         int rc;
 852
 853         ENTRY;
 854
 855         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 856         rc = mdo_xattr_list(env, mdd_obj, buf, mdd_object_capa(env, mdd_obj));
 857         mdd_read_unlock(env, mdd_obj);
 858
 859         RETURN(rc);
 860 }
 861
 862 int mdd_object_create_internal(const struct lu_env *env, struct mdd_object *p,
 863                                struct mdd_object *c, struct md_attr *ma,
 864                                struct thandle *handle,
 865                                const struct md_op_spec *spec)
 866 {
 867         struct lu_attr *attr = &ma->ma_attr;
 868         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
 869         struct dt_object_format *dof = &mdd_env_info(env)->mti_dof;
 870         const struct dt_index_features *feat = spec->sp_feat;
 871         int rc;
 872         ENTRY;
 873
 874         if (!mdd_object_exists(c)) {
 875                 struct dt_object *next = mdd_object_child(c);
 876                 LASSERT(next);
 877
 878                 if (feat != &dt_directory_features && feat != NULL)
 879                         dof->dof_type = DFT_INDEX;
 880                 else
 881                         dof->dof_type = dt_mode_to_dft(attr->la_mode);
 882
 883                 dof->u.dof_idx.di_feat = feat;
 884
 885                 /* @hint will be initialized by underlying device. */
 886                 next->do_ops->do_ah_init(env, hint,
 887                                          p ? mdd_object_child(p) : NULL,
 888                                          attr->la_mode & S_IFMT);
 889
 890                 rc = mdo_create_obj(env, c, attr, hint, dof, handle);
 891                 LASSERT(ergo(rc == 0, mdd_object_exists(c)));
 892         } else
 893                 rc = -EEXIST;
 894
 895         RETURN(rc);
 896 }
 897
 898 /**
 899  * Make sure the ctime is increased only.
 900  */
 901 static inline int mdd_attr_check(const struct lu_env *env,
 902                                  struct mdd_object *obj,
 903                                  struct lu_attr *attr)
 904 {
 905         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
 906         int rc;
 907         ENTRY;
 908
 909         if (attr->la_valid & LA_CTIME) {
 910                 rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 911                 if (rc)
 912                         RETURN(rc);
 913
 914                 if (attr->la_ctime < tmp_la->la_ctime)
 915                         attr->la_valid &= ~(LA_MTIME | LA_CTIME);
 916                 else if (attr->la_valid == LA_CTIME &&
 917                          attr->la_ctime == tmp_la->la_ctime)
 918                         attr->la_valid &= ~LA_CTIME;
 919         }
 920         RETURN(0);
 921 }
 922
 923 int mdd_attr_set_internal(const struct lu_env *env,
 924                           struct mdd_object *obj,
 925                           struct lu_attr *attr,
 926                           struct thandle *handle,
 927                           int needacl)
 928 {
 929         int rc;
 930         ENTRY;
 931
 932         rc = mdo_attr_set(env, obj, attr, handle, mdd_object_capa(env, obj));
 933 #ifdef CONFIG_FS_POSIX_ACL
 934         if (!rc && (attr->la_valid & LA_MODE) && needacl)
 935                 rc = mdd_acl_chmod(env, obj, attr->la_mode, handle);
 936 #endif
 937         RETURN(rc);
 938 }
 939
 940 int mdd_attr_check_set_internal(const struct lu_env *env,
 941                                 struct mdd_object *obj,
 942                                 struct lu_attr *attr,
 943                                 struct thandle *handle,
 944                                 int needacl)
 945 {
 946         int rc;
 947         ENTRY;
 948
 949         rc = mdd_attr_check(env, obj, attr);
 950         if (rc)
 951                 RETURN(rc);
 952
 953         if (attr->la_valid)
 954                 rc = mdd_attr_set_internal(env, obj, attr, handle, needacl);
 955         RETURN(rc);
 956 }
 957
 958 static int mdd_attr_set_internal_locked(const struct lu_env *env,
 959                                         struct mdd_object *obj,
 960                                         struct lu_attr *attr,
 961                                         struct thandle *handle,
 962                                         int needacl)
 963 {
 964         int rc;
 965         ENTRY;
 966
 967         needacl = needacl && (attr->la_valid & LA_MODE);
 968         if (needacl)
 969                 mdd_write_lock(env, obj, MOR_TGT_CHILD);
 970         rc = mdd_attr_set_internal(env, obj, attr, handle, needacl);
 971         if (needacl)
 972                 mdd_write_unlock(env, obj);
 973         RETURN(rc);
 974 }
 975
 976 int mdd_attr_check_set_internal_locked(const struct lu_env *env,
 977                                        struct mdd_object *obj,
 978                                        struct lu_attr *attr,
 979                                        struct thandle *handle,
 980                                        int needacl)
 981 {
 982         int rc;
 983         ENTRY;
 984
 985         needacl = needacl && (attr->la_valid & LA_MODE);
 986         if (needacl)
 987                 mdd_write_lock(env, obj, MOR_TGT_CHILD);
 988         rc = mdd_attr_check_set_internal(env, obj, attr, handle, needacl);
 989         if (needacl)
 990                 mdd_write_unlock(env, obj);
 991         RETURN(rc);
 992 }
 993
 994 int __mdd_xattr_set(const struct lu_env *env, struct mdd_object *obj,
 995                     const struct lu_buf *buf, const char *name,
 996                     int fl, struct thandle *handle)
 997 {
 998         struct lustre_capa *capa = mdd_object_capa(env, obj);
 999         int rc = -EINVAL;
1000         ENTRY;
1001
1002         if (buf->lb_buf && buf->lb_len > 0)
1003                 rc = mdo_xattr_set(env, obj, buf, name, 0, handle, capa);
1004         else if (buf->lb_buf == NULL && buf->lb_len == 0)
1005                 rc = mdo_xattr_del(env, obj, name, handle, capa);
1006
1007         RETURN(rc);
1008 }
1009
1010 /*
1011  * This gives the same functionality as the code between
1012  * sys_chmod and inode_setattr
1013  * chown_common and inode_setattr
1014  * utimes and inode_setattr
1015  * This API is ported from mds_fix_attr but remove some unnecesssary stuff.
1016  */
1017 static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
1018                         struct lu_attr *la, const struct md_attr *ma)
1019 {
1020         struct lu_attr   *tmp_la     = &mdd_env_info(env)->mti_la;
1021         struct md_ucred  *uc;
1022         int               rc;
1023         ENTRY;
1024
1025         if (!la->la_valid)
1026                 RETURN(0);
1027
1028         /* Do not permit change file type */
1029         if (la->la_valid & LA_TYPE)
1030                 RETURN(-EPERM);
1031
1032         /* They should not be processed by setattr */
1033         if (la->la_valid & (LA_NLINK | LA_RDEV | LA_BLKSIZE))
1034                 RETURN(-EPERM);
1035
1036         /* export destroy does not have ->le_ses, but we may want
1037          * to drop LUSTRE_SOM_FL. */
1038         if (!env->le_ses)
1039                 RETURN(0);
1040
1041         uc = md_ucred(env);
1042
1043         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
1044         if (rc)
1045                 RETURN(rc);
1046
1047         if (la->la_valid == LA_CTIME) {
1048                 if (!(ma->ma_attr_flags & MDS_PERM_BYPASS))
1049                         /* This is only for set ctime when rename's source is
1050                          * on remote MDS. */
1051                         rc = mdd_may_delete(env, NULL, obj,
1052                                             (struct md_attr *)ma, 1, 0);
1053                 if (rc == 0 && la->la_ctime <= tmp_la->la_ctime)
1054                         la->la_valid &= ~LA_CTIME;
1055                 RETURN(rc);
1056         }
1057
1058         if (la->la_valid == LA_ATIME) {
1059                 /* This is atime only set for read atime update on close. */
1060                 if (la->la_atime <= tmp_la->la_atime +
1061                                     mdd_obj2mdd_dev(obj)->mdd_atime_diff)
1062                         la->la_valid &= ~LA_ATIME;
1063                 RETURN(0);
1064         }
1065
1066         /* Check if flags change. */
1067         if (la->la_valid & LA_FLAGS) {
1068                 unsigned int oldflags = 0;
1069                 unsigned int newflags = la->la_flags &
1070                                 (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL);
1071
1072                 if ((uc->mu_fsuid != tmp_la->la_uid) &&
1073                     !mdd_capable(uc, CFS_CAP_FOWNER))
1074                         RETURN(-EPERM);
1075
1076                 /* XXX: the IMMUTABLE and APPEND_ONLY flags can
1077                  * only be changed by the relevant capability. */
1078                 if (mdd_is_immutable(obj))
1079                         oldflags |= LUSTRE_IMMUTABLE_FL;
1080                 if (mdd_is_append(obj))
1081                         oldflags |= LUSTRE_APPEND_FL;
1082                 if ((oldflags ^ newflags) &&
1083                     !mdd_capable(uc, CFS_CAP_LINUX_IMMUTABLE))
1084                         RETURN(-EPERM);
1085
1086                 if (!S_ISDIR(tmp_la->la_mode))
1087                         la->la_flags &= ~LUSTRE_DIRSYNC_FL;
1088         }
1089
1090         if ((mdd_is_immutable(obj) || mdd_is_append(obj)) &&
1091             (la->la_valid & ~LA_FLAGS) &&
1092             !(ma->ma_attr_flags & MDS_PERM_BYPASS))
1093                 RETURN(-EPERM);
1094
1095         /* Check for setting the obj time. */
1096         if ((la->la_valid & (LA_MTIME | LA_ATIME | LA_CTIME)) &&
1097             !(la->la_valid & ~(LA_MTIME | LA_ATIME | LA_CTIME))) {
1098                 if ((uc->mu_fsuid != tmp_la->la_uid) &&
1099                     !mdd_capable(uc, CFS_CAP_FOWNER)) {
1100                         rc = mdd_permission_internal_locked(env, obj, tmp_la,
1101                                                             MAY_WRITE,
1102                                                             MOR_TGT_CHILD);
1103                         if (rc)
1104                                 RETURN(rc);
1105                 }
1106         }
1107
1108         /* Make sure a caller can chmod. */
1109         if (la->la_valid & LA_MODE) {
1110                 /* Bypass la_vaild == LA_MODE,
1111                  * this is for changing file with SUID or SGID. */
1112                 if ((la->la_valid & ~LA_MODE) &&
1113                     !(ma->ma_attr_flags & MDS_PERM_BYPASS) &&
1114                     (uc->mu_fsuid != tmp_la->la_uid) &&
1115                     !mdd_capable(uc, CFS_CAP_FOWNER))
1116                         RETURN(-EPERM);
1117
1118                 if (la->la_mode == (cfs_umode_t) -1)
1119                         la->la_mode = tmp_la->la_mode;
1120                 else
1121                         la->la_mode = (la->la_mode & S_IALLUGO) |
1122                                       (tmp_la->la_mode & ~S_IALLUGO);
1123
1124                 /* Also check the setgid bit! */
1125                 if (!lustre_in_group_p(uc, (la->la_valid & LA_GID) ?
1126                                        la->la_gid : tmp_la->la_gid) &&
1127                     !mdd_capable(uc, CFS_CAP_FSETID))
1128                         la->la_mode &= ~S_ISGID;
1129         } else {
1130                la->la_mode = tmp_la->la_mode;
1131         }
1132
1133         /* Make sure a caller can chown. */
1134         if (la->la_valid & LA_UID) {
1135                 if (la->la_uid == (uid_t) -1)
1136                         la->la_uid = tmp_la->la_uid;
1137                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
1138                     (la->la_uid != tmp_la->la_uid)) &&
1139                     !mdd_capable(uc, CFS_CAP_CHOWN))
1140                         RETURN(-EPERM);
1141
1142                 /* If the user or group of a non-directory has been
1143                  * changed by a non-root user, remove the setuid bit.
1144                  * 19981026 David C Niemi <niemi@tux.org>
1145                  *
1146                  * Changed this to apply to all users, including root,
1147                  * to avoid some races. This is the behavior we had in
1148                  * 2.0. The check for non-root was definitely wrong
1149                  * for 2.2 anyway, as it should have been using
1150                  * CAP_FSETID rather than fsuid -- 19990830 SD. */
1151                 if (((tmp_la->la_mode & S_ISUID) == S_ISUID) &&
1152                     !S_ISDIR(tmp_la->la_mode)) {
1153                         la->la_mode &= ~S_ISUID;
1154                         la->la_valid |= LA_MODE;
1155                 }
1156         }
1157
1158         /* Make sure caller can chgrp. */
1159         if (la->la_valid & LA_GID) {
1160                 if (la->la_gid == (gid_t) -1)
1161                         la->la_gid = tmp_la->la_gid;
1162                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
1163                     ((la->la_gid != tmp_la->la_gid) &&
1164                     !lustre_in_group_p(uc, la->la_gid))) &&
1165                     !mdd_capable(uc, CFS_CAP_CHOWN))
1166                         RETURN(-EPERM);
1167
1168                 /* Likewise, if the user or group of a non-directory
1169                  * has been changed by a non-root user, remove the
1170                  * setgid bit UNLESS there is no group execute bit
1171                  * (this would be a file marked for mandatory
1172                  * locking).  19981026 David C Niemi <niemi@tux.org>
1173                  *
1174                  * Removed the fsuid check (see the comment above) --
1175                  * 19990830 SD. */
1176                 if (((tmp_la->la_mode & (S_ISGID | S_IXGRP)) ==
1177                      (S_ISGID | S_IXGRP)) && !S_ISDIR(tmp_la->la_mode)) {
1178                         la->la_mode &= ~S_ISGID;
1179                         la->la_valid |= LA_MODE;
1180                 }
1181         }
1182
1183         /* For both Size-on-MDS case and truncate case,
1184          * "la->la_valid & (LA_SIZE | LA_BLOCKS)" are ture.
1185          * We distinguish them by "ma->ma_attr_flags & MDS_SOM".
1186          * For SOM case, it is true, the MAY_WRITE perm has been checked
1187          * when open, no need check again. For truncate case, it is false,
1188          * the MAY_WRITE perm should be checked here. */
1189         if (ma->ma_attr_flags & MDS_SOM) {
1190                 /* For the "Size-on-MDS" setattr update, merge coming
1191                  * attributes with the set in the inode. BUG 10641 */
1192                 if ((la->la_valid & LA_ATIME) &&
1193                     (la->la_atime <= tmp_la->la_atime))
1194                         la->la_valid &= ~LA_ATIME;
1195
1196                 /* OST attributes do not have a priority over MDS attributes,
1197                  * so drop times if ctime is equal. */
1198                 if ((la->la_valid & LA_CTIME) &&
1199                     (la->la_ctime <= tmp_la->la_ctime))
1200                         la->la_valid &= ~(LA_MTIME | LA_CTIME);
1201         } else {
1202                 if (la->la_valid & (LA_SIZE | LA_BLOCKS)) {
1203                         if (!((ma->ma_attr_flags & MDS_OPEN_OWNEROVERRIDE) &&
1204                               (uc->mu_fsuid == tmp_la->la_uid)) &&
1205                             !(ma->ma_attr_flags & MDS_PERM_BYPASS)) {
1206                                 rc = mdd_permission_internal_locked(env, obj,
1207                                                             tmp_la, MAY_WRITE,
1208                                                             MOR_TGT_CHILD);
1209                                 if (rc)
1210                                         RETURN(rc);
1211                         }
1212                 }
1213                 if (la->la_valid & LA_CTIME) {
1214                         /* The pure setattr, it has the priority over what is
1215                          * already set, do not drop it if ctime is equal. */
1216                         if (la->la_ctime < tmp_la->la_ctime)
1217                                 la->la_valid &= ~(LA_ATIME | LA_MTIME |
1218                                                   LA_CTIME);
1219                 }
1220         }
1221
1222         RETURN(0);
1223 }
1224
1225 /** Store a data change changelog record
1226  * If this fails, we must fail the whole transaction; we don't
1227  * want the change to commit without the log entry.
1228  * \param mdd_obj - mdd_object of change
1229  * \param handle - transacion handle
1230  */
1231 static int mdd_changelog_data_store(const struct lu_env     *env,
1232                                     struct mdd_device       *mdd,
1233                                     enum changelog_rec_type type,
1234                                     int                     flags,
1235                                     struct mdd_object       *mdd_obj,
1236                                     struct thandle          *handle)
1237 {
1238         const struct lu_fid *tfid = mdo2fid(mdd_obj);
1239         struct llog_changelog_rec *rec;
1240         struct lu_buf *buf;
1241         int reclen;
1242         int rc;
1243
1244         /* Not recording */
1245         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
1246                 RETURN(0);
1247         if ((mdd->mdd_cl.mc_mask & (1 << type)) == 0)
1248                 RETURN(0);
1249
1250         LASSERT(handle != NULL);
1251         LASSERT(mdd_obj != NULL);
1252
1253         if ((type >= CL_MTIME) && (type <= CL_ATIME) &&
1254             cfs_time_before_64(mdd->mdd_cl.mc_starttime, mdd_obj->mod_cltime)) {
1255                 /* Don't need multiple updates in this log */
1256                 /* Don't check under lock - no big deal if we get an extra
1257                    entry */
1258                 RETURN(0);
1259         }
1260
1261         reclen = llog_data_len(sizeof(*rec));
1262         buf = mdd_buf_alloc(env, reclen);
1263         if (buf->lb_buf == NULL)
1264                 RETURN(-ENOMEM);
1265         rec = (struct llog_changelog_rec *)buf->lb_buf;
1266
1267         rec->cr.cr_flags = CLF_VERSION | (CLF_FLAGMASK & flags);
1268         rec->cr.cr_type = (__u32)type;
1269         rec->cr.cr_tfid = *tfid;
1270         rec->cr.cr_namelen = 0;
1271         mdd_obj->mod_cltime = cfs_time_current_64();
1272
1273         rc = mdd_changelog_llog_write(mdd, rec, handle);
1274         if (rc < 0) {
1275                 CERROR("changelog failed: rc=%d op%d t"DFID"\n",
1276                        rc, type, PFID(tfid));
1277                 return -EFAULT;
1278         }
1279
1280         return 0;
1281 }
1282
1283 /**
1284  * Should be called with write lock held.
1285  *
1286  * \see mdd_lma_set_locked().
1287  */
1288 static int __mdd_lma_set(const struct lu_env *env, struct mdd_object *mdd_obj,
1289                        const struct md_attr *ma, struct thandle *handle)
1290 {
1291         struct mdd_thread_info *info = mdd_env_info(env);
1292         struct lu_buf *buf;
1293         struct lustre_mdt_attrs *lma =
1294                                 (struct lustre_mdt_attrs *) info->mti_xattr_buf;
1295         int lmasize = sizeof(struct lustre_mdt_attrs);
1296         int rc = 0;
1297
1298         ENTRY;
1299
1300         /* Either HSM or SOM part is not valid, we need to read it before */
1301         if ((!ma->ma_valid) & (MA_HSM | MA_SOM)) {
1302                 rc = mdd_get_md(env, mdd_obj, lma, &lmasize, XATTR_NAME_LMA);
1303                 if (rc <= 0)
1304                         RETURN(rc);
1305
1306                 lustre_lma_swab(lma);
1307         } else {
1308                 memset(lma, 0, lmasize);
1309         }
1310
1311         /* Copy HSM data */
1312         if (ma->ma_valid & MA_HSM) {
1313                 lma->lma_flags  |= ma->ma_hsm_flags & HSM_FLAGS_MASK;
1314                 lma->lma_compat |= LMAC_HSM;
1315         }
1316
1317         /* Copy SOM data */
1318         if (ma->ma_valid & MA_SOM) {
1319                 LASSERT(ma->ma_som != NULL);
1320                 if (ma->ma_som->msd_ioepoch == IOEPOCH_INVAL) {
1321                         lma->lma_compat     &= ~LMAC_SOM;
1322                 } else {
1323                         lma->lma_compat     |= LMAC_SOM;
1324                         lma->lma_ioepoch     = ma->ma_som->msd_ioepoch;
1325                         lma->lma_som_size    = ma->ma_som->msd_size;
1326                         lma->lma_som_blocks  = ma->ma_som->msd_blocks;
1327                         lma->lma_som_mountid = ma->ma_som->msd_mountid;
1328                 }
1329         }
1330
1331         /* Copy FID */
1332         memcpy(&lma->lma_self_fid, mdo2fid(mdd_obj), sizeof(lma->lma_self_fid));
1333
1334         lustre_lma_swab(lma);
1335         buf = mdd_buf_get(env, lma, lmasize);
1336         rc = __mdd_xattr_set(env, mdd_obj, buf, XATTR_NAME_LMA, 0, handle);
1337
1338         RETURN(rc);
1339 }
1340
1341 /**
1342  * Save LMA extended attributes with data from \a ma.
1343  *
1344  * HSM and Size-On-MDS data will be extracted from \ma if they are valid, if
1345  * not, LMA EA will be first read from disk, modified and write back.
1346  *
1347  */
1348 static int mdd_lma_set_locked(const struct lu_env *env,
1349                               struct mdd_object *mdd_obj,
1350                               const struct md_attr *ma, struct thandle *handle)
1351 {
1352         int rc;
1353
1354         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1355         rc = __mdd_lma_set(env, mdd_obj, ma, handle);
1356         mdd_write_unlock(env, mdd_obj);
1357         return rc;
1358 }
1359
1360 /* Precedence for choosing record type when multiple
1361  * attributes change: setattr > mtime > ctime > atime
1362  * (ctime changes when mtime does, plus chmod/chown.
1363  * atime and ctime are independent.) */
1364 static int mdd_attr_set_changelog(const struct lu_env *env,
1365                                   struct md_object *obj, struct thandle *handle,
1366                                   __u64 valid)
1367 {
1368         struct mdd_device *mdd = mdo2mdd(obj);
1369         int bits, type = 0;
1370
1371         bits = (valid & ~(LA_CTIME|LA_MTIME|LA_ATIME)) ? 1 << CL_SETATTR : 0;
1372         bits |= (valid & LA_MTIME) ? 1 << CL_MTIME : 0;
1373         bits |= (valid & LA_CTIME) ? 1 << CL_CTIME : 0;
1374         bits |= (valid & LA_ATIME) ? 1 << CL_ATIME : 0;
1375         bits = bits & mdd->mdd_cl.mc_mask;
1376         if (bits == 0)
1377                 return 0;
1378
1379         /* The record type is the lowest non-masked set bit */
1380         while (bits && ((bits & 1) == 0)) {
1381                 bits = bits >> 1;
1382                 type++;
1383         }
1384
1385         /* FYI we only store the first CLF_FLAGMASK bits of la_valid */
1386         return mdd_changelog_data_store(env, mdd, type, (int)valid,
1387                                         md2mdd_obj(obj), handle);
1388 }
1389
1390 /* set attr and LOV EA at once, return updated attr */
1391 static int mdd_attr_set(const struct lu_env *env, struct md_object *obj,
1392                         const struct md_attr *ma)
1393 {
1394         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1395         struct mdd_device *mdd = mdo2mdd(obj);
1396         struct thandle *handle;
1397         struct lov_mds_md *lmm = NULL;
1398         struct llog_cookie *logcookies = NULL;
1399         int  rc, lmm_size = 0, cookie_size = 0;
1400         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
1401 #ifdef HAVE_QUOTA_SUPPORT
1402         struct obd_device *obd = mdd->mdd_obd_dev;
1403         struct mds_obd *mds = &obd->u.mds;
1404         unsigned int qnids[MAXQUOTAS] = { 0, 0 };
1405         unsigned int qoids[MAXQUOTAS] = { 0, 0 };
1406         int quota_opc = 0, block_count = 0;
1407         int inode_pending[MAXQUOTAS] = { 0, 0 };
1408         int block_pending[MAXQUOTAS] = { 0, 0 };
1409 #endif
1410         ENTRY;
1411
1412         mdd_setattr_txn_param_build(env, obj, (struct md_attr *)ma,
1413                                     MDD_TXN_ATTR_SET_OP);
1414         handle = mdd_trans_start(env, mdd);
1415         if (IS_ERR(handle))
1416                 RETURN(PTR_ERR(handle));
1417         /*TODO: add lock here*/
1418         /* start a log jounal handle if needed */
1419         if (S_ISREG(mdd_object_type(mdd_obj)) &&
1420             ma->ma_attr.la_valid & (LA_UID | LA_GID)) {
1421                 lmm_size = mdd_lov_mdsize(env, mdd);
1422                 lmm = mdd_max_lmm_get(env, mdd);
1423                 if (lmm == NULL)
1424                         GOTO(cleanup, rc = -ENOMEM);
1425
1426                 rc = mdd_get_md_locked(env, mdd_obj, lmm, &lmm_size,
1427                                 XATTR_NAME_LOV);
1428
1429                 if (rc < 0)
1430                         GOTO(cleanup, rc);
1431         }
1432
1433         if (ma->ma_attr.la_valid & (LA_MTIME | LA_CTIME))
1434                 CDEBUG(D_INODE, "setting mtime "LPU64", ctime "LPU64"\n",
1435                        ma->ma_attr.la_mtime, ma->ma_attr.la_ctime);
1436
1437         *la_copy = ma->ma_attr;
1438         rc = mdd_fix_attr(env, mdd_obj, la_copy, ma);
1439         if (rc)
1440                 GOTO(cleanup, rc);
1441
1442 #ifdef HAVE_QUOTA_SUPPORT
1443         if (mds->mds_quota && la_copy->la_valid & (LA_UID | LA_GID)) {
1444                 struct obd_export *exp = md_quota(env)->mq_exp;
1445                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1446
1447                 rc = mdd_la_get(env, mdd_obj, la_tmp, BYPASS_CAPA);
1448                 if (!rc) {
1449                         quota_opc = FSFILT_OP_SETATTR;
1450                         mdd_quota_wrapper(la_copy, qnids);
1451                         mdd_quota_wrapper(la_tmp, qoids);
1452                         /* get file quota for new owner */
1453                         lquota_chkquota(mds_quota_interface_ref, obd, exp,
1454                                         qnids, inode_pending, 1, NULL, 0,
1455                                         NULL, 0);
1456                         block_count = (la_tmp->la_blocks + 7) >> 3;
1457                         if (block_count) {
1458                                 void *data = NULL;
1459                                 mdd_data_get(env, mdd_obj, &data);
1460                                 /* get block quota for new owner */
1461                                 lquota_chkquota(mds_quota_interface_ref, obd,
1462                                                 exp, qnids, block_pending,
1463                                                 block_count, NULL,
1464                                                 LQUOTA_FLAGS_BLK, data, 1);
1465                         }
1466                 }
1467         }
1468 #endif
1469
1470         if (la_copy->la_valid & LA_FLAGS) {
1471                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
1472                                                   handle, 1);
1473                 if (rc == 0)
1474                         mdd_flags_xlate(mdd_obj, la_copy->la_flags);
1475         } else if (la_copy->la_valid) {            /* setattr */
1476                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
1477                                                   handle, 1);
1478                 /* journal chown/chgrp in llog, just like unlink */
1479                 if (rc == 0 && lmm_size){
1480                         cookie_size = mdd_lov_cookiesize(env, mdd);
1481                         logcookies = mdd_max_cookie_get(env, mdd);
1482                         if (logcookies == NULL)
1483                                 GOTO(cleanup, rc = -ENOMEM);
1484
1485                         if (mdd_setattr_log(env, mdd, ma, lmm, lmm_size,
1486                                             logcookies, cookie_size) <= 0)
1487                                 logcookies = NULL;
1488                 }
1489         }
1490
1491         if (rc == 0 && ma->ma_valid & MA_LOV) {
1492                 cfs_umode_t mode;
1493
1494                 mode = mdd_object_type(mdd_obj);
1495                 if (S_ISREG(mode) || S_ISDIR(mode)) {
1496                         rc = mdd_lsm_sanity_check(env, mdd_obj);
1497                         if (rc)
1498                                 GOTO(cleanup, rc);
1499
1500                         rc = mdd_lov_set_md(env, NULL, mdd_obj, ma->ma_lmm,
1501                                             ma->ma_lmm_size, handle, 1);
1502                 }
1503
1504         }
1505         if (rc == 0 && ma->ma_valid & (MA_HSM | MA_SOM)) {
1506                 cfs_umode_t mode;
1507
1508                 mode = mdd_object_type(mdd_obj);
1509                 if (S_ISREG(mode))
1510                         rc = mdd_lma_set_locked(env, mdd_obj, ma, handle);
1511
1512         }
1513 cleanup:
1514         if (rc == 0)
1515                 rc = mdd_attr_set_changelog(env, obj, handle,
1516                                             ma->ma_attr.la_valid);
1517         mdd_trans_stop(env, mdd, rc, handle);
1518         if (rc == 0 && (lmm != NULL && lmm_size > 0 )) {
1519                 /*set obd attr, if needed*/
1520                 rc = mdd_lov_setattr_async(env, mdd_obj, lmm, lmm_size,
1521                                            logcookies);
1522         }
1523 #ifdef HAVE_QUOTA_SUPPORT
1524         if (quota_opc) {
1525                 lquota_pending_commit(mds_quota_interface_ref, obd, qnids,
1526                                       inode_pending, 0);
1527                 lquota_pending_commit(mds_quota_interface_ref, obd, qnids,
1528                                       block_pending, 1);
1529                 /* Trigger dqrel/dqacq for original owner and new owner.
1530                  * If failed, the next call for lquota_chkquota will
1531                  * process it. */
1532                 lquota_adjust(mds_quota_interface_ref, obd, qnids, qoids, rc,
1533                               quota_opc);
1534         }
1535 #endif
1536         RETURN(rc);
1537 }
1538
1539 int mdd_xattr_set_txn(const struct lu_env *env, struct mdd_object *obj,
1540                       const struct lu_buf *buf, const char *name, int fl,
1541                       struct thandle *handle)
1542 {
1543         int  rc;
1544         ENTRY;
1545
1546         mdd_write_lock(env, obj, MOR_TGT_CHILD);
1547         rc = __mdd_xattr_set(env, obj, buf, name, fl, handle);
1548         mdd_write_unlock(env, obj);
1549
1550         RETURN(rc);
1551 }
1552
1553 static int mdd_xattr_sanity_check(const struct lu_env *env,
1554                                   struct mdd_object *obj)
1555 {
1556         struct lu_attr  *tmp_la = &mdd_env_info(env)->mti_la;
1557         struct md_ucred *uc     = md_ucred(env);
1558         int rc;
1559         ENTRY;
1560
1561         if (mdd_is_immutable(obj) || mdd_is_append(obj))
1562                 RETURN(-EPERM);
1563
1564         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
1565         if (rc)
1566                 RETURN(rc);
1567
1568         if ((uc->mu_fsuid != tmp_la->la_uid) &&
1569             !mdd_capable(uc, CFS_CAP_FOWNER))
1570                 RETURN(-EPERM);
1571
1572         RETURN(rc);
1573 }
1574
1575 /**
1576  * The caller should guarantee to update the object ctime
1577  * after xattr_set if needed.
1578  */
1579 static int mdd_xattr_set(const struct lu_env *env, struct md_object *obj,
1580                          const struct lu_buf *buf, const char *name,
1581                          int fl)
1582 {
1583         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1584         struct mdd_device *mdd = mdo2mdd(obj);
1585         struct thandle *handle;
1586         int  rc;
1587         ENTRY;
1588
1589         rc = mdd_xattr_sanity_check(env, mdd_obj);
1590         if (rc)
1591                 RETURN(rc);
1592
1593         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
1594         /* security-replated changes may require sync */
1595         if (!strcmp(name, XATTR_NAME_ACL_ACCESS) &&
1596             mdd->mdd_sync_permission == 1)
1597                 txn_param_sync(&mdd_env_info(env)->mti_param);
1598
1599         handle = mdd_trans_start(env, mdd);
1600         if (IS_ERR(handle))
1601                 RETURN(PTR_ERR(handle));
1602
1603         rc = mdd_xattr_set_txn(env, mdd_obj, buf, name, fl, handle);
1604
1605         /* Only record user xattr changes */
1606         if ((rc == 0) && (strncmp("user.", name, 5) == 0))
1607                 rc = mdd_changelog_data_store(env, mdd, CL_XATTR, 0, mdd_obj,
1608                                               handle);
1609         mdd_trans_stop(env, mdd, rc, handle);
1610
1611         RETURN(rc);
1612 }
1613
1614 /**
1615  * The caller should guarantee to update the object ctime
1616  * after xattr_set if needed.
1617  */
1618 int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
1619                   const char *name)
1620 {
1621         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1622         struct mdd_device *mdd = mdo2mdd(obj);
1623         struct thandle *handle;
1624         int  rc;
1625         ENTRY;
1626
1627         rc = mdd_xattr_sanity_check(env, mdd_obj);
1628         if (rc)
1629                 RETURN(rc);
1630
1631         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
1632         handle = mdd_trans_start(env, mdd);
1633         if (IS_ERR(handle))
1634                 RETURN(PTR_ERR(handle));
1635
1636         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1637         rc = mdo_xattr_del(env, mdd_obj, name, handle,
1638                            mdd_object_capa(env, mdd_obj));
1639         mdd_write_unlock(env, mdd_obj);
1640
1641         /* Only record user xattr changes */
1642         if ((rc == 0) && (strncmp("user.", name, 5) != 0))
1643                 rc = mdd_changelog_data_store(env, mdd, CL_XATTR, 0, mdd_obj,
1644                                               handle);
1645
1646         mdd_trans_stop(env, mdd, rc, handle);
1647
1648         RETURN(rc);
1649 }
1650
1651 /* partial unlink */
1652 static int mdd_ref_del(const struct lu_env *env, struct md_object *obj,
1653                        struct md_attr *ma)
1654 {
1655         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
1656         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1657         struct mdd_device *mdd = mdo2mdd(obj);
1658         struct thandle *handle;
1659 #ifdef HAVE_QUOTA_SUPPORT
1660         struct obd_device *obd = mdd->mdd_obd_dev;
1661         struct mds_obd *mds = &obd->u.mds;
1662         unsigned int qids[MAXQUOTAS] = { 0, 0 };
1663         int quota_opc = 0;
1664 #endif
1665         int rc;
1666         ENTRY;
1667
1668         /*
1669          * Check -ENOENT early here because we need to get object type
1670          * to calculate credits before transaction start
1671          */
1672         if (!mdd_object_exists(mdd_obj))
1673                 RETURN(-ENOENT);
1674
1675         LASSERT(mdd_object_exists(mdd_obj) > 0);
1676
1677         rc = mdd_log_txn_param_build(env, obj, ma, MDD_TXN_UNLINK_OP);
1678         if (rc)
1679                 RETURN(rc);
1680
1681         handle = mdd_trans_start(env, mdd);
1682         if (IS_ERR(handle))
1683                 RETURN(-ENOMEM);
1684
1685         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1686
1687         rc = mdd_unlink_sanity_check(env, NULL, mdd_obj, ma);
1688         if (rc)
1689                 GOTO(cleanup, rc);
1690
1691         __mdd_ref_del(env, mdd_obj, handle, 0);
1692
1693         if (S_ISDIR(lu_object_attr(&obj->mo_lu))) {
1694                 /* unlink dot */
1695                 __mdd_ref_del(env, mdd_obj, handle, 1);
1696         }
1697
1698         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1699         la_copy->la_ctime = ma->ma_attr.la_ctime;
1700
1701         la_copy->la_valid = LA_CTIME;
1702         rc = mdd_attr_check_set_internal(env, mdd_obj, la_copy, handle, 0);
1703         if (rc)
1704                 GOTO(cleanup, rc);
1705
1706         rc = mdd_finish_unlink(env, mdd_obj, ma, handle);
1707 #ifdef HAVE_QUOTA_SUPPORT
1708         if (mds->mds_quota && ma->ma_valid & MA_INODE &&
1709             ma->ma_attr.la_nlink == 0 && mdd_obj->mod_count == 0) {
1710                 quota_opc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
1711                 mdd_quota_wrapper(&ma->ma_attr, qids);
1712         }
1713 #endif
1714
1715
1716         EXIT;
1717 cleanup:
1718         mdd_write_unlock(env, mdd_obj);
1719         mdd_trans_stop(env, mdd, rc, handle);
1720 #ifdef HAVE_QUOTA_SUPPORT
1721         if (quota_opc)
1722                 /* Trigger dqrel on the owner of child. If failed,
1723                  * the next call for lquota_chkquota will process it */
1724                 lquota_adjust(mds_quota_interface_ref, obd, qids, 0, rc,
1725                               quota_opc);
1726 #endif
1727         return rc;
1728 }
1729
1730 /* partial operation */
1731 static int mdd_oc_sanity_check(const struct lu_env *env,
1732                                struct mdd_object *obj,
1733                                struct md_attr *ma)
1734 {
1735         int rc;
1736         ENTRY;
1737
1738         switch (ma->ma_attr.la_mode & S_IFMT) {
1739         case S_IFREG:
1740         case S_IFDIR:
1741         case S_IFLNK:
1742         case S_IFCHR:
1743         case S_IFBLK:
1744         case S_IFIFO:
1745         case S_IFSOCK:
1746                 rc = 0;
1747                 break;
1748         default:
1749                 rc = -EINVAL;
1750                 break;
1751         }
1752         RETURN(rc);
1753 }
1754
1755 static int mdd_object_create(const struct lu_env *env,
1756                              struct md_object *obj,
1757                              const struct md_op_spec *spec,
1758                              struct md_attr *ma)
1759 {
1760
1761         struct mdd_device *mdd = mdo2mdd(obj);
1762         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1763         const struct lu_fid *pfid = spec->u.sp_pfid;
1764         struct thandle *handle;
1765 #ifdef HAVE_QUOTA_SUPPORT
1766         struct obd_device *obd = mdd->mdd_obd_dev;
1767         struct obd_export *exp = md_quota(env)->mq_exp;
1768         struct mds_obd *mds = &obd->u.mds;
1769         unsigned int qids[MAXQUOTAS] = { 0, 0 };
1770         int quota_opc = 0, block_count = 0;
1771         int inode_pending[MAXQUOTAS] = { 0, 0 };
1772         int block_pending[MAXQUOTAS] = { 0, 0 };
1773 #endif
1774         int rc = 0;
1775         ENTRY;
1776
1777 #ifdef HAVE_QUOTA_SUPPORT
1778         if (mds->mds_quota) {
1779                 quota_opc = FSFILT_OP_CREATE_PARTIAL_CHILD;
1780                 mdd_quota_wrapper(&ma->ma_attr, qids);
1781                 /* get file quota for child */
1782                 lquota_chkquota(mds_quota_interface_ref, obd, exp,
1783                                 qids, inode_pending, 1, NULL, 0,
1784                                 NULL, 0);
1785                 switch (ma->ma_attr.la_mode & S_IFMT) {
1786                 case S_IFLNK:
1787                 case S_IFDIR:
1788                         block_count = 2;
1789                         break;
1790                 case S_IFREG:
1791                         block_count = 1;
1792                         break;
1793                 }
1794                 /* get block quota for child */
1795                 if (block_count)
1796                         lquota_chkquota(mds_quota_interface_ref, obd, exp,
1797                                         qids, block_pending, block_count,
1798                                         NULL, LQUOTA_FLAGS_BLK, NULL, 0);
1799         }
1800 #endif
1801
1802         mdd_txn_param_build(env, mdd, MDD_TXN_OBJECT_CREATE_OP);
1803         handle = mdd_trans_start(env, mdd);
1804         if (IS_ERR(handle))
1805                 GOTO(out_pending, rc = PTR_ERR(handle));
1806
1807         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1808         rc = mdd_oc_sanity_check(env, mdd_obj, ma);
1809         if (rc)
1810                 GOTO(unlock, rc);
1811
1812         rc = mdd_object_create_internal(env, NULL, mdd_obj, ma, handle, spec);
1813         if (rc)
1814                 GOTO(unlock, rc);
1815
1816         if (spec->sp_cr_flags & MDS_CREATE_SLAVE_OBJ) {
1817                 /* If creating the slave object, set slave EA here. */
1818                 int lmv_size = spec->u.sp_ea.eadatalen;
1819                 struct lmv_stripe_md *lmv;
1820
1821                 lmv = (struct lmv_stripe_md *)spec->u.sp_ea.eadata;
1822                 LASSERT(lmv != NULL && lmv_size > 0);
1823
1824                 rc = __mdd_xattr_set(env, mdd_obj,
1825                                      mdd_buf_get_const(env, lmv, lmv_size),
1826                                      XATTR_NAME_LMV, 0, handle);
1827                 if (rc)
1828                         GOTO(unlock, rc);
1829
1830                 rc = mdd_attr_set_internal(env, mdd_obj, &ma->ma_attr,
1831                                            handle, 0);
1832         } else {
1833 #ifdef CONFIG_FS_POSIX_ACL
1834                 if (spec->sp_cr_flags & MDS_CREATE_RMT_ACL) {
1835                         struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
1836
1837                         buf->lb_buf = (void *)spec->u.sp_ea.eadata;
1838                         buf->lb_len = spec->u.sp_ea.eadatalen;
1839                         if ((buf->lb_len > 0) && (buf->lb_buf != NULL)) {
1840                                 rc = __mdd_acl_init(env, mdd_obj, buf,
1841                                                     &ma->ma_attr.la_mode,
1842                                                     handle);
1843                                 if (rc)
1844                                         GOTO(unlock, rc);
1845                                 else
1846                                         ma->ma_attr.la_valid |= LA_MODE;
1847                         }
1848
1849                         pfid = spec->u.sp_ea.fid;
1850                 }
1851 #endif
1852                 rc = mdd_object_initialize(env, pfid, NULL, mdd_obj, ma, handle,
1853                                            spec);
1854         }
1855         EXIT;
1856 unlock:
1857         if (rc == 0)
1858                 rc = mdd_attr_get_internal(env, mdd_obj, ma);
1859         mdd_write_unlock(env, mdd_obj);
1860
1861         mdd_trans_stop(env, mdd, rc, handle);
1862 out_pending:
1863 #ifdef HAVE_QUOTA_SUPPORT
1864         if (quota_opc) {
1865                 lquota_pending_commit(mds_quota_interface_ref, obd, qids,
1866                                       inode_pending, 0);
1867                 lquota_pending_commit(mds_quota_interface_ref, obd, qids,
1868                                       block_pending, 1);
1869                 /* Trigger dqacq on the owner of child. If failed,
1870                  * the next call for lquota_chkquota will process it. */
1871                 lquota_adjust(mds_quota_interface_ref, obd, qids, 0, rc,
1872                               quota_opc);
1873         }
1874 #endif
1875         return rc;
1876 }
1877
1878 /* partial link */
1879 static int mdd_ref_add(const struct lu_env *env, struct md_object *obj,
1880                        const struct md_attr *ma)
1881 {
1882         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
1883         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1884         struct mdd_device *mdd = mdo2mdd(obj);
1885         struct thandle *handle;
1886         int rc;
1887         ENTRY;
1888
1889         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
1890         handle = mdd_trans_start(env, mdd);
1891         if (IS_ERR(handle))
1892                 RETURN(-ENOMEM);
1893
1894         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1895         rc = mdd_link_sanity_check(env, NULL, NULL, mdd_obj);
1896         if (rc == 0)
1897                 __mdd_ref_add(env, mdd_obj, handle);
1898         mdd_write_unlock(env, mdd_obj);
1899         if (rc == 0) {
1900                 LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1901                 la_copy->la_ctime = ma->ma_attr.la_ctime;
1902
1903                 la_copy->la_valid = LA_CTIME;
1904                 rc = mdd_attr_check_set_internal_locked(env, mdd_obj, la_copy,
1905                                                         handle, 0);
1906         }
1907         mdd_trans_stop(env, mdd, 0, handle);
1908
1909         RETURN(rc);
1910 }
1911
1912 /*
1913  * do NOT or the MAY_*'s, you'll get the weakest
1914  */
1915 int accmode(const struct lu_env *env, struct lu_attr *la, int flags)
1916 {
1917         int res = 0;
1918
1919         /* Sadly, NFSD reopens a file repeatedly during operation, so the
1920          * "acc_mode = 0" allowance for newly-created files isn't honoured.
1921          * NFSD uses the MDS_OPEN_OWNEROVERRIDE flag to say that a file
1922          * owner can write to a file even if it is marked readonly to hide
1923          * its brokenness. (bug 5781) */
1924         if (flags & MDS_OPEN_OWNEROVERRIDE) {
1925                 struct md_ucred *uc = md_ucred(env);
1926
1927                 if ((uc == NULL) || (uc->mu_valid == UCRED_INIT) ||
1928                     (la->la_uid == uc->mu_fsuid))
1929                         return 0;
1930         }
1931
1932         if (flags & FMODE_READ)
1933                 res |= MAY_READ;
1934         if (flags & (FMODE_WRITE | MDS_OPEN_TRUNC | MDS_OPEN_APPEND))
1935                 res |= MAY_WRITE;
1936         if (flags & MDS_FMODE_EXEC)
1937                 res |= MAY_EXEC;
1938         return res;
1939 }
1940
1941 static int mdd_open_sanity_check(const struct lu_env *env,
1942                                  struct mdd_object *obj, int flag)
1943 {
1944         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
1945         int mode, rc;
1946         ENTRY;
1947
1948         /* EEXIST check */
1949         if (mdd_is_dead_obj(obj))
1950                 RETURN(-ENOENT);
1951
1952         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
1953         if (rc)
1954                RETURN(rc);
1955
1956         if (S_ISLNK(tmp_la->la_mode))
1957                 RETURN(-ELOOP);
1958
1959         mode = accmode(env, tmp_la, flag);
1960
1961         if (S_ISDIR(tmp_la->la_mode) && (mode & MAY_WRITE))
1962                 RETURN(-EISDIR);
1963
1964         if (!(flag & MDS_OPEN_CREATED)) {
1965                 rc = mdd_permission_internal(env, obj, tmp_la, mode);
1966                 if (rc)
1967                         RETURN(rc);
1968         }
1969
1970         if (S_ISFIFO(tmp_la->la_mode) || S_ISSOCK(tmp_la->la_mode) ||
1971             S_ISBLK(tmp_la->la_mode) || S_ISCHR(tmp_la->la_mode))
1972                 flag &= ~MDS_OPEN_TRUNC;
1973
1974         /* For writing append-only file must open it with append mode. */
1975         if (mdd_is_append(obj)) {
1976                 if ((flag & FMODE_WRITE) && !(flag & MDS_OPEN_APPEND))
1977                         RETURN(-EPERM);
1978                 if (flag & MDS_OPEN_TRUNC)
1979                         RETURN(-EPERM);
1980         }
1981
1982 #if 0
1983         /*
1984          * Now, flag -- O_NOATIME does not be packed by client.
1985          */
1986         if (flag & O_NOATIME) {
1987                 struct md_ucred *uc = md_ucred(env);
1988
1989                 if (uc && ((uc->mu_valid == UCRED_OLD) ||
1990                     (uc->mu_valid == UCRED_NEW)) &&
1991                     (uc->mu_fsuid != tmp_la->la_uid) &&
1992                     !mdd_capable(uc, CFS_CAP_FOWNER))
1993                         RETURN(-EPERM);
1994         }
1995 #endif
1996
1997         RETURN(0);
1998 }
1999
2000 static int mdd_open(const struct lu_env *env, struct md_object *obj,
2001                     int flags)
2002 {
2003         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2004         int rc = 0;
2005
2006         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
2007
2008         rc = mdd_open_sanity_check(env, mdd_obj, flags);
2009         if (rc == 0)
2010                 mdd_obj->mod_count++;
2011
2012         mdd_write_unlock(env, mdd_obj);
2013         return rc;
2014 }
2015
2016 /* return md_attr back,
2017  * if it is last unlink then return lov ea + llog cookie*/
2018 int mdd_object_kill(const struct lu_env *env, struct mdd_object *obj,
2019                     struct md_attr *ma)
2020 {
2021         int rc = 0;
2022         ENTRY;
2023
2024         if (S_ISREG(mdd_object_type(obj))) {
2025                 /* Return LOV & COOKIES unconditionally here. We clean evth up.
2026                  * Caller must be ready for that. */
2027
2028                 rc = __mdd_lmm_get(env, obj, ma);
2029                 if ((ma->ma_valid & MA_LOV))
2030                         rc = mdd_unlink_log(env, mdo2mdd(&obj->mod_obj),
2031                                             obj, ma);
2032         }
2033         RETURN(rc);
2034 }
2035
2036 /*
2037  * No permission check is needed.
2038  */
2039 static int mdd_close(const struct lu_env *env, struct md_object *obj,
2040                      struct md_attr *ma)
2041 {
2042         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2043         struct mdd_device *mdd = mdo2mdd(obj);
2044         struct thandle    *handle;
2045         int rc;
2046         int reset = 1;
2047
2048 #ifdef HAVE_QUOTA_SUPPORT
2049         struct obd_device *obd = mdo2mdd(obj)->mdd_obd_dev;
2050         struct mds_obd *mds = &obd->u.mds;
2051         unsigned int qids[MAXQUOTAS] = { 0, 0 };
2052         int quota_opc = 0;
2053 #endif
2054         ENTRY;
2055
2056         rc = mdd_log_txn_param_build(env, obj, ma, MDD_TXN_UNLINK_OP);
2057         if (rc)
2058                 RETURN(rc);
2059         handle = mdd_trans_start(env, mdo2mdd(obj));
2060         if (IS_ERR(handle))
2061                 RETURN(PTR_ERR(handle));
2062
2063         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
2064         /* release open count */
2065         mdd_obj->mod_count --;
2066
2067         if (mdd_obj->mod_count == 0 && mdd_obj->mod_flags & ORPHAN_OBJ) {
2068                 /* remove link to object from orphan index */
2069                 rc = __mdd_orphan_del(env, mdd_obj, handle);
2070                 if (rc == 0) {
2071                         CDEBUG(D_HA, "Object "DFID" is deleted from orphan "
2072                                "list, OSS objects to be destroyed.\n",
2073                                PFID(mdd_object_fid(mdd_obj)));
2074                 } else {
2075                         CERROR("Object "DFID" can not be deleted from orphan "
2076                                 "list, maybe cause OST objects can not be "
2077                                 "destroyed (err: %d).\n",
2078                                 PFID(mdd_object_fid(mdd_obj)), rc);
2079                         /* If object was not deleted from orphan list, do not
2080                          * destroy OSS objects, which will be done when next
2081                          * recovery. */
2082                         GOTO(out, rc);
2083                 }
2084         }
2085
2086         rc = mdd_iattr_get(env, mdd_obj, ma);
2087         /* Object maybe not in orphan list originally, it is rare case for
2088          * mdd_finish_unlink() failure. */
2089         if (rc == 0 && ma->ma_attr.la_nlink == 0) {
2090 #ifdef HAVE_QUOTA_SUPPORT
2091                 if (mds->mds_quota) {
2092                         quota_opc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
2093                         mdd_quota_wrapper(&ma->ma_attr, qids);
2094                 }
2095 #endif
2096                 /* MDS_CLOSE_CLEANUP means destroy OSS objects by MDS. */
2097                 if (ma->ma_valid & MA_FLAGS &&
2098                     ma->ma_attr_flags & MDS_CLOSE_CLEANUP) {
2099                         rc = mdd_lov_destroy(env, mdd, mdd_obj, &ma->ma_attr);
2100                 } else {
2101                         rc = mdd_object_kill(env, mdd_obj, ma);
2102                                 if (rc == 0)
2103                                         reset = 0;
2104                 }
2105
2106                 if (rc != 0)
2107                         CERROR("Error when prepare to delete Object "DFID" , "
2108                                "which will cause OST objects can not be "
2109                                "destroyed.\n",  PFID(mdd_object_fid(mdd_obj)));
2110         }
2111         EXIT;
2112
2113 out:
2114         if (reset)
2115                 ma->ma_valid &= ~(MA_LOV | MA_COOKIE);
2116
2117         mdd_write_unlock(env, mdd_obj);
2118         mdd_trans_stop(env, mdo2mdd(obj), rc, handle);
2119 #ifdef HAVE_QUOTA_SUPPORT
2120         if (quota_opc)
2121                 /* Trigger dqrel on the owner of child. If failed,
2122                  * the next call for lquota_chkquota will process it */
2123                 lquota_adjust(mds_quota_interface_ref, obd, qids, 0, rc,
2124                               quota_opc);
2125 #endif
2126         return rc;
2127 }
2128
2129 /*
2130  * Permission check is done when open,
2131  * no need check again.
2132  */
2133 static int mdd_readpage_sanity_check(const struct lu_env *env,
2134                                      struct mdd_object *obj)
2135 {
2136         struct dt_object *next = mdd_object_child(obj);
2137         int rc;
2138         ENTRY;
2139
2140         if (S_ISDIR(mdd_object_type(obj)) && dt_try_as_dir(env, next))
2141                 rc = 0;
2142         else
2143                 rc = -ENOTDIR;
2144
2145         RETURN(rc);
2146 }
2147
2148 static int mdd_dir_page_build(const struct lu_env *env, struct mdd_device *mdd,
2149                               int first, void *area, int nob,
2150                               const struct dt_it_ops *iops, struct dt_it *it,
2151                               __u64 *start, __u64 *end,
2152                               struct lu_dirent **last, __u32 attr)
2153 {
2154         int                     result;
2155         __u64                   hash = 0;
2156         struct lu_dirent       *ent;
2157
2158         if (first) {
2159                 memset(area, 0, sizeof (struct lu_dirpage));
2160                 area += sizeof (struct lu_dirpage);
2161                 nob  -= sizeof (struct lu_dirpage);
2162         }
2163
2164         ent  = area;
2165         do {
2166                 int    len;
2167                 int    recsize;
2168
2169                 len  = iops->key_size(env, it);
2170
2171                 /* IAM iterator can return record with zero len. */
2172                 if (len == 0)
2173                         goto next;
2174
2175                 hash = iops->store(env, it);
2176                 if (unlikely(first)) {
2177                         first = 0;
2178                         *start = hash;
2179                 }
2180
2181                 /* calculate max space required for lu_dirent */
2182                 recsize = lu_dirent_calc_size(len, attr);
2183
2184                 if (nob >= recsize) {
2185                         result = iops->rec(env, it, ent, attr);
2186                         if (result == -ESTALE)
2187                                 goto next;
2188                         if (result != 0)
2189                                 goto out;
2190
2191                         /* osd might not able to pack all attributes,
2192                          * so recheck rec length */
2193                         recsize = le16_to_cpu(ent->lde_reclen);
2194                 } else {
2195                         /*
2196                          * record doesn't fit into page, enlarge previous one.
2197                          */
2198                         if (*last) {
2199                                 (*last)->lde_reclen =
2200                                         cpu_to_le16(le16_to_cpu((*last)->lde_reclen) +
2201                                                         nob);
2202                                 result = 0;
2203                         } else
2204                                 result = -EINVAL;
2205
2206                         goto out;
2207                 }
2208                 *last = ent;
2209                 ent = (void *)ent + recsize;
2210                 nob -= recsize;
2211
2212 next:
2213                 result = iops->next(env, it);
2214                 if (result == -ESTALE)
2215                         goto next;
2216         } while (result == 0);
2217
2218 out:
2219         *end = hash;
2220         return result;
2221 }
2222
2223 static int __mdd_readpage(const struct lu_env *env, struct mdd_object *obj,
2224                           const struct lu_rdpg *rdpg)
2225 {
2226         struct dt_it      *it;
2227         struct dt_object  *next = mdd_object_child(obj);
2228         const struct dt_it_ops  *iops;
2229         struct page       *pg;
2230         struct lu_dirent  *last = NULL;
2231         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
2232         int i;
2233         int rc;
2234         int nob;
2235         __u64 hash_start;
2236         __u64 hash_end = 0;
2237
2238         LASSERT(rdpg->rp_pages != NULL);
2239         LASSERT(next->do_index_ops != NULL);
2240
2241         if (rdpg->rp_count <= 0)
2242                 return -EFAULT;
2243
2244         /*
2245          * iterate through directory and fill pages from @rdpg
2246          */
2247         iops = &next->do_index_ops->dio_it;
2248         it = iops->init(env, next, mdd_object_capa(env, obj));
2249         if (IS_ERR(it))
2250                 return PTR_ERR(it);
2251
2252         rc = iops->load(env, it, rdpg->rp_hash);
2253
2254         if (rc == 0){
2255                 /*
2256                  * Iterator didn't find record with exactly the key requested.
2257                  *
2258                  * It is currently either
2259                  *
2260                  *     - positioned above record with key less than
2261                  *     requested---skip it.
2262                  *
2263                  *     - or not positioned at all (is in IAM_IT_SKEWED
2264                  *     state)---position it on the next item.
2265                  */
2266                 rc = iops->next(env, it);
2267         } else if (rc > 0)
2268                 rc = 0;
2269
2270         /*
2271          * At this point and across for-loop:
2272          *
2273          *  rc == 0 -> ok, proceed.
2274          *  rc >  0 -> end of directory.
2275          *  rc <  0 -> error.
2276          */
2277         for (i = 0, nob = rdpg->rp_count; rc == 0 && nob > 0;
2278              i++, nob -= CFS_PAGE_SIZE) {
2279                 LASSERT(i < rdpg->rp_npages);
2280                 pg = rdpg->rp_pages[i];
2281                 rc = mdd_dir_page_build(env, mdd, !i, cfs_kmap(pg),
2282                                         min_t(int, nob, CFS_PAGE_SIZE), iops,
2283                                         it, &hash_start, &hash_end, &last,
2284                                         rdpg->rp_attrs);
2285                 if (rc != 0 || i == rdpg->rp_npages - 1) {
2286                         if (last)
2287                                 last->lde_reclen = 0;
2288                 }
2289                 cfs_kunmap(pg);
2290         }
2291         if (rc > 0) {
2292                 /*
2293                  * end of directory.
2294                  */
2295                 hash_end = DIR_END_OFF;
2296                 rc = 0;
2297         }
2298         if (rc == 0) {
2299                 struct lu_dirpage *dp;
2300
2301                 dp = cfs_kmap(rdpg->rp_pages[0]);
2302                 dp->ldp_hash_start = cpu_to_le64(rdpg->rp_hash);
2303                 dp->ldp_hash_end   = cpu_to_le64(hash_end);
2304                 if (i == 0)
2305                         /*
2306                          * No pages were processed, mark this.
2307                          */
2308                         dp->ldp_flags |= LDF_EMPTY;
2309
2310                 dp->ldp_flags = cpu_to_le32(dp->ldp_flags);
2311                 cfs_kunmap(rdpg->rp_pages[0]);
2312         }
2313         iops->put(env, it);
2314         iops->fini(env, it);
2315
2316         return rc;
2317 }
2318
2319 int mdd_readpage(const struct lu_env *env, struct md_object *obj,
2320                  const struct lu_rdpg *rdpg)
2321 {
2322         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2323         int rc;
2324         ENTRY;
2325
2326         LASSERT(mdd_object_exists(mdd_obj));
2327
2328         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
2329         rc = mdd_readpage_sanity_check(env, mdd_obj);
2330         if (rc)
2331                 GOTO(out_unlock, rc);
2332
2333         if (mdd_is_dead_obj(mdd_obj)) {
2334                 struct page *pg;
2335                 struct lu_dirpage *dp;
2336
2337                 /*
2338                  * According to POSIX, please do not return any entry to client:
2339                  * even dot and dotdot should not be returned.
2340                  */
2341                 CWARN("readdir from dead object: "DFID"\n",
2342                         PFID(mdd_object_fid(mdd_obj)));
2343
2344                 if (rdpg->rp_count <= 0)
2345                         GOTO(out_unlock, rc = -EFAULT);
2346                 LASSERT(rdpg->rp_pages != NULL);
2347
2348                 pg = rdpg->rp_pages[0];
2349                 dp = (struct lu_dirpage*)cfs_kmap(pg);
2350                 memset(dp, 0 , sizeof(struct lu_dirpage));
2351                 dp->ldp_hash_start = cpu_to_le64(rdpg->rp_hash);
2352                 dp->ldp_hash_end   = cpu_to_le64(DIR_END_OFF);
2353                 dp->ldp_flags |= LDF_EMPTY;
2354                 dp->ldp_flags = cpu_to_le32(dp->ldp_flags);
2355                 cfs_kunmap(pg);
2356                 GOTO(out_unlock, rc = 0);
2357         }
2358
2359         rc = __mdd_readpage(env, mdd_obj, rdpg);
2360
2361         EXIT;
2362 out_unlock:
2363         mdd_read_unlock(env, mdd_obj);
2364         return rc;
2365 }
2366
2367 static int mdd_object_sync(const struct lu_env *env, struct md_object *obj)
2368 {
2369         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2370         struct dt_object *next;
2371
2372         LASSERT(mdd_object_exists(mdd_obj));
2373         next = mdd_object_child(mdd_obj);
2374         return next->do_ops->do_object_sync(env, next);
2375 }
2376
2377 static dt_obj_version_t mdd_version_get(const struct lu_env *env,
2378                                         struct md_object *obj)
2379 {
2380         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2381
2382         LASSERT(mdd_object_exists(mdd_obj));
2383         return do_version_get(env, mdd_object_child(mdd_obj));
2384 }
2385
2386 static void mdd_version_set(const struct lu_env *env, struct md_object *obj,
2387                             dt_obj_version_t version)
2388 {
2389         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2390
2391         LASSERT(mdd_object_exists(mdd_obj));
2392         do_version_set(env, mdd_object_child(mdd_obj), version);
2393 }
2394
2395 const struct md_object_operations mdd_obj_ops = {
2396         .moo_permission    = mdd_permission,
2397         .moo_attr_get      = mdd_attr_get,
2398         .moo_attr_set      = mdd_attr_set,
2399         .moo_xattr_get     = mdd_xattr_get,
2400         .moo_xattr_set     = mdd_xattr_set,
2401         .moo_xattr_list    = mdd_xattr_list,
2402         .moo_xattr_del     = mdd_xattr_del,
2403         .moo_object_create = mdd_object_create,
2404         .moo_ref_add       = mdd_ref_add,
2405         .moo_ref_del       = mdd_ref_del,
2406         .moo_open          = mdd_open,
2407         .moo_close         = mdd_close,
2408         .moo_readpage      = mdd_readpage,
2409         .moo_readlink      = mdd_readlink,
2410         .moo_capa_get      = mdd_capa_get,
2411         .moo_object_sync   = mdd_object_sync,
2412         .moo_version_get   = mdd_version_get,
2413         .moo_version_set   = mdd_version_set,
2414         .moo_path          = mdd_path,
2415 };