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  *  mdd/mdd_handler.c
   5  *  Lustre Metadata Server (mdd) routines
   6  *
   7  *  Copyright (C) 2006 Cluster File Systems, Inc.
   8  *   Author: Wang Di <wangdi@clusterfs.com>
   9  *
  10  *   This file is part of the Lustre file system, http://www.lustre.org
  11  *   Lustre is a trademark of Cluster File Systems, Inc.
  12  *
  13  *   You may have signed or agreed to another license before downloading
  14  *   this software.  If so, you are bound by the terms and conditions
  15  *   of that agreement, and the following does not apply to you.  See the
  16  *   LICENSE file included with this distribution for more information.
  17  *
  18  *   If you did not agree to a different license, then this copy of Lustre
  19  *   is open source software; you can redistribute it and/or modify it
  20  *   under the terms of version 2 of the GNU General Public License as
  21  *   published by the Free Software Foundation.
  22  *
  23  *   In either case, Lustre is distributed in the hope that it will be
  24  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  25  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  26  *   license text for more details.
  27  */
  28 #ifndef EXPORT_SYMTAB
  29 # define EXPORT_SYMTAB
  30 #endif
  31 #define DEBUG_SUBSYSTEM S_MDS
  32
  33 #include <linux/module.h>
  34 #include <linux/jbd.h>
  35 #include <obd.h>
  36 #include <obd_class.h>
  37 #include <obd_support.h>
  38 #include <lprocfs_status.h>
  39 /* fid_be_cpu(), fid_cpu_to_be(). */
  40 #include <lustre_fid.h>
  41
  42 #include <linux/ldiskfs_fs.h>
  43 #include <lustre_mds.h>
  44 #include <lustre/lustre_idl.h>
  45
  46 #include "mdd_internal.h"
  47
  48 static struct lu_object_operations mdd_lu_obj_ops;
  49
  50 int mdd_la_get(const struct lu_env *env, struct mdd_object *obj,
  51                struct lu_attr *la, struct lustre_capa *capa)
  52 {
  53         LASSERTF(mdd_object_exists(obj), "FID is "DFID"\n",
  54                  PFID(mdd_object_fid(obj)));
  55         return mdo_attr_get(env, obj, la, capa);
  56 }
  57
  58 static void mdd_flags_xlate(struct mdd_object *obj, __u32 flags)
  59 {
  60         obj->mod_flags &= ~(APPEND_OBJ|IMMUTE_OBJ);
  61
  62         if (flags & LUSTRE_APPEND_FL)
  63                 obj->mod_flags |= APPEND_OBJ;
  64
  65         if (flags & LUSTRE_IMMUTABLE_FL)
  66                 obj->mod_flags |= IMMUTE_OBJ;
  67 }
  68
  69 struct lu_buf *mdd_buf_get(const struct lu_env *env, void *area, ssize_t len)
  70 {
  71         struct lu_buf *buf;
  72
  73         buf = &mdd_env_info(env)->mti_buf;
  74         buf->lb_buf = area;
  75         buf->lb_len = len;
  76         return buf;
  77 }
  78
  79 struct llog_cookie *mdd_max_cookie_get(const struct lu_env *env,
  80                                        struct mdd_device *mdd)
  81 {
  82         struct mdd_thread_info *mti = mdd_env_info(env);
  83         int                     max_cookie_size;
  84
  85         max_cookie_size = mdd_lov_cookiesize(env, mdd);
  86         if (unlikely(mti->mti_max_cookie_size < max_cookie_size)) {
  87                 if (mti->mti_max_cookie)
  88                         OBD_FREE(mti->mti_max_cookie, mti->mti_max_cookie_size);
  89                 mti->mti_max_cookie = NULL;
  90                 mti->mti_max_cookie_size = 0;
  91         }
  92         if (unlikely(mti->mti_max_cookie == NULL)) {
  93                 OBD_ALLOC(mti->mti_max_cookie, max_cookie_size);
  94                 if (unlikely(mti->mti_max_cookie != NULL))
  95                         mti->mti_max_cookie_size = max_cookie_size;
  96         }
  97         return mti->mti_max_cookie;
  98 }
  99
 100 struct lov_mds_md *mdd_max_lmm_get(const struct lu_env *env,
 101                                    struct mdd_device *mdd)
 102 {
 103         struct mdd_thread_info *mti = mdd_env_info(env);
 104         int                     max_lmm_size;
 105
 106         max_lmm_size = mdd_lov_mdsize(env, mdd);
 107         if (unlikely(mti->mti_max_lmm_size < max_lmm_size)) {
 108                 if (mti->mti_max_lmm)
 109                         OBD_FREE(mti->mti_max_lmm, mti->mti_max_lmm_size);
 110                 mti->mti_max_lmm = NULL;
 111                 mti->mti_max_lmm_size = 0;
 112         }
 113         if (unlikely(mti->mti_max_lmm == NULL)) {
 114                 OBD_ALLOC(mti->mti_max_lmm, max_lmm_size);
 115                 if (unlikely(mti->mti_max_lmm != NULL))
 116                         mti->mti_max_lmm_size = max_lmm_size;
 117         }
 118         return mti->mti_max_lmm;
 119 }
 120
 121 const struct lu_buf *mdd_buf_get_const(const struct lu_env *env,
 122                                        const void *area, ssize_t len)
 123 {
 124         struct lu_buf *buf;
 125
 126         buf = &mdd_env_info(env)->mti_buf;
 127         buf->lb_buf = (void *)area;
 128         buf->lb_len = len;
 129         return buf;
 130 }
 131
 132 struct mdd_thread_info *mdd_env_info(const struct lu_env *env)
 133 {
 134         struct mdd_thread_info *info;
 135
 136         info = lu_context_key_get(&env->le_ctx, &mdd_thread_key);
 137         LASSERT(info != NULL);
 138         return info;
 139 }
 140
 141 struct lu_object *mdd_object_alloc(const struct lu_env *env,
 142                                    const struct lu_object_header *hdr,
 143                                    struct lu_device *d)
 144 {
 145         struct mdd_object *mdd_obj;
 146
 147         OBD_ALLOC_PTR(mdd_obj);
 148         if (mdd_obj != NULL) {
 149                 struct lu_object *o;
 150
 151                 o = mdd2lu_obj(mdd_obj);
 152                 lu_object_init(o, NULL, d);
 153                 mdd_obj->mod_obj.mo_ops = &mdd_obj_ops;
 154                 mdd_obj->mod_obj.mo_dir_ops = &mdd_dir_ops;
 155                 mdd_obj->mod_count = 0;
 156                 o->lo_ops = &mdd_lu_obj_ops;
 157                 return o;
 158         } else {
 159                 return NULL;
 160         }
 161 }
 162
 163 static int mdd_object_init(const struct lu_env *env, struct lu_object *o)
 164 {
 165         struct mdd_device *d = lu2mdd_dev(o->lo_dev);
 166         struct lu_object  *below;
 167         struct lu_device  *under;
 168         ENTRY;
 169
 170         under = &d->mdd_child->dd_lu_dev;
 171         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
 172         mdd_pdlock_init(lu2mdd_obj(o));
 173         if (below == NULL)
 174                 RETURN(-ENOMEM);
 175
 176         lu_object_add(o, below);
 177         RETURN(0);
 178 }
 179
 180 static int mdd_object_start(const struct lu_env *env, struct lu_object *o)
 181 {
 182         if (lu_object_exists(o))
 183                 return mdd_get_flags(env, lu2mdd_obj(o));
 184         else
 185                 return 0;
 186 }
 187
 188 static void mdd_object_free(const struct lu_env *env, struct lu_object *o)
 189 {
 190         struct mdd_object *mdd = lu2mdd_obj(o);
 191
 192         lu_object_fini(o);
 193         OBD_FREE_PTR(mdd);
 194 }
 195
 196 static int mdd_object_print(const struct lu_env *env, void *cookie,
 197                             lu_printer_t p, const struct lu_object *o)
 198 {
 199         return (*p)(env, cookie, LUSTRE_MDD_NAME"-object@%p", o);
 200 }
 201
 202 /* orphan handling is here */
 203 static void mdd_object_delete(const struct lu_env *env,
 204                                struct lu_object *o)
 205 {
 206         struct mdd_object *mdd_obj = lu2mdd_obj(o);
 207         struct thandle *handle = NULL;
 208         ENTRY;
 209
 210         if (lu2mdd_dev(o->lo_dev)->mdd_orphans == NULL)
 211                 return;
 212
 213         if (mdd_obj->mod_flags & ORPHAN_OBJ) {
 214                 mdd_txn_param_build(env, lu2mdd_dev(o->lo_dev),
 215                                     MDD_TXN_INDEX_DELETE_OP);
 216                 handle = mdd_trans_start(env, lu2mdd_dev(o->lo_dev));
 217                 if (IS_ERR(handle))
 218                         CERROR("Cannot get thandle\n");
 219                 else {
 220                         mdd_write_lock(env, mdd_obj);
 221                         /* let's remove obj from the orphan list */
 222                         __mdd_orphan_del(env, mdd_obj, handle);
 223                         mdd_write_unlock(env, mdd_obj);
 224                         mdd_trans_stop(env, lu2mdd_dev(o->lo_dev),
 225                                        0, handle);
 226                 }
 227         }
 228 }
 229
 230 static struct lu_object_operations mdd_lu_obj_ops = {
 231         .loo_object_init    = mdd_object_init,
 232         .loo_object_start   = mdd_object_start,
 233         .loo_object_free    = mdd_object_free,
 234         .loo_object_print   = mdd_object_print,
 235         .loo_object_delete  = mdd_object_delete
 236 };
 237
 238 struct mdd_object *mdd_object_find(const struct lu_env *env,
 239                                    struct mdd_device *d,
 240                                    const struct lu_fid *f)
 241 {
 242         struct lu_object *o, *lo;
 243         struct mdd_object *m;
 244         ENTRY;
 245
 246         o = lu_object_find(env, mdd2lu_dev(d)->ld_site, f);
 247         if (IS_ERR(o))
 248                 m = (struct mdd_object *)o;
 249         else {
 250                 lo = lu_object_locate(o->lo_header, mdd2lu_dev(d)->ld_type);
 251                 /* remote object can't be located and should be put then */
 252                 if (lo == NULL)
 253                         lu_object_put(env, o);
 254                 m = lu2mdd_obj(lo);
 255         }
 256         RETURN(m);
 257 }
 258
 259 int mdd_get_flags(const struct lu_env *env, struct mdd_object *obj)
 260 {
 261         struct lu_attr *la = &mdd_env_info(env)->mti_la;
 262         int rc;
 263
 264         ENTRY;
 265         rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
 266         if (rc == 0) {
 267                 mdd_flags_xlate(obj, la->la_flags);
 268                 if (S_ISDIR(la->la_mode) && la->la_nlink == 1)
 269                         obj->mod_flags |= MNLINK_OBJ;
 270         }
 271         RETURN(rc);
 272 }
 273
 274 /* get only inode attributes */
 275 int mdd_iattr_get(const struct lu_env *env, struct mdd_object *mdd_obj,
 276                   struct md_attr *ma)
 277 {
 278         int rc = 0;
 279         ENTRY;
 280
 281         if (ma->ma_valid & MA_INODE)
 282                 RETURN(0);
 283
 284         rc = mdd_la_get(env, mdd_obj, &ma->ma_attr,
 285                           mdd_object_capa(env, mdd_obj));
 286         if (rc == 0)
 287                 ma->ma_valid |= MA_INODE;
 288         RETURN(rc);
 289 }
 290
 291 static int mdd_get_default_md(struct mdd_object *mdd_obj,
 292                 struct lov_mds_md *lmm, int *size)
 293 {
 294         struct lov_desc *ldesc;
 295         struct mdd_device *mdd = mdo2mdd(&mdd_obj->mod_obj);
 296         ENTRY;
 297
 298         ldesc = &mdd->mdd_obd_dev->u.mds.mds_lov_desc;
 299         LASSERT(ldesc != NULL);
 300
 301         if (!lmm)
 302                 RETURN(0);
 303
 304         lmm->lmm_magic = LOV_MAGIC_V1;
 305         lmm->lmm_object_gr = LOV_OBJECT_GROUP_DEFAULT;
 306         lmm->lmm_pattern = ldesc->ld_pattern;
 307         lmm->lmm_stripe_size = ldesc->ld_default_stripe_size;
 308         lmm->lmm_stripe_count = ldesc->ld_default_stripe_count;
 309         *size = sizeof(struct lov_mds_md);
 310
 311         RETURN(sizeof(struct lov_mds_md));
 312 }
 313
 314 /* get lov EA only */
 315 static int __mdd_lmm_get(const struct lu_env *env,
 316                          struct mdd_object *mdd_obj, struct md_attr *ma)
 317 {
 318         int rc;
 319         ENTRY;
 320
 321         if (ma->ma_valid & MA_LOV)
 322                 RETURN(0);
 323
 324         rc = mdd_get_md(env, mdd_obj, ma->ma_lmm, &ma->ma_lmm_size,
 325                         MDS_LOV_MD_NAME);
 326
 327         if (rc == 0 && (ma->ma_need & MA_LOV_DEF)) {
 328                 rc = mdd_get_default_md(mdd_obj, ma->ma_lmm,
 329                                 &ma->ma_lmm_size);
 330         }
 331
 332         if (rc > 0) {
 333                 ma->ma_valid |= MA_LOV;
 334                 rc = 0;
 335         }
 336         RETURN(rc);
 337 }
 338
 339 int mdd_lmm_get_locked(const struct lu_env *env, struct mdd_object *mdd_obj,
 340                        struct md_attr *ma)
 341 {
 342         int rc;
 343         ENTRY;
 344
 345         mdd_read_lock(env, mdd_obj);
 346         rc = __mdd_lmm_get(env, mdd_obj, ma);
 347         mdd_read_unlock(env, mdd_obj);
 348         RETURN(rc);
 349 }
 350
 351 /* get lmv EA only*/
 352 static int __mdd_lmv_get(const struct lu_env *env,
 353                          struct mdd_object *mdd_obj, struct md_attr *ma)
 354 {
 355         int rc;
 356         ENTRY;
 357
 358         if (ma->ma_valid & MA_LMV)
 359                 RETURN(0);
 360
 361         rc = mdd_get_md(env, mdd_obj, ma->ma_lmv, &ma->ma_lmv_size,
 362                         MDS_LMV_MD_NAME);
 363         if (rc > 0) {
 364                 ma->ma_valid |= MA_LMV;
 365                 rc = 0;
 366         }
 367         RETURN(rc);
 368 }
 369
 370 static int mdd_attr_get_internal(const struct lu_env *env,
 371                                  struct mdd_object *mdd_obj,
 372                                  struct md_attr *ma)
 373 {
 374         int rc = 0;
 375         ENTRY;
 376
 377         if (ma->ma_need & MA_INODE)
 378                 rc = mdd_iattr_get(env, mdd_obj, ma);
 379
 380         if (rc == 0 && ma->ma_need & MA_LOV) {
 381                 if (S_ISREG(mdd_object_type(mdd_obj)) ||
 382                     S_ISDIR(mdd_object_type(mdd_obj)))
 383                         rc = __mdd_lmm_get(env, mdd_obj, ma);
 384         }
 385         if (rc == 0 && ma->ma_need & MA_LMV) {
 386                 if (S_ISDIR(mdd_object_type(mdd_obj)))
 387                         rc = __mdd_lmv_get(env, mdd_obj, ma);
 388         }
 389 #ifdef CONFIG_FS_POSIX_ACL
 390         if (rc == 0 && ma->ma_need & MA_ACL_DEF) {
 391                 if (S_ISDIR(mdd_object_type(mdd_obj)))
 392                         rc = mdd_acl_def_get(env, mdd_obj, ma);
 393         }
 394 #endif
 395         CDEBUG(D_INODE, "after getattr rc = %d, ma_valid = "LPX64"\n",
 396                rc, ma->ma_valid);
 397         RETURN(rc);
 398 }
 399
 400 int mdd_attr_get_internal_locked(const struct lu_env *env,
 401                                  struct mdd_object *mdd_obj, struct md_attr *ma)
 402 {
 403         int rc;
 404         int needlock = ma->ma_need & (MA_LOV | MA_LMV | MA_ACL_DEF);
 405
 406         if (needlock)
 407                 mdd_read_lock(env, mdd_obj);
 408         rc = mdd_attr_get_internal(env, mdd_obj, ma);
 409         if (needlock)
 410                 mdd_read_unlock(env, mdd_obj);
 411         return rc;
 412 }
 413
 414 /*
 415  * No permission check is needed.
 416  */
 417 static int mdd_attr_get(const struct lu_env *env, struct md_object *obj,
 418                         struct md_attr *ma)
 419 {
 420         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 421         int                rc;
 422
 423         ENTRY;
 424         rc = mdd_attr_get_internal_locked(env, mdd_obj, ma);
 425         RETURN(rc);
 426 }
 427
 428 /*
 429  * No permission check is needed.
 430  */
 431 static int mdd_xattr_get(const struct lu_env *env,
 432                          struct md_object *obj, struct lu_buf *buf,
 433                          const char *name)
 434 {
 435         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 436         int rc;
 437
 438         ENTRY;
 439
 440         LASSERT(mdd_object_exists(mdd_obj));
 441
 442         mdd_read_lock(env, mdd_obj);
 443         rc = mdo_xattr_get(env, mdd_obj, buf, name,
 444                            mdd_object_capa(env, mdd_obj));
 445         mdd_read_unlock(env, mdd_obj);
 446
 447         RETURN(rc);
 448 }
 449
 450 /*
 451  * Permission check is done when open,
 452  * no need check again.
 453  */
 454 static int mdd_readlink(const struct lu_env *env, struct md_object *obj,
 455                         struct lu_buf *buf)
 456 {
 457         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 458         struct dt_object  *next;
 459         loff_t             pos = 0;
 460         int                rc;
 461         ENTRY;
 462
 463         LASSERT(mdd_object_exists(mdd_obj));
 464
 465         next = mdd_object_child(mdd_obj);
 466         mdd_read_lock(env, mdd_obj);
 467         rc = next->do_body_ops->dbo_read(env, next, buf, &pos,
 468                                          mdd_object_capa(env, mdd_obj));
 469         mdd_read_unlock(env, mdd_obj);
 470         RETURN(rc);
 471 }
 472
 473 /*
 474  * No permission check is needed.
 475  */
 476 static int mdd_xattr_list(const struct lu_env *env, struct md_object *obj,
 477                           struct lu_buf *buf)
 478 {
 479         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 480         int rc;
 481
 482         ENTRY;
 483
 484         mdd_read_lock(env, mdd_obj);
 485         rc = mdo_xattr_list(env, mdd_obj, buf, mdd_object_capa(env, mdd_obj));
 486         mdd_read_unlock(env, mdd_obj);
 487
 488         RETURN(rc);
 489 }
 490
 491 int mdd_object_create_internal(const struct lu_env *env, struct mdd_object *p,
 492                                struct mdd_object *c, struct md_attr *ma,
 493                                struct thandle *handle)
 494 {
 495         struct lu_attr *attr = &ma->ma_attr;
 496         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
 497         int rc;
 498         ENTRY;
 499
 500         if (!mdd_object_exists(c)) {
 501                 struct dt_object *next = mdd_object_child(c);
 502                 LASSERT(next);
 503
 504                 /* @hint will be initialized by underlying device. */
 505                 next->do_ops->do_ah_init(env, hint,
 506                                          p ? mdd_object_child(p) : NULL,
 507                                          attr->la_mode & S_IFMT);
 508                 rc = mdo_create_obj(env, c, attr, hint, handle);
 509                 LASSERT(ergo(rc == 0, mdd_object_exists(c)));
 510         } else
 511                 rc = -EEXIST;
 512
 513         RETURN(rc);
 514 }
 515
 516 static inline int mdd_attr_check(const struct lu_env *env,
 517                                  struct mdd_object *obj,
 518                                  struct lu_attr *attr)
 519 {
 520         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
 521         int rc;
 522         ENTRY;
 523
 524         if (attr->la_valid & LA_CTIME) {
 525                 rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 526                 if (rc)
 527                         RETURN(rc);
 528
 529                 if (attr->la_ctime < tmp_la->la_ctime)
 530                         attr->la_valid &= ~(LA_MTIME | LA_CTIME);
 531                 else if (attr->la_valid == LA_CTIME &&
 532                          attr->la_ctime == tmp_la->la_ctime)
 533                         attr->la_valid &= ~LA_CTIME;
 534         }
 535         RETURN(0);
 536 }
 537
 538 int mdd_attr_set_internal(const struct lu_env *env,
 539                           struct mdd_object *obj,
 540                           struct lu_attr *attr,
 541                           struct thandle *handle,
 542                           int needacl)
 543 {
 544         int rc;
 545         ENTRY;
 546
 547         rc = mdo_attr_set(env, obj, attr, handle, mdd_object_capa(env, obj));
 548 #ifdef CONFIG_FS_POSIX_ACL
 549         if (!rc && (attr->la_valid & LA_MODE) && needacl)
 550                 rc = mdd_acl_chmod(env, obj, attr->la_mode, handle);
 551 #endif
 552         RETURN(rc);
 553 }
 554
 555 int mdd_attr_check_set_internal(const struct lu_env *env,
 556                                 struct mdd_object *obj,
 557                                 struct lu_attr *attr,
 558                                 struct thandle *handle,
 559                                 int needacl)
 560 {
 561         int rc;
 562         ENTRY;
 563
 564         rc = mdd_attr_check(env, obj, attr);
 565         if (rc)
 566                 RETURN(rc);
 567
 568         if (attr->la_valid)
 569                 rc = mdd_attr_set_internal(env, obj, attr, handle, needacl);
 570         RETURN(rc);
 571 }
 572
 573 static int mdd_attr_set_internal_locked(const struct lu_env *env,
 574                                         struct mdd_object *obj,
 575                                         struct lu_attr *attr,
 576                                         struct thandle *handle,
 577                                         int needacl)
 578 {
 579         int rc;
 580         ENTRY;
 581
 582         needacl = needacl && (attr->la_valid & LA_MODE);
 583
 584         if (needacl)
 585                 mdd_write_lock(env, obj);
 586
 587         rc = mdd_attr_set_internal(env, obj, attr, handle, needacl);
 588
 589         if (needacl)
 590                 mdd_write_unlock(env, obj);
 591         RETURN(rc);
 592 }
 593
 594 int mdd_attr_check_set_internal_locked(const struct lu_env *env,
 595                                        struct mdd_object *obj,
 596                                        struct lu_attr *attr,
 597                                        struct thandle *handle,
 598                                        int needacl)
 599 {
 600         int rc;
 601         ENTRY;
 602
 603         rc = mdd_attr_check(env, obj, attr);
 604         if (rc)
 605                 RETURN(rc);
 606
 607         if (attr->la_valid)
 608                 rc = mdd_attr_set_internal_locked(env, obj, attr, handle,
 609                                                   needacl);
 610         RETURN(rc);
 611 }
 612
 613 static int __mdd_xattr_set(const struct lu_env *env, struct mdd_object *obj,
 614                            const struct lu_buf *buf, const char *name,
 615                            int fl, struct thandle *handle)
 616 {
 617         struct lustre_capa *capa = mdd_object_capa(env, obj);
 618         int rc = -EINVAL;
 619         ENTRY;
 620
 621         if (buf->lb_buf && buf->lb_len > 0)
 622                 rc = mdo_xattr_set(env, obj, buf, name, 0, handle, capa);
 623         else if (buf->lb_buf == NULL && buf->lb_len == 0)
 624                 rc = mdo_xattr_del(env, obj, name, handle, capa);
 625
 626         RETURN(rc);
 627 }
 628
 629 /*
 630  * This gives the same functionality as the code between
 631  * sys_chmod and inode_setattr
 632  * chown_common and inode_setattr
 633  * utimes and inode_setattr
 634  * This API is ported from mds_fix_attr but remove some unnecesssary stuff.
 635  */
 636 static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
 637                         struct lu_attr *la, const struct md_attr *ma)
 638 {
 639         struct lu_attr   *tmp_la     = &mdd_env_info(env)->mti_la;
 640         struct md_ucred  *uc         = md_ucred(env);
 641         int               rc;
 642         ENTRY;
 643
 644         if (!la->la_valid)
 645                 RETURN(0);
 646
 647         /* Do not permit change file type */
 648         if (la->la_valid & LA_TYPE)
 649                 RETURN(-EPERM);
 650
 651         /* They should not be processed by setattr */
 652         if (la->la_valid & (LA_NLINK | LA_RDEV | LA_BLKSIZE))
 653                 RETURN(-EPERM);
 654
 655         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 656         if (rc)
 657                 RETURN(rc);
 658
 659         /* This is only for set ctime when rename's source is on remote MDS. */
 660         if (unlikely(la->la_valid == LA_CTIME)) {
 661                 rc = mdd_may_delete(env, NULL, obj, (struct md_attr *)ma, 1, 0);
 662                 if (rc == 0 && la->la_ctime <= tmp_la->la_ctime)
 663                         la->la_valid &= ~LA_CTIME;
 664                 RETURN(rc);
 665         }
 666
 667         if (la->la_valid == LA_ATIME) {
 668                 /* This is atime only set for read atime update on close. */
 669                 if (la->la_atime <= tmp_la->la_atime +
 670                                     mdd_obj2mdd_dev(obj)->mdd_atime_diff)
 671                         la->la_valid &= ~LA_ATIME;
 672                 RETURN(0);
 673         }
 674
 675         /* Check if flags change. */
 676         if (la->la_valid & LA_FLAGS) {
 677                 unsigned int oldflags = 0;
 678                 unsigned int newflags = la->la_flags &
 679                                 (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL);
 680
 681                 if ((uc->mu_fsuid != tmp_la->la_uid) &&
 682                     !mdd_capable(uc, CAP_FOWNER))
 683                         RETURN(-EPERM);
 684
 685                 /* XXX: the IMMUTABLE and APPEND_ONLY flags can
 686                  * only be changed by the relevant capability. */
 687                 if (mdd_is_immutable(obj))
 688                         oldflags |= LUSTRE_IMMUTABLE_FL;
 689                 if (mdd_is_append(obj))
 690                         oldflags |= LUSTRE_APPEND_FL;
 691                 if ((oldflags ^ newflags) &&
 692                     !mdd_capable(uc, CAP_LINUX_IMMUTABLE))
 693                         RETURN(-EPERM);
 694
 695                 if (!S_ISDIR(tmp_la->la_mode))
 696                         la->la_flags &= ~LUSTRE_DIRSYNC_FL;
 697         }
 698
 699         if ((mdd_is_immutable(obj) || mdd_is_append(obj)) &&
 700             (la->la_valid & ~LA_FLAGS) &&
 701             !(ma->ma_attr_flags & MDS_PERM_BYPASS))
 702                 RETURN(-EPERM);
 703
 704         /* Check for setting the obj time. */
 705         if ((la->la_valid & (LA_MTIME | LA_ATIME | LA_CTIME)) &&
 706             !(la->la_valid & ~(LA_MTIME | LA_ATIME | LA_CTIME))) {
 707                 if ((uc->mu_fsuid != tmp_la->la_uid) &&
 708                     !mdd_capable(uc, CAP_FOWNER)) {
 709                         rc = mdd_permission_internal_locked(env, obj, tmp_la,
 710                                                             MAY_WRITE);
 711                         if (rc)
 712                                 RETURN(rc);
 713                 }
 714         }
 715
 716         /* Make sure a caller can chmod. */
 717         if (la->la_valid & LA_MODE) {
 718                 /*
 719                  * Bypass la_vaild == LA_MODE,
 720                  * this is for changing file with SUID or SGID.
 721                  */
 722                 if ((la->la_valid & ~LA_MODE) &&
 723                     (uc->mu_fsuid != tmp_la->la_uid) &&
 724                     !mdd_capable(uc, CAP_FOWNER))
 725                         RETURN(-EPERM);
 726
 727                 if (la->la_mode == (umode_t) -1)
 728                         la->la_mode = tmp_la->la_mode;
 729                 else
 730                         la->la_mode = (la->la_mode & S_IALLUGO) |
 731                                       (tmp_la->la_mode & ~S_IALLUGO);
 732
 733                 /* Also check the setgid bit! */
 734                 if (!lustre_in_group_p(uc, (la->la_valid & LA_GID) ? la->la_gid :
 735                                 tmp_la->la_gid) && !mdd_capable(uc, CAP_FSETID))
 736                         la->la_mode &= ~S_ISGID;
 737         } else {
 738                la->la_mode = tmp_la->la_mode;
 739         }
 740
 741         /* Make sure a caller can chown. */
 742         if (la->la_valid & LA_UID) {
 743                 if (la->la_uid == (uid_t) -1)
 744                         la->la_uid = tmp_la->la_uid;
 745                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
 746                     (la->la_uid != tmp_la->la_uid)) &&
 747                     !mdd_capable(uc, CAP_CHOWN))
 748                         RETURN(-EPERM);
 749
 750                 /*
 751                  * If the user or group of a non-directory has been
 752                  * changed by a non-root user, remove the setuid bit.
 753                  * 19981026 David C Niemi <niemi@tux.org>
 754                  *
 755                  * Changed this to apply to all users, including root,
 756                  * to avoid some races. This is the behavior we had in
 757                  * 2.0. The check for non-root was definitely wrong
 758                  * for 2.2 anyway, as it should have been using
 759                  * CAP_FSETID rather than fsuid -- 19990830 SD.
 760                  */
 761                 if (((tmp_la->la_mode & S_ISUID) == S_ISUID) &&
 762                     !S_ISDIR(tmp_la->la_mode)) {
 763                         la->la_mode &= ~S_ISUID;
 764                         la->la_valid |= LA_MODE;
 765                 }
 766         }
 767
 768         /* Make sure caller can chgrp. */
 769         if (la->la_valid & LA_GID) {
 770                 if (la->la_gid == (gid_t) -1)
 771                         la->la_gid = tmp_la->la_gid;
 772                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
 773                     ((la->la_gid != tmp_la->la_gid) &&
 774                     !lustre_in_group_p(uc, la->la_gid))) &&
 775                     !mdd_capable(uc, CAP_CHOWN))
 776                         RETURN(-EPERM);
 777
 778                 /*
 779                  * Likewise, if the user or group of a non-directory
 780                  * has been changed by a non-root user, remove the
 781                  * setgid bit UNLESS there is no group execute bit
 782                  * (this would be a file marked for mandatory
 783                  * locking).  19981026 David C Niemi <niemi@tux.org>
 784                  *
 785                  * Removed the fsuid check (see the comment above) --
 786                  * 19990830 SD.
 787                  */
 788                 if (((tmp_la->la_mode & (S_ISGID | S_IXGRP)) ==
 789                      (S_ISGID | S_IXGRP)) && !S_ISDIR(tmp_la->la_mode)) {
 790                         la->la_mode &= ~S_ISGID;
 791                         la->la_valid |= LA_MODE;
 792                 }
 793         }
 794
 795         if (la->la_valid & (LA_SIZE | LA_BLOCKS) &&
 796             !(la->la_valid & LA_TRUNC)) {
 797                 /* For "Size-on-MDS" case, the MAY_WRITE perm
 798                  * has been checked when file open. */
 799
 800                 /* For the "Size-on-MDS" setattr update, merge coming
 801                  * attributes with the set in the inode. BUG 10641 */
 802                 if ((la->la_valid & LA_ATIME) &&
 803                     (la->la_atime <= tmp_la->la_atime))
 804                         la->la_valid &= ~LA_ATIME;
 805
 806                 /* OST attributes do not have a priority over MDS attributes,
 807                  * so drop times if ctime is equal. */
 808                 if ((la->la_valid & LA_CTIME) &&
 809                     (la->la_ctime <= tmp_la->la_ctime))
 810                         la->la_valid &= ~(LA_MTIME | LA_CTIME);
 811         } else {
 812                 if (la->la_valid & LA_TRUNC) {
 813                         /* For truncate, we should have MAY_WRITE perm. */
 814                         if (!((la->la_valid & MDS_OPEN_OWNEROVERRIDE) &&
 815                               (uc->mu_fsuid == tmp_la->la_uid)) &&
 816                             !(ma->ma_attr_flags & MDS_PERM_BYPASS)) {
 817                                 rc = mdd_permission_internal_locked(env, obj,
 818                                                             tmp_la, MAY_WRITE);
 819                                 if (rc)
 820                                         RETURN(rc);
 821                         }
 822                 }
 823                 if (la->la_valid & LA_CTIME) {
 824                         /* The pure setattr, it has the priority over what is
 825                          * already set, do not drop it if ctime is equal. */
 826                         if (la->la_ctime < tmp_la->la_ctime)
 827                                 la->la_valid &= ~(LA_ATIME | LA_MTIME |
 828                                                   LA_CTIME);
 829                 }
 830         }
 831
 832         RETURN(0);
 833 }
 834
 835 /* set attr and LOV EA at once, return updated attr */
 836 static int mdd_attr_set(const struct lu_env *env, struct md_object *obj,
 837                         const struct md_attr *ma)
 838 {
 839         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 840         struct mdd_device *mdd = mdo2mdd(obj);
 841         struct thandle *handle;
 842         struct lov_mds_md *lmm = NULL;
 843         struct llog_cookie *logcookies = NULL;
 844         int  rc, lmm_size = 0, cookie_size = 0;
 845         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
 846         ENTRY;
 847
 848         mdd_txn_param_build(env, mdd, MDD_TXN_ATTR_SET_OP);
 849         handle = mdd_trans_start(env, mdd);
 850         if (IS_ERR(handle))
 851                 RETURN(PTR_ERR(handle));
 852         /*TODO: add lock here*/
 853         /* start a log jounal handle if needed */
 854         if (S_ISREG(mdd_object_type(mdd_obj)) &&
 855             ma->ma_attr.la_valid & (LA_UID | LA_GID)) {
 856                 lmm_size = mdd_lov_mdsize(env, mdd);
 857                 lmm = mdd_max_lmm_get(env, mdd);
 858                 if (lmm == NULL)
 859                         GOTO(cleanup, rc = -ENOMEM);
 860
 861                 rc = mdd_get_md_locked(env, mdd_obj, lmm, &lmm_size,
 862                                 MDS_LOV_MD_NAME);
 863
 864                 if (rc < 0)
 865                         GOTO(cleanup, rc);
 866         }
 867
 868         if (ma->ma_attr.la_valid & (ATTR_MTIME | ATTR_CTIME))
 869                 CDEBUG(D_INODE, "setting mtime "LPU64", ctime "LPU64"\n",
 870                        ma->ma_attr.la_mtime, ma->ma_attr.la_ctime);
 871
 872         *la_copy = ma->ma_attr;
 873         rc = mdd_fix_attr(env, mdd_obj, la_copy, ma);
 874         if (rc)
 875                 GOTO(cleanup, rc);
 876
 877         if (la_copy->la_valid & LA_FLAGS) {
 878                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
 879                                                   handle, 1);
 880                 if (rc == 0)
 881                         mdd_flags_xlate(mdd_obj, la_copy->la_flags);
 882         } else if (la_copy->la_valid) {            /* setattr */
 883                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
 884                                                   handle, 1);
 885                 /* journal chown/chgrp in llog, just like unlink */
 886                 if (rc == 0 && lmm_size){
 887                         cookie_size = mdd_lov_cookiesize(env, mdd);
 888                         logcookies = mdd_max_cookie_get(env, mdd);
 889                         if (logcookies == NULL)
 890                                 GOTO(cleanup, rc = -ENOMEM);
 891
 892                         if (mdd_setattr_log(env, mdd, ma, lmm, lmm_size,
 893                                             logcookies, cookie_size) <= 0)
 894                                 logcookies = NULL;
 895                 }
 896         }
 897
 898         if (rc == 0 && ma->ma_valid & MA_LOV) {
 899                 umode_t mode;
 900
 901                 mode = mdd_object_type(mdd_obj);
 902                 if (S_ISREG(mode) || S_ISDIR(mode)) {
 903                         rc = mdd_lsm_sanity_check(env, mdd_obj);
 904                         if (rc)
 905                                 GOTO(cleanup, rc);
 906
 907                         rc = mdd_lov_set_md(env, NULL, mdd_obj, ma->ma_lmm,
 908                                             ma->ma_lmm_size, handle, 1);
 909                 }
 910
 911         }
 912 cleanup:
 913         mdd_trans_stop(env, mdd, rc, handle);
 914         if (rc == 0 && (lmm != NULL && lmm_size > 0 )) {
 915                 /*set obd attr, if needed*/
 916                 rc = mdd_lov_setattr_async(env, mdd_obj, lmm, lmm_size,
 917                                            logcookies);
 918         }
 919         RETURN(rc);
 920 }
 921
 922 int mdd_xattr_set_txn(const struct lu_env *env, struct mdd_object *obj,
 923                       const struct lu_buf *buf, const char *name, int fl,
 924                       struct thandle *handle)
 925 {
 926         int  rc;
 927         ENTRY;
 928
 929         mdd_write_lock(env, obj);
 930         rc = __mdd_xattr_set(env, obj, buf, name, fl, handle);
 931         mdd_write_unlock(env, obj);
 932
 933         RETURN(rc);
 934 }
 935
 936 static int mdd_xattr_sanity_check(const struct lu_env *env,
 937                                   struct mdd_object *obj)
 938 {
 939         struct lu_attr  *tmp_la = &mdd_env_info(env)->mti_la;
 940         struct md_ucred *uc     = md_ucred(env);
 941         int rc;
 942         ENTRY;
 943
 944         if (mdd_is_immutable(obj) || mdd_is_append(obj))
 945                 RETURN(-EPERM);
 946
 947         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 948         if (rc)
 949                 RETURN(rc);
 950
 951         if ((uc->mu_fsuid != tmp_la->la_uid) && !mdd_capable(uc, CAP_FOWNER))
 952                 RETURN(-EPERM);
 953
 954         RETURN(rc);
 955 }
 956
 957 static int mdd_xattr_set(const struct lu_env *env, struct md_object *obj,
 958                          const struct lu_buf *buf, const char *name,
 959                          int fl, const struct lu_attr *la)
 960 {
 961         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
 962         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 963         struct mdd_device *mdd = mdo2mdd(obj);
 964         struct thandle *handle;
 965         int  rc;
 966         ENTRY;
 967
 968         rc = mdd_xattr_sanity_check(env, mdd_obj);
 969         if (rc)
 970                 RETURN(rc);
 971
 972         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
 973         handle = mdd_trans_start(env, mdd);
 974         if (IS_ERR(handle))
 975                 RETURN(PTR_ERR(handle));
 976
 977         rc = mdd_xattr_set_txn(env, mdd_obj, buf, name, fl, handle);
 978         if (rc == 0 && likely(la != NULL)) {
 979                 LASSERT(la->la_valid & LA_CTIME);
 980                 la_copy->la_ctime = la->la_ctime;
 981                 la_copy->la_valid = LA_CTIME;
 982                 rc = mdd_attr_check_set_internal_locked(env, mdd_obj, la_copy,
 983                                                         handle, 0);
 984         }
 985         mdd_trans_stop(env, mdd, rc, handle);
 986
 987         RETURN(rc);
 988 }
 989
 990 int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
 991                   const char *name, const struct lu_attr *la)
 992 {
 993         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
 994         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 995         struct mdd_device *mdd = mdo2mdd(obj);
 996         struct thandle *handle;
 997         int  rc;
 998         ENTRY;
 999
1000         rc = mdd_xattr_sanity_check(env, mdd_obj);
1001         if (rc)
1002                 RETURN(rc);
1003
1004         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
1005         handle = mdd_trans_start(env, mdd);
1006         if (IS_ERR(handle))
1007                 RETURN(PTR_ERR(handle));
1008
1009         mdd_write_lock(env, mdd_obj);
1010         rc = mdo_xattr_del(env, mdd_obj, name, handle,
1011                            mdd_object_capa(env, mdd_obj));
1012         mdd_write_unlock(env, mdd_obj);
1013         if (rc == 0 && likely(la != NULL)) {
1014                 LASSERT(la->la_valid & LA_CTIME);
1015                 la_copy->la_ctime = la->la_ctime;
1016                 la_copy->la_valid = LA_CTIME;
1017                 rc = mdd_attr_check_set_internal_locked(env, mdd_obj, la_copy,
1018                                                         handle, 0);
1019         }
1020
1021         mdd_trans_stop(env, mdd, rc, handle);
1022
1023         RETURN(rc);
1024 }
1025
1026 /* partial unlink */
1027 static int mdd_ref_del(const struct lu_env *env, struct md_object *obj,
1028                        struct md_attr *ma)
1029 {
1030         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
1031         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1032         struct mdd_device *mdd = mdo2mdd(obj);
1033         struct thandle *handle;
1034         int rc;
1035         ENTRY;
1036
1037         /*
1038          * Check -ENOENT early here because we need to get object type
1039          * to calculate credits before transaction start
1040          */
1041         if (!mdd_object_exists(mdd_obj))
1042                 RETURN(-ENOENT);
1043
1044         LASSERT(mdd_object_exists(mdd_obj) > 0);
1045
1046         rc = mdd_log_txn_param_build(env, obj, ma, MDD_TXN_UNLINK_OP);
1047         if (rc)
1048                 RETURN(rc);
1049
1050         handle = mdd_trans_start(env, mdd);
1051         if (IS_ERR(handle))
1052                 RETURN(-ENOMEM);
1053
1054         mdd_write_lock(env, mdd_obj);
1055
1056         rc = mdd_unlink_sanity_check(env, NULL, mdd_obj, ma);
1057         if (rc)
1058                 GOTO(cleanup, rc);
1059
1060         __mdd_ref_del(env, mdd_obj, handle, 0);
1061
1062         if (S_ISDIR(lu_object_attr(&obj->mo_lu))) {
1063                 /* unlink dot */
1064                 __mdd_ref_del(env, mdd_obj, handle, 1);
1065         }
1066
1067         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1068         la_copy->la_ctime = ma->ma_attr.la_ctime;
1069
1070         la_copy->la_valid = LA_CTIME;
1071         rc = mdd_attr_check_set_internal(env, mdd_obj, la_copy, handle, 0);
1072         if (rc)
1073                 GOTO(cleanup, rc);
1074
1075         rc = mdd_finish_unlink(env, mdd_obj, ma, handle);
1076
1077         EXIT;
1078 cleanup:
1079         mdd_write_unlock(env, mdd_obj);
1080         mdd_trans_stop(env, mdd, rc, handle);
1081         return rc;
1082 }
1083
1084 /* partial operation */
1085 static int mdd_oc_sanity_check(const struct lu_env *env,
1086                                struct mdd_object *obj,
1087                                struct md_attr *ma)
1088 {
1089         int rc;
1090         ENTRY;
1091
1092         switch (ma->ma_attr.la_mode & S_IFMT) {
1093         case S_IFREG:
1094         case S_IFDIR:
1095         case S_IFLNK:
1096         case S_IFCHR:
1097         case S_IFBLK:
1098         case S_IFIFO:
1099         case S_IFSOCK:
1100                 rc = 0;
1101                 break;
1102         default:
1103                 rc = -EINVAL;
1104                 break;
1105         }
1106         RETURN(rc);
1107 }
1108
1109 static int mdd_object_create(const struct lu_env *env,
1110                              struct md_object *obj,
1111                              const struct md_op_spec *spec,
1112                              struct md_attr *ma)
1113 {
1114
1115         struct mdd_device *mdd = mdo2mdd(obj);
1116         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1117         const struct lu_fid *pfid = spec->u.sp_pfid;
1118         struct thandle *handle;
1119         int rc;
1120         ENTRY;
1121
1122         mdd_txn_param_build(env, mdd, MDD_TXN_OBJECT_CREATE_OP);
1123         handle = mdd_trans_start(env, mdd);
1124         if (IS_ERR(handle))
1125                 RETURN(PTR_ERR(handle));
1126
1127         mdd_write_lock(env, mdd_obj);
1128         rc = mdd_oc_sanity_check(env, mdd_obj, ma);
1129         if (rc)
1130                 GOTO(unlock, rc);
1131
1132         rc = mdd_object_create_internal(env, NULL, mdd_obj, ma, handle);
1133         if (rc)
1134                 GOTO(unlock, rc);
1135
1136         if (spec->sp_cr_flags & MDS_CREATE_SLAVE_OBJ) {
1137                 /* If creating the slave object, set slave EA here. */
1138                 int lmv_size = spec->u.sp_ea.eadatalen;
1139                 struct lmv_stripe_md *lmv;
1140
1141                 lmv = (struct lmv_stripe_md *)spec->u.sp_ea.eadata;
1142                 LASSERT(lmv != NULL && lmv_size > 0);
1143
1144                 rc = __mdd_xattr_set(env, mdd_obj,
1145                                      mdd_buf_get_const(env, lmv, lmv_size),
1146                                      MDS_LMV_MD_NAME, 0, handle);
1147                 if (rc)
1148                         GOTO(unlock, rc);
1149
1150                 rc = mdd_attr_set_internal(env, mdd_obj, &ma->ma_attr,
1151                                            handle, 0);
1152         } else {
1153 #ifdef CONFIG_FS_POSIX_ACL
1154                 if (spec->sp_cr_flags & MDS_CREATE_RMT_ACL) {
1155                         struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
1156
1157                         buf->lb_buf = (void *)spec->u.sp_ea.eadata;
1158                         buf->lb_len = spec->u.sp_ea.eadatalen;
1159                         if ((buf->lb_len > 0) && (buf->lb_buf != NULL)) {
1160                                 rc = __mdd_acl_init(env, mdd_obj, buf,
1161                                                     &ma->ma_attr.la_mode,
1162                                                     handle);
1163                                 if (rc)
1164                                         GOTO(unlock, rc);
1165                                 else
1166                                         ma->ma_attr.la_valid |= LA_MODE;
1167                         }
1168
1169                         pfid = spec->u.sp_ea.fid;
1170                 }
1171 #endif
1172                 rc = mdd_object_initialize(env, pfid, mdd_obj, ma, handle);
1173         }
1174         EXIT;
1175 unlock:
1176         mdd_write_unlock(env, mdd_obj);
1177         if (rc == 0)
1178                 rc = mdd_attr_get_internal_locked(env, mdd_obj, ma);
1179
1180         mdd_trans_stop(env, mdd, rc, handle);
1181         return rc;
1182 }
1183
1184 /* partial link */
1185 static int mdd_ref_add(const struct lu_env *env, struct md_object *obj,
1186                        const struct md_attr *ma)
1187 {
1188         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
1189         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1190         struct mdd_device *mdd = mdo2mdd(obj);
1191         struct thandle *handle;
1192         int rc;
1193         ENTRY;
1194
1195         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
1196         handle = mdd_trans_start(env, mdd);
1197         if (IS_ERR(handle))
1198                 RETURN(-ENOMEM);
1199
1200         mdd_write_lock(env, mdd_obj);
1201         rc = mdd_link_sanity_check(env, NULL, NULL, mdd_obj);
1202         if (rc == 0)
1203                 __mdd_ref_add(env, mdd_obj, handle);
1204         mdd_write_unlock(env, mdd_obj);
1205         if (rc == 0) {
1206                 LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1207                 la_copy->la_ctime = ma->ma_attr.la_ctime;
1208
1209                 la_copy->la_valid = LA_CTIME;
1210                 rc = mdd_attr_check_set_internal_locked(env, mdd_obj, la_copy,
1211                                                         handle, 0);
1212         }
1213         mdd_trans_stop(env, mdd, 0, handle);
1214
1215         RETURN(rc);
1216 }
1217
1218 /*
1219  * do NOT or the MAY_*'s, you'll get the weakest
1220  */
1221 int accmode(const struct lu_env *env, struct lu_attr *la, int flags)
1222 {
1223         int res = 0;
1224
1225         /* Sadly, NFSD reopens a file repeatedly during operation, so the
1226          * "acc_mode = 0" allowance for newly-created files isn't honoured.
1227          * NFSD uses the MDS_OPEN_OWNEROVERRIDE flag to say that a file
1228          * owner can write to a file even if it is marked readonly to hide
1229          * its brokenness. (bug 5781) */
1230         if (flags & MDS_OPEN_OWNEROVERRIDE) {
1231                 struct md_ucred *uc = md_ucred(env);
1232
1233                 if ((uc == NULL) || (uc->mu_valid == UCRED_INIT) ||
1234                     (la->la_uid == uc->mu_fsuid))
1235                         return 0;
1236         }
1237
1238         if (flags & FMODE_READ)
1239                 res |= MAY_READ;
1240         if (flags & (FMODE_WRITE | MDS_OPEN_TRUNC | MDS_OPEN_APPEND))
1241                 res |= MAY_WRITE;
1242         if (flags & MDS_FMODE_EXEC)
1243                 res |= MAY_EXEC;
1244         return res;
1245 }
1246
1247 static int mdd_open_sanity_check(const struct lu_env *env,
1248                                  struct mdd_object *obj, int flag)
1249 {
1250         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
1251         int mode, rc;
1252         ENTRY;
1253
1254         /* EEXIST check */
1255         if (mdd_is_dead_obj(obj))
1256                 RETURN(-ENOENT);
1257
1258         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
1259         if (rc)
1260                RETURN(rc);
1261
1262         if (S_ISLNK(tmp_la->la_mode))
1263                 RETURN(-ELOOP);
1264
1265         mode = accmode(env, tmp_la, flag);
1266
1267         if (S_ISDIR(tmp_la->la_mode) && (mode & MAY_WRITE))
1268                 RETURN(-EISDIR);
1269
1270         if (!(flag & MDS_OPEN_CREATED)) {
1271                 rc = mdd_permission_internal(env, obj, tmp_la, mode);
1272                 if (rc)
1273                         RETURN(rc);
1274         }
1275
1276         if (S_ISFIFO(tmp_la->la_mode) || S_ISSOCK(tmp_la->la_mode) ||
1277             S_ISBLK(tmp_la->la_mode) || S_ISCHR(tmp_la->la_mode))
1278                 flag &= ~MDS_OPEN_TRUNC;
1279
1280         /* For writing append-only file must open it with append mode. */
1281         if (mdd_is_append(obj)) {
1282                 if ((flag & FMODE_WRITE) && !(flag & MDS_OPEN_APPEND))
1283                         RETURN(-EPERM);
1284                 if (flag & MDS_OPEN_TRUNC)
1285                         RETURN(-EPERM);
1286         }
1287
1288 #if 0
1289         /*
1290          * Now, flag -- O_NOATIME does not be packed by client.
1291          */
1292         if (flag & O_NOATIME) {
1293                 struct md_ucred *uc = md_ucred(env);
1294
1295                 if (uc && ((uc->mu_valid == UCRED_OLD) ||
1296                     (uc->mu_valid == UCRED_NEW)) &&
1297                     (uc->mu_fsuid != tmp_la->la_uid) &&
1298                     !mdd_capable(uc, CAP_FOWNER))
1299                         RETURN(-EPERM);
1300         }
1301 #endif
1302
1303         RETURN(0);
1304 }
1305
1306 static int mdd_open(const struct lu_env *env, struct md_object *obj,
1307                     int flags)
1308 {
1309         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1310         int rc = 0;
1311
1312         mdd_write_lock(env, mdd_obj);
1313
1314         rc = mdd_open_sanity_check(env, mdd_obj, flags);
1315         if (rc == 0)
1316                 mdd_obj->mod_count++;
1317
1318         mdd_write_unlock(env, mdd_obj);
1319         return rc;
1320 }
1321
1322 /* return md_attr back,
1323  * if it is last unlink then return lov ea + llog cookie*/
1324 int mdd_object_kill(const struct lu_env *env, struct mdd_object *obj,
1325                     struct md_attr *ma)
1326 {
1327         int rc = 0;
1328         ENTRY;
1329
1330         if (S_ISREG(mdd_object_type(obj))) {
1331                 /* Return LOV & COOKIES unconditionally here. We clean evth up.
1332                  * Caller must be ready for that. */
1333                 rc = __mdd_lmm_get(env, obj, ma);
1334                 if ((ma->ma_valid & MA_LOV))
1335                         rc = mdd_unlink_log(env, mdo2mdd(&obj->mod_obj),
1336                                             obj, ma);
1337         }
1338         RETURN(rc);
1339 }
1340
1341 /*
1342  * No permission check is needed.
1343  */
1344 static int mdd_close(const struct lu_env *env, struct md_object *obj,
1345                      struct md_attr *ma)
1346 {
1347         int rc;
1348         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1349         struct thandle    *handle;
1350         ENTRY;
1351
1352         rc = mdd_log_txn_param_build(env, obj, ma, MDD_TXN_UNLINK_OP);
1353         if (rc)
1354                 RETURN(rc);
1355         handle = mdd_trans_start(env, mdo2mdd(obj));
1356         if (IS_ERR(handle))
1357                 RETURN(PTR_ERR(handle));
1358
1359         mdd_write_lock(env, mdd_obj);
1360         /* release open count */
1361         mdd_obj->mod_count --;
1362
1363         rc = mdd_iattr_get(env, mdd_obj, ma);
1364         if (rc == 0 && mdd_obj->mod_count == 0 && ma->ma_attr.la_nlink == 0)
1365                 rc = mdd_object_kill(env, mdd_obj, ma);
1366         else
1367                 ma->ma_valid &= ~(MA_LOV | MA_COOKIE);
1368
1369         mdd_write_unlock(env, mdd_obj);
1370         mdd_trans_stop(env, mdo2mdd(obj), rc, handle);
1371         RETURN(rc);
1372 }
1373
1374 /*
1375  * Permission check is done when open,
1376  * no need check again.
1377  */
1378 static int mdd_readpage_sanity_check(const struct lu_env *env,
1379                                      struct mdd_object *obj)
1380 {
1381         struct dt_object *next = mdd_object_child(obj);
1382         int rc;
1383         ENTRY;
1384
1385         if (S_ISDIR(mdd_object_type(obj)) && dt_try_as_dir(env, next))
1386                 rc = 0;
1387         else
1388                 rc = -ENOTDIR;
1389
1390         RETURN(rc);
1391 }
1392
1393 static int mdd_dir_page_build(const struct lu_env *env, int first,
1394                               void *area, int nob, struct dt_it_ops *iops,
1395                               struct dt_it *it, __u64 *start, __u64 *end,
1396                               struct lu_dirent **last)
1397 {
1398         struct lu_fid          *fid  = &mdd_env_info(env)->mti_fid2;
1399         struct mdd_thread_info *info = mdd_env_info(env);
1400         struct lu_fid_pack     *pack = &info->mti_pack;
1401         int                     result;
1402         struct lu_dirent       *ent;
1403
1404         if (first) {
1405                 memset(area, 0, sizeof (struct lu_dirpage));
1406                 area += sizeof (struct lu_dirpage);
1407                 nob  -= sizeof (struct lu_dirpage);
1408         }
1409
1410         LASSERT(nob > sizeof *ent);
1411
1412         ent  = area;
1413         result = 0;
1414         do {
1415                 char  *name;
1416                 int    len;
1417                 int    recsize;
1418                 __u64  hash;
1419
1420                 name = (char *)iops->key(env, it);
1421                 len  = iops->key_size(env, it);
1422
1423                 pack = (struct lu_fid_pack *)iops->rec(env, it);
1424                 result = fid_unpack(pack, fid);
1425                 if (result != 0)
1426                         break;
1427
1428                 recsize = (sizeof(*ent) + len + 7) & ~7;
1429                 hash = iops->store(env, it);
1430                 *end = hash;
1431
1432                 CDEBUG(D_INFO, "%p %p %d "DFID": "LPU64" (%d) \"%*.*s\"\n",
1433                        name, ent, nob, PFID(fid), hash, len, len, len, name);
1434
1435                 if (nob >= recsize) {
1436                         ent->lde_fid = *fid;
1437                         fid_cpu_to_le(&ent->lde_fid, &ent->lde_fid);
1438                         ent->lde_hash = hash;
1439                         ent->lde_namelen = cpu_to_le16(len);
1440                         ent->lde_reclen  = cpu_to_le16(recsize);
1441                         memcpy(ent->lde_name, name, len);
1442                         if (first && ent == area)
1443                                 *start = hash;
1444                         *last = ent;
1445                         ent = (void *)ent + recsize;
1446                         nob -= recsize;
1447                         result = iops->next(env, it);
1448                 } else {
1449                         /*
1450                          * record doesn't fit into page, enlarge previous one.
1451                          */
1452                         LASSERT(*last != NULL);
1453                         (*last)->lde_reclen =
1454                                 cpu_to_le16(le16_to_cpu((*last)->lde_reclen) +
1455                                             nob);
1456                         break;
1457                 }
1458         } while (result == 0);
1459
1460         return result;
1461 }
1462
1463 static int __mdd_readpage(const struct lu_env *env, struct mdd_object *obj,
1464                           const struct lu_rdpg *rdpg)
1465 {
1466         struct dt_it      *it;
1467         struct dt_object  *next = mdd_object_child(obj);
1468         struct dt_it_ops  *iops;
1469         struct page       *pg;
1470         struct lu_dirent  *last;
1471         int i;
1472         int rc;
1473         int nob;
1474         __u64 hash_start;
1475         __u64 hash_end;
1476
1477         LASSERT(rdpg->rp_pages != NULL);
1478         LASSERT(next->do_index_ops != NULL);
1479
1480         if (rdpg->rp_count <= 0)
1481                 return -EFAULT;
1482
1483         /*
1484          * iterate through directory and fill pages from @rdpg
1485          */
1486         iops = &next->do_index_ops->dio_it;
1487         it = iops->init(env, next, 0, mdd_object_capa(env, obj));
1488         if (IS_ERR(it))
1489                 return PTR_ERR(it);
1490
1491         rc = iops->load(env, it, rdpg->rp_hash);
1492
1493         if (rc == 0)
1494                 /*
1495                  * Iterator didn't find record with exactly the key requested.
1496                  *
1497                  * It is currently either
1498                  *
1499                  *     - positioned above record with key less than
1500                  *     requested---skip it.
1501                  *
1502                  *     - or not positioned at all (is in IAM_IT_SKEWED
1503                  *     state)---position it on the next item.
1504                  */
1505                 rc = iops->next(env, it);
1506         else if (rc > 0)
1507                 rc = 0;
1508
1509         /*
1510          * At this point and across for-loop:
1511          *
1512          *  rc == 0 -> ok, proceed.
1513          *  rc >  0 -> end of directory.
1514          *  rc <  0 -> error.
1515          */
1516         for (i = 0, nob = rdpg->rp_count; rc == 0 && nob > 0;
1517              i++, nob -= CFS_PAGE_SIZE) {
1518                 LASSERT(i < rdpg->rp_npages);
1519                 pg = rdpg->rp_pages[i];
1520                 rc = mdd_dir_page_build(env, !i, cfs_kmap(pg),
1521                                         min_t(int, nob, CFS_PAGE_SIZE), iops,
1522                                         it, &hash_start, &hash_end, &last);
1523                 if (rc != 0 || i == rdpg->rp_npages - 1)
1524                         last->lde_reclen = 0;
1525                 cfs_kunmap(pg);
1526         }
1527         if (rc > 0) {
1528                 /*
1529                  * end of directory.
1530                  */
1531                 hash_end = DIR_END_OFF;
1532                 rc = 0;
1533         }
1534         if (rc == 0) {
1535                 struct lu_dirpage *dp;
1536
1537                 dp = cfs_kmap(rdpg->rp_pages[0]);
1538                 dp->ldp_hash_start = rdpg->rp_hash;
1539                 dp->ldp_hash_end   = hash_end;
1540                 if (i == 0)
1541                         /*
1542                          * No pages were processed, mark this.
1543                          */
1544                         dp->ldp_flags |= LDF_EMPTY;
1545                 dp->ldp_flags = cpu_to_le32(dp->ldp_flags);
1546                 cfs_kunmap(rdpg->rp_pages[0]);
1547         }
1548         iops->put(env, it);
1549         iops->fini(env, it);
1550
1551         return rc;
1552 }
1553
1554 static int mdd_readpage(const struct lu_env *env, struct md_object *obj,
1555                         const struct lu_rdpg *rdpg)
1556 {
1557         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1558         int rc;
1559         ENTRY;
1560
1561         LASSERT(mdd_object_exists(mdd_obj));
1562
1563         mdd_read_lock(env, mdd_obj);
1564         rc = mdd_readpage_sanity_check(env, mdd_obj);
1565         if (rc)
1566                 GOTO(out_unlock, rc);
1567
1568         if (mdd_is_dead_obj(mdd_obj)) {
1569                 struct page *pg;
1570                 struct lu_dirpage *dp;
1571
1572                 /*
1573                  * According to POSIX, please do not return any entry to client:
1574                  * even dot and dotdot should not be returned.
1575                  */
1576                 CWARN("readdir from dead object: "DFID"\n",
1577                         PFID(mdd_object_fid(mdd_obj)));
1578
1579                 if (rdpg->rp_count <= 0)
1580                         GOTO(out_unlock, rc = -EFAULT);
1581                 LASSERT(rdpg->rp_pages != NULL);
1582
1583                 pg = rdpg->rp_pages[0];
1584                 dp = (struct lu_dirpage*)cfs_kmap(pg);
1585                 memset(dp, 0 , sizeof(struct lu_dirpage));
1586                 dp->ldp_hash_start = rdpg->rp_hash;
1587                 dp->ldp_hash_end   = DIR_END_OFF;
1588                 dp->ldp_flags |= LDF_EMPTY;
1589                 dp->ldp_flags = cpu_to_le32(dp->ldp_flags);
1590                 cfs_kunmap(pg);
1591                 GOTO(out_unlock, rc = 0);
1592         }
1593
1594         rc = __mdd_readpage(env, mdd_obj, rdpg);
1595
1596         EXIT;
1597 out_unlock:
1598         mdd_read_unlock(env, mdd_obj);
1599         return rc;
1600 }
1601
1602 struct md_object_operations mdd_obj_ops = {
1603         .moo_permission    = mdd_permission,
1604         .moo_attr_get      = mdd_attr_get,
1605         .moo_attr_set      = mdd_attr_set,
1606         .moo_xattr_get     = mdd_xattr_get,
1607         .moo_xattr_set     = mdd_xattr_set,
1608         .moo_xattr_list    = mdd_xattr_list,
1609         .moo_xattr_del     = mdd_xattr_del,
1610         .moo_object_create = mdd_object_create,
1611         .moo_ref_add       = mdd_ref_add,
1612         .moo_ref_del       = mdd_ref_del,
1613         .moo_open          = mdd_open,
1614         .moo_close         = mdd_close,
1615         .moo_readpage      = mdd_readpage,
1616         .moo_readlink      = mdd_readlink,
1617         .moo_capa_get      = mdd_capa_get
1618 };