lustre/osd-zfs/osd_object.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.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  *
  31  * lustre/osd-zfs/osd_object.c
  32  *
  33  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
  34  * Author: Mike Pershin <tappro@whamcloud.com>
  35  * Author: Johann Lombardi <johann@whamcloud.com>
  36  */
  37
  38 #define DEBUG_SUBSYSTEM S_OSD
  39
  40 #include <libcfs/libcfs.h>
  41 #include <obd_support.h>
  42 #include <lustre_net.h>
  43 #include <obd.h>
  44 #include <obd_class.h>
  45 #include <lustre_disk.h>
  46 #include <lustre_fid.h>
  47
  48 #include "osd_internal.h"
  49
  50 #include <sys/dnode.h>
  51 #include <sys/dbuf.h>
  52 #include <sys/spa.h>
  53 #include <sys/stat.h>
  54 #include <sys/zap.h>
  55 #include <sys/spa_impl.h>
  56 #include <sys/zfs_znode.h>
  57 #include <sys/dmu_tx.h>
  58 #include <sys/dmu_objset.h>
  59 #include <sys/dsl_prop.h>
  60 #include <sys/sa_impl.h>
  61 #include <sys/txg.h>
  62
  63 char *osd_obj_tag = "osd_object";
  64 static int osd_object_sync_delay_us = -1;
  65
  66 static const struct dt_object_operations osd_obj_ops;
  67 static const struct lu_object_operations osd_lu_obj_ops;
  68 static const struct dt_object_operations osd_obj_otable_it_ops;
  69
  70 static void
  71 osd_object_sa_fini(struct osd_object *obj)
  72 {
  73         if (obj->oo_sa_hdl) {
  74                 sa_handle_destroy(obj->oo_sa_hdl);
  75                 obj->oo_sa_hdl = NULL;
  76         }
  77 }
  78
  79 static int
  80 osd_object_sa_init(struct osd_object *obj, struct osd_device *o)
  81 {
  82         int rc;
  83
  84         LASSERT(obj->oo_sa_hdl == NULL);
  85         LASSERT(obj->oo_dn != NULL);
  86
  87         rc = osd_sa_handle_get(obj);
  88         if (rc)
  89                 return rc;
  90
  91         /* Cache the xattr object id, valid for the life of the object */
  92         rc = -sa_lookup(obj->oo_sa_hdl, SA_ZPL_XATTR(o), &obj->oo_xattr, 8);
  93         if (rc == -ENOENT) {
  94                 obj->oo_xattr = ZFS_NO_OBJECT;
  95                 rc = 0;
  96         } else if (rc) {
  97                 osd_object_sa_fini(obj);
  98         }
  99
 100         return rc;
 101 }
 102
 103 /*
 104  * Add object to list of dirty objects in tx handle.
 105  */
 106 void osd_object_sa_dirty_add(struct osd_object *obj, struct osd_thandle *oh)
 107 {
 108         if (!list_empty(&obj->oo_sa_linkage))
 109                 return;
 110
 111         write_lock(&obj->oo_attr_lock);
 112         if (likely(list_empty(&obj->oo_sa_linkage)))
 113                 list_add(&obj->oo_sa_linkage, &oh->ot_sa_list);
 114         write_unlock(&obj->oo_attr_lock);
 115 }
 116
 117 /*
 118  * Release spill block dbuf hold for all dirty SAs.
 119  */
 120 void osd_object_sa_dirty_rele(const struct lu_env *env, struct osd_thandle *oh)
 121 {
 122         struct osd_object *obj;
 123
 124         while (!list_empty(&oh->ot_sa_list)) {
 125                 obj = list_first_entry(&oh->ot_sa_list,
 126                                        struct osd_object, oo_sa_linkage);
 127                 write_lock(&obj->oo_attr_lock);
 128                 list_del_init(&obj->oo_sa_linkage);
 129                 write_unlock(&obj->oo_attr_lock);
 130                 if (obj->oo_late_xattr && obj->oo_destroyed == 0) {
 131                         /*
 132                          * take oo_guard to protect oo_sa_xattr buffer
 133                          * from concurrent update by osd_xattr_set()
 134                          */
 135                         LASSERT(oh->ot_assigned != 0);
 136                         down_write(&obj->oo_guard);
 137                         if (obj->oo_destroyed == 0) {
 138                                 if (obj->oo_late_attr_set)
 139                                         __osd_sa_attr_init(env, obj, oh);
 140                                 else if (obj->oo_late_xattr)
 141                                         __osd_sa_xattr_update(env, obj, oh);
 142                         }
 143                         up_write(&obj->oo_guard);
 144                 }
 145                 sa_spill_rele(obj->oo_sa_hdl);
 146         }
 147 }
 148
 149 /*
 150  * Update the SA and add the object to the dirty list.
 151  */
 152 int osd_object_sa_update(struct osd_object *obj, sa_attr_type_t type,
 153                          void *buf, uint32_t buflen, struct osd_thandle *oh)
 154 {
 155         int rc;
 156
 157         LASSERT(obj->oo_sa_hdl != NULL);
 158         LASSERT(oh->ot_tx != NULL);
 159
 160         rc = -sa_update(obj->oo_sa_hdl, type, buf, buflen, oh->ot_tx);
 161         osd_object_sa_dirty_add(obj, oh);
 162
 163         return rc;
 164 }
 165
 166 /*
 167  * Bulk update the SA and add the object to the dirty list.
 168  */
 169 static int
 170 osd_object_sa_bulk_update(struct osd_object *obj, sa_bulk_attr_t *attrs,
 171                           int count, struct osd_thandle *oh)
 172 {
 173         int rc;
 174
 175         LASSERT(obj->oo_sa_hdl != NULL);
 176         LASSERT(oh->ot_tx != NULL);
 177
 178         rc = -sa_bulk_update(obj->oo_sa_hdl, attrs, count, oh->ot_tx);
 179         osd_object_sa_dirty_add(obj, oh);
 180
 181         return rc;
 182 }
 183
 184 /*
 185  * Retrieve the attributes of a DMU object
 186  */
 187 static int __osd_object_attr_get(const struct lu_env *env, struct osd_device *o,
 188                                  struct osd_object *obj, struct lu_attr *la)
 189 {
 190         struct osa_attr *osa = &osd_oti_get(env)->oti_osa;
 191         sa_bulk_attr_t *bulk = osd_oti_get(env)->oti_attr_bulk;
 192         struct lustre_mdt_attrs *lma;
 193         struct lu_buf buf;
 194         int cnt = 0;
 195         int              rc;
 196
 197         ENTRY;
 198         LASSERT(obj->oo_dn != NULL);
 199
 200         la->la_valid |= LA_ATIME | LA_MTIME | LA_CTIME | LA_BTIME | LA_MODE |
 201                         LA_TYPE | LA_SIZE | LA_UID | LA_GID | LA_FLAGS |
 202                         LA_NLINK;
 203
 204         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(o), NULL, osa->atime, 16);
 205         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(o), NULL, osa->mtime, 16);
 206         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(o), NULL, osa->ctime, 16);
 207         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CRTIME(o), NULL, osa->btime, 16);
 208         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(o), NULL, &osa->mode, 8);
 209         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(o), NULL, &osa->size, 8);
 210         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(o), NULL, &osa->nlink, 8);
 211         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(o), NULL, &osa->uid, 8);
 212         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(o), NULL, &osa->gid, 8);
 213         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(o), NULL, &osa->flags, 8);
 214         LASSERT(cnt <= ARRAY_SIZE(osd_oti_get(env)->oti_attr_bulk));
 215
 216         rc = -sa_bulk_lookup(obj->oo_sa_hdl, bulk, cnt);
 217         if (rc)
 218                 GOTO(out_sa, rc);
 219
 220 #ifdef ZFS_PROJINHERIT
 221         if (o->od_projectused_dn && osa->flags & ZFS_PROJID) {
 222                 rc = -sa_lookup(obj->oo_sa_hdl, SA_ZPL_PROJID(o),
 223                                 &osa->projid, 8);
 224                 if (rc)
 225                         GOTO(out_sa, rc);
 226
 227                 la->la_projid = osa->projid;
 228                 la->la_valid |= LA_PROJID;
 229                 obj->oo_with_projid = 1;
 230         } else {
 231                 la->la_projid = ZFS_DEFAULT_PROJID;
 232                 la->la_valid &= ~LA_PROJID;
 233         }
 234 #else
 235         la->la_projid = 0;
 236         la->la_valid &= ~LA_PROJID;
 237 #endif
 238
 239         la->la_atime = osa->atime[0];
 240         la->la_mtime = osa->mtime[0];
 241         la->la_ctime = osa->ctime[0];
 242         la->la_btime = osa->btime[0];
 243         la->la_mode = osa->mode;
 244         la->la_uid = osa->uid;
 245         la->la_gid = osa->gid;
 246         la->la_nlink = osa->nlink;
 247         la->la_flags = attrs_zfs2fs(osa->flags);
 248         la->la_size = osa->size;
 249
 250         /* Try to get extra flags from LMA */
 251         lma = (struct lustre_mdt_attrs *)osd_oti_get(env)->oti_buf;
 252         buf.lb_buf = lma;
 253         buf.lb_len = sizeof(osd_oti_get(env)->oti_buf);
 254         down_read(&obj->oo_guard);
 255         rc = osd_xattr_get_lma(env, obj, &buf);
 256         if (!rc) {
 257                 lma->lma_incompat = le32_to_cpu(lma->lma_incompat);
 258                 obj->oo_lma_flags =
 259                         lma_to_lustre_flags(lma->lma_incompat);
 260         } else if (rc == -ENODATA ||
 261                    !(S_ISDIR(la->la_mode) &&
 262                      dt_object_exists(&obj->oo_dt))) {
 263                 rc = 0;
 264         }
 265         up_read(&obj->oo_guard);
 266
 267         if (S_ISCHR(la->la_mode) || S_ISBLK(la->la_mode)) {
 268                 rc = -sa_lookup(obj->oo_sa_hdl, SA_ZPL_RDEV(o), &osa->rdev, 8);
 269                 if (rc)
 270                         GOTO(out_sa, rc);
 271                 la->la_rdev = osa->rdev;
 272                 la->la_valid |= LA_RDEV;
 273         }
 274 out_sa:
 275
 276         RETURN(rc);
 277 }
 278
 279 int __osd_obj2dnode(objset_t *os, uint64_t oid, dnode_t **dnp)
 280 {
 281         dmu_buf_t *db;
 282         dmu_buf_impl_t *dbi;
 283         int rc;
 284
 285         rc = -dmu_bonus_hold(os, oid, osd_obj_tag, &db);
 286         if (rc)
 287                 return rc;
 288
 289         dbi = (dmu_buf_impl_t *)db;
 290         DB_DNODE_ENTER(dbi);
 291         *dnp = DB_DNODE(dbi);
 292         DB_DNODE_EXIT(dbi);
 293         LASSERT(*dnp != NULL);
 294
 295         return 0;
 296 }
 297
 298 /*
 299  * Concurrency: no concurrent access is possible that early in object
 300  * life-cycle.
 301  */
 302 struct lu_object *osd_object_alloc(const struct lu_env *env,
 303                                    const struct lu_object_header *hdr,
 304                                    struct lu_device *d)
 305 {
 306         struct osd_object *mo;
 307
 308         OBD_SLAB_ALLOC_PTR_GFP(mo, osd_object_kmem, GFP_NOFS);
 309         if (mo != NULL) {
 310                 struct lu_object *l;
 311                 struct lu_object_header *h;
 312                 struct osd_device *o = osd_dev(d);
 313                 int i;
 314
 315                 l = &mo->oo_dt.do_lu;
 316                 if (unlikely(o->od_in_init)) {
 317                         OBD_ALLOC_PTR(h);
 318                         if (!h) {
 319                                 OBD_FREE_PTR(mo);
 320                                 return NULL;
 321                         }
 322
 323                         lu_object_header_init(h);
 324                         lu_object_init(l, h, d);
 325                         lu_object_add_top(h, l);
 326                         mo->oo_header = h;
 327                 } else {
 328                         dt_object_init(&mo->oo_dt, NULL, d);
 329                         mo->oo_header = NULL;
 330                 }
 331
 332                 mo->oo_dt.do_ops = &osd_obj_ops;
 333                 l->lo_ops = &osd_lu_obj_ops;
 334                 INIT_LIST_HEAD(&mo->oo_sa_linkage);
 335                 INIT_LIST_HEAD(&mo->oo_unlinked_linkage);
 336                 init_rwsem(&mo->oo_sem);
 337                 init_rwsem(&mo->oo_guard);
 338                 rwlock_init(&mo->oo_attr_lock);
 339                 mo->oo_destroy = OSD_DESTROY_NONE;
 340                 for (i = 0; i < OSD_MAX_DBUFS; i++)
 341                         mo->oo_dbs[i] = NULL;
 342                 return l;
 343         } else {
 344                 return NULL;
 345         }
 346 }
 347
 348 static void osd_obj_set_blksize(const struct lu_env *env,
 349                                 struct osd_device *osd, struct osd_object *obj)
 350 {
 351         const struct lu_fid *fid = lu_object_fid(&obj->oo_dt.do_lu);
 352         dmu_tx_t *tx;
 353         dnode_t *dn = obj->oo_dn;
 354         uint32_t blksz;
 355         int rc = 0;
 356
 357         ENTRY;
 358         LASSERT(!osd_oti_get(env)->oti_in_trans);
 359
 360         tx = dmu_tx_create(osd->od_os);
 361         if (!tx) {
 362                 CERROR("%s: fail to create tx to set blksize for "DFID"\n",
 363                        osd->od_svname, PFID(fid));
 364                 RETURN_EXIT;
 365         }
 366
 367         dmu_tx_hold_bonus(tx, dn->dn_object);
 368         rc = -dmu_tx_assign(tx, TXG_WAIT);
 369         if (rc) {
 370                 dmu_tx_abort(tx);
 371                 CERROR("%s: fail to assign tx to set blksize for "DFID": rc = %d\n",
 372                        osd->od_svname, PFID(fid), rc);
 373                 RETURN_EXIT;
 374         }
 375
 376         down_write(&obj->oo_guard);
 377         if (unlikely((1 << dn->dn_datablkshift) >= PAGE_SIZE))
 378                 GOTO(out, rc = 1);
 379
 380         blksz = dn->dn_datablksz;
 381         if (!is_power_of_2(blksz))
 382                 blksz = size_roundup_power2(blksz);
 383
 384         if (blksz > osd->od_max_blksz)
 385                 blksz = osd->od_max_blksz;
 386         else if (blksz < PAGE_SIZE)
 387                 blksz = PAGE_SIZE;
 388         rc = -dmu_object_set_blocksize(osd->od_os, dn->dn_object, blksz, 0, tx);
 389
 390         GOTO(out, rc);
 391
 392 out:
 393         up_write(&obj->oo_guard);
 394         if (rc) {
 395                 dmu_tx_abort(tx);
 396                 if (unlikely(obj->oo_dn->dn_maxblkid > 0))
 397                         rc = 1;
 398                 if (rc < 0)
 399                         CERROR("%s: fail to set blksize for "DFID": rc = %d\n",
 400                                osd->od_svname, PFID(fid), rc);
 401         } else {
 402                 dmu_tx_commit(tx);
 403                 CDEBUG(D_INODE, "%s: set blksize as %u for "DFID"\n",
 404                        osd->od_svname, blksz, PFID(fid));
 405         }
 406 }
 407
 408 /*
 409  * Concurrency: shouldn't matter.
 410  */
 411 static int osd_object_init0(const struct lu_env *env, struct osd_object *obj)
 412 {
 413         struct osd_device       *osd = osd_obj2dev(obj);
 414         const struct lu_fid     *fid = lu_object_fid(&obj->oo_dt.do_lu);
 415         int                      rc = 0;
 416
 417         ENTRY;
 418         LASSERT(obj->oo_dn);
 419
 420         rc = osd_object_sa_init(obj, osd);
 421         if (rc)
 422                 RETURN(rc);
 423
 424         /* cache attrs in object */
 425         rc = __osd_object_attr_get(env, osd, obj, &obj->oo_attr);
 426         if (rc)
 427                 RETURN(rc);
 428
 429         if (likely(!fid_is_acct(fid))) {
 430                 /* no body operations for accounting objects */
 431                 obj->oo_dt.do_body_ops = &osd_body_ops;
 432
 433                 if (S_ISREG(obj->oo_attr.la_mode) &&
 434                     obj->oo_dn->dn_maxblkid == 0 &&
 435                     (1 << obj->oo_dn->dn_datablkshift) < PAGE_SIZE &&
 436                     (fid_is_idif(fid) || fid_is_norm(fid) ||
 437                      fid_is_echo(fid)) &&
 438                     osd->od_is_ost && !osd->od_dt_dev.dd_rdonly)
 439                         osd_obj_set_blksize(env, osd, obj);
 440         }
 441
 442         /*
 443          * initialize object before marking it existing
 444          */
 445         obj->oo_dt.do_lu.lo_header->loh_attr |= obj->oo_attr.la_mode & S_IFMT;
 446
 447         smp_mb();
 448         obj->oo_dt.do_lu.lo_header->loh_attr |= LOHA_EXISTS;
 449
 450         RETURN(0);
 451 }
 452
 453 static int osd_check_lma(const struct lu_env *env, struct osd_object *obj)
 454 {
 455         struct osd_thread_info  *info = osd_oti_get(env);
 456         struct lu_buf           buf;
 457         int                     rc;
 458         struct lustre_mdt_attrs *lma;
 459         const struct lu_fid *rfid = lu_object_fid(&obj->oo_dt.do_lu);
 460
 461         ENTRY;
 462         BUILD_BUG_ON(sizeof(info->oti_buf) < sizeof(*lma));
 463         lma = (struct lustre_mdt_attrs *)info->oti_buf;
 464         buf.lb_buf = lma;
 465         buf.lb_len = sizeof(info->oti_buf);
 466
 467         rc = osd_xattr_get(env, &obj->oo_dt, &buf, XATTR_NAME_LMA);
 468         if (rc > 0) {
 469                 rc = 0;
 470                 lustre_lma_swab(lma);
 471                 if (unlikely((lma->lma_incompat & ~LMA_INCOMPAT_SUPP) ||
 472                              CFS_FAIL_CHECK(OBD_FAIL_OSD_LMA_INCOMPAT))) {
 473                         rc = -EOPNOTSUPP;
 474                         CWARN("%s: unsupported incompat LMA feature(s) %#x for fid = "DFID": rc = %d\n",
 475                               osd_obj2dev(obj)->od_svname,
 476                               lma->lma_incompat & ~LMA_INCOMPAT_SUPP,
 477                               PFID(rfid), rc);
 478                 } else if (unlikely(!lu_fid_eq(rfid, &lma->lma_self_fid))) {
 479                         rc = -EREMCHG;
 480                         CERROR("%s: FID-in-LMA "DFID" does not match the object self-fid "DFID": rc = %d\n",
 481                               osd_obj2dev(obj)->od_svname,
 482                               PFID(&lma->lma_self_fid), PFID(rfid), rc);
 483                 } else {
 484                         struct osd_device *osd = osd_obj2dev(obj);
 485
 486                         if (lma->lma_compat & LMAC_STRIPE_INFO &&
 487                             osd->od_is_ost)
 488                                 obj->oo_pfid_in_lma = 1;
 489                         if (unlikely(lma->lma_incompat & LMAI_REMOTE_PARENT) &&
 490                             osd->od_remote_parent_dir != ZFS_NO_OBJECT)
 491                                 lu_object_set_agent_entry(&obj->oo_dt.do_lu);
 492                 }
 493         } else if (rc == -ENODATA) {
 494                 /* haven't initialize LMA xattr */
 495                 rc = 0;
 496         }
 497
 498         RETURN(rc);
 499 }
 500
 501 /**
 502  * Helper function to retrieve DMU object id from fid for accounting object
 503  */
 504 static dnode_t *osd_quota_fid2dmu(const struct osd_device *osd,
 505                                   const struct lu_fid *fid)
 506 {
 507         dnode_t *dn = NULL;
 508
 509         LASSERT(fid_is_acct(fid));
 510
 511         switch (fid_oid(fid)) {
 512         case ACCT_USER_OID:
 513                 dn = osd->od_userused_dn;
 514                 break;
 515         case ACCT_GROUP_OID:
 516                 dn = osd->od_groupused_dn;
 517                 break;
 518 #ifdef ZFS_PROJINHERIT
 519         case ACCT_PROJECT_OID:
 520                 dn = osd->od_projectused_dn;
 521                 break;
 522 #endif
 523         default:
 524                 break;
 525         }
 526
 527         return dn;
 528 }
 529
 530 /*
 531  * Concurrency: no concurrent access is possible that early in object
 532  * life-cycle.
 533  */
 534 static int osd_object_init(const struct lu_env *env, struct lu_object *l,
 535                            const struct lu_object_conf *conf)
 536 {
 537         struct osd_object *obj = osd_obj(l);
 538         struct osd_device *osd = osd_obj2dev(obj);
 539         const struct lu_fid *fid = lu_object_fid(l);
 540         struct lustre_scrub *scrub = &osd->od_scrub;
 541         struct osd_thread_info *info = osd_oti_get(env);
 542         struct luz_direntry *zde = &info->oti_zde;
 543         struct osd_idmap_cache *idc;
 544         char *name = info->oti_str;
 545         uint64_t oid;
 546         int rc = 0;
 547         int rc1;
 548         bool remote = false;
 549
 550         ENTRY;
 551         LASSERT(osd_invariant(obj));
 552
 553         if (fid_is_otable_it(&l->lo_header->loh_fid)) {
 554                 obj->oo_dt.do_ops = &osd_obj_otable_it_ops;
 555                 l->lo_header->loh_attr |= LOHA_EXISTS;
 556
 557                 GOTO(out, rc = 0);
 558         }
 559
 560         if (conf && conf->loc_flags & LOC_F_NEW)
 561                 GOTO(out, rc = 0);
 562
 563         if (unlikely(fid_is_acct(fid))) {
 564                 obj->oo_dn = osd_quota_fid2dmu(osd, fid);
 565                 if (obj->oo_dn) {
 566                         obj->oo_dt.do_index_ops = &osd_acct_index_ops;
 567                         l->lo_header->loh_attr |= LOHA_EXISTS;
 568                 }
 569
 570                 GOTO(out, rc = 0);
 571         }
 572
 573         idc = osd_idc_find(env, osd, fid);
 574         if (idc && !idc->oic_remote && idc->oic_dnode != ZFS_NO_OBJECT) {
 575                 oid = idc->oic_dnode;
 576                 goto zget;
 577         }
 578
 579         rc = -ENOENT;
 580         if (!list_empty(&osd->od_scrub.os_inconsistent_items))
 581                 rc = osd_oii_lookup(osd, fid, &oid);
 582
 583         if (rc)
 584                 rc = osd_fid_lookup(env, osd, fid, &oid);
 585
 586         if (rc == -ENOENT) {
 587                 if (likely(!(fid_is_norm(fid) || fid_is_igif(fid)) ||
 588                            fid_is_on_ost(env, osd, fid) ||
 589                            !zfs_test_bit(osd_oi_fid2idx(osd, fid),
 590                                          scrub->os_file.sf_oi_bitmap)))
 591                         GOTO(out, rc = 0);
 592
 593                 rc = -EREMCHG;
 594                 goto trigger;
 595         }
 596
 597         if (rc)
 598                 GOTO(out, rc);
 599
 600 zget:
 601         LASSERT(obj->oo_dn == NULL);
 602
 603         rc = __osd_obj2dnode(osd->od_os, oid, &obj->oo_dn);
 604         /* EEXIST will be returned if object is being deleted in ZFS */
 605         if (rc == -EEXIST)
 606                 GOTO(out, rc = 0);
 607
 608         if (rc) {
 609                 CERROR("%s: lookup "DFID"/%#llx failed: rc = %d\n",
 610                        osd->od_svname, PFID(lu_object_fid(l)), oid, rc);
 611                 GOTO(out, rc);
 612         }
 613
 614         rc = osd_object_init0(env, obj);
 615         if (rc)
 616                 GOTO(out, rc);
 617
 618         if (unlikely(obj->oo_header))
 619                 GOTO(out, rc = 0);
 620
 621         rc = osd_check_lma(env, obj);
 622         if (rc != -EREMCHG)
 623                 GOTO(out, rc);
 624
 625         osd_scrub_refresh_mapping(env, osd, fid, oid, DTO_INDEX_DELETE, true,
 626                                   NULL);
 627
 628 trigger:
 629         /* We still have chance to get the valid dnode: for the object that is
 630          * referenced by remote name entry, the object on the local MDT will be
 631          * linked under the dir /REMOTE_PARENT_DIR with its FID string as name.
 632          *
 633          * During the OI scrub, if we cannot find the OI mapping, we may still
 634          * have change to map the FID to local OID via lookup the dir
 635          * /REMOTE_PARENT_DIR.
 636          */
 637         if (!remote && !fid_is_on_ost(env, osd, fid)) {
 638                 osd_fid2str(name, fid, sizeof(info->oti_str));
 639                 rc = osd_zap_lookup(osd, osd->od_remote_parent_dir,
 640                                     NULL, name, 8, 3, (void *)zde);
 641                 if (!rc) {
 642                         oid = zde->lzd_reg.zde_dnode;
 643                         osd_dnode_rele(obj->oo_dn);
 644                         obj->oo_dn = NULL;
 645                         remote = true;
 646                         goto zget;
 647                 }
 648         }
 649
 650         /* The case someone triggered the OI scrub already. */
 651         if (scrub->os_running) {
 652                 if (!rc) {
 653                         LASSERT(remote);
 654
 655                         lu_object_set_agent_entry(l);
 656                         osd_oii_insert(env, osd, fid, oid, false);
 657                 } else {
 658                         rc = -EINPROGRESS;
 659                 }
 660
 661                 GOTO(out, rc);
 662         }
 663
 664         /* The case NOT allow to trigger OI scrub automatically. */
 665         if (osd->od_scrub.os_auto_scrub_interval == AS_NEVER)
 666                 GOTO(out, rc);
 667
 668         /* It is me to trigger the OI scrub. */
 669         rc1 = osd_scrub_start(env, osd, SS_CLEAR_DRYRUN |
 670                               SS_CLEAR_FAILOUT | SS_AUTO_FULL);
 671         CDEBUG_LIMIT(D_LFSCK | D_CONSOLE | D_WARNING,
 672                      "%s: trigger OI scrub by RPC for "DFID"/%#llx: rc = %d\n",
 673                      osd_name(osd), PFID(fid), oid, rc1);
 674         if (!rc) {
 675                 LASSERT(remote);
 676
 677                 lu_object_set_agent_entry(l);
 678                 if (!rc1)
 679                         osd_oii_insert(env, osd, fid, oid, false);
 680         } else {
 681                 if (!rc1)
 682                         rc = -EINPROGRESS;
 683                 else
 684                         rc = -EREMCHG;
 685         }
 686
 687         GOTO(out, rc);
 688
 689 out:
 690         RETURN(rc);
 691 }
 692
 693 /*
 694  * Concurrency: no concurrent access is possible that late in object
 695  * life-cycle.
 696  */
 697 static void osd_object_free(const struct lu_env *env, struct lu_object *l)
 698 {
 699         struct osd_object *obj = osd_obj(l);
 700         struct lu_object_header *h = obj->oo_header;
 701
 702         LASSERT(osd_invariant(obj));
 703
 704         dt_object_fini(&obj->oo_dt);
 705         /* obj doesn't contain an lu_object_header, so we don't need call_rcu */
 706         OBD_SLAB_FREE_PTR(obj, osd_object_kmem);
 707         if (unlikely(h))
 708                 lu_object_header_free(h);
 709 }
 710
 711 static int
 712 osd_object_unlinked_add(struct osd_object *obj, struct osd_thandle *oh)
 713 {
 714         int rc = -EBUSY;
 715
 716         LASSERT(obj->oo_destroy == OSD_DESTROY_ASYNC);
 717
 718         /* the object is supposed to be exclusively locked by
 719          * the caller (osd_destroy()), while the transaction
 720          * (oh) is per-thread and not shared
 721          */
 722         if (likely(list_empty(&obj->oo_unlinked_linkage))) {
 723                 list_add(&obj->oo_unlinked_linkage, &oh->ot_unlinked_list);
 724                 rc = 0;
 725         }
 726
 727         return rc;
 728 }
 729
 730 /* Default to max data size covered by a level-1 indirect block */
 731 static unsigned long osd_sync_destroy_max_size =
 732         1UL << (DN_MAX_INDBLKSHIFT - SPA_BLKPTRSHIFT + SPA_MAXBLOCKSHIFT);
 733 module_param(osd_sync_destroy_max_size, ulong, 0444);
 734 MODULE_PARM_DESC(osd_sync_destroy_max_size, "Maximum object size to use synchronous destroy.");
 735
 736 static inline void
 737 osd_object_set_destroy_type(struct osd_object *obj)
 738 {
 739         /*
 740          * Lock-less OST_WRITE can race with OST_DESTROY, so set destroy type
 741          * only once and use it consistently thereafter.
 742          */
 743         down_write(&obj->oo_guard);
 744         if (obj->oo_destroy == OSD_DESTROY_NONE) {
 745                 if (obj->oo_attr.la_size <= osd_sync_destroy_max_size)
 746                         obj->oo_destroy = OSD_DESTROY_SYNC;
 747                 else /* Larger objects are destroyed asynchronously */
 748                         obj->oo_destroy = OSD_DESTROY_ASYNC;
 749         }
 750         up_write(&obj->oo_guard);
 751 }
 752
 753 static int osd_declare_destroy(const struct lu_env *env, struct dt_object *dt,
 754                                struct thandle *th)
 755 {
 756         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
 757         struct osd_object       *obj = osd_dt_obj(dt);
 758         struct osd_device       *osd = osd_obj2dev(obj);
 759         struct osd_thandle      *oh;
 760         dnode_t *dn;
 761         int                      rc;
 762         uint64_t                 zapid;
 763
 764         ENTRY;
 765         LASSERT(th != NULL);
 766         if (unlikely(obj->oo_dn == NULL))
 767                 RETURN(-ENOENT);
 768
 769         oh = container_of(th, struct osd_thandle, ot_super);
 770         LASSERT(oh->ot_tx != NULL);
 771
 772         dmu_tx_mark_netfree(oh->ot_tx);
 773
 774         /* declare that we'll remove object from fid-dnode mapping */
 775         zapid = osd_get_name_n_idx(env, osd, fid, NULL, 0, &dn);
 776         osd_tx_hold_zap(oh->ot_tx, zapid, dn, FALSE, NULL);
 777
 778         osd_declare_xattrs_destroy(env, obj, oh);
 779
 780         /* one less inode */
 781         rc = osd_declare_quota(env, osd, obj->oo_attr.la_uid,
 782                                obj->oo_attr.la_gid, obj->oo_attr.la_projid,
 783                                -1, oh, NULL, OSD_QID_INODE);
 784         if (rc)
 785                 RETURN(rc);
 786
 787         /* data to be truncated */
 788         rc = osd_declare_quota(env, osd, obj->oo_attr.la_uid,
 789                                obj->oo_attr.la_gid, obj->oo_attr.la_projid,
 790                                0, oh, NULL, OSD_QID_BLK);
 791         if (rc)
 792                 RETURN(rc);
 793
 794         osd_object_set_destroy_type(obj);
 795         if (obj->oo_destroy == OSD_DESTROY_SYNC)
 796                 dmu_tx_hold_free(oh->ot_tx, obj->oo_dn->dn_object,
 797                                  0, DMU_OBJECT_END);
 798         else
 799                 osd_tx_hold_zap(oh->ot_tx, osd->od_unlinked->dn_object,
 800                                 osd->od_unlinked, TRUE, NULL);
 801
 802         /* remove agent entry (if have) from remote parent */
 803         if (lu_object_has_agent_entry(&obj->oo_dt.do_lu))
 804                 osd_tx_hold_zap(oh->ot_tx, osd->od_remote_parent_dir,
 805                                 NULL, FALSE, NULL);
 806
 807         /* will help to find FID->ino when this obj is being added to PENDING */
 808         osd_idc_find_and_init(env, osd, obj);
 809
 810         RETURN(0);
 811 }
 812
 813 static int osd_destroy(const struct lu_env *env, struct dt_object *dt,
 814                        struct thandle *th)
 815 {
 816         struct osd_thread_info  *info = osd_oti_get(env);
 817         char                    *buf = info->oti_str;
 818         struct osd_object       *obj = osd_dt_obj(dt);
 819         struct osd_device       *osd = osd_obj2dev(obj);
 820         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
 821         struct osd_thandle      *oh;
 822         int                      rc;
 823         uint64_t                 oid, zapid;
 824         dnode_t *zdn;
 825
 826         ENTRY;
 827         down_write(&obj->oo_guard);
 828
 829         if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
 830                 GOTO(out, rc = -ENOENT);
 831
 832         LASSERT(obj->oo_dn != NULL);
 833
 834         oh = container_of(th, struct osd_thandle, ot_super);
 835         LASSERT(oh != NULL);
 836         LASSERT(oh->ot_tx != NULL);
 837
 838         /* remove obj ref from index dir (it depends) */
 839         zapid = osd_get_name_n_idx(env, osd, fid, buf,
 840                                    sizeof(info->oti_str), &zdn);
 841         rc = osd_xattrs_destroy(env, obj, oh);
 842         if (rc) {
 843                 CERROR("%s: cannot destroy xattrs for %s: rc = %d\n",
 844                        osd->od_svname, buf, rc);
 845                 GOTO(out, rc);
 846         }
 847
 848         if (lu_object_has_agent_entry(&obj->oo_dt.do_lu)) {
 849                 rc = osd_delete_from_remote_parent(env, osd, obj, oh, true);
 850                 if (rc)
 851                         GOTO(out, rc);
 852         }
 853
 854         oid = obj->oo_dn->dn_object;
 855         if (unlikely(obj->oo_destroy == OSD_DESTROY_NONE)) {
 856                 /* this may happen if the destroy wasn't declared
 857                  * e.g. when the object is created and then destroyed
 858                  * in the same transaction - we don't need additional
 859                  * space for destroy specifically
 860                  */
 861                 LASSERT(obj->oo_attr.la_size <= osd_sync_destroy_max_size);
 862                 rc = -dmu_object_free(osd->od_os, oid, oh->ot_tx);
 863                 if (rc)
 864                         CERROR("%s: failed to free %s/%#llx: rc = %d\n",
 865                                osd->od_svname, buf, oid, rc);
 866         } else if (obj->oo_destroy == OSD_DESTROY_SYNC) {
 867                 rc = -dmu_object_free(osd->od_os, oid, oh->ot_tx);
 868                 if (rc)
 869                         CERROR("%s: failed to free %s/%#llx: rc = %d\n",
 870                                osd->od_svname, buf, oid, rc);
 871         } else { /* asynchronous destroy */
 872                 char *key = info->oti_key;
 873
 874                 rc = osd_object_unlinked_add(obj, oh);
 875                 if (rc)
 876                         GOTO(out, rc);
 877
 878                 snprintf(key, sizeof(info->oti_key), "%llx", oid);
 879                 rc = osd_zap_add(osd, osd->od_unlinked->dn_object,
 880                                  osd->od_unlinked, key, 8, 1, &oid, oh->ot_tx);
 881                 if (rc)
 882                         CERROR("%s: zap_add_int() failed %s/%#llx: rc = %d\n",
 883                                osd->od_svname, buf, oid, rc);
 884         }
 885
 886         /* Remove the OI mapping after the destroy to handle the race with
 887          * OI scrub that may insert missed OI mapping during the interval.
 888          */
 889         rc = osd_zap_remove(osd, zapid, zdn, buf, oh->ot_tx);
 890         if (unlikely(rc == -ENOENT))
 891                 rc = 0;
 892         if (rc)
 893                 CERROR("%s: zap_remove(%s) failed: rc = %d\n",
 894                        osd->od_svname, buf, rc);
 895
 896         GOTO(out, rc);
 897
 898 out:
 899         /* not needed in the cache anymore */
 900         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
 901         if (rc == 0)
 902                 obj->oo_destroyed = 1;
 903         up_write(&obj->oo_guard);
 904         RETURN(0);
 905 }
 906
 907 static void osd_object_delete(const struct lu_env *env, struct lu_object *l)
 908 {
 909         struct osd_object *obj = osd_obj(l);
 910         const struct lu_fid *fid = lu_object_fid(l);
 911         dmu_buf_t **dbs = obj->oo_dbs;
 912         int i;
 913
 914         for (i = 0; i < OSD_MAX_DBUFS; i++) {
 915                 if (dbs[i])
 916                         dbuf_rele((dmu_buf_impl_t *)dbs[i], osd_0copy_tag);
 917         }
 918
 919         if (obj->oo_dn) {
 920                 if (likely(!fid_is_acct(fid))) {
 921                         osd_object_sa_fini(obj);
 922                         if (obj->oo_sa_xattr) {
 923                                 nvlist_free(obj->oo_sa_xattr);
 924                                 obj->oo_sa_xattr = NULL;
 925                         }
 926                         osd_dnode_rele(obj->oo_dn);
 927                         list_del(&obj->oo_sa_linkage);
 928                 }
 929                 obj->oo_dn = NULL;
 930         }
 931 }
 932
 933 /*
 934  * Concurrency: ->loo_object_release() is called under site spin-lock.
 935  */
 936 static void osd_object_release(const struct lu_env *env,
 937                                struct lu_object *l)
 938 {
 939 }
 940
 941 /*
 942  * Concurrency: shouldn't matter.
 943  */
 944 static int osd_object_print(const struct lu_env *env, void *cookie,
 945                             lu_printer_t p, const struct lu_object *l)
 946 {
 947         struct osd_object *o = osd_obj(l);
 948
 949         return (*p)(env, cookie, LUSTRE_OSD_ZFS_NAME"-object@%p", o);
 950 }
 951
 952 static void osd_read_lock(const struct lu_env *env, struct dt_object *dt,
 953                           unsigned int role)
 954 {
 955         struct osd_object *obj = osd_dt_obj(dt);
 956
 957         LASSERT(osd_invariant(obj));
 958
 959         down_read_nested(&obj->oo_sem, role);
 960 }
 961
 962 static void osd_write_lock(const struct lu_env *env, struct dt_object *dt,
 963                            unsigned int role)
 964 {
 965         struct osd_object *obj = osd_dt_obj(dt);
 966
 967         LASSERT(osd_invariant(obj));
 968
 969         down_write_nested(&obj->oo_sem, role);
 970 }
 971
 972 static void osd_read_unlock(const struct lu_env *env, struct dt_object *dt)
 973 {
 974         struct osd_object *obj = osd_dt_obj(dt);
 975
 976         LASSERT(osd_invariant(obj));
 977         up_read(&obj->oo_sem);
 978 }
 979
 980 static void osd_write_unlock(const struct lu_env *env, struct dt_object *dt)
 981 {
 982         struct osd_object *obj = osd_dt_obj(dt);
 983
 984         LASSERT(osd_invariant(obj));
 985         up_write(&obj->oo_sem);
 986 }
 987
 988 static int osd_write_locked(const struct lu_env *env, struct dt_object *dt)
 989 {
 990         struct osd_object *obj = osd_dt_obj(dt);
 991         int rc = 1;
 992
 993         LASSERT(osd_invariant(obj));
 994
 995         if (down_write_trylock(&obj->oo_sem)) {
 996                 rc = 0;
 997                 up_write(&obj->oo_sem);
 998         }
 999         return rc;
1000 }
1001
1002 static int osd_attr_get(const struct lu_env *env, struct dt_object *dt,
1003                         struct lu_attr *attr)
1004 {
1005         struct osd_object *obj = osd_dt_obj(dt);
1006         struct osd_device *osd = osd_obj2dev(obj);
1007         uint64_t blocks;
1008         uint32_t blksize;
1009         int rc = 0;
1010
1011         down_read(&obj->oo_guard);
1012
1013         if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
1014                 GOTO(out, rc = -ENOENT);
1015
1016         if (unlikely(fid_is_acct(lu_object_fid(&dt->do_lu))))
1017                 GOTO(out, rc = 0);
1018
1019         LASSERT(osd_invariant(obj));
1020         LASSERT(obj->oo_dn);
1021
1022         read_lock(&obj->oo_attr_lock);
1023         *attr = obj->oo_attr;
1024         if (obj->oo_lma_flags & LUSTRE_ORPHAN_FL) {
1025                 attr->la_valid |= LA_FLAGS;
1026                 attr->la_flags |= LUSTRE_ORPHAN_FL;
1027         }
1028         if (obj->oo_lma_flags & LUSTRE_ENCRYPT_FL) {
1029                 attr->la_valid |= LA_FLAGS;
1030                 attr->la_flags |= LUSTRE_ENCRYPT_FL;
1031         }
1032         read_unlock(&obj->oo_attr_lock);
1033         if (attr->la_valid & LA_FLAGS && attr->la_flags & LUSTRE_ORPHAN_FL)
1034                 CDEBUG(D_INFO, "%s: set orphan flag on "DFID" (%#llx/%#x)\n",
1035                        osd_obj2dev(obj)->od_svname,
1036                        PFID(lu_object_fid(&dt->do_lu)),
1037                        attr->la_valid, obj->oo_lma_flags);
1038
1039         /* with ZFS_DEBUG zrl_add_debug() called by DB_DNODE_ENTER()
1040          * from within sa_object_size() can block on a mutex, so
1041          * we can't call sa_object_size() holding rwlock
1042          */
1043         sa_object_size(obj->oo_sa_hdl, &blksize, &blocks);
1044         /* we do not control size of indices, so always calculate
1045          * it from number of blocks reported by DMU
1046          */
1047         if (S_ISDIR(attr->la_mode)) {
1048                 attr->la_size = 512 * blocks;
1049                 rc = -zap_count(osd->od_os, obj->oo_dn->dn_object,
1050                                 &attr->la_dirent_count);
1051         }
1052         /* Block size may be not set; suggest maximal I/O transfers. */
1053         if (blksize == 0)
1054                 blksize = spa_maxblocksize(
1055                         dmu_objset_spa(osd_obj2dev(obj)->od_os));
1056
1057         attr->la_blksize = blksize;
1058         attr->la_blocks = blocks;
1059         attr->la_valid |= LA_BLOCKS | LA_BLKSIZE;
1060
1061 out:
1062         up_read(&obj->oo_guard);
1063         return rc;
1064 }
1065
1066 #ifdef ZFS_PROJINHERIT
1067 /*
1068  * For the existed object that is upgraded from old system, its ondisk layout
1069  * has no slot for the project ID attribute. But quota accounting logic needs
1070  * to access related slots by offset directly. So we need to adjust these old
1071  * objects' layout to make the project ID to some unified and fixed offset.
1072  */
1073 static int osd_add_projid(const struct lu_env *env, struct osd_object *obj,
1074                           struct osd_thandle *oh, uint64_t projid)
1075 {
1076         sa_bulk_attr_t *bulk = osd_oti_get(env)->oti_attr_bulk;
1077         struct osa_attr *osa = &osd_oti_get(env)->oti_osa;
1078         struct osd_device *osd = osd_obj2dev(obj);
1079         uint64_t gen;
1080         size_t sa_size;
1081         char *dxattr = NULL;
1082         int rc, cnt;
1083
1084         rc = -sa_lookup(obj->oo_sa_hdl, SA_ZPL_PROJID(osd), &osa->projid, 8);
1085         if (unlikely(rc == 0))
1086                 rc = -EEXIST;
1087         if (rc != -ENOENT)
1088                 GOTO(out, rc);
1089
1090         gen = dmu_tx_get_txg(oh->ot_tx);
1091         osa->atime[0] = obj->oo_attr.la_atime;
1092         osa->ctime[0] = obj->oo_attr.la_ctime;
1093         osa->mtime[0] = obj->oo_attr.la_mtime;
1094         osa->btime[0] = obj->oo_attr.la_btime;
1095         osa->mode = obj->oo_attr.la_mode;
1096         osa->uid = obj->oo_attr.la_uid;
1097         osa->gid = obj->oo_attr.la_gid;
1098         osa->rdev = obj->oo_attr.la_rdev;
1099         osa->nlink = obj->oo_attr.la_nlink;
1100         osa->flags = attrs_fs2zfs(obj->oo_attr.la_flags) | ZFS_PROJID;
1101         osa->size  = obj->oo_attr.la_size;
1102         osa->projid = projid;
1103
1104         cnt = 0;
1105         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(osd), NULL, &osa->mode, 8);
1106         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(osd), NULL, &osa->size, 8);
1107         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GEN(osd), NULL, &gen, 8);
1108         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(osd), NULL, &osa->uid, 8);
1109         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(osd), NULL, &osa->gid, 8);
1110         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PARENT(osd), NULL,
1111                          &obj->oo_parent, 8);
1112         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(osd), NULL, &osa->flags, 8);
1113         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(osd), NULL, osa->atime, 16);
1114         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(osd), NULL, osa->mtime, 16);
1115         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(osd), NULL, osa->ctime, 16);
1116         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CRTIME(osd), NULL, osa->btime, 16);
1117         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(osd), NULL, &osa->nlink, 8);
1118         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PROJID(osd), NULL, &osa->projid, 8);
1119         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(osd), NULL, &osa->rdev, 8);
1120         LASSERT(cnt <= ARRAY_SIZE(osd_oti_get(env)->oti_attr_bulk));
1121
1122         if (obj->oo_sa_xattr == NULL) {
1123                 rc = __osd_xattr_load(osd, obj->oo_sa_hdl, &obj->oo_sa_xattr);
1124                 if (rc)
1125                         GOTO(out, rc);
1126         }
1127
1128         if (obj->oo_sa_xattr) {
1129                 rc = -nvlist_size(obj->oo_sa_xattr, &sa_size, NV_ENCODE_XDR);
1130                 if (rc)
1131                         GOTO(out, rc);
1132
1133                 dxattr = zio_buf_alloc(sa_size);
1134                 if (dxattr == NULL)
1135                         GOTO(out, rc = -ENOMEM);
1136
1137                 rc = -nvlist_pack(obj->oo_sa_xattr, &dxattr, &sa_size,
1138                                 NV_ENCODE_XDR, KM_SLEEP);
1139                 if (rc)
1140                         GOTO(out, rc);
1141
1142                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_DXATTR(osd),
1143                                 NULL, dxattr, sa_size);
1144         }
1145
1146         rc = -sa_replace_all_by_template(obj->oo_sa_hdl, bulk, cnt, oh->ot_tx);
1147 out:
1148         if (dxattr)
1149                 zio_buf_free(dxattr, sa_size);
1150         return rc;
1151 }
1152 #endif
1153
1154 static int osd_declare_attr_set(const struct lu_env *env,
1155                                 struct dt_object *dt,
1156                                 const struct lu_attr *attr,
1157                                 struct thandle *handle)
1158 {
1159         struct osd_thread_info  *info = osd_oti_get(env);
1160         struct osd_object       *obj = osd_dt_obj(dt);
1161         struct osd_device       *osd = osd_obj2dev(obj);
1162         dmu_tx_hold_t           *txh;
1163         struct osd_thandle      *oh;
1164         uint64_t                 bspace;
1165         uint32_t                 blksize;
1166         int                      rc = 0;
1167         bool                     found;
1168
1169         ENTRY;
1170         LASSERT(handle != NULL);
1171         LASSERT(osd_invariant(obj));
1172
1173         oh = container_of(handle, struct osd_thandle, ot_super);
1174
1175         down_read(&obj->oo_guard);
1176         if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
1177                 GOTO(out_sem, rc = 0);
1178
1179         LASSERT(obj->oo_sa_hdl != NULL);
1180         LASSERT(oh->ot_tx != NULL);
1181         /* regular attributes are part of the bonus buffer */
1182         /* let's check whether this object is already part of
1183          * transaction..
1184          */
1185         found = false;
1186         for (txh = list_head(&oh->ot_tx->tx_holds); txh;
1187              txh = list_next(&oh->ot_tx->tx_holds, txh)) {
1188                 if (txh->txh_dnode == NULL)
1189                         continue;
1190                 if (txh->txh_dnode->dn_object != obj->oo_dn->dn_object)
1191                         continue;
1192                 /* this object is part of the transaction already
1193                  * we don't need to declare bonus again
1194                  */
1195                 found = true;
1196                 break;
1197         }
1198         if (!found)
1199                 dmu_tx_hold_bonus(oh->ot_tx, obj->oo_dn->dn_object);
1200         if (oh->ot_tx->tx_err != 0)
1201                 GOTO(out_sem, rc = -oh->ot_tx->tx_err);
1202
1203         if (attr && attr->la_valid & LA_FLAGS) {
1204                 /* punch must be aware we are dealing with an encrypted file */
1205                 if (attr->la_flags & LUSTRE_ENCRYPT_FL)
1206                         obj->oo_lma_flags |= LUSTRE_ENCRYPT_FL;
1207         }
1208
1209         if (attr && (attr->la_valid & (LA_UID | LA_GID | LA_PROJID))) {
1210                 sa_object_size(obj->oo_sa_hdl, &blksize, &bspace);
1211                 bspace = toqb(bspace * 512);
1212
1213                 CDEBUG(D_QUOTA,
1214                        "%s: enforce quota on UID %u, GID %u, the quota space is %lld (%u)\n",
1215                        osd->od_svname,
1216                        attr->la_uid, attr->la_gid, bspace, blksize);
1217         }
1218         /* to preserve locking order - qsd_transfer() may need to flush
1219          * currently running transaction when we're out of quota.
1220          */
1221         up_read(&obj->oo_guard);
1222
1223         /* quota enforcement for user */
1224         if (attr && attr->la_valid & LA_UID &&
1225             attr->la_uid != obj->oo_attr.la_uid) {
1226                 rc = qsd_transfer(env, osd_def_qsd(osd),
1227                                   &oh->ot_quota_trans, USRQUOTA,
1228                                   obj->oo_attr.la_uid, attr->la_uid,
1229                                   bspace, &info->oti_qi);
1230                 if (rc)
1231                         GOTO(out, rc);
1232         }
1233
1234         /* quota enforcement for group */
1235         if (attr && attr->la_valid & LA_GID &&
1236             attr->la_gid != obj->oo_attr.la_gid) {
1237                 rc = qsd_transfer(env, osd_def_qsd(osd),
1238                                   &oh->ot_quota_trans, GRPQUOTA,
1239                                   obj->oo_attr.la_gid, attr->la_gid,
1240                                   bspace, &info->oti_qi);
1241                 if (rc)
1242                         GOTO(out, rc);
1243         }
1244 #ifdef ZFS_PROJINHERIT
1245         /* quota enforcement for project */
1246         if (attr && attr->la_valid & LA_PROJID &&
1247             attr->la_projid != obj->oo_attr.la_projid) {
1248                 if (!osd->od_projectused_dn)
1249                         GOTO(out, rc = -EOPNOTSUPP);
1250
1251                 if (!zpl_is_valid_projid(attr->la_projid))
1252                         GOTO(out, rc = -EINVAL);
1253
1254                 rc = qsd_transfer(env, osd_def_qsd(osd),
1255                                   &oh->ot_quota_trans, PRJQUOTA,
1256                                   obj->oo_attr.la_projid,
1257                                   attr->la_projid, bspace,
1258                                   &info->oti_qi);
1259                 if (rc)
1260                         GOTO(out, rc);
1261         }
1262 #endif
1263 out:
1264         RETURN(rc);
1265 out_sem:
1266         up_read(&obj->oo_guard);
1267         RETURN(rc);
1268 }
1269
1270 /*
1271  * Set the attributes of an object
1272  *
1273  * The transaction passed to this routine must have
1274  * dmu_tx_hold_bonus(tx, oid) called and then assigned
1275  * to a transaction group.
1276  */
1277 static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
1278                         const struct lu_attr *la, struct thandle *handle)
1279 {
1280         struct osd_thread_info  *info = osd_oti_get(env);
1281         sa_bulk_attr_t          *bulk = osd_oti_get(env)->oti_attr_bulk;
1282         struct osd_object       *obj = osd_dt_obj(dt);
1283         struct osd_device       *osd = osd_obj2dev(obj);
1284         struct osd_thandle      *oh;
1285         struct osa_attr         *osa = &info->oti_osa;
1286         __u64                    valid = la->la_valid;
1287         int                      cnt;
1288         int                      rc = 0;
1289
1290         ENTRY;
1291         down_read(&obj->oo_guard);
1292         if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
1293                 GOTO(out, rc = -ENOENT);
1294
1295         LASSERT(handle != NULL);
1296         LASSERT(osd_invariant(obj));
1297         LASSERT(obj->oo_sa_hdl);
1298
1299         oh = container_of(handle, struct osd_thandle, ot_super);
1300         /* Assert that the transaction has been assigned to a transaction grp */
1301         LASSERT(oh->ot_tx->tx_txg != 0);
1302
1303         if (CFS_FAIL_CHECK(OBD_FAIL_OSD_FID_MAPPING) && !osd->od_is_ost) {
1304                 struct zpl_direntry *zde = &info->oti_zde.lzd_reg;
1305                 char *buf = info->oti_str;
1306                 dnode_t *zdn = NULL;
1307                 uint64_t zapid;
1308
1309                 zapid = osd_get_name_n_idx(env, osd, lu_object_fid(&dt->do_lu),
1310                                            buf, sizeof(info->oti_str), &zdn);
1311                 rc = osd_zap_lookup(osd, zapid, zdn, buf, 8,
1312                                     sizeof(*zde) / 8, zde);
1313                 if (!rc) {
1314                         zde->zde_dnode -= 1;
1315                         rc = -zap_update(osd->od_os, zapid, buf, 8,
1316                                          sizeof(*zde) / 8, zde, oh->ot_tx);
1317                 }
1318                 if (rc > 0)
1319                         rc = 0;
1320                 GOTO(out, rc);
1321         }
1322
1323         /* Only allow set size for regular file */
1324         if (!S_ISREG(dt->do_lu.lo_header->loh_attr))
1325                 valid &= ~(LA_SIZE | LA_BLOCKS);
1326
1327         if (valid & LA_CTIME && la->la_ctime == obj->oo_attr.la_ctime)
1328                 valid &= ~LA_CTIME;
1329
1330         if (valid & LA_MTIME && la->la_mtime == obj->oo_attr.la_mtime)
1331                 valid &= ~LA_MTIME;
1332
1333         if (valid & LA_ATIME && la->la_atime == obj->oo_attr.la_atime)
1334                 valid &= ~LA_ATIME;
1335
1336         if (valid == 0)
1337                 GOTO(out, rc = 0);
1338
1339         if (valid & LA_FLAGS) {
1340                 struct lustre_mdt_attrs *lma;
1341                 struct lu_buf buf;
1342                 int size = 0;
1343
1344                 if (la->la_flags & LUSTRE_LMA_FL_MASKS) {
1345                         LASSERT(!obj->oo_pfid_in_lma);
1346                         BUILD_BUG_ON(sizeof(info->oti_buf) < sizeof(*lma));
1347                         lma = (struct lustre_mdt_attrs *)&info->oti_buf;
1348                         buf.lb_buf = lma;
1349                         buf.lb_len = sizeof(info->oti_buf);
1350
1351                         /* Please do NOT call osd_xattr_get() directly, that
1352                          * will cause recursive down_read() on oo_guard.
1353                          */
1354                         rc = osd_xattr_get_internal(env, obj, &buf,
1355                                                     XATTR_NAME_LMA, &size);
1356                         if (!rc && unlikely(size < sizeof(*lma))) {
1357                                 rc = -EINVAL;
1358                         } else if (!rc) {
1359                                 lma->lma_incompat =
1360                                         le32_to_cpu(lma->lma_incompat);
1361
1362                                 if ((lma->lma_incompat &
1363                                      lustre_to_lma_flags(la->la_flags)) ==
1364                                     lustre_to_lma_flags(la->la_flags))
1365                                         /* save a useless xattr set if lma
1366                                          * incompat already has the flags
1367                                          */
1368                                         GOTO(lock, rc = 0);
1369
1370                                 lma->lma_incompat |=
1371                                         lustre_to_lma_flags(la->la_flags);
1372                                 lma->lma_incompat =
1373                                         cpu_to_le32(lma->lma_incompat);
1374                                 buf.lb_buf = lma;
1375                                 buf.lb_len = sizeof(*lma);
1376                                 rc = osd_xattr_set_internal(env, obj, &buf,
1377                                                             XATTR_NAME_LMA,
1378                                                             LU_XATTR_REPLACE,
1379                                                             oh);
1380                         }
1381                         if (rc < 0) {
1382                                 CWARN("%s: failed to set LMA flags: rc = %d\n",
1383                                        osd->od_svname, rc);
1384                                 GOTO(out, rc);
1385                         } else {
1386                                 obj->oo_lma_flags =
1387                                         la->la_flags & LUSTRE_LMA_FL_MASKS;
1388                         }
1389                 }
1390         }
1391
1392 lock:
1393         write_lock(&obj->oo_attr_lock);
1394         cnt = 0;
1395
1396         if (valid & LA_PROJID) {
1397 #ifdef ZFS_PROJINHERIT
1398                 if (osd->od_projectused_dn) {
1399                         if (obj->oo_with_projid) {
1400                                 osa->projid  = la->la_projid;
1401                                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PROJID(osd),
1402                                                  NULL, &osa->projid, 8);
1403                         } else {
1404                                 rc = osd_add_projid(env, obj, oh,
1405                                                     la->la_projid);
1406                                 if (unlikely(rc == -EEXIST)) {
1407                                         rc = 0;
1408                                 } else if (rc != 0) {
1409                                         write_unlock(&obj->oo_attr_lock);
1410                                         GOTO(out, rc);
1411                                 }
1412                                 obj->oo_with_projid = 1;
1413                         }
1414                         obj->oo_attr.la_projid = la->la_projid;
1415                 } else
1416 #endif
1417                         valid &= ~LA_PROJID;
1418         }
1419
1420         if (valid & LA_ATIME) {
1421                 osa->atime[0] = obj->oo_attr.la_atime = la->la_atime;
1422                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(osd), NULL,
1423                                  osa->atime, 16);
1424         }
1425         if (valid & LA_MTIME) {
1426                 osa->mtime[0] = obj->oo_attr.la_mtime = la->la_mtime;
1427                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(osd), NULL,
1428                                  osa->mtime, 16);
1429         }
1430         if (valid & LA_CTIME) {
1431                 osa->ctime[0] = obj->oo_attr.la_ctime = la->la_ctime;
1432                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(osd), NULL,
1433                                  osa->ctime, 16);
1434         }
1435         if (valid & LA_MODE) {
1436                 /* mode is stored along with type, so read it first */
1437                 obj->oo_attr.la_mode = (obj->oo_attr.la_mode & S_IFMT) |
1438                         (la->la_mode & ~S_IFMT);
1439                 osa->mode = obj->oo_attr.la_mode;
1440                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(osd), NULL,
1441                                  &osa->mode, 8);
1442         }
1443         if (valid & LA_SIZE) {
1444                 osa->size = obj->oo_attr.la_size = la->la_size;
1445                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(osd), NULL,
1446                                  &osa->size, 8);
1447         }
1448         if (valid & LA_NLINK) {
1449                 osa->nlink = obj->oo_attr.la_nlink = la->la_nlink;
1450                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(osd), NULL,
1451                                  &osa->nlink, 8);
1452         }
1453         if (valid & LA_RDEV) {
1454                 osa->rdev = obj->oo_attr.la_rdev = la->la_rdev;
1455                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(osd), NULL,
1456                                  &osa->rdev, 8);
1457         }
1458         if (valid & LA_FLAGS) {
1459                 osa->flags = attrs_fs2zfs(la->la_flags);
1460                 /* many flags not supported by zfs, ensure good cached copy */
1461                 obj->oo_attr.la_flags = attrs_zfs2fs(osa->flags);
1462 #ifdef ZFS_PROJINHERIT
1463                 if (obj->oo_with_projid && osd->od_projectused_dn)
1464                         osa->flags |= ZFS_PROJID;
1465 #endif
1466                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(osd), NULL,
1467                                  &osa->flags, 8);
1468         }
1469         if (valid & LA_UID) {
1470                 osa->uid = obj->oo_attr.la_uid = la->la_uid;
1471                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(osd), NULL,
1472                                  &osa->uid, 8);
1473         }
1474         if (valid & LA_GID) {
1475                 osa->gid = obj->oo_attr.la_gid = la->la_gid;
1476                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(osd), NULL,
1477                                  &osa->gid, 8);
1478         }
1479         obj->oo_attr.la_valid |= valid;
1480         write_unlock(&obj->oo_attr_lock);
1481
1482         LASSERT(cnt <= ARRAY_SIZE(osd_oti_get(env)->oti_attr_bulk));
1483         rc = osd_object_sa_bulk_update(obj, bulk, cnt, oh);
1484
1485 out:
1486         up_read(&obj->oo_guard);
1487         RETURN(rc);
1488 }
1489
1490 /*
1491  * Object creation.
1492  *
1493  * XXX temporary solution.
1494  */
1495 static void osd_ah_init(const struct lu_env *env, struct dt_allocation_hint *ah,
1496                         struct dt_object *parent, struct dt_object *child,
1497                         umode_t child_mode)
1498 {
1499         LASSERT(ah);
1500
1501         ah->dah_parent = parent;
1502
1503         if (parent != NULL && !dt_object_remote(parent)) {
1504                 /* will help to find FID->ino at dt_insert("..") */
1505                 struct osd_object *pobj = osd_dt_obj(parent);
1506
1507                 osd_idc_find_and_init(env, osd_obj2dev(pobj), pobj);
1508         }
1509 }
1510
1511 static int osd_declare_create(const struct lu_env *env, struct dt_object *dt,
1512                               struct lu_attr *attr,
1513                               struct dt_allocation_hint *hint,
1514                               struct dt_object_format *dof,
1515                               struct thandle *handle)
1516 {
1517         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1518         struct osd_object       *obj = osd_dt_obj(dt);
1519         struct osd_device       *osd = osd_obj2dev(obj);
1520         struct osd_thandle      *oh;
1521         uint64_t                 zapid;
1522         dnode_t                 *dn;
1523         int                      rc, dnode_size;
1524
1525         ENTRY;
1526         LASSERT(dof);
1527
1528         switch (dof->dof_type) {
1529         case DFT_REGULAR:
1530         case DFT_SYM:
1531         case DFT_NODE:
1532                 if (obj->oo_dt.do_body_ops == NULL)
1533                         obj->oo_dt.do_body_ops = &osd_body_ops;
1534                 break;
1535         default:
1536                 break;
1537         }
1538
1539         LASSERT(handle != NULL);
1540         oh = container_of(handle, struct osd_thandle, ot_super);
1541         LASSERT(oh->ot_tx != NULL);
1542
1543         /* this is the minimum set of EAs on every Lustre object */
1544         obj->oo_ea_in_bonus = OSD_BASE_EA_IN_BONUS;
1545         /* reserve 32 bytes for extra stuff like ACLs */
1546         dnode_size = size_roundup_power2(obj->oo_ea_in_bonus + 32);
1547
1548         switch (dof->dof_type) {
1549         case DFT_DIR:
1550                 dt->do_index_ops = &osd_dir_ops;
1551                 fallthrough;
1552         case DFT_INDEX:
1553                 /* for zap create */
1554                 dmu_tx_hold_zap(oh->ot_tx, DMU_NEW_OBJECT, FALSE, NULL);
1555                 dmu_tx_hold_sa_create(oh->ot_tx, dnode_size);
1556                 break;
1557         case DFT_REGULAR:
1558         case DFT_SYM:
1559         case DFT_NODE:
1560                 /* first, we'll create new object */
1561                 dmu_tx_hold_sa_create(oh->ot_tx, dnode_size);
1562                 break;
1563
1564         default:
1565                 LBUG();
1566                 break;
1567         }
1568
1569         /* and we'll add it to some mapping */
1570         zapid = osd_get_name_n_idx(env, osd, fid, NULL, 0, &dn);
1571         osd_tx_hold_zap(oh->ot_tx, zapid, dn, TRUE, NULL);
1572
1573         /* will help to find FID->ino mapping at dt_insert() */
1574         osd_idc_find_and_init(env, osd, obj);
1575
1576         rc = osd_declare_quota(env, osd, attr->la_uid, attr->la_gid,
1577                                attr->la_projid, 1, oh, NULL, OSD_QID_INODE);
1578
1579         RETURN(rc);
1580 }
1581
1582 int __osd_attr_init(const struct lu_env *env, struct osd_device *osd,
1583                     struct osd_object *obj, sa_handle_t *sa_hdl, dmu_tx_t *tx,
1584                     struct lu_attr *la, uint64_t parent,
1585                     nvlist_t *xattr)
1586 {
1587         sa_bulk_attr_t *bulk = osd_oti_get(env)->oti_attr_bulk;
1588         struct osa_attr *osa = &osd_oti_get(env)->oti_osa;
1589         uint64_t gen;
1590         inode_timespec_t now;
1591         int cnt;
1592         int rc;
1593         char *dxattr = NULL;
1594         size_t sa_size;
1595
1596
1597         LASSERT(sa_hdl);
1598
1599         gen = dmu_tx_get_txg(tx);
1600         gethrestime(&now);
1601         ZFS_TIME_ENCODE(&now, osa->btime);
1602
1603         osa->atime[0] = la->la_atime;
1604         osa->ctime[0] = la->la_ctime;
1605         osa->mtime[0] = la->la_mtime;
1606         osa->mode = la->la_mode;
1607         osa->uid = la->la_uid;
1608         osa->gid = la->la_gid;
1609         osa->rdev = la->la_rdev;
1610         osa->nlink = la->la_nlink;
1611         if (la->la_valid & LA_FLAGS)
1612                 osa->flags = attrs_fs2zfs(la->la_flags);
1613         else
1614                 osa->flags = 0;
1615         osa->size  = la->la_size;
1616 #ifdef ZFS_PROJINHERIT
1617         if (osd->od_projectused_dn) {
1618                 if (la->la_valid & LA_PROJID)
1619                         osa->projid = la->la_projid;
1620                 else
1621                         osa->projid = ZFS_DEFAULT_PROJID;
1622                 osa->flags |= ZFS_PROJID;
1623                 if (obj)
1624                         obj->oo_with_projid = 1;
1625         } else {
1626                 osa->flags &= ~ZFS_PROJID;
1627         }
1628 #endif
1629
1630         /*
1631          * we need to create all SA below upon object create.
1632          *
1633          * XXX The attribute order matters since the accounting callback relies
1634          * on static offsets (i.e. SA_*_OFFSET, see zfs_space_delta_cb()) to
1635          * look up the UID/GID/PROJID attributes. Moreover, the callback does
1636          * not seem to support the spill block.
1637          * We define attributes in the same order as SA_*_OFFSET in order to
1638          * work around the problem. See ORI-610.
1639          */
1640         cnt = 0;
1641         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(osd), NULL, &osa->mode, 8);
1642         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(osd), NULL, &osa->size, 8);
1643         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GEN(osd), NULL, &gen, 8);
1644         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(osd), NULL, &osa->uid, 8);
1645         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(osd), NULL, &osa->gid, 8);
1646         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PARENT(osd), NULL, &parent, 8);
1647         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(osd), NULL, &osa->flags, 8);
1648         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(osd), NULL, osa->atime, 16);
1649         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(osd), NULL, osa->mtime, 16);
1650         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(osd), NULL, osa->ctime, 16);
1651         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CRTIME(osd), NULL, osa->btime, 16);
1652         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(osd), NULL, &osa->nlink, 8);
1653 #ifdef ZFS_PROJINHERIT
1654         if (osd->od_projectused_dn)
1655                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PROJID(osd), NULL,
1656                                  &osa->projid, 8);
1657 #endif
1658         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(osd), NULL, &osa->rdev, 8);
1659         LASSERT(cnt <= ARRAY_SIZE(osd_oti_get(env)->oti_attr_bulk));
1660
1661         if (xattr) {
1662                 rc = -nvlist_size(xattr, &sa_size, NV_ENCODE_XDR);
1663                 LASSERT(rc == 0);
1664
1665                 dxattr = zio_buf_alloc(sa_size);
1666                 LASSERT(dxattr);
1667
1668                 rc = -nvlist_pack(xattr, &dxattr, &sa_size,
1669                                 NV_ENCODE_XDR, KM_SLEEP);
1670                 LASSERT(rc == 0);
1671
1672                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_DXATTR(osd),
1673                                 NULL, dxattr, sa_size);
1674         }
1675
1676         rc = -sa_replace_all_by_template(sa_hdl, bulk, cnt, tx);
1677         if (dxattr)
1678                 zio_buf_free(dxattr, sa_size);
1679
1680         return rc;
1681 }
1682
1683 int osd_find_new_dnode(const struct lu_env *env, dmu_tx_t *tx,
1684                        uint64_t oid, dnode_t **dnp)
1685 {
1686         dmu_tx_hold_t *txh;
1687         int rc = 0;
1688
1689         /* take dnode_t from tx to save on dnode#->dnode_t lookup */
1690         for (txh = list_tail(&tx->tx_holds); txh;
1691              txh = list_prev(&tx->tx_holds, txh)) {
1692                 dnode_t *dn = txh->txh_dnode;
1693                 dmu_buf_impl_t *db;
1694
1695                 if (dn == NULL)
1696                         continue;
1697                 if (dn->dn_object != oid)
1698                         continue;
1699                 db = dn->dn_bonus;
1700                 if (db == NULL) {
1701                         rw_enter(&dn->dn_struct_rwlock, RW_WRITER);
1702                         if (dn->dn_bonus == NULL)
1703                                 dbuf_create_bonus(dn);
1704                         rw_exit(&dn->dn_struct_rwlock);
1705                 }
1706                 db = dn->dn_bonus;
1707                 LASSERT(db);
1708                 LASSERT(dn->dn_handle);
1709                 DB_DNODE_ENTER(db);
1710                 if (zfs_refcount_add(&db->db_holds, osd_obj_tag) == 1) {
1711                         zfs_refcount_add(&dn->dn_holds, osd_obj_tag);
1712                         atomic_inc_32(&dn->dn_dbufs_count);
1713                 }
1714                 *dnp = dn;
1715                 DB_DNODE_EXIT(db);
1716                 dbuf_read(db, NULL, DB_RF_MUST_SUCCEED | DB_RF_NOPREFETCH);
1717                 break;
1718         }
1719
1720         if (unlikely(*dnp == NULL))
1721                 rc = __osd_obj2dnode(tx->tx_objset, oid, dnp);
1722
1723         return rc;
1724 }
1725
1726 int osd_find_dnsize(struct osd_device *osd, int ea_in_bonus)
1727 {
1728         int dnsize;
1729
1730         if (osd->od_dnsize == ZFS_DNSIZE_AUTO) {
1731                 dnsize = DNODE_MIN_SIZE;
1732                 do {
1733                         if (DN_BONUS_SIZE(dnsize) >= ea_in_bonus + 32)
1734                                 break;
1735                         dnsize <<= 1;
1736                 } while (dnsize < DNODE_MAX_SIZE);
1737                 if (dnsize > DNODE_MAX_SIZE)
1738                         dnsize = DNODE_MAX_SIZE;
1739         } else if (osd->od_dnsize == ZFS_DNSIZE_1K) {
1740                 dnsize = 1024;
1741         } else if (osd->od_dnsize == ZFS_DNSIZE_2K) {
1742                 dnsize = 2048;
1743         } else if (osd->od_dnsize == ZFS_DNSIZE_4K) {
1744                 dnsize = 4096;
1745         } else if (osd->od_dnsize == ZFS_DNSIZE_8K) {
1746                 dnsize = 8192;
1747         } else if (osd->od_dnsize == ZFS_DNSIZE_16K) {
1748                 dnsize = 16384;
1749         } else {
1750                 dnsize = DNODE_MIN_SIZE;
1751         }
1752         return dnsize;
1753 }
1754
1755 /*
1756  * The transaction passed to this routine must have
1757  * dmu_tx_hold_bonus(tx, DMU_NEW_OBJECT) called and then assigned
1758  * to a transaction group.
1759  */
1760 int __osd_object_create(const struct lu_env *env, struct osd_device *osd,
1761                         struct osd_object *obj, const struct lu_fid *fid,
1762                         dnode_t **dnp, dmu_tx_t *tx, struct lu_attr *la)
1763 {
1764         dmu_object_type_t type = DMU_OT_PLAIN_FILE_CONTENTS;
1765         uint64_t oid;
1766         int size;
1767
1768         /* Use DMU_OTN_UINT8_METADATA for local objects so their data blocks
1769          * would get an additional ditto copy
1770          */
1771         if (unlikely(S_ISREG(la->la_mode) &&
1772                      fid_seq_is_local_file(fid_seq(fid))))
1773                 type = DMU_OTN_UINT8_METADATA;
1774
1775         /* Create a new DMU object using the default dnode size. */
1776         if (obj)
1777                 size = obj->oo_ea_in_bonus;
1778         else
1779                 size = OSD_BASE_EA_IN_BONUS;
1780         oid = osd_dmu_object_alloc(osd->od_os, type, 0,
1781                                    osd_find_dnsize(osd, size), tx);
1782
1783         LASSERT(la->la_valid & LA_MODE);
1784         la->la_size = 0;
1785         la->la_nlink = 1;
1786
1787         return osd_find_new_dnode(env, tx, oid, dnp);
1788 }
1789
1790 /*
1791  * The transaction passed to this routine must have
1792  * dmu_tx_hold_zap(tx, DMU_NEW_OBJECT, ...) called and then assigned
1793  * to a transaction group.
1794  *
1795  * Using ZAP_FLAG_HASH64 will force the ZAP to always be a FAT ZAP.
1796  * This is fine for directories today, because storing the FID in the dirent
1797  * will also require a FAT ZAP.  If there is a new type of micro ZAP created
1798  * then we might need to re-evaluate the use of this flag and instead do
1799  * a conversion from the different internal ZAP hash formats being used.
1800  */
1801 int __osd_zap_create(const struct lu_env *env, struct osd_device *osd,
1802                      dnode_t **dnp, dmu_tx_t *tx, struct lu_attr *la,
1803                      unsigned int dnsize, zap_flags_t flags)
1804 {
1805         uint64_t oid;
1806
1807         /* Assert that the transaction has been assigned to a transaction grp */
1808         LASSERT(tx->tx_txg != 0);
1809         *dnp = NULL;
1810
1811         oid = osd_zap_create_flags(osd->od_os, 0, flags | ZAP_FLAG_HASH64,
1812                                    DMU_OT_DIRECTORY_CONTENTS,
1813                                    14, /* == ZFS fzap_default_blockshift */
1814                                    DN_MAX_INDBLKSHIFT, /* indirect blockshift */
1815                                    dnsize, tx);
1816
1817         la->la_size = 2;
1818         la->la_nlink = 1;
1819
1820         return osd_find_new_dnode(env, tx, oid, dnp);
1821 }
1822
1823 static dnode_t *osd_mkidx(const struct lu_env *env, struct osd_object *obj,
1824                           struct lu_attr *la, struct osd_thandle *oh)
1825 {
1826         struct osd_device *osd = osd_obj2dev(obj);
1827         dnode_t *dn;
1828         int rc;
1829
1830         /* Index file should be created as regular file in order not to confuse
1831          * ZPL which could interpret them as directory.
1832          * We set ZAP_FLAG_UINT64_KEY to let ZFS know than we are going to use
1833          * binary keys
1834          */
1835         LASSERT(S_ISREG(la->la_mode));
1836         rc = __osd_zap_create(env, osd, &dn, oh->ot_tx, la,
1837                 osd_find_dnsize(osd, obj->oo_ea_in_bonus), ZAP_FLAG_UINT64_KEY);
1838         if (rc)
1839                 return ERR_PTR(rc);
1840         return dn;
1841 }
1842
1843 static dnode_t *osd_mkdir(const struct lu_env *env, struct osd_object *obj,
1844                           struct lu_attr *la, struct osd_thandle *oh)
1845 {
1846         struct osd_device *osd = osd_obj2dev(obj);
1847         dnode_t *dn;
1848         int rc;
1849
1850         LASSERT(S_ISDIR(la->la_mode));
1851         rc = __osd_zap_create(env, osd, &dn, oh->ot_tx, la,
1852                               osd_find_dnsize(osd, obj->oo_ea_in_bonus), 0);
1853         if (rc)
1854                 return ERR_PTR(rc);
1855         return dn;
1856 }
1857
1858 static dnode_t *osd_mkreg(const struct lu_env *env, struct osd_object *obj,
1859                           struct lu_attr *la, struct osd_thandle *oh)
1860 {
1861         const struct lu_fid *fid = lu_object_fid(&obj->oo_dt.do_lu);
1862         struct osd_device *osd = osd_obj2dev(obj);
1863         dnode_t *dn;
1864         int rc;
1865
1866         LASSERT(S_ISREG(la->la_mode));
1867         rc = __osd_object_create(env, osd, obj, fid, &dn, oh->ot_tx, la);
1868         if (rc)
1869                 return ERR_PTR(rc);
1870
1871         if ((fid_is_idif(fid) || fid_is_norm(fid) || fid_is_echo(fid))) {
1872                 /* The minimum block size must be at least page size otherwise
1873                  * it will break the assumption in tgt_thread_big_cache where
1874                  * the array size is PTLRPC_MAX_BRW_PAGES. It will also affect
1875                  * RDMA due to subpage transfer size
1876                  */
1877                 rc = -dmu_object_set_blocksize(osd->od_os, dn->dn_object,
1878                                                PAGE_SIZE, 0, oh->ot_tx);
1879                 if (unlikely(rc)) {
1880                         CERROR("%s: can't change blocksize: %d\n",
1881                                osd->od_svname, rc);
1882                         return ERR_PTR(rc);
1883                 }
1884         } else if ((fid_is_llog(fid))) {
1885                 rc = -dmu_object_set_blocksize(osd->od_os, dn->dn_object,
1886                                                LLOG_MIN_CHUNK_SIZE, 0,
1887                                                oh->ot_tx);
1888                 if (unlikely(rc)) {
1889                         CERROR("%s: can't change blocksize: %d\n",
1890                                osd->od_svname, rc);
1891                         return ERR_PTR(rc);
1892                 }
1893         }
1894
1895         return dn;
1896 }
1897
1898 static dnode_t *osd_mksym(const struct lu_env *env, struct osd_object *obj,
1899                           struct lu_attr *la, struct osd_thandle *oh)
1900 {
1901         dnode_t *dn;
1902         int rc;
1903
1904         LASSERT(S_ISLNK(la->la_mode));
1905         rc = __osd_object_create(env, osd_obj2dev(obj), obj,
1906                                  lu_object_fid(&obj->oo_dt.do_lu),
1907                                  &dn, oh->ot_tx, la);
1908         if (rc)
1909                 return ERR_PTR(rc);
1910         return dn;
1911 }
1912
1913 static dnode_t *osd_mknod(const struct lu_env *env, struct osd_object *obj,
1914                           struct lu_attr *la, struct osd_thandle *oh)
1915 {
1916         dnode_t *dn;
1917         int rc;
1918
1919         if (S_ISCHR(la->la_mode) || S_ISBLK(la->la_mode))
1920                 la->la_valid |= LA_RDEV;
1921
1922         rc = __osd_object_create(env, osd_obj2dev(obj), obj,
1923                                  lu_object_fid(&obj->oo_dt.do_lu),
1924                                  &dn, oh->ot_tx, la);
1925         if (rc)
1926                 return ERR_PTR(rc);
1927         return dn;
1928 }
1929
1930 typedef dnode_t *(*osd_obj_type_f)(const struct lu_env *env,
1931                                    struct osd_object *obj,
1932                                    struct lu_attr *la,
1933                                    struct osd_thandle *oh);
1934
1935 static osd_obj_type_f osd_create_type_f(enum dt_format_type type)
1936 {
1937         osd_obj_type_f result;
1938
1939         switch (type) {
1940         case DFT_DIR:
1941                 result = osd_mkdir;
1942                 break;
1943         case DFT_INDEX:
1944                 result = osd_mkidx;
1945                 break;
1946         case DFT_REGULAR:
1947                 result = osd_mkreg;
1948                 break;
1949         case DFT_SYM:
1950                 result = osd_mksym;
1951                 break;
1952         case DFT_NODE:
1953                 result = osd_mknod;
1954                 break;
1955         default:
1956                 LBUG();
1957                 break;
1958         }
1959         return result;
1960 }
1961
1962 /*
1963  * Concurrency: @dt is write locked.
1964  */
1965 static int osd_create(const struct lu_env *env, struct dt_object *dt,
1966                       struct lu_attr *attr, struct dt_allocation_hint *hint,
1967                       struct dt_object_format *dof, struct thandle *th)
1968 {
1969         struct osd_thread_info  *info = osd_oti_get(env);
1970         struct lustre_mdt_attrs *lma = &info->oti_mdt_attrs;
1971         struct zpl_direntry     *zde = &info->oti_zde.lzd_reg;
1972         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1973         struct osd_object       *obj = osd_dt_obj(dt);
1974         struct osd_device       *osd = osd_obj2dev(obj);
1975         char                    *buf = info->oti_str;
1976         struct osd_thandle      *oh;
1977         dnode_t *dn = NULL, *zdn = NULL;
1978         uint64_t                 zapid, parent = 0;
1979         int                      rc;
1980         __u32 compat = 0;
1981
1982         ENTRY;
1983         LASSERT(!fid_is_acct(fid));
1984
1985         /* concurrent create declarations should not see
1986          * the object inconsistent (db, attr, etc).
1987          * in regular cases acquisition should be cheap
1988          */
1989         down_write(&obj->oo_guard);
1990
1991         if (unlikely(dt_object_exists(dt)))
1992                 GOTO(out, rc = -EEXIST);
1993
1994         LASSERT(osd_invariant(obj));
1995         LASSERT(dof != NULL);
1996
1997         LASSERT(th != NULL);
1998         oh = container_of(th, struct osd_thandle, ot_super);
1999
2000         LASSERT(obj->oo_dn == NULL);
2001
2002         /* to follow ZFS on-disk format we need to initialize parent dnode
2003          * properly
2004          */
2005         if (hint != NULL && hint->dah_parent != NULL &&
2006             !dt_object_remote(hint->dah_parent))
2007                 parent = osd_dt_obj(hint->dah_parent)->oo_dn->dn_object;
2008
2009         /* we may fix some attributes, better do not change the source */
2010         obj->oo_attr = *attr;
2011         obj->oo_attr.la_size = 0;
2012         obj->oo_attr.la_nlink = 0;
2013         obj->oo_attr.la_valid |= LA_SIZE | LA_NLINK | LA_TYPE;
2014         if (!(obj->oo_attr.la_valid & LA_FLAGS))
2015                 obj->oo_attr.la_flags = 0;
2016
2017 #ifdef ZFS_PROJINHERIT
2018         if (osd->od_projectused_dn) {
2019                 if (!(obj->oo_attr.la_valid & LA_PROJID))
2020                         obj->oo_attr.la_projid = ZFS_DEFAULT_PROJID;
2021                 obj->oo_with_projid = 1;
2022         }
2023 #endif
2024
2025         dn = osd_create_type_f(dof->dof_type)(env, obj, &obj->oo_attr, oh);
2026         if (IS_ERR(dn)) {
2027                 rc = PTR_ERR(dn);
2028                 dn = NULL;
2029                 GOTO(out, rc);
2030         }
2031
2032         zde->zde_pad = 0;
2033         zde->zde_dnode = dn->dn_object;
2034         zde->zde_type = S_DT(attr->la_mode & S_IFMT);
2035
2036         zapid = osd_get_name_n_idx(env, osd, fid, buf,
2037                                    sizeof(info->oti_str), &zdn);
2038         if (CFS_FAIL_CHECK(OBD_FAIL_OSD_NO_OI_ENTRY) ||
2039             (osd->od_is_ost && CFS_FAIL_CHECK(OBD_FAIL_OSD_COMPAT_NO_ENTRY)))
2040                 goto skip_add;
2041
2042         if (osd->od_is_ost && CFS_FAIL_CHECK(OBD_FAIL_OSD_COMPAT_INVALID_ENTRY))
2043                 zde->zde_dnode++;
2044
2045         rc = osd_zap_add(osd, zapid, zdn, buf, 8, 1, zde, oh->ot_tx);
2046         if (rc)
2047                 GOTO(out, rc);
2048
2049 skip_add:
2050         obj->oo_dn = dn;
2051         /* Now add in all of the "SA" attributes */
2052         rc = osd_sa_handle_get(obj);
2053         if (rc)
2054                 GOTO(out, rc);
2055
2056         rc = -nvlist_alloc(&obj->oo_sa_xattr, NV_UNIQUE_NAME, KM_SLEEP);
2057         if (rc)
2058                 GOTO(out, rc);
2059
2060         /* initialize LMA */
2061         if (fid_is_idif(fid) || (fid_is_norm(fid) && osd->od_is_ost))
2062                 compat |= LMAC_FID_ON_OST;
2063         lustre_lma_init(lma, fid, compat, 0);
2064         lustre_lma_swab(lma);
2065         rc = -nvlist_add_byte_array(obj->oo_sa_xattr, XATTR_NAME_LMA,
2066                                     (uchar_t *)lma, sizeof(*lma));
2067         if (rc)
2068                 GOTO(out, rc);
2069
2070         /* configure new osd object */
2071         obj->oo_parent = parent != 0 ? parent : zapid;
2072         obj->oo_late_attr_set = 1;
2073         rc = __osd_sa_xattr_schedule_update(env, obj, oh);
2074         if (rc)
2075                 GOTO(out, rc);
2076
2077         /* XXX: oo_lma_flags */
2078         obj->oo_dt.do_lu.lo_header->loh_attr |= obj->oo_attr.la_mode & S_IFMT;
2079         if (likely(!fid_is_acct(lu_object_fid(&obj->oo_dt.do_lu))))
2080                 /* no body operations for accounting objects */
2081                 obj->oo_dt.do_body_ops = &osd_body_ops;
2082
2083         osd_idc_find_and_init(env, osd, obj);
2084
2085 out:
2086         if (unlikely(rc && dn)) {
2087                 dmu_object_free(osd->od_os, dn->dn_object, oh->ot_tx);
2088                 osd_dnode_rele(dn);
2089                 obj->oo_dn = NULL;
2090         } else if (!rc) {
2091                 obj->oo_dt.do_lu.lo_header->loh_attr |= LOHA_EXISTS;
2092         }
2093         up_write(&obj->oo_guard);
2094         RETURN(rc);
2095 }
2096
2097 static int osd_declare_ref_add(const struct lu_env *env, struct dt_object *dt,
2098                                struct thandle *th)
2099 {
2100         osd_idc_find_and_init(env, osd_dev(dt->do_lu.lo_dev), osd_dt_obj(dt));
2101         return osd_declare_attr_set(env, dt, NULL, th);
2102 }
2103
2104 /*
2105  * Concurrency: @dt is write locked.
2106  */
2107 static int osd_ref_add(const struct lu_env *env, struct dt_object *dt,
2108                        struct thandle *handle)
2109 {
2110         struct osd_object       *obj = osd_dt_obj(dt);
2111         struct osd_thandle      *oh;
2112         struct osd_device       *osd = osd_obj2dev(obj);
2113         uint64_t                 nlink;
2114         int rc;
2115
2116         ENTRY;
2117         down_read(&obj->oo_guard);
2118         if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
2119                 GOTO(out, rc = -ENOENT);
2120
2121         LASSERT(osd_invariant(obj));
2122         LASSERT(obj->oo_sa_hdl != NULL);
2123
2124         oh = container_of(handle, struct osd_thandle, ot_super);
2125
2126         write_lock(&obj->oo_attr_lock);
2127         nlink = ++obj->oo_attr.la_nlink;
2128         write_unlock(&obj->oo_attr_lock);
2129
2130         rc = osd_object_sa_update(obj, SA_ZPL_LINKS(osd), &nlink, 8, oh);
2131
2132 out:
2133         up_read(&obj->oo_guard);
2134         RETURN(rc);
2135 }
2136
2137 static int osd_declare_ref_del(const struct lu_env *env, struct dt_object *dt,
2138                                struct thandle *handle)
2139 {
2140         osd_idc_find_and_init(env, osd_dev(dt->do_lu.lo_dev), osd_dt_obj(dt));
2141         return osd_declare_attr_set(env, dt, NULL, handle);
2142 }
2143
2144 /*
2145  * Concurrency: @dt is write locked.
2146  */
2147 static int osd_ref_del(const struct lu_env *env, struct dt_object *dt,
2148                        struct thandle *handle)
2149 {
2150         struct osd_object       *obj = osd_dt_obj(dt);
2151         struct osd_thandle      *oh;
2152         struct osd_device       *osd = osd_obj2dev(obj);
2153         uint64_t                 nlink;
2154         int                      rc;
2155
2156         ENTRY;
2157         down_read(&obj->oo_guard);
2158
2159         if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))
2160                 GOTO(out, rc = -ENOENT);
2161
2162         LASSERT(osd_invariant(obj));
2163         LASSERT(obj->oo_sa_hdl != NULL);
2164
2165         oh = container_of(handle, struct osd_thandle, ot_super);
2166         LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));
2167
2168         write_lock(&obj->oo_attr_lock);
2169         nlink = --obj->oo_attr.la_nlink;
2170         write_unlock(&obj->oo_attr_lock);
2171
2172         rc = osd_object_sa_update(obj, SA_ZPL_LINKS(osd), &nlink, 8, oh);
2173
2174 out:
2175         up_read(&obj->oo_guard);
2176         RETURN(rc);
2177 }
2178
2179 static int osd_object_sync(const struct lu_env *env, struct dt_object *dt,
2180                            __u64 start, __u64 end)
2181 {
2182         struct osd_device *osd = osd_obj2dev(osd_dt_obj(dt));
2183         uint64_t txg = 0;
2184
2185         ENTRY;
2186         if (osd->od_dt_dev.dd_rdonly)
2187                 RETURN(0);
2188
2189         txg = osd_db_dirty_txg(osd_dt_obj(dt)->oo_dn->dn_dbuf);
2190         if (txg) {
2191                 /* the object is dirty or being synced */
2192                 if (osd_object_sync_delay_us < 0)
2193                         txg_wait_synced(dmu_objset_pool(osd->od_os), txg);
2194                 else
2195                         udelay(osd_object_sync_delay_us);
2196         }
2197
2198         RETURN(0);
2199 }
2200
2201 static int osd_invalidate(const struct lu_env *env, struct dt_object *dt)
2202 {
2203         return 0;
2204 }
2205
2206 static bool osd_check_stale(struct dt_object *dt)
2207 {
2208         return false;
2209 }
2210
2211 static const struct dt_object_operations osd_obj_ops = {
2212         .do_read_lock           = osd_read_lock,
2213         .do_write_lock          = osd_write_lock,
2214         .do_read_unlock         = osd_read_unlock,
2215         .do_write_unlock        = osd_write_unlock,
2216         .do_write_locked        = osd_write_locked,
2217         .do_attr_get            = osd_attr_get,
2218         .do_declare_attr_set    = osd_declare_attr_set,
2219         .do_attr_set            = osd_attr_set,
2220         .do_ah_init             = osd_ah_init,
2221         .do_declare_create      = osd_declare_create,
2222         .do_create              = osd_create,
2223         .do_declare_destroy     = osd_declare_destroy,
2224         .do_destroy             = osd_destroy,
2225         .do_index_try           = osd_index_try,
2226         .do_declare_ref_add     = osd_declare_ref_add,
2227         .do_ref_add             = osd_ref_add,
2228         .do_declare_ref_del     = osd_declare_ref_del,
2229         .do_ref_del             = osd_ref_del,
2230         .do_xattr_get           = osd_xattr_get,
2231         .do_declare_xattr_set   = osd_declare_xattr_set,
2232         .do_xattr_set           = osd_xattr_set,
2233         .do_declare_xattr_del   = osd_declare_xattr_del,
2234         .do_xattr_del           = osd_xattr_del,
2235         .do_xattr_list          = osd_xattr_list,
2236         .do_object_sync         = osd_object_sync,
2237         .do_invalidate          = osd_invalidate,
2238         .do_check_stale         = osd_check_stale,
2239 };
2240
2241 static const struct lu_object_operations osd_lu_obj_ops = {
2242         .loo_object_init        = osd_object_init,
2243         .loo_object_delete      = osd_object_delete,
2244         .loo_object_release     = osd_object_release,
2245         .loo_object_free        = osd_object_free,
2246         .loo_object_print       = osd_object_print,
2247         .loo_object_invariant   = osd_object_invariant,
2248 };
2249
2250 static int osd_otable_it_attr_get(const struct lu_env *env,
2251                                 struct dt_object *dt,
2252                                 struct lu_attr *attr)
2253 {
2254         attr->la_valid = 0;
2255         return 0;
2256 }
2257
2258 static const struct dt_object_operations osd_obj_otable_it_ops = {
2259         .do_attr_get            = osd_otable_it_attr_get,
2260         .do_index_try           = osd_index_try,
2261 };
2262
2263 module_param(osd_object_sync_delay_us, int, 0644);
2264 MODULE_PARM_DESC(osd_object_sync_delay_us,
2265                  "If zero or larger delay N usec instead of doing object sync");