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