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         RETURN(rc);
 269 }
 270
 271 /* get only inode attributes */
 272 int mdd_iattr_get(const struct lu_env *env, struct mdd_object *mdd_obj,
 273                   struct md_attr *ma)
 274 {
 275         int rc = 0;
 276         ENTRY;
 277
 278         if (ma->ma_valid & MA_INODE)
 279                 RETURN(0);
 280
 281         rc = mdd_la_get(env, mdd_obj, &ma->ma_attr,
 282                           mdd_object_capa(env, mdd_obj));
 283         if (rc == 0)
 284                 ma->ma_valid |= MA_INODE;
 285         RETURN(rc);
 286 }
 287
 288 static int mdd_get_default_md(struct mdd_object *mdd_obj,
 289                 struct lov_mds_md *lmm, int *size)
 290 {
 291         struct lov_desc *ldesc;
 292         struct mdd_device *mdd = mdo2mdd(&mdd_obj->mod_obj);
 293         ENTRY;
 294
 295         ldesc = &mdd->mdd_obd_dev->u.mds.mds_lov_desc;
 296         LASSERT(ldesc != NULL);
 297
 298         if (!lmm)
 299                 RETURN(0);
 300
 301         lmm->lmm_magic = LOV_MAGIC_V1;
 302         lmm->lmm_object_gr = LOV_OBJECT_GROUP_DEFAULT;
 303         lmm->lmm_pattern = ldesc->ld_pattern;
 304         lmm->lmm_stripe_size = ldesc->ld_default_stripe_size;
 305         lmm->lmm_stripe_count = ldesc->ld_default_stripe_count;
 306         *size = sizeof(struct lov_mds_md);
 307
 308         RETURN(sizeof(struct lov_mds_md));
 309 }
 310
 311 /* get lov EA only */
 312 static int __mdd_lmm_get(const struct lu_env *env,
 313                          struct mdd_object *mdd_obj, struct md_attr *ma)
 314 {
 315         int rc;
 316         ENTRY;
 317
 318         if (ma->ma_valid & MA_LOV)
 319                 RETURN(0);
 320
 321         rc = mdd_get_md(env, mdd_obj, ma->ma_lmm, &ma->ma_lmm_size,
 322                         MDS_LOV_MD_NAME);
 323
 324         if (rc == 0 && (ma->ma_need & MA_LOV_DEF)) {
 325                 rc = mdd_get_default_md(mdd_obj, ma->ma_lmm,
 326                                 &ma->ma_lmm_size);
 327         }
 328
 329         if (rc > 0) {
 330                 ma->ma_valid |= MA_LOV;
 331                 rc = 0;
 332         }
 333         RETURN(rc);
 334 }
 335
 336 int mdd_lmm_get_locked(const struct lu_env *env, struct mdd_object *mdd_obj,
 337                        struct md_attr *ma)
 338 {
 339         int rc;
 340         ENTRY;
 341
 342         mdd_read_lock(env, mdd_obj);
 343         rc = __mdd_lmm_get(env, mdd_obj, ma);
 344         mdd_read_unlock(env, mdd_obj);
 345         RETURN(rc);
 346 }
 347
 348 /* get lmv EA only*/
 349 static int __mdd_lmv_get(const struct lu_env *env,
 350                          struct mdd_object *mdd_obj, struct md_attr *ma)
 351 {
 352         int rc;
 353         ENTRY;
 354
 355         if (ma->ma_valid & MA_LMV)
 356                 RETURN(0);
 357
 358         rc = mdd_get_md(env, mdd_obj, ma->ma_lmv, &ma->ma_lmv_size,
 359                         MDS_LMV_MD_NAME);
 360         if (rc > 0) {
 361                 ma->ma_valid |= MA_LMV;
 362                 rc = 0;
 363         }
 364         RETURN(rc);
 365 }
 366
 367 static int mdd_attr_get_internal(const struct lu_env *env,
 368                                  struct mdd_object *mdd_obj,
 369                                  struct md_attr *ma)
 370 {
 371         int rc = 0;
 372         ENTRY;
 373
 374         if (ma->ma_need & MA_INODE)
 375                 rc = mdd_iattr_get(env, mdd_obj, ma);
 376
 377         if (rc == 0 && ma->ma_need & MA_LOV) {
 378                 if (S_ISREG(mdd_object_type(mdd_obj)) ||
 379                     S_ISDIR(mdd_object_type(mdd_obj)))
 380                         rc = __mdd_lmm_get(env, mdd_obj, ma);
 381         }
 382         if (rc == 0 && ma->ma_need & MA_LMV) {
 383                 if (S_ISDIR(mdd_object_type(mdd_obj)))
 384                         rc = __mdd_lmv_get(env, mdd_obj, ma);
 385         }
 386 #ifdef CONFIG_FS_POSIX_ACL
 387         if (rc == 0 && ma->ma_need & MA_ACL_DEF) {
 388                 if (S_ISDIR(mdd_object_type(mdd_obj)))
 389                         rc = mdd_acl_def_get(env, mdd_obj, ma);
 390         }
 391 #endif
 392         CDEBUG(D_INODE, "after getattr rc = %d, ma_valid = "LPX64"\n",
 393                rc, ma->ma_valid);
 394         RETURN(rc);
 395 }
 396
 397 int mdd_attr_get_internal_locked(const struct lu_env *env,
 398                                  struct mdd_object *mdd_obj, struct md_attr *ma)
 399 {
 400         int rc;
 401         int needlock = ma->ma_need & (MA_LOV | MA_LMV | MA_ACL_DEF);
 402
 403         if (needlock)
 404                 mdd_read_lock(env, mdd_obj);
 405         rc = mdd_attr_get_internal(env, mdd_obj, ma);
 406         if (needlock)
 407                 mdd_read_unlock(env, mdd_obj);
 408         return rc;
 409 }
 410
 411 /*
 412  * No permission check is needed.
 413  */
 414 static int mdd_attr_get(const struct lu_env *env, struct md_object *obj,
 415                         struct md_attr *ma)
 416 {
 417         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 418         int                rc;
 419
 420         ENTRY;
 421         rc = mdd_attr_get_internal_locked(env, mdd_obj, ma);
 422         RETURN(rc);
 423 }
 424
 425 /*
 426  * No permission check is needed.
 427  */
 428 static int mdd_xattr_get(const struct lu_env *env,
 429                          struct md_object *obj, struct lu_buf *buf,
 430                          const char *name)
 431 {
 432         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 433         int rc;
 434
 435         ENTRY;
 436
 437         LASSERT(mdd_object_exists(mdd_obj));
 438
 439         mdd_read_lock(env, mdd_obj);
 440         rc = mdo_xattr_get(env, mdd_obj, buf, name,
 441                            mdd_object_capa(env, mdd_obj));
 442         mdd_read_unlock(env, mdd_obj);
 443
 444         RETURN(rc);
 445 }
 446
 447 /*
 448  * Permission check is done when open,
 449  * no need check again.
 450  */
 451 static int mdd_readlink(const struct lu_env *env, struct md_object *obj,
 452                         struct lu_buf *buf)
 453 {
 454         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 455         struct dt_object  *next;
 456         loff_t             pos = 0;
 457         int                rc;
 458         ENTRY;
 459
 460         LASSERT(mdd_object_exists(mdd_obj));
 461
 462         next = mdd_object_child(mdd_obj);
 463         mdd_read_lock(env, mdd_obj);
 464         rc = next->do_body_ops->dbo_read(env, next, buf, &pos,
 465                                          mdd_object_capa(env, mdd_obj));
 466         mdd_read_unlock(env, mdd_obj);
 467         RETURN(rc);
 468 }
 469
 470 /*
 471  * No permission check is needed.
 472  */
 473 static int mdd_xattr_list(const struct lu_env *env, struct md_object *obj,
 474                           struct lu_buf *buf)
 475 {
 476         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 477         int rc;
 478
 479         ENTRY;
 480
 481         mdd_read_lock(env, mdd_obj);
 482         rc = mdo_xattr_list(env, mdd_obj, buf, mdd_object_capa(env, mdd_obj));
 483         mdd_read_unlock(env, mdd_obj);
 484
 485         RETURN(rc);
 486 }
 487
 488 int mdd_object_create_internal(const struct lu_env *env, struct mdd_object *p,
 489                                struct mdd_object *c, struct md_attr *ma,
 490                                struct thandle *handle)
 491 {
 492         struct lu_attr *attr = &ma->ma_attr;
 493         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
 494         int rc;
 495         ENTRY;
 496
 497         if (!mdd_object_exists(c)) {
 498                 struct dt_object *next = mdd_object_child(c);
 499                 LASSERT(next);
 500
 501                 /* @hint will be initialized by underlying device. */
 502                 next->do_ops->do_ah_init(env, hint,
 503                                          p ? mdd_object_child(p) : NULL,
 504                                          attr->la_mode & S_IFMT);
 505                 rc = mdo_create_obj(env, c, attr, hint, handle);
 506                 LASSERT(ergo(rc == 0, mdd_object_exists(c)));
 507         } else
 508                 rc = -EEXIST;
 509
 510         RETURN(rc);
 511 }
 512
 513
 514 int mdd_attr_set_internal(const struct lu_env *env, struct mdd_object *obj,
 515                           const struct lu_attr *attr, struct thandle *handle,
 516                           const int needacl)
 517 {
 518         int rc;
 519         ENTRY;
 520
 521         rc = mdo_attr_set(env, obj, attr, handle, mdd_object_capa(env, obj));
 522 #ifdef CONFIG_FS_POSIX_ACL
 523         if (!rc && (attr->la_valid & LA_MODE) && needacl)
 524                 rc = mdd_acl_chmod(env, obj, attr->la_mode, handle);
 525 #endif
 526         RETURN(rc);
 527 }
 528
 529 int mdd_attr_set_internal_locked(const struct lu_env *env,
 530                                  struct mdd_object *o,
 531                                  const struct lu_attr *attr,
 532                                  struct thandle *handle, int needacl)
 533 {
 534         int rc;
 535         ENTRY;
 536
 537         needacl = needacl && (attr->la_valid & LA_MODE);
 538
 539         if (needacl)
 540                 mdd_write_lock(env, o);
 541
 542         rc = mdd_attr_set_internal(env, o, attr, handle, needacl);
 543
 544         if (needacl)
 545                 mdd_write_unlock(env, o);
 546         RETURN(rc);
 547 }
 548
 549 static int __mdd_xattr_set(const struct lu_env *env, struct mdd_object *obj,
 550                            const struct lu_buf *buf, const char *name,
 551                            int fl, struct thandle *handle)
 552 {
 553         struct lustre_capa *capa = mdd_object_capa(env, obj);
 554         int rc = -EINVAL;
 555         ENTRY;
 556
 557         if (buf->lb_buf && buf->lb_len > 0)
 558                 rc = mdo_xattr_set(env, obj, buf, name, 0, handle, capa);
 559         else if (buf->lb_buf == NULL && buf->lb_len == 0)
 560                 rc = mdo_xattr_del(env, obj, name, handle, capa);
 561
 562         RETURN(rc);
 563 }
 564
 565 /*
 566  * This gives the same functionality as the code between
 567  * sys_chmod and inode_setattr
 568  * chown_common and inode_setattr
 569  * utimes and inode_setattr
 570  * This API is ported from mds_fix_attr but remove some unnecesssary stuff.
 571  */
 572 static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
 573                         struct lu_attr *la, const struct md_attr *ma)
 574 {
 575         struct lu_attr   *tmp_la     = &mdd_env_info(env)->mti_la;
 576         struct md_ucred  *uc         = md_ucred(env);
 577         int               rc;
 578         ENTRY;
 579
 580         if (!la->la_valid)
 581                 RETURN(0);
 582
 583         /* Do not permit change file type */
 584         if (la->la_valid & LA_TYPE)
 585                 RETURN(-EPERM);
 586
 587         /* They should not be processed by setattr */
 588         if (la->la_valid & (LA_NLINK | LA_RDEV | LA_BLKSIZE))
 589                 RETURN(-EPERM);
 590
 591         /* This is only for set ctime when rename's source is on remote MDS. */
 592         if (unlikely(la->la_valid == LA_CTIME)) {
 593                 rc = mdd_may_delete(env, NULL, obj, (struct md_attr *)ma, 1, 0);
 594                 RETURN(rc);
 595         }
 596
 597         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 598         if (rc)
 599                 RETURN(rc);
 600
 601         if (la->la_valid == LA_ATIME) {
 602                 /* This is atime only set for read atime update on close. */
 603                 if (la->la_atime <= tmp_la->la_atime +
 604                                     mdd_obj2mdd_dev(obj)->mdd_atime_diff)
 605                         la->la_valid &= ~LA_ATIME;
 606                 RETURN(0);
 607         }
 608
 609         /* Check if flags change. */
 610         if (la->la_valid & LA_FLAGS) {
 611                 unsigned int oldflags = 0;
 612                 unsigned int newflags = la->la_flags &
 613                                 (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL);
 614
 615                 if ((uc->mu_fsuid != tmp_la->la_uid) &&
 616                     !mdd_capable(uc, CAP_FOWNER))
 617                         RETURN(-EPERM);
 618
 619                 /* XXX: the IMMUTABLE and APPEND_ONLY flags can
 620                  * only be changed by the relevant capability. */
 621                 if (mdd_is_immutable(obj))
 622                         oldflags |= LUSTRE_IMMUTABLE_FL;
 623                 if (mdd_is_append(obj))
 624                         oldflags |= LUSTRE_APPEND_FL;
 625                 if ((oldflags ^ newflags) &&
 626                     !mdd_capable(uc, CAP_LINUX_IMMUTABLE))
 627                         RETURN(-EPERM);
 628
 629                 if (!S_ISDIR(tmp_la->la_mode))
 630                         la->la_flags &= ~LUSTRE_DIRSYNC_FL;
 631         }
 632
 633         if ((mdd_is_immutable(obj) || mdd_is_append(obj)) &&
 634             (la->la_valid & ~LA_FLAGS) &&
 635             !(ma->ma_attr_flags & MDS_PERM_BYPASS))
 636                 RETURN(-EPERM);
 637
 638         /* Check for setting the obj time. */
 639         if ((la->la_valid & (LA_MTIME | LA_ATIME | LA_CTIME)) &&
 640             !(la->la_valid & ~(LA_MTIME | LA_ATIME | LA_CTIME))) {
 641                 if ((uc->mu_fsuid != tmp_la->la_uid) &&
 642                     !mdd_capable(uc, CAP_FOWNER)) {
 643                         rc = mdd_permission_internal_locked(env, obj, tmp_la,
 644                                                             MAY_WRITE);
 645                         if (rc)
 646                                 RETURN(rc);
 647                 }
 648         }
 649
 650         /* Make sure a caller can chmod. */
 651         if (la->la_valid & LA_MODE) {
 652                 /*
 653                  * Bypass la_vaild == LA_MODE,
 654                  * this is for changing file with SUID or SGID.
 655                  */
 656                 if ((la->la_valid & ~LA_MODE) &&
 657                     (uc->mu_fsuid != tmp_la->la_uid) &&
 658                     !mdd_capable(uc, CAP_FOWNER))
 659                         RETURN(-EPERM);
 660
 661                 if (la->la_mode == (umode_t) -1)
 662                         la->la_mode = tmp_la->la_mode;
 663                 else
 664                         la->la_mode = (la->la_mode & S_IALLUGO) |
 665                                       (tmp_la->la_mode & ~S_IALLUGO);
 666
 667                 /* Also check the setgid bit! */
 668                 if (!lustre_in_group_p(uc, (la->la_valid & LA_GID) ? la->la_gid :
 669                                 tmp_la->la_gid) && !mdd_capable(uc, CAP_FSETID))
 670                         la->la_mode &= ~S_ISGID;
 671         } else {
 672                la->la_mode = tmp_la->la_mode;
 673         }
 674
 675         /* Make sure a caller can chown. */
 676         if (la->la_valid & LA_UID) {
 677                 if (la->la_uid == (uid_t) -1)
 678                         la->la_uid = tmp_la->la_uid;
 679                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
 680                     (la->la_uid != tmp_la->la_uid)) &&
 681                     !mdd_capable(uc, CAP_CHOWN))
 682                         RETURN(-EPERM);
 683
 684                 /*
 685                  * If the user or group of a non-directory has been
 686                  * changed by a non-root user, remove the setuid bit.
 687                  * 19981026 David C Niemi <niemi@tux.org>
 688                  *
 689                  * Changed this to apply to all users, including root,
 690                  * to avoid some races. This is the behavior we had in
 691                  * 2.0. The check for non-root was definitely wrong
 692                  * for 2.2 anyway, as it should have been using
 693                  * CAP_FSETID rather than fsuid -- 19990830 SD.
 694                  */
 695                 if (((tmp_la->la_mode & S_ISUID) == S_ISUID) &&
 696                     !S_ISDIR(tmp_la->la_mode)) {
 697                         la->la_mode &= ~S_ISUID;
 698                         la->la_valid |= LA_MODE;
 699                 }
 700         }
 701
 702         /* Make sure caller can chgrp. */
 703         if (la->la_valid & LA_GID) {
 704                 if (la->la_gid == (gid_t) -1)
 705                         la->la_gid = tmp_la->la_gid;
 706                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
 707                     ((la->la_gid != tmp_la->la_gid) &&
 708                     !lustre_in_group_p(uc, la->la_gid))) &&
 709                     !mdd_capable(uc, CAP_CHOWN))
 710                         RETURN(-EPERM);
 711
 712                 /*
 713                  * Likewise, if the user or group of a non-directory
 714                  * has been changed by a non-root user, remove the
 715                  * setgid bit UNLESS there is no group execute bit
 716                  * (this would be a file marked for mandatory
 717                  * locking).  19981026 David C Niemi <niemi@tux.org>
 718                  *
 719                  * Removed the fsuid check (see the comment above) --
 720                  * 19990830 SD.
 721                  */
 722                 if (((tmp_la->la_mode & (S_ISGID | S_IXGRP)) ==
 723                      (S_ISGID | S_IXGRP)) && !S_ISDIR(tmp_la->la_mode)) {
 724                         la->la_mode &= ~S_ISGID;
 725                         la->la_valid |= LA_MODE;
 726                 }
 727         }
 728
 729         /* For truncate (or setsize), we should have MAY_WRITE perm */
 730         if (la->la_valid & (LA_SIZE | LA_BLOCKS)) {
 731                 if (!((la->la_valid & MDS_OPEN_OWNEROVERRIDE) &&
 732                     (uc->mu_fsuid == tmp_la->la_uid)) &&
 733                     !(ma->ma_attr_flags & MDS_PERM_BYPASS)) {
 734                         rc = mdd_permission_internal_locked(env, obj, tmp_la,
 735                                                             MAY_WRITE);
 736                         if (rc)
 737                                 RETURN(rc);
 738                 }
 739
 740                 /* For the "Size-on-MDS" setattr update, merge coming
 741                  * attributes with the set in the inode. BUG 10641 */
 742                 if ((la->la_valid & LA_ATIME) &&
 743                     (la->la_atime <= tmp_la->la_atime))
 744                         la->la_valid &= ~LA_ATIME;
 745
 746                 /* OST attributes do not have a priority over MDS attributes,
 747                  * so drop times if ctime is equal. */
 748                 if ((la->la_valid & LA_CTIME) &&
 749                     (la->la_ctime <= tmp_la->la_ctime))
 750                         la->la_valid &= ~(LA_MTIME | LA_CTIME);
 751         } else if (la->la_valid & LA_CTIME) {
 752                 /* The pure setattr, it has the priority over what is already
 753                  * set, do not drop it if ctime is equal. */
 754                 if (la->la_ctime < tmp_la->la_ctime)
 755                         la->la_valid &= ~(LA_ATIME | LA_MTIME | LA_CTIME);
 756         }
 757
 758         RETURN(0);
 759 }
 760
 761 /* set attr and LOV EA at once, return updated attr */
 762 static int mdd_attr_set(const struct lu_env *env, struct md_object *obj,
 763                         const struct md_attr *ma)
 764 {
 765         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 766         struct mdd_device *mdd = mdo2mdd(obj);
 767         struct thandle *handle;
 768         struct lov_mds_md *lmm = NULL;
 769         struct llog_cookie *logcookies = NULL;
 770         int  rc, lmm_size = 0, cookie_size = 0;
 771         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
 772         ENTRY;
 773
 774         mdd_txn_param_build(env, mdd, MDD_TXN_ATTR_SET_OP);
 775         handle = mdd_trans_start(env, mdd);
 776         if (IS_ERR(handle))
 777                 RETURN(PTR_ERR(handle));
 778         /*TODO: add lock here*/
 779         /* start a log jounal handle if needed */
 780         if (S_ISREG(mdd_object_type(mdd_obj)) &&
 781             ma->ma_attr.la_valid & (LA_UID | LA_GID)) {
 782                 lmm_size = mdd_lov_mdsize(env, mdd);
 783                 lmm = mdd_max_lmm_get(env, mdd);
 784                 if (lmm == NULL)
 785                         GOTO(cleanup, rc = -ENOMEM);
 786
 787                 rc = mdd_get_md_locked(env, mdd_obj, lmm, &lmm_size,
 788                                 MDS_LOV_MD_NAME);
 789
 790                 if (rc < 0)
 791                         GOTO(cleanup, rc);
 792         }
 793
 794         if (ma->ma_attr.la_valid & (ATTR_MTIME | ATTR_CTIME))
 795                 CDEBUG(D_INODE, "setting mtime "LPU64", ctime "LPU64"\n",
 796                        ma->ma_attr.la_mtime, ma->ma_attr.la_ctime);
 797
 798         *la_copy = ma->ma_attr;
 799         rc = mdd_fix_attr(env, mdd_obj, la_copy, ma);
 800         if (rc)
 801                 GOTO(cleanup, rc);
 802
 803         if (la_copy->la_valid & LA_FLAGS) {
 804                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
 805                                                   handle, 1);
 806                 if (rc == 0)
 807                         mdd_flags_xlate(mdd_obj, la_copy->la_flags);
 808         } else if (la_copy->la_valid) {            /* setattr */
 809                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
 810                                                   handle, 1);
 811                 /* journal chown/chgrp in llog, just like unlink */
 812                 if (rc == 0 && lmm_size){
 813                         cookie_size = mdd_lov_cookiesize(env, mdd);
 814                         logcookies = mdd_max_cookie_get(env, mdd);
 815                         if (logcookies == NULL)
 816                                 GOTO(cleanup, rc = -ENOMEM);
 817
 818                         if (mdd_setattr_log(env, mdd, ma, lmm, lmm_size,
 819                                             logcookies, cookie_size) <= 0)
 820                                 logcookies = NULL;
 821                 }
 822         }
 823
 824         if (rc == 0 && ma->ma_valid & MA_LOV) {
 825                 umode_t mode;
 826
 827                 mode = mdd_object_type(mdd_obj);
 828                 if (S_ISREG(mode) || S_ISDIR(mode)) {
 829                         rc = mdd_lsm_sanity_check(env, mdd_obj);
 830                         if (rc)
 831                                 GOTO(cleanup, rc);
 832
 833                         rc = mdd_lov_set_md(env, NULL, mdd_obj, ma->ma_lmm,
 834                                             ma->ma_lmm_size, handle, 1);
 835                 }
 836
 837         }
 838 cleanup:
 839         mdd_trans_stop(env, mdd, rc, handle);
 840         if (rc == 0 && (lmm != NULL && lmm_size > 0 )) {
 841                 /*set obd attr, if needed*/
 842                 rc = mdd_lov_setattr_async(env, mdd_obj, lmm, lmm_size,
 843                                            logcookies);
 844         }
 845         RETURN(rc);
 846 }
 847
 848 int mdd_xattr_set_txn(const struct lu_env *env, struct mdd_object *obj,
 849                       const struct lu_buf *buf, const char *name, int fl,
 850                       struct thandle *handle)
 851 {
 852         int  rc;
 853         ENTRY;
 854
 855         mdd_write_lock(env, obj);
 856         rc = __mdd_xattr_set(env, obj, buf, name, fl, handle);
 857         mdd_write_unlock(env, obj);
 858
 859         RETURN(rc);
 860 }
 861
 862 static int mdd_xattr_sanity_check(const struct lu_env *env,
 863                                   struct mdd_object *obj)
 864 {
 865         struct lu_attr  *tmp_la = &mdd_env_info(env)->mti_la;
 866         struct md_ucred *uc     = md_ucred(env);
 867         int rc;
 868         ENTRY;
 869
 870         if (mdd_is_immutable(obj) || mdd_is_append(obj))
 871                 RETURN(-EPERM);
 872
 873         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 874         if (rc)
 875                 RETURN(rc);
 876
 877         if ((uc->mu_fsuid != tmp_la->la_uid) && !mdd_capable(uc, CAP_FOWNER))
 878                 RETURN(-EPERM);
 879
 880         RETURN(rc);
 881 }
 882
 883 static int mdd_xattr_set(const struct lu_env *env, struct md_object *obj,
 884                          const struct lu_buf *buf, const char *name, int fl)
 885 {
 886         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
 887         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 888         struct mdd_device *mdd = mdo2mdd(obj);
 889         struct thandle *handle;
 890         int  rc;
 891         ENTRY;
 892
 893         rc = mdd_xattr_sanity_check(env, mdd_obj);
 894         if (rc)
 895                 RETURN(rc);
 896
 897         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
 898         handle = mdd_trans_start(env, mdd);
 899         if (IS_ERR(handle))
 900                 RETURN(PTR_ERR(handle));
 901
 902         rc = mdd_xattr_set_txn(env, md2mdd_obj(obj), buf, name,
 903                                fl, handle);
 904         if (rc == 0) {
 905                 la_copy->la_ctime = cfs_time_current_sec();
 906                 la_copy->la_valid = LA_CTIME;
 907                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
 908                                                   handle, 0);
 909         }
 910         mdd_trans_stop(env, mdd, rc, handle);
 911
 912         RETURN(rc);
 913 }
 914
 915 int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
 916                   const char *name)
 917 {
 918         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
 919         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 920         struct mdd_device *mdd = mdo2mdd(obj);
 921         struct thandle *handle;
 922         int  rc;
 923         ENTRY;
 924
 925         rc = mdd_xattr_sanity_check(env, mdd_obj);
 926         if (rc)
 927                 RETURN(rc);
 928
 929         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
 930         handle = mdd_trans_start(env, mdd);
 931         if (IS_ERR(handle))
 932                 RETURN(PTR_ERR(handle));
 933
 934         mdd_write_lock(env, mdd_obj);
 935         rc = mdo_xattr_del(env, md2mdd_obj(obj), name, handle,
 936                            mdd_object_capa(env, mdd_obj));
 937         mdd_write_unlock(env, mdd_obj);
 938         if (rc == 0) {
 939                 la_copy->la_ctime = cfs_time_current_sec();
 940                 la_copy->la_valid = LA_CTIME;
 941                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
 942                                                   handle, 0);
 943         }
 944
 945         mdd_trans_stop(env, mdd, rc, handle);
 946
 947         RETURN(rc);
 948 }
 949
 950 /* partial unlink */
 951 static int mdd_ref_del(const struct lu_env *env, struct md_object *obj,
 952                        struct md_attr *ma)
 953 {
 954         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
 955         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 956         struct mdd_device *mdd = mdo2mdd(obj);
 957         struct thandle *handle;
 958         int rc;
 959         ENTRY;
 960
 961         /*
 962          * Check -ENOENT early here because we need to get object type
 963          * to calculate credits before transaction start
 964          */
 965         if (!mdd_object_exists(mdd_obj))
 966                 RETURN(-ENOENT);
 967
 968         LASSERT(mdd_object_exists(mdd_obj) > 0);
 969
 970         rc = mdd_log_txn_param_build(env, obj, ma, MDD_TXN_UNLINK_OP);
 971         if (rc)
 972                 RETURN(rc);
 973
 974         handle = mdd_trans_start(env, mdd);
 975         if (IS_ERR(handle))
 976                 RETURN(-ENOMEM);
 977
 978         mdd_write_lock(env, mdd_obj);
 979
 980         rc = mdd_unlink_sanity_check(env, NULL, mdd_obj, ma);
 981         if (rc)
 982                 GOTO(cleanup, rc);
 983
 984         mdo_ref_del(env, mdd_obj, handle);
 985
 986         if (S_ISDIR(lu_object_attr(&obj->mo_lu))) {
 987                 /* unlink dot */
 988                 mdo_ref_del(env, mdd_obj, handle);
 989         }
 990
 991         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
 992         la_copy->la_ctime = ma->ma_attr.la_ctime;
 993
 994         la_copy->la_valid = LA_CTIME;
 995         rc = mdd_attr_set_internal(env, mdd_obj, la_copy, handle, 0);
 996         if (rc)
 997                 GOTO(cleanup, rc);
 998
 999         rc = mdd_finish_unlink(env, mdd_obj, ma, handle);
1000
1001         EXIT;
1002 cleanup:
1003         mdd_write_unlock(env, mdd_obj);
1004         mdd_trans_stop(env, mdd, rc, handle);
1005         return rc;
1006 }
1007
1008 /* partial operation */
1009 static int mdd_oc_sanity_check(const struct lu_env *env,
1010                                struct mdd_object *obj,
1011                                struct md_attr *ma)
1012 {
1013         int rc;
1014         ENTRY;
1015
1016         switch (ma->ma_attr.la_mode & S_IFMT) {
1017         case S_IFREG:
1018         case S_IFDIR:
1019         case S_IFLNK:
1020         case S_IFCHR:
1021         case S_IFBLK:
1022         case S_IFIFO:
1023         case S_IFSOCK:
1024                 rc = 0;
1025                 break;
1026         default:
1027                 rc = -EINVAL;
1028                 break;
1029         }
1030         RETURN(rc);
1031 }
1032
1033 static int mdd_object_create(const struct lu_env *env,
1034                              struct md_object *obj,
1035                              const struct md_op_spec *spec,
1036                              struct md_attr *ma)
1037 {
1038
1039         struct mdd_device *mdd = mdo2mdd(obj);
1040         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1041         const struct lu_fid *pfid = spec->u.sp_pfid;
1042         struct thandle *handle;
1043         int rc;
1044         ENTRY;
1045
1046         mdd_txn_param_build(env, mdd, MDD_TXN_OBJECT_CREATE_OP);
1047         handle = mdd_trans_start(env, mdd);
1048         if (IS_ERR(handle))
1049                 RETURN(PTR_ERR(handle));
1050
1051         mdd_write_lock(env, mdd_obj);
1052         rc = mdd_oc_sanity_check(env, mdd_obj, ma);
1053         if (rc)
1054                 GOTO(unlock, rc);
1055
1056         rc = mdd_object_create_internal(env, NULL, mdd_obj, ma, handle);
1057         if (rc)
1058                 GOTO(unlock, rc);
1059
1060         if (spec->sp_cr_flags & MDS_CREATE_SLAVE_OBJ) {
1061                 /* If creating the slave object, set slave EA here. */
1062                 int lmv_size = spec->u.sp_ea.eadatalen;
1063                 struct lmv_stripe_md *lmv;
1064
1065                 lmv = (struct lmv_stripe_md *)spec->u.sp_ea.eadata;
1066                 LASSERT(lmv != NULL && lmv_size > 0);
1067
1068                 rc = __mdd_xattr_set(env, mdd_obj,
1069                                      mdd_buf_get_const(env, lmv, lmv_size),
1070                                      MDS_LMV_MD_NAME, 0, handle);
1071                 if (rc)
1072                         GOTO(unlock, rc);
1073
1074                 rc = mdd_attr_set_internal(env, mdd_obj, &ma->ma_attr, handle, 0);
1075         } else {
1076 #ifdef CONFIG_FS_POSIX_ACL
1077                 if (spec->sp_cr_flags & MDS_CREATE_RMT_ACL) {
1078                         struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
1079
1080                         buf->lb_buf = (void *)spec->u.sp_ea.eadata;
1081                         buf->lb_len = spec->u.sp_ea.eadatalen;
1082                         if ((buf->lb_len > 0) && (buf->lb_buf != NULL)) {
1083                                 rc = __mdd_acl_init(env, mdd_obj, buf,
1084                                                     &ma->ma_attr.la_mode,
1085                                                     handle);
1086                                 if (rc)
1087                                         GOTO(unlock, rc);
1088                                 else
1089                                         ma->ma_attr.la_valid |= LA_MODE;
1090                         }
1091
1092                         pfid = spec->u.sp_ea.fid;
1093                 }
1094 #endif
1095                 rc = mdd_object_initialize(env, pfid, mdd_obj, ma, handle);
1096         }
1097         EXIT;
1098 unlock:
1099         mdd_write_unlock(env, mdd_obj);
1100         if (rc == 0)
1101                 rc = mdd_attr_get_internal_locked(env, mdd_obj, ma);
1102
1103         mdd_trans_stop(env, mdd, rc, handle);
1104         return rc;
1105 }
1106
1107 /* partial link */
1108 static int mdd_ref_add(const struct lu_env *env, struct md_object *obj,
1109                        const struct md_attr *ma)
1110 {
1111         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
1112         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1113         struct mdd_device *mdd = mdo2mdd(obj);
1114         struct thandle *handle;
1115         int rc;
1116         ENTRY;
1117
1118         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
1119         handle = mdd_trans_start(env, mdd);
1120         if (IS_ERR(handle))
1121                 RETURN(-ENOMEM);
1122
1123         mdd_write_lock(env, mdd_obj);
1124         rc = mdd_link_sanity_check(env, NULL, NULL, mdd_obj);
1125         if (rc == 0)
1126                 mdo_ref_add(env, mdd_obj, handle);
1127         mdd_write_unlock(env, mdd_obj);
1128         if (rc == 0) {
1129                 LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1130                 la_copy->la_ctime = ma->ma_attr.la_ctime;
1131
1132                 la_copy->la_valid = LA_CTIME;
1133                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
1134                                                   handle, 0);
1135         }
1136         mdd_trans_stop(env, mdd, 0, handle);
1137
1138         RETURN(rc);
1139 }
1140
1141 /*
1142  * do NOT or the MAY_*'s, you'll get the weakest
1143  */
1144 int accmode(const struct lu_env *env, struct lu_attr *la, int flags)
1145 {
1146         int res = 0;
1147
1148         /* Sadly, NFSD reopens a file repeatedly during operation, so the
1149          * "acc_mode = 0" allowance for newly-created files isn't honoured.
1150          * NFSD uses the MDS_OPEN_OWNEROVERRIDE flag to say that a file
1151          * owner can write to a file even if it is marked readonly to hide
1152          * its brokenness. (bug 5781) */
1153         if (flags & MDS_OPEN_OWNEROVERRIDE) {
1154                 struct md_ucred *uc = md_ucred(env);
1155
1156                 if ((uc == NULL) || (uc->mu_valid == UCRED_INIT) ||
1157                     (la->la_uid == uc->mu_fsuid))
1158                         return 0;
1159         }
1160
1161         if (flags & FMODE_READ)
1162                 res |= MAY_READ;
1163         if (flags & (FMODE_WRITE | MDS_OPEN_TRUNC | MDS_OPEN_APPEND))
1164                 res |= MAY_WRITE;
1165         if (flags & MDS_FMODE_EXEC)
1166                 res |= MAY_EXEC;
1167         return res;
1168 }
1169
1170 static int mdd_open_sanity_check(const struct lu_env *env,
1171                                  struct mdd_object *obj, int flag)
1172 {
1173         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
1174         int mode, rc;
1175         ENTRY;
1176
1177         /* EEXIST check */
1178         if (mdd_is_dead_obj(obj))
1179                 RETURN(-ENOENT);
1180
1181         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
1182         if (rc)
1183                RETURN(rc);
1184
1185         if (S_ISLNK(tmp_la->la_mode))
1186                 RETURN(-ELOOP);
1187
1188         mode = accmode(env, tmp_la, flag);
1189
1190         if (S_ISDIR(tmp_la->la_mode) && (mode & MAY_WRITE))
1191                 RETURN(-EISDIR);
1192
1193         if (!(flag & MDS_OPEN_CREATED)) {
1194                 rc = mdd_permission_internal(env, obj, tmp_la, mode);
1195                 if (rc)
1196                         RETURN(rc);
1197         }
1198
1199         if (S_ISFIFO(tmp_la->la_mode) || S_ISSOCK(tmp_la->la_mode) ||
1200             S_ISBLK(tmp_la->la_mode) || S_ISCHR(tmp_la->la_mode))
1201                 flag &= ~MDS_OPEN_TRUNC;
1202
1203         /* For writing append-only file must open it with append mode. */
1204         if (mdd_is_append(obj)) {
1205                 if ((flag & FMODE_WRITE) && !(flag & MDS_OPEN_APPEND))
1206                         RETURN(-EPERM);
1207                 if (flag & MDS_OPEN_TRUNC)
1208                         RETURN(-EPERM);
1209         }
1210
1211 #if 0
1212         /*
1213          * Now, flag -- O_NOATIME does not be packed by client.
1214          */
1215         if (flag & O_NOATIME) {
1216                 struct md_ucred *uc = md_ucred(env);
1217
1218                 if (uc && ((uc->mu_valid == UCRED_OLD) ||
1219                     (uc->mu_valid == UCRED_NEW)) &&
1220                     (uc->mu_fsuid != tmp_la->la_uid) &&
1221                     !mdd_capable(uc, CAP_FOWNER))
1222                         RETURN(-EPERM);
1223         }
1224 #endif
1225
1226         RETURN(0);
1227 }
1228
1229 static int mdd_open(const struct lu_env *env, struct md_object *obj,
1230                     int flags)
1231 {
1232         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1233         int rc = 0;
1234
1235         mdd_write_lock(env, mdd_obj);
1236
1237         rc = mdd_open_sanity_check(env, mdd_obj, flags);
1238         if (rc == 0)
1239                 mdd_obj->mod_count++;
1240
1241         mdd_write_unlock(env, mdd_obj);
1242         return rc;
1243 }
1244
1245 /* return md_attr back,
1246  * if it is last unlink then return lov ea + llog cookie*/
1247 int mdd_object_kill(const struct lu_env *env, struct mdd_object *obj,
1248                     struct md_attr *ma)
1249 {
1250         int rc = 0;
1251         ENTRY;
1252
1253         if (S_ISREG(mdd_object_type(obj))) {
1254                 /* Return LOV & COOKIES unconditionally here. We clean evth up.
1255                  * Caller must be ready for that. */
1256                 rc = __mdd_lmm_get(env, obj, ma);
1257                 if ((ma->ma_valid & MA_LOV))
1258                         rc = mdd_unlink_log(env, mdo2mdd(&obj->mod_obj),
1259                                             obj, ma);
1260         }
1261         RETURN(rc);
1262 }
1263
1264 /*
1265  * No permission check is needed.
1266  */
1267 static int mdd_close(const struct lu_env *env, struct md_object *obj,
1268                      struct md_attr *ma)
1269 {
1270         int rc;
1271         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1272         struct thandle    *handle;
1273         ENTRY;
1274
1275         rc = mdd_log_txn_param_build(env, obj, ma, MDD_TXN_UNLINK_OP);
1276         if (rc)
1277                 RETURN(rc);
1278         handle = mdd_trans_start(env, mdo2mdd(obj));
1279         if (IS_ERR(handle))
1280                 RETURN(PTR_ERR(handle));
1281
1282         mdd_write_lock(env, mdd_obj);
1283         /* release open count */
1284         mdd_obj->mod_count --;
1285
1286         rc = mdd_iattr_get(env, mdd_obj, ma);
1287         if (rc == 0 && mdd_obj->mod_count == 0 && ma->ma_attr.la_nlink == 0)
1288                 rc = mdd_object_kill(env, mdd_obj, ma);
1289         else
1290                 ma->ma_valid &= ~(MA_LOV | MA_COOKIE);
1291
1292         mdd_write_unlock(env, mdd_obj);
1293         mdd_trans_stop(env, mdo2mdd(obj), rc, handle);
1294         RETURN(rc);
1295 }
1296
1297 /*
1298  * Permission check is done when open,
1299  * no need check again.
1300  */
1301 static int mdd_readpage_sanity_check(const struct lu_env *env,
1302                                      struct mdd_object *obj)
1303 {
1304         struct dt_object *next = mdd_object_child(obj);
1305         int rc;
1306         ENTRY;
1307
1308         if (S_ISDIR(mdd_object_type(obj)) && dt_try_as_dir(env, next))
1309                 rc = 0;
1310         else
1311                 rc = -ENOTDIR;
1312
1313         RETURN(rc);
1314 }
1315
1316 static int mdd_dir_page_build(const struct lu_env *env, int first,
1317                               void *area, int nob, struct dt_it_ops *iops,
1318                               struct dt_it *it, __u64 *start, __u64 *end,
1319                               struct lu_dirent **last)
1320 {
1321         struct lu_fid          *fid  = &mdd_env_info(env)->mti_fid2;
1322         struct mdd_thread_info *info = mdd_env_info(env);
1323         struct lu_fid_pack     *pack = &info->mti_pack;
1324         int                     result;
1325         struct lu_dirent       *ent;
1326
1327         if (first) {
1328                 memset(area, 0, sizeof (struct lu_dirpage));
1329                 area += sizeof (struct lu_dirpage);
1330                 nob  -= sizeof (struct lu_dirpage);
1331         }
1332
1333         LASSERT(nob > sizeof *ent);
1334
1335         ent  = area;
1336         result = 0;
1337         do {
1338                 char  *name;
1339                 int    len;
1340                 int    recsize;
1341                 __u64  hash;
1342
1343                 name = (char *)iops->key(env, it);
1344                 len  = iops->key_size(env, it);
1345
1346                 pack = (struct lu_fid_pack *)iops->rec(env, it);
1347                 result = fid_unpack(pack, fid);
1348                 if (result != 0)
1349                         break;
1350
1351                 recsize = (sizeof(*ent) + len + 7) & ~7;
1352                 hash = iops->store(env, it);
1353                 *end = hash;
1354
1355                 CDEBUG(D_INFO, "%p %p %d "DFID": "LPU64" (%d) \"%*.*s\"\n",
1356                        name, ent, nob, PFID(fid), hash, len, len, len, name);
1357
1358                 if (nob >= recsize) {
1359                         ent->lde_fid = *fid;
1360                         fid_cpu_to_le(&ent->lde_fid, &ent->lde_fid);
1361                         ent->lde_hash = hash;
1362                         ent->lde_namelen = cpu_to_le16(len);
1363                         ent->lde_reclen  = cpu_to_le16(recsize);
1364                         memcpy(ent->lde_name, name, len);
1365                         if (first && ent == area)
1366                                 *start = hash;
1367                         *last = ent;
1368                         ent = (void *)ent + recsize;
1369                         nob -= recsize;
1370                         result = iops->next(env, it);
1371                 } else {
1372                         /*
1373                          * record doesn't fit into page, enlarge previous one.
1374                          */
1375                         LASSERT(*last != NULL);
1376                         (*last)->lde_reclen =
1377                                 cpu_to_le16(le16_to_cpu((*last)->lde_reclen) +
1378                                             nob);
1379                         break;
1380                 }
1381         } while (result == 0);
1382
1383         return result;
1384 }
1385
1386 static int __mdd_readpage(const struct lu_env *env, struct mdd_object *obj,
1387                           const struct lu_rdpg *rdpg)
1388 {
1389         struct dt_it      *it;
1390         struct dt_object  *next = mdd_object_child(obj);
1391         struct dt_it_ops  *iops;
1392         struct page       *pg;
1393         struct lu_dirent  *last;
1394         int i;
1395         int rc;
1396         int nob;
1397         __u64 hash_start;
1398         __u64 hash_end;
1399
1400         LASSERT(rdpg->rp_pages != NULL);
1401         LASSERT(next->do_index_ops != NULL);
1402
1403         if (rdpg->rp_count <= 0)
1404                 return -EFAULT;
1405
1406         /*
1407          * iterate through directory and fill pages from @rdpg
1408          */
1409         iops = &next->do_index_ops->dio_it;
1410         it = iops->init(env, next, 0, mdd_object_capa(env, obj));
1411         if (IS_ERR(it))
1412                 return PTR_ERR(it);
1413
1414         rc = iops->load(env, it, rdpg->rp_hash);
1415
1416         if (rc == 0)
1417                 /*
1418                  * Iterator didn't find record with exactly the key requested.
1419                  *
1420                  * It is currently either
1421                  *
1422                  *     - positioned above record with key less than
1423                  *     requested---skip it.
1424                  *
1425                  *     - or not positioned at all (is in IAM_IT_SKEWED
1426                  *     state)---position it on the next item.
1427                  */
1428                 rc = iops->next(env, it);
1429         else if (rc > 0)
1430                 rc = 0;
1431
1432         /*
1433          * At this point and across for-loop:
1434          *
1435          *  rc == 0 -> ok, proceed.
1436          *  rc >  0 -> end of directory.
1437          *  rc <  0 -> error.
1438          */
1439         for (i = 0, nob = rdpg->rp_count; rc == 0 && nob > 0;
1440              i++, nob -= CFS_PAGE_SIZE) {
1441                 LASSERT(i < rdpg->rp_npages);
1442                 pg = rdpg->rp_pages[i];
1443                 rc = mdd_dir_page_build(env, !i, cfs_kmap(pg),
1444                                         min_t(int, nob, CFS_PAGE_SIZE), iops,
1445                                         it, &hash_start, &hash_end, &last);
1446                 if (rc != 0 || i == rdpg->rp_npages - 1)
1447                         last->lde_reclen = 0;
1448                 cfs_kunmap(pg);
1449         }
1450         if (rc > 0) {
1451                 /*
1452                  * end of directory.
1453                  */
1454                 hash_end = DIR_END_OFF;
1455                 rc = 0;
1456         }
1457         if (rc == 0) {
1458                 struct lu_dirpage *dp;
1459
1460                 dp = cfs_kmap(rdpg->rp_pages[0]);
1461                 dp->ldp_hash_start = rdpg->rp_hash;
1462                 dp->ldp_hash_end   = hash_end;
1463                 if (i == 0)
1464                         /*
1465                          * No pages were processed, mark this.
1466                          */
1467                         dp->ldp_flags |= LDF_EMPTY;
1468                 dp->ldp_flags = cpu_to_le32(dp->ldp_flags);
1469                 cfs_kunmap(rdpg->rp_pages[0]);
1470         }
1471         iops->put(env, it);
1472         iops->fini(env, it);
1473
1474         return rc;
1475 }
1476
1477 static int mdd_readpage(const struct lu_env *env, struct md_object *obj,
1478                         const struct lu_rdpg *rdpg)
1479 {
1480         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1481         int rc;
1482         ENTRY;
1483
1484         LASSERT(mdd_object_exists(mdd_obj));
1485
1486         mdd_read_lock(env, mdd_obj);
1487         rc = mdd_readpage_sanity_check(env, mdd_obj);
1488         if (rc)
1489                 GOTO(out_unlock, rc);
1490
1491         if (mdd_is_dead_obj(mdd_obj)) {
1492                 struct page *pg;
1493                 struct lu_dirpage *dp;
1494
1495                 /*
1496                  * According to POSIX, please do not return any entry to client:
1497                  * even dot and dotdot should not be returned.
1498                  */
1499                 CWARN("readdir from dead object: "DFID"\n",
1500                         PFID(mdd_object_fid(mdd_obj)));
1501
1502                 if (rdpg->rp_count <= 0)
1503                         GOTO(out_unlock, rc = -EFAULT);
1504                 LASSERT(rdpg->rp_pages != NULL);
1505
1506                 pg = rdpg->rp_pages[0];
1507                 dp = (struct lu_dirpage*)cfs_kmap(pg);
1508                 memset(dp, 0 , sizeof(struct lu_dirpage));
1509                 dp->ldp_hash_start = rdpg->rp_hash;
1510                 dp->ldp_hash_end   = DIR_END_OFF;
1511                 dp->ldp_flags |= LDF_EMPTY;
1512                 dp->ldp_flags = cpu_to_le32(dp->ldp_flags);
1513                 cfs_kunmap(pg);
1514                 GOTO(out_unlock, rc = 0);
1515         }
1516
1517         rc = __mdd_readpage(env, mdd_obj, rdpg);
1518
1519         EXIT;
1520 out_unlock:
1521         mdd_read_unlock(env, mdd_obj);
1522         return rc;
1523 }
1524
1525 struct md_object_operations mdd_obj_ops = {
1526         .moo_permission    = mdd_permission,
1527         .moo_attr_get      = mdd_attr_get,
1528         .moo_attr_set      = mdd_attr_set,
1529         .moo_xattr_get     = mdd_xattr_get,
1530         .moo_xattr_set     = mdd_xattr_set,
1531         .moo_xattr_list    = mdd_xattr_list,
1532         .moo_xattr_del     = mdd_xattr_del,
1533         .moo_object_create = mdd_object_create,
1534         .moo_ref_add       = mdd_ref_add,
1535         .moo_ref_del       = mdd_ref_del,
1536         .moo_open          = mdd_open,
1537         .moo_close         = mdd_close,
1538         .moo_readpage      = mdd_readpage,
1539         .moo_readlink      = mdd_readlink,
1540         .moo_capa_get      = mdd_capa_get
1541 };