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.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) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  */
  30 /*
  31  * Copyright (c) 2011, 2012 Whamcloud, Inc.
  32  * Use is subject to license terms.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  *
  38  * lustre/osd-zfs/osd_object.c
  39  *
  40  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
  41  * Author: Mike Pershin <tappro@whamcloud.com>
  42  * Author: Johann Lombardi <johann@whamcloud.com>
  43  */
  44
  45 #ifndef EXPORT_SYMTAB
  46 # define EXPORT_SYMTAB
  47 #endif
  48 #define DEBUG_SUBSYSTEM S_OSD
  49
  50 #include <lustre_ver.h>
  51 #include <libcfs/libcfs.h>
  52 #include <lustre_fsfilt.h>
  53 #include <obd_support.h>
  54 #include <lustre_net.h>
  55 #include <obd.h>
  56 #include <obd_class.h>
  57 #include <lustre_disk.h>
  58 #include <lustre_fid.h>
  59
  60 #include "osd_internal.h"
  61
  62 #include <sys/dnode.h>
  63 #include <sys/dbuf.h>
  64 #include <sys/spa.h>
  65 #include <sys/stat.h>
  66 #include <sys/zap.h>
  67 #include <sys/spa_impl.h>
  68 #include <sys/zfs_znode.h>
  69 #include <sys/dmu_tx.h>
  70 #include <sys/dmu_objset.h>
  71 #include <sys/dsl_prop.h>
  72 #include <sys/sa_impl.h>
  73 #include <sys/txg.h>
  74
  75 static char *osd_obj_tag = "osd_object";
  76
  77 static struct dt_object_operations osd_obj_ops;
  78 static struct lu_object_operations osd_lu_obj_ops;
  79 extern struct dt_body_operations osd_body_ops;
  80 static struct dt_object_operations osd_obj_otable_it_ops;
  81
  82 extern cfs_mem_cache_t *osd_object_kmem;
  83
  84 static void
  85 osd_object_sa_fini(struct osd_object *obj)
  86 {
  87         if (obj->oo_sa_hdl) {
  88                 sa_handle_destroy(obj->oo_sa_hdl);
  89                 obj->oo_sa_hdl = NULL;
  90         }
  91 }
  92
  93 static int
  94 osd_object_sa_init(struct osd_object *obj, udmu_objset_t *uos)
  95 {
  96         int rc;
  97
  98         LASSERT(obj->oo_sa_hdl == NULL);
  99         LASSERT(obj->oo_db != NULL);
 100
 101         rc = -sa_handle_get(uos->os, obj->oo_db->db_object, obj,
 102                             SA_HDL_PRIVATE, &obj->oo_sa_hdl);
 103         if (rc)
 104                 return rc;
 105
 106         /* Cache the xattr object id, valid for the life of the object */
 107         rc = -sa_lookup(obj->oo_sa_hdl, SA_ZPL_XATTR(uos), &obj->oo_xattr, 8);
 108         if (rc == -ENOENT) {
 109                 obj->oo_xattr = ZFS_NO_OBJECT;
 110                 rc = 0;
 111         } else if (rc) {
 112                 osd_object_sa_fini(obj);
 113         }
 114
 115         return rc;
 116 }
 117
 118 /*
 119  * Add object to list of dirty objects in tx handle.
 120  */
 121 static void
 122 osd_object_sa_dirty_add(struct osd_object *obj, struct osd_thandle *oh)
 123 {
 124         if (!cfs_list_empty(&obj->oo_sa_linkage))
 125                 return;
 126
 127         cfs_down(&oh->ot_sa_lock);
 128         if (likely(cfs_list_empty(&obj->oo_sa_linkage)))
 129                 cfs_list_add(&obj->oo_sa_linkage, &oh->ot_sa_list);
 130         cfs_up(&oh->ot_sa_lock);
 131 }
 132
 133 /*
 134  * Release spill block dbuf hold for all dirty SAs.
 135  */
 136 void osd_object_sa_dirty_rele(struct osd_thandle *oh)
 137 {
 138         struct osd_object *obj;
 139
 140         cfs_down(&oh->ot_sa_lock);
 141         while (!cfs_list_empty(&oh->ot_sa_list)) {
 142                 obj = cfs_list_entry(oh->ot_sa_list.next,
 143                                      struct osd_object, oo_sa_linkage);
 144                 sa_spill_rele(obj->oo_sa_hdl);
 145                 cfs_list_del_init(&obj->oo_sa_linkage);
 146         }
 147         cfs_up(&oh->ot_sa_lock);
 148 }
 149
 150 /*
 151  * Update the SA and add the object to the dirty list.
 152  */
 153 int osd_object_sa_update(struct osd_object *obj, sa_attr_type_t type,
 154                          void *buf, uint32_t buflen, struct osd_thandle *oh)
 155 {
 156         int rc;
 157
 158         LASSERT(obj->oo_sa_hdl != NULL);
 159         LASSERT(oh->ot_tx != NULL);
 160
 161         rc = -sa_update(obj->oo_sa_hdl, type, buf, buflen, oh->ot_tx);
 162         osd_object_sa_dirty_add(obj, oh);
 163
 164         return rc;
 165 }
 166
 167 /*
 168  * Bulk update the SA and add the object to the dirty list.
 169  */
 170 static int
 171 osd_object_sa_bulk_update(struct osd_object *obj, sa_bulk_attr_t *attrs,
 172                           int count, struct osd_thandle *oh)
 173 {
 174         int rc;
 175
 176         LASSERT(obj->oo_sa_hdl != NULL);
 177         LASSERT(oh->ot_tx != NULL);
 178
 179         rc = -sa_bulk_update(obj->oo_sa_hdl, attrs, count, oh->ot_tx);
 180         osd_object_sa_dirty_add(obj, oh);
 181
 182         return rc;
 183 }
 184
 185 /*
 186  * Retrieve the attributes of a DMU object
 187  */
 188 int __osd_object_attr_get(const struct lu_env *env, udmu_objset_t *uos,
 189                           struct osd_object *obj, struct lu_attr *la)
 190 {
 191         struct osa_attr *osa = &osd_oti_get(env)->oti_osa;
 192         sa_handle_t     *sa_hdl;
 193         sa_bulk_attr_t  *bulk;
 194         int              cnt = 0;
 195         int              rc;
 196         ENTRY;
 197
 198         LASSERT(obj->oo_db != NULL);
 199
 200         rc = -sa_handle_get(uos->os, obj->oo_db->db_object, NULL,
 201                             SA_HDL_PRIVATE, &sa_hdl);
 202         if (rc)
 203                 RETURN(rc);
 204
 205         OBD_ALLOC(bulk, sizeof(sa_bulk_attr_t) * 9);
 206         if (bulk == NULL)
 207                 GOTO(out_sa, rc = -ENOMEM);
 208
 209         la->la_valid |= LA_ATIME | LA_MTIME | LA_CTIME | LA_MODE | LA_TYPE |
 210                         LA_SIZE | LA_UID | LA_GID | LA_FLAGS | LA_NLINK;
 211
 212         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(uos), NULL, osa->atime, 16);
 213         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(uos), NULL, osa->mtime, 16);
 214         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(uos), NULL, osa->ctime, 16);
 215         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(uos), NULL, &osa->mode, 8);
 216         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(uos), NULL, &osa->size, 8);
 217         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(uos), NULL, &osa->nlink, 8);
 218         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(uos), NULL, &osa->uid, 8);
 219         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(uos), NULL, &osa->gid, 8);
 220         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(uos), NULL, &osa->flags, 8);
 221
 222         rc = -sa_bulk_lookup(sa_hdl, bulk, cnt);
 223         if (rc)
 224                 GOTO(out_bulk, rc);
 225
 226         la->la_atime = osa->atime[0];
 227         la->la_mtime = osa->mtime[0];
 228         la->la_ctime = osa->ctime[0];
 229         la->la_mode = osa->mode;
 230         la->la_uid = osa->uid;
 231         la->la_gid = osa->gid;
 232         la->la_nlink = osa->nlink;
 233         la->la_flags = osa->flags;
 234         la->la_size = osa->size;
 235
 236         if (S_ISCHR(la->la_mode) || S_ISBLK(la->la_mode)) {
 237                 rc = -sa_lookup(sa_hdl, SA_ZPL_RDEV(uos), &osa->rdev, 8);
 238                 if (rc)
 239                         GOTO(out_bulk, rc);
 240                 la->la_rdev = osa->rdev;
 241                 la->la_valid |= LA_RDEV;
 242         }
 243 out_bulk:
 244         OBD_FREE(bulk, sizeof(sa_bulk_attr_t) * 9);
 245 out_sa:
 246         sa_handle_destroy(sa_hdl);
 247
 248         RETURN(rc);
 249 }
 250
 251 int __osd_obj2dbuf(const struct lu_env *env, objset_t *os,
 252                    uint64_t oid, dmu_buf_t **dbp, void *tag)
 253 {
 254         dmu_object_info_t *doi = &osd_oti_get(env)->oti_doi;
 255         int rc;
 256
 257         LASSERT(tag);
 258
 259         rc = -sa_buf_hold(os, oid, tag, dbp);
 260         if (rc)
 261                 return rc;
 262
 263         dmu_object_info_from_db(*dbp, doi);
 264         if (unlikely (oid != DMU_USERUSED_OBJECT &&
 265             oid != DMU_GROUPUSED_OBJECT && doi->doi_bonus_type != DMU_OT_SA)) {
 266                 sa_buf_rele(*dbp, tag);
 267                 *dbp = NULL;
 268                 return -EINVAL;
 269         }
 270
 271         LASSERT(*dbp);
 272         LASSERT((*dbp)->db_object == oid);
 273         LASSERT((*dbp)->db_offset == -1);
 274         LASSERT((*dbp)->db_data != NULL);
 275
 276         return 0;
 277 }
 278
 279 /*
 280  * Concurrency: no concurrent access is possible that early in object
 281  * life-cycle.
 282  */
 283 struct lu_object *osd_object_alloc(const struct lu_env *env,
 284                                    const struct lu_object_header *hdr,
 285                                    struct lu_device *d)
 286 {
 287         struct osd_object *mo;
 288
 289         OBD_SLAB_ALLOC_PTR_GFP(mo, osd_object_kmem, CFS_ALLOC_IO);
 290         if (mo != NULL) {
 291                 struct lu_object *l;
 292
 293                 l = &mo->oo_dt.do_lu;
 294                 dt_object_init(&mo->oo_dt, NULL, d);
 295                 mo->oo_dt.do_ops = &osd_obj_ops;
 296                 l->lo_ops = &osd_lu_obj_ops;
 297                 CFS_INIT_LIST_HEAD(&mo->oo_sa_linkage);
 298                 cfs_init_rwsem(&mo->oo_sem);
 299                 cfs_sema_init(&mo->oo_guard, 1);
 300                 cfs_rwlock_init(&mo->oo_attr_lock);
 301                 return l;
 302         } else {
 303                 return NULL;
 304         }
 305 }
 306
 307 /*
 308  * Concurrency: shouldn't matter.
 309  */
 310 int osd_object_init0(const struct lu_env *env, struct osd_object *obj)
 311 {
 312         struct osd_device       *osd = osd_obj2dev(obj);
 313         udmu_objset_t           *uos = &osd->od_objset;
 314         const struct lu_fid     *fid  = lu_object_fid(&obj->oo_dt.do_lu);
 315         int                      rc = 0;
 316         ENTRY;
 317
 318         if (obj->oo_db == NULL)
 319                 RETURN(0);
 320
 321         /* object exist */
 322
 323         rc = osd_object_sa_init(obj, uos);
 324         if (rc)
 325                 RETURN(rc);
 326
 327         /* cache attrs in object */
 328         rc = __osd_object_attr_get(env, &osd->od_objset,
 329                                    obj, &obj->oo_attr);
 330         if (rc)
 331                 RETURN(rc);
 332
 333         if (likely(!fid_is_acct(fid)))
 334                 /* no body operations for accounting objects */
 335                 obj->oo_dt.do_body_ops = &osd_body_ops;
 336
 337         /*
 338          * initialize object before marking it existing
 339          */
 340         obj->oo_dt.do_lu.lo_header->loh_attr |= obj->oo_attr.la_mode & S_IFMT;
 341
 342         cfs_mb();
 343         obj->oo_dt.do_lu.lo_header->loh_attr |= LOHA_EXISTS;
 344
 345         RETURN(0);
 346 }
 347
 348 /*
 349  * Concurrency: no concurrent access is possible that early in object
 350  * life-cycle.
 351  */
 352 static int osd_object_init(const struct lu_env *env, struct lu_object *l,
 353                            const struct lu_object_conf *conf)
 354 {
 355         struct osd_object       *obj = osd_obj(l);
 356         struct osd_device       *osd = osd_obj2dev(obj);
 357         uint64_t                 oid;
 358         int                      rc;
 359         ENTRY;
 360
 361         LASSERT(osd_invariant(obj));
 362
 363         rc = osd_fid_lookup(env, osd, lu_object_fid(l), &oid);
 364         if (rc == 0) {
 365                 LASSERT(obj->oo_db == NULL);
 366                 rc = __osd_obj2dbuf(env, osd->od_objset.os, oid,
 367                                         &obj->oo_db, osd_obj_tag);
 368                 if (rc == 0) {
 369                         LASSERT(obj->oo_db);
 370                         rc = osd_object_init0(env, obj);
 371                 } else {
 372                         CERROR("%s: lookup "DFID"/"LPX64" failed: rc = %d\n",
 373                                osd->od_svname, PFID(lu_object_fid(l)), oid, rc);
 374                 }
 375         } else if (rc == -ENOENT) {
 376                 if (fid_is_otable_it(&l->lo_header->loh_fid)) {
 377                         obj->oo_dt.do_ops = &osd_obj_otable_it_ops;
 378                         /* LFSCK iterator object is special without inode */
 379                         l->lo_header->loh_attr |= LOHA_EXISTS;
 380                 }
 381                 rc = 0;
 382         }
 383         LASSERT(osd_invariant(obj));
 384         RETURN(rc);
 385 }
 386
 387 /*
 388  * Concurrency: no concurrent access is possible that late in object
 389  * life-cycle.
 390  */
 391 static void osd_object_free(const struct lu_env *env, struct lu_object *l)
 392 {
 393         struct osd_object *obj = osd_obj(l);
 394
 395         LASSERT(osd_invariant(obj));
 396
 397         dt_object_fini(&obj->oo_dt);
 398         OBD_SLAB_FREE_PTR(obj, osd_object_kmem);
 399 }
 400
 401 static void __osd_declare_object_destroy(const struct lu_env *env,
 402                                          struct osd_object *obj,
 403                                          struct osd_thandle *oh)
 404 {
 405         struct osd_device       *osd = osd_obj2dev(obj);
 406         udmu_objset_t           *uos = &osd->od_objset;
 407         dmu_buf_t               *db = obj->oo_db;
 408         zap_attribute_t         *za = &osd_oti_get(env)->oti_za;
 409         uint64_t                 oid = db->db_object, xid;
 410         dmu_tx_t                *tx = oh->ot_tx;
 411         zap_cursor_t            *zc;
 412         int                      rc = 0;
 413
 414         dmu_tx_hold_free(tx, oid, 0, DMU_OBJECT_END);
 415
 416         /* zap holding xattrs */
 417         if (obj->oo_xattr != ZFS_NO_OBJECT) {
 418                 oid = obj->oo_xattr;
 419
 420                 dmu_tx_hold_free(tx, oid, 0, DMU_OBJECT_END);
 421
 422                 rc = -udmu_zap_cursor_init(&zc, uos, oid, 0);
 423                 if (rc)
 424                         goto out;
 425
 426                 while ((rc = -zap_cursor_retrieve(zc, za)) == 0) {
 427                         BUG_ON(za->za_integer_length != sizeof(uint64_t));
 428                         BUG_ON(za->za_num_integers != 1);
 429
 430                         rc = -zap_lookup(uos->os, obj->oo_xattr, za->za_name,
 431                                          sizeof(uint64_t), 1, &xid);
 432                         if (rc) {
 433                                 CERROR("%s: xattr lookup failed: rc = %d\n",
 434                                        osd->od_svname, rc);
 435                                 goto out_err;
 436                         }
 437                         dmu_tx_hold_free(tx, xid, 0, DMU_OBJECT_END);
 438
 439                         zap_cursor_advance(zc);
 440                 }
 441                 if (rc == -ENOENT)
 442                         rc = 0;
 443 out_err:
 444                 udmu_zap_cursor_fini(zc);
 445         }
 446 out:
 447         if (rc && tx->tx_err == 0)
 448                 tx->tx_err = -rc;
 449 }
 450
 451 static int osd_declare_object_destroy(const struct lu_env *env,
 452                                       struct dt_object *dt,
 453                                       struct thandle *th)
 454 {
 455         char                    *buf = osd_oti_get(env)->oti_str;
 456         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
 457         struct osd_object       *obj = osd_dt_obj(dt);
 458         struct osd_device       *osd = osd_obj2dev(obj);
 459         struct osd_thandle      *oh;
 460         uint64_t                 zapid;
 461         ENTRY;
 462
 463         LASSERT(th != NULL);
 464         LASSERT(dt_object_exists(dt));
 465
 466         oh = container_of0(th, struct osd_thandle, ot_super);
 467         LASSERT(oh->ot_tx != NULL);
 468
 469         /* declare that we'll destroy the object */
 470         __osd_declare_object_destroy(env, obj, oh);
 471
 472         /* declare that we'll remove object from fid-dnode mapping */
 473         zapid = osd_get_name_n_idx(env, osd, fid, buf);
 474         dmu_tx_hold_bonus(oh->ot_tx, zapid);
 475         dmu_tx_hold_zap(oh->ot_tx, zapid, 0, buf);
 476
 477         /* declare that we'll remove object from inode accounting ZAPs */
 478         dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
 479         dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, 0, buf);
 480         dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
 481         dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, 0, buf);
 482
 483         RETURN(0);
 484 }
 485
 486 int __osd_object_free(udmu_objset_t *uos, uint64_t oid, dmu_tx_t *tx)
 487 {
 488         LASSERT(uos->objects != 0);
 489         cfs_spin_lock(&uos->lock);
 490         uos->objects--;
 491         cfs_spin_unlock(&uos->lock);
 492
 493         return -dmu_object_free(uos->os, oid, tx);
 494 }
 495
 496 /*
 497  * Delete a DMU object
 498  *
 499  * The transaction passed to this routine must have
 500  * dmu_tx_hold_free(tx, oid, 0, DMU_OBJECT_END) called
 501  * and then assigned to a transaction group.
 502  *
 503  * This will release db and set it to NULL to prevent further dbuf releases.
 504  */
 505 static int __osd_object_destroy(const struct lu_env *env,
 506                                 struct osd_object *obj,
 507                                 dmu_tx_t *tx, void *tag)
 508 {
 509         struct osd_device       *osd = osd_obj2dev(obj);
 510         udmu_objset_t           *uos = &osd->od_objset;
 511         uint64_t                 xid;
 512         zap_attribute_t         *za = &osd_oti_get(env)->oti_za;
 513         zap_cursor_t            *zc;
 514         int                      rc;
 515
 516         /* Assert that the transaction has been assigned to a
 517            transaction group. */
 518         LASSERT(tx->tx_txg != 0);
 519
 520         /* zap holding xattrs */
 521         if (obj->oo_xattr != ZFS_NO_OBJECT) {
 522                 rc = -udmu_zap_cursor_init(&zc, uos, obj->oo_xattr, 0);
 523                 if (rc)
 524                         return rc;
 525                 while ((rc = -zap_cursor_retrieve(zc, za)) == 0) {
 526                         BUG_ON(za->za_integer_length != sizeof(uint64_t));
 527                         BUG_ON(za->za_num_integers != 1);
 528
 529                         rc = -zap_lookup(uos->os, obj->oo_xattr, za->za_name,
 530                                          sizeof(uint64_t), 1, &xid);
 531                         if (rc) {
 532                                 CERROR("%s: lookup xattr %s failed: rc = %d\n",
 533                                        osd->od_svname, za->za_name, rc);
 534                                 continue;
 535                         }
 536                         rc = __osd_object_free(uos, xid, tx);
 537                         if (rc)
 538                                 CERROR("%s: fetch xattr %s failed: rc = %d\n",
 539                                        osd->od_svname, za->za_name, rc);
 540                         zap_cursor_advance(zc);
 541                 }
 542                 udmu_zap_cursor_fini(zc);
 543
 544                 rc = __osd_object_free(uos, obj->oo_xattr, tx);
 545                 if (rc)
 546                         CERROR("%s: freeing xattr failed: rc = %d\n",
 547                                osd->od_svname, rc);
 548         }
 549
 550         return __osd_object_free(uos, obj->oo_db->db_object, tx);
 551 }
 552
 553 static int osd_object_destroy(const struct lu_env *env,
 554                               struct dt_object *dt,
 555                               struct thandle *th)
 556 {
 557         char                    *buf = osd_oti_get(env)->oti_str;
 558         struct osd_object       *obj = osd_dt_obj(dt);
 559         struct osd_device       *osd = osd_obj2dev(obj);
 560         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
 561         struct osd_thandle      *oh;
 562         int                      rc;
 563         uint64_t                 zapid;
 564         ENTRY;
 565
 566         LASSERT(obj->oo_db != NULL);
 567         LASSERT(dt_object_exists(dt));
 568         LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));
 569
 570         oh = container_of0(th, struct osd_thandle, ot_super);
 571         LASSERT(oh != NULL);
 572         LASSERT(oh->ot_tx != NULL);
 573
 574         zapid = osd_get_name_n_idx(env, osd, fid, buf);
 575
 576         /* remove obj ref from index dir (it depends) */
 577         rc = -zap_remove(osd->od_objset.os, zapid, buf, oh->ot_tx);
 578         if (rc) {
 579                 CERROR("%s: zap_remove() failed: rc = %d\n",
 580                        osd->od_svname, rc);
 581                 GOTO(out, rc);
 582         }
 583
 584         /* Remove object from inode accounting. It is not fatal for the destroy
 585          * operation if something goes wrong while updating accounting, but we
 586          * still log an error message to notify the administrator */
 587         rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
 588                         obj->oo_attr.la_uid, -1, oh->ot_tx);
 589         if (rc)
 590                 CERROR("%s: failed to remove "DFID" from accounting ZAP for usr"
 591                         " %d: rc = %d\n", osd->od_svname, PFID(fid),
 592                         obj->oo_attr.la_uid, rc);
 593         rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
 594                                 obj->oo_attr.la_gid, -1, oh->ot_tx);
 595         if (rc)
 596                 CERROR("%s: failed to remove "DFID" from accounting ZAP for grp"
 597                         " %d: rc = %d\n", osd->od_svname, PFID(fid),
 598                         obj->oo_attr.la_gid, rc);
 599
 600         /* kill object */
 601         rc = __osd_object_destroy(env, obj, oh->ot_tx, osd_obj_tag);
 602         if (rc) {
 603                 CERROR("%s: __osd_object_destroy() failed: rc = %d\n",
 604                        osd->od_svname, rc);
 605                 GOTO(out, rc);
 606         }
 607
 608 out:
 609         /* not needed in the cache anymore */
 610         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
 611
 612         RETURN (0);
 613 }
 614
 615 static void osd_object_delete(const struct lu_env *env, struct lu_object *l)
 616 {
 617         struct osd_object *obj = osd_obj(l);
 618
 619         if (obj->oo_db != NULL) {
 620                 osd_object_sa_fini(obj);
 621                 if (obj->oo_sa_xattr) {
 622                         nvlist_free(obj->oo_sa_xattr);
 623                         obj->oo_sa_xattr = NULL;
 624                 }
 625                 sa_buf_rele(obj->oo_db, osd_obj_tag);
 626                 cfs_list_del(&obj->oo_sa_linkage);
 627                 obj->oo_db = NULL;
 628         }
 629 }
 630
 631 /*
 632  * Concurrency: ->loo_object_release() is called under site spin-lock.
 633  */
 634 static void osd_object_release(const struct lu_env *env,
 635                                struct lu_object *l)
 636 {
 637 }
 638
 639 /*
 640  * Concurrency: shouldn't matter.
 641  */
 642 static int osd_object_print(const struct lu_env *env, void *cookie,
 643                             lu_printer_t p, const struct lu_object *l)
 644 {
 645         struct osd_object *o = osd_obj(l);
 646
 647         return (*p)(env, cookie, LUSTRE_OSD_ZFS_NAME"-object@%p", o);
 648 }
 649
 650 static void osd_object_read_lock(const struct lu_env *env,
 651                                  struct dt_object *dt, unsigned role)
 652 {
 653         struct osd_object *obj = osd_dt_obj(dt);
 654
 655         LASSERT(osd_invariant(obj));
 656
 657         cfs_down_read(&obj->oo_sem);
 658 }
 659
 660 static void osd_object_write_lock(const struct lu_env *env,
 661                                   struct dt_object *dt, unsigned role)
 662 {
 663         struct osd_object *obj = osd_dt_obj(dt);
 664
 665         LASSERT(osd_invariant(obj));
 666
 667         cfs_down_write(&obj->oo_sem);
 668 }
 669
 670 static void osd_object_read_unlock(const struct lu_env *env,
 671                                    struct dt_object *dt)
 672 {
 673         struct osd_object *obj = osd_dt_obj(dt);
 674
 675         LASSERT(osd_invariant(obj));
 676         cfs_up_read(&obj->oo_sem);
 677 }
 678
 679 static void osd_object_write_unlock(const struct lu_env *env,
 680                                     struct dt_object *dt)
 681 {
 682         struct osd_object *obj = osd_dt_obj(dt);
 683
 684         LASSERT(osd_invariant(obj));
 685         cfs_up_write(&obj->oo_sem);
 686 }
 687
 688 static int osd_object_write_locked(const struct lu_env *env,
 689                                    struct dt_object *dt)
 690 {
 691         struct osd_object *obj = osd_dt_obj(dt);
 692         int rc = 1;
 693
 694         LASSERT(osd_invariant(obj));
 695
 696         if (cfs_down_write_trylock(&obj->oo_sem)) {
 697                 rc = 0;
 698                 cfs_up_write(&obj->oo_sem);
 699         }
 700         return rc;
 701 }
 702
 703 static int osd_attr_get(const struct lu_env *env,
 704                         struct dt_object *dt,
 705                         struct lu_attr *attr,
 706                         struct lustre_capa *capa)
 707 {
 708         struct osd_object       *obj = osd_dt_obj(dt);
 709         uint64_t                 blocks;
 710         uint32_t                 blksize;
 711
 712         LASSERT(dt_object_exists(dt));
 713         LASSERT(osd_invariant(obj));
 714         LASSERT(obj->oo_db);
 715
 716         cfs_read_lock(&obj->oo_attr_lock);
 717         *attr = obj->oo_attr;
 718         cfs_read_unlock(&obj->oo_attr_lock);
 719
 720         /* with ZFS_DEBUG zrl_add_debug() called by DB_DNODE_ENTER()
 721          * from within sa_object_size() can block on a mutex, so
 722          * we can't call sa_object_size() holding rwlock */
 723         sa_object_size(obj->oo_sa_hdl, &blksize, &blocks);
 724         /* Block size may be not set; suggest maximal I/O transfers. */
 725         if (blksize == 0)
 726                 blksize = 1ULL << SPA_MAXBLOCKSHIFT;
 727
 728         attr->la_blksize = blksize;
 729         attr->la_blocks = blocks;
 730         attr->la_valid |= LA_BLOCKS | LA_BLKSIZE;
 731
 732         return 0;
 733 }
 734
 735 static int osd_declare_attr_set(const struct lu_env *env,
 736                                 struct dt_object *dt,
 737                                 const struct lu_attr *attr,
 738                                 struct thandle *handle)
 739 {
 740         char                    *buf = osd_oti_get(env)->oti_str;
 741         struct osd_object       *obj = osd_dt_obj(dt);
 742         struct osd_device       *osd = osd_obj2dev(obj);
 743         struct osd_thandle      *oh;
 744         ENTRY;
 745
 746         if (!dt_object_exists(dt)) {
 747                 /* XXX: sanity check that object creation is declared */
 748                 RETURN(0);
 749         }
 750
 751         LASSERT(handle != NULL);
 752         LASSERT(osd_invariant(obj));
 753
 754         oh = container_of0(handle, struct osd_thandle, ot_super);
 755
 756         LASSERT(obj->oo_sa_hdl != NULL);
 757         dmu_tx_hold_sa(oh->ot_tx, obj->oo_sa_hdl, 0);
 758
 759         if (attr && attr->la_valid & LA_UID) {
 760                 /* account for user inode tracking ZAP update */
 761                 dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
 762                 dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, TRUE, buf);
 763         }
 764         if (attr && attr->la_valid & LA_GID) {
 765                 /* account for user inode tracking ZAP update */
 766                 dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
 767                 dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, TRUE, buf);
 768         }
 769
 770         RETURN(0);
 771 }
 772
 773 /*
 774  * Set the attributes of an object
 775  *
 776  * The transaction passed to this routine must have
 777  * dmu_tx_hold_bonus(tx, oid) called and then assigned
 778  * to a transaction group.
 779  */
 780 static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
 781                         const struct lu_attr *la, struct thandle *handle,
 782                         struct lustre_capa *capa)
 783 {
 784         struct osd_object       *obj = osd_dt_obj(dt);
 785         struct osd_device       *osd = osd_obj2dev(obj);
 786         udmu_objset_t           *uos = &osd->od_objset;
 787         struct osd_thandle      *oh;
 788         struct osa_attr         *osa = &osd_oti_get(env)->oti_osa;
 789         sa_bulk_attr_t          *bulk;
 790         int                      cnt;
 791         int                      rc = 0;
 792
 793         ENTRY;
 794         LASSERT(handle != NULL);
 795         LASSERT(dt_object_exists(dt));
 796         LASSERT(osd_invariant(obj));
 797         LASSERT(obj->oo_sa_hdl);
 798
 799         oh = container_of0(handle, struct osd_thandle, ot_super);
 800         /* Assert that the transaction has been assigned to a
 801            transaction group. */
 802         LASSERT(oh->ot_tx->tx_txg != 0);
 803
 804         if (la->la_valid == 0)
 805                 RETURN(0);
 806
 807         OBD_ALLOC(bulk, sizeof(sa_bulk_attr_t) * 10);
 808         if (bulk == NULL)
 809                 RETURN(-ENOMEM);
 810
 811         /* do both accounting updates outside oo_attr_lock below */
 812         if ((la->la_valid & LA_UID) && (la->la_uid != obj->oo_attr.la_uid)) {
 813                 /* Update user accounting. Failure isn't fatal, but we still
 814                  * log an error message */
 815                 rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
 816                                         la->la_uid, 1, oh->ot_tx);
 817                 if (rc)
 818                         CERROR("%s: failed to update accounting ZAP for user "
 819                                 "%d (%d)\n", osd->od_svname, la->la_uid, rc);
 820                 rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
 821                                         obj->oo_attr.la_uid, -1, oh->ot_tx);
 822                 if (rc)
 823                         CERROR("%s: failed to update accounting ZAP for user "
 824                                 "%d (%d)\n", osd->od_svname,
 825                                 obj->oo_attr.la_uid, rc);
 826         }
 827         if ((la->la_valid & LA_GID) && (la->la_gid != obj->oo_attr.la_gid)) {
 828                 /* Update group accounting. Failure isn't fatal, but we still
 829                  * log an error message */
 830                 rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
 831                                         la->la_gid, 1, oh->ot_tx);
 832                 if (rc)
 833                         CERROR("%s: failed to update accounting ZAP for user "
 834                                 "%d (%d)\n", osd->od_svname, la->la_gid, rc);
 835                 rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
 836                                         obj->oo_attr.la_gid, -1, oh->ot_tx);
 837                 if (rc)
 838                         CERROR("%s: failed to update accounting ZAP for user "
 839                                 "%d (%d)\n", osd->od_svname,
 840                                 obj->oo_attr.la_gid, rc);
 841         }
 842
 843         cfs_write_lock(&obj->oo_attr_lock);
 844         cnt = 0;
 845         if (la->la_valid & LA_ATIME) {
 846                 osa->atime[0] = obj->oo_attr.la_atime = la->la_atime;
 847                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(uos), NULL,
 848                                  osa->atime, 16);
 849         }
 850         if (la->la_valid & LA_MTIME) {
 851                 osa->mtime[0] = obj->oo_attr.la_mtime = la->la_mtime;
 852                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(uos), NULL,
 853                                  osa->mtime, 16);
 854         }
 855         if (la->la_valid & LA_CTIME) {
 856                 osa->ctime[0] = obj->oo_attr.la_ctime = la->la_ctime;
 857                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(uos), NULL,
 858                                  osa->ctime, 16);
 859         }
 860         if (la->la_valid & LA_MODE) {
 861                 /* mode is stored along with type, so read it first */
 862                 obj->oo_attr.la_mode = (obj->oo_attr.la_mode & S_IFMT) |
 863                         (la->la_mode & ~S_IFMT);
 864                 osa->mode = obj->oo_attr.la_mode;
 865                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(uos), NULL,
 866                                  &osa->mode, 8);
 867         }
 868         if (la->la_valid & LA_SIZE) {
 869                 osa->size = obj->oo_attr.la_size = la->la_size;
 870                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(uos), NULL,
 871                                  &osa->size, 8);
 872         }
 873         if (la->la_valid & LA_NLINK) {
 874                 osa->nlink = obj->oo_attr.la_nlink = la->la_nlink;
 875                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(uos), NULL,
 876                                  &osa->nlink, 8);
 877         }
 878         if (la->la_valid & LA_RDEV) {
 879                 osa->rdev = obj->oo_attr.la_rdev = la->la_rdev;
 880                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(uos), NULL,
 881                                  &osa->rdev, 8);
 882         }
 883         if (la->la_valid & LA_FLAGS) {
 884                 osa->flags = obj->oo_attr.la_flags = la->la_flags;
 885                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(uos), NULL,
 886                                  &osa->flags, 8);
 887         }
 888         if (la->la_valid & LA_UID) {
 889                 osa->uid = obj->oo_attr.la_uid = la->la_uid;
 890                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(uos), NULL,
 891                                  &osa->uid, 8);
 892         }
 893         if (la->la_valid & LA_GID) {
 894                 osa->gid = obj->oo_attr.la_gid = la->la_gid;
 895                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(uos), NULL,
 896                                  &osa->gid, 8);
 897         }
 898         obj->oo_attr.la_valid |= la->la_valid;
 899         cfs_write_unlock(&obj->oo_attr_lock);
 900
 901         rc = osd_object_sa_bulk_update(obj, bulk, cnt, oh);
 902
 903         OBD_FREE(bulk, sizeof(sa_bulk_attr_t) * 10);
 904         RETURN(rc);
 905 }
 906
 907 /*
 908  * Object creation.
 909  *
 910  * XXX temporary solution.
 911  */
 912
 913 static void osd_ah_init(const struct lu_env *env, struct dt_allocation_hint *ah,
 914                         struct dt_object *parent, cfs_umode_t child_mode)
 915 {
 916         LASSERT(ah);
 917
 918         memset(ah, 0, sizeof(*ah));
 919         ah->dah_mode = child_mode;
 920 }
 921
 922 static int osd_declare_object_create(const struct lu_env *env,
 923                                      struct dt_object *dt,
 924                                      struct lu_attr *attr,
 925                                      struct dt_allocation_hint *hint,
 926                                      struct dt_object_format *dof,
 927                                      struct thandle *handle)
 928 {
 929         char                    *buf = osd_oti_get(env)->oti_str;
 930         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
 931         struct osd_object       *obj = osd_dt_obj(dt);
 932         struct osd_device       *osd = osd_obj2dev(obj);
 933         struct osd_thandle      *oh;
 934         uint64_t                 zapid;
 935         ENTRY;
 936
 937         LASSERT(dof);
 938
 939         switch (dof->dof_type) {
 940                 case DFT_REGULAR:
 941                 case DFT_SYM:
 942                 case DFT_NODE:
 943                         if (obj->oo_dt.do_body_ops == NULL)
 944                                 obj->oo_dt.do_body_ops = &osd_body_ops;
 945                         break;
 946                 default:
 947                         break;
 948         }
 949
 950         LASSERT(handle != NULL);
 951         oh = container_of0(handle, struct osd_thandle, ot_super);
 952         LASSERT(oh->ot_tx != NULL);
 953
 954         switch (dof->dof_type) {
 955                 case DFT_DIR:
 956                 case DFT_INDEX:
 957                         /* for zap create */
 958                         dmu_tx_hold_zap(oh->ot_tx, DMU_NEW_OBJECT, 1, NULL);
 959                         break;
 960                 case DFT_REGULAR:
 961                 case DFT_SYM:
 962                 case DFT_NODE:
 963                         /* first, we'll create new object */
 964                         dmu_tx_hold_bonus(oh->ot_tx, DMU_NEW_OBJECT);
 965                         break;
 966
 967                 default:
 968                         LBUG();
 969                         break;
 970         }
 971
 972         /* and we'll add it to some mapping */
 973         zapid = osd_get_name_n_idx(env, osd, fid, buf);
 974         dmu_tx_hold_bonus(oh->ot_tx, zapid);
 975         dmu_tx_hold_zap(oh->ot_tx, zapid, TRUE, buf);
 976
 977         /* we will also update inode accounting ZAPs */
 978         dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
 979         dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, TRUE, buf);
 980         dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
 981         dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, TRUE, buf);
 982
 983         dmu_tx_hold_sa_create(oh->ot_tx, ZFS_SA_BASE_ATTR_SIZE);
 984
 985         RETURN(0);
 986 }
 987
 988 int __osd_attr_init(const struct lu_env *env, udmu_objset_t *uos,
 989                     uint64_t oid, dmu_tx_t *tx, struct lu_attr *la)
 990 {
 991         sa_bulk_attr_t  *bulk;
 992         sa_handle_t     *sa_hdl;
 993         struct osa_attr *osa = &osd_oti_get(env)->oti_osa;
 994         uint64_t         gen;
 995         uint64_t         parent;
 996         uint64_t         crtime[2];
 997         timestruc_t      now;
 998         int              cnt;
 999         int              rc;
1000
1001         gethrestime(&now);
1002         gen = dmu_tx_get_txg(tx);
1003
1004         ZFS_TIME_ENCODE(&now, crtime);
1005         /* XXX: this should be real id of parent for ZPL access, but we have no
1006          * such info in OSD, probably it can be part of dt_object_format */
1007         parent = 0;
1008
1009         osa->atime[0] = la->la_atime;
1010         osa->ctime[0] = la->la_ctime;
1011         osa->mtime[0] = la->la_mtime;
1012         osa->mode = la->la_mode;
1013         osa->uid = la->la_uid;
1014         osa->gid = la->la_gid;
1015         osa->rdev = la->la_rdev;
1016         osa->nlink = la->la_nlink;
1017         osa->flags = la->la_flags;
1018         osa->size  = la->la_size;
1019
1020         /* Now add in all of the "SA" attributes */
1021         rc = -sa_handle_get(uos->os, oid, NULL, SA_HDL_PRIVATE, &sa_hdl);
1022         if (rc)
1023                 return rc;
1024
1025         OBD_ALLOC(bulk, sizeof(sa_bulk_attr_t) * 13);
1026         if (bulk == NULL) {
1027                 rc = -ENOMEM;
1028                 goto out;
1029         }
1030         /*
1031          * we need to create all SA below upon object create.
1032          *
1033          * XXX The attribute order matters since the accounting callback relies
1034          * on static offsets (i.e. SA_*_OFFSET, see zfs_space_delta_cb()) to
1035          * look up the UID/GID attributes. Moreover, the callback does not seem
1036          * to support the spill block.
1037          * We define attributes in the same order as SA_*_OFFSET in order to
1038          * work around the problem. See ORI-610.
1039          */
1040         cnt = 0;
1041         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(uos), NULL, &osa->mode, 8);
1042         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(uos), NULL, &osa->size, 8);
1043         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GEN(uos), NULL, &gen, 8);
1044         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(uos), NULL, &osa->uid, 8);
1045         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(uos), NULL, &osa->gid, 8);
1046         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PARENT(uos), NULL, &parent, 8);
1047         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(uos), NULL, &osa->flags, 8);
1048         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(uos), NULL, osa->atime, 16);
1049         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(uos), NULL, osa->mtime, 16);
1050         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(uos), NULL, osa->ctime, 16);
1051         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CRTIME(uos), NULL, crtime, 16);
1052         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(uos), NULL, &osa->nlink, 8);
1053         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(uos), NULL, &osa->rdev, 8);
1054
1055         rc = -sa_replace_all_by_template(sa_hdl, bulk, cnt, tx);
1056
1057         OBD_FREE(bulk, sizeof(sa_bulk_attr_t) * 13);
1058 out:
1059         sa_handle_destroy(sa_hdl);
1060         return rc;
1061 }
1062
1063 /*
1064  * The transaction passed to this routine must have
1065  * dmu_tx_hold_bonus(tx, DMU_NEW_OBJECT) called and then assigned
1066  * to a transaction group.
1067  */
1068 int __osd_object_create(const struct lu_env *env, udmu_objset_t *uos,
1069                         dmu_buf_t **dbp, dmu_tx_t *tx,
1070                         struct lu_attr *la, void *tag)
1071 {
1072         uint64_t oid;
1073         int      rc;
1074
1075         LASSERT(tag);
1076         cfs_spin_lock(&uos->lock);
1077         uos->objects++;
1078         cfs_spin_unlock(&uos->lock);
1079
1080         /* Assert that the transaction has been assigned to a
1081            transaction group. */
1082         LASSERT(tx->tx_txg != 0);
1083
1084         /* Create a new DMU object. */
1085         oid = dmu_object_alloc(uos->os, DMU_OT_PLAIN_FILE_CONTENTS, 0,
1086                                DMU_OT_SA, DN_MAX_BONUSLEN, tx);
1087         rc = -sa_buf_hold(uos->os, oid, tag, dbp);
1088         if (rc)
1089                 return rc;
1090
1091         LASSERT(la->la_valid & LA_MODE);
1092         la->la_size = 0;
1093         la->la_nlink = 1;
1094
1095         return __osd_attr_init(env, uos, oid, tx, la);
1096 }
1097
1098 /*
1099  * The transaction passed to this routine must have
1100  * dmu_tx_hold_zap(tx, DMU_NEW_OBJECT, ...) called and then assigned
1101  * to a transaction group.
1102  *
1103  * Using ZAP_FLAG_HASH64 will force the ZAP to always be a FAT ZAP.
1104  * This is fine for directories today, because storing the FID in the dirent
1105  * will also require a FAT ZAP.  If there is a new type of micro ZAP created
1106  * then we might need to re-evaluate the use of this flag and instead do
1107  * a conversion from the different internal ZAP hash formats being used. */
1108 int __osd_zap_create(const struct lu_env *env, udmu_objset_t *uos,
1109                      dmu_buf_t **zap_dbp, dmu_tx_t *tx,
1110                      struct lu_attr *la, void *tag, zap_flags_t flags)
1111 {
1112         uint64_t oid;
1113         int      rc;
1114
1115         LASSERT(tag);
1116
1117         cfs_spin_lock(&uos->lock);
1118         uos->objects++;
1119         cfs_spin_unlock(&uos->lock);
1120
1121         /* Assert that the transaction has been assigned to a
1122            transaction group. */
1123         LASSERT(tx->tx_txg != 0);
1124
1125         oid = zap_create_flags(uos->os, 0, flags | ZAP_FLAG_HASH64,
1126                                DMU_OT_DIRECTORY_CONTENTS, 12, 12,
1127                                DMU_OT_SA, DN_MAX_BONUSLEN, tx);
1128
1129         rc = -sa_buf_hold(uos->os, oid, tag, zap_dbp);
1130         if (rc)
1131                 return rc;
1132
1133         LASSERT(la->la_valid & LA_MODE);
1134         la->la_size = 2;
1135         la->la_nlink = 1;
1136
1137         return __osd_attr_init(env, uos, oid, tx, la);
1138 }
1139
1140 static dmu_buf_t *osd_mkidx(const struct lu_env *env, struct osd_device *osd,
1141                             struct lu_attr *la, struct osd_thandle *oh)
1142 {
1143         dmu_buf_t *db;
1144         int        rc;
1145
1146         /* Index file should be created as regular file in order not to confuse
1147          * ZPL which could interpret them as directory.
1148          * We set ZAP_FLAG_UINT64_KEY to let ZFS know than we are going to use
1149          * binary keys */
1150         LASSERT(S_ISREG(la->la_mode));
1151         rc = __osd_zap_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1152                               osd_obj_tag, ZAP_FLAG_UINT64_KEY);
1153         if (rc)
1154                 return ERR_PTR(rc);
1155         return db;
1156 }
1157
1158 static dmu_buf_t *osd_mkdir(const struct lu_env *env, struct osd_device *osd,
1159                             struct lu_attr *la, struct osd_thandle *oh)
1160 {
1161         dmu_buf_t *db;
1162         int        rc;
1163
1164         LASSERT(S_ISDIR(la->la_mode));
1165         rc = __osd_zap_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1166                               osd_obj_tag, 0);
1167         if (rc)
1168                 return ERR_PTR(rc);
1169         return db;
1170 }
1171
1172 static dmu_buf_t* osd_mkreg(const struct lu_env *env, struct osd_device *osd,
1173                             struct lu_attr *la, struct osd_thandle *oh)
1174 {
1175         dmu_buf_t *db;
1176         int         rc;
1177
1178         LASSERT(S_ISREG(la->la_mode));
1179         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1180                                  osd_obj_tag);
1181         if (rc)
1182                 return ERR_PTR(rc);
1183
1184         /*
1185          * XXX: a hack, OST to use bigger blocksize. we need
1186          * a method in OSD API to control this from OFD/MDD
1187          */
1188         if (!lu_device_is_md(osd2lu_dev(osd))) {
1189                 rc = -dmu_object_set_blocksize(osd->od_objset.os,
1190                                                db->db_object,
1191                                 128 << 10, 0, oh->ot_tx);
1192                 if (unlikely(rc)) {
1193                         CERROR("%s: can't change blocksize: %d\n",
1194                                osd->od_svname, rc);
1195                         return ERR_PTR(rc);
1196                 }
1197         }
1198
1199         return db;
1200 }
1201
1202 static dmu_buf_t *osd_mksym(const struct lu_env *env, struct osd_device *osd,
1203                             struct lu_attr *la, struct osd_thandle *oh)
1204 {
1205         dmu_buf_t *db;
1206         int        rc;
1207
1208         LASSERT(S_ISLNK(la->la_mode));
1209         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1210                                  osd_obj_tag);
1211         if (rc)
1212                 return ERR_PTR(rc);
1213         return db;
1214 }
1215
1216 static dmu_buf_t *osd_mknod(const struct lu_env *env, struct osd_device *osd,
1217                             struct lu_attr *la, struct osd_thandle *oh)
1218 {
1219         dmu_buf_t *db;
1220         int        rc;
1221
1222         la->la_valid = LA_MODE;
1223         if (S_ISCHR(la->la_mode) || S_ISBLK(la->la_mode))
1224                 la->la_valid |= LA_RDEV;
1225
1226         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1227                                  osd_obj_tag);
1228         if (rc)
1229                 return ERR_PTR(rc);
1230         return db;
1231 }
1232
1233 typedef dmu_buf_t *(*osd_obj_type_f)(const struct lu_env *env, struct osd_device *osd,
1234                                      struct lu_attr *la, struct osd_thandle *oh);
1235
1236 static osd_obj_type_f osd_create_type_f(enum dt_format_type type)
1237 {
1238         osd_obj_type_f result;
1239
1240         switch (type) {
1241         case DFT_DIR:
1242                 result = osd_mkdir;
1243                 break;
1244         case DFT_INDEX:
1245                 result = osd_mkidx;
1246                 break;
1247         case DFT_REGULAR:
1248                 result = osd_mkreg;
1249                 break;
1250         case DFT_SYM:
1251                 result = osd_mksym;
1252                 break;
1253         case DFT_NODE:
1254                 result = osd_mknod;
1255                 break;
1256         default:
1257                 LBUG();
1258                 break;
1259         }
1260         return result;
1261 }
1262
1263 /*
1264  * Primitives for directory (i.e. ZAP) handling
1265  */
1266
1267 /*
1268  * Concurrency: @dt is write locked.
1269  */
1270 static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
1271                              struct lu_attr *attr,
1272                              struct dt_allocation_hint *hint,
1273                              struct dt_object_format *dof,
1274                              struct thandle *th)
1275 {
1276         struct zpl_direntry     *zde = &osd_oti_get(env)->oti_zde.lzd_reg;
1277         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1278         struct osd_object       *obj = osd_dt_obj(dt);
1279         struct osd_device       *osd = osd_obj2dev(obj);
1280         char                    *buf = osd_oti_get(env)->oti_str;
1281         struct osd_thandle      *oh;
1282         dmu_buf_t               *db;
1283         uint64_t                 zapid;
1284         int                      rc;
1285
1286         ENTRY;
1287
1288         /* concurrent create declarations should not see
1289          * the object inconsistent (db, attr, etc).
1290          * in regular cases acquisition should be cheap */
1291         cfs_down(&obj->oo_guard);
1292
1293         LASSERT(osd_invariant(obj));
1294         LASSERT(!dt_object_exists(dt));
1295         LASSERT(dof != NULL);
1296
1297         LASSERT(th != NULL);
1298         oh = container_of0(th, struct osd_thandle, ot_super);
1299
1300         /*
1301          * XXX missing: Quote handling.
1302          */
1303
1304         LASSERT(obj->oo_db == NULL);
1305
1306         db = osd_create_type_f(dof->dof_type)(env, osd, attr, oh);
1307         if (IS_ERR(db))
1308                 GOTO(out, rc = PTR_ERR(th));
1309
1310         zde->zde_pad = 0;
1311         zde->zde_dnode = db->db_object;
1312         zde->zde_type = IFTODT(attr->la_mode & S_IFMT);
1313
1314         zapid = osd_get_name_n_idx(env, osd, fid, buf);
1315
1316         rc = -zap_add(osd->od_objset.os, zapid, buf, 8, 1, zde, oh->ot_tx);
1317         if (rc)
1318                 GOTO(out, rc);
1319
1320         /* Add new object to inode accounting.
1321          * Errors are not considered as fatal */
1322         rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
1323                                 (attr->la_valid & LA_UID) ? attr->la_uid : 0, 1,
1324                                 oh->ot_tx);
1325         if (rc)
1326                 CERROR("%s: failed to add "DFID" to accounting ZAP for usr %d "
1327                         "(%d)\n", osd->od_svname, PFID(fid), attr->la_uid, rc);
1328         rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
1329                                 (attr->la_valid & LA_GID) ? attr->la_gid : 0, 1,
1330                                 oh->ot_tx);
1331         if (rc)
1332                 CERROR("%s: failed to add "DFID" to accounting ZAP for grp %d "
1333                         "(%d)\n", osd->od_svname, PFID(fid), attr->la_gid, rc);
1334
1335         /* configure new osd object */
1336         obj->oo_db = db;
1337         rc = osd_object_init0(env, obj);
1338         LASSERT(ergo(rc == 0, dt_object_exists(dt)));
1339         LASSERT(osd_invariant(obj));
1340
1341 out:
1342         cfs_up(&obj->oo_guard);
1343         RETURN(rc);
1344 }
1345
1346 static int osd_declare_object_ref_add(const struct lu_env *env,
1347                                       struct dt_object *dt,
1348                                       struct thandle *th)
1349 {
1350         return osd_declare_attr_set(env, dt, NULL, th);
1351 }
1352
1353 /*
1354  * Concurrency: @dt is write locked.
1355  */
1356 static int osd_object_ref_add(const struct lu_env *env,
1357                               struct dt_object *dt,
1358                               struct thandle *handle)
1359 {
1360         struct osd_object       *obj = osd_dt_obj(dt);
1361         struct osd_thandle      *oh;
1362         struct osd_device       *osd = osd_obj2dev(obj);
1363         udmu_objset_t           *uos = &osd->od_objset;
1364         uint64_t                 nlink;
1365         int rc;
1366
1367         ENTRY;
1368
1369         LASSERT(osd_invariant(obj));
1370         LASSERT(dt_object_exists(dt));
1371         LASSERT(obj->oo_sa_hdl != NULL);
1372
1373         oh = container_of0(handle, struct osd_thandle, ot_super);
1374
1375         cfs_write_lock(&obj->oo_attr_lock);
1376         nlink = ++obj->oo_attr.la_nlink;
1377         cfs_write_unlock(&obj->oo_attr_lock);
1378
1379         rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh);
1380         return rc;
1381 }
1382
1383 static int osd_declare_object_ref_del(const struct lu_env *env,
1384                                       struct dt_object *dt,
1385                                       struct thandle *handle)
1386 {
1387         return osd_declare_attr_set(env, dt, NULL, handle);
1388 }
1389
1390 /*
1391  * Concurrency: @dt is write locked.
1392  */
1393 static int osd_object_ref_del(const struct lu_env *env,
1394                               struct dt_object *dt,
1395                               struct thandle *handle)
1396 {
1397         struct osd_object       *obj = osd_dt_obj(dt);
1398         struct osd_thandle      *oh;
1399         struct osd_device       *osd = osd_obj2dev(obj);
1400         udmu_objset_t           *uos = &osd->od_objset;
1401         uint64_t                 nlink;
1402         int                      rc;
1403
1404         ENTRY;
1405
1406         LASSERT(osd_invariant(obj));
1407         LASSERT(dt_object_exists(dt));
1408         LASSERT(obj->oo_sa_hdl != NULL);
1409
1410         oh = container_of0(handle, struct osd_thandle, ot_super);
1411         LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));
1412
1413         cfs_write_lock(&obj->oo_attr_lock);
1414         nlink = --obj->oo_attr.la_nlink;
1415         cfs_write_unlock(&obj->oo_attr_lock);
1416
1417         rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh);
1418         return rc;
1419 }
1420
1421 static int capa_is_sane(const struct lu_env *env, struct osd_device *dev,
1422                         struct lustre_capa *capa, struct lustre_capa_key *keys)
1423 {
1424         struct osd_thread_info  *oti = osd_oti_get(env);
1425         struct obd_capa         *oc;
1426         int                      i, rc = 0;
1427         ENTRY;
1428
1429         oc = capa_lookup(dev->od_capa_hash, capa, 0);
1430         if (oc) {
1431                 if (capa_is_expired(oc)) {
1432                         DEBUG_CAPA(D_ERROR, capa, "expired");
1433                         rc = -ESTALE;
1434                 }
1435                 capa_put(oc);
1436                 RETURN(rc);
1437         }
1438
1439         cfs_spin_lock(&capa_lock);
1440         for (i = 0; i < 2; i++) {
1441                 if (keys[i].lk_keyid == capa->lc_keyid) {
1442                         oti->oti_capa_key = keys[i];
1443                         break;
1444                 }
1445         }
1446         cfs_spin_unlock(&capa_lock);
1447
1448         if (i == 2) {
1449                 DEBUG_CAPA(D_ERROR, capa, "no matched capa key");
1450                 RETURN(-ESTALE);
1451         }
1452
1453         rc = capa_hmac(oti->oti_capa.lc_hmac, capa, oti->oti_capa_key.lk_key);
1454         if (rc)
1455                 RETURN(rc);
1456         if (memcmp(oti->oti_capa.lc_hmac, capa->lc_hmac, sizeof(capa->lc_hmac)))
1457         {
1458                 DEBUG_CAPA(D_ERROR, capa, "HMAC mismatch");
1459                 RETURN(-EACCES);
1460         }
1461
1462         oc = capa_add(dev->od_capa_hash, capa);
1463         capa_put(oc);
1464
1465         RETURN(0);
1466 }
1467
1468 static int osd_object_auth(const struct lu_env *env, struct dt_object *dt,
1469                            struct lustre_capa *capa, __u64 opc)
1470 {
1471         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1472         struct osd_device       *dev = osd_dev(dt->do_lu.lo_dev);
1473         int                      rc;
1474
1475         if (!dev->od_fl_capa)
1476                 return 0;
1477
1478         if (capa == BYPASS_CAPA)
1479                 return 0;
1480
1481         if (!capa) {
1482                 CERROR("no capability is provided for fid "DFID"\n", PFID(fid));
1483                 return -EACCES;
1484         }
1485
1486         if (!lu_fid_eq(fid, &capa->lc_fid)) {
1487                 DEBUG_CAPA(D_ERROR, capa, "fid "DFID" mismatch with",PFID(fid));
1488                 return -EACCES;
1489         }
1490
1491         if (!capa_opc_supported(capa, opc)) {
1492                 DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
1493                 return -EACCES;
1494         }
1495
1496         if ((rc = capa_is_sane(env, dev, capa, dev->od_capa_keys))) {
1497                 DEBUG_CAPA(D_ERROR, capa, "insane (rc %d)", rc);
1498                 return -EACCES;
1499         }
1500
1501         return 0;
1502 }
1503
1504 static struct obd_capa *osd_capa_get(const struct lu_env *env,
1505                                      struct dt_object *dt,
1506                                      struct lustre_capa *old,
1507                                      __u64 opc)
1508 {
1509         struct osd_thread_info  *info = osd_oti_get(env);
1510         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1511         struct osd_object       *obj = osd_dt_obj(dt);
1512         struct osd_device       *dev = osd_obj2dev(obj);
1513         struct lustre_capa_key  *key = &info->oti_capa_key;
1514         struct lustre_capa      *capa = &info->oti_capa;
1515         struct obd_capa         *oc;
1516         int                      rc;
1517         ENTRY;
1518
1519         if (!dev->od_fl_capa)
1520                 RETURN(ERR_PTR(-ENOENT));
1521
1522         LASSERT(dt_object_exists(dt));
1523         LASSERT(osd_invariant(obj));
1524
1525         /* renewal sanity check */
1526         if (old && osd_object_auth(env, dt, old, opc))
1527                 RETURN(ERR_PTR(-EACCES));
1528
1529         capa->lc_fid = *fid;
1530         capa->lc_opc = opc;
1531         capa->lc_uid = 0;
1532         capa->lc_flags = dev->od_capa_alg << 24;
1533         capa->lc_timeout = dev->od_capa_timeout;
1534         capa->lc_expiry = 0;
1535
1536         oc = capa_lookup(dev->od_capa_hash, capa, 1);
1537         if (oc) {
1538                 LASSERT(!capa_is_expired(oc));
1539                 RETURN(oc);
1540         }
1541
1542         cfs_spin_lock(&capa_lock);
1543         *key = dev->od_capa_keys[1];
1544         cfs_spin_unlock(&capa_lock);
1545
1546         capa->lc_keyid = key->lk_keyid;
1547         capa->lc_expiry = cfs_time_current_sec() + dev->od_capa_timeout;
1548
1549         rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
1550         if (rc) {
1551                 DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
1552                 LBUG();
1553                 RETURN(ERR_PTR(rc));
1554         }
1555
1556         oc = capa_add(dev->od_capa_hash, capa);
1557         RETURN(oc);
1558 }
1559
1560 static int osd_object_sync(const struct lu_env *env, struct dt_object *dt)
1561 {
1562         struct osd_device *osd = osd_obj2dev(osd_dt_obj(dt));
1563         ENTRY;
1564
1565         /* XXX: no other option than syncing the whole filesystem until we
1566          * support ZIL */
1567         txg_wait_synced(dmu_objset_pool(osd->od_objset.os), 0ULL);
1568
1569         RETURN(0);
1570 }
1571
1572 static struct dt_object_operations osd_obj_ops = {
1573         .do_read_lock           = osd_object_read_lock,
1574         .do_write_lock          = osd_object_write_lock,
1575         .do_read_unlock         = osd_object_read_unlock,
1576         .do_write_unlock        = osd_object_write_unlock,
1577         .do_write_locked        = osd_object_write_locked,
1578         .do_attr_get            = osd_attr_get,
1579         .do_declare_attr_set    = osd_declare_attr_set,
1580         .do_attr_set            = osd_attr_set,
1581         .do_ah_init             = osd_ah_init,
1582         .do_declare_create      = osd_declare_object_create,
1583         .do_create              = osd_object_create,
1584         .do_declare_destroy     = osd_declare_object_destroy,
1585         .do_destroy             = osd_object_destroy,
1586         .do_index_try           = osd_index_try,
1587         .do_declare_ref_add     = osd_declare_object_ref_add,
1588         .do_ref_add             = osd_object_ref_add,
1589         .do_declare_ref_del     = osd_declare_object_ref_del,
1590         .do_ref_del             = osd_object_ref_del,
1591         .do_xattr_get           = osd_xattr_get,
1592         .do_declare_xattr_set   = osd_declare_xattr_set,
1593         .do_xattr_set           = osd_xattr_set,
1594         .do_declare_xattr_del   = osd_declare_xattr_del,
1595         .do_xattr_del           = osd_xattr_del,
1596         .do_xattr_list          = osd_xattr_list,
1597         .do_capa_get            = osd_capa_get,
1598         .do_object_sync         = osd_object_sync,
1599 };
1600
1601 static struct lu_object_operations osd_lu_obj_ops = {
1602         .loo_object_init        = osd_object_init,
1603         .loo_object_delete      = osd_object_delete,
1604         .loo_object_release     = osd_object_release,
1605         .loo_object_free        = osd_object_free,
1606         .loo_object_print       = osd_object_print,
1607         .loo_object_invariant   = osd_object_invariant,
1608 };
1609
1610 static int osd_otable_it_attr_get(const struct lu_env *env,
1611                                 struct dt_object *dt,
1612                                 struct lu_attr *attr,
1613                                 struct lustre_capa *capa)
1614 {
1615         attr->la_valid = 0;
1616         return 0;
1617 }
1618
1619 static struct dt_object_operations osd_obj_otable_it_ops = {
1620         .do_attr_get    = osd_otable_it_attr_get,
1621         .do_index_try   = osd_index_try,
1622 };
1623