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 <lustre_param.h>
  58 #include <linux/ldiskfs_fs.h>
  59 #include <lustre_mds.h>
  60 #include <lustre/lustre_idl.h>
  61
  62 #include "mdd_internal.h"
  63
  64 static const struct lu_object_operations mdd_lu_obj_ops;
  65
  66 static int mdd_xattr_get(const struct lu_env *env,
  67                          struct md_object *obj, struct lu_buf *buf,
  68                          const char *name);
  69
  70 int mdd_data_get(const struct lu_env *env, struct mdd_object *obj,
  71                  void **data)
  72 {
  73         LASSERTF(mdd_object_exists(obj), "FID is "DFID"\n",
  74                  PFID(mdd_object_fid(obj)));
  75         mdo_data_get(env, obj, data);
  76         return 0;
  77 }
  78
  79 int mdd_la_get(const struct lu_env *env, struct mdd_object *obj,
  80                struct lu_attr *la, struct lustre_capa *capa)
  81 {
  82         LASSERTF(mdd_object_exists(obj), "FID is "DFID"\n",
  83                  PFID(mdd_object_fid(obj)));
  84         return mdo_attr_get(env, obj, la, capa);
  85 }
  86
  87 static void mdd_flags_xlate(struct mdd_object *obj, __u32 flags)
  88 {
  89         obj->mod_flags &= ~(APPEND_OBJ|IMMUTE_OBJ);
  90
  91         if (flags & LUSTRE_APPEND_FL)
  92                 obj->mod_flags |= APPEND_OBJ;
  93
  94         if (flags & LUSTRE_IMMUTABLE_FL)
  95                 obj->mod_flags |= IMMUTE_OBJ;
  96 }
  97
  98 struct mdd_thread_info *mdd_env_info(const struct lu_env *env)
  99 {
 100         struct mdd_thread_info *info;
 101
 102         info = lu_context_key_get(&env->le_ctx, &mdd_thread_key);
 103         LASSERT(info != NULL);
 104         return info;
 105 }
 106
 107 struct lu_buf *mdd_buf_get(const struct lu_env *env, void *area, ssize_t len)
 108 {
 109         struct lu_buf *buf;
 110
 111         buf = &mdd_env_info(env)->mti_buf;
 112         buf->lb_buf = area;
 113         buf->lb_len = len;
 114         return buf;
 115 }
 116
 117 void mdd_buf_put(struct lu_buf *buf)
 118 {
 119         if (buf == NULL || buf->lb_buf == NULL)
 120                 return;
 121         if (buf->lb_vmalloc)
 122                 OBD_VFREE(buf->lb_buf, buf->lb_len);
 123         else
 124                 OBD_FREE(buf->lb_buf, buf->lb_len);
 125         buf->lb_buf = NULL;
 126 }
 127
 128 const struct lu_buf *mdd_buf_get_const(const struct lu_env *env,
 129                                        const void *area, ssize_t len)
 130 {
 131         struct lu_buf *buf;
 132
 133         buf = &mdd_env_info(env)->mti_buf;
 134         buf->lb_buf = (void *)area;
 135         buf->lb_len = len;
 136         return buf;
 137 }
 138
 139 #define BUF_VMALLOC_SIZE (CFS_PAGE_SIZE<<2) /* 16k */
 140 struct lu_buf *mdd_buf_alloc(const struct lu_env *env, ssize_t len)
 141 {
 142         struct lu_buf *buf = &mdd_env_info(env)->mti_big_buf;
 143
 144         if ((len > buf->lb_len) && (buf->lb_buf != NULL)) {
 145                 if (buf->lb_vmalloc)
 146                         OBD_VFREE(buf->lb_buf, buf->lb_len);
 147                 else
 148                         OBD_FREE(buf->lb_buf, buf->lb_len);
 149                 buf->lb_buf = NULL;
 150         }
 151         if (buf->lb_buf == NULL) {
 152                 buf->lb_len = len;
 153                 if (buf->lb_len <= BUF_VMALLOC_SIZE) {
 154                         OBD_ALLOC(buf->lb_buf, buf->lb_len);
 155                         buf->lb_vmalloc = 0;
 156                 }
 157                 if (buf->lb_buf == NULL) {
 158                         OBD_VMALLOC(buf->lb_buf, buf->lb_len);
 159                         buf->lb_vmalloc = 1;
 160                 }
 161                 if (buf->lb_buf == NULL)
 162                         buf->lb_len = 0;
 163         }
 164         return buf;
 165 }
 166
 167 /** Increase the size of the \a mti_big_buf.
 168  * preserves old data in buffer
 169  * old buffer remains unchanged on error
 170  * \retval 0 or -ENOMEM
 171  */
 172 int mdd_buf_grow(const struct lu_env *env, ssize_t len)
 173 {
 174         struct lu_buf *oldbuf = &mdd_env_info(env)->mti_big_buf;
 175         struct lu_buf buf;
 176
 177         LASSERT(len >= oldbuf->lb_len);
 178         if (len > BUF_VMALLOC_SIZE) {
 179                 OBD_VMALLOC(buf.lb_buf, len);
 180                 buf.lb_vmalloc = 1;
 181         } else {
 182                 OBD_ALLOC(buf.lb_buf, len);
 183                 buf.lb_vmalloc = 0;
 184         }
 185         if (buf.lb_buf == NULL)
 186                 return -ENOMEM;
 187
 188         buf.lb_len = len;
 189         memcpy(buf.lb_buf, oldbuf->lb_buf, oldbuf->lb_len);
 190
 191         if (oldbuf->lb_vmalloc)
 192                 OBD_VFREE(oldbuf->lb_buf, oldbuf->lb_len);
 193         else
 194                 OBD_FREE(oldbuf->lb_buf, oldbuf->lb_len);
 195
 196         memcpy(oldbuf, &buf, sizeof(buf));
 197
 198         return 0;
 199 }
 200
 201 struct llog_cookie *mdd_max_cookie_get(const struct lu_env *env,
 202                                        struct mdd_device *mdd)
 203 {
 204         struct mdd_thread_info *mti = mdd_env_info(env);
 205         int                     max_cookie_size;
 206
 207         max_cookie_size = mdd_lov_cookiesize(env, mdd);
 208         if (unlikely(mti->mti_max_cookie_size < max_cookie_size)) {
 209                 if (mti->mti_max_cookie)
 210                         OBD_FREE(mti->mti_max_cookie, mti->mti_max_cookie_size);
 211                 mti->mti_max_cookie = NULL;
 212                 mti->mti_max_cookie_size = 0;
 213         }
 214         if (unlikely(mti->mti_max_cookie == NULL)) {
 215                 OBD_ALLOC(mti->mti_max_cookie, max_cookie_size);
 216                 if (likely(mti->mti_max_cookie != NULL))
 217                         mti->mti_max_cookie_size = max_cookie_size;
 218         }
 219         if (likely(mti->mti_max_cookie != NULL))
 220                 memset(mti->mti_max_cookie, 0, mti->mti_max_cookie_size);
 221         return mti->mti_max_cookie;
 222 }
 223
 224 struct lov_mds_md *mdd_max_lmm_get(const struct lu_env *env,
 225                                    struct mdd_device *mdd)
 226 {
 227         struct mdd_thread_info *mti = mdd_env_info(env);
 228         int                     max_lmm_size;
 229
 230         max_lmm_size = mdd_lov_mdsize(env, mdd);
 231         if (unlikely(mti->mti_max_lmm_size < max_lmm_size)) {
 232                 if (mti->mti_max_lmm)
 233                         OBD_FREE(mti->mti_max_lmm, mti->mti_max_lmm_size);
 234                 mti->mti_max_lmm = NULL;
 235                 mti->mti_max_lmm_size = 0;
 236         }
 237         if (unlikely(mti->mti_max_lmm == NULL)) {
 238                 OBD_ALLOC(mti->mti_max_lmm, max_lmm_size);
 239                 if (unlikely(mti->mti_max_lmm != NULL))
 240                         mti->mti_max_lmm_size = max_lmm_size;
 241         }
 242         return mti->mti_max_lmm;
 243 }
 244
 245 struct lu_object *mdd_object_alloc(const struct lu_env *env,
 246                                    const struct lu_object_header *hdr,
 247                                    struct lu_device *d)
 248 {
 249         struct mdd_object *mdd_obj;
 250
 251         OBD_ALLOC_PTR(mdd_obj);
 252         if (mdd_obj != NULL) {
 253                 struct lu_object *o;
 254
 255                 o = mdd2lu_obj(mdd_obj);
 256                 lu_object_init(o, NULL, d);
 257                 mdd_obj->mod_obj.mo_ops = &mdd_obj_ops;
 258                 mdd_obj->mod_obj.mo_dir_ops = &mdd_dir_ops;
 259                 mdd_obj->mod_count = 0;
 260                 o->lo_ops = &mdd_lu_obj_ops;
 261                 return o;
 262         } else {
 263                 return NULL;
 264         }
 265 }
 266
 267 static int mdd_object_init(const struct lu_env *env, struct lu_object *o,
 268                            const struct lu_object_conf *_)
 269 {
 270         struct mdd_device *d = lu2mdd_dev(o->lo_dev);
 271         struct mdd_object *mdd_obj = lu2mdd_obj(o);
 272         struct lu_object  *below;
 273         struct lu_device  *under;
 274         ENTRY;
 275
 276         mdd_obj->mod_cltime = 0;
 277         under = &d->mdd_child->dd_lu_dev;
 278         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
 279         mdd_pdlock_init(mdd_obj);
 280         if (below == NULL)
 281                 RETURN(-ENOMEM);
 282
 283         lu_object_add(o, below);
 284
 285         RETURN(0);
 286 }
 287
 288 static int mdd_object_start(const struct lu_env *env, struct lu_object *o)
 289 {
 290         if (lu_object_exists(o))
 291                 return mdd_get_flags(env, lu2mdd_obj(o));
 292         else
 293                 return 0;
 294 }
 295
 296 static void mdd_object_free(const struct lu_env *env, struct lu_object *o)
 297 {
 298         struct mdd_object *mdd = lu2mdd_obj(o);
 299
 300         lu_object_fini(o);
 301         OBD_FREE_PTR(mdd);
 302 }
 303
 304 static int mdd_object_print(const struct lu_env *env, void *cookie,
 305                             lu_printer_t p, const struct lu_object *o)
 306 {
 307         struct mdd_object *mdd = lu2mdd_obj((struct lu_object *)o);
 308         return (*p)(env, cookie, LUSTRE_MDD_NAME"-object@%p(open_count=%d, "
 309                     "valid=%x, cltime=%llu, flags=%lx",
 310                     mdd, mdd->mod_count, mdd->mod_valid,
 311                     mdd->mod_cltime, mdd->mod_flags);
 312 }
 313
 314 static const struct lu_object_operations mdd_lu_obj_ops = {
 315         .loo_object_init    = mdd_object_init,
 316         .loo_object_start   = mdd_object_start,
 317         .loo_object_free    = mdd_object_free,
 318         .loo_object_print   = mdd_object_print,
 319 };
 320
 321 struct mdd_object *mdd_object_find(const struct lu_env *env,
 322                                    struct mdd_device *d,
 323                                    const struct lu_fid *f)
 324 {
 325         return md2mdd_obj(md_object_find_slice(env, &d->mdd_md_dev, f));
 326 }
 327
 328 static int mdd_path2fid(const struct lu_env *env, struct mdd_device *mdd,
 329                         const char *path, struct lu_fid *fid)
 330 {
 331         struct lu_buf *buf;
 332         struct lu_fid *f = &mdd_env_info(env)->mti_fid;
 333         struct mdd_object *obj;
 334         struct lu_name *lname = &mdd_env_info(env)->mti_name;
 335         char *name;
 336         int rc = 0;
 337         ENTRY;
 338
 339         /* temp buffer for path element */
 340         buf = mdd_buf_alloc(env, PATH_MAX);
 341         if (buf->lb_buf == NULL)
 342                 RETURN(-ENOMEM);
 343
 344         lname->ln_name = name = buf->lb_buf;
 345         lname->ln_namelen = 0;
 346         *f = mdd->mdd_root_fid;
 347
 348         while(1) {
 349                 while (*path == '/')
 350                         path++;
 351                 if (*path == '\0')
 352                         break;
 353                 while (*path != '/' && *path != '\0') {
 354                         *name = *path;
 355                         path++;
 356                         name++;
 357                         lname->ln_namelen++;
 358                 }
 359
 360                 *name = '\0';
 361                 /* find obj corresponding to fid */
 362                 obj = mdd_object_find(env, mdd, f);
 363                 if (obj == NULL)
 364                         GOTO(out, rc = -EREMOTE);
 365                 if (IS_ERR(obj))
 366                         GOTO(out, rc = -PTR_ERR(obj));
 367                 /* get child fid from parent and name */
 368                 rc = mdd_lookup(env, &obj->mod_obj, lname, f, NULL);
 369                 mdd_object_put(env, obj);
 370                 if (rc)
 371                         break;
 372
 373                 name = buf->lb_buf;
 374                 lname->ln_namelen = 0;
 375         }
 376
 377         if (!rc)
 378                 *fid = *f;
 379 out:
 380         RETURN(rc);
 381 }
 382
 383 /** The maximum depth that fid2path() will search.
 384  * This is limited only because we want to store the fids for
 385  * historical path lookup purposes.
 386  */
 387 #define MAX_PATH_DEPTH 100
 388
 389 /** mdd_path() lookup structure. */
 390 struct path_lookup_info {
 391         __u64                pli_recno;        /**< history point */
 392         struct lu_fid        pli_fid;
 393         struct lu_fid        pli_fids[MAX_PATH_DEPTH]; /**< path, in fids */
 394         struct mdd_object   *pli_mdd_obj;
 395         char                *pli_path;         /**< full path */
 396         int                  pli_pathlen;
 397         int                  pli_linkno;       /**< which hardlink to follow */
 398         int                  pli_fidcount;     /**< number of \a pli_fids */
 399 };
 400
 401 static int mdd_path_current(const struct lu_env *env,
 402                             struct path_lookup_info *pli)
 403 {
 404         struct mdd_device *mdd = mdo2mdd(&pli->pli_mdd_obj->mod_obj);
 405         struct mdd_object *mdd_obj;
 406         struct lu_buf     *buf = NULL;
 407         struct link_ea_header *leh;
 408         struct link_ea_entry  *lee;
 409         struct lu_name *tmpname = &mdd_env_info(env)->mti_name;
 410         struct lu_fid  *tmpfid = &mdd_env_info(env)->mti_fid;
 411         char *ptr;
 412         int reclen;
 413         int rc;
 414         ENTRY;
 415
 416         ptr = pli->pli_path + pli->pli_pathlen - 1;
 417         *ptr = 0;
 418         --ptr;
 419         pli->pli_fidcount = 0;
 420         pli->pli_fids[0] = *(struct lu_fid *)mdd_object_fid(pli->pli_mdd_obj);
 421
 422         while (!mdd_is_root(mdd, &pli->pli_fids[pli->pli_fidcount])) {
 423                 mdd_obj = mdd_object_find(env, mdd,
 424                                           &pli->pli_fids[pli->pli_fidcount]);
 425                 if (mdd_obj == NULL)
 426                         GOTO(out, rc = -EREMOTE);
 427                 if (IS_ERR(mdd_obj))
 428                         GOTO(out, rc = -PTR_ERR(mdd_obj));
 429                 rc = lu_object_exists(&mdd_obj->mod_obj.mo_lu);
 430                 if (rc <= 0) {
 431                         mdd_object_put(env, mdd_obj);
 432                         if (rc == -1)
 433                                 rc = -EREMOTE;
 434                         else if (rc == 0)
 435                                 /* Do I need to error out here? */
 436                                 rc = -ENOENT;
 437                         GOTO(out, rc);
 438                 }
 439
 440                 /* Get parent fid and object name */
 441                 mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 442                 buf = mdd_links_get(env, mdd_obj);
 443                 mdd_read_unlock(env, mdd_obj);
 444                 mdd_object_put(env, mdd_obj);
 445                 if (IS_ERR(buf))
 446                         GOTO(out, rc = PTR_ERR(buf));
 447
 448                 leh = buf->lb_buf;
 449                 lee = (struct link_ea_entry *)(leh + 1); /* link #0 */
 450                 mdd_lee_unpack(lee, &reclen, tmpname, tmpfid);
 451
 452                 /* If set, use link #linkno for path lookup, otherwise use
 453                    link #0.  Only do this for the final path element. */
 454                 if ((pli->pli_fidcount == 0) &&
 455                     (pli->pli_linkno < leh->leh_reccount)) {
 456                         int count;
 457                         for (count = 0; count < pli->pli_linkno; count++) {
 458                                 lee = (struct link_ea_entry *)
 459                                      ((char *)lee + reclen);
 460                                 mdd_lee_unpack(lee, &reclen, tmpname, tmpfid);
 461                         }
 462                         if (pli->pli_linkno < leh->leh_reccount - 1)
 463                                 /* indicate to user there are more links */
 464                                 pli->pli_linkno++;
 465                 }
 466
 467                 /* Pack the name in the end of the buffer */
 468                 ptr -= tmpname->ln_namelen;
 469                 if (ptr - 1 <= pli->pli_path)
 470                         GOTO(out, rc = -EOVERFLOW);
 471                 strncpy(ptr, tmpname->ln_name, tmpname->ln_namelen);
 472                 *(--ptr) = '/';
 473
 474                 /* Store the parent fid for historic lookup */
 475                 if (++pli->pli_fidcount >= MAX_PATH_DEPTH)
 476                         GOTO(out, rc = -EOVERFLOW);
 477                 pli->pli_fids[pli->pli_fidcount] = *tmpfid;
 478         }
 479
 480         /* Verify that our path hasn't changed since we started the lookup */
 481         rc = mdd_path2fid(env, mdd, ptr, &pli->pli_fid);
 482         if (rc) {
 483                 CDEBUG(D_INFO, "mdd_path2fid(%s) failed %d\n", ptr, rc);
 484                 GOTO (out, rc = -EAGAIN);
 485         }
 486         if (!lu_fid_eq(&pli->pli_fids[0], &pli->pli_fid)) {
 487                 CDEBUG(D_INFO, "mdd_path2fid(%s) found another FID o="DFID
 488                        " n="DFID"\n", ptr, PFID(&pli->pli_fids[0]),
 489                        PFID(&pli->pli_fid));
 490                 GOTO(out, rc = -EAGAIN);
 491         }
 492
 493         memmove(pli->pli_path, ptr, pli->pli_path + pli->pli_pathlen - ptr);
 494
 495         EXIT;
 496 out:
 497         if (buf && !IS_ERR(buf) && buf->lb_vmalloc)
 498                 /* if we vmalloced a large buffer drop it */
 499                 mdd_buf_put(buf);
 500
 501         return rc;
 502 }
 503
 504 /* Returns the full path to this fid, as of changelog record recno. */
 505 static int mdd_path(const struct lu_env *env, struct md_object *obj,
 506                     char *path, int pathlen, __u64 recno, int *linkno)
 507 {
 508         struct path_lookup_info *pli;
 509         int tries = 3;
 510         int rc = -EAGAIN;
 511         ENTRY;
 512
 513         if (pathlen < 3)
 514                 RETURN(-EOVERFLOW);
 515
 516         if (mdd_is_root(mdo2mdd(obj), mdd_object_fid(md2mdd_obj(obj)))) {
 517                 path[0] = '/';
 518                 path[1] = '\0';
 519                 RETURN(0);
 520         }
 521
 522         OBD_ALLOC_PTR(pli);
 523         if (pli == NULL)
 524                 RETURN(-ENOMEM);
 525
 526         pli->pli_mdd_obj = md2mdd_obj(obj);
 527         pli->pli_recno = recno;
 528         pli->pli_path = path;
 529         pli->pli_pathlen = pathlen;
 530         pli->pli_linkno = *linkno;
 531
 532         /* Retry multiple times in case file is being moved */
 533         while (tries-- && rc == -EAGAIN)
 534                 rc = mdd_path_current(env, pli);
 535
 536 #if 0   /* We need old path names only for replication */
 537         /* For historical path lookup, the current links may not have existed
 538          * at "recno" time.  We must switch over to earlier links/parents
 539          * by using the changelog records.  If the earlier parent doesn't
 540          * exist, we must search back through the changelog to reconstruct
 541          * its parents, then check if it exists, etc.
 542          * We may ignore this problem for the initial implementation and
 543          * state that an "original" hardlink must still exist for us to find
 544          * historic path name. */
 545         if (pli->pli_recno != -1)
 546                 rc = mdd_path_historic(env, pli);
 547 #endif
 548
 549         /* return next link index to caller */
 550         *linkno = pli->pli_linkno;
 551
 552         OBD_FREE_PTR(pli);
 553
 554         RETURN (rc);
 555 }
 556
 557 int mdd_get_flags(const struct lu_env *env, struct mdd_object *obj)
 558 {
 559         struct lu_attr *la = &mdd_env_info(env)->mti_la;
 560         int rc;
 561
 562         ENTRY;
 563         rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
 564         if (rc == 0) {
 565                 mdd_flags_xlate(obj, la->la_flags);
 566                 if (S_ISDIR(la->la_mode) && la->la_nlink == 1)
 567                         obj->mod_flags |= MNLINK_OBJ;
 568         }
 569         RETURN(rc);
 570 }
 571
 572 /* get only inode attributes */
 573 int mdd_iattr_get(const struct lu_env *env, struct mdd_object *mdd_obj,
 574                   struct md_attr *ma)
 575 {
 576         int rc = 0;
 577         ENTRY;
 578
 579         if (ma->ma_valid & MA_INODE)
 580                 RETURN(0);
 581
 582         rc = mdd_la_get(env, mdd_obj, &ma->ma_attr,
 583                           mdd_object_capa(env, mdd_obj));
 584         if (rc == 0)
 585                 ma->ma_valid |= MA_INODE;
 586         RETURN(rc);
 587 }
 588
 589 int mdd_get_default_md(struct mdd_object *mdd_obj, struct lov_mds_md *lmm,
 590                        int *size)
 591 {
 592         struct lov_desc *ldesc;
 593         struct mdd_device *mdd = mdo2mdd(&mdd_obj->mod_obj);
 594         ENTRY;
 595
 596         ldesc = &mdd->mdd_obd_dev->u.mds.mds_lov_desc;
 597         LASSERT(ldesc != NULL);
 598
 599         if (!lmm)
 600                 RETURN(0);
 601
 602         lmm->lmm_magic = LOV_MAGIC_V1;
 603         lmm->lmm_object_gr = LOV_OBJECT_GROUP_DEFAULT;
 604         lmm->lmm_pattern = ldesc->ld_pattern;
 605         lmm->lmm_stripe_size = ldesc->ld_default_stripe_size;
 606         lmm->lmm_stripe_count = ldesc->ld_default_stripe_count;
 607         *size = sizeof(struct lov_mds_md);
 608
 609         RETURN(sizeof(struct lov_mds_md));
 610 }
 611
 612 /* get lov EA only */
 613 static int __mdd_lmm_get(const struct lu_env *env,
 614                          struct mdd_object *mdd_obj, struct md_attr *ma)
 615 {
 616         int rc;
 617         ENTRY;
 618
 619         if (ma->ma_valid & MA_LOV)
 620                 RETURN(0);
 621
 622         rc = mdd_get_md(env, mdd_obj, ma->ma_lmm, &ma->ma_lmm_size,
 623                         XATTR_NAME_LOV);
 624
 625         if (rc == 0 && (ma->ma_need & MA_LOV_DEF)) {
 626                 rc = mdd_get_default_md(mdd_obj, ma->ma_lmm,
 627                                 &ma->ma_lmm_size);
 628         }
 629
 630         if (rc > 0) {
 631                 ma->ma_valid |= MA_LOV;
 632                 rc = 0;
 633         }
 634         RETURN(rc);
 635 }
 636
 637 int mdd_lmm_get_locked(const struct lu_env *env, struct mdd_object *mdd_obj,
 638                        struct md_attr *ma)
 639 {
 640         int rc;
 641         ENTRY;
 642
 643         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 644         rc = __mdd_lmm_get(env, mdd_obj, ma);
 645         mdd_read_unlock(env, mdd_obj);
 646         RETURN(rc);
 647 }
 648
 649 /* get lmv EA only*/
 650 static int __mdd_lmv_get(const struct lu_env *env,
 651                          struct mdd_object *mdd_obj, struct md_attr *ma)
 652 {
 653         int rc;
 654         ENTRY;
 655
 656         if (ma->ma_valid & MA_LMV)
 657                 RETURN(0);
 658
 659         rc = mdd_get_md(env, mdd_obj, ma->ma_lmv, &ma->ma_lmv_size,
 660                         XATTR_NAME_LMV);
 661         if (rc > 0) {
 662                 ma->ma_valid |= MA_LMV;
 663                 rc = 0;
 664         }
 665         RETURN(rc);
 666 }
 667
 668 static int mdd_attr_get_internal(const struct lu_env *env,
 669                                  struct mdd_object *mdd_obj,
 670                                  struct md_attr *ma)
 671 {
 672         int rc = 0;
 673         ENTRY;
 674
 675         if (ma->ma_need & MA_INODE)
 676                 rc = mdd_iattr_get(env, mdd_obj, ma);
 677
 678         if (rc == 0 && ma->ma_need & MA_LOV) {
 679                 if (S_ISREG(mdd_object_type(mdd_obj)) ||
 680                     S_ISDIR(mdd_object_type(mdd_obj)))
 681                         rc = __mdd_lmm_get(env, mdd_obj, ma);
 682         }
 683         if (rc == 0 && ma->ma_need & MA_LMV) {
 684                 if (S_ISDIR(mdd_object_type(mdd_obj)))
 685                         rc = __mdd_lmv_get(env, mdd_obj, ma);
 686         }
 687 #ifdef CONFIG_FS_POSIX_ACL
 688         if (rc == 0 && ma->ma_need & MA_ACL_DEF) {
 689                 if (S_ISDIR(mdd_object_type(mdd_obj)))
 690                         rc = mdd_def_acl_get(env, mdd_obj, ma);
 691         }
 692 #endif
 693         CDEBUG(D_INODE, "after getattr rc = %d, ma_valid = "LPX64"\n",
 694                rc, ma->ma_valid);
 695         RETURN(rc);
 696 }
 697
 698 int mdd_attr_get_internal_locked(const struct lu_env *env,
 699                                  struct mdd_object *mdd_obj, struct md_attr *ma)
 700 {
 701         int rc;
 702         int needlock = ma->ma_need & (MA_LOV | MA_LMV | MA_ACL_DEF);
 703
 704         if (needlock)
 705                 mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 706         rc = mdd_attr_get_internal(env, mdd_obj, ma);
 707         if (needlock)
 708                 mdd_read_unlock(env, mdd_obj);
 709         return rc;
 710 }
 711
 712 /*
 713  * No permission check is needed.
 714  */
 715 static int mdd_attr_get(const struct lu_env *env, struct md_object *obj,
 716                         struct md_attr *ma)
 717 {
 718         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 719         int                rc;
 720
 721         ENTRY;
 722         rc = mdd_attr_get_internal_locked(env, mdd_obj, ma);
 723         RETURN(rc);
 724 }
 725
 726 /*
 727  * No permission check is needed.
 728  */
 729 static int mdd_xattr_get(const struct lu_env *env,
 730                          struct md_object *obj, struct lu_buf *buf,
 731                          const char *name)
 732 {
 733         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 734         int rc;
 735
 736         ENTRY;
 737
 738         LASSERT(mdd_object_exists(mdd_obj));
 739
 740         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 741         rc = mdo_xattr_get(env, mdd_obj, buf, name,
 742                            mdd_object_capa(env, mdd_obj));
 743         mdd_read_unlock(env, mdd_obj);
 744
 745         RETURN(rc);
 746 }
 747
 748 /*
 749  * Permission check is done when open,
 750  * no need check again.
 751  */
 752 static int mdd_readlink(const struct lu_env *env, struct md_object *obj,
 753                         struct lu_buf *buf)
 754 {
 755         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 756         struct dt_object  *next;
 757         loff_t             pos = 0;
 758         int                rc;
 759         ENTRY;
 760
 761         LASSERT(mdd_object_exists(mdd_obj));
 762
 763         next = mdd_object_child(mdd_obj);
 764         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 765         rc = next->do_body_ops->dbo_read(env, next, buf, &pos,
 766                                          mdd_object_capa(env, mdd_obj));
 767         mdd_read_unlock(env, mdd_obj);
 768         RETURN(rc);
 769 }
 770
 771 /*
 772  * No permission check is needed.
 773  */
 774 static int mdd_xattr_list(const struct lu_env *env, struct md_object *obj,
 775                           struct lu_buf *buf)
 776 {
 777         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 778         int rc;
 779
 780         ENTRY;
 781
 782         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
 783         rc = mdo_xattr_list(env, mdd_obj, buf, mdd_object_capa(env, mdd_obj));
 784         mdd_read_unlock(env, mdd_obj);
 785
 786         RETURN(rc);
 787 }
 788
 789 int mdd_object_create_internal(const struct lu_env *env, struct mdd_object *p,
 790                                struct mdd_object *c, struct md_attr *ma,
 791                                struct thandle *handle,
 792                                const struct md_op_spec *spec)
 793 {
 794         struct lu_attr *attr = &ma->ma_attr;
 795         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
 796         struct dt_object_format *dof = &mdd_env_info(env)->mti_dof;
 797         const struct dt_index_features *feat = spec->sp_feat;
 798         int rc;
 799         ENTRY;
 800
 801         if (!mdd_object_exists(c)) {
 802                 struct dt_object *next = mdd_object_child(c);
 803                 LASSERT(next);
 804
 805                 if (feat != &dt_directory_features && feat != NULL)
 806                         dof->dof_type = DFT_INDEX;
 807                 else
 808                         dof->dof_type = dt_mode_to_dft(attr->la_mode);
 809
 810                 dof->u.dof_idx.di_feat = feat;
 811
 812                 /* @hint will be initialized by underlying device. */
 813                 next->do_ops->do_ah_init(env, hint,
 814                                          p ? mdd_object_child(p) : NULL,
 815                                          attr->la_mode & S_IFMT);
 816
 817                 rc = mdo_create_obj(env, c, attr, hint, dof, handle);
 818                 LASSERT(ergo(rc == 0, mdd_object_exists(c)));
 819         } else
 820                 rc = -EEXIST;
 821
 822         RETURN(rc);
 823 }
 824
 825 /**
 826  * Make sure the ctime is increased only.
 827  */
 828 static inline int mdd_attr_check(const struct lu_env *env,
 829                                  struct mdd_object *obj,
 830                                  struct lu_attr *attr)
 831 {
 832         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
 833         int rc;
 834         ENTRY;
 835
 836         if (attr->la_valid & LA_CTIME) {
 837                 rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 838                 if (rc)
 839                         RETURN(rc);
 840
 841                 if (attr->la_ctime < tmp_la->la_ctime)
 842                         attr->la_valid &= ~(LA_MTIME | LA_CTIME);
 843                 else if (attr->la_valid == LA_CTIME &&
 844                          attr->la_ctime == tmp_la->la_ctime)
 845                         attr->la_valid &= ~LA_CTIME;
 846         }
 847         RETURN(0);
 848 }
 849
 850 int mdd_attr_set_internal(const struct lu_env *env,
 851                           struct mdd_object *obj,
 852                           struct lu_attr *attr,
 853                           struct thandle *handle,
 854                           int needacl)
 855 {
 856         int rc;
 857         ENTRY;
 858
 859         rc = mdo_attr_set(env, obj, attr, handle, mdd_object_capa(env, obj));
 860 #ifdef CONFIG_FS_POSIX_ACL
 861         if (!rc && (attr->la_valid & LA_MODE) && needacl)
 862                 rc = mdd_acl_chmod(env, obj, attr->la_mode, handle);
 863 #endif
 864         RETURN(rc);
 865 }
 866
 867 int mdd_attr_check_set_internal(const struct lu_env *env,
 868                                 struct mdd_object *obj,
 869                                 struct lu_attr *attr,
 870                                 struct thandle *handle,
 871                                 int needacl)
 872 {
 873         int rc;
 874         ENTRY;
 875
 876         rc = mdd_attr_check(env, obj, attr);
 877         if (rc)
 878                 RETURN(rc);
 879
 880         if (attr->la_valid)
 881                 rc = mdd_attr_set_internal(env, obj, attr, handle, needacl);
 882         RETURN(rc);
 883 }
 884
 885 static int mdd_attr_set_internal_locked(const struct lu_env *env,
 886                                         struct mdd_object *obj,
 887                                         struct lu_attr *attr,
 888                                         struct thandle *handle,
 889                                         int needacl)
 890 {
 891         int rc;
 892         ENTRY;
 893
 894         needacl = needacl && (attr->la_valid & LA_MODE);
 895         if (needacl)
 896                 mdd_write_lock(env, obj, MOR_TGT_CHILD);
 897         rc = mdd_attr_set_internal(env, obj, attr, handle, needacl);
 898         if (needacl)
 899                 mdd_write_unlock(env, obj);
 900         RETURN(rc);
 901 }
 902
 903 int mdd_attr_check_set_internal_locked(const struct lu_env *env,
 904                                        struct mdd_object *obj,
 905                                        struct lu_attr *attr,
 906                                        struct thandle *handle,
 907                                        int needacl)
 908 {
 909         int rc;
 910         ENTRY;
 911
 912         needacl = needacl && (attr->la_valid & LA_MODE);
 913         if (needacl)
 914                 mdd_write_lock(env, obj, MOR_TGT_CHILD);
 915         rc = mdd_attr_check_set_internal(env, obj, attr, handle, needacl);
 916         if (needacl)
 917                 mdd_write_unlock(env, obj);
 918         RETURN(rc);
 919 }
 920
 921 int __mdd_xattr_set(const struct lu_env *env, struct mdd_object *obj,
 922                     const struct lu_buf *buf, const char *name,
 923                     int fl, struct thandle *handle)
 924 {
 925         struct lustre_capa *capa = mdd_object_capa(env, obj);
 926         int rc = -EINVAL;
 927         ENTRY;
 928
 929         if (buf->lb_buf && buf->lb_len > 0)
 930                 rc = mdo_xattr_set(env, obj, buf, name, 0, handle, capa);
 931         else if (buf->lb_buf == NULL && buf->lb_len == 0)
 932                 rc = mdo_xattr_del(env, obj, name, handle, capa);
 933
 934         RETURN(rc);
 935 }
 936
 937 /*
 938  * This gives the same functionality as the code between
 939  * sys_chmod and inode_setattr
 940  * chown_common and inode_setattr
 941  * utimes and inode_setattr
 942  * This API is ported from mds_fix_attr but remove some unnecesssary stuff.
 943  */
 944 static int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
 945                         struct lu_attr *la, const struct md_attr *ma)
 946 {
 947         struct lu_attr   *tmp_la     = &mdd_env_info(env)->mti_la;
 948         struct md_ucred  *uc         = md_ucred(env);
 949         int               rc;
 950         ENTRY;
 951
 952         if (!la->la_valid)
 953                 RETURN(0);
 954
 955         /* Do not permit change file type */
 956         if (la->la_valid & LA_TYPE)
 957                 RETURN(-EPERM);
 958
 959         /* They should not be processed by setattr */
 960         if (la->la_valid & (LA_NLINK | LA_RDEV | LA_BLKSIZE))
 961                 RETURN(-EPERM);
 962
 963         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 964         if (rc)
 965                 RETURN(rc);
 966
 967         if (la->la_valid == LA_CTIME) {
 968                 if (!(ma->ma_attr_flags & MDS_PERM_BYPASS))
 969                         /* This is only for set ctime when rename's source is
 970                          * on remote MDS. */
 971                         rc = mdd_may_delete(env, NULL, obj,
 972                                             (struct md_attr *)ma, 1, 0);
 973                 if (rc == 0 && la->la_ctime <= tmp_la->la_ctime)
 974                         la->la_valid &= ~LA_CTIME;
 975                 RETURN(rc);
 976         }
 977
 978         if (la->la_valid == LA_ATIME) {
 979                 /* This is atime only set for read atime update on close. */
 980                 if (la->la_atime <= tmp_la->la_atime +
 981                                     mdd_obj2mdd_dev(obj)->mdd_atime_diff)
 982                         la->la_valid &= ~LA_ATIME;
 983                 RETURN(0);
 984         }
 985
 986         /* Check if flags change. */
 987         if (la->la_valid & LA_FLAGS) {
 988                 unsigned int oldflags = 0;
 989                 unsigned int newflags = la->la_flags &
 990                                 (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL);
 991
 992                 if ((uc->mu_fsuid != tmp_la->la_uid) &&
 993                     !mdd_capable(uc, CFS_CAP_FOWNER))
 994                         RETURN(-EPERM);
 995
 996                 /* XXX: the IMMUTABLE and APPEND_ONLY flags can
 997                  * only be changed by the relevant capability. */
 998                 if (mdd_is_immutable(obj))
 999                         oldflags |= LUSTRE_IMMUTABLE_FL;
1000                 if (mdd_is_append(obj))
1001                         oldflags |= LUSTRE_APPEND_FL;
1002                 if ((oldflags ^ newflags) &&
1003                     !mdd_capable(uc, CFS_CAP_LINUX_IMMUTABLE))
1004                         RETURN(-EPERM);
1005
1006                 if (!S_ISDIR(tmp_la->la_mode))
1007                         la->la_flags &= ~LUSTRE_DIRSYNC_FL;
1008         }
1009
1010         if ((mdd_is_immutable(obj) || mdd_is_append(obj)) &&
1011             (la->la_valid & ~LA_FLAGS) &&
1012             !(ma->ma_attr_flags & MDS_PERM_BYPASS))
1013                 RETURN(-EPERM);
1014
1015         /* Check for setting the obj time. */
1016         if ((la->la_valid & (LA_MTIME | LA_ATIME | LA_CTIME)) &&
1017             !(la->la_valid & ~(LA_MTIME | LA_ATIME | LA_CTIME))) {
1018                 if ((uc->mu_fsuid != tmp_la->la_uid) &&
1019                     !mdd_capable(uc, CFS_CAP_FOWNER)) {
1020                         rc = mdd_permission_internal_locked(env, obj, tmp_la,
1021                                                             MAY_WRITE,
1022                                                             MOR_TGT_CHILD);
1023                         if (rc)
1024                                 RETURN(rc);
1025                 }
1026         }
1027
1028         /* Make sure a caller can chmod. */
1029         if (la->la_valid & LA_MODE) {
1030                 /* Bypass la_vaild == LA_MODE,
1031                  * this is for changing file with SUID or SGID. */
1032                 if ((la->la_valid & ~LA_MODE) &&
1033                     !(ma->ma_attr_flags & MDS_PERM_BYPASS) &&
1034                     (uc->mu_fsuid != tmp_la->la_uid) &&
1035                     !mdd_capable(uc, CFS_CAP_FOWNER))
1036                         RETURN(-EPERM);
1037
1038                 if (la->la_mode == (umode_t) -1)
1039                         la->la_mode = tmp_la->la_mode;
1040                 else
1041                         la->la_mode = (la->la_mode & S_IALLUGO) |
1042                                       (tmp_la->la_mode & ~S_IALLUGO);
1043
1044                 /* Also check the setgid bit! */
1045                 if (!lustre_in_group_p(uc, (la->la_valid & LA_GID) ?
1046                                        la->la_gid : tmp_la->la_gid) &&
1047                     !mdd_capable(uc, CFS_CAP_FSETID))
1048                         la->la_mode &= ~S_ISGID;
1049         } else {
1050                la->la_mode = tmp_la->la_mode;
1051         }
1052
1053         /* Make sure a caller can chown. */
1054         if (la->la_valid & LA_UID) {
1055                 if (la->la_uid == (uid_t) -1)
1056                         la->la_uid = tmp_la->la_uid;
1057                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
1058                     (la->la_uid != tmp_la->la_uid)) &&
1059                     !mdd_capable(uc, CFS_CAP_CHOWN))
1060                         RETURN(-EPERM);
1061
1062                 /* If the user or group of a non-directory has been
1063                  * changed by a non-root user, remove the setuid bit.
1064                  * 19981026 David C Niemi <niemi@tux.org>
1065                  *
1066                  * Changed this to apply to all users, including root,
1067                  * to avoid some races. This is the behavior we had in
1068                  * 2.0. The check for non-root was definitely wrong
1069                  * for 2.2 anyway, as it should have been using
1070                  * CAP_FSETID rather than fsuid -- 19990830 SD. */
1071                 if (((tmp_la->la_mode & S_ISUID) == S_ISUID) &&
1072                     !S_ISDIR(tmp_la->la_mode)) {
1073                         la->la_mode &= ~S_ISUID;
1074                         la->la_valid |= LA_MODE;
1075                 }
1076         }
1077
1078         /* Make sure caller can chgrp. */
1079         if (la->la_valid & LA_GID) {
1080                 if (la->la_gid == (gid_t) -1)
1081                         la->la_gid = tmp_la->la_gid;
1082                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
1083                     ((la->la_gid != tmp_la->la_gid) &&
1084                     !lustre_in_group_p(uc, la->la_gid))) &&
1085                     !mdd_capable(uc, CFS_CAP_CHOWN))
1086                         RETURN(-EPERM);
1087
1088                 /* Likewise, if the user or group of a non-directory
1089                  * has been changed by a non-root user, remove the
1090                  * setgid bit UNLESS there is no group execute bit
1091                  * (this would be a file marked for mandatory
1092                  * locking).  19981026 David C Niemi <niemi@tux.org>
1093                  *
1094                  * Removed the fsuid check (see the comment above) --
1095                  * 19990830 SD. */
1096                 if (((tmp_la->la_mode & (S_ISGID | S_IXGRP)) ==
1097                      (S_ISGID | S_IXGRP)) && !S_ISDIR(tmp_la->la_mode)) {
1098                         la->la_mode &= ~S_ISGID;
1099                         la->la_valid |= LA_MODE;
1100                 }
1101         }
1102
1103         /* For both Size-on-MDS case and truncate case,
1104          * "la->la_valid & (LA_SIZE | LA_BLOCKS)" are ture.
1105          * We distinguish them by "ma->ma_attr_flags & MDS_SOM".
1106          * For SOM case, it is true, the MAY_WRITE perm has been checked
1107          * when open, no need check again. For truncate case, it is false,
1108          * the MAY_WRITE perm should be checked here. */
1109         if (ma->ma_attr_flags & MDS_SOM) {
1110                 /* For the "Size-on-MDS" setattr update, merge coming
1111                  * attributes with the set in the inode. BUG 10641 */
1112                 if ((la->la_valid & LA_ATIME) &&
1113                     (la->la_atime <= tmp_la->la_atime))
1114                         la->la_valid &= ~LA_ATIME;
1115
1116                 /* OST attributes do not have a priority over MDS attributes,
1117                  * so drop times if ctime is equal. */
1118                 if ((la->la_valid & LA_CTIME) &&
1119                     (la->la_ctime <= tmp_la->la_ctime))
1120                         la->la_valid &= ~(LA_MTIME | LA_CTIME);
1121         } else {
1122                 if (la->la_valid & (LA_SIZE | LA_BLOCKS)) {
1123                         if (!((ma->ma_attr_flags & MDS_OPEN_OWNEROVERRIDE) &&
1124                               (uc->mu_fsuid == tmp_la->la_uid)) &&
1125                             !(ma->ma_attr_flags & MDS_PERM_BYPASS)) {
1126                                 rc = mdd_permission_internal_locked(env, obj,
1127                                                             tmp_la, MAY_WRITE,
1128                                                             MOR_TGT_CHILD);
1129                                 if (rc)
1130                                         RETURN(rc);
1131                         }
1132                 }
1133                 if (la->la_valid & LA_CTIME) {
1134                         /* The pure setattr, it has the priority over what is
1135                          * already set, do not drop it if ctime is equal. */
1136                         if (la->la_ctime < tmp_la->la_ctime)
1137                                 la->la_valid &= ~(LA_ATIME | LA_MTIME |
1138                                                   LA_CTIME);
1139                 }
1140         }
1141
1142         RETURN(0);
1143 }
1144
1145 /** Store a data change changelog record
1146  * If this fails, we must fail the whole transaction; we don't
1147  * want the change to commit without the log entry.
1148  * \param mdd_obj - mdd_object of change
1149  * \param handle - transacion handle
1150  */
1151 static int mdd_changelog_data_store(const struct lu_env     *env,
1152                                     struct mdd_device       *mdd,
1153                                     enum changelog_rec_type type,
1154                                     struct mdd_object       *mdd_obj,
1155                                     struct thandle          *handle)
1156 {
1157         const struct lu_fid *tfid = mdo2fid(mdd_obj);
1158         struct llog_changelog_rec *rec;
1159         struct lu_buf *buf;
1160         int reclen;
1161         int rc;
1162
1163         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
1164                 RETURN(0);
1165
1166         LASSERT(handle != NULL);
1167         LASSERT(mdd_obj != NULL);
1168
1169         if ((type == CL_SETATTR) &&
1170             cfs_time_before_64(mdd->mdd_cl.mc_starttime, mdd_obj->mod_cltime)) {
1171                 /* Don't need multiple updates in this log */
1172                 /* Don't check under lock - no big deal if we get an extra
1173                    entry */
1174                 RETURN(0);
1175         }
1176
1177         reclen = llog_data_len(sizeof(*rec));
1178         buf = mdd_buf_alloc(env, reclen);
1179         if (buf->lb_buf == NULL)
1180                 RETURN(-ENOMEM);
1181         rec = (struct llog_changelog_rec *)buf->lb_buf;
1182
1183         rec->cr_flags = CLF_VERSION;
1184         rec->cr_type = (__u32)type;
1185         rec->cr_tfid = *tfid;
1186         rec->cr_namelen = 0;
1187         mdd_obj->mod_cltime = cfs_time_current_64();
1188
1189         rc = mdd_changelog_llog_write(mdd, rec, handle);
1190         if (rc < 0) {
1191                 CERROR("changelog failed: rc=%d op%d t"DFID"\n",
1192                        rc, type, PFID(tfid));
1193                 return -EFAULT;
1194         }
1195
1196         return 0;
1197 }
1198
1199 /* set attr and LOV EA at once, return updated attr */
1200 static int mdd_attr_set(const struct lu_env *env, struct md_object *obj,
1201                         const struct md_attr *ma)
1202 {
1203         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1204         struct mdd_device *mdd = mdo2mdd(obj);
1205         struct thandle *handle;
1206         struct lov_mds_md *lmm = NULL;
1207         struct llog_cookie *logcookies = NULL;
1208         int  rc, lmm_size = 0, cookie_size = 0;
1209         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
1210 #ifdef HAVE_QUOTA_SUPPORT
1211         struct obd_device *obd = mdd->mdd_obd_dev;
1212         struct mds_obd *mds = &obd->u.mds;
1213         unsigned int qnids[MAXQUOTAS] = { 0, 0 };
1214         unsigned int qoids[MAXQUOTAS] = { 0, 0 };
1215         int quota_opc = 0, block_count = 0;
1216         int inode_pending = 0, block_pending = 0;
1217 #endif
1218         ENTRY;
1219
1220         mdd_setattr_txn_param_build(env, obj, (struct md_attr *)ma,
1221                                     MDD_TXN_ATTR_SET_OP);
1222         handle = mdd_trans_start(env, mdd);
1223         if (IS_ERR(handle))
1224                 RETURN(PTR_ERR(handle));
1225         /*TODO: add lock here*/
1226         /* start a log jounal handle if needed */
1227         if (S_ISREG(mdd_object_type(mdd_obj)) &&
1228             ma->ma_attr.la_valid & (LA_UID | LA_GID)) {
1229                 lmm_size = mdd_lov_mdsize(env, mdd);
1230                 lmm = mdd_max_lmm_get(env, mdd);
1231                 if (lmm == NULL)
1232                         GOTO(cleanup, rc = -ENOMEM);
1233
1234                 rc = mdd_get_md_locked(env, mdd_obj, lmm, &lmm_size,
1235                                 XATTR_NAME_LOV);
1236
1237                 if (rc < 0)
1238                         GOTO(cleanup, rc);
1239         }
1240
1241         if (ma->ma_attr.la_valid & (LA_MTIME | LA_CTIME))
1242                 CDEBUG(D_INODE, "setting mtime "LPU64", ctime "LPU64"\n",
1243                        ma->ma_attr.la_mtime, ma->ma_attr.la_ctime);
1244
1245         *la_copy = ma->ma_attr;
1246         rc = mdd_fix_attr(env, mdd_obj, la_copy, ma);
1247         if (rc)
1248                 GOTO(cleanup, rc);
1249
1250 #ifdef HAVE_QUOTA_SUPPORT
1251         if (mds->mds_quota && la_copy->la_valid & (LA_UID | LA_GID)) {
1252                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1253
1254                 rc = mdd_la_get(env, mdd_obj, la_tmp, BYPASS_CAPA);
1255                 if (!rc) {
1256                         quota_opc = FSFILT_OP_SETATTR;
1257                         mdd_quota_wrapper(la_copy, qnids);
1258                         mdd_quota_wrapper(la_tmp, qoids);
1259                         /* get file quota for new owner */
1260                         lquota_chkquota(mds_quota_interface_ref, obd,
1261                                         qnids[USRQUOTA], qnids[GRPQUOTA], 1,
1262                                         &inode_pending, NULL, 0, NULL, 0);
1263                         block_count = (la_tmp->la_blocks + 7) >> 3;
1264                         if (block_count) {
1265                                 void *data = NULL;
1266                                 mdd_data_get(env, mdd_obj, &data);
1267                                 /* get block quota for new owner */
1268                                 lquota_chkquota(mds_quota_interface_ref, obd,
1269                                                 qnids[USRQUOTA],
1270                                                 qnids[GRPQUOTA],
1271                                                 block_count, &block_pending,
1272                                                 NULL, LQUOTA_FLAGS_BLK,
1273                                                 data, 1);
1274                         }
1275                 }
1276         }
1277 #endif
1278
1279         if (la_copy->la_valid & LA_FLAGS) {
1280                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
1281                                                   handle, 1);
1282                 if (rc == 0)
1283                         mdd_flags_xlate(mdd_obj, la_copy->la_flags);
1284         } else if (la_copy->la_valid) {            /* setattr */
1285                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
1286                                                   handle, 1);
1287                 /* journal chown/chgrp in llog, just like unlink */
1288                 if (rc == 0 && lmm_size){
1289                         cookie_size = mdd_lov_cookiesize(env, mdd);
1290                         logcookies = mdd_max_cookie_get(env, mdd);
1291                         if (logcookies == NULL)
1292                                 GOTO(cleanup, rc = -ENOMEM);
1293
1294                         if (mdd_setattr_log(env, mdd, ma, lmm, lmm_size,
1295                                             logcookies, cookie_size) <= 0)
1296                                 logcookies = NULL;
1297                 }
1298         }
1299
1300         if (rc == 0 && ma->ma_valid & MA_LOV) {
1301                 umode_t mode;
1302
1303                 mode = mdd_object_type(mdd_obj);
1304                 if (S_ISREG(mode) || S_ISDIR(mode)) {
1305                         rc = mdd_lsm_sanity_check(env, mdd_obj);
1306                         if (rc)
1307                                 GOTO(cleanup, rc);
1308
1309                         rc = mdd_lov_set_md(env, NULL, mdd_obj, ma->ma_lmm,
1310                                             ma->ma_lmm_size, handle, 1);
1311                 }
1312
1313         }
1314 cleanup:
1315         if ((rc == 0) && (ma->ma_attr.la_valid & (LA_MTIME | LA_CTIME)))
1316                 rc = mdd_changelog_data_store(env, mdd, CL_SETATTR, mdd_obj,
1317                                               handle);
1318         mdd_trans_stop(env, mdd, rc, handle);
1319         if (rc == 0 && (lmm != NULL && lmm_size > 0 )) {
1320                 /*set obd attr, if needed*/
1321                 rc = mdd_lov_setattr_async(env, mdd_obj, lmm, lmm_size,
1322                                            logcookies);
1323         }
1324 #ifdef HAVE_QUOTA_SUPPORT
1325         if (quota_opc) {
1326                 if (inode_pending)
1327                         lquota_pending_commit(mds_quota_interface_ref, obd,
1328                                               qnids[USRQUOTA], qnids[GRPQUOTA],
1329                                               inode_pending, 0);
1330                 if (block_pending)
1331                         lquota_pending_commit(mds_quota_interface_ref, obd,
1332                                               qnids[USRQUOTA], qnids[GRPQUOTA],
1333                                               block_pending, 1);
1334                 /* Trigger dqrel/dqacq for original owner and new owner.
1335                  * If failed, the next call for lquota_chkquota will
1336                  * process it. */
1337                 lquota_adjust(mds_quota_interface_ref, obd, qnids, qoids, rc,
1338                               quota_opc);
1339         }
1340 #endif
1341         RETURN(rc);
1342 }
1343
1344 int mdd_xattr_set_txn(const struct lu_env *env, struct mdd_object *obj,
1345                       const struct lu_buf *buf, const char *name, int fl,
1346                       struct thandle *handle)
1347 {
1348         int  rc;
1349         ENTRY;
1350
1351         mdd_write_lock(env, obj, MOR_TGT_CHILD);
1352         rc = __mdd_xattr_set(env, obj, buf, name, fl, handle);
1353         mdd_write_unlock(env, obj);
1354
1355         RETURN(rc);
1356 }
1357
1358 static int mdd_xattr_sanity_check(const struct lu_env *env,
1359                                   struct mdd_object *obj)
1360 {
1361         struct lu_attr  *tmp_la = &mdd_env_info(env)->mti_la;
1362         struct md_ucred *uc     = md_ucred(env);
1363         int rc;
1364         ENTRY;
1365
1366         if (mdd_is_immutable(obj) || mdd_is_append(obj))
1367                 RETURN(-EPERM);
1368
1369         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
1370         if (rc)
1371                 RETURN(rc);
1372
1373         if ((uc->mu_fsuid != tmp_la->la_uid) &&
1374             !mdd_capable(uc, CFS_CAP_FOWNER))
1375                 RETURN(-EPERM);
1376
1377         RETURN(rc);
1378 }
1379
1380 /**
1381  * The caller should guarantee to update the object ctime
1382  * after xattr_set if needed.
1383  */
1384 static int mdd_xattr_set(const struct lu_env *env, struct md_object *obj,
1385                          const struct lu_buf *buf, const char *name,
1386                          int fl)
1387 {
1388         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1389         struct mdd_device *mdd = mdo2mdd(obj);
1390         struct thandle *handle;
1391         int  rc;
1392         ENTRY;
1393
1394         rc = mdd_xattr_sanity_check(env, mdd_obj);
1395         if (rc)
1396                 RETURN(rc);
1397
1398         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
1399         handle = mdd_trans_start(env, mdd);
1400         if (IS_ERR(handle))
1401                 RETURN(PTR_ERR(handle));
1402
1403         rc = mdd_xattr_set_txn(env, mdd_obj, buf, name, fl, handle);
1404
1405         /* Only record user xattr changes */
1406         if ((rc == 0) && (mdd->mdd_cl.mc_flags & CLM_ON) &&
1407             (strncmp("user.", name, 5) == 0))
1408                 rc = mdd_changelog_data_store(env, mdd, CL_XATTR, mdd_obj,
1409                                               handle);
1410         mdd_trans_stop(env, mdd, rc, handle);
1411
1412         RETURN(rc);
1413 }
1414
1415 /**
1416  * The caller should guarantee to update the object ctime
1417  * after xattr_set if needed.
1418  */
1419 int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
1420                   const char *name)
1421 {
1422         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1423         struct mdd_device *mdd = mdo2mdd(obj);
1424         struct thandle *handle;
1425         int  rc;
1426         ENTRY;
1427
1428         rc = mdd_xattr_sanity_check(env, mdd_obj);
1429         if (rc)
1430                 RETURN(rc);
1431
1432         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
1433         handle = mdd_trans_start(env, mdd);
1434         if (IS_ERR(handle))
1435                 RETURN(PTR_ERR(handle));
1436
1437         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1438         rc = mdo_xattr_del(env, mdd_obj, name, handle,
1439                            mdd_object_capa(env, mdd_obj));
1440         mdd_write_unlock(env, mdd_obj);
1441
1442         /* Only record user xattr changes */
1443         if ((rc == 0) && (mdd->mdd_cl.mc_flags & CLM_ON) &&
1444             (strncmp("user.", name, 5) != 0))
1445                 rc = mdd_changelog_data_store(env, mdd, CL_XATTR, mdd_obj,
1446                                               handle);
1447
1448         mdd_trans_stop(env, mdd, rc, handle);
1449
1450         RETURN(rc);
1451 }
1452
1453 /* partial unlink */
1454 static int mdd_ref_del(const struct lu_env *env, struct md_object *obj,
1455                        struct md_attr *ma)
1456 {
1457         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
1458         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1459         struct mdd_device *mdd = mdo2mdd(obj);
1460         struct thandle *handle;
1461 #ifdef HAVE_QUOTA_SUPPORT
1462         struct obd_device *obd = mdd->mdd_obd_dev;
1463         struct mds_obd *mds = &obd->u.mds;
1464         unsigned int qids[MAXQUOTAS] = { 0, 0 };
1465         int quota_opc = 0;
1466 #endif
1467         int rc;
1468         ENTRY;
1469
1470         /*
1471          * Check -ENOENT early here because we need to get object type
1472          * to calculate credits before transaction start
1473          */
1474         if (!mdd_object_exists(mdd_obj))
1475                 RETURN(-ENOENT);
1476
1477         LASSERT(mdd_object_exists(mdd_obj) > 0);
1478
1479         rc = mdd_log_txn_param_build(env, obj, ma, MDD_TXN_UNLINK_OP);
1480         if (rc)
1481                 RETURN(rc);
1482
1483         handle = mdd_trans_start(env, mdd);
1484         if (IS_ERR(handle))
1485                 RETURN(-ENOMEM);
1486
1487         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1488
1489         rc = mdd_unlink_sanity_check(env, NULL, mdd_obj, ma);
1490         if (rc)
1491                 GOTO(cleanup, rc);
1492
1493         __mdd_ref_del(env, mdd_obj, handle, 0);
1494
1495         if (S_ISDIR(lu_object_attr(&obj->mo_lu))) {
1496                 /* unlink dot */
1497                 __mdd_ref_del(env, mdd_obj, handle, 1);
1498         }
1499
1500         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1501         la_copy->la_ctime = ma->ma_attr.la_ctime;
1502
1503         la_copy->la_valid = LA_CTIME;
1504         rc = mdd_attr_check_set_internal(env, mdd_obj, la_copy, handle, 0);
1505         if (rc)
1506                 GOTO(cleanup, rc);
1507
1508         rc = mdd_finish_unlink(env, mdd_obj, ma, handle);
1509 #ifdef HAVE_QUOTA_SUPPORT
1510         if (mds->mds_quota && ma->ma_valid & MA_INODE &&
1511             ma->ma_attr.la_nlink == 0 && mdd_obj->mod_count == 0) {
1512                 quota_opc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
1513                 mdd_quota_wrapper(&ma->ma_attr, qids);
1514         }
1515 #endif
1516
1517
1518         EXIT;
1519 cleanup:
1520         mdd_write_unlock(env, mdd_obj);
1521         mdd_trans_stop(env, mdd, rc, handle);
1522 #ifdef HAVE_QUOTA_SUPPORT
1523         if (quota_opc)
1524                 /* Trigger dqrel on the owner of child. If failed,
1525                  * the next call for lquota_chkquota will process it */
1526                 lquota_adjust(mds_quota_interface_ref, obd, qids, 0, rc,
1527                               quota_opc);
1528 #endif
1529         return rc;
1530 }
1531
1532 /* partial operation */
1533 static int mdd_oc_sanity_check(const struct lu_env *env,
1534                                struct mdd_object *obj,
1535                                struct md_attr *ma)
1536 {
1537         int rc;
1538         ENTRY;
1539
1540         switch (ma->ma_attr.la_mode & S_IFMT) {
1541         case S_IFREG:
1542         case S_IFDIR:
1543         case S_IFLNK:
1544         case S_IFCHR:
1545         case S_IFBLK:
1546         case S_IFIFO:
1547         case S_IFSOCK:
1548                 rc = 0;
1549                 break;
1550         default:
1551                 rc = -EINVAL;
1552                 break;
1553         }
1554         RETURN(rc);
1555 }
1556
1557 static int mdd_object_create(const struct lu_env *env,
1558                              struct md_object *obj,
1559                              const struct md_op_spec *spec,
1560                              struct md_attr *ma)
1561 {
1562
1563         struct mdd_device *mdd = mdo2mdd(obj);
1564         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1565         const struct lu_fid *pfid = spec->u.sp_pfid;
1566         struct thandle *handle;
1567 #ifdef HAVE_QUOTA_SUPPORT
1568         struct obd_device *obd = mdd->mdd_obd_dev;
1569         struct mds_obd *mds = &obd->u.mds;
1570         unsigned int qids[MAXQUOTAS] = { 0, 0 };
1571         int quota_opc = 0, block_count = 0;
1572         int inode_pending = 0, block_pending = 0;
1573 #endif
1574         int rc = 0;
1575         ENTRY;
1576
1577 #ifdef HAVE_QUOTA_SUPPORT
1578         if (mds->mds_quota) {
1579                 quota_opc = FSFILT_OP_CREATE_PARTIAL_CHILD;
1580                 mdd_quota_wrapper(&ma->ma_attr, qids);
1581                 /* get file quota for child */
1582                 lquota_chkquota(mds_quota_interface_ref, obd, qids[USRQUOTA],
1583                                 qids[GRPQUOTA], 1, &inode_pending, NULL, 0,
1584                                 NULL, 0);
1585                 switch (ma->ma_attr.la_mode & S_IFMT) {
1586                 case S_IFLNK:
1587                 case S_IFDIR:
1588                         block_count = 2;
1589                         break;
1590                 case S_IFREG:
1591                         block_count = 1;
1592                         break;
1593                 }
1594                 /* get block quota for child */
1595                 if (block_count)
1596                         lquota_chkquota(mds_quota_interface_ref, obd,
1597                                         qids[USRQUOTA], qids[GRPQUOTA],
1598                                         block_count, &block_pending, NULL,
1599                                         LQUOTA_FLAGS_BLK, NULL, 0);
1600         }
1601 #endif
1602
1603         mdd_txn_param_build(env, mdd, MDD_TXN_OBJECT_CREATE_OP);
1604         handle = mdd_trans_start(env, mdd);
1605         if (IS_ERR(handle))
1606                 GOTO(out_pending, rc = PTR_ERR(handle));
1607
1608         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1609         rc = mdd_oc_sanity_check(env, mdd_obj, ma);
1610         if (rc)
1611                 GOTO(unlock, rc);
1612
1613         rc = mdd_object_create_internal(env, NULL, mdd_obj, ma, handle, spec);
1614         if (rc)
1615                 GOTO(unlock, rc);
1616
1617         if (spec->sp_cr_flags & MDS_CREATE_SLAVE_OBJ) {
1618                 /* If creating the slave object, set slave EA here. */
1619                 int lmv_size = spec->u.sp_ea.eadatalen;
1620                 struct lmv_stripe_md *lmv;
1621
1622                 lmv = (struct lmv_stripe_md *)spec->u.sp_ea.eadata;
1623                 LASSERT(lmv != NULL && lmv_size > 0);
1624
1625                 rc = __mdd_xattr_set(env, mdd_obj,
1626                                      mdd_buf_get_const(env, lmv, lmv_size),
1627                                      XATTR_NAME_LMV, 0, handle);
1628                 if (rc)
1629                         GOTO(unlock, rc);
1630
1631                 rc = mdd_attr_set_internal(env, mdd_obj, &ma->ma_attr,
1632                                            handle, 0);
1633         } else {
1634 #ifdef CONFIG_FS_POSIX_ACL
1635                 if (spec->sp_cr_flags & MDS_CREATE_RMT_ACL) {
1636                         struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
1637
1638                         buf->lb_buf = (void *)spec->u.sp_ea.eadata;
1639                         buf->lb_len = spec->u.sp_ea.eadatalen;
1640                         if ((buf->lb_len > 0) && (buf->lb_buf != NULL)) {
1641                                 rc = __mdd_acl_init(env, mdd_obj, buf,
1642                                                     &ma->ma_attr.la_mode,
1643                                                     handle);
1644                                 if (rc)
1645                                         GOTO(unlock, rc);
1646                                 else
1647                                         ma->ma_attr.la_valid |= LA_MODE;
1648                         }
1649
1650                         pfid = spec->u.sp_ea.fid;
1651                 }
1652 #endif
1653                 rc = mdd_object_initialize(env, pfid, NULL, mdd_obj, ma, handle,
1654                                            spec);
1655         }
1656         EXIT;
1657 unlock:
1658         if (rc == 0)
1659                 rc = mdd_attr_get_internal(env, mdd_obj, ma);
1660         mdd_write_unlock(env, mdd_obj);
1661
1662         mdd_trans_stop(env, mdd, rc, handle);
1663 out_pending:
1664 #ifdef HAVE_QUOTA_SUPPORT
1665         if (quota_opc) {
1666                 if (inode_pending)
1667                         lquota_pending_commit(mds_quota_interface_ref, obd,
1668                                               qids[USRQUOTA], qids[GRPQUOTA],
1669                                               inode_pending, 0);
1670                 if (block_pending)
1671                         lquota_pending_commit(mds_quota_interface_ref, obd,
1672                                               qids[USRQUOTA], qids[GRPQUOTA],
1673                                               block_pending, 1);
1674                 /* Trigger dqacq on the owner of child. If failed,
1675                  * the next call for lquota_chkquota will process it. */
1676                 lquota_adjust(mds_quota_interface_ref, obd, qids, 0, rc,
1677                               FSFILT_OP_CREATE_PARTIAL_CHILD);
1678         }
1679 #endif
1680         return rc;
1681 }
1682
1683 /* partial link */
1684 static int mdd_ref_add(const struct lu_env *env, struct md_object *obj,
1685                        const struct md_attr *ma)
1686 {
1687         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
1688         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1689         struct mdd_device *mdd = mdo2mdd(obj);
1690         struct thandle *handle;
1691         int rc;
1692         ENTRY;
1693
1694         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
1695         handle = mdd_trans_start(env, mdd);
1696         if (IS_ERR(handle))
1697                 RETURN(-ENOMEM);
1698
1699         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1700         rc = mdd_link_sanity_check(env, NULL, NULL, mdd_obj);
1701         if (rc == 0)
1702                 __mdd_ref_add(env, mdd_obj, handle);
1703         mdd_write_unlock(env, mdd_obj);
1704         if (rc == 0) {
1705                 LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1706                 la_copy->la_ctime = ma->ma_attr.la_ctime;
1707
1708                 la_copy->la_valid = LA_CTIME;
1709                 rc = mdd_attr_check_set_internal_locked(env, mdd_obj, la_copy,
1710                                                         handle, 0);
1711         }
1712         mdd_trans_stop(env, mdd, 0, handle);
1713
1714         RETURN(rc);
1715 }
1716
1717 /*
1718  * do NOT or the MAY_*'s, you'll get the weakest
1719  */
1720 int accmode(const struct lu_env *env, struct lu_attr *la, int flags)
1721 {
1722         int res = 0;
1723
1724         /* Sadly, NFSD reopens a file repeatedly during operation, so the
1725          * "acc_mode = 0" allowance for newly-created files isn't honoured.
1726          * NFSD uses the MDS_OPEN_OWNEROVERRIDE flag to say that a file
1727          * owner can write to a file even if it is marked readonly to hide
1728          * its brokenness. (bug 5781) */
1729         if (flags & MDS_OPEN_OWNEROVERRIDE) {
1730                 struct md_ucred *uc = md_ucred(env);
1731
1732                 if ((uc == NULL) || (uc->mu_valid == UCRED_INIT) ||
1733                     (la->la_uid == uc->mu_fsuid))
1734                         return 0;
1735         }
1736
1737         if (flags & FMODE_READ)
1738                 res |= MAY_READ;
1739         if (flags & (FMODE_WRITE | MDS_OPEN_TRUNC | MDS_OPEN_APPEND))
1740                 res |= MAY_WRITE;
1741         if (flags & MDS_FMODE_EXEC)
1742                 res |= MAY_EXEC;
1743         return res;
1744 }
1745
1746 static int mdd_open_sanity_check(const struct lu_env *env,
1747                                  struct mdd_object *obj, int flag)
1748 {
1749         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
1750         int mode, rc;
1751         ENTRY;
1752
1753         /* EEXIST check */
1754         if (mdd_is_dead_obj(obj))
1755                 RETURN(-ENOENT);
1756
1757         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
1758         if (rc)
1759                RETURN(rc);
1760
1761         if (S_ISLNK(tmp_la->la_mode))
1762                 RETURN(-ELOOP);
1763
1764         mode = accmode(env, tmp_la, flag);
1765
1766         if (S_ISDIR(tmp_la->la_mode) && (mode & MAY_WRITE))
1767                 RETURN(-EISDIR);
1768
1769         if (!(flag & MDS_OPEN_CREATED)) {
1770                 rc = mdd_permission_internal(env, obj, tmp_la, mode);
1771                 if (rc)
1772                         RETURN(rc);
1773         }
1774
1775         if (S_ISFIFO(tmp_la->la_mode) || S_ISSOCK(tmp_la->la_mode) ||
1776             S_ISBLK(tmp_la->la_mode) || S_ISCHR(tmp_la->la_mode))
1777                 flag &= ~MDS_OPEN_TRUNC;
1778
1779         /* For writing append-only file must open it with append mode. */
1780         if (mdd_is_append(obj)) {
1781                 if ((flag & FMODE_WRITE) && !(flag & MDS_OPEN_APPEND))
1782                         RETURN(-EPERM);
1783                 if (flag & MDS_OPEN_TRUNC)
1784                         RETURN(-EPERM);
1785         }
1786
1787 #if 0
1788         /*
1789          * Now, flag -- O_NOATIME does not be packed by client.
1790          */
1791         if (flag & O_NOATIME) {
1792                 struct md_ucred *uc = md_ucred(env);
1793
1794                 if (uc && ((uc->mu_valid == UCRED_OLD) ||
1795                     (uc->mu_valid == UCRED_NEW)) &&
1796                     (uc->mu_fsuid != tmp_la->la_uid) &&
1797                     !mdd_capable(uc, CFS_CAP_FOWNER))
1798                         RETURN(-EPERM);
1799         }
1800 #endif
1801
1802         RETURN(0);
1803 }
1804
1805 static int mdd_open(const struct lu_env *env, struct md_object *obj,
1806                     int flags)
1807 {
1808         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1809         int rc = 0;
1810
1811         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1812
1813         rc = mdd_open_sanity_check(env, mdd_obj, flags);
1814         if (rc == 0)
1815                 mdd_obj->mod_count++;
1816
1817         mdd_write_unlock(env, mdd_obj);
1818         return rc;
1819 }
1820
1821 /* return md_attr back,
1822  * if it is last unlink then return lov ea + llog cookie*/
1823 int mdd_object_kill(const struct lu_env *env, struct mdd_object *obj,
1824                     struct md_attr *ma)
1825 {
1826         int rc = 0;
1827         ENTRY;
1828
1829         if (S_ISREG(mdd_object_type(obj))) {
1830                 /* Return LOV & COOKIES unconditionally here. We clean evth up.
1831                  * Caller must be ready for that. */
1832
1833                 rc = __mdd_lmm_get(env, obj, ma);
1834                 if ((ma->ma_valid & MA_LOV))
1835                         rc = mdd_unlink_log(env, mdo2mdd(&obj->mod_obj),
1836                                             obj, ma);
1837         }
1838         RETURN(rc);
1839 }
1840
1841 /*
1842  * No permission check is needed.
1843  */
1844 static int mdd_close(const struct lu_env *env, struct md_object *obj,
1845                      struct md_attr *ma)
1846 {
1847         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1848         struct thandle    *handle;
1849         int rc;
1850         int reset = 1;
1851
1852 #ifdef HAVE_QUOTA_SUPPORT
1853         struct obd_device *obd = mdo2mdd(obj)->mdd_obd_dev;
1854         struct mds_obd *mds = &obd->u.mds;
1855         unsigned int qids[MAXQUOTAS] = { 0, 0 };
1856         int quota_opc = 0;
1857 #endif
1858         ENTRY;
1859
1860         rc = mdd_log_txn_param_build(env, obj, ma, MDD_TXN_UNLINK_OP);
1861         if (rc)
1862                 RETURN(rc);
1863         handle = mdd_trans_start(env, mdo2mdd(obj));
1864         if (IS_ERR(handle))
1865                 RETURN(PTR_ERR(handle));
1866
1867         mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD);
1868         /* release open count */
1869         mdd_obj->mod_count --;
1870
1871         if (mdd_obj->mod_count == 0) {
1872                 /* remove link to object from orphan index */
1873                 if (mdd_obj->mod_flags & ORPHAN_OBJ)
1874                         __mdd_orphan_del(env, mdd_obj, handle);
1875         }
1876
1877         rc = mdd_iattr_get(env, mdd_obj, ma);
1878         if (rc == 0) {
1879                 if (mdd_obj->mod_count == 0 && ma->ma_attr.la_nlink == 0) {
1880                         rc = mdd_object_kill(env, mdd_obj, ma);
1881 #ifdef HAVE_QUOTA_SUPPORT
1882                         if (mds->mds_quota) {
1883                                 quota_opc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
1884                                 mdd_quota_wrapper(&ma->ma_attr, qids);
1885                         }
1886 #endif
1887                         if (rc == 0)
1888                                 reset = 0;
1889                 }
1890         }
1891
1892         if (reset)
1893                 ma->ma_valid &= ~(MA_LOV | MA_COOKIE);
1894
1895         mdd_write_unlock(env, mdd_obj);
1896         mdd_trans_stop(env, mdo2mdd(obj), rc, handle);
1897 #ifdef HAVE_QUOTA_SUPPORT
1898         if (quota_opc)
1899                 /* Trigger dqrel on the owner of child. If failed,
1900                  * the next call for lquota_chkquota will process it */
1901                 lquota_adjust(mds_quota_interface_ref, obd, qids, 0, rc,
1902                               quota_opc);
1903 #endif
1904         RETURN(rc);
1905 }
1906
1907 /*
1908  * Permission check is done when open,
1909  * no need check again.
1910  */
1911 static int mdd_readpage_sanity_check(const struct lu_env *env,
1912                                      struct mdd_object *obj)
1913 {
1914         struct dt_object *next = mdd_object_child(obj);
1915         int rc;
1916         ENTRY;
1917
1918         if (S_ISDIR(mdd_object_type(obj)) && dt_try_as_dir(env, next))
1919                 rc = 0;
1920         else
1921                 rc = -ENOTDIR;
1922
1923         RETURN(rc);
1924 }
1925
1926 static int mdd_append_attrs(const struct lu_env *env,
1927                              struct mdd_device *mdd,
1928                              __u32 attr,
1929                              const struct dt_it_ops *iops,
1930                              struct dt_it *it,
1931                              struct lu_dirent*ent)
1932 {
1933         struct mdd_thread_info  *info = mdd_env_info(env);
1934         struct lu_fid           *fid  = &info->mti_fid2;
1935         int                      len = cpu_to_le16(ent->lde_namelen);
1936         const unsigned           align = sizeof(struct luda_type) - 1;
1937         struct lu_fid_pack      *pack;
1938         struct mdd_object       *obj;
1939         struct luda_type        *lt;
1940         int rc = 0;
1941
1942         if (attr & LUDA_FID) {
1943                 pack = (struct lu_fid_pack *)iops->rec(env, it);
1944                 if (IS_ERR(pack)) {
1945                         rc = PTR_ERR(pack);
1946                         ent->lde_attrs = 0;
1947                         goto out;
1948                 }
1949                 rc = fid_unpack(pack, fid);
1950                 if (rc != 0) {
1951                         ent->lde_attrs = 0;
1952                         goto out;
1953                 }
1954
1955                 fid_cpu_to_le(&ent->lde_fid, fid);
1956                 ent->lde_attrs = LUDA_FID;
1957         }
1958
1959         /* check if file type is required */
1960         if (attr & LUDA_TYPE) {
1961                 if (!(attr & LUDA_FID)) {
1962                         CERROR("wrong attr : [%x]\n",attr);
1963                         rc = -EINVAL;
1964                         goto out;
1965                 }
1966
1967                 obj = mdd_object_find(env, mdd, fid);
1968                 if (obj == NULL) /* remote object */
1969                         goto out;
1970
1971                 if (IS_ERR(obj)) {
1972                         rc = PTR_ERR(obj);
1973                         goto out;
1974                 }
1975
1976                 if (mdd_object_exists(obj) == +1) {
1977                         len = (len + align) & ~align;
1978
1979                         lt = (void *) ent->lde_name + len;
1980                         lt->lt_type = cpu_to_le16(mdd_object_type(obj));
1981
1982                         ent->lde_attrs |= LUDA_TYPE;
1983                 }
1984                 mdd_object_put(env, obj);
1985         }
1986 out:
1987         ent->lde_attrs = cpu_to_le32(ent->lde_attrs);
1988         return rc;
1989 }
1990
1991 static int mdd_dir_page_build(const struct lu_env *env, struct mdd_device *mdd,
1992                               int first, void *area, int nob,
1993                               const struct dt_it_ops *iops, struct dt_it *it,
1994                               __u64 *start, __u64 *end,
1995                               struct lu_dirent **last, __u32 attr)
1996 {
1997         int                     result;
1998         struct lu_dirent       *ent;
1999         __u64  hash = 0;
2000
2001         if (first) {
2002                 memset(area, 0, sizeof (struct lu_dirpage));
2003                 area += sizeof (struct lu_dirpage);
2004                 nob  -= sizeof (struct lu_dirpage);
2005         }
2006
2007         ent  = area;
2008         do {
2009                 char  *name;
2010                 int    len;
2011                 int    recsize;
2012
2013                 len  = iops->key_size(env, it);
2014
2015                 /* IAM iterator can return record with zero len. */
2016                 if (len == 0)
2017                         goto next;
2018
2019                 name = (char *)iops->key(env, it);
2020                 hash = iops->store(env, it);
2021
2022                 if (unlikely(first)) {
2023                         first = 0;
2024                         *start = hash;
2025                 }
2026
2027                 recsize = lu_dirent_calc_size(len, attr);
2028
2029                 CDEBUG(D_INFO, "%p %p %d "LPU64" (%d) \"%*.*s\"\n",
2030                                 name, ent, nob, hash, len, len, len, name);
2031
2032                 if (nob >= recsize) {
2033                         ent->lde_hash    = cpu_to_le64(hash);
2034                         ent->lde_namelen = cpu_to_le16(len);
2035                         ent->lde_reclen  = cpu_to_le16(recsize);
2036                         memcpy(ent->lde_name, name, len);
2037
2038                         result = mdd_append_attrs(env, mdd, attr, iops, it, ent);
2039                         if (result != 0)
2040                                 goto out;
2041                 } else {
2042                         /*
2043                          * record doesn't fit into page, enlarge previous one.
2044                          */
2045                         if (*last) {
2046                                 (*last)->lde_reclen =
2047                                         cpu_to_le16(le16_to_cpu((*last)->lde_reclen) +
2048                                                         nob);
2049                                 result = 0;
2050                         } else
2051                                 result = -EINVAL;
2052
2053                         goto out;
2054                 }
2055                 *last = ent;
2056                 ent = (void *)ent + recsize;
2057                 nob -= recsize;
2058
2059 next:
2060                 result = iops->next(env, it);
2061         } while (result == 0);
2062
2063 out:
2064         *end = hash;
2065         return result;
2066 }
2067
2068 static int __mdd_readpage(const struct lu_env *env, struct mdd_object *obj,
2069                           const struct lu_rdpg *rdpg)
2070 {
2071         struct dt_it      *it;
2072         struct dt_object  *next = mdd_object_child(obj);
2073         const struct dt_it_ops  *iops;
2074         struct page       *pg;
2075         struct lu_dirent  *last = NULL;
2076         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
2077         int i;
2078         int rc;
2079         int nob;
2080         __u64 hash_start;
2081         __u64 hash_end = 0;
2082
2083         LASSERT(rdpg->rp_pages != NULL);
2084         LASSERT(next->do_index_ops != NULL);
2085
2086         if (rdpg->rp_count <= 0)
2087                 return -EFAULT;
2088
2089         /*
2090          * iterate through directory and fill pages from @rdpg
2091          */
2092         iops = &next->do_index_ops->dio_it;
2093         it = iops->init(env, next, mdd_object_capa(env, obj));
2094         if (IS_ERR(it))
2095                 return PTR_ERR(it);
2096
2097         rc = iops->load(env, it, rdpg->rp_hash);
2098
2099         if (rc == 0)
2100                 /*
2101                  * Iterator didn't find record with exactly the key requested.
2102                  *
2103                  * It is currently either
2104                  *
2105                  *     - positioned above record with key less than
2106                  *     requested---skip it.
2107                  *
2108                  *     - or not positioned at all (is in IAM_IT_SKEWED
2109                  *     state)---position it on the next item.
2110                  */
2111                 rc = iops->next(env, it);
2112         else if (rc > 0)
2113                 rc = 0;
2114
2115         /*
2116          * At this point and across for-loop:
2117          *
2118          *  rc == 0 -> ok, proceed.
2119          *  rc >  0 -> end of directory.
2120          *  rc <  0 -> error.
2121          */
2122         for (i = 0, nob = rdpg->rp_count; rc == 0 && nob > 0;
2123              i++, nob -= CFS_PAGE_SIZE) {
2124                 LASSERT(i < rdpg->rp_npages);
2125                 pg = rdpg->rp_pages[i];
2126                 rc = mdd_dir_page_build(env, mdd, !i, cfs_kmap(pg),
2127                                         min_t(int, nob, CFS_PAGE_SIZE), iops,
2128                                         it, &hash_start, &hash_end, &last,
2129                                         rdpg->rp_attrs);
2130                 if (rc != 0 || i == rdpg->rp_npages - 1) {
2131                         if (last)
2132                                 last->lde_reclen = 0;
2133                 }
2134                 cfs_kunmap(pg);
2135         }
2136         if (rc > 0) {
2137                 /*
2138                  * end of directory.
2139                  */
2140                 hash_end = DIR_END_OFF;
2141                 rc = 0;
2142         }
2143         if (rc == 0) {
2144                 struct lu_dirpage *dp;
2145
2146                 dp = cfs_kmap(rdpg->rp_pages[0]);
2147                 dp->ldp_hash_start = cpu_to_le64(rdpg->rp_hash);
2148                 dp->ldp_hash_end   = cpu_to_le64(hash_end);
2149                 if (i == 0)
2150                         /*
2151                          * No pages were processed, mark this.
2152                          */
2153                         dp->ldp_flags |= LDF_EMPTY;
2154
2155                 dp->ldp_flags = cpu_to_le32(dp->ldp_flags);
2156                 cfs_kunmap(rdpg->rp_pages[0]);
2157         }
2158         iops->put(env, it);
2159         iops->fini(env, it);
2160
2161         return rc;
2162 }
2163
2164 int mdd_readpage(const struct lu_env *env, struct md_object *obj,
2165                  const struct lu_rdpg *rdpg)
2166 {
2167         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2168         int rc;
2169         ENTRY;
2170
2171         LASSERT(mdd_object_exists(mdd_obj));
2172
2173         mdd_read_lock(env, mdd_obj, MOR_TGT_CHILD);
2174         rc = mdd_readpage_sanity_check(env, mdd_obj);
2175         if (rc)
2176                 GOTO(out_unlock, rc);
2177
2178         if (mdd_is_dead_obj(mdd_obj)) {
2179                 struct page *pg;
2180                 struct lu_dirpage *dp;
2181
2182                 /*
2183                  * According to POSIX, please do not return any entry to client:
2184                  * even dot and dotdot should not be returned.
2185                  */
2186                 CWARN("readdir from dead object: "DFID"\n",
2187                         PFID(mdd_object_fid(mdd_obj)));
2188
2189                 if (rdpg->rp_count <= 0)
2190                         GOTO(out_unlock, rc = -EFAULT);
2191                 LASSERT(rdpg->rp_pages != NULL);
2192
2193                 pg = rdpg->rp_pages[0];
2194                 dp = (struct lu_dirpage*)cfs_kmap(pg);
2195                 memset(dp, 0 , sizeof(struct lu_dirpage));
2196                 dp->ldp_hash_start = cpu_to_le64(rdpg->rp_hash);
2197                 dp->ldp_hash_end   = cpu_to_le64(DIR_END_OFF);
2198                 dp->ldp_flags |= LDF_EMPTY;
2199                 dp->ldp_flags = cpu_to_le32(dp->ldp_flags);
2200                 cfs_kunmap(pg);
2201                 GOTO(out_unlock, rc = 0);
2202         }
2203
2204         rc = __mdd_readpage(env, mdd_obj, rdpg);
2205
2206         EXIT;
2207 out_unlock:
2208         mdd_read_unlock(env, mdd_obj);
2209         return rc;
2210 }
2211
2212 static int mdd_object_sync(const struct lu_env *env, struct md_object *obj)
2213 {
2214         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2215         struct dt_object *next;
2216
2217         LASSERT(mdd_object_exists(mdd_obj));
2218         next = mdd_object_child(mdd_obj);
2219         return next->do_ops->do_object_sync(env, next);
2220 }
2221
2222 const struct md_object_operations mdd_obj_ops = {
2223         .moo_permission    = mdd_permission,
2224         .moo_attr_get      = mdd_attr_get,
2225         .moo_attr_set      = mdd_attr_set,
2226         .moo_xattr_get     = mdd_xattr_get,
2227         .moo_xattr_set     = mdd_xattr_set,
2228         .moo_xattr_list    = mdd_xattr_list,
2229         .moo_xattr_del     = mdd_xattr_del,
2230         .moo_object_create = mdd_object_create,
2231         .moo_ref_add       = mdd_ref_add,
2232         .moo_ref_del       = mdd_ref_del,
2233         .moo_open          = mdd_open,
2234         .moo_close         = mdd_close,
2235         .moo_readpage      = mdd_readpage,
2236         .moo_readlink      = mdd_readlink,
2237         .moo_capa_get      = mdd_capa_get,
2238         .moo_object_sync   = mdd_object_sync,
2239         .moo_path          = mdd_path,
2240 };