lustre/mdd/mdd_object.c

   1 /* -*- MODE: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  mdd/mdd_handler.c
   5  *  Lustre Metadata Server (mdd) routines
   6  *
   7  *  Copyright (C) 2006 Cluster File Systems, Inc.
   8  *   Author: Wang Di <wangdi@clusterfs.com>
   9  *
  10  *   This file is part of the Lustre file system, http://www.lustre.org
  11  *   Lustre is a trademark of Cluster File Systems, Inc.
  12  *
  13  *   You may have signed or agreed to another license before downloading
  14  *   this software.  If so, you are bound by the terms and conditions
  15  *   of that agreement, and the following does not apply to you.  See the
  16  *   LICENSE file included with this distribution for more information.
  17  *
  18  *   If you did not agree to a different license, then this copy of Lustre
  19  *   is open source software; you can redistribute it and/or modify it
  20  *   under the terms of version 2 of the GNU General Public License as
  21  *   published by the Free Software Foundation.
  22  *
  23  *   In either case, Lustre is distributed in the hope that it will be
  24  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  25  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  26  *   license text for more details.
  27  */
  28 #ifndef EXPORT_SYMTAB
  29 # define EXPORT_SYMTAB
  30 #endif
  31 #define DEBUG_SUBSYSTEM S_MDS
  32
  33 #include <linux/module.h>
  34 #include <linux/jbd.h>
  35 #include <obd.h>
  36 #include <obd_class.h>
  37 #include <lustre_ver.h>
  38 #include <obd_support.h>
  39 #include <lprocfs_status.h>
  40 /* fid_be_cpu(), fid_cpu_to_be(). */
  41 #include <lustre_fid.h>
  42
  43 #include <linux/ldiskfs_fs.h>
  44 #include <lustre_mds.h>
  45 #include <lustre/lustre_idl.h>
  46
  47 #include "mdd_internal.h"
  48
  49 static struct lu_object_operations mdd_lu_obj_ops;
  50
  51 static inline void mdd_set_dead_obj(struct mdd_object *obj)
  52 {
  53         if (obj)
  54                 obj->mod_flags |= DEAD_OBJ;
  55 }
  56
  57 int mdd_la_get(const struct lu_env *env, struct mdd_object *obj,
  58                struct lu_attr *la, struct lustre_capa *capa)
  59 {
  60         struct dt_object *next = mdd_object_child(obj);
  61         LASSERT(lu_object_exists(mdd2lu_obj(obj)));
  62         return next->do_ops->do_attr_get(env, next, la, capa);
  63 }
  64
  65 static void mdd_flags_xlate(struct mdd_object *obj, __u32 flags)
  66 {
  67         obj->mod_flags &= ~(APPEND_OBJ|IMMUTE_OBJ);
  68
  69         if (flags & LUSTRE_APPEND_FL)
  70                 obj->mod_flags |= APPEND_OBJ;
  71
  72         if (flags & LUSTRE_IMMUTABLE_FL)
  73                 obj->mod_flags |= IMMUTE_OBJ;
  74 }
  75
  76 struct lu_buf *mdd_buf_get(const struct lu_env *env, void *area, ssize_t len)
  77 {
  78         struct lu_buf *buf;
  79
  80         buf = &mdd_env_info(env)->mti_buf;
  81         buf->lb_buf = area;
  82         buf->lb_len = len;
  83         return buf;
  84 }
  85
  86 const struct lu_buf *mdd_buf_get_const(const struct lu_env *env,
  87                                        const void *area, ssize_t len)
  88 {
  89         struct lu_buf *buf;
  90
  91         buf = &mdd_env_info(env)->mti_buf;
  92         buf->lb_buf = (void *)area;
  93         buf->lb_len = len;
  94         return buf;
  95 }
  96
  97 struct mdd_thread_info *mdd_env_info(const struct lu_env *env)
  98 {
  99         struct mdd_thread_info *info;
 100
 101         info = lu_context_key_get(&env->le_ctx, &mdd_thread_key);
 102         LASSERT(info != NULL);
 103         return info;
 104 }
 105
 106 struct lu_object *mdd_object_alloc(const struct lu_env *env,
 107                                    const struct lu_object_header *hdr,
 108                                    struct lu_device *d)
 109 {
 110         struct mdd_object *mdd_obj;
 111
 112         OBD_ALLOC_PTR(mdd_obj);
 113         if (mdd_obj != NULL) {
 114                 struct lu_object *o;
 115
 116                 o = mdd2lu_obj(mdd_obj);
 117                 lu_object_init(o, NULL, d);
 118                 mdd_obj->mod_obj.mo_ops = &mdd_obj_ops;
 119                 mdd_obj->mod_obj.mo_dir_ops = &mdd_dir_ops;
 120                 mdd_obj->mod_count = 0;
 121                 o->lo_ops = &mdd_lu_obj_ops;
 122                 return o;
 123         } else {
 124                 return NULL;
 125         }
 126 }
 127
 128 static int mdd_object_init(const struct lu_env *env, struct lu_object *o)
 129 {
 130         struct mdd_device *d = lu2mdd_dev(o->lo_dev);
 131         struct lu_object  *below;
 132         struct lu_device  *under;
 133         ENTRY;
 134
 135         under = &d->mdd_child->dd_lu_dev;
 136         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
 137         mdd_pdlock_init(lu2mdd_obj(o));
 138         if (below == NULL)
 139                 RETURN(-ENOMEM);
 140
 141         lu_object_add(o, below);
 142         RETURN(0);
 143 }
 144
 145 static int mdd_object_start(const struct lu_env *env, struct lu_object *o)
 146 {
 147         if (lu_object_exists(o))
 148                 return mdd_get_flags(env, lu2mdd_obj(o));
 149         else
 150                 return 0;
 151 }
 152
 153 static void mdd_object_free(const struct lu_env *env, struct lu_object *o)
 154 {
 155         struct mdd_object *mdd = lu2mdd_obj(o);
 156
 157         lu_object_fini(o);
 158         OBD_FREE_PTR(mdd);
 159 }
 160
 161 static int mdd_object_print(const struct lu_env *env, void *cookie,
 162                             lu_printer_t p, const struct lu_object *o)
 163 {
 164         return (*p)(env, cookie, LUSTRE_MDD_NAME"-object@%p", o);
 165 }
 166
 167 /* orphan handling is here */
 168 static void mdd_object_delete(const struct lu_env *env,
 169                                struct lu_object *o)
 170 {
 171         struct mdd_object *mdd_obj = lu2mdd_obj(o);
 172         struct thandle *handle = NULL;
 173         ENTRY;
 174
 175         if (lu2mdd_dev(o->lo_dev)->mdd_orphans == NULL)
 176                 return;
 177
 178         if (mdd_obj->mod_flags & ORPHAN_OBJ) {
 179                 mdd_txn_param_build(env, lu2mdd_dev(o->lo_dev),
 180                                     MDD_TXN_INDEX_DELETE_OP);
 181                 handle = mdd_trans_start(env, lu2mdd_dev(o->lo_dev));
 182                 if (IS_ERR(handle))
 183                         CERROR("Cannot get thandle\n");
 184                 else {
 185                         mdd_write_lock(env, mdd_obj);
 186                         /* let's remove obj from the orphan list */
 187                         __mdd_orphan_del(env, mdd_obj, handle);
 188                         mdd_write_unlock(env, mdd_obj);
 189                         mdd_trans_stop(env, lu2mdd_dev(o->lo_dev),
 190                                        0, handle);
 191                 }
 192         }
 193 }
 194
 195 static struct lu_object_operations mdd_lu_obj_ops = {
 196         .loo_object_init    = mdd_object_init,
 197         .loo_object_start   = mdd_object_start,
 198         .loo_object_free    = mdd_object_free,
 199         .loo_object_print   = mdd_object_print,
 200         .loo_object_delete  = mdd_object_delete
 201 };
 202
 203 struct mdd_object *mdd_object_find(const struct lu_env *env,
 204                                    struct mdd_device *d,
 205                                    const struct lu_fid *f)
 206 {
 207         struct lu_object *o, *lo;
 208         struct mdd_object *m;
 209         ENTRY;
 210
 211         o = lu_object_find(env, mdd2lu_dev(d)->ld_site, f);
 212         if (IS_ERR(o))
 213                 m = (struct mdd_object *)o;
 214         else {
 215                 lo = lu_object_locate(o->lo_header, mdd2lu_dev(d)->ld_type);
 216                 /* remote object can't be located and should be put then */
 217                 if (lo == NULL)
 218                         lu_object_put(env, o);
 219                 m = lu2mdd_obj(lo);
 220         }
 221         RETURN(m);
 222 }
 223
 224 int mdd_get_flags(const struct lu_env *env, struct mdd_object *obj)
 225 {
 226         struct lu_attr *la = &mdd_env_info(env)->mti_la;
 227         int rc;
 228
 229         ENTRY;
 230         mdd_read_lock(env, obj);
 231         rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
 232         mdd_read_unlock(env, obj);
 233         if (rc == 0)
 234                 mdd_flags_xlate(obj, la->la_flags);
 235         RETURN(rc);
 236 }
 237
 238 void mdd_ref_add_internal(const struct lu_env *env, struct mdd_object *obj,
 239                           struct thandle *handle)
 240 {
 241         struct dt_object *next;
 242
 243         LASSERT(lu_object_exists(mdd2lu_obj(obj)));
 244         next = mdd_object_child(obj);
 245         next->do_ops->do_ref_add(env, next, handle);
 246 }
 247
 248 void mdd_ref_del_internal(const struct lu_env *env, struct mdd_object *obj,
 249                           struct thandle *handle)
 250 {
 251         struct dt_object *next = mdd_object_child(obj);
 252         ENTRY;
 253
 254         LASSERT(lu_object_exists(mdd2lu_obj(obj)));
 255
 256         next->do_ops->do_ref_del(env, next, handle);
 257         EXIT;
 258 }
 259
 260 /* get only inode attributes */
 261 int mdd_iattr_get(const struct lu_env *env, struct mdd_object *mdd_obj,
 262                   struct md_attr *ma)
 263 {
 264         int rc = 0;
 265         ENTRY;
 266
 267         if (ma->ma_valid & MA_INODE)
 268                 RETURN(0);
 269
 270         rc = mdd_la_get(env, mdd_obj, &ma->ma_attr,
 271                           mdd_object_capa(env, mdd_obj));
 272         if (rc == 0)
 273                 ma->ma_valid |= MA_INODE;
 274         RETURN(rc);
 275 }
 276
 277 /* get lov EA only */
 278 static int __mdd_lmm_get(const struct lu_env *env,
 279                          struct mdd_object *mdd_obj, struct md_attr *ma)
 280 {
 281         int rc, lmm_size;
 282         ENTRY;
 283
 284         if (ma->ma_valid & MA_LOV)
 285                 RETURN(0);
 286
 287         lmm_size = ma->ma_lmm_size;
 288         rc = mdd_get_md(env, mdd_obj, ma->ma_lmm, &lmm_size,
 289                         MDS_LOV_MD_NAME);
 290         if (rc > 0) {
 291                 ma->ma_valid |= MA_LOV;
 292                 ma->ma_lmm_size = lmm_size;
 293                 rc = 0;
 294         }
 295         RETURN(rc);
 296 }
 297
 298 int mdd_lmm_get_locked(const struct lu_env *env, struct mdd_object *mdd_obj,
 299                        struct md_attr *ma)
 300 {
 301         int rc;
 302         ENTRY;
 303
 304         mdd_read_lock(env, mdd_obj);
 305         rc = __mdd_lmm_get(env, mdd_obj, ma);
 306         mdd_read_unlock(env, mdd_obj);
 307         RETURN(rc);
 308 }
 309
 310 /* get lmv EA only*/
 311 static int __mdd_lmv_get(const struct lu_env *env,
 312                          struct mdd_object *mdd_obj, struct md_attr *ma)
 313 {
 314         int rc;
 315
 316         if (ma->ma_valid & MA_LMV)
 317                 RETURN(0);
 318
 319         rc = mdd_get_md(env, mdd_obj, ma->ma_lmv, &ma->ma_lmv_size,
 320                         MDS_LMV_MD_NAME);
 321         if (rc > 0) {
 322                 ma->ma_valid |= MA_LMV;
 323                 rc = 0;
 324         }
 325         RETURN(rc);
 326 }
 327
 328 static int mdd_attr_get_internal(const struct lu_env *env,
 329                                  struct mdd_object *mdd_obj,
 330                                  struct md_attr *ma)
 331 {
 332         int rc = 0;
 333         ENTRY;
 334
 335         if (ma->ma_need & MA_INODE)
 336                 rc = mdd_iattr_get(env, mdd_obj, ma);
 337
 338         if (rc == 0 && ma->ma_need & MA_LOV) {
 339                 if (S_ISREG(mdd_object_type(mdd_obj)) ||
 340                     S_ISDIR(mdd_object_type(mdd_obj)))
 341                         rc = __mdd_lmm_get(env, mdd_obj, ma);
 342         }
 343         if (rc == 0 && ma->ma_need & MA_LMV) {
 344                 if (S_ISDIR(mdd_object_type(mdd_obj)))
 345                         rc = __mdd_lmv_get(env, mdd_obj, ma);
 346         }
 347 #ifdef CONFIG_FS_POSIX_ACL
 348         if (rc == 0 && ma->ma_need & MA_ACL_DEF) {
 349                 if (S_ISDIR(mdd_object_type(mdd_obj)))
 350                         rc = mdd_acl_def_get(env, mdd_obj, ma);
 351         }
 352 #endif
 353         CDEBUG(D_INODE, "after getattr rc = %d, ma_valid = "LPX64"\n",
 354                rc, ma->ma_valid);
 355         RETURN(rc);
 356 }
 357
 358 int mdd_attr_get_internal_locked(const struct lu_env *env,
 359                                  struct mdd_object *mdd_obj, struct md_attr *ma)
 360 {
 361         int rc;
 362         mdd_read_lock(env, mdd_obj);
 363         rc = mdd_attr_get_internal(env, mdd_obj, ma);
 364         mdd_read_unlock(env, mdd_obj);
 365         return rc;
 366 }
 367
 368 /*
 369  * No permission check is needed.
 370  */
 371 static int mdd_attr_get(const struct lu_env *env, struct md_object *obj,
 372                         struct md_attr *ma)
 373 {
 374         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 375         int                rc;
 376
 377         ENTRY;
 378         rc = mdd_attr_get_internal_locked(env, mdd_obj, ma);
 379         RETURN(rc);
 380 }
 381
 382 /*
 383  * No permission check is needed.
 384  */
 385 static int mdd_xattr_get(const struct lu_env *env,
 386                          struct md_object *obj, struct lu_buf *buf,
 387                          const char *name)
 388 {
 389         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 390         struct dt_object  *next;
 391         int rc;
 392
 393         ENTRY;
 394
 395         LASSERT(lu_object_exists(&obj->mo_lu));
 396
 397         next = mdd_object_child(mdd_obj);
 398         mdd_read_lock(env, mdd_obj);
 399         rc = next->do_ops->do_xattr_get(env, next, buf, name,
 400                                         mdd_object_capa(env, mdd_obj));
 401         mdd_read_unlock(env, mdd_obj);
 402
 403         RETURN(rc);
 404 }
 405
 406 /*
 407  * Permission check is done when open,
 408  * no need check again.
 409  */
 410 static int mdd_readlink(const struct lu_env *env, struct md_object *obj,
 411                         struct lu_buf *buf)
 412 {
 413         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 414         struct dt_object  *next;
 415         loff_t             pos = 0;
 416         int                rc;
 417         ENTRY;
 418
 419         LASSERT(lu_object_exists(&obj->mo_lu));
 420
 421         next = mdd_object_child(mdd_obj);
 422         mdd_read_lock(env, mdd_obj);
 423         rc = next->do_body_ops->dbo_read(env, next, buf, &pos,
 424                                          mdd_object_capa(env, mdd_obj));
 425         mdd_read_unlock(env, mdd_obj);
 426         RETURN(rc);
 427 }
 428
 429 static int mdd_xattr_list(const struct lu_env *env, struct md_object *obj,
 430                           struct lu_buf *buf)
 431 {
 432         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 433         struct dt_object  *next;
 434         int rc;
 435
 436         ENTRY;
 437
 438         LASSERT(lu_object_exists(&obj->mo_lu));
 439
 440         next = mdd_object_child(mdd_obj);
 441         mdd_read_lock(env, mdd_obj);
 442         rc = next->do_ops->do_xattr_list(env, next, buf,
 443                                          mdd_object_capa(env, mdd_obj));
 444         mdd_read_unlock(env, mdd_obj);
 445
 446         RETURN(rc);
 447 }
 448
 449 int mdd_object_create_internal(const struct lu_env *env,
 450                                struct mdd_object *obj, struct md_attr *ma,
 451                                struct thandle *handle)
 452 {
 453         struct dt_object *next;
 454         struct lu_attr *attr = &ma->ma_attr;
 455         int rc;
 456         ENTRY;
 457
 458         if (!lu_object_exists(mdd2lu_obj(obj))) {
 459                 next = mdd_object_child(obj);
 460                 rc = next->do_ops->do_create(env, next, attr, handle);
 461         } else
 462                 rc = -EEXIST;
 463
 464         LASSERT(ergo(rc == 0, lu_object_exists(mdd2lu_obj(obj))));
 465
 466         RETURN(rc);
 467 }
 468
 469
 470 int mdd_attr_set_internal(const struct lu_env *env, struct mdd_object *o,
 471                           const struct lu_attr *attr, struct thandle *handle,
 472                           const int needacl)
 473 {
 474         struct dt_object *next;
 475         int rc;
 476
 477         LASSERT(lu_object_exists(mdd2lu_obj(o)));
 478         next = mdd_object_child(o);
 479         rc = next->do_ops->do_attr_set(env, next, attr, handle,
 480                                        mdd_object_capa(env, o));
 481 #ifdef CONFIG_FS_POSIX_ACL
 482         if (!rc && (attr->la_valid & LA_MODE) && needacl)
 483                 rc = mdd_acl_chmod(env, o, attr->la_mode, handle);
 484 #endif
 485         return rc;
 486 }
 487
 488 int mdd_attr_set_internal_locked(const struct lu_env *env,
 489                                  struct mdd_object *o,
 490                                  const struct lu_attr *attr,
 491                                  struct thandle *handle, int needacl)
 492 {
 493         int rc;
 494         mdd_write_lock(env, o);
 495         rc = mdd_attr_set_internal(env, o, attr, handle, needacl);
 496         mdd_write_unlock(env, o);
 497         return rc;
 498 }
 499
 500 static int __mdd_xattr_set(const struct lu_env *env, struct mdd_object *o,
 501                            const struct lu_buf *buf, const char *name,
 502                            int fl, struct thandle *handle)
 503 {
 504         struct dt_object *next;
 505         struct lustre_capa *capa = mdd_object_capa(env, o);
 506         int rc = 0;
 507         ENTRY;
 508
 509         LASSERT(lu_object_exists(mdd2lu_obj(o)));
 510         next = mdd_object_child(o);
 511         if (buf->lb_buf && buf->lb_len > 0) {
 512                 rc = next->do_ops->do_xattr_set(env, next, buf, name, 0, handle,
 513                                                 capa);
 514         } else if (buf->lb_buf == NULL && buf->lb_len == 0) {
 515                 rc = next->do_ops->do_xattr_del(env, next, name, handle, capa);
 516         }
 517         RETURN(rc);
 518 }
 519
 520 /* this gives the same functionality as the code between
 521  * sys_chmod and inode_setattr
 522  * chown_common and inode_setattr
 523  * utimes and inode_setattr
 524  * This API is ported from mds_fix_attr but remove some unnecesssary stuff.
 525  * and port to
 526  */
 527 int mdd_fix_attr(const struct lu_env *env, struct mdd_object *obj,
 528                  struct lu_attr *la)
 529 {
 530         struct lu_attr   *tmp_la     = &mdd_env_info(env)->mti_la;
 531         struct md_ucred  *uc         = md_ucred(env);
 532         time_t            now        = CURRENT_SECONDS;
 533         int               rc;
 534         ENTRY;
 535
 536         if (!la->la_valid)
 537                 RETURN(0);
 538
 539         /* Do not permit change file type */
 540         if (la->la_valid & LA_TYPE)
 541                 RETURN(-EPERM);
 542
 543         /* They should not be processed by setattr */
 544         if (la->la_valid & (LA_NLINK | LA_RDEV | LA_BLKSIZE))
 545                 RETURN(-EPERM);
 546
 547         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 548         if (rc)
 549                 RETURN(rc);
 550
 551         if (mdd_is_immutable(obj) || mdd_is_append(obj)) {
 552
 553                 /*
 554                  * If only change flags of the object, we should
 555                  * let it pass, but also need capability check
 556                  * here if (!capable(CAP_LINUX_IMMUTABLE)),
 557                  * fix it, when implement capable in mds
 558                  */
 559                 if (la->la_valid & ~LA_FLAGS)
 560                         RETURN(-EPERM);
 561
 562                 if (!mdd_capable(uc, CAP_LINUX_IMMUTABLE))
 563                         RETURN(-EPERM);
 564
 565                 if ((uc->mu_fsuid != tmp_la->la_uid) &&
 566                     !mdd_capable(uc, CAP_FOWNER))
 567                         RETURN(-EPERM);
 568
 569                 /*
 570                  * According to Ext3 implementation on this, the
 571                  * Ctime will be changed, but not clear why?
 572                  */
 573                 la->la_ctime = now;
 574                 la->la_valid |= LA_CTIME;
 575                 RETURN(0);
 576         }
 577
 578         /* Check for setting the obj time. */
 579         if ((la->la_valid & (LA_MTIME | LA_ATIME | LA_CTIME)) &&
 580             !(la->la_valid & ~(LA_MTIME | LA_ATIME | LA_CTIME))) {
 581                 rc = __mdd_permission_internal(env, obj, MAY_WRITE, 0);
 582                 if (rc)
 583                         RETURN(rc);
 584         }
 585
 586         /* Make sure a caller can chmod. */
 587         if (la->la_valid & LA_MODE) {
 588                 /*
 589                  * Bypass la_vaild == LA_MODE,
 590                  * this is for changing file with SUID or SGID.
 591                  */
 592                 if ((la->la_valid & ~LA_MODE) &&
 593                     (uc->mu_fsuid != tmp_la->la_uid) &&
 594                     !mdd_capable(uc, CAP_FOWNER))
 595                         RETURN(-EPERM);
 596
 597                 if (la->la_mode == (umode_t) -1)
 598                         la->la_mode = tmp_la->la_mode;
 599                 else
 600                         la->la_mode = (la->la_mode & S_IALLUGO) |
 601                                       (tmp_la->la_mode & ~S_IALLUGO);
 602
 603                 /* Also check the setgid bit! */
 604                 if (!mdd_in_group_p(uc, (la->la_valid & LA_GID) ? la->la_gid :
 605                                 tmp_la->la_gid) && !mdd_capable(uc, CAP_FSETID))
 606                         la->la_mode &= ~S_ISGID;
 607         } else {
 608                la->la_mode = tmp_la->la_mode;
 609         }
 610
 611         /* Make sure a caller can chown. */
 612         if (la->la_valid & LA_UID) {
 613                 if (la->la_uid == (uid_t) -1)
 614                         la->la_uid = tmp_la->la_uid;
 615                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
 616                     (la->la_uid != tmp_la->la_uid)) &&
 617                     !mdd_capable(uc, CAP_CHOWN))
 618                         RETURN(-EPERM);
 619
 620                 /*
 621                  * If the user or group of a non-directory has been
 622                  * changed by a non-root user, remove the setuid bit.
 623                  * 19981026 David C Niemi <niemi@tux.org>
 624                  *
 625                  * Changed this to apply to all users, including root,
 626                  * to avoid some races. This is the behavior we had in
 627                  * 2.0. The check for non-root was definitely wrong
 628                  * for 2.2 anyway, as it should have been using
 629                  * CAP_FSETID rather than fsuid -- 19990830 SD.
 630                  */
 631                 if (((tmp_la->la_mode & S_ISUID) == S_ISUID) &&
 632                     !S_ISDIR(tmp_la->la_mode)) {
 633                         la->la_mode &= ~S_ISUID;
 634                         la->la_valid |= LA_MODE;
 635                 }
 636         }
 637
 638         /* Make sure caller can chgrp. */
 639         if (la->la_valid & LA_GID) {
 640                 if (la->la_gid == (gid_t) -1)
 641                         la->la_gid = tmp_la->la_gid;
 642                 if (((uc->mu_fsuid != tmp_la->la_uid) ||
 643                     ((la->la_gid != tmp_la->la_gid) &&
 644                     !mdd_in_group_p(uc, la->la_gid))) &&
 645                     !mdd_capable(uc, CAP_CHOWN))
 646                         RETURN(-EPERM);
 647
 648                 /*
 649                  * Likewise, if the user or group of a non-directory
 650                  * has been changed by a non-root user, remove the
 651                  * setgid bit UNLESS there is no group execute bit
 652                  * (this would be a file marked for mandatory
 653                  * locking).  19981026 David C Niemi <niemi@tux.org>
 654                  *
 655                  * Removed the fsuid check (see the comment above) --
 656                  * 19990830 SD.
 657                  */
 658                 if (((tmp_la->la_mode & (S_ISGID | S_IXGRP)) ==
 659                      (S_ISGID | S_IXGRP)) && !S_ISDIR(tmp_la->la_mode)) {
 660                         la->la_mode &= ~S_ISGID;
 661                         la->la_valid |= LA_MODE;
 662                 }
 663         }
 664
 665         /* For tuncate (or setsize), we should have MAY_WRITE perm */
 666         if (la->la_valid & (LA_SIZE | LA_BLOCKS)) {
 667                 rc = __mdd_permission_internal(env, obj, MAY_WRITE, 0);
 668                 if (rc)
 669                         RETURN(rc);
 670
 671                 /*
 672                  * For the "Size-on-MDS" setattr update, merge coming
 673                  * attributes with the set in the inode. BUG 10641
 674                  */
 675                 if ((la->la_valid & LA_ATIME) &&
 676                     (la->la_atime < tmp_la->la_atime))
 677                         la->la_valid &= ~LA_ATIME;
 678
 679                 if ((la->la_valid & LA_CTIME) &&
 680                     (la->la_ctime < tmp_la->la_ctime))
 681                         la->la_valid &= ~(LA_MTIME | LA_CTIME);
 682
 683                 if (!(la->la_valid & LA_MTIME) && (now > tmp_la->la_mtime)) {
 684                         la->la_mtime = now;
 685                         la->la_valid |= LA_MTIME;
 686                 }
 687         }
 688
 689         /* For last, ctime must be fixed */
 690         if (!(la->la_valid & LA_CTIME) && (now > tmp_la->la_ctime)) {
 691                 la->la_ctime = now;
 692                 la->la_valid |= LA_CTIME;
 693         }
 694
 695         RETURN(0);
 696 }
 697
 698 /* set attr and LOV EA at once, return updated attr */
 699 static int mdd_attr_set(const struct lu_env *env, struct md_object *obj,
 700                         const struct md_attr *ma)
 701 {
 702         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 703         struct mdd_device *mdd = mdo2mdd(obj);
 704         struct thandle *handle;
 705         struct lov_mds_md *lmm = NULL;
 706         int  rc, lmm_size = 0, max_size = 0;
 707         struct lu_attr *la_copy = &mdd_env_info(env)->mti_la_for_fix;
 708         ENTRY;
 709
 710         mdd_txn_param_build(env, mdd, MDD_TXN_ATTR_SET_OP);
 711         handle = mdd_trans_start(env, mdd);
 712         if (IS_ERR(handle))
 713                 RETURN(PTR_ERR(handle));
 714         /*TODO: add lock here*/
 715         /* start a log jounal handle if needed */
 716         if (S_ISREG(mdd_object_type(mdd_obj)) &&
 717             ma->ma_attr.la_valid & (LA_UID | LA_GID)) {
 718                 max_size = mdd_lov_mdsize(env, mdd);
 719                 OBD_ALLOC(lmm, max_size);
 720                 lmm_size = max_size;
 721                 if (lmm == NULL)
 722                         GOTO(cleanup, rc = -ENOMEM);
 723
 724                 rc = mdd_get_md_locked(env, mdd_obj, lmm, &lmm_size,
 725                                 MDS_LOV_MD_NAME);
 726
 727                 if (rc < 0)
 728                         GOTO(cleanup, rc);
 729         }
 730
 731         if (ma->ma_attr.la_valid & (ATTR_MTIME | ATTR_CTIME))
 732                 CDEBUG(D_INODE, "setting mtime "LPU64", ctime "LPU64"\n",
 733                        ma->ma_attr.la_mtime, ma->ma_attr.la_ctime);
 734
 735         *la_copy = ma->ma_attr;
 736         mdd_write_lock(env, mdd_obj);
 737         rc = mdd_fix_attr(env, mdd_obj, la_copy);
 738         mdd_write_unlock(env, mdd_obj);
 739         if (rc)
 740                 GOTO(cleanup, rc);
 741
 742         if (la_copy->la_valid & LA_FLAGS) {
 743                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
 744                                                   handle, 1);
 745                 if (rc == 0)
 746                         mdd_flags_xlate(mdd_obj, la_copy->la_flags);
 747         } else if (la_copy->la_valid) {            /* setattr */
 748                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la_copy,
 749                                                   handle, 1);
 750                 /* journal chown/chgrp in llog, just like unlink */
 751                 if (rc == 0 && lmm_size){
 752                         /*TODO set_attr llog */
 753                 }
 754         }
 755
 756         if (rc == 0 && ma->ma_valid & MA_LOV) {
 757                 umode_t mode;
 758
 759                 mode = mdd_object_type(mdd_obj);
 760                 if (S_ISREG(mode) || S_ISDIR(mode)) {
 761                         /*TODO check permission*/
 762                         rc = mdd_lov_set_md(env, NULL, mdd_obj, ma->ma_lmm,
 763                                             ma->ma_lmm_size, handle, 1);
 764                 }
 765
 766         }
 767 cleanup:
 768         mdd_trans_stop(env, mdd, rc, handle);
 769         if (rc == 0 && (lmm != NULL && lmm_size > 0 )) {
 770                 /*set obd attr, if needed*/
 771                 rc = mdd_lov_setattr_async(env, mdd_obj, lmm, lmm_size);
 772         }
 773         if (lmm != NULL) {
 774                 OBD_FREE(lmm, max_size);
 775         }
 776
 777         RETURN(rc);
 778 }
 779
 780 int mdd_xattr_set_txn(const struct lu_env *env, struct mdd_object *obj,
 781                       const struct lu_buf *buf, const char *name, int fl,
 782                       struct thandle *handle)
 783 {
 784         int  rc;
 785         ENTRY;
 786
 787         mdd_write_lock(env, obj);
 788         rc = __mdd_xattr_set(env, obj, buf, name, fl, handle);
 789         mdd_write_unlock(env, obj);
 790
 791         RETURN(rc);
 792 }
 793
 794 static int mdd_xattr_sanity_check(const struct lu_env *env,
 795                                   struct mdd_object *obj)
 796 {
 797         struct lu_attr  *tmp_la = &mdd_env_info(env)->mti_la;
 798         struct md_ucred *uc     = md_ucred(env);
 799         int rc;
 800         ENTRY;
 801
 802         if (mdd_is_immutable(obj) || mdd_is_append(obj))
 803                 RETURN(-EPERM);
 804
 805         mdd_read_lock(env, obj);
 806         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
 807         mdd_read_unlock(env, obj);
 808         if (rc)
 809                 RETURN(rc);
 810
 811         if ((uc->mu_fsuid != tmp_la->la_uid) && !mdd_capable(uc, CAP_FOWNER))
 812                 RETURN(-EPERM);
 813
 814         RETURN(rc);
 815 }
 816
 817 static int mdd_xattr_set(const struct lu_env *env, struct md_object *obj,
 818                          const struct lu_buf *buf, const char *name, int fl)
 819 {
 820         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 821         struct mdd_device *mdd = mdo2mdd(obj);
 822         struct thandle *handle;
 823         int  rc;
 824         ENTRY;
 825
 826         rc = mdd_xattr_sanity_check(env, mdd_obj);
 827         if (rc)
 828                 RETURN(rc);
 829
 830         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
 831         handle = mdd_trans_start(env, mdd);
 832         if (IS_ERR(handle))
 833                 RETURN(PTR_ERR(handle));
 834
 835         rc = mdd_xattr_set_txn(env, md2mdd_obj(obj), buf, name,
 836                                fl, handle);
 837
 838         mdd_trans_stop(env, mdd, rc, handle);
 839
 840         RETURN(rc);
 841 }
 842
 843 static int __mdd_xattr_del(const struct lu_env *env,struct mdd_device *mdd,
 844                            struct mdd_object *obj,
 845                            const char *name, struct thandle *handle)
 846 {
 847         struct dt_object *next;
 848
 849         LASSERT(lu_object_exists(mdd2lu_obj(obj)));
 850         next = mdd_object_child(obj);
 851         return next->do_ops->do_xattr_del(env, next, name, handle,
 852                                           mdd_object_capa(env, obj));
 853 }
 854
 855 int mdd_xattr_del(const struct lu_env *env, struct md_object *obj,
 856                   const char *name)
 857 {
 858         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 859         struct mdd_device *mdd = mdo2mdd(obj);
 860         struct thandle *handle;
 861         int  rc;
 862         ENTRY;
 863
 864         rc = mdd_xattr_sanity_check(env, mdd_obj);
 865         if (rc)
 866                 RETURN(rc);
 867
 868         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
 869         handle = mdd_trans_start(env, mdd);
 870         if (IS_ERR(handle))
 871                 RETURN(PTR_ERR(handle));
 872
 873         mdd_write_lock(env, mdd_obj);
 874         rc = __mdd_xattr_del(env, mdd, md2mdd_obj(obj), name, handle);
 875         mdd_write_unlock(env, mdd_obj);
 876
 877         mdd_trans_stop(env, mdd, rc, handle);
 878
 879         RETURN(rc);
 880 }
 881
 882 /* partial unlink */
 883 static int mdd_ref_del(const struct lu_env *env, struct md_object *obj,
 884                        struct md_attr *ma)
 885 {
 886         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 887         struct mdd_device *mdd = mdo2mdd(obj);
 888         struct thandle *handle;
 889         int rc;
 890         ENTRY;
 891
 892         rc = mdd_log_txn_param_build(env, obj, ma, MDD_TXN_UNLINK_OP);
 893         if (rc)
 894                 RETURN(rc);
 895
 896         handle = mdd_trans_start(env, mdd);
 897         if (IS_ERR(handle))
 898                 RETURN(-ENOMEM);
 899
 900         mdd_write_lock(env, mdd_obj);
 901
 902         rc = mdd_unlink_sanity_check(env, NULL, mdd_obj, ma);
 903         if (rc)
 904                 GOTO(cleanup, rc);
 905
 906         mdd_ref_del_internal(env, mdd_obj, handle);
 907
 908         if (S_ISDIR(lu_object_attr(&obj->mo_lu))) {
 909                 /* unlink dot */
 910                 mdd_ref_del_internal(env, mdd_obj, handle);
 911         }
 912
 913         rc = mdd_finish_unlink(env, mdd_obj, ma, handle);
 914
 915         EXIT;
 916 cleanup:
 917         mdd_write_unlock(env, mdd_obj);
 918         mdd_trans_stop(env, mdd, rc, handle);
 919         return rc;
 920 }
 921
 922 /* partial operation */
 923 static int mdd_oc_sanity_check(const struct lu_env *env,
 924                                struct mdd_object *obj,
 925                                struct md_attr *ma)
 926 {
 927         int rc;
 928         ENTRY;
 929
 930         switch (ma->ma_attr.la_mode & S_IFMT) {
 931         case S_IFREG:
 932         case S_IFDIR:
 933         case S_IFLNK:
 934         case S_IFCHR:
 935         case S_IFBLK:
 936         case S_IFIFO:
 937         case S_IFSOCK:
 938                 rc = 0;
 939                 break;
 940         default:
 941                 rc = -EINVAL;
 942                 break;
 943         }
 944         RETURN(rc);
 945 }
 946
 947 static int mdd_object_create(const struct lu_env *env,
 948                              struct md_object *obj,
 949                              const struct md_create_spec *spec,
 950                              struct md_attr *ma)
 951 {
 952
 953         struct mdd_device *mdd = mdo2mdd(obj);
 954         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 955         const struct lu_fid *pfid = spec->u.sp_pfid;
 956         struct thandle *handle;
 957         int rc;
 958         ENTRY;
 959
 960         rc = mdd_oc_sanity_check(env, mdd_obj, ma);
 961         if (rc)
 962                 RETURN(rc);
 963
 964         mdd_txn_param_build(env, mdd, MDD_TXN_OBJECT_CREATE_OP);
 965         handle = mdd_trans_start(env, mdd);
 966         if (IS_ERR(handle))
 967                 RETURN(PTR_ERR(handle));
 968
 969         mdd_write_lock(env, mdd_obj);
 970         rc = mdd_object_create_internal(env, mdd_obj, ma, handle);
 971         if (rc)
 972                 GOTO(unlock, rc);
 973
 974         if (spec->sp_cr_flags & MDS_CREATE_SLAVE_OBJ) {
 975                 /* If creating the slave object, set slave EA here. */
 976                 int lmv_size = spec->u.sp_ea.eadatalen;
 977                 struct lmv_stripe_md *lmv;
 978
 979                 lmv = (struct lmv_stripe_md *)spec->u.sp_ea.eadata;
 980                 LASSERT(lmv != NULL && lmv_size > 0);
 981
 982                 rc = __mdd_xattr_set(env, mdd_obj,
 983                                      mdd_buf_get_const(env, lmv, lmv_size),
 984                                      MDS_LMV_MD_NAME, 0, handle);
 985                 if (rc)
 986                         GOTO(unlock, rc);
 987                 pfid = spec->u.sp_ea.fid;
 988
 989                 CDEBUG(D_INFO, "Set slave ea "DFID", eadatalen %d, rc %d\n",
 990                        PFID(mdo2fid(mdd_obj)), spec->u.sp_ea.eadatalen, rc);
 991                 rc = mdd_attr_set_internal(env, mdd_obj, &ma->ma_attr, handle, 0);
 992         } else {
 993 #ifdef CONFIG_FS_POSIX_ACL
 994                 if (spec->sp_cr_flags & MDS_CREATE_RMT_ACL) {
 995                         struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
 996
 997                         buf->lb_buf = (void *)spec->u.sp_ea.eadata;
 998                         buf->lb_len = spec->u.sp_ea.eadatalen;
 999                         if ((buf->lb_len > 0) && (buf->lb_buf != NULL)) {
1000                                 rc = __mdd_acl_init(env, mdd_obj, buf,
1001                                                     &ma->ma_attr.la_mode,
1002                                                     handle);
1003                                 if (rc)
1004                                         GOTO(unlock, rc);
1005                                 else
1006                                         ma->ma_attr.la_valid |= LA_MODE;
1007                         }
1008                 }
1009 #endif
1010                 rc = mdd_object_initialize(env, pfid, mdd_obj, ma, handle);
1011         }
1012         EXIT;
1013 unlock:
1014         mdd_write_unlock(env, mdd_obj);
1015         if (rc == 0)
1016                 rc = mdd_attr_get_internal_locked(env, mdd_obj, ma);
1017
1018         mdd_trans_stop(env, mdd, rc, handle);
1019         return rc;
1020 }
1021
1022 /*
1023  * XXX: if permission check is needed here?
1024  */
1025 static int mdd_ref_add(const struct lu_env *env,
1026                        struct md_object *obj)
1027 {
1028         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1029         struct mdd_device *mdd = mdo2mdd(obj);
1030         struct thandle *handle;
1031         int rc;
1032         ENTRY;
1033
1034         mdd_txn_param_build(env, mdd, MDD_TXN_XATTR_SET_OP);
1035         handle = mdd_trans_start(env, mdd);
1036         if (IS_ERR(handle))
1037                 RETURN(-ENOMEM);
1038
1039         mdd_write_lock(env, mdd_obj);
1040         rc = mdd_link_sanity_check(env, NULL, mdd_obj);
1041         if (!rc)
1042                 mdd_ref_add_internal(env, mdd_obj, handle);
1043         mdd_write_unlock(env, mdd_obj);
1044
1045         mdd_trans_stop(env, mdd, 0, handle);
1046
1047         RETURN(0);
1048 }
1049
1050 /* do NOT or the MAY_*'s, you'll get the weakest */
1051 static int accmode(struct mdd_object *mdd_obj, int flags)
1052 {
1053         int res = 0;
1054
1055 #if 0
1056         /* Sadly, NFSD reopens a file repeatedly during operation, so the
1057          * "acc_mode = 0" allowance for newly-created files isn't honoured.
1058          * NFSD uses the MDS_OPEN_OWNEROVERRIDE flag to say that a file
1059          * owner can write to a file even if it is marked readonly to hide
1060          * its brokenness. (bug 5781) */
1061         if (flags & MDS_OPEN_OWNEROVERRIDE && inode->i_uid == current->fsuid)
1062                 return 0;
1063 #endif
1064         if (flags & FMODE_READ)
1065                 res = MAY_READ;
1066         if (flags & (FMODE_WRITE | MDS_OPEN_TRUNC | MDS_OPEN_APPEND))
1067                 res |= MAY_WRITE;
1068         if (flags & MDS_FMODE_EXEC)
1069                 res = MAY_EXEC;
1070         return res;
1071 }
1072
1073 static int mdd_open_sanity_check(const struct lu_env *env,
1074                                  struct mdd_object *obj, int flag)
1075 {
1076         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
1077         int mode = accmode(obj, flag);
1078         int rc;
1079         ENTRY;
1080
1081         /* EEXIST check */
1082         if (mdd_is_dead_obj(obj))
1083                 RETURN(-ENOENT);
1084
1085         rc = mdd_la_get(env, obj, tmp_la, BYPASS_CAPA);
1086         if (rc)
1087                RETURN(rc);
1088
1089         if (S_ISLNK(tmp_la->la_mode))
1090                 RETURN(-ELOOP);
1091
1092         if (S_ISDIR(tmp_la->la_mode) && (mode & MAY_WRITE))
1093                 RETURN(-EISDIR);
1094
1095         if (!(flag & MDS_OPEN_CREATED)) {
1096                 rc = __mdd_permission_internal(env, obj, mode, 0);
1097                 if (rc)
1098                         RETURN(rc);
1099         }
1100
1101         /*
1102          * FIFO's, sockets and device files are special: they don't
1103          * actually live on the filesystem itself, and as such you
1104          * can write to them even if the filesystem is read-only.
1105          */
1106         if (S_ISFIFO(tmp_la->la_mode) || S_ISSOCK(tmp_la->la_mode) ||
1107             S_ISBLK(tmp_la->la_mode) || S_ISCHR(tmp_la->la_mode))
1108                 flag &= ~O_TRUNC;
1109
1110         /*
1111          * An append-only file must be opened in append mode for writing.
1112          */
1113         if (mdd_is_append(obj)) {
1114                 if ((flag & FMODE_WRITE) && !(flag & O_APPEND))
1115                         RETURN(-EPERM);
1116                 if (flag & O_TRUNC)
1117                         RETURN(-EPERM);
1118         }
1119
1120         /* O_NOATIME can only be set by the owner or superuser */
1121         if (flag & O_NOATIME) {
1122                 struct md_ucred *uc = md_ucred(env);
1123
1124                 if (uc->mu_fsuid != tmp_la->la_uid &&
1125                     !mdd_capable(uc, CAP_FOWNER))
1126                         RETURN(-EPERM);
1127         }
1128
1129         RETURN(0);
1130 }
1131
1132 static int mdd_open(const struct lu_env *env, struct md_object *obj,
1133                     int flags)
1134 {
1135         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1136         int rc = 0;
1137
1138         mdd_write_lock(env, mdd_obj);
1139
1140         rc = mdd_open_sanity_check(env, mdd_obj, flags);
1141         if (rc == 0)
1142                 mdd_obj->mod_count++;
1143
1144         mdd_write_unlock(env, mdd_obj);
1145         return rc;
1146 }
1147
1148 /* return md_attr back,
1149  * if it is last unlink then return lov ea + llog cookie*/
1150 int mdd_object_kill(const struct lu_env *env, struct mdd_object *obj,
1151                     struct md_attr *ma)
1152 {
1153         int rc = 0;
1154         ENTRY;
1155
1156         mdd_set_dead_obj(obj);
1157         if (S_ISREG(mdd_object_type(obj))) {
1158                 /* Return LOV & COOKIES unconditionally here. We clean evth up.
1159                  * Caller must be ready for that. */
1160                 rc = __mdd_lmm_get(env, obj, ma);
1161                 if ((ma->ma_valid & MA_LOV))
1162                         rc = mdd_unlink_log(env, mdo2mdd(&obj->mod_obj),
1163                                             obj, ma);
1164         }
1165         RETURN(rc);
1166 }
1167
1168 /*
1169  * No permission check is needed.
1170  */
1171 static int mdd_close(const struct lu_env *env, struct md_object *obj,
1172                      struct md_attr *ma)
1173 {
1174         int rc;
1175         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1176         ENTRY;
1177
1178         mdd_write_lock(env, mdd_obj);
1179         /* release open count */
1180         mdd_obj->mod_count --;
1181
1182         rc = mdd_iattr_get(env, mdd_obj, ma);
1183         if (rc == 0 && mdd_obj->mod_count == 0) {
1184                 if (ma->ma_attr.la_nlink == 0)
1185                         rc = mdd_object_kill(env, mdd_obj, ma);
1186         }
1187         mdd_write_unlock(env, mdd_obj);
1188         RETURN(rc);
1189 }
1190
1191 /*
1192  * Permission check is done when open,
1193  * no need check again.
1194  */
1195 static int mdd_readpage_sanity_check(const struct lu_env *env,
1196                                      struct mdd_object *obj)
1197 {
1198         struct dt_object *next = mdd_object_child(obj);
1199         int rc;
1200         ENTRY;
1201
1202         if (S_ISDIR(mdd_object_type(obj)) && dt_try_as_dir(env, next))
1203 #if 0
1204                 rc = mdd_permission_internal(env, obj, MAY_READ);
1205 #else
1206                 rc = 0;
1207 #endif
1208         else
1209                 rc = -ENOTDIR;
1210
1211         RETURN(rc);
1212 }
1213
1214 static int mdd_dir_page_build(const struct lu_env *env, int first,
1215                               void *area, int nob,
1216                               struct dt_it_ops  *iops, struct dt_it *it,
1217                               __u32 *start, __u32 *end, struct lu_dirent **last)
1218 {
1219         int result;
1220         struct mdd_thread_info *info = mdd_env_info(env);
1221         struct lu_fid          *fid  = &info->mti_fid;
1222         struct lu_dirent       *ent;
1223
1224         if (first) {
1225                 memset(area, 0, sizeof (struct lu_dirpage));
1226                 area += sizeof (struct lu_dirpage);
1227                 nob  -= sizeof (struct lu_dirpage);
1228         }
1229
1230         LASSERT(nob > sizeof *ent);
1231
1232         ent  = area;
1233         result = 0;
1234         do {
1235                 char  *name;
1236                 int    len;
1237                 int    recsize;
1238                 __u32  hash;
1239
1240                 name = (char *)iops->key(env, it);
1241                 len  = iops->key_size(env, it);
1242
1243                 fid  = (struct lu_fid *)iops->rec(env, it);
1244
1245                 recsize = (sizeof *ent + len + 3) & ~3;
1246                 hash = iops->store(env, it);
1247                 *end = hash;
1248                 CDEBUG(D_INFO, "%p %p %d "DFID": %#8.8x (%d) \"%*.*s\"\n",
1249                        name, ent, nob, PFID(fid), hash, len, len, len, name);
1250
1251                 if (nob >= recsize) {
1252                         fid_be_to_cpu(&ent->lde_fid, fid);
1253                         fid_cpu_to_le(&ent->lde_fid, &ent->lde_fid);
1254                         ent->lde_hash = hash;
1255                         ent->lde_namelen = cpu_to_le16(len);
1256                         ent->lde_reclen  = cpu_to_le16(recsize);
1257                         memcpy(ent->lde_name, name, len);
1258                         if (first && ent == area)
1259                                 *start = hash;
1260                         *last = ent;
1261                         ent = (void *)ent + recsize;
1262                         nob -= recsize;
1263                         result = iops->next(env, it);
1264                 } else {
1265                         /*
1266                          * record doesn't fit into page, enlarge previous one.
1267                          */
1268                         LASSERT(*last != NULL);
1269                         (*last)->lde_reclen =
1270                                 cpu_to_le16(le16_to_cpu((*last)->lde_reclen) +
1271                                             nob);
1272                         break;
1273                 }
1274         } while (result == 0);
1275
1276         return result;
1277 }
1278
1279 static int __mdd_readpage(const struct lu_env *env, struct mdd_object *obj,
1280                           const struct lu_rdpg *rdpg)
1281 {
1282         struct dt_it      *it;
1283         struct dt_object  *next = mdd_object_child(obj);
1284         struct dt_it_ops  *iops;
1285         struct page       *pg;
1286         struct lu_dirent  *last;
1287         int i;
1288         int rc;
1289         int nob;
1290         __u32 hash_start;
1291         __u32 hash_end;
1292
1293         LASSERT(rdpg->rp_pages != NULL);
1294         LASSERT(next->do_index_ops != NULL);
1295
1296         if (rdpg->rp_count <= 0)
1297                 return -EFAULT;
1298
1299         /*
1300          * iterate through directory and fill pages from @rdpg
1301          */
1302         iops = &next->do_index_ops->dio_it;
1303         it = iops->init(env, next, 0, mdd_object_capa(env, obj));
1304         if (it == NULL)
1305                 return -ENOMEM;
1306
1307         rc = iops->load(env, it, rdpg->rp_hash);
1308
1309         if (rc == 0)
1310                 /*
1311                  * Iterator didn't find record with exactly the key requested.
1312                  *
1313                  * It is currently either
1314                  *
1315                  *     - positioned above record with key less than
1316                  *     requested---skip it.
1317                  *
1318                  *     - or not positioned at all (is in IAM_IT_SKEWED
1319                  *     state)---position it on the next item.
1320                  */
1321                 rc = iops->next(env, it);
1322         else if (rc > 0)
1323                 rc = 0;
1324
1325         /*
1326          * At this point and across for-loop:
1327          *
1328          *  rc == 0 -> ok, proceed.
1329          *  rc >  0 -> end of directory.
1330          *  rc <  0 -> error.
1331          */
1332         for (i = 0, nob = rdpg->rp_count; rc == 0 && nob > 0;
1333              i++, nob -= CFS_PAGE_SIZE) {
1334                 LASSERT(i < rdpg->rp_npages);
1335                 pg = rdpg->rp_pages[i];
1336                 rc = mdd_dir_page_build(env, !i, kmap(pg),
1337                                         min_t(int, nob, CFS_PAGE_SIZE), iops,
1338                                         it, &hash_start, &hash_end, &last);
1339                 if (rc != 0 || i == rdpg->rp_npages - 1)
1340                         last->lde_reclen = 0;
1341                 kunmap(pg);
1342         }
1343         if (rc > 0) {
1344                 /*
1345                  * end of directory.
1346                  */
1347                 hash_end = ~0ul;
1348                 rc = 0;
1349         }
1350         if (rc == 0) {
1351                 struct lu_dirpage *dp;
1352
1353                 dp = kmap(rdpg->rp_pages[0]);
1354                 dp->ldp_hash_start = rdpg->rp_hash;
1355                 dp->ldp_hash_end   = hash_end;
1356                 if (i == 0)
1357                         /*
1358                          * No pages were processed, mark this.
1359                          */
1360                         dp->ldp_flags |= LDF_EMPTY;
1361                 dp->ldp_flags = cpu_to_le16(dp->ldp_flags);
1362                 kunmap(rdpg->rp_pages[0]);
1363         }
1364         iops->put(env, it);
1365         iops->fini(env, it);
1366
1367         return rc;
1368 }
1369
1370 static int mdd_readpage(const struct lu_env *env, struct md_object *obj,
1371                         const struct lu_rdpg *rdpg)
1372 {
1373         struct dt_object *next;
1374         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1375         int rc;
1376         ENTRY;
1377
1378         LASSERT(lu_object_exists(mdd2lu_obj(mdd_obj)));
1379         next = mdd_object_child(mdd_obj);
1380
1381         mdd_read_lock(env, mdd_obj);
1382         rc = mdd_readpage_sanity_check(env, mdd_obj);
1383         if (rc)
1384                 GOTO(out_unlock, rc);
1385
1386         rc = __mdd_readpage(env, mdd_obj, rdpg);
1387
1388 out_unlock:
1389         mdd_read_unlock(env, mdd_obj);
1390         RETURN(rc);
1391 }
1392
1393 struct md_object_operations mdd_obj_ops = {
1394         .moo_permission    = mdd_permission,
1395         .moo_attr_get      = mdd_attr_get,
1396         .moo_attr_set      = mdd_attr_set,
1397         .moo_xattr_get     = mdd_xattr_get,
1398         .moo_xattr_set     = mdd_xattr_set,
1399         .moo_xattr_list    = mdd_xattr_list,
1400         .moo_xattr_del     = mdd_xattr_del,
1401         .moo_object_create = mdd_object_create,
1402         .moo_ref_add       = mdd_ref_add,
1403         .moo_ref_del       = mdd_ref_del,
1404         .moo_open          = mdd_open,
1405         .moo_close         = mdd_close,
1406         .moo_readpage      = mdd_readpage,
1407         .moo_readlink      = mdd_readlink,
1408         .moo_capa_get      = mdd_capa_get
1409 };