lustre/mdd/mdd_object.c

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