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