lustre/osd-ldiskfs/osd_handler.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2015, Intel Corporation.
  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/osd/osd_handler.c
  37  *
  38  * Top-level entry points into osd module
  39  *
  40  * Author: Nikita Danilov <nikita@clusterfs.com>
  41  *         Pravin Shelar <pravin.shelar@sun.com> : Added fid in dirent
  42  */
  43
  44 #define DEBUG_SUBSYSTEM S_OSD
  45
  46 #include <linux/module.h>
  47 #include <linux/user_namespace.h>
  48 #ifdef HAVE_UIDGID_HEADER
  49 # include <linux/uidgid.h>
  50 #endif
  51
  52 /* LUSTRE_VERSION_CODE */
  53 #include <lustre_ver.h>
  54 /* prerequisite for linux/xattr.h */
  55 #include <linux/types.h>
  56 /* prerequisite for linux/xattr.h */
  57 #include <linux/fs.h>
  58 /* XATTR_{REPLACE,CREATE} */
  59 #include <linux/xattr.h>
  60
  61 #include <ldiskfs/ldiskfs.h>
  62 #include <ldiskfs/xattr.h>
  63 #include <ldiskfs/ldiskfs_extents.h>
  64 #undef ENTRY
  65 /*
  66  * struct OBD_{ALLOC,FREE}*()
  67  * OBD_FAIL_CHECK
  68  */
  69 #include <obd_support.h>
  70 /* struct ptlrpc_thread */
  71 #include <lustre_net.h>
  72 #include <lustre_fid.h>
  73 /* process_config */
  74 #include <lustre_param.h>
  75
  76 #include "osd_internal.h"
  77 #include "osd_dynlocks.h"
  78
  79 /* llo_* api support */
  80 #include <md_object.h>
  81 #include <lustre_quota.h>
  82
  83 #include <lustre_linkea.h>
  84
  85 int ldiskfs_pdo = 1;
  86 module_param(ldiskfs_pdo, int, 0644);
  87 MODULE_PARM_DESC(ldiskfs_pdo, "ldiskfs with parallel directory operations");
  88
  89 int ldiskfs_track_declares_assert;
  90 module_param(ldiskfs_track_declares_assert, int, 0644);
  91 MODULE_PARM_DESC(ldiskfs_track_declares_assert, "LBUG during tracking of declares");
  92
  93 /* Slab to allocate dynlocks */
  94 struct kmem_cache *dynlock_cachep;
  95
  96 /* Slab to allocate osd_it_ea */
  97 struct kmem_cache *osd_itea_cachep;
  98
  99 static struct lu_kmem_descr ldiskfs_caches[] = {
 100         {
 101                 .ckd_cache = &dynlock_cachep,
 102                 .ckd_name  = "dynlock_cache",
 103                 .ckd_size  = sizeof(struct dynlock_handle)
 104         },
 105         {
 106                 .ckd_cache = &osd_itea_cachep,
 107                 .ckd_name  = "osd_itea_cache",
 108                 .ckd_size  = sizeof(struct osd_it_ea)
 109         },
 110         {
 111                 .ckd_cache = NULL
 112         }
 113 };
 114
 115 static const char dot[] = ".";
 116 static const char dotdot[] = "..";
 117 static const char remote_obj_dir[] = "REM_OBJ_DIR";
 118
 119 static const struct lu_object_operations      osd_lu_obj_ops;
 120 static const struct dt_object_operations      osd_obj_ops;
 121 static const struct dt_object_operations      osd_obj_ea_ops;
 122 static const struct dt_object_operations      osd_obj_otable_it_ops;
 123 static const struct dt_index_operations       osd_index_iam_ops;
 124 static const struct dt_index_operations       osd_index_ea_ops;
 125
 126 static int osd_remote_fid(const struct lu_env *env, struct osd_device *osd,
 127                           const struct lu_fid *fid);
 128 static int osd_process_scheduled_agent_removals(const struct lu_env *env,
 129                                                 struct osd_device *osd);
 130
 131 int osd_trans_declare_op2rb[] = {
 132         [OSD_OT_ATTR_SET]       = OSD_OT_ATTR_SET,
 133         [OSD_OT_PUNCH]          = OSD_OT_MAX,
 134         [OSD_OT_XATTR_SET]      = OSD_OT_XATTR_SET,
 135         [OSD_OT_CREATE]         = OSD_OT_DESTROY,
 136         [OSD_OT_DESTROY]        = OSD_OT_CREATE,
 137         [OSD_OT_REF_ADD]        = OSD_OT_REF_DEL,
 138         [OSD_OT_REF_DEL]        = OSD_OT_REF_ADD,
 139         [OSD_OT_WRITE]          = OSD_OT_WRITE,
 140         [OSD_OT_INSERT]         = OSD_OT_DELETE,
 141         [OSD_OT_DELETE]         = OSD_OT_INSERT,
 142         [OSD_OT_QUOTA]          = OSD_OT_MAX,
 143 };
 144
 145 static int osd_has_index(const struct osd_object *obj)
 146 {
 147         return obj->oo_dt.do_index_ops != NULL;
 148 }
 149
 150 static int osd_object_invariant(const struct lu_object *l)
 151 {
 152         return osd_invariant(osd_obj(l));
 153 }
 154
 155 /*
 156  * Concurrency: doesn't matter
 157  */
 158
 159 /*
 160  * Concurrency: doesn't matter
 161  */
 162 static int osd_write_locked(const struct lu_env *env, struct osd_object *o)
 163 {
 164         struct osd_thread_info *oti = osd_oti_get(env);
 165         return oti->oti_w_locks > 0 && o->oo_owner == env;
 166 }
 167
 168 /*
 169  * Concurrency: doesn't access mutable data
 170  */
 171 static int osd_root_get(const struct lu_env *env,
 172                         struct dt_device *dev, struct lu_fid *f)
 173 {
 174         lu_local_obj_fid(f, OSD_FS_ROOT_OID);
 175         return 0;
 176 }
 177
 178 /*
 179  * the following set of functions are used to maintain per-thread
 180  * cache of FID->ino mapping. this mechanism is needed to resolve
 181  * FID to inode at dt_insert() which in turn stores ino in the
 182  * directory entries to keep ldiskfs compatible with ext[34].
 183  * due to locking-originated restrictions we can't lookup ino
 184  * using LU cache (deadlock is possible). lookup using OI is quite
 185  * expensive. so instead we maintain this cache and methods like
 186  * dt_create() fill it. so in the majority of cases dt_insert() is
 187  * able to find needed mapping in lockless manner.
 188  */
 189 static struct osd_idmap_cache *
 190 osd_idc_find(const struct lu_env *env, struct osd_device *osd,
 191              const struct lu_fid *fid)
 192 {
 193         struct osd_thread_info  *oti   = osd_oti_get(env);
 194         struct osd_idmap_cache  *idc    = oti->oti_ins_cache;
 195         int i;
 196         for (i = 0; i < oti->oti_ins_cache_used; i++) {
 197                 if (!lu_fid_eq(&idc[i].oic_fid, fid))
 198                         continue;
 199                 if (idc[i].oic_dev != osd)
 200                         continue;
 201
 202                 return idc + i;
 203         }
 204
 205         return NULL;
 206 }
 207
 208 static struct osd_idmap_cache *
 209 osd_idc_add(const struct lu_env *env, struct osd_device *osd,
 210             const struct lu_fid *fid)
 211 {
 212         struct osd_thread_info  *oti   = osd_oti_get(env);
 213         struct osd_idmap_cache  *idc;
 214         int i;
 215
 216         if (unlikely(oti->oti_ins_cache_used >= oti->oti_ins_cache_size)) {
 217                 i = oti->oti_ins_cache_size * 2;
 218                 if (i == 0)
 219                         i = OSD_INS_CACHE_SIZE;
 220                 OBD_ALLOC(idc, sizeof(*idc) * i);
 221                 if (idc == NULL)
 222                         return ERR_PTR(-ENOMEM);
 223                 if (oti->oti_ins_cache != NULL) {
 224                         memcpy(idc, oti->oti_ins_cache,
 225                                oti->oti_ins_cache_used * sizeof(*idc));
 226                         OBD_FREE(oti->oti_ins_cache,
 227                                  oti->oti_ins_cache_used * sizeof(*idc));
 228                 }
 229                 oti->oti_ins_cache = idc;
 230                 oti->oti_ins_cache_size = i;
 231         }
 232
 233         idc = oti->oti_ins_cache + oti->oti_ins_cache_used++;
 234         idc->oic_fid = *fid;
 235         idc->oic_dev = osd;
 236         idc->oic_lid.oii_ino = 0;
 237         idc->oic_lid.oii_gen = 0;
 238         idc->oic_remote = 0;
 239
 240         return idc;
 241 }
 242
 243 /*
 244  * lookup mapping for the given fid in the cache, initialize a
 245  * new one if not found. the initialization checks whether the
 246  * object is local or remote. for local objects, OI is used to
 247  * learn ino/generation. the function is used when the caller
 248  * has no information about the object, e.g. at dt_insert().
 249  */
 250 static struct osd_idmap_cache *
 251 osd_idc_find_or_init(const struct lu_env *env, struct osd_device *osd,
 252                      const struct lu_fid *fid)
 253 {
 254         struct osd_idmap_cache *idc;
 255         int rc;
 256
 257         idc = osd_idc_find(env, osd, fid);
 258         LASSERT(!IS_ERR(idc));
 259         if (idc != NULL)
 260                 return idc;
 261
 262         /* new mapping is needed */
 263         idc = osd_idc_add(env, osd, fid);
 264         if (IS_ERR(idc))
 265                 return idc;
 266
 267         /* initialize it */
 268         rc = osd_remote_fid(env, osd, fid);
 269         if (unlikely(rc < 0))
 270                 return ERR_PTR(rc);
 271
 272         if (rc == 0) {
 273                 /* the object is local, lookup in OI */
 274                 /* XXX: probably cheaper to lookup in LU first? */
 275                 rc = osd_oi_lookup(osd_oti_get(env), osd, fid,
 276                                    &idc->oic_lid, 0);
 277                 if (unlikely(rc < 0)) {
 278                         CERROR("can't lookup: rc = %d\n", rc);
 279                         return ERR_PTR(rc);
 280                 }
 281         } else {
 282                 /* the object is remote */
 283                 idc->oic_remote = 1;
 284         }
 285
 286         return idc;
 287 }
 288
 289 /*
 290  * lookup mapping for given FID and fill it from the given object.
 291  * the object is lolcal by definition.
 292  */
 293 static int osd_idc_find_and_init(const struct lu_env *env,
 294                                  struct osd_device *osd,
 295                                  struct osd_object *obj)
 296 {
 297         const struct lu_fid     *fid = lu_object_fid(&obj->oo_dt.do_lu);
 298         struct osd_idmap_cache  *idc;
 299
 300         idc = osd_idc_find(env, osd, fid);
 301         LASSERT(!IS_ERR(idc));
 302         if (idc != NULL) {
 303                 if (obj->oo_inode == NULL)
 304                         return 0;
 305                 if (idc->oic_lid.oii_ino != obj->oo_inode->i_ino) {
 306                         LASSERT(idc->oic_lid.oii_ino == 0);
 307                         idc->oic_lid.oii_ino = obj->oo_inode->i_ino;
 308                         idc->oic_lid.oii_gen = obj->oo_inode->i_generation;
 309                 }
 310                 return 0;
 311         }
 312
 313         /* new mapping is needed */
 314         idc = osd_idc_add(env, osd, fid);
 315         if (IS_ERR(idc))
 316                 return PTR_ERR(idc);
 317
 318         if (obj->oo_inode != NULL) {
 319                 idc->oic_lid.oii_ino = obj->oo_inode->i_ino;
 320                 idc->oic_lid.oii_gen = obj->oo_inode->i_generation;
 321         }
 322         return 0;
 323 }
 324
 325 /*
 326  * OSD object methods.
 327  */
 328
 329 /*
 330  * Concurrency: no concurrent access is possible that early in object
 331  * life-cycle.
 332  */
 333 static struct lu_object *osd_object_alloc(const struct lu_env *env,
 334                                           const struct lu_object_header *hdr,
 335                                           struct lu_device *d)
 336 {
 337         struct osd_object *mo;
 338
 339         OBD_ALLOC_PTR(mo);
 340         if (mo != NULL) {
 341                 struct lu_object *l;
 342
 343                 l = &mo->oo_dt.do_lu;
 344                 dt_object_init(&mo->oo_dt, NULL, d);
 345                 mo->oo_dt.do_ops = &osd_obj_ea_ops;
 346                 l->lo_ops = &osd_lu_obj_ops;
 347                 init_rwsem(&mo->oo_sem);
 348                 init_rwsem(&mo->oo_ext_idx_sem);
 349                 spin_lock_init(&mo->oo_guard);
 350                 INIT_LIST_HEAD(&mo->oo_xattr_list);
 351                 return l;
 352         } else {
 353                 return NULL;
 354         }
 355 }
 356
 357 int osd_get_lma(struct osd_thread_info *info, struct inode *inode,
 358                 struct dentry *dentry, struct lustre_mdt_attrs *lma)
 359 {
 360         int rc;
 361
 362         CLASSERT(LMA_OLD_SIZE >= sizeof(*lma));
 363         rc = __osd_xattr_get(inode, dentry, XATTR_NAME_LMA,
 364                              info->oti_mdt_attrs_old, LMA_OLD_SIZE);
 365         if (rc > 0) {
 366                 if ((void *)lma != (void *)info->oti_mdt_attrs_old)
 367                         memcpy(lma, info->oti_mdt_attrs_old, sizeof(*lma));
 368                 rc = 0;
 369                 lustre_lma_swab(lma);
 370                 /* Check LMA compatibility */
 371                 if (lma->lma_incompat & ~LMA_INCOMPAT_SUPP) {
 372                         CWARN("%.16s: unsupported incompat LMA feature(s) %#x "
 373                               "for fid = "DFID", ino = %lu\n",
 374                               LDISKFS_SB(inode->i_sb)->s_es->s_volume_name,
 375                               lma->lma_incompat & ~LMA_INCOMPAT_SUPP,
 376                               PFID(&lma->lma_self_fid), inode->i_ino);
 377                         rc = -EOPNOTSUPP;
 378                 }
 379         } else if (rc == 0) {
 380                 rc = -ENODATA;
 381         }
 382
 383         return rc;
 384 }
 385
 386 /*
 387  * retrieve object from backend ext fs.
 388  **/
 389 struct inode *osd_iget(struct osd_thread_info *info, struct osd_device *dev,
 390                        struct osd_inode_id *id)
 391 {
 392         struct inode *inode = NULL;
 393
 394         /* if we look for an inode withing a running
 395          * transaction, then we risk to deadlock */
 396         /* osd_dirent_check_repair() breaks this */
 397         /*LASSERT(current->journal_info == NULL);*/
 398
 399         inode = ldiskfs_iget(osd_sb(dev), id->oii_ino);
 400         if (IS_ERR(inode)) {
 401                 CDEBUG(D_INODE, "no inode: ino = %u, rc = %ld\n",
 402                        id->oii_ino, PTR_ERR(inode));
 403         } else if (id->oii_gen != OSD_OII_NOGEN &&
 404                    inode->i_generation != id->oii_gen) {
 405                 CDEBUG(D_INODE, "unmatched inode: ino = %u, oii_gen = %u, "
 406                        "i_generation = %u\n",
 407                        id->oii_ino, id->oii_gen, inode->i_generation);
 408                 iput(inode);
 409                 inode = ERR_PTR(-ESTALE);
 410         } else if (inode->i_nlink == 0) {
 411                 /* due to parallel readdir and unlink,
 412                 * we can have dead inode here. */
 413                 CDEBUG(D_INODE, "stale inode: ino = %u\n", id->oii_ino);
 414                 iput(inode);
 415                 inode = ERR_PTR(-ESTALE);
 416         } else if (is_bad_inode(inode)) {
 417                 CWARN("%.16s: bad inode: ino = %u\n",
 418                 LDISKFS_SB(osd_sb(dev))->s_es->s_volume_name, id->oii_ino);
 419                 iput(inode);
 420                 inode = ERR_PTR(-ENOENT);
 421         } else {
 422                 ldiskfs_clear_inode_state(inode, LDISKFS_STATE_LUSTRE_DESTROY);
 423                 if (id->oii_gen == OSD_OII_NOGEN)
 424                         osd_id_gen(id, inode->i_ino, inode->i_generation);
 425
 426                 /* Do not update file c/mtime in ldiskfs.
 427                  * NB: we don't have any lock to protect this because we don't
 428                  * have reference on osd_object now, but contention with
 429                  * another lookup + attr_set can't happen in the tiny window
 430                  * between if (...) and set S_NOCMTIME. */
 431                 if (!(inode->i_flags & S_NOCMTIME))
 432                         inode->i_flags |= S_NOCMTIME;
 433         }
 434         return inode;
 435 }
 436
 437 int osd_ldiskfs_add_entry(struct osd_thread_info *info,
 438                           handle_t *handle, struct dentry *child,
 439                           struct inode *inode, struct htree_lock *hlock)
 440 {
 441         int rc, rc2;
 442
 443         rc = __ldiskfs_add_entry(handle, child, inode, hlock);
 444         if (rc == -ENOBUFS || rc == -ENOSPC) {
 445                 char fidbuf[FID_LEN + 1];
 446                 struct lustre_mdt_attrs lma;
 447                 struct lu_fid fid = { };
 448                 char *errstr;
 449                 struct dentry *p_dentry = child->d_parent;
 450
 451                 rc2 = osd_get_lma(info, p_dentry->d_inode, p_dentry,
 452                                  &lma);
 453                 if (rc2 == 0) {
 454                         fid = lma.lma_self_fid;
 455                         snprintf(fidbuf, sizeof(fidbuf), DFID, PFID(&fid));
 456                 } else if (rc2 == -ENODATA) {
 457                         if (unlikely(p_dentry->d_inode ==
 458                                      inode->i_sb->s_root->d_inode))
 459                                 lu_local_obj_fid(&fid, OSD_FS_ROOT_OID);
 460                         else if (info->oti_dev && !info->oti_dev->od_is_ost &&
 461                                  fid_seq_is_mdt0(fid_seq(&fid)))
 462                                 lu_igif_build(&fid, p_dentry->d_inode->i_ino,
 463                                               p_dentry->d_inode->i_generation);
 464                         snprintf(fidbuf, sizeof(fidbuf), DFID, PFID(&fid));
 465                 } else {
 466                         snprintf(fidbuf, FID_LEN, "%s", "unknown");
 467                 }
 468
 469                 if (rc == -ENOSPC)
 470                         errstr = "has reached";
 471                 else
 472                         errstr = "is approaching";
 473                 CWARN("%.16s: directory (inode: %lu FID: %s) %s maximum entry limit\n",
 474                         LDISKFS_SB(inode->i_sb)->s_es->s_volume_name,
 475                         p_dentry->d_inode->i_ino, fidbuf, errstr);
 476                 /* ignore such error now */
 477                 if (rc == -ENOBUFS)
 478                         rc = 0;
 479         }
 480         return rc;
 481 }
 482
 483
 484 static struct inode *
 485 osd_iget_fid(struct osd_thread_info *info, struct osd_device *dev,
 486              struct osd_inode_id *id, struct lu_fid *fid)
 487 {
 488         struct lustre_mdt_attrs *lma   = &info->oti_mdt_attrs;
 489         struct inode            *inode;
 490         int                      rc;
 491
 492         inode = osd_iget(info, dev, id);
 493         if (IS_ERR(inode))
 494                 return inode;
 495
 496         rc = osd_get_lma(info, inode, &info->oti_obj_dentry, lma);
 497         if (rc == 0) {
 498                 *fid = lma->lma_self_fid;
 499         } else if (rc == -ENODATA) {
 500                 if (unlikely(inode == osd_sb(dev)->s_root->d_inode))
 501                         lu_local_obj_fid(fid, OSD_FS_ROOT_OID);
 502                 else
 503                         lu_igif_build(fid, inode->i_ino, inode->i_generation);
 504         } else {
 505                 iput(inode);
 506                 inode = ERR_PTR(rc);
 507         }
 508         return inode;
 509 }
 510
 511 static struct inode *osd_iget_check(struct osd_thread_info *info,
 512                                     struct osd_device *dev,
 513                                     const struct lu_fid *fid,
 514                                     struct osd_inode_id *id,
 515                                     bool cached)
 516 {
 517         struct inode    *inode;
 518         int              rc     = 0;
 519         ENTRY;
 520
 521         /* The cached OI mapping is trustable. If we cannot locate the inode
 522          * via the cached OI mapping, then return the failure to the caller
 523          * directly without further OI checking. */
 524
 525         inode = ldiskfs_iget(osd_sb(dev), id->oii_ino);
 526         if (IS_ERR(inode)) {
 527                 rc = PTR_ERR(inode);
 528                 if (cached || (rc != -ENOENT && rc != -ESTALE)) {
 529                         CDEBUG(D_INODE, "no inode: ino = %u, rc = %d\n",
 530                                id->oii_ino, rc);
 531
 532                         GOTO(put, rc);
 533                 }
 534
 535                 goto check_oi;
 536         }
 537
 538         if (is_bad_inode(inode)) {
 539                 rc = -ENOENT;
 540                 if (cached) {
 541                         CDEBUG(D_INODE, "bad inode: ino = %u\n", id->oii_ino);
 542
 543                         GOTO(put, rc);
 544                 }
 545
 546                 goto check_oi;
 547         }
 548
 549         if (id->oii_gen != OSD_OII_NOGEN &&
 550             inode->i_generation != id->oii_gen) {
 551                 rc = -ESTALE;
 552                 if (cached) {
 553                         CDEBUG(D_INODE, "unmatched inode: ino = %u, "
 554                                "oii_gen = %u, i_generation = %u\n",
 555                                id->oii_ino, id->oii_gen, inode->i_generation);
 556
 557                         GOTO(put, rc);
 558                 }
 559
 560                 goto check_oi;
 561         }
 562
 563         if (inode->i_nlink == 0) {
 564                 rc = -ENOENT;
 565                 if (cached) {
 566                         CDEBUG(D_INODE, "stale inode: ino = %u\n", id->oii_ino);
 567
 568                         GOTO(put, rc);
 569                 }
 570
 571                 goto check_oi;
 572         }
 573
 574         ldiskfs_clear_inode_state(inode, LDISKFS_STATE_LUSTRE_DESTROY);
 575
 576 check_oi:
 577         if (rc != 0) {
 578                 LASSERTF(rc == -ESTALE || rc == -ENOENT, "rc = %d\n", rc);
 579
 580                 rc = osd_oi_lookup(info, dev, fid, id, OI_CHECK_FLD);
 581                 /* XXX: There are four possible cases:
 582                  *      1. rc = 0.
 583                  *         Backup/restore caused the OI invalid.
 584                  *      2. rc = 0.
 585                  *         Someone unlinked the object but NOT removed
 586                  *         the OI mapping, such as mount target device
 587                  *         as ldiskfs, and modify something directly.
 588                  *      3. rc = -ENOENT.
 589                  *         Someone just removed the object between the
 590                  *         former oi_lookup and the iget. It is normal.
 591                  *      4. Other failure cases.
 592                  *
 593                  *      Generally, when the device is mounted, it will
 594                  *      auto check whether the system is restored from
 595                  *      file-level backup or not. We trust such detect
 596                  *      to distinguish the 1st case from the 2nd case. */
 597                 if (rc == 0) {
 598                         if (!IS_ERR(inode) && inode->i_generation != 0 &&
 599                             inode->i_generation == id->oii_gen)
 600                                 /* "id->oii_gen != OSD_OII_NOGEN" is for
 601                                  * "@cached == false" case. */
 602                                 rc = -ENOENT;
 603                         else
 604                                 rc = -EREMCHG;
 605                 } else {
 606                         /* If the OI mapping was in OI file before the
 607                          * osd_iget_check(), but now, it is disappear,
 608                          * then it must be removed by race. That is a
 609                          * normal race case. */
 610                 }
 611         } else {
 612                 if (id->oii_gen == OSD_OII_NOGEN)
 613                         osd_id_gen(id, inode->i_ino, inode->i_generation);
 614
 615                 /* Do not update file c/mtime in ldiskfs.
 616                  * NB: we don't have any lock to protect this because we don't
 617                  * have reference on osd_object now, but contention with
 618                  * another lookup + attr_set can't happen in the tiny window
 619                  * between if (...) and set S_NOCMTIME. */
 620                 if (!(inode->i_flags & S_NOCMTIME))
 621                         inode->i_flags |= S_NOCMTIME;
 622         }
 623
 624         GOTO(put, rc);
 625
 626 put:
 627         if (rc != 0) {
 628                 if (!IS_ERR(inode))
 629                         iput(inode);
 630
 631                 inode = ERR_PTR(rc);
 632         }
 633
 634         return inode;
 635 }
 636
 637 /**
 638  * \retval +v: new filter_fid, does not contain self-fid
 639  * \retval 0:  filter_fid_old, contains self-fid
 640  * \retval -v: other failure cases
 641  */
 642 int osd_get_idif(struct osd_thread_info *info, struct inode *inode,
 643                  struct dentry *dentry, struct lu_fid *fid)
 644 {
 645         struct filter_fid_old   *ff     = &info->oti_ff;
 646         struct ost_id           *ostid  = &info->oti_ostid;
 647         int                      rc;
 648
 649         rc = __osd_xattr_get(inode, dentry, XATTR_NAME_FID, ff, sizeof(*ff));
 650         if (rc == sizeof(*ff)) {
 651                 rc = 0;
 652                 ostid_set_seq(ostid, le64_to_cpu(ff->ff_seq));
 653                 ostid_set_id(ostid, le64_to_cpu(ff->ff_objid));
 654                 /* XXX: use 0 as the index for compatibility, the caller will
 655                  *      handle index related issues when necessarry. */
 656                 ostid_to_fid(fid, ostid, 0);
 657         } else if (rc == sizeof(struct filter_fid)) {
 658                 rc = 1;
 659         } else if (rc >= 0) {
 660                 rc = -EINVAL;
 661         }
 662
 663         return rc;
 664 }
 665
 666 static int osd_lma_self_repair(struct osd_thread_info *info,
 667                                struct osd_device *osd, struct inode *inode,
 668                                const struct lu_fid *fid, __u32 compat)
 669 {
 670         handle_t *jh;
 671         int       rc;
 672
 673         LASSERT(current->journal_info == NULL);
 674
 675         jh = osd_journal_start_sb(osd_sb(osd), LDISKFS_HT_MISC,
 676                                   osd_dto_credits_noquota[DTO_XATTR_SET]);
 677         if (IS_ERR(jh)) {
 678                 rc = PTR_ERR(jh);
 679                 CWARN("%s: cannot start journal for lma_self_repair: rc = %d\n",
 680                       osd_name(osd), rc);
 681                 return rc;
 682         }
 683
 684         rc = osd_ea_fid_set(info, inode, fid, compat, 0);
 685         if (rc != 0)
 686                 CWARN("%s: cannot self repair the LMA: rc = %d\n",
 687                       osd_name(osd), rc);
 688         ldiskfs_journal_stop(jh);
 689         return rc;
 690 }
 691
 692 static int osd_check_lma(const struct lu_env *env, struct osd_object *obj)
 693 {
 694         struct osd_thread_info  *info   = osd_oti_get(env);
 695         struct osd_device       *osd    = osd_obj2dev(obj);
 696         struct lustre_mdt_attrs *lma    = &info->oti_mdt_attrs;
 697         struct inode            *inode  = obj->oo_inode;
 698         struct dentry           *dentry = &info->oti_obj_dentry;
 699         struct lu_fid           *fid    = NULL;
 700         const struct lu_fid     *rfid   = lu_object_fid(&obj->oo_dt.do_lu);
 701         int                      rc;
 702         ENTRY;
 703
 704         CLASSERT(LMA_OLD_SIZE >= sizeof(*lma));
 705         rc = __osd_xattr_get(inode, dentry, XATTR_NAME_LMA,
 706                              info->oti_mdt_attrs_old, LMA_OLD_SIZE);
 707         if (rc == -ENODATA && !fid_is_igif(rfid) && osd->od_check_ff) {
 708                 fid = &lma->lma_self_fid;
 709                 rc = osd_get_idif(info, inode, dentry, fid);
 710                 if ((rc > 0) || (rc == -ENODATA && osd->od_index_in_idif)) {
 711                         /* For the given OST-object, if it has neither LMA nor
 712                          * FID in XATTR_NAME_FID, then the given FID (which is
 713                          * contained in the @obj, from client RPC for locating
 714                          * the OST-object) is trusted. We use it to generate
 715                          * the LMA. */
 716                         osd_lma_self_repair(info, osd, inode, rfid,
 717                                             LMAC_FID_ON_OST);
 718                         RETURN(0);
 719                 }
 720         }
 721
 722         if (rc < 0)
 723                 RETURN(rc);
 724
 725         if (rc > 0) {
 726                 rc = 0;
 727                 lustre_lma_swab(lma);
 728                 if (unlikely((lma->lma_incompat & ~LMA_INCOMPAT_SUPP) ||
 729                              CFS_FAIL_CHECK(OBD_FAIL_OSD_LMA_INCOMPAT))) {
 730                         CWARN("%s: unsupported incompat LMA feature(s) %#x for "
 731                               "fid = "DFID", ino = %lu\n", osd_name(osd),
 732                               lma->lma_incompat & ~LMA_INCOMPAT_SUPP,
 733                               PFID(rfid), inode->i_ino);
 734                         rc = -EOPNOTSUPP;
 735                 } else {
 736                         fid = &lma->lma_self_fid;
 737                 }
 738         }
 739
 740         if (fid != NULL && unlikely(!lu_fid_eq(rfid, fid))) {
 741                 if (fid_is_idif(rfid) && fid_is_idif(fid)) {
 742                         struct ost_id   *oi   = &info->oti_ostid;
 743                         struct lu_fid   *fid1 = &info->oti_fid3;
 744                         __u32            idx  = fid_idif_ost_idx(rfid);
 745
 746                         /* For old IDIF, the OST index is not part of the IDIF,
 747                          * Means that different OSTs may have the same IDIFs.
 748                          * Under such case, we need to make some compatible
 749                          * check to make sure to trigger OI scrub properly. */
 750                         if (idx != 0 && fid_idif_ost_idx(fid) == 0) {
 751                                 /* Given @rfid is new, LMA is old. */
 752                                 fid_to_ostid(fid, oi);
 753                                 ostid_to_fid(fid1, oi, idx);
 754                                 if (lu_fid_eq(fid1, rfid)) {
 755                                         if (osd->od_index_in_idif)
 756                                                 osd_lma_self_repair(info, osd,
 757                                                         inode, rfid,
 758                                                         LMAC_FID_ON_OST);
 759                                         RETURN(0);
 760                                 }
 761                         }
 762                 }
 763
 764                 rc = -EREMCHG;
 765         }
 766
 767         RETURN(rc);
 768 }
 769
 770 struct osd_check_lmv_buf {
 771 #ifdef HAVE_DIR_CONTEXT
 772         /* please keep it as first member */
 773         struct dir_context       ctx;
 774 #endif
 775         struct osd_thread_info  *oclb_info;
 776         struct osd_device       *oclb_dev;
 777         struct osd_idmap_cache  *oclb_oic;
 778 };
 779
 780 /**
 781  * It is called internally by ->readdir() to filter out the
 782  * local slave object's FID of the striped directory.
 783  *
 784  * \retval      1 found the local slave's FID
 785  * \retval      0 continue to check next item
 786  * \retval      -ve for failure
 787  */
 788 #ifdef HAVE_FILLDIR_USE_CTX
 789 static int osd_stripe_dir_filldir(struct dir_context *buf,
 790 #else
 791 static int osd_stripe_dir_filldir(void *buf,
 792 #endif
 793                                   const char *name, int namelen,
 794                                   loff_t offset, __u64 ino, unsigned d_type)
 795 {
 796         struct osd_check_lmv_buf *oclb = (struct osd_check_lmv_buf *)buf;
 797         struct osd_thread_info *oti = oclb->oclb_info;
 798         struct lu_fid *fid = &oti->oti_fid3;
 799         struct osd_inode_id *id = &oti->oti_id3;
 800         struct osd_device *dev = oclb->oclb_dev;
 801         struct osd_idmap_cache *oic = oclb->oclb_oic;
 802         struct inode *inode;
 803         int rc;
 804
 805         if (name[0] == '.')
 806                 return 0;
 807
 808         fid_zero(fid);
 809         sscanf(name + 1, SFID, RFID(fid));
 810         if (!fid_is_sane(fid))
 811                 return 0;
 812
 813         if (osd_remote_fid(oti->oti_env, dev, fid))
 814                 return 0;
 815
 816         osd_id_gen(id, ino, OSD_OII_NOGEN);
 817         inode = osd_iget(oti, dev, id);
 818         if (IS_ERR(inode))
 819                 return PTR_ERR(inode);
 820
 821         iput(inode);
 822         osd_add_oi_cache(oti, dev, id, fid);
 823         oic->oic_fid = *fid;
 824         oic->oic_lid = *id;
 825         oic->oic_dev = dev;
 826         rc = osd_oii_insert(dev, oic, true);
 827
 828         return rc == 0 ? 1 : rc;
 829 }
 830
 831 /* When lookup item under striped directory, we need to locate the master
 832  * MDT-object of the striped directory firstly, then the client will send
 833  * lookup (getattr_by_name) RPC to the MDT with some slave MDT-object's FID
 834  * and the item's name. If the system is restored from MDT file level backup,
 835  * then before the OI scrub completely built the OI files, the OI mappings of
 836  * the master MDT-object and slave MDT-object may be invalid. Usually, it is
 837  * not a problem for the master MDT-object. Because when locate the master
 838  * MDT-object, we will do name based lookup (for the striped directory itself)
 839  * firstly, during such process we can setup the correct OI mapping for the
 840  * master MDT-object. But it will be trouble for the slave MDT-object. Because
 841  * the client will not trigger name based lookup on the MDT to locate the slave
 842  * MDT-object before locating item under the striped directory, then when
 843  * osd_fid_lookup(), it will find that the OI mapping for the slave MDT-object
 844  * is invalid and does not know what the right OI mapping is, then the MDT has
 845  * to return -EINPROGRESS to the client to notify that the OI scrub is rebuiding
 846  * the OI file, related OI mapping is unknown yet, please try again later. And
 847  * then client will re-try the RPC again and again until related OI mapping has
 848  * been updated. That is quite inefficient.
 849  *
 850  * To resolve above trouble, we will handle it as the following two cases:
 851  *
 852  * 1) The slave MDT-object and the master MDT-object are on different MDTs.
 853  *    It is relative easy. Be as one of remote MDT-objects, the slave MDT-object
 854  *    is linked under /REMOTE_PARENT_DIR with the name of its FID string.
 855  *    We can locate the slave MDT-object via lookup the /REMOTE_PARENT_DIR
 856  *    directly. Please check osd_fid_lookup().
 857  *
 858  * 2) The slave MDT-object and the master MDT-object reside on the same MDT.
 859  *    Under such case, during lookup the master MDT-object, we will lookup the
 860  *    slave MDT-object via readdir against the master MDT-object, because the
 861  *    slave MDT-objects information are stored as sub-directories with the name
 862  *    "${FID}:${index}". Then when find the local slave MDT-object, its OI
 863  *    mapping will be recorded. Then subsequent osd_fid_lookup() will know
 864  *    the correct OI mapping for the slave MDT-object. */
 865 static int osd_check_lmv(struct osd_thread_info *oti, struct osd_device *dev,
 866                          struct inode *inode, struct osd_idmap_cache *oic)
 867 {
 868         struct lu_buf *buf = &oti->oti_big_buf;
 869         struct dentry *dentry = &oti->oti_obj_dentry;
 870         struct file *filp = &oti->oti_file;
 871         const struct file_operations *fops;
 872         struct lmv_mds_md_v1 *lmv1;
 873         struct osd_check_lmv_buf oclb = {
 874 #ifdef HAVE_DIR_CONTEXT
 875                 .ctx.actor = osd_stripe_dir_filldir,
 876 #endif
 877                 .oclb_info = oti,
 878                 .oclb_dev = dev,
 879                 .oclb_oic = oic
 880         };
 881         int rc = 0;
 882         ENTRY;
 883
 884 again:
 885         rc = __osd_xattr_get(inode, dentry, XATTR_NAME_LMV, buf->lb_buf,
 886                              buf->lb_len);
 887         if (rc == -ERANGE) {
 888                 rc = __osd_xattr_get(inode, dentry, XATTR_NAME_LMV, NULL, 0);
 889                 if (rc > 0) {
 890                         lu_buf_realloc(buf, rc);
 891                         if (buf->lb_buf == NULL)
 892                                 GOTO(out, rc = -ENOMEM);
 893
 894                         goto again;
 895                 }
 896         }
 897
 898         if (unlikely(rc == 0 || rc == -ENODATA))
 899                 GOTO(out, rc = 0);
 900
 901         if (rc < 0)
 902                 GOTO(out, rc);
 903
 904         if (unlikely(buf->lb_buf == NULL)) {
 905                 lu_buf_realloc(buf, rc);
 906                 if (buf->lb_buf == NULL)
 907                         GOTO(out, rc = -ENOMEM);
 908
 909                 goto again;
 910         }
 911
 912         lmv1 = buf->lb_buf;
 913         if (le32_to_cpu(lmv1->lmv_magic) != LMV_MAGIC_V1)
 914                 GOTO(out, rc = 0);
 915
 916         fops = inode->i_fop;
 917         dentry->d_inode = inode;
 918         dentry->d_sb = inode->i_sb;
 919         filp->f_pos = 0;
 920         filp->f_path.dentry = dentry;
 921         filp->f_mode = FMODE_64BITHASH;
 922         filp->f_mapping = inode->i_mapping;
 923         filp->f_op = fops;
 924         filp->private_data = NULL;
 925         set_file_inode(filp, inode);
 926
 927 #ifdef HAVE_DIR_CONTEXT
 928         oclb.ctx.pos = filp->f_pos;
 929         rc = fops->iterate(filp, &oclb.ctx);
 930         filp->f_pos = oclb.ctx.pos;
 931 #else
 932         rc = fops->readdir(filp, &oclb, osd_stripe_dir_filldir);
 933 #endif
 934         fops->release(inode, filp);
 935
 936 out:
 937         if (rc < 0)
 938                 CDEBUG(D_LFSCK, "%.16s: fail to check LMV EA, inode = %lu/%u,"
 939                        DFID": rc = %d\n",
 940                        LDISKFS_SB(inode->i_sb)->s_es->s_volume_name,
 941                        inode->i_ino, inode->i_generation,
 942                        PFID(&oic->oic_fid), rc);
 943         else
 944                 rc = 0;
 945
 946         RETURN(rc);
 947 }
 948
 949 static int osd_fid_lookup(const struct lu_env *env, struct osd_object *obj,
 950                           const struct lu_fid *fid,
 951                           const struct lu_object_conf *conf)
 952 {
 953         struct osd_thread_info *info;
 954         struct lu_device       *ldev   = obj->oo_dt.do_lu.lo_dev;
 955         struct osd_device      *dev;
 956         struct osd_idmap_cache *oic;
 957         struct osd_inode_id    *id;
 958         struct inode           *inode = NULL;
 959         struct osd_scrub       *scrub;
 960         struct scrub_file      *sf;
 961         __u32                   flags = SS_CLEAR_DRYRUN | SS_CLEAR_FAILOUT |
 962                                         SS_AUTO_FULL;
 963         int                     result  = 0;
 964         int                     rc1     = 0;
 965         bool                    cached  = true;
 966         bool                    remote  = false;
 967         ENTRY;
 968
 969         LINVRNT(osd_invariant(obj));
 970         LASSERT(obj->oo_inode == NULL);
 971         LASSERTF(fid_is_sane(fid) || fid_is_idif(fid), DFID"\n", PFID(fid));
 972
 973         dev = osd_dev(ldev);
 974         scrub = &dev->od_scrub;
 975         sf = &scrub->os_file;
 976         info = osd_oti_get(env);
 977         LASSERT(info);
 978         oic = &info->oti_cache;
 979
 980         if (OBD_FAIL_CHECK(OBD_FAIL_SRV_ENOENT))
 981                 RETURN(-ENOENT);
 982
 983         /* For the object is created as locking anchor, or for the object to
 984          * be created on disk. No need to osd_oi_lookup() at here because FID
 985          * shouldn't never be re-used, if it's really a duplicate FID from
 986          * unexpected reason, we should be able to detect it later by calling
 987          * do_create->osd_oi_insert(). */
 988         if (conf != NULL && conf->loc_flags & LOC_F_NEW)
 989                 GOTO(out, result = 0);
 990
 991         /* Search order: 1. per-thread cache. */
 992         if (lu_fid_eq(fid, &oic->oic_fid) &&
 993             likely(oic->oic_dev == dev)) {
 994                 id = &oic->oic_lid;
 995                 goto iget;
 996         }
 997
 998         id = &info->oti_id;
 999         if (!list_empty(&scrub->os_inconsistent_items)) {
1000                 /* Search order: 2. OI scrub pending list. */
1001                 result = osd_oii_lookup(dev, fid, id);
1002                 if (result == 0)
1003                         goto iget;
1004         }
1005
1006         cached = false;
1007         /* Search order: 3. OI files. */
1008         result = osd_oi_lookup(info, dev, fid, id, OI_CHECK_FLD);
1009         if (result == -ENOENT) {
1010                 if (!(fid_is_norm(fid) || fid_is_igif(fid)) ||
1011                     fid_is_on_ost(info, dev, fid, OI_CHECK_FLD) ||
1012                     !ldiskfs_test_bit(osd_oi_fid2idx(dev,fid),
1013                                       sf->sf_oi_bitmap))
1014                         GOTO(out, result = 0);
1015
1016                 goto trigger;
1017         }
1018
1019         if (result != 0)
1020                 GOTO(out, result);
1021
1022 iget:
1023         inode = osd_iget_check(info, dev, fid, id, cached);
1024         if (IS_ERR(inode)) {
1025                 result = PTR_ERR(inode);
1026                 if (result == -ENOENT || result == -ESTALE)
1027                         GOTO(out, result = -ENOENT);
1028
1029                 if (result == -EREMCHG) {
1030
1031 trigger:
1032                         /* We still have chance to get the valid inode: for the
1033                          * object which is referenced by remote name entry, the
1034                          * object on the local MDT will be linked under the dir
1035                          * of "/REMOTE_PARENT_DIR" with its FID string as name.
1036                          *
1037                          * We do not know whether the object for the given FID
1038                          * is referenced by some remote name entry or not, and
1039                          * especially for DNE II, a multiple-linked object may
1040                          * have many name entries reside on many MDTs.
1041                          *
1042                          * To simplify the operation, OSD will not distinguish
1043                          * more, just lookup "/REMOTE_PARENT_DIR". Usually, it
1044                          * only happened for the RPC from other MDT during the
1045                          * OI scrub, or for the client side RPC with FID only,
1046                          * such as FID to path, or from old connected client. */
1047                         if (!remote &&
1048                             !fid_is_on_ost(info, dev, fid, OI_CHECK_FLD)) {
1049                                 rc1 = osd_lookup_in_remote_parent(info, dev,
1050                                                                   fid, id);
1051                                 if (rc1 == 0) {
1052                                         remote = true;
1053                                         cached = true;
1054                                         flags |= SS_AUTO_PARTIAL;
1055                                         flags &= ~SS_AUTO_FULL;
1056                                         goto iget;
1057                                 }
1058                         }
1059
1060                         if (thread_is_running(&scrub->os_thread)) {
1061                                 if (scrub->os_partial_scan &&
1062                                     !scrub->os_in_join) {
1063                                         goto join;
1064                                 } else {
1065                                         if (inode != NULL && !IS_ERR(inode)) {
1066                                                 LASSERT(remote);
1067
1068                                                 osd_add_oi_cache(info, dev, id,
1069                                                                  fid);
1070                                                 osd_oii_insert(dev, oic, true);
1071                                         } else {
1072                                                 result = -EINPROGRESS;
1073                                         }
1074                                 }
1075                         } else if (!dev->od_noscrub) {
1076
1077 join:
1078                                 rc1 = osd_scrub_start(dev, flags);
1079                                 LCONSOLE_WARN("%.16s: trigger OI scrub by RPC "
1080                                               "for the "DFID" with flags 0x%x,"
1081                                               " rc = %d\n", osd_name(dev),
1082                                               PFID(fid), flags, rc1);
1083                                 if (rc1 == 0 || rc1 == -EALREADY) {
1084                                         if (inode != NULL && !IS_ERR(inode)) {
1085                                                 LASSERT(remote);
1086
1087                                                 osd_add_oi_cache(info, dev, id,
1088                                                                  fid);
1089                                                 osd_oii_insert(dev, oic, true);
1090                                         } else {
1091                                                 result = -EINPROGRESS;
1092                                         }
1093                                 } else {
1094                                         result = -EREMCHG;
1095                                 }
1096                         } else {
1097                                 result = -EREMCHG;
1098                         }
1099                 }
1100
1101                 if (inode == NULL || IS_ERR(inode))
1102                         GOTO(out, result);
1103         } else if (remote) {
1104                 goto trigger;
1105         }
1106
1107         obj->oo_inode = inode;
1108         LASSERT(obj->oo_inode->i_sb == osd_sb(dev));
1109
1110         result = osd_check_lma(env, obj);
1111         if (result != 0) {
1112                 if (result == -ENODATA) {
1113                         if (cached) {
1114                                 result = osd_oi_lookup(info, dev, fid, id,
1115                                                        OI_CHECK_FLD);
1116                                 if (result != 0) {
1117                                         /* result == -ENOENT means that the OI
1118                                          * mapping has been removed by race,
1119                                          * the target inode belongs to other
1120                                          * object.
1121                                          *
1122                                          * Others error also can be returned
1123                                          * directly. */
1124                                         iput(inode);
1125                                         obj->oo_inode = NULL;
1126                                         GOTO(out, result);
1127                                 } else {
1128                                         /* result == 0 means the cached OI
1129                                          * mapping is still in the OI file,
1130                                          * the target the inode is valid. */
1131                                 }
1132                         } else {
1133                                 /* The current OI mapping is from the OI file,
1134                                  * since the inode has been found via
1135                                  * osd_iget_check(), no need recheck OI. */
1136                         }
1137
1138                         goto found;
1139                 }
1140
1141                 iput(inode);
1142                 inode = NULL;
1143                 obj->oo_inode = NULL;
1144                 if (result != -EREMCHG)
1145                         GOTO(out, result);
1146
1147                 if (cached) {
1148                         result = osd_oi_lookup(info, dev, fid, id,
1149                                                OI_CHECK_FLD);
1150                         /* result == -ENOENT means the cached OI mapping
1151                          * has been removed from the OI file by race,
1152                          * above target inode belongs to other object.
1153                          *
1154                          * Others error also can be returned directly. */
1155                         if (result != 0)
1156                                 GOTO(out, result);
1157
1158                         /* result == 0, goto trigger */
1159                 } else {
1160                         /* The current OI mapping is from the OI file,
1161                          * since the inode has been found via
1162                          * osd_iget_check(), no need recheck OI. */
1163                 }
1164
1165                 goto trigger;
1166         }
1167
1168 found:
1169         obj->oo_compat_dot_created = 1;
1170         obj->oo_compat_dotdot_created = 1;
1171
1172         if (S_ISDIR(inode->i_mode) &&
1173             (flags & SS_AUTO_PARTIAL || sf->sf_status == SS_SCANNING))
1174                 osd_check_lmv(info, dev, inode, oic);
1175
1176         if (!ldiskfs_pdo)
1177                 GOTO(out, result = 0);
1178
1179         LASSERT(obj->oo_hl_head == NULL);
1180         obj->oo_hl_head = ldiskfs_htree_lock_head_alloc(HTREE_HBITS_DEF);
1181         if (obj->oo_hl_head == NULL) {
1182                 obj->oo_inode = NULL;
1183                 iput(inode);
1184                 GOTO(out, result = -ENOMEM);
1185         }
1186         GOTO(out, result = 0);
1187
1188 out:
1189         if (result != 0 && cached)
1190                 fid_zero(&oic->oic_fid);
1191
1192         LINVRNT(osd_invariant(obj));
1193         return result;
1194 }
1195
1196 /*
1197  * Concurrency: shouldn't matter.
1198  */
1199 static void osd_object_init0(struct osd_object *obj)
1200 {
1201         LASSERT(obj->oo_inode != NULL);
1202         obj->oo_dt.do_body_ops = &osd_body_ops;
1203         obj->oo_dt.do_lu.lo_header->loh_attr |=
1204                 (LOHA_EXISTS | (obj->oo_inode->i_mode & S_IFMT));
1205 }
1206
1207 /*
1208  * Concurrency: no concurrent access is possible that early in object
1209  * life-cycle.
1210  */
1211 static int osd_object_init(const struct lu_env *env, struct lu_object *l,
1212                            const struct lu_object_conf *conf)
1213 {
1214         struct osd_object *obj = osd_obj(l);
1215         int result;
1216
1217         LINVRNT(osd_invariant(obj));
1218
1219         if (fid_is_otable_it(&l->lo_header->loh_fid)) {
1220                 obj->oo_dt.do_ops = &osd_obj_otable_it_ops;
1221                 l->lo_header->loh_attr |= LOHA_EXISTS;
1222                 return 0;
1223         }
1224
1225         result = osd_fid_lookup(env, obj, lu_object_fid(l), conf);
1226         obj->oo_dt.do_body_ops = &osd_body_ops_new;
1227         if (result == 0 && obj->oo_inode != NULL) {
1228                 struct osd_thread_info *oti = osd_oti_get(env);
1229                 struct lustre_mdt_attrs *lma = &oti->oti_mdt_attrs;
1230
1231                 osd_object_init0(obj);
1232                 result = osd_get_lma(oti, obj->oo_inode,
1233                                      &oti->oti_obj_dentry, lma);
1234                 if (result == 0) {
1235                         /* Convert LMAI flags to lustre LMA flags
1236                          * and cache it to oo_lma_flags */
1237                         obj->oo_lma_flags =
1238                                 lma_to_lustre_flags(lma->lma_incompat);
1239                 } else if (result == -ENODATA) {
1240                         result = 0;
1241                 }
1242         }
1243
1244         LINVRNT(osd_invariant(obj));
1245         return result;
1246 }
1247
1248 /* The first part of oxe_buf is xattr name, and is '\0' terminated.
1249  * The left part is for value, binary mode. */
1250 struct osd_xattr_entry {
1251         struct list_head        oxe_list;
1252         size_t                  oxe_len;
1253         size_t                  oxe_namelen;
1254         bool                    oxe_exist;
1255         struct rcu_head         oxe_rcu;
1256         char                    oxe_buf[0];
1257 };
1258
1259 static struct osd_xattr_entry *osd_oxc_lookup(struct osd_object *obj,
1260                                               const char *name,
1261                                               size_t namelen)
1262 {
1263         struct osd_xattr_entry *oxe;
1264
1265         list_for_each_entry(oxe, &obj->oo_xattr_list, oxe_list) {
1266                 if (namelen == oxe->oxe_namelen &&
1267                     strncmp(name, oxe->oxe_buf, namelen) == 0)
1268                         return oxe;
1269         }
1270
1271         return NULL;
1272 }
1273
1274 static int osd_oxc_get(struct osd_object *obj, const char *name,
1275                        struct lu_buf *buf)
1276 {
1277         struct osd_xattr_entry *oxe;
1278         size_t vallen;
1279         ENTRY;
1280
1281         rcu_read_lock();
1282         oxe = osd_oxc_lookup(obj, name, strlen(name));
1283         if (oxe == NULL) {
1284                 rcu_read_unlock();
1285                 RETURN(-ENOENT);
1286         }
1287
1288         if (!oxe->oxe_exist) {
1289                 rcu_read_unlock();
1290                 RETURN(-ENODATA);
1291         }
1292
1293         vallen = oxe->oxe_len - sizeof(*oxe) - oxe->oxe_namelen - 1;
1294         LASSERT(vallen > 0);
1295
1296         if (buf->lb_buf == NULL) {
1297                 rcu_read_unlock();
1298                 RETURN(vallen);
1299         }
1300
1301         if (buf->lb_len < vallen) {
1302                 rcu_read_unlock();
1303                 RETURN(-ERANGE);
1304         }
1305
1306         memcpy(buf->lb_buf, oxe->oxe_buf + oxe->oxe_namelen + 1, vallen);
1307         rcu_read_unlock();
1308
1309         RETURN(vallen);
1310 }
1311
1312 static void osd_oxc_free(struct rcu_head *head)
1313 {
1314         struct osd_xattr_entry *oxe;
1315
1316         oxe = container_of(head, struct osd_xattr_entry, oxe_rcu);
1317         OBD_FREE(oxe, oxe->oxe_len);
1318 }
1319
1320 static inline void __osd_oxc_del(struct osd_object *obj, const char *name)
1321 {
1322         struct osd_xattr_entry *oxe;
1323
1324         oxe = osd_oxc_lookup(obj, name, strlen(name));
1325         if (oxe != NULL) {
1326                 list_del(&oxe->oxe_list);
1327                 call_rcu(&oxe->oxe_rcu, osd_oxc_free);
1328         }
1329 }
1330
1331 static void osd_oxc_add(struct osd_object *obj, const char *name,
1332                         const char *buf, int buflen)
1333 {
1334         struct osd_xattr_entry *oxe;
1335         size_t namelen = strlen(name);
1336         size_t len = sizeof(*oxe) + namelen + 1 + buflen;
1337
1338         OBD_ALLOC(oxe, len);
1339         if (oxe == NULL)
1340                 return;
1341
1342         INIT_LIST_HEAD(&oxe->oxe_list);
1343         oxe->oxe_len = len;
1344         oxe->oxe_namelen = namelen;
1345         memcpy(oxe->oxe_buf, name, namelen);
1346         if (buflen > 0) {
1347                 LASSERT(buf != NULL);
1348                 memcpy(oxe->oxe_buf + namelen + 1, buf, buflen);
1349                 oxe->oxe_exist = true;
1350         } else {
1351                 oxe->oxe_exist = false;
1352         }
1353
1354         /* this should be rarely called, just remove old and add new */
1355         spin_lock(&obj->oo_guard);
1356         __osd_oxc_del(obj, name);
1357         list_add_tail(&oxe->oxe_list, &obj->oo_xattr_list);
1358         spin_unlock(&obj->oo_guard);
1359 }
1360
1361 static void osd_oxc_del(struct osd_object *obj, const char *name)
1362 {
1363         spin_lock(&obj->oo_guard);
1364         __osd_oxc_del(obj, name);
1365         spin_unlock(&obj->oo_guard);
1366 }
1367
1368 static void osd_oxc_fini(struct osd_object *obj)
1369 {
1370         struct osd_xattr_entry *oxe, *next;
1371
1372         list_for_each_entry_safe(oxe, next, &obj->oo_xattr_list, oxe_list) {
1373                 list_del(&oxe->oxe_list);
1374                 OBD_FREE(oxe, oxe->oxe_len);
1375         }
1376 }
1377
1378 /*
1379  * Concurrency: no concurrent access is possible that late in object
1380  * life-cycle.
1381  */
1382 static void osd_object_free(const struct lu_env *env, struct lu_object *l)
1383 {
1384         struct osd_object *obj = osd_obj(l);
1385
1386         LINVRNT(osd_invariant(obj));
1387
1388         osd_oxc_fini(obj);
1389         dt_object_fini(&obj->oo_dt);
1390         if (obj->oo_hl_head != NULL)
1391                 ldiskfs_htree_lock_head_free(obj->oo_hl_head);
1392         OBD_FREE_PTR(obj);
1393 }
1394
1395 /*
1396  * Concurrency: no concurrent access is possible that late in object
1397  * life-cycle.
1398  */
1399 static void osd_index_fini(struct osd_object *o)
1400 {
1401         struct iam_container *bag;
1402
1403         if (o->oo_dir != NULL) {
1404                 bag = &o->oo_dir->od_container;
1405                 if (o->oo_inode != NULL) {
1406                         if (bag->ic_object == o->oo_inode)
1407                                 iam_container_fini(bag);
1408                 }
1409                 OBD_FREE_PTR(o->oo_dir);
1410                 o->oo_dir = NULL;
1411         }
1412 }
1413
1414 /*
1415  * Concurrency: no concurrent access is possible that late in object
1416  * life-cycle (for all existing callers, that is. New callers have to provide
1417  * their own locking.)
1418  */
1419 static int osd_inode_unlinked(const struct inode *inode)
1420 {
1421         return inode->i_nlink == 0;
1422 }
1423
1424 enum {
1425         OSD_TXN_OI_DELETE_CREDITS    = 20,
1426         OSD_TXN_INODE_DELETE_CREDITS = 20
1427 };
1428
1429 /*
1430  * Journal
1431  */
1432
1433 #if OSD_THANDLE_STATS
1434 /**
1435  * Set time when the handle is allocated
1436  */
1437 static void osd_th_alloced(struct osd_thandle *oth)
1438 {
1439         oth->oth_alloced = cfs_time_current();
1440 }
1441
1442 /**
1443  * Set time when the handle started
1444  */
1445 static void osd_th_started(struct osd_thandle *oth)
1446 {
1447         oth->oth_started = cfs_time_current();
1448 }
1449
1450 /**
1451  * Helper function to convert time interval to microseconds packed in
1452  * long int.
1453  */
1454 static long interval_to_usec(cfs_time_t start, cfs_time_t end)
1455 {
1456         struct timeval val;
1457
1458         cfs_duration_usec(cfs_time_sub(end, start), &val);
1459         return val.tv_sec * 1000000 + val.tv_usec;
1460 }
1461
1462 /**
1463  * Check whether the we deal with this handle for too long.
1464  */
1465 static void __osd_th_check_slow(void *oth, struct osd_device *dev,
1466                                 cfs_time_t alloced, cfs_time_t started,
1467                                 cfs_time_t closed)
1468 {
1469         cfs_time_t now = cfs_time_current();
1470
1471         LASSERT(dev != NULL);
1472
1473         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_STARTING,
1474                             interval_to_usec(alloced, started));
1475         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_OPEN,
1476                             interval_to_usec(started, closed));
1477         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_CLOSING,
1478                             interval_to_usec(closed, now));
1479
1480         if (cfs_time_before(cfs_time_add(alloced, cfs_time_seconds(30)), now)) {
1481                 CWARN("transaction handle %p was open for too long: "
1482                       "now "CFS_TIME_T" ,"
1483                       "alloced "CFS_TIME_T" ,"
1484                       "started "CFS_TIME_T" ,"
1485                       "closed "CFS_TIME_T"\n",
1486                       oth, now, alloced, started, closed);
1487                 libcfs_debug_dumpstack(NULL);
1488         }
1489 }
1490
1491 #define OSD_CHECK_SLOW_TH(oth, dev, expr)                               \
1492 {                                                                       \
1493         cfs_time_t __closed = cfs_time_current();                       \
1494         cfs_time_t __alloced = oth->oth_alloced;                        \
1495         cfs_time_t __started = oth->oth_started;                        \
1496                                                                         \
1497         expr;                                                           \
1498         __osd_th_check_slow(oth, dev, __alloced, __started, __closed);  \
1499 }
1500
1501 #else /* OSD_THANDLE_STATS */
1502
1503 #define osd_th_alloced(h)                  do {} while(0)
1504 #define osd_th_started(h)                  do {} while(0)
1505 #define OSD_CHECK_SLOW_TH(oth, dev, expr)  expr
1506
1507 #endif /* OSD_THANDLE_STATS */
1508
1509 /*
1510  * Concurrency: doesn't access mutable data.
1511  */
1512 static int osd_param_is_not_sane(const struct osd_device *dev,
1513                                  const struct thandle *th)
1514 {
1515         struct osd_thandle *oh = container_of(th, typeof(*oh), ot_super);
1516
1517         return oh->ot_credits > osd_transaction_size(dev);
1518 }
1519
1520 /*
1521  * Concurrency: shouldn't matter.
1522  */
1523 static void osd_trans_commit_cb(struct super_block *sb,
1524                                 struct ldiskfs_journal_cb_entry *jcb, int error)
1525 {
1526         struct osd_thandle *oh = container_of0(jcb, struct osd_thandle, ot_jcb);
1527         struct thandle     *th  = &oh->ot_super;
1528         struct lu_device   *lud = &th->th_dev->dd_lu_dev;
1529         struct dt_txn_commit_cb *dcb, *tmp;
1530
1531         LASSERT(oh->ot_handle == NULL);
1532
1533         if (error)
1534                 CERROR("transaction @0x%p commit error: %d\n", th, error);
1535
1536         dt_txn_hook_commit(th);
1537
1538         /* call per-transaction callbacks if any */
1539         list_for_each_entry_safe(dcb, tmp, &oh->ot_commit_dcb_list,
1540                                  dcb_linkage) {
1541                 LASSERTF(dcb->dcb_magic == TRANS_COMMIT_CB_MAGIC,
1542                          "commit callback entry: magic=%x name='%s'\n",
1543                          dcb->dcb_magic, dcb->dcb_name);
1544                 list_del_init(&dcb->dcb_linkage);
1545                 dcb->dcb_func(NULL, th, dcb, error);
1546         }
1547
1548         lu_ref_del_at(&lud->ld_reference, &oh->ot_dev_link, "osd-tx", th);
1549         lu_device_put(lud);
1550         th->th_dev = NULL;
1551
1552         lu_context_exit(&th->th_ctx);
1553         lu_context_fini(&th->th_ctx);
1554         OBD_FREE_PTR(oh);
1555 }
1556
1557 #ifndef HAVE_SB_START_WRITE
1558 # define sb_start_write(sb) do {} while (0)
1559 # define sb_end_write(sb) do {} while (0)
1560 #endif
1561
1562 static struct thandle *osd_trans_create(const struct lu_env *env,
1563                                         struct dt_device *d)
1564 {
1565         struct osd_thread_info  *oti = osd_oti_get(env);
1566         struct osd_iobuf        *iobuf = &oti->oti_iobuf;
1567         struct osd_thandle      *oh;
1568         struct thandle          *th;
1569         ENTRY;
1570
1571         /* on pending IO in this thread should left from prev. request */
1572         LASSERT(atomic_read(&iobuf->dr_numreqs) == 0);
1573
1574         sb_start_write(osd_sb(osd_dt_dev(d)));
1575
1576         OBD_ALLOC_GFP(oh, sizeof *oh, GFP_NOFS);
1577         if (oh != NULL) {
1578                 oh->ot_quota_trans = &oti->oti_quota_trans;
1579                 memset(oh->ot_quota_trans, 0, sizeof(*oh->ot_quota_trans));
1580                 th = &oh->ot_super;
1581                 th->th_dev = d;
1582                 th->th_result = 0;
1583                 th->th_tags = LCT_TX_HANDLE;
1584                 oh->ot_credits = 0;
1585                 INIT_LIST_HEAD(&oh->ot_commit_dcb_list);
1586                 INIT_LIST_HEAD(&oh->ot_stop_dcb_list);
1587                 osd_th_alloced(oh);
1588
1589                 memset(oti->oti_declare_ops, 0,
1590                        sizeof(oti->oti_declare_ops));
1591                 memset(oti->oti_declare_ops_cred, 0,
1592                        sizeof(oti->oti_declare_ops_cred));
1593                 memset(oti->oti_declare_ops_used, 0,
1594                        sizeof(oti->oti_declare_ops_used));
1595         } else {
1596                 sb_end_write(osd_sb(osd_dt_dev(d)));
1597                 th = ERR_PTR(-ENOMEM);
1598         }
1599         RETURN(th);
1600 }
1601
1602 void osd_trans_dump_creds(const struct lu_env *env, struct thandle *th)
1603 {
1604         struct osd_thread_info  *oti = osd_oti_get(env);
1605         struct osd_thandle      *oh;
1606
1607         oh = container_of0(th, struct osd_thandle, ot_super);
1608         LASSERT(oh != NULL);
1609
1610         CWARN("  create: %u/%u/%u, destroy: %u/%u/%u\n",
1611               oti->oti_declare_ops[OSD_OT_CREATE],
1612               oti->oti_declare_ops_cred[OSD_OT_CREATE],
1613               oti->oti_declare_ops_used[OSD_OT_CREATE],
1614               oti->oti_declare_ops[OSD_OT_DESTROY],
1615               oti->oti_declare_ops_cred[OSD_OT_DESTROY],
1616               oti->oti_declare_ops_used[OSD_OT_DESTROY]);
1617         CWARN("  attr_set: %u/%u/%u, xattr_set: %u/%u/%u\n",
1618               oti->oti_declare_ops[OSD_OT_ATTR_SET],
1619               oti->oti_declare_ops_cred[OSD_OT_ATTR_SET],
1620               oti->oti_declare_ops_used[OSD_OT_ATTR_SET],
1621               oti->oti_declare_ops[OSD_OT_XATTR_SET],
1622               oti->oti_declare_ops_cred[OSD_OT_XATTR_SET],
1623               oti->oti_declare_ops_used[OSD_OT_XATTR_SET]);
1624         CWARN("  write: %u/%u/%u, punch: %u/%u/%u, quota %u/%u/%u\n",
1625               oti->oti_declare_ops[OSD_OT_WRITE],
1626               oti->oti_declare_ops_cred[OSD_OT_WRITE],
1627               oti->oti_declare_ops_used[OSD_OT_WRITE],
1628               oti->oti_declare_ops[OSD_OT_PUNCH],
1629               oti->oti_declare_ops_cred[OSD_OT_PUNCH],
1630               oti->oti_declare_ops_used[OSD_OT_PUNCH],
1631               oti->oti_declare_ops[OSD_OT_QUOTA],
1632               oti->oti_declare_ops_cred[OSD_OT_QUOTA],
1633               oti->oti_declare_ops_used[OSD_OT_QUOTA]);
1634         CWARN("  insert: %u/%u/%u, delete: %u/%u/%u\n",
1635               oti->oti_declare_ops[OSD_OT_INSERT],
1636               oti->oti_declare_ops_cred[OSD_OT_INSERT],
1637               oti->oti_declare_ops_used[OSD_OT_INSERT],
1638               oti->oti_declare_ops[OSD_OT_DELETE],
1639               oti->oti_declare_ops_cred[OSD_OT_DELETE],
1640               oti->oti_declare_ops_used[OSD_OT_DELETE]);
1641         CWARN("  ref_add: %u/%u/%u, ref_del: %u/%u/%u\n",
1642               oti->oti_declare_ops[OSD_OT_REF_ADD],
1643               oti->oti_declare_ops_cred[OSD_OT_REF_ADD],
1644               oti->oti_declare_ops_used[OSD_OT_REF_ADD],
1645               oti->oti_declare_ops[OSD_OT_REF_DEL],
1646               oti->oti_declare_ops_cred[OSD_OT_REF_DEL],
1647               oti->oti_declare_ops_used[OSD_OT_REF_DEL]);
1648 }
1649
1650 /*
1651  * Concurrency: shouldn't matter.
1652  */
1653 static int osd_trans_start(const struct lu_env *env, struct dt_device *d,
1654                            struct thandle *th)
1655 {
1656         struct osd_thread_info *oti = osd_oti_get(env);
1657         struct osd_device  *dev = osd_dt_dev(d);
1658         handle_t           *jh;
1659         struct osd_thandle *oh;
1660         int rc;
1661
1662         ENTRY;
1663
1664         LASSERT(current->journal_info == NULL);
1665
1666         oh = container_of0(th, struct osd_thandle, ot_super);
1667         LASSERT(oh != NULL);
1668         LASSERT(oh->ot_handle == NULL);
1669
1670         rc = dt_txn_hook_start(env, d, th);
1671         if (rc != 0)
1672                 GOTO(out, rc);
1673
1674         if (unlikely(osd_param_is_not_sane(dev, th))) {
1675                 static unsigned long last_printed;
1676                 static int last_credits;
1677
1678                 /* don't make noise on a tiny testing systems
1679                  * actual credits misuse will be caught anyway */
1680                 if (last_credits != oh->ot_credits &&
1681                     time_after(jiffies, last_printed +
1682                                msecs_to_jiffies(60 * MSEC_PER_SEC)) &&
1683                     osd_transaction_size(dev) > 512) {
1684                         osd_trans_dump_creds(env, th);
1685                         libcfs_debug_dumpstack(NULL);
1686                         last_credits = oh->ot_credits;
1687                         last_printed = jiffies;
1688                 }
1689                 /* XXX Limit the credits to 'max_transaction_buffers', and
1690                  *     let the underlying filesystem to catch the error if
1691                  *     we really need so many credits.
1692                  *
1693                  *     This should be removed when we can calculate the
1694                  *     credits precisely. */
1695                 oh->ot_credits = osd_transaction_size(dev);
1696         }
1697
1698         /*
1699          * XXX temporary stuff. Some abstraction layer should
1700          * be used.
1701          */
1702         jh = osd_journal_start_sb(osd_sb(dev), LDISKFS_HT_MISC, oh->ot_credits);
1703         osd_th_started(oh);
1704         if (!IS_ERR(jh)) {
1705                 oh->ot_handle = jh;
1706                 LASSERT(oti->oti_txns == 0);
1707                 lu_context_init(&th->th_ctx, th->th_tags);
1708                 lu_context_enter(&th->th_ctx);
1709
1710                 lu_device_get(&d->dd_lu_dev);
1711                 lu_ref_add_at(&d->dd_lu_dev.ld_reference, &oh->ot_dev_link,
1712                               "osd-tx", th);
1713                 oti->oti_txns++;
1714                 rc = 0;
1715         } else {
1716                 rc = PTR_ERR(jh);
1717         }
1718 out:
1719         RETURN(rc);
1720 }
1721
1722 static int osd_seq_exists(const struct lu_env *env,
1723                           struct osd_device *osd, u64 seq)
1724 {
1725         struct lu_seq_range     *range = &osd_oti_get(env)->oti_seq_range;
1726         struct seq_server_site  *ss = osd_seq_site(osd);
1727         int                     rc;
1728         ENTRY;
1729
1730         LASSERT(ss != NULL);
1731         LASSERT(ss->ss_server_fld != NULL);
1732
1733         rc = osd_fld_lookup(env, osd, seq, range);
1734         if (rc != 0) {
1735                 if (rc != -ENOENT)
1736                         CERROR("%s: can't lookup FLD sequence "LPX64
1737                                ": rc = %d\n", osd_name(osd), seq, rc);
1738                 RETURN(0);
1739         }
1740
1741         RETURN(ss->ss_node_id == range->lsr_index);
1742 }
1743
1744 static void osd_trans_stop_cb(struct osd_thandle *oth, int result)
1745 {
1746         struct dt_txn_commit_cb *dcb;
1747         struct dt_txn_commit_cb *tmp;
1748
1749         /* call per-transaction stop callbacks if any */
1750         list_for_each_entry_safe(dcb, tmp, &oth->ot_stop_dcb_list,
1751                                  dcb_linkage) {
1752                 LASSERTF(dcb->dcb_magic == TRANS_COMMIT_CB_MAGIC,
1753                          "commit callback entry: magic=%x name='%s'\n",
1754                          dcb->dcb_magic, dcb->dcb_name);
1755                 list_del_init(&dcb->dcb_linkage);
1756                 dcb->dcb_func(NULL, &oth->ot_super, dcb, result);
1757         }
1758 }
1759
1760 /*
1761  * Concurrency: shouldn't matter.
1762  */
1763 static int osd_trans_stop(const struct lu_env *env, struct dt_device *dt,
1764                           struct thandle *th)
1765 {
1766         int                     rc = 0, remove_agents = 0;
1767         struct osd_thandle     *oh;
1768         struct osd_thread_info *oti = osd_oti_get(env);
1769         struct osd_iobuf       *iobuf = &oti->oti_iobuf;
1770         struct osd_device      *osd = osd_dt_dev(th->th_dev);
1771         struct qsd_instance    *qsd = osd->od_quota_slave;
1772         struct lquota_trans    *qtrans;
1773         ENTRY;
1774
1775         oh = container_of0(th, struct osd_thandle, ot_super);
1776
1777         /* reset OI cache for safety */
1778         oti->oti_ins_cache_used = 0;
1779
1780         remove_agents = oh->ot_remove_agents;
1781
1782         qtrans = oh->ot_quota_trans;
1783         oh->ot_quota_trans = NULL;
1784
1785         if (oh->ot_handle != NULL) {
1786                 handle_t *hdl = oh->ot_handle;
1787
1788                 /*
1789                  * add commit callback
1790                  * notice we don't do this in osd_trans_start()
1791                  * as underlying transaction can change during truncate
1792                  */
1793                 ldiskfs_journal_callback_add(hdl, osd_trans_commit_cb,
1794                                          &oh->ot_jcb);
1795
1796                 LASSERT(oti->oti_txns == 1);
1797                 oti->oti_txns--;
1798
1799                 rc = dt_txn_hook_stop(env, th);
1800                 if (rc != 0)
1801                         CERROR("%s: failed in transaction hook: rc = %d\n",
1802                                osd_name(osd), rc);
1803
1804                 osd_trans_stop_cb(oh, rc);
1805                 /* hook functions might modify th_sync */
1806                 hdl->h_sync = th->th_sync;
1807
1808                 oh->ot_handle = NULL;
1809                 OSD_CHECK_SLOW_TH(oh, osd, rc = ldiskfs_journal_stop(hdl));
1810                 if (rc != 0)
1811                         CERROR("%s: failed to stop transaction: rc = %d\n",
1812                                osd_name(osd), rc);
1813         } else {
1814                 osd_trans_stop_cb(oh, th->th_result);
1815                 OBD_FREE_PTR(oh);
1816         }
1817
1818         /* inform the quota slave device that the transaction is stopping */
1819         qsd_op_end(env, qsd, qtrans);
1820
1821         /* as we want IO to journal and data IO be concurrent, we don't block
1822          * awaiting data IO completion in osd_do_bio(), instead we wait here
1823          * once transaction is submitted to the journal. all reqular requests
1824          * don't do direct IO (except read/write), thus this wait_event becomes
1825          * no-op for them.
1826          *
1827          * IMPORTANT: we have to wait till any IO submited by the thread is
1828          * completed otherwise iobuf may be corrupted by different request
1829          */
1830         wait_event(iobuf->dr_wait,
1831                        atomic_read(&iobuf->dr_numreqs) == 0);
1832         osd_fini_iobuf(osd, iobuf);
1833         if (!rc)
1834                 rc = iobuf->dr_error;
1835
1836         if (unlikely(remove_agents != 0))
1837                 osd_process_scheduled_agent_removals(env, osd);
1838
1839         sb_end_write(osd_sb(osd));
1840
1841         RETURN(rc);
1842 }
1843
1844 static int osd_trans_cb_add(struct thandle *th, struct dt_txn_commit_cb *dcb)
1845 {
1846         struct osd_thandle *oh = container_of0(th, struct osd_thandle,
1847                                                ot_super);
1848
1849         LASSERT(dcb->dcb_magic == TRANS_COMMIT_CB_MAGIC);
1850         LASSERT(&dcb->dcb_func != NULL);
1851         if (dcb->dcb_flags & DCB_TRANS_STOP)
1852                 list_add(&dcb->dcb_linkage, &oh->ot_stop_dcb_list);
1853         else
1854                 list_add(&dcb->dcb_linkage, &oh->ot_commit_dcb_list);
1855
1856         return 0;
1857 }
1858
1859 /*
1860  * Called just before object is freed. Releases all resources except for
1861  * object itself (that is released by osd_object_free()).
1862  *
1863  * Concurrency: no concurrent access is possible that late in object
1864  * life-cycle.
1865  */
1866 static void osd_object_delete(const struct lu_env *env, struct lu_object *l)
1867 {
1868         struct osd_object *obj   = osd_obj(l);
1869         struct inode      *inode = obj->oo_inode;
1870
1871         LINVRNT(osd_invariant(obj));
1872
1873         /*
1874          * If object is unlinked remove fid->ino mapping from object index.
1875          */
1876
1877         osd_index_fini(obj);
1878         if (inode != NULL) {
1879                 struct qsd_instance     *qsd = osd_obj2dev(obj)->od_quota_slave;
1880                 qid_t                    uid = i_uid_read(inode);
1881                 qid_t                    gid = i_gid_read(inode);
1882
1883                 iput(inode);
1884                 obj->oo_inode = NULL;
1885
1886                 if (qsd != NULL) {
1887                         struct osd_thread_info  *info = osd_oti_get(env);
1888                         struct lquota_id_info   *qi = &info->oti_qi;
1889
1890                         /* Release granted quota to master if necessary */
1891                         qi->lqi_id.qid_uid = uid;
1892                         qsd_op_adjust(env, qsd, &qi->lqi_id, USRQUOTA);
1893
1894                         qi->lqi_id.qid_uid = gid;
1895                         qsd_op_adjust(env, qsd, &qi->lqi_id, GRPQUOTA);
1896                 }
1897         }
1898 }
1899
1900 /*
1901  * Concurrency: ->loo_object_release() is called under site spin-lock.
1902  */
1903 static void osd_object_release(const struct lu_env *env,
1904                                struct lu_object *l)
1905 {
1906         struct osd_object *o = osd_obj(l);
1907         /* nobody should be releasing a non-destroyed object with nlink=0
1908          * the API allows this, but ldiskfs doesn't like and then report
1909          * this inode as deleted */
1910         if (unlikely(!o->oo_destroyed && o->oo_inode && o->oo_inode->i_nlink == 0))
1911                 LBUG();
1912 }
1913
1914 /*
1915  * Concurrency: shouldn't matter.
1916  */
1917 static int osd_object_print(const struct lu_env *env, void *cookie,
1918                             lu_printer_t p, const struct lu_object *l)
1919 {
1920         struct osd_object *o = osd_obj(l);
1921         struct iam_descr  *d;
1922
1923         if (o->oo_dir != NULL)
1924                 d = o->oo_dir->od_container.ic_descr;
1925         else
1926                 d = NULL;
1927         return (*p)(env, cookie,
1928                     LUSTRE_OSD_LDISKFS_NAME"-object@%p(i:%p:%lu/%u)[%s]",
1929                     o, o->oo_inode,
1930                     o->oo_inode ? o->oo_inode->i_ino : 0UL,
1931                     o->oo_inode ? o->oo_inode->i_generation : 0,
1932                     d ? d->id_ops->id_name : "plain");
1933 }
1934
1935 /*
1936  * Concurrency: shouldn't matter.
1937  */
1938 int osd_statfs(const struct lu_env *env, struct dt_device *d,
1939                struct obd_statfs *sfs)
1940 {
1941         struct osd_device       *osd = osd_dt_dev(d);
1942         struct super_block      *sb = osd_sb(osd);
1943         struct kstatfs          *ksfs;
1944         __u64                    reserved;
1945         int                      result = 0;
1946
1947         if (unlikely(osd->od_mnt == NULL))
1948                 return -EINPROGRESS;
1949
1950         /* osd_lproc.c call this without env, allocate ksfs for that case */
1951         if (unlikely(env == NULL)) {
1952                 OBD_ALLOC_PTR(ksfs);
1953                 if (ksfs == NULL)
1954                         return -ENOMEM;
1955         } else {
1956                 ksfs = &osd_oti_get(env)->oti_ksfs;
1957         }
1958
1959         result = sb->s_op->statfs(sb->s_root, ksfs);
1960         if (result)
1961                 goto out;
1962
1963         statfs_pack(sfs, ksfs);
1964         if (unlikely(sb->s_flags & MS_RDONLY))
1965                 sfs->os_state = OS_STATE_READONLY;
1966         if (LDISKFS_HAS_INCOMPAT_FEATURE(sb,
1967                                          LDISKFS_FEATURE_INCOMPAT_EXTENTS))
1968                 sfs->os_maxbytes = sb->s_maxbytes;
1969         else
1970                 sfs->os_maxbytes = LDISKFS_SB(sb)->s_bitmap_maxbytes;
1971
1972         /*
1973          * Reserve some space so to avoid fragmenting the filesystem too much.
1974          * Fragmentation not only impacts performance, but can also increase
1975          * metadata overhead significantly, causing grant calculation to be
1976          * wrong.
1977          *
1978          * Reserve 0.78% of total space, at least 8MB for small filesystems.
1979          */
1980         CLASSERT(OSD_STATFS_RESERVED > LDISKFS_MAX_BLOCK_SIZE);
1981         reserved = OSD_STATFS_RESERVED >> sb->s_blocksize_bits;
1982         if (likely(sfs->os_blocks >= reserved << OSD_STATFS_RESERVED_SHIFT))
1983                 reserved = sfs->os_blocks >> OSD_STATFS_RESERVED_SHIFT;
1984
1985         sfs->os_blocks -= reserved;
1986         sfs->os_bfree  -= min(reserved, sfs->os_bfree);
1987         sfs->os_bavail -= min(reserved, sfs->os_bavail);
1988
1989 out:
1990         if (unlikely(env == NULL))
1991                 OBD_FREE_PTR(ksfs);
1992         return result;
1993 }
1994
1995 /**
1996  * Estimate space needed for file creations. We assume the largest filename
1997  * which is 2^64 - 1, hence a filename of 20 chars.
1998  * This is 28 bytes per object which is 28MB for 1M objects ... no so bad.
1999  */
2000 #ifdef __LDISKFS_DIR_REC_LEN
2001 #define PER_OBJ_USAGE __LDISKFS_DIR_REC_LEN(20)
2002 #else
2003 #define PER_OBJ_USAGE LDISKFS_DIR_REC_LEN(20)
2004 #endif
2005
2006 /*
2007  * Concurrency: doesn't access mutable data.
2008  */
2009 static void osd_conf_get(const struct lu_env *env,
2010                          const struct dt_device *dev,
2011                          struct dt_device_param *param)
2012 {
2013         struct super_block *sb = osd_sb(osd_dt_dev(dev));
2014         int                ea_overhead;
2015
2016         /*
2017          * XXX should be taken from not-yet-existing fs abstraction layer.
2018          */
2019         param->ddp_max_name_len = LDISKFS_NAME_LEN;
2020         param->ddp_max_nlink    = LDISKFS_LINK_MAX;
2021         param->ddp_symlink_max  = sb->s_blocksize;
2022         param->ddp_mount_type     = LDD_MT_LDISKFS;
2023         if (LDISKFS_HAS_INCOMPAT_FEATURE(sb, LDISKFS_FEATURE_INCOMPAT_EXTENTS))
2024                 param->ddp_maxbytes = sb->s_maxbytes;
2025         else
2026                 param->ddp_maxbytes = LDISKFS_SB(sb)->s_bitmap_maxbytes;
2027         /* inode are statically allocated, so per-inode space consumption
2028          * is the space consumed by the directory entry */
2029         param->ddp_inodespace     = PER_OBJ_USAGE;
2030         /* EXT_INIT_MAX_LEN is the theoretical maximum extent size  (32k blocks
2031          * = 128MB) which is unlikely to be hit in real life. Report a smaller
2032          * maximum length to not under count the actual number of extents
2033          * needed for writing a file. */
2034         param->ddp_max_extent_blks = EXT_INIT_MAX_LEN >> 2;
2035         /* worst-case extent insertion metadata overhead */
2036         param->ddp_extent_tax = 6 * LDISKFS_BLOCK_SIZE(sb);
2037         param->ddp_mntopts      = 0;
2038         if (test_opt(sb, XATTR_USER))
2039                 param->ddp_mntopts |= MNTOPT_USERXATTR;
2040         if (test_opt(sb, POSIX_ACL))
2041                 param->ddp_mntopts |= MNTOPT_ACL;
2042
2043         /* LOD might calculate the max stripe count based on max_ea_size,
2044          * so we need take account in the overhead as well,
2045          * xattr_header + magic + xattr_entry_head */
2046         ea_overhead = sizeof(struct ldiskfs_xattr_header) + sizeof(__u32) +
2047                       LDISKFS_XATTR_LEN(XATTR_NAME_MAX_LEN);
2048
2049 #if defined(LDISKFS_FEATURE_INCOMPAT_EA_INODE)
2050         if (LDISKFS_HAS_INCOMPAT_FEATURE(sb, LDISKFS_FEATURE_INCOMPAT_EA_INODE))
2051                 param->ddp_max_ea_size = LDISKFS_XATTR_MAX_LARGE_EA_SIZE -
2052                                                                 ea_overhead;
2053         else
2054 #endif
2055                 param->ddp_max_ea_size = sb->s_blocksize - ea_overhead;
2056 }
2057
2058 /*
2059  * Concurrency: shouldn't matter.
2060  */
2061 static int osd_sync(const struct lu_env *env, struct dt_device *d)
2062 {
2063         int rc;
2064
2065         CDEBUG(D_CACHE, "syncing OSD %s\n", LUSTRE_OSD_LDISKFS_NAME);
2066
2067         rc = ldiskfs_force_commit(osd_sb(osd_dt_dev(d)));
2068
2069         CDEBUG(D_CACHE, "synced OSD %s: rc = %d\n",
2070                LUSTRE_OSD_LDISKFS_NAME, rc);
2071
2072         return rc;
2073 }
2074
2075 /**
2076  * Start commit for OSD device.
2077  *
2078  * An implementation of dt_commit_async method for OSD device.
2079  * Asychronously starts underlayng fs sync and thereby a transaction
2080  * commit.
2081  *
2082  * \param env environment
2083  * \param d dt device
2084  *
2085  * \see dt_device_operations
2086  */
2087 static int osd_commit_async(const struct lu_env *env,
2088                             struct dt_device *d)
2089 {
2090         struct super_block *s = osd_sb(osd_dt_dev(d));
2091         ENTRY;
2092
2093         CDEBUG(D_HA, "async commit OSD %s\n", LUSTRE_OSD_LDISKFS_NAME);
2094         RETURN(s->s_op->sync_fs(s, 0));
2095 }
2096
2097 /*
2098  * Concurrency: shouldn't matter.
2099  */
2100
2101 static int osd_ro(const struct lu_env *env, struct dt_device *d)
2102 {
2103         struct super_block *sb = osd_sb(osd_dt_dev(d));
2104         struct block_device *dev = sb->s_bdev;
2105 #ifdef HAVE_DEV_SET_RDONLY
2106         struct block_device *jdev = LDISKFS_SB(sb)->journal_bdev;
2107         int rc = 0;
2108 #else
2109         int rc = -EOPNOTSUPP;
2110 #endif
2111         ENTRY;
2112
2113 #ifdef HAVE_DEV_SET_RDONLY
2114         CERROR("*** setting %s read-only ***\n", osd_dt_dev(d)->od_svname);
2115
2116         if (sb->s_op->freeze_fs) {
2117                 rc = sb->s_op->freeze_fs(sb);
2118                 if (rc)
2119                         goto out;
2120         }
2121
2122         if (jdev && (jdev != dev)) {
2123                 CDEBUG(D_IOCTL | D_HA, "set journal dev %lx rdonly\n",
2124                        (long)jdev);
2125                 dev_set_rdonly(jdev);
2126         }
2127         CDEBUG(D_IOCTL | D_HA, "set dev %lx rdonly\n", (long)dev);
2128         dev_set_rdonly(dev);
2129
2130         if (sb->s_op->unfreeze_fs)
2131                 sb->s_op->unfreeze_fs(sb);
2132
2133 out:
2134 #endif
2135         if (rc)
2136                 CERROR("%s: %lx CANNOT BE SET READONLY: rc = %d\n",
2137                        osd_dt_dev(d)->od_svname, (long)dev, rc);
2138
2139         RETURN(rc);
2140 }
2141
2142 /**
2143  * Note: we do not count into QUOTA here.
2144  * If we mount with --data_journal we may need more.
2145  */
2146 const int osd_dto_credits_noquota[DTO_NR] = {
2147         /**
2148          * Insert.
2149          * INDEX_EXTRA_TRANS_BLOCKS(8) +
2150          * SINGLEDATA_TRANS_BLOCKS(8)
2151          * XXX Note: maybe iam need more, since iam have more level than
2152          *           EXT3 htree.
2153          */
2154         [DTO_INDEX_INSERT]  = 16,
2155         /**
2156          * Delete
2157          * just modify a single entry, probably merge few within a block
2158          */
2159         [DTO_INDEX_DELETE]  = 1,
2160         /**
2161          * Used for OI scrub
2162          */
2163         [DTO_INDEX_UPDATE]  = 16,
2164         /**
2165          * 4(inode, inode bits, groups, GDT)
2166          *   notice: OI updates are counted separately with DTO_INDEX_INSERT
2167          */
2168         [DTO_OBJECT_CREATE] = 4,
2169         /**
2170          * 4(inode, inode bits, groups, GDT)
2171          *   notice: OI updates are counted separately with DTO_INDEX_DELETE
2172          */
2173         [DTO_OBJECT_DELETE] = 4,
2174         /**
2175          * Attr set credits (inode)
2176          */
2177         [DTO_ATTR_SET_BASE] = 1,
2178         /**
2179          * Xattr set. The same as xattr of EXT3.
2180          * DATA_TRANS_BLOCKS(14)
2181          * XXX Note: in original MDS implmentation INDEX_EXTRA_TRANS_BLOCKS
2182          * are also counted in. Do not know why?
2183          */
2184         [DTO_XATTR_SET]     = 14,
2185         /**
2186          * credits for inode change during write.
2187          */
2188         [DTO_WRITE_BASE]    = 3,
2189         /**
2190          * credits for single block write.
2191          */
2192         [DTO_WRITE_BLOCK]   = 14,
2193         /**
2194          * Attr set credits for chown.
2195          * This is extra credits for setattr, and it is null without quota
2196          */
2197         [DTO_ATTR_SET_CHOWN] = 0
2198 };
2199
2200 static const struct dt_device_operations osd_dt_ops = {
2201         .dt_root_get       = osd_root_get,
2202         .dt_statfs         = osd_statfs,
2203         .dt_trans_create   = osd_trans_create,
2204         .dt_trans_start    = osd_trans_start,
2205         .dt_trans_stop     = osd_trans_stop,
2206         .dt_trans_cb_add   = osd_trans_cb_add,
2207         .dt_conf_get       = osd_conf_get,
2208         .dt_sync           = osd_sync,
2209         .dt_ro             = osd_ro,
2210         .dt_commit_async   = osd_commit_async,
2211 };
2212
2213 static void osd_object_read_lock(const struct lu_env *env,
2214                                  struct dt_object *dt, unsigned role)
2215 {
2216         struct osd_object *obj = osd_dt_obj(dt);
2217         struct osd_thread_info *oti = osd_oti_get(env);
2218
2219         LINVRNT(osd_invariant(obj));
2220
2221         LASSERT(obj->oo_owner != env);
2222         down_read_nested(&obj->oo_sem, role);
2223
2224         LASSERT(obj->oo_owner == NULL);
2225         oti->oti_r_locks++;
2226 }
2227
2228 static void osd_object_write_lock(const struct lu_env *env,
2229                                   struct dt_object *dt, unsigned role)
2230 {
2231         struct osd_object *obj = osd_dt_obj(dt);
2232         struct osd_thread_info *oti = osd_oti_get(env);
2233
2234         LINVRNT(osd_invariant(obj));
2235
2236         LASSERT(obj->oo_owner != env);
2237         down_write_nested(&obj->oo_sem, role);
2238
2239         LASSERT(obj->oo_owner == NULL);
2240         obj->oo_owner = env;
2241         oti->oti_w_locks++;
2242 }
2243
2244 static void osd_object_read_unlock(const struct lu_env *env,
2245                                    struct dt_object *dt)
2246 {
2247         struct osd_object *obj = osd_dt_obj(dt);
2248         struct osd_thread_info *oti = osd_oti_get(env);
2249
2250         LINVRNT(osd_invariant(obj));
2251
2252         LASSERT(oti->oti_r_locks > 0);
2253         oti->oti_r_locks--;
2254         up_read(&obj->oo_sem);
2255 }
2256
2257 static void osd_object_write_unlock(const struct lu_env *env,
2258                                     struct dt_object *dt)
2259 {
2260         struct osd_object *obj = osd_dt_obj(dt);
2261         struct osd_thread_info *oti = osd_oti_get(env);
2262
2263         LINVRNT(osd_invariant(obj));
2264
2265         LASSERT(obj->oo_owner == env);
2266         LASSERT(oti->oti_w_locks > 0);
2267         oti->oti_w_locks--;
2268         obj->oo_owner = NULL;
2269         up_write(&obj->oo_sem);
2270 }
2271
2272 static int osd_object_write_locked(const struct lu_env *env,
2273                                    struct dt_object *dt)
2274 {
2275         struct osd_object *obj = osd_dt_obj(dt);
2276
2277         LINVRNT(osd_invariant(obj));
2278
2279         return obj->oo_owner == env;
2280 }
2281
2282 static struct timespec *osd_inode_time(const struct lu_env *env,
2283                                        struct inode *inode, __u64 seconds)
2284 {
2285         struct osd_thread_info  *oti = osd_oti_get(env);
2286         struct timespec         *t   = &oti->oti_time;
2287
2288         t->tv_sec = seconds;
2289         t->tv_nsec = 0;
2290         *t = timespec_trunc(*t, inode->i_sb->s_time_gran);
2291         return t;
2292 }
2293
2294 static void osd_inode_getattr(const struct lu_env *env,
2295                               struct inode *inode, struct lu_attr *attr)
2296 {
2297         attr->la_valid  |= LA_ATIME | LA_MTIME | LA_CTIME | LA_MODE |
2298                            LA_SIZE | LA_BLOCKS | LA_UID | LA_GID |
2299                            LA_FLAGS | LA_NLINK | LA_RDEV | LA_BLKSIZE |
2300                            LA_TYPE;
2301
2302         attr->la_atime   = LTIME_S(inode->i_atime);
2303         attr->la_mtime   = LTIME_S(inode->i_mtime);
2304         attr->la_ctime   = LTIME_S(inode->i_ctime);
2305         attr->la_mode    = inode->i_mode;
2306         attr->la_size    = i_size_read(inode);
2307         attr->la_blocks  = inode->i_blocks;
2308         attr->la_uid     = i_uid_read(inode);
2309         attr->la_gid     = i_gid_read(inode);
2310         attr->la_flags   = ll_inode_to_ext_flags(inode->i_flags);
2311         attr->la_nlink   = inode->i_nlink;
2312         attr->la_rdev    = inode->i_rdev;
2313         attr->la_blksize = 1 << inode->i_blkbits;
2314         attr->la_blkbits = inode->i_blkbits;
2315 }
2316
2317 static int osd_attr_get(const struct lu_env *env,
2318                         struct dt_object *dt,
2319                         struct lu_attr *attr)
2320 {
2321         struct osd_object *obj = osd_dt_obj(dt);
2322
2323         if (unlikely(!dt_object_exists(dt)))
2324                 return -ENOENT;
2325         if (unlikely(obj->oo_destroyed))
2326                 return -ENOENT;
2327
2328         LASSERT(!dt_object_remote(dt));
2329         LINVRNT(osd_invariant(obj));
2330
2331         spin_lock(&obj->oo_guard);
2332         osd_inode_getattr(env, obj->oo_inode, attr);
2333         if (obj->oo_lma_flags & LUSTRE_ORPHAN_FL)
2334                 attr->la_flags |= LUSTRE_ORPHAN_FL;
2335         spin_unlock(&obj->oo_guard);
2336
2337         return 0;
2338 }
2339
2340 static int osd_declare_attr_set(const struct lu_env *env,
2341                                 struct dt_object *dt,
2342                                 const struct lu_attr *attr,
2343                                 struct thandle *handle)
2344 {
2345         struct osd_thandle     *oh;
2346         struct osd_object      *obj;
2347         struct osd_thread_info *info = osd_oti_get(env);
2348         struct lquota_id_info  *qi = &info->oti_qi;
2349         qid_t                   uid;
2350         qid_t                   gid;
2351         long long               bspace;
2352         int                     rc = 0;
2353         bool                    enforce;
2354         ENTRY;
2355
2356         LASSERT(dt != NULL);
2357         LASSERT(handle != NULL);
2358
2359         obj = osd_dt_obj(dt);
2360         LASSERT(osd_invariant(obj));
2361
2362         oh = container_of0(handle, struct osd_thandle, ot_super);
2363         LASSERT(oh->ot_handle == NULL);
2364
2365         osd_trans_declare_op(env, oh, OSD_OT_ATTR_SET,
2366                              osd_dto_credits_noquota[DTO_ATTR_SET_BASE]);
2367
2368         osd_trans_declare_op(env, oh, OSD_OT_XATTR_SET,
2369                              osd_dto_credits_noquota[DTO_XATTR_SET]);
2370
2371         if (attr == NULL || obj->oo_inode == NULL)
2372                 RETURN(rc);
2373
2374         bspace   = obj->oo_inode->i_blocks;
2375         bspace <<= obj->oo_inode->i_sb->s_blocksize_bits;
2376         bspace   = toqb(bspace);
2377
2378         /* Changing ownership is always preformed by super user, it should not
2379          * fail with EDQUOT.
2380          *
2381          * We still need to call the osd_declare_qid() to calculate the journal
2382          * credits for updating quota accounting files and to trigger quota
2383          * space adjustment once the operation is completed.*/
2384         if (attr->la_valid & LA_UID || attr->la_valid & LA_GID) {
2385                 /* USERQUOTA */
2386                 uid = i_uid_read(obj->oo_inode);
2387                 qi->lqi_type = USRQUOTA;
2388                 enforce = (attr->la_valid & LA_UID) && (attr->la_uid != uid);
2389                 /* inode accounting */
2390                 qi->lqi_is_blk = false;
2391
2392                 /* one more inode for the new uid ... */
2393                 qi->lqi_id.qid_uid = attr->la_uid;
2394                 qi->lqi_space      = 1;
2395                 /* Reserve credits for the new uid */
2396                 rc = osd_declare_qid(env, oh, qi, NULL, enforce, NULL);
2397                 if (rc == -EDQUOT || rc == -EINPROGRESS)
2398                         rc = 0;
2399                 if (rc)
2400                         RETURN(rc);
2401
2402                 /* and one less inode for the current uid */
2403                 qi->lqi_id.qid_uid = uid;
2404                 qi->lqi_space      = -1;
2405                 rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
2406                 if (rc == -EDQUOT || rc == -EINPROGRESS)
2407                         rc = 0;
2408                 if (rc)
2409                         RETURN(rc);
2410
2411                 /* block accounting */
2412                 qi->lqi_is_blk = true;
2413
2414                 /* more blocks for the new uid ... */
2415                 qi->lqi_id.qid_uid = attr->la_uid;
2416                 qi->lqi_space      = bspace;
2417                 /*
2418                  * Credits for the new uid has been reserved, re-use "obj"
2419                  * to save credit reservation.
2420                  */
2421                 rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
2422                 if (rc == -EDQUOT || rc == -EINPROGRESS)
2423                         rc = 0;
2424                 if (rc)
2425                         RETURN(rc);
2426
2427                 /* and finally less blocks for the current uid */
2428                 qi->lqi_id.qid_uid = uid;
2429                 qi->lqi_space      = -bspace;
2430                 rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
2431                 if (rc == -EDQUOT || rc == -EINPROGRESS)
2432                         rc = 0;
2433                 if (rc)
2434                         RETURN(rc);
2435
2436                 /* GROUP QUOTA */
2437                 gid = i_gid_read(obj->oo_inode);
2438                 qi->lqi_type = GRPQUOTA;
2439                 enforce = (attr->la_valid & LA_GID) && (attr->la_gid != gid);
2440
2441                 /* inode accounting */
2442                 qi->lqi_is_blk = false;
2443
2444                 /* one more inode for the new gid ... */
2445                 qi->lqi_id.qid_gid = attr->la_gid;
2446                 qi->lqi_space      = 1;
2447                 rc = osd_declare_qid(env, oh, qi, NULL, enforce, NULL);
2448                 if (rc == -EDQUOT || rc == -EINPROGRESS)
2449                         rc = 0;
2450                 if (rc)
2451                         RETURN(rc);
2452
2453                 /* and one less inode for the current gid */
2454                 qi->lqi_id.qid_gid = gid;
2455                 qi->lqi_space      = -1;
2456                 rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
2457                 if (rc == -EDQUOT || rc == -EINPROGRESS)
2458                         rc = 0;
2459                 if (rc)
2460                         RETURN(rc);
2461
2462                 /* block accounting */
2463                 qi->lqi_is_blk = true;
2464
2465                 /* more blocks for the new gid ... */
2466                 qi->lqi_id.qid_gid = attr->la_gid;
2467                 qi->lqi_space      = bspace;
2468                 rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
2469                 if (rc == -EDQUOT || rc == -EINPROGRESS)
2470                         rc = 0;
2471                 if (rc)
2472                         RETURN(rc);
2473
2474                 /* and finally less blocks for the current gid */
2475                 qi->lqi_id.qid_gid = gid;
2476                 qi->lqi_space      = -bspace;
2477                 rc = osd_declare_qid(env, oh, qi, obj, enforce, NULL);
2478                 if (rc == -EDQUOT || rc == -EINPROGRESS)
2479                         rc = 0;
2480                 if (rc)
2481                         RETURN(rc);
2482         }
2483
2484         RETURN(rc);
2485 }
2486
2487 static int osd_inode_setattr(const struct lu_env *env,
2488                              struct inode *inode, const struct lu_attr *attr)
2489 {
2490         __u64 bits = attr->la_valid;
2491
2492         /* Only allow set size for regular file */
2493         if (!S_ISREG(inode->i_mode))
2494                 bits &= ~(LA_SIZE | LA_BLOCKS);
2495
2496         if (bits == 0)
2497                 return 0;
2498
2499         if (bits & LA_ATIME)
2500                 inode->i_atime  = *osd_inode_time(env, inode, attr->la_atime);
2501         if (bits & LA_CTIME)
2502                 inode->i_ctime  = *osd_inode_time(env, inode, attr->la_ctime);
2503         if (bits & LA_MTIME)
2504                 inode->i_mtime  = *osd_inode_time(env, inode, attr->la_mtime);
2505         if (bits & LA_SIZE) {
2506                 LDISKFS_I(inode)->i_disksize = attr->la_size;
2507                 i_size_write(inode, attr->la_size);
2508         }
2509
2510         /* OSD should not change "i_blocks" which is used by quota.
2511          * "i_blocks" should be changed by ldiskfs only. */
2512         if (bits & LA_MODE)
2513                 inode->i_mode = (inode->i_mode & S_IFMT) |
2514                                 (attr->la_mode & ~S_IFMT);
2515         if (bits & LA_UID)
2516                 i_uid_write(inode, attr->la_uid);
2517         if (bits & LA_GID)
2518                 i_gid_write(inode, attr->la_gid);
2519         if (bits & LA_NLINK)
2520                 set_nlink(inode, attr->la_nlink);
2521         if (bits & LA_RDEV)
2522                 inode->i_rdev = attr->la_rdev;
2523
2524         if (bits & LA_FLAGS) {
2525                 /* always keep S_NOCMTIME */
2526                 inode->i_flags = ll_ext_to_inode_flags(attr->la_flags) |
2527                                  S_NOCMTIME;
2528         }
2529         return 0;
2530 }
2531
2532 static int osd_quota_transfer(struct inode *inode, const struct lu_attr *attr)
2533 {
2534         if ((attr->la_valid & LA_UID && attr->la_uid != i_uid_read(inode)) ||
2535             (attr->la_valid & LA_GID && attr->la_gid != i_gid_read(inode))) {
2536                 struct iattr    iattr;
2537                 int             rc;
2538
2539                 ll_vfs_dq_init(inode);
2540                 iattr.ia_valid = 0;
2541                 if (attr->la_valid & LA_UID)
2542                         iattr.ia_valid |= ATTR_UID;
2543                 if (attr->la_valid & LA_GID)
2544                         iattr.ia_valid |= ATTR_GID;
2545                 iattr.ia_uid = make_kuid(&init_user_ns, attr->la_uid);
2546                 iattr.ia_gid = make_kgid(&init_user_ns, attr->la_gid);
2547
2548                 rc = ll_vfs_dq_transfer(inode, &iattr);
2549                 if (rc) {
2550                         CERROR("%s: quota transfer failed: rc = %d. Is quota "
2551                                "enforcement enabled on the ldiskfs "
2552                                "filesystem?\n", inode->i_sb->s_id, rc);
2553                         return rc;
2554                 }
2555         }
2556         return 0;
2557 }
2558
2559 static int osd_attr_set(const struct lu_env *env,
2560                         struct dt_object *dt,
2561                         const struct lu_attr *attr,
2562                         struct thandle *handle)
2563 {
2564         struct osd_object *obj = osd_dt_obj(dt);
2565         struct inode      *inode;
2566         int rc;
2567
2568         if (!dt_object_exists(dt))
2569                 return -ENOENT;
2570
2571         LASSERT(handle != NULL);
2572         LASSERT(!dt_object_remote(dt));
2573         LASSERT(osd_invariant(obj));
2574
2575         osd_trans_exec_op(env, handle, OSD_OT_ATTR_SET);
2576
2577         if (OBD_FAIL_CHECK(OBD_FAIL_OSD_FID_MAPPING)) {
2578                 struct osd_thread_info  *oti  = osd_oti_get(env);
2579                 const struct lu_fid     *fid0 = lu_object_fid(&dt->do_lu);
2580                 struct lu_fid           *fid1 = &oti->oti_fid;
2581                 struct osd_inode_id     *id   = &oti->oti_id;
2582                 struct iam_path_descr   *ipd;
2583                 struct iam_container    *bag;
2584                 struct osd_thandle      *oh;
2585                 int                      rc;
2586
2587                 fid_cpu_to_be(fid1, fid0);
2588                 memset(id, 1, sizeof(*id));
2589                 bag = &osd_fid2oi(osd_dev(dt->do_lu.lo_dev),
2590                                   fid0)->oi_dir.od_container;
2591                 ipd = osd_idx_ipd_get(env, bag);
2592                 if (unlikely(ipd == NULL))
2593                         RETURN(-ENOMEM);
2594
2595                 oh = container_of0(handle, struct osd_thandle, ot_super);
2596                 rc = iam_update(oh->ot_handle, bag, (const struct iam_key *)fid1,
2597                                 (const struct iam_rec *)id, ipd);
2598                 osd_ipd_put(env, bag, ipd);
2599                 return(rc > 0 ? 0 : rc);
2600         }
2601
2602         inode = obj->oo_inode;
2603
2604         rc = osd_quota_transfer(inode, attr);
2605         if (rc)
2606                 return rc;
2607
2608         spin_lock(&obj->oo_guard);
2609         rc = osd_inode_setattr(env, inode, attr);
2610         spin_unlock(&obj->oo_guard);
2611         if (rc != 0)
2612                 GOTO(out, rc);
2613
2614         ll_dirty_inode(inode, I_DIRTY_DATASYNC);
2615
2616         if (!(attr->la_valid & LA_FLAGS))
2617                 GOTO(out, rc);
2618
2619         /* Let's check if there are extra flags need to be set into LMA */
2620         if (attr->la_flags & LUSTRE_LMA_FL_MASKS) {
2621                 struct osd_thread_info *info = osd_oti_get(env);
2622                 struct lustre_mdt_attrs *lma = &info->oti_mdt_attrs;
2623
2624                 rc = osd_get_lma(info, inode, &info->oti_obj_dentry, lma);
2625                 if (rc != 0)
2626                         GOTO(out, rc);
2627
2628                 lma->lma_incompat |=
2629                         lustre_to_lma_flags(attr->la_flags);
2630                 lustre_lma_swab(lma);
2631                 rc = __osd_xattr_set(info, inode, XATTR_NAME_LMA,
2632                                      lma, sizeof(*lma), XATTR_REPLACE);
2633                 if (rc != 0) {
2634                         struct osd_device *osd = osd_obj2dev(obj);
2635
2636                         CWARN("%s: set "DFID" lma flags %u failed: rc = %d\n",
2637                               osd_name(osd), PFID(lu_object_fid(&dt->do_lu)),
2638                               lma->lma_incompat, rc);
2639                 } else {
2640                         obj->oo_lma_flags =
2641                                 attr->la_flags & LUSTRE_LMA_FL_MASKS;
2642                 }
2643                 osd_trans_exec_check(env, handle, OSD_OT_XATTR_SET);
2644         }
2645 out:
2646         osd_trans_exec_check(env, handle, OSD_OT_ATTR_SET);
2647
2648         return rc;
2649 }
2650
2651 static struct dentry *osd_child_dentry_get(const struct lu_env *env,
2652                                            struct osd_object *obj,
2653                                            const char *name, const int namelen)
2654 {
2655         return osd_child_dentry_by_inode(env, obj->oo_inode, name, namelen);
2656 }
2657
2658 static int osd_mkfile(struct osd_thread_info *info, struct osd_object *obj,
2659                       umode_t mode, struct dt_allocation_hint *hint,
2660                       struct thandle *th)
2661 {
2662         int result;
2663         struct osd_device  *osd = osd_obj2dev(obj);
2664         struct osd_thandle *oth;
2665         struct dt_object   *parent = NULL;
2666         struct inode       *inode;
2667
2668         LINVRNT(osd_invariant(obj));
2669         LASSERT(obj->oo_inode == NULL);
2670         LASSERT(obj->oo_hl_head == NULL);
2671
2672         if (S_ISDIR(mode) && ldiskfs_pdo) {
2673                 obj->oo_hl_head =ldiskfs_htree_lock_head_alloc(HTREE_HBITS_DEF);
2674                 if (obj->oo_hl_head == NULL)
2675                         return -ENOMEM;
2676         }
2677
2678         oth = container_of(th, struct osd_thandle, ot_super);
2679         LASSERT(oth->ot_handle->h_transaction != NULL);
2680
2681         if (hint != NULL && hint->dah_parent != NULL &&
2682             !dt_object_remote(hint->dah_parent))
2683                 parent = hint->dah_parent;
2684
2685         inode = ldiskfs_create_inode(oth->ot_handle,
2686                                      parent ? osd_dt_obj(parent)->oo_inode :
2687                                               osd_sb(osd)->s_root->d_inode,
2688                                      mode);
2689         if (!IS_ERR(inode)) {
2690                 /* Do not update file c/mtime in ldiskfs. */
2691                 inode->i_flags |= S_NOCMTIME;
2692
2693                 /* For new created object, it must be consistent,
2694                  * and it is unnecessary to scrub against it. */
2695                 ldiskfs_set_inode_state(inode, LDISKFS_STATE_LUSTRE_NOSCRUB);
2696
2697                 obj->oo_inode = inode;
2698                 result = 0;
2699         } else {
2700                 if (obj->oo_hl_head != NULL) {
2701                         ldiskfs_htree_lock_head_free(obj->oo_hl_head);
2702                         obj->oo_hl_head = NULL;
2703                 }
2704                 result = PTR_ERR(inode);
2705         }
2706         LINVRNT(osd_invariant(obj));
2707         return result;
2708 }
2709
2710 enum {
2711         OSD_NAME_LEN = 255
2712 };
2713
2714 static int osd_mkdir(struct osd_thread_info *info, struct osd_object *obj,
2715                      struct lu_attr *attr,
2716                      struct dt_allocation_hint *hint,
2717                      struct dt_object_format *dof,
2718                      struct thandle *th)
2719 {
2720         int result;
2721         struct osd_thandle *oth;
2722         __u32 mode = (attr->la_mode & (S_IFMT | S_IRWXUGO | S_ISVTX));
2723
2724         LASSERT(S_ISDIR(attr->la_mode));
2725
2726         oth = container_of(th, struct osd_thandle, ot_super);
2727         LASSERT(oth->ot_handle->h_transaction != NULL);
2728         result = osd_mkfile(info, obj, mode, hint, th);
2729
2730         return result;
2731 }
2732
2733 static int osd_mk_index(struct osd_thread_info *info, struct osd_object *obj,
2734                         struct lu_attr *attr,
2735                         struct dt_allocation_hint *hint,
2736                         struct dt_object_format *dof,
2737                         struct thandle *th)
2738 {
2739         int result;
2740         struct osd_thandle *oth;
2741         const struct dt_index_features *feat = dof->u.dof_idx.di_feat;
2742
2743         __u32 mode = (attr->la_mode & (S_IFMT | S_IALLUGO | S_ISVTX));
2744
2745         LASSERT(S_ISREG(attr->la_mode));
2746
2747         oth = container_of(th, struct osd_thandle, ot_super);
2748         LASSERT(oth->ot_handle->h_transaction != NULL);
2749
2750         result = osd_mkfile(info, obj, mode, hint, th);
2751         if (result == 0) {
2752                 LASSERT(obj->oo_inode != NULL);
2753                 if (feat->dif_flags & DT_IND_VARKEY)
2754                         result = iam_lvar_create(obj->oo_inode,
2755                                                  feat->dif_keysize_max,
2756                                                  feat->dif_ptrsize,
2757                                                  feat->dif_recsize_max,
2758                                                  oth->ot_handle);
2759                 else
2760                         result = iam_lfix_create(obj->oo_inode,
2761                                                  feat->dif_keysize_max,
2762                                                  feat->dif_ptrsize,
2763                                                  feat->dif_recsize_max,
2764                                                  oth->ot_handle);
2765
2766         }
2767         return result;
2768 }
2769
2770 static int osd_mkreg(struct osd_thread_info *info, struct osd_object *obj,
2771                      struct lu_attr *attr,
2772                      struct dt_allocation_hint *hint,
2773                      struct dt_object_format *dof,
2774                      struct thandle *th)
2775 {
2776         LASSERT(S_ISREG(attr->la_mode));
2777         return osd_mkfile(info, obj, (attr->la_mode &
2778                                (S_IFMT | S_IALLUGO | S_ISVTX)), hint, th);
2779 }
2780
2781 static int osd_mksym(struct osd_thread_info *info, struct osd_object *obj,
2782                      struct lu_attr *attr,
2783                      struct dt_allocation_hint *hint,
2784                      struct dt_object_format *dof,
2785                      struct thandle *th)
2786 {
2787         LASSERT(S_ISLNK(attr->la_mode));
2788         return osd_mkfile(info, obj, (attr->la_mode &
2789                               (S_IFMT | S_IALLUGO | S_ISVTX)), hint, th);
2790 }
2791
2792 static int osd_mknod(struct osd_thread_info *info, struct osd_object *obj,
2793                      struct lu_attr *attr,
2794                      struct dt_allocation_hint *hint,
2795                      struct dt_object_format *dof,
2796                      struct thandle *th)
2797 {
2798         umode_t mode = attr->la_mode & (S_IFMT | S_IALLUGO | S_ISVTX);
2799         int result;
2800
2801         LINVRNT(osd_invariant(obj));
2802         LASSERT(obj->oo_inode == NULL);
2803         LASSERT(S_ISCHR(mode) || S_ISBLK(mode) ||
2804                 S_ISFIFO(mode) || S_ISSOCK(mode));
2805
2806         result = osd_mkfile(info, obj, mode, hint, th);
2807         if (result == 0) {
2808                 LASSERT(obj->oo_inode != NULL);
2809                 /*
2810                  * This inode should be marked dirty for i_rdev.  Currently
2811                  * that is done in the osd_attr_init().
2812                  */
2813                 init_special_inode(obj->oo_inode, obj->oo_inode->i_mode,
2814                                    attr->la_rdev);
2815         }
2816         LINVRNT(osd_invariant(obj));
2817         return result;
2818 }
2819
2820 typedef int (*osd_obj_type_f)(struct osd_thread_info *, struct osd_object *,
2821                               struct lu_attr *,
2822                               struct dt_allocation_hint *hint,
2823                               struct dt_object_format *dof,
2824                               struct thandle *);
2825
2826 static osd_obj_type_f osd_create_type_f(enum dt_format_type type)
2827 {
2828         osd_obj_type_f result;
2829
2830         switch (type) {
2831         case DFT_DIR:
2832                 result = osd_mkdir;
2833                 break;
2834         case DFT_REGULAR:
2835                 result = osd_mkreg;
2836                 break;
2837         case DFT_SYM:
2838                 result = osd_mksym;
2839                 break;
2840         case DFT_NODE:
2841                 result = osd_mknod;
2842                 break;
2843         case DFT_INDEX:
2844                 result = osd_mk_index;
2845                 break;
2846
2847         default:
2848                 LBUG();
2849                 break;
2850         }
2851         return result;
2852 }
2853
2854
2855 static void osd_ah_init(const struct lu_env *env, struct dt_allocation_hint *ah,
2856                         struct dt_object *parent, struct dt_object *child,
2857                         umode_t child_mode)
2858 {
2859         LASSERT(ah);
2860
2861         ah->dah_parent = parent;
2862         ah->dah_mode = child_mode;
2863
2864         if (parent != NULL && !dt_object_remote(parent)) {
2865                 /* will help to find FID->ino at dt_insert("..") */
2866                 struct osd_object *pobj = osd_dt_obj(parent);
2867                 osd_idc_find_and_init(env, osd_obj2dev(pobj), pobj);
2868         }
2869 }
2870
2871 static void osd_attr_init(struct osd_thread_info *info, struct osd_object *obj,
2872                           struct lu_attr *attr, struct dt_object_format *dof)
2873 {
2874         struct inode   *inode = obj->oo_inode;
2875         __u64           valid = attr->la_valid;
2876         int             result;
2877
2878         attr->la_valid &= ~(LA_TYPE | LA_MODE);
2879
2880         if (dof->dof_type != DFT_NODE)
2881                 attr->la_valid &= ~LA_RDEV;
2882         if ((valid & LA_ATIME) && (attr->la_atime == LTIME_S(inode->i_atime)))
2883                 attr->la_valid &= ~LA_ATIME;
2884         if ((valid & LA_CTIME) && (attr->la_ctime == LTIME_S(inode->i_ctime)))
2885                 attr->la_valid &= ~LA_CTIME;
2886         if ((valid & LA_MTIME) && (attr->la_mtime == LTIME_S(inode->i_mtime)))
2887                 attr->la_valid &= ~LA_MTIME;
2888
2889         result = osd_quota_transfer(inode, attr);
2890         if (result)
2891                 return;
2892
2893         if (attr->la_valid != 0) {
2894                 result = osd_inode_setattr(info->oti_env, inode, attr);
2895                 /*
2896                  * The osd_inode_setattr() should always succeed here.  The
2897                  * only error that could be returned is EDQUOT when we are
2898                  * trying to change the UID or GID of the inode. However, this
2899                  * should not happen since quota enforcement is no longer
2900                  * enabled on ldiskfs (lquota takes care of it).
2901                  */
2902                 LASSERTF(result == 0, "%d\n", result);
2903                 ll_dirty_inode(inode, I_DIRTY_DATASYNC);
2904         }
2905
2906         attr->la_valid = valid;
2907 }
2908
2909 /**
2910  * Helper function for osd_object_create()
2911  *
2912  * \retval 0, on success
2913  */
2914 static int __osd_object_create(struct osd_thread_info *info,
2915                                struct osd_object *obj, struct lu_attr *attr,
2916                                struct dt_allocation_hint *hint,
2917                                struct dt_object_format *dof,
2918                                struct thandle *th)
2919 {
2920         int     result;
2921         __u32   umask;
2922
2923         osd_trans_exec_op(info->oti_env, th, OSD_OT_CREATE);
2924
2925         /* we drop umask so that permissions we pass are not affected */
2926         umask = current->fs->umask;
2927         current->fs->umask = 0;
2928
2929         result = osd_create_type_f(dof->dof_type)(info, obj, attr, hint, dof,
2930                                                   th);
2931         if (likely(obj->oo_inode != NULL)) {
2932                 LASSERT(obj->oo_inode->i_state & I_NEW);
2933
2934                 /* Unlock the inode before attr initialization to avoid
2935                  * unnecessary dqget operations. LU-6378 */
2936                 unlock_new_inode(obj->oo_inode);
2937         }
2938
2939         if (likely(result == 0)) {
2940                 osd_attr_init(info, obj, attr, dof);
2941                 osd_object_init0(obj);
2942         }
2943
2944         /* restore previous umask value */
2945         current->fs->umask = umask;
2946
2947         osd_trans_exec_check(info->oti_env, th, OSD_OT_CREATE);
2948
2949         return result;
2950 }
2951
2952 /**
2953  * Helper function for osd_object_create()
2954  *
2955  * \retval 0, on success
2956  */
2957 static int __osd_oi_insert(const struct lu_env *env, struct osd_object *obj,
2958                            const struct lu_fid *fid, struct thandle *th)
2959 {
2960         struct osd_thread_info *info = osd_oti_get(env);
2961         struct osd_inode_id    *id   = &info->oti_id;
2962         struct osd_device      *osd  = osd_obj2dev(obj);
2963         struct osd_thandle     *oh;
2964         int                     rc;
2965
2966         LASSERT(obj->oo_inode != NULL);
2967
2968         oh = container_of0(th, struct osd_thandle, ot_super);
2969         LASSERT(oh->ot_handle);
2970         osd_trans_exec_op(env, th, OSD_OT_INSERT);
2971
2972         osd_id_gen(id, obj->oo_inode->i_ino, obj->oo_inode->i_generation);
2973         rc = osd_oi_insert(info, osd, fid, id, oh->ot_handle,
2974                            OI_CHECK_FLD, NULL);
2975         osd_trans_exec_check(env, th, OSD_OT_INSERT);
2976
2977         return rc;
2978 }
2979
2980 int osd_fld_lookup(const struct lu_env *env, struct osd_device *osd,
2981                    u64 seq, struct lu_seq_range *range)
2982 {
2983         struct seq_server_site  *ss = osd_seq_site(osd);
2984
2985         if (fid_seq_is_idif(seq)) {
2986                 fld_range_set_ost(range);
2987                 range->lsr_index = idif_ost_idx(seq);
2988                 return 0;
2989         }
2990
2991         if (!fid_seq_in_fldb(seq)) {
2992                 fld_range_set_mdt(range);
2993                 if (ss != NULL)
2994                         /* FIXME: If ss is NULL, it suppose not get lsr_index
2995                          * at all */
2996                         range->lsr_index = ss->ss_node_id;
2997                 return 0;
2998         }
2999
3000         LASSERT(ss != NULL);
3001         fld_range_set_any(range);
3002         /* OSD will only do local fld lookup */
3003         return fld_local_lookup(env, ss->ss_server_fld, seq, range);
3004 }
3005
3006 /*
3007  * Concurrency: no external locking is necessary.
3008  */
3009 static int osd_declare_object_create(const struct lu_env *env,
3010                                      struct dt_object *dt,
3011                                      struct lu_attr *attr,
3012                                      struct dt_allocation_hint *hint,
3013                                      struct dt_object_format *dof,
3014                                      struct thandle *handle)
3015 {
3016         struct osd_thandle      *oh;
3017         int                      rc;
3018         ENTRY;
3019
3020         LASSERT(handle != NULL);
3021
3022         oh = container_of0(handle, struct osd_thandle, ot_super);
3023         LASSERT(oh->ot_handle == NULL);
3024
3025         /* EA object consumes more credits than regular object: osd_mk_index
3026          * vs. osd_mkreg: osd_mk_index will create 2 blocks for root_node and
3027          * leaf_node, could involves the block, block bitmap, groups, GDT
3028          * change for each block, so add 4 * 2 credits in that case. */
3029         osd_trans_declare_op(env, oh, OSD_OT_CREATE,
3030                              osd_dto_credits_noquota[DTO_OBJECT_CREATE] +
3031                              (dof->dof_type == DFT_INDEX) ? 4 * 2 : 0);
3032         /* Reuse idle OI block may cause additional one OI block
3033          * to be changed. */
3034         osd_trans_declare_op(env, oh, OSD_OT_INSERT,
3035                              osd_dto_credits_noquota[DTO_INDEX_INSERT] + 1);
3036
3037         if (!attr)
3038                 RETURN(0);
3039
3040         rc = osd_declare_inode_qid(env, attr->la_uid, attr->la_gid, 1, oh,
3041                                    osd_dt_obj(dt), false, NULL, false);
3042         if (rc != 0)
3043                 RETURN(rc);
3044
3045         /* will help to find FID->ino mapping at dt_insert() */
3046         rc = osd_idc_find_and_init(env, osd_obj2dev(osd_dt_obj(dt)),
3047                                    osd_dt_obj(dt));
3048
3049         RETURN(rc);
3050 }
3051
3052 static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
3053                              struct lu_attr *attr,
3054                              struct dt_allocation_hint *hint,
3055                              struct dt_object_format *dof, struct thandle *th)
3056 {
3057         const struct lu_fid     *fid    = lu_object_fid(&dt->do_lu);
3058         struct osd_object       *obj    = osd_dt_obj(dt);
3059         struct osd_thread_info  *info   = osd_oti_get(env);
3060         int result;
3061         ENTRY;
3062
3063         if (dt_object_exists(dt))
3064                 return -EEXIST;
3065
3066         LINVRNT(osd_invariant(obj));
3067         LASSERT(!dt_object_remote(dt));
3068         LASSERT(osd_write_locked(env, obj));
3069         LASSERT(th != NULL);
3070
3071         if (unlikely(fid_is_acct(fid)))
3072                 /* Quota files can't be created from the kernel any more,
3073                  * 'tune2fs -O quota' will take care of creating them */
3074                 RETURN(-EPERM);
3075
3076         result = __osd_object_create(info, obj, attr, hint, dof, th);
3077         if (result == 0) {
3078                 result = __osd_oi_insert(env, obj, fid, th);
3079                 if (obj->oo_dt.do_body_ops == &osd_body_ops_new)
3080                         obj->oo_dt.do_body_ops = &osd_body_ops;
3081         }
3082         LASSERT(ergo(result == 0,
3083                 dt_object_exists(dt) && !dt_object_remote(dt)));
3084
3085         LASSERT(osd_invariant(obj));
3086         RETURN(result);
3087 }
3088
3089 /**
3090  * Called to destroy on-disk representation of the object
3091  *
3092  * Concurrency: must be locked
3093  */
3094 static int osd_declare_object_destroy(const struct lu_env *env,
3095                                       struct dt_object *dt,
3096                                       struct thandle *th)
3097 {
3098         struct osd_object  *obj = osd_dt_obj(dt);
3099         struct inode       *inode = obj->oo_inode;
3100         struct osd_thandle *oh;
3101         int                 rc;
3102         ENTRY;
3103
3104         if (inode == NULL)
3105                 RETURN(-ENOENT);
3106
3107         oh = container_of0(th, struct osd_thandle, ot_super);
3108         LASSERT(oh->ot_handle == NULL);
3109
3110         osd_trans_declare_op(env, oh, OSD_OT_DESTROY,
3111                              osd_dto_credits_noquota[DTO_OBJECT_DELETE]);
3112         /* Recycle idle OI leaf may cause additional three OI blocks
3113          * to be changed. */
3114         osd_trans_declare_op(env, oh, OSD_OT_DELETE,
3115                              osd_dto_credits_noquota[DTO_INDEX_DELETE] + 3);
3116         /* one less inode */
3117         rc = osd_declare_inode_qid(env, i_uid_read(inode), i_gid_read(inode),
3118                                    -1, oh, obj, false, NULL, false);
3119         if (rc)
3120                 RETURN(rc);
3121         /* data to be truncated */
3122         rc = osd_declare_inode_qid(env, i_uid_read(inode), i_gid_read(inode),
3123                                    0, oh, obj, true, NULL, false);
3124         if (rc)
3125                 RETURN(rc);
3126
3127         /* will help to find FID->ino when this object is being
3128          * added to PENDING/ */
3129         rc = osd_idc_find_and_init(env, osd_obj2dev(obj), obj);
3130
3131         RETURN(rc);
3132 }
3133
3134 static int osd_object_destroy(const struct lu_env *env,
3135                               struct dt_object *dt,
3136                               struct thandle *th)
3137 {
3138         const struct lu_fid    *fid = lu_object_fid(&dt->do_lu);
3139         struct osd_object      *obj = osd_dt_obj(dt);
3140         struct inode           *inode = obj->oo_inode;
3141         struct osd_device      *osd = osd_obj2dev(obj);
3142         struct osd_thandle     *oh;
3143         int                     result;
3144         ENTRY;
3145