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         int                      rc;
 462         ENTRY;
 463
 464         LASSERT(th != NULL);
 465         LASSERT(dt_object_exists(dt));
 466
 467         oh = container_of0(th, struct osd_thandle, ot_super);
 468         LASSERT(oh->ot_tx != NULL);
 469
 470         /* declare that we'll destroy the object */
 471         __osd_declare_object_destroy(env, obj, oh);
 472
 473         /* declare that we'll remove object from fid-dnode mapping */
 474         zapid = osd_get_name_n_idx(env, osd, fid, buf);
 475         dmu_tx_hold_bonus(oh->ot_tx, zapid);
 476         dmu_tx_hold_zap(oh->ot_tx, zapid, 0, buf);
 477
 478         /* declare that we'll remove object from inode accounting ZAPs */
 479         dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
 480         dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, 0, buf);
 481         dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
 482         dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, 0, buf);
 483
 484         /* one less inode */
 485         rc = osd_declare_quota(env, osd, obj->oo_attr.la_uid,
 486                                obj->oo_attr.la_gid, -1, oh, false, NULL, false);
 487         if (rc)
 488                 RETURN(rc);
 489
 490         /* data to be truncated */
 491         rc = osd_declare_quota(env, osd, obj->oo_attr.la_uid,
 492                                obj->oo_attr.la_gid, 0, oh, true, NULL, false);
 493         RETURN(rc);
 494 }
 495
 496 int __osd_object_free(udmu_objset_t *uos, uint64_t oid, dmu_tx_t *tx)
 497 {
 498         LASSERT(uos->objects != 0);
 499         cfs_spin_lock(&uos->lock);
 500         uos->objects--;
 501         cfs_spin_unlock(&uos->lock);
 502
 503         return -dmu_object_free(uos->os, oid, tx);
 504 }
 505
 506 /*
 507  * Delete a DMU object
 508  *
 509  * The transaction passed to this routine must have
 510  * dmu_tx_hold_free(tx, oid, 0, DMU_OBJECT_END) called
 511  * and then assigned to a transaction group.
 512  *
 513  * This will release db and set it to NULL to prevent further dbuf releases.
 514  */
 515 static int __osd_object_destroy(const struct lu_env *env,
 516                                 struct osd_object *obj,
 517                                 dmu_tx_t *tx, void *tag)
 518 {
 519         struct osd_device       *osd = osd_obj2dev(obj);
 520         udmu_objset_t           *uos = &osd->od_objset;
 521         uint64_t                 xid;
 522         zap_attribute_t         *za = &osd_oti_get(env)->oti_za;
 523         zap_cursor_t            *zc;
 524         int                      rc;
 525
 526         /* Assert that the transaction has been assigned to a
 527            transaction group. */
 528         LASSERT(tx->tx_txg != 0);
 529
 530         /* zap holding xattrs */
 531         if (obj->oo_xattr != ZFS_NO_OBJECT) {
 532                 rc = -udmu_zap_cursor_init(&zc, uos, obj->oo_xattr, 0);
 533                 if (rc)
 534                         return rc;
 535                 while ((rc = -zap_cursor_retrieve(zc, za)) == 0) {
 536                         BUG_ON(za->za_integer_length != sizeof(uint64_t));
 537                         BUG_ON(za->za_num_integers != 1);
 538
 539                         rc = -zap_lookup(uos->os, obj->oo_xattr, za->za_name,
 540                                          sizeof(uint64_t), 1, &xid);
 541                         if (rc) {
 542                                 CERROR("%s: lookup xattr %s failed: rc = %d\n",
 543                                        osd->od_svname, za->za_name, rc);
 544                                 continue;
 545                         }
 546                         rc = __osd_object_free(uos, xid, tx);
 547                         if (rc)
 548                                 CERROR("%s: fetch xattr %s failed: rc = %d\n",
 549                                        osd->od_svname, za->za_name, rc);
 550                         zap_cursor_advance(zc);
 551                 }
 552                 udmu_zap_cursor_fini(zc);
 553
 554                 rc = __osd_object_free(uos, obj->oo_xattr, tx);
 555                 if (rc)
 556                         CERROR("%s: freeing xattr failed: rc = %d\n",
 557                                osd->od_svname, rc);
 558         }
 559
 560         return __osd_object_free(uos, obj->oo_db->db_object, tx);
 561 }
 562
 563 static int osd_object_destroy(const struct lu_env *env,
 564                               struct dt_object *dt,
 565                               struct thandle *th)
 566 {
 567         char                    *buf = osd_oti_get(env)->oti_str;
 568         struct osd_object       *obj = osd_dt_obj(dt);
 569         struct osd_device       *osd = osd_obj2dev(obj);
 570         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
 571         struct osd_thandle      *oh;
 572         int                      rc;
 573         uint64_t                 zapid;
 574         ENTRY;
 575
 576         LASSERT(obj->oo_db != NULL);
 577         LASSERT(dt_object_exists(dt));
 578         LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));
 579
 580         oh = container_of0(th, struct osd_thandle, ot_super);
 581         LASSERT(oh != NULL);
 582         LASSERT(oh->ot_tx != NULL);
 583
 584         zapid = osd_get_name_n_idx(env, osd, fid, buf);
 585
 586         /* remove obj ref from index dir (it depends) */
 587         rc = -zap_remove(osd->od_objset.os, zapid, buf, oh->ot_tx);
 588         if (rc) {
 589                 CERROR("%s: zap_remove() failed: rc = %d\n",
 590                        osd->od_svname, rc);
 591                 GOTO(out, rc);
 592         }
 593
 594         /* Remove object from inode accounting. It is not fatal for the destroy
 595          * operation if something goes wrong while updating accounting, but we
 596          * still log an error message to notify the administrator */
 597         rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
 598                         obj->oo_attr.la_uid, -1, oh->ot_tx);
 599         if (rc)
 600                 CERROR("%s: failed to remove "DFID" from accounting ZAP for usr"
 601                         " %d: rc = %d\n", osd->od_svname, PFID(fid),
 602                         obj->oo_attr.la_uid, rc);
 603         rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
 604                                 obj->oo_attr.la_gid, -1, oh->ot_tx);
 605         if (rc)
 606                 CERROR("%s: failed to remove "DFID" from accounting ZAP for grp"
 607                         " %d: rc = %d\n", osd->od_svname, PFID(fid),
 608                         obj->oo_attr.la_gid, rc);
 609
 610         /* kill object */
 611         rc = __osd_object_destroy(env, obj, oh->ot_tx, osd_obj_tag);
 612         if (rc) {
 613                 CERROR("%s: __osd_object_destroy() failed: rc = %d\n",
 614                        osd->od_svname, rc);
 615                 GOTO(out, rc);
 616         }
 617
 618 out:
 619         /* not needed in the cache anymore */
 620         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
 621
 622         RETURN (0);
 623 }
 624
 625 static void osd_object_delete(const struct lu_env *env, struct lu_object *l)
 626 {
 627         struct osd_object *obj = osd_obj(l);
 628
 629         if (obj->oo_db != NULL) {
 630                 osd_object_sa_fini(obj);
 631                 if (obj->oo_sa_xattr) {
 632                         nvlist_free(obj->oo_sa_xattr);
 633                         obj->oo_sa_xattr = NULL;
 634                 }
 635                 sa_buf_rele(obj->oo_db, osd_obj_tag);
 636                 cfs_list_del(&obj->oo_sa_linkage);
 637                 obj->oo_db = NULL;
 638         }
 639 }
 640
 641 /*
 642  * Concurrency: ->loo_object_release() is called under site spin-lock.
 643  */
 644 static void osd_object_release(const struct lu_env *env,
 645                                struct lu_object *l)
 646 {
 647 }
 648
 649 /*
 650  * Concurrency: shouldn't matter.
 651  */
 652 static int osd_object_print(const struct lu_env *env, void *cookie,
 653                             lu_printer_t p, const struct lu_object *l)
 654 {
 655         struct osd_object *o = osd_obj(l);
 656
 657         return (*p)(env, cookie, LUSTRE_OSD_ZFS_NAME"-object@%p", o);
 658 }
 659
 660 static void osd_object_read_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_read(&obj->oo_sem);
 668 }
 669
 670 static void osd_object_write_lock(const struct lu_env *env,
 671                                   struct dt_object *dt, unsigned role)
 672 {
 673         struct osd_object *obj = osd_dt_obj(dt);
 674
 675         LASSERT(osd_invariant(obj));
 676
 677         cfs_down_write(&obj->oo_sem);
 678 }
 679
 680 static void osd_object_read_unlock(const struct lu_env *env,
 681                                    struct dt_object *dt)
 682 {
 683         struct osd_object *obj = osd_dt_obj(dt);
 684
 685         LASSERT(osd_invariant(obj));
 686         cfs_up_read(&obj->oo_sem);
 687 }
 688
 689 static void osd_object_write_unlock(const struct lu_env *env,
 690                                     struct dt_object *dt)
 691 {
 692         struct osd_object *obj = osd_dt_obj(dt);
 693
 694         LASSERT(osd_invariant(obj));
 695         cfs_up_write(&obj->oo_sem);
 696 }
 697
 698 static int osd_object_write_locked(const struct lu_env *env,
 699                                    struct dt_object *dt)
 700 {
 701         struct osd_object *obj = osd_dt_obj(dt);
 702         int rc = 1;
 703
 704         LASSERT(osd_invariant(obj));
 705
 706         if (cfs_down_write_trylock(&obj->oo_sem)) {
 707                 rc = 0;
 708                 cfs_up_write(&obj->oo_sem);
 709         }
 710         return rc;
 711 }
 712
 713 static int osd_attr_get(const struct lu_env *env,
 714                         struct dt_object *dt,
 715                         struct lu_attr *attr,
 716                         struct lustre_capa *capa)
 717 {
 718         struct osd_object       *obj = osd_dt_obj(dt);
 719         uint64_t                 blocks;
 720         uint32_t                 blksize;
 721
 722         LASSERT(dt_object_exists(dt));
 723         LASSERT(osd_invariant(obj));
 724         LASSERT(obj->oo_db);
 725
 726         cfs_read_lock(&obj->oo_attr_lock);
 727         *attr = obj->oo_attr;
 728         cfs_read_unlock(&obj->oo_attr_lock);
 729
 730         /* with ZFS_DEBUG zrl_add_debug() called by DB_DNODE_ENTER()
 731          * from within sa_object_size() can block on a mutex, so
 732          * we can't call sa_object_size() holding rwlock */
 733         sa_object_size(obj->oo_sa_hdl, &blksize, &blocks);
 734         /* Block size may be not set; suggest maximal I/O transfers. */
 735         if (blksize == 0)
 736                 blksize = 1ULL << SPA_MAXBLOCKSHIFT;
 737
 738         attr->la_blksize = blksize;
 739         attr->la_blocks = blocks;
 740         attr->la_valid |= LA_BLOCKS | LA_BLKSIZE;
 741
 742         return 0;
 743 }
 744
 745 /* Simple wrapper on top of qsd API which implement quota transfer for osd
 746  * setattr needs. As a reminder, only the root user can change ownership of
 747  * a file, that's why EDQUOT & EINPROGRESS errors are discarded */
 748 static inline int qsd_transfer(const struct lu_env *env,
 749                                struct qsd_instance *qsd,
 750                                struct lquota_trans *trans, int qtype,
 751                                __u64 orig_id, __u64 new_id, __u64 bspace,
 752                                struct lquota_id_info *qi)
 753 {
 754         int     rc;
 755
 756         if (unlikely(qsd == NULL))
 757                 return 0;
 758
 759         LASSERT(qtype >= 0 && qtype < MAXQUOTAS);
 760         qi->lqi_type = qtype;
 761
 762         /* inode accounting */
 763         qi->lqi_is_blk = false;
 764
 765         /* one more inode for the new owner ... */
 766         qi->lqi_id.qid_uid = new_id;
 767         qi->lqi_space      = 1;
 768         rc = qsd_op_begin(env, qsd, trans, qi, NULL);
 769         if (rc == -EDQUOT || rc == -EINPROGRESS)
 770                 rc = 0;
 771         if (rc)
 772                 return rc;
 773
 774         /* and one less inode for the current id */
 775         qi->lqi_id.qid_uid = orig_id;;
 776         qi->lqi_space      = -1;
 777         rc = qsd_op_begin(env, qsd, trans, qi, NULL);
 778         if (rc == -EDQUOT || rc == -EINPROGRESS)
 779                 rc = 0;
 780         if (rc)
 781                 return rc;
 782
 783         /* block accounting */
 784         qi->lqi_is_blk = true;
 785
 786         /* more blocks for the new owner ... */
 787         qi->lqi_id.qid_uid = new_id;
 788         qi->lqi_space      = bspace;
 789         rc = qsd_op_begin(env, qsd, trans, qi, NULL);
 790         if (rc == -EDQUOT || rc == -EINPROGRESS)
 791                 rc = 0;
 792         if (rc)
 793                 return rc;
 794
 795         /* and finally less blocks for the current owner */
 796         qi->lqi_id.qid_uid = orig_id;
 797         qi->lqi_space      = -bspace;
 798         rc = qsd_op_begin(env, qsd, trans, qi, NULL);
 799         if (rc == -EDQUOT || rc == -EINPROGRESS)
 800                 rc = 0;
 801         return rc;
 802 }
 803
 804 static int osd_declare_attr_set(const struct lu_env *env,
 805                                 struct dt_object *dt,
 806                                 const struct lu_attr *attr,
 807                                 struct thandle *handle)
 808 {
 809         struct osd_thread_info  *info = osd_oti_get(env);
 810         char                    *buf = osd_oti_get(env)->oti_str;
 811         struct osd_object       *obj = osd_dt_obj(dt);
 812         struct osd_device       *osd = osd_obj2dev(obj);
 813         struct osd_thandle      *oh;
 814         uint64_t                 bspace;
 815         uint32_t                 blksize;
 816         int                      rc;
 817         ENTRY;
 818
 819         if (!dt_object_exists(dt)) {
 820                 /* XXX: sanity check that object creation is declared */
 821                 RETURN(0);
 822         }
 823
 824         LASSERT(handle != NULL);
 825         LASSERT(osd_invariant(obj));
 826
 827         oh = container_of0(handle, struct osd_thandle, ot_super);
 828
 829         LASSERT(obj->oo_sa_hdl != NULL);
 830         dmu_tx_hold_sa(oh->ot_tx, obj->oo_sa_hdl, 0);
 831
 832         sa_object_size(obj->oo_sa_hdl, &blksize, &bspace);
 833         bspace = toqb(bspace * blksize);
 834
 835         if (attr && attr->la_valid & LA_UID) {
 836                 /* account for user inode tracking ZAP update */
 837                 dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
 838                 dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, TRUE, buf);
 839
 840                 /* quota enforcement for user */
 841                 if (attr->la_uid != obj->oo_attr.la_uid) {
 842                         rc = qsd_transfer(env, osd->od_quota_slave,
 843                                           &oh->ot_quota_trans, USRQUOTA,
 844                                           obj->oo_attr.la_uid, attr->la_uid,
 845                                           bspace, &info->oti_qi);
 846                         if (rc)
 847                                 RETURN(rc);
 848                 }
 849         }
 850         if (attr && attr->la_valid & LA_GID) {
 851                 /* account for user inode tracking ZAP update */
 852                 dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
 853                 dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, TRUE, buf);
 854
 855                 /* quota enforcement for group */
 856                 if (attr->la_gid != obj->oo_attr.la_gid) {
 857                         rc = qsd_transfer(env, osd->od_quota_slave,
 858                                           &oh->ot_quota_trans, GRPQUOTA,
 859                                           obj->oo_attr.la_gid, attr->la_gid,
 860                                           bspace, &info->oti_qi);
 861                         if (rc)
 862                                 RETURN(rc);
 863                 }
 864         }
 865
 866         RETURN(0);
 867 }
 868
 869 /*
 870  * Set the attributes of an object
 871  *
 872  * The transaction passed to this routine must have
 873  * dmu_tx_hold_bonus(tx, oid) called and then assigned
 874  * to a transaction group.
 875  */
 876 static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
 877                         const struct lu_attr *la, struct thandle *handle,
 878                         struct lustre_capa *capa)
 879 {
 880         struct osd_object       *obj = osd_dt_obj(dt);
 881         struct osd_device       *osd = osd_obj2dev(obj);
 882         udmu_objset_t           *uos = &osd->od_objset;
 883         struct osd_thandle      *oh;
 884         struct osa_attr         *osa = &osd_oti_get(env)->oti_osa;
 885         sa_bulk_attr_t          *bulk;
 886         int                      cnt;
 887         int                      rc = 0;
 888
 889         ENTRY;
 890         LASSERT(handle != NULL);
 891         LASSERT(dt_object_exists(dt));
 892         LASSERT(osd_invariant(obj));
 893         LASSERT(obj->oo_sa_hdl);
 894
 895         oh = container_of0(handle, struct osd_thandle, ot_super);
 896         /* Assert that the transaction has been assigned to a
 897            transaction group. */
 898         LASSERT(oh->ot_tx->tx_txg != 0);
 899
 900         if (la->la_valid == 0)
 901                 RETURN(0);
 902
 903         OBD_ALLOC(bulk, sizeof(sa_bulk_attr_t) * 10);
 904         if (bulk == NULL)
 905                 RETURN(-ENOMEM);
 906
 907         /* do both accounting updates outside oo_attr_lock below */
 908         if ((la->la_valid & LA_UID) && (la->la_uid != obj->oo_attr.la_uid)) {
 909                 /* Update user accounting. Failure isn't fatal, but we still
 910                  * log an error message */
 911                 rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
 912                                         la->la_uid, 1, oh->ot_tx);
 913                 if (rc)
 914                         CERROR("%s: failed to update accounting ZAP for user "
 915                                 "%d (%d)\n", osd->od_svname, la->la_uid, rc);
 916                 rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
 917                                         obj->oo_attr.la_uid, -1, oh->ot_tx);
 918                 if (rc)
 919                         CERROR("%s: failed to update accounting ZAP for user "
 920                                 "%d (%d)\n", osd->od_svname,
 921                                 obj->oo_attr.la_uid, rc);
 922         }
 923         if ((la->la_valid & LA_GID) && (la->la_gid != obj->oo_attr.la_gid)) {
 924                 /* Update group accounting. Failure isn't fatal, but we still
 925                  * log an error message */
 926                 rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
 927                                         la->la_gid, 1, oh->ot_tx);
 928                 if (rc)
 929                         CERROR("%s: failed to update accounting ZAP for user "
 930                                 "%d (%d)\n", osd->od_svname, la->la_gid, rc);
 931                 rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
 932                                         obj->oo_attr.la_gid, -1, oh->ot_tx);
 933                 if (rc)
 934                         CERROR("%s: failed to update accounting ZAP for user "
 935                                 "%d (%d)\n", osd->od_svname,
 936                                 obj->oo_attr.la_gid, rc);
 937         }
 938
 939         cfs_write_lock(&obj->oo_attr_lock);
 940         cnt = 0;
 941         if (la->la_valid & LA_ATIME) {
 942                 osa->atime[0] = obj->oo_attr.la_atime = la->la_atime;
 943                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(uos), NULL,
 944                                  osa->atime, 16);
 945         }
 946         if (la->la_valid & LA_MTIME) {
 947                 osa->mtime[0] = obj->oo_attr.la_mtime = la->la_mtime;
 948                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(uos), NULL,
 949                                  osa->mtime, 16);
 950         }
 951         if (la->la_valid & LA_CTIME) {
 952                 osa->ctime[0] = obj->oo_attr.la_ctime = la->la_ctime;
 953                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(uos), NULL,
 954                                  osa->ctime, 16);
 955         }
 956         if (la->la_valid & LA_MODE) {
 957                 /* mode is stored along with type, so read it first */
 958                 obj->oo_attr.la_mode = (obj->oo_attr.la_mode & S_IFMT) |
 959                         (la->la_mode & ~S_IFMT);
 960                 osa->mode = obj->oo_attr.la_mode;
 961                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(uos), NULL,
 962                                  &osa->mode, 8);
 963         }
 964         if (la->la_valid & LA_SIZE) {
 965                 osa->size = obj->oo_attr.la_size = la->la_size;
 966                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(uos), NULL,
 967                                  &osa->size, 8);
 968         }
 969         if (la->la_valid & LA_NLINK) {
 970                 osa->nlink = obj->oo_attr.la_nlink = la->la_nlink;
 971                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(uos), NULL,
 972                                  &osa->nlink, 8);
 973         }
 974         if (la->la_valid & LA_RDEV) {
 975                 osa->rdev = obj->oo_attr.la_rdev = la->la_rdev;
 976                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(uos), NULL,
 977                                  &osa->rdev, 8);
 978         }
 979         if (la->la_valid & LA_FLAGS) {
 980                 osa->flags = obj->oo_attr.la_flags = la->la_flags;
 981                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(uos), NULL,
 982                                  &osa->flags, 8);
 983         }
 984         if (la->la_valid & LA_UID) {
 985                 osa->uid = obj->oo_attr.la_uid = la->la_uid;
 986                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(uos), NULL,
 987                                  &osa->uid, 8);
 988         }
 989         if (la->la_valid & LA_GID) {
 990                 osa->gid = obj->oo_attr.la_gid = la->la_gid;
 991                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(uos), NULL,
 992                                  &osa->gid, 8);
 993         }
 994         obj->oo_attr.la_valid |= la->la_valid;
 995         cfs_write_unlock(&obj->oo_attr_lock);
 996
 997         rc = osd_object_sa_bulk_update(obj, bulk, cnt, oh);
 998
 999         OBD_FREE(bulk, sizeof(sa_bulk_attr_t) * 10);
1000         RETURN(rc);
1001 }
1002
1003 /*
1004  * Object creation.
1005  *
1006  * XXX temporary solution.
1007  */
1008
1009 static void osd_ah_init(const struct lu_env *env, struct dt_allocation_hint *ah,
1010                         struct dt_object *parent, struct dt_object *child,
1011                         cfs_umode_t child_mode)
1012 {
1013         LASSERT(ah);
1014
1015         memset(ah, 0, sizeof(*ah));
1016         ah->dah_mode = child_mode;
1017 }
1018
1019 static int osd_declare_object_create(const struct lu_env *env,
1020                                      struct dt_object *dt,
1021                                      struct lu_attr *attr,
1022                                      struct dt_allocation_hint *hint,
1023                                      struct dt_object_format *dof,
1024                                      struct thandle *handle)
1025 {
1026         char                    *buf = osd_oti_get(env)->oti_str;
1027         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1028         struct osd_object       *obj = osd_dt_obj(dt);
1029         struct osd_device       *osd = osd_obj2dev(obj);
1030         struct osd_thandle      *oh;
1031         uint64_t                 zapid;
1032         ENTRY;
1033
1034         LASSERT(dof);
1035
1036         switch (dof->dof_type) {
1037                 case DFT_REGULAR:
1038                 case DFT_SYM:
1039                 case DFT_NODE:
1040                         if (obj->oo_dt.do_body_ops == NULL)
1041                                 obj->oo_dt.do_body_ops = &osd_body_ops;
1042                         break;
1043                 default:
1044                         break;
1045         }
1046
1047         LASSERT(handle != NULL);
1048         oh = container_of0(handle, struct osd_thandle, ot_super);
1049         LASSERT(oh->ot_tx != NULL);
1050
1051         switch (dof->dof_type) {
1052                 case DFT_DIR:
1053                 case DFT_INDEX:
1054                         /* for zap create */
1055                         dmu_tx_hold_zap(oh->ot_tx, DMU_NEW_OBJECT, 1, NULL);
1056                         break;
1057                 case DFT_REGULAR:
1058                 case DFT_SYM:
1059                 case DFT_NODE:
1060                         /* first, we'll create new object */
1061                         dmu_tx_hold_bonus(oh->ot_tx, DMU_NEW_OBJECT);
1062                         break;
1063
1064                 default:
1065                         LBUG();
1066                         break;
1067         }
1068
1069         /* and we'll add it to some mapping */
1070         zapid = osd_get_name_n_idx(env, osd, fid, buf);
1071         dmu_tx_hold_bonus(oh->ot_tx, zapid);
1072         dmu_tx_hold_zap(oh->ot_tx, zapid, TRUE, buf);
1073
1074         /* we will also update inode accounting ZAPs */
1075         dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
1076         dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, TRUE, buf);
1077         dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
1078         dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, TRUE, buf);
1079
1080         dmu_tx_hold_sa_create(oh->ot_tx, ZFS_SA_BASE_ATTR_SIZE);
1081
1082         RETURN(osd_declare_quota(env, osd, attr->la_uid, attr->la_gid, 1, oh,
1083                                  false, NULL, false));
1084 }
1085
1086 int __osd_attr_init(const struct lu_env *env, udmu_objset_t *uos,
1087                     uint64_t oid, dmu_tx_t *tx, struct lu_attr *la)
1088 {
1089         sa_bulk_attr_t  *bulk;
1090         sa_handle_t     *sa_hdl;
1091         struct osa_attr *osa = &osd_oti_get(env)->oti_osa;
1092         uint64_t         gen;
1093         uint64_t         parent;
1094         uint64_t         crtime[2];
1095         timestruc_t      now;
1096         int              cnt;
1097         int              rc;
1098
1099         gethrestime(&now);
1100         gen = dmu_tx_get_txg(tx);
1101
1102         ZFS_TIME_ENCODE(&now, crtime);
1103         /* XXX: this should be real id of parent for ZPL access, but we have no
1104          * such info in OSD, probably it can be part of dt_object_format */
1105         parent = 0;
1106
1107         osa->atime[0] = la->la_atime;
1108         osa->ctime[0] = la->la_ctime;
1109         osa->mtime[0] = la->la_mtime;
1110         osa->mode = la->la_mode;
1111         osa->uid = la->la_uid;
1112         osa->gid = la->la_gid;
1113         osa->rdev = la->la_rdev;
1114         osa->nlink = la->la_nlink;
1115         osa->flags = la->la_flags;
1116         osa->size  = la->la_size;
1117
1118         /* Now add in all of the "SA" attributes */
1119         rc = -sa_handle_get(uos->os, oid, NULL, SA_HDL_PRIVATE, &sa_hdl);
1120         if (rc)
1121                 return rc;
1122
1123         OBD_ALLOC(bulk, sizeof(sa_bulk_attr_t) * 13);
1124         if (bulk == NULL) {
1125                 rc = -ENOMEM;
1126                 goto out;
1127         }
1128         /*
1129          * we need to create all SA below upon object create.
1130          *
1131          * XXX The attribute order matters since the accounting callback relies
1132          * on static offsets (i.e. SA_*_OFFSET, see zfs_space_delta_cb()) to
1133          * look up the UID/GID attributes. Moreover, the callback does not seem
1134          * to support the spill block.
1135          * We define attributes in the same order as SA_*_OFFSET in order to
1136          * work around the problem. See ORI-610.
1137          */
1138         cnt = 0;
1139         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(uos), NULL, &osa->mode, 8);
1140         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(uos), NULL, &osa->size, 8);
1141         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GEN(uos), NULL, &gen, 8);
1142         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(uos), NULL, &osa->uid, 8);
1143         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(uos), NULL, &osa->gid, 8);
1144         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PARENT(uos), NULL, &parent, 8);
1145         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(uos), NULL, &osa->flags, 8);
1146         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(uos), NULL, osa->atime, 16);
1147         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(uos), NULL, osa->mtime, 16);
1148         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(uos), NULL, osa->ctime, 16);
1149         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CRTIME(uos), NULL, crtime, 16);
1150         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(uos), NULL, &osa->nlink, 8);
1151         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(uos), NULL, &osa->rdev, 8);
1152
1153         rc = -sa_replace_all_by_template(sa_hdl, bulk, cnt, tx);
1154
1155         OBD_FREE(bulk, sizeof(sa_bulk_attr_t) * 13);
1156 out:
1157         sa_handle_destroy(sa_hdl);
1158         return rc;
1159 }
1160
1161 /*
1162  * The transaction passed to this routine must have
1163  * dmu_tx_hold_bonus(tx, DMU_NEW_OBJECT) called and then assigned
1164  * to a transaction group.
1165  */
1166 int __osd_object_create(const struct lu_env *env, udmu_objset_t *uos,
1167                         dmu_buf_t **dbp, dmu_tx_t *tx,
1168                         struct lu_attr *la, void *tag)
1169 {
1170         uint64_t oid;
1171         int      rc;
1172
1173         LASSERT(tag);
1174         cfs_spin_lock(&uos->lock);
1175         uos->objects++;
1176         cfs_spin_unlock(&uos->lock);
1177
1178         /* Assert that the transaction has been assigned to a
1179            transaction group. */
1180         LASSERT(tx->tx_txg != 0);
1181
1182         /* Create a new DMU object. */
1183         oid = dmu_object_alloc(uos->os, DMU_OT_PLAIN_FILE_CONTENTS, 0,
1184                                DMU_OT_SA, DN_MAX_BONUSLEN, tx);
1185         rc = -sa_buf_hold(uos->os, oid, tag, dbp);
1186         if (rc)
1187                 return rc;
1188
1189         LASSERT(la->la_valid & LA_MODE);
1190         la->la_size = 0;
1191         la->la_nlink = 1;
1192
1193         return __osd_attr_init(env, uos, oid, tx, la);
1194 }
1195
1196 /*
1197  * The transaction passed to this routine must have
1198  * dmu_tx_hold_zap(tx, DMU_NEW_OBJECT, ...) called and then assigned
1199  * to a transaction group.
1200  *
1201  * Using ZAP_FLAG_HASH64 will force the ZAP to always be a FAT ZAP.
1202  * This is fine for directories today, because storing the FID in the dirent
1203  * will also require a FAT ZAP.  If there is a new type of micro ZAP created
1204  * then we might need to re-evaluate the use of this flag and instead do
1205  * a conversion from the different internal ZAP hash formats being used. */
1206 int __osd_zap_create(const struct lu_env *env, udmu_objset_t *uos,
1207                      dmu_buf_t **zap_dbp, dmu_tx_t *tx,
1208                      struct lu_attr *la, void *tag, zap_flags_t flags)
1209 {
1210         uint64_t oid;
1211         int      rc;
1212
1213         LASSERT(tag);
1214
1215         cfs_spin_lock(&uos->lock);
1216         uos->objects++;
1217         cfs_spin_unlock(&uos->lock);
1218
1219         /* Assert that the transaction has been assigned to a
1220            transaction group. */
1221         LASSERT(tx->tx_txg != 0);
1222
1223         oid = zap_create_flags(uos->os, 0, flags | ZAP_FLAG_HASH64,
1224                                DMU_OT_DIRECTORY_CONTENTS, 12, 12,
1225                                DMU_OT_SA, DN_MAX_BONUSLEN, tx);
1226
1227         rc = -sa_buf_hold(uos->os, oid, tag, zap_dbp);
1228         if (rc)
1229                 return rc;
1230
1231         LASSERT(la->la_valid & LA_MODE);
1232         la->la_size = 2;
1233         la->la_nlink = 1;
1234
1235         return __osd_attr_init(env, uos, oid, tx, la);
1236 }
1237
1238 static dmu_buf_t *osd_mkidx(const struct lu_env *env, struct osd_device *osd,
1239                             struct lu_attr *la, struct osd_thandle *oh)
1240 {
1241         dmu_buf_t *db;
1242         int        rc;
1243
1244         /* Index file should be created as regular file in order not to confuse
1245          * ZPL which could interpret them as directory.
1246          * We set ZAP_FLAG_UINT64_KEY to let ZFS know than we are going to use
1247          * binary keys */
1248         LASSERT(S_ISREG(la->la_mode));
1249         rc = __osd_zap_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1250                               osd_obj_tag, ZAP_FLAG_UINT64_KEY);
1251         if (rc)
1252                 return ERR_PTR(rc);
1253         return db;
1254 }
1255
1256 static dmu_buf_t *osd_mkdir(const struct lu_env *env, struct osd_device *osd,
1257                             struct lu_attr *la, struct osd_thandle *oh)
1258 {
1259         dmu_buf_t *db;
1260         int        rc;
1261
1262         LASSERT(S_ISDIR(la->la_mode));
1263         rc = __osd_zap_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1264                               osd_obj_tag, 0);
1265         if (rc)
1266                 return ERR_PTR(rc);
1267         return db;
1268 }
1269
1270 static dmu_buf_t* osd_mkreg(const struct lu_env *env, struct osd_device *osd,
1271                             struct lu_attr *la, struct osd_thandle *oh)
1272 {
1273         dmu_buf_t *db;
1274         int         rc;
1275
1276         LASSERT(S_ISREG(la->la_mode));
1277         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1278                                  osd_obj_tag);
1279         if (rc)
1280                 return ERR_PTR(rc);
1281
1282         /*
1283          * XXX: a hack, OST to use bigger blocksize. we need
1284          * a method in OSD API to control this from OFD/MDD
1285          */
1286         if (!lu_device_is_md(osd2lu_dev(osd))) {
1287                 rc = -dmu_object_set_blocksize(osd->od_objset.os,
1288                                                db->db_object,
1289                                 128 << 10, 0, oh->ot_tx);
1290                 if (unlikely(rc)) {
1291                         CERROR("%s: can't change blocksize: %d\n",
1292                                osd->od_svname, rc);
1293                         return ERR_PTR(rc);
1294                 }
1295         }
1296
1297         return db;
1298 }
1299
1300 static dmu_buf_t *osd_mksym(const struct lu_env *env, struct osd_device *osd,
1301                             struct lu_attr *la, struct osd_thandle *oh)
1302 {
1303         dmu_buf_t *db;
1304         int        rc;
1305
1306         LASSERT(S_ISLNK(la->la_mode));
1307         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1308                                  osd_obj_tag);
1309         if (rc)
1310                 return ERR_PTR(rc);
1311         return db;
1312 }
1313
1314 static dmu_buf_t *osd_mknod(const struct lu_env *env, struct osd_device *osd,
1315                             struct lu_attr *la, struct osd_thandle *oh)
1316 {
1317         dmu_buf_t *db;
1318         int        rc;
1319
1320         la->la_valid = LA_MODE;
1321         if (S_ISCHR(la->la_mode) || S_ISBLK(la->la_mode))
1322                 la->la_valid |= LA_RDEV;
1323
1324         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1325                                  osd_obj_tag);
1326         if (rc)
1327                 return ERR_PTR(rc);
1328         return db;
1329 }
1330
1331 typedef dmu_buf_t *(*osd_obj_type_f)(const struct lu_env *env, struct osd_device *osd,
1332                                      struct lu_attr *la, struct osd_thandle *oh);
1333
1334 static osd_obj_type_f osd_create_type_f(enum dt_format_type type)
1335 {
1336         osd_obj_type_f result;
1337
1338         switch (type) {
1339         case DFT_DIR:
1340                 result = osd_mkdir;
1341                 break;
1342         case DFT_INDEX:
1343                 result = osd_mkidx;
1344                 break;
1345         case DFT_REGULAR:
1346                 result = osd_mkreg;
1347                 break;
1348         case DFT_SYM:
1349                 result = osd_mksym;
1350                 break;
1351         case DFT_NODE:
1352                 result = osd_mknod;
1353                 break;
1354         default:
1355                 LBUG();
1356                 break;
1357         }
1358         return result;
1359 }
1360
1361 /*
1362  * Primitives for directory (i.e. ZAP) handling
1363  */
1364
1365 /*
1366  * Concurrency: @dt is write locked.
1367  */
1368 static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
1369                              struct lu_attr *attr,
1370                              struct dt_allocation_hint *hint,
1371                              struct dt_object_format *dof,
1372                              struct thandle *th)
1373 {
1374         struct zpl_direntry     *zde = &osd_oti_get(env)->oti_zde.lzd_reg;
1375         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1376         struct osd_object       *obj = osd_dt_obj(dt);
1377         struct osd_device       *osd = osd_obj2dev(obj);
1378         char                    *buf = osd_oti_get(env)->oti_str;
1379         struct osd_thandle      *oh;
1380         dmu_buf_t               *db;
1381         uint64_t                 zapid;
1382         int                      rc;
1383
1384         ENTRY;
1385
1386         /* concurrent create declarations should not see
1387          * the object inconsistent (db, attr, etc).
1388          * in regular cases acquisition should be cheap */
1389         cfs_down(&obj->oo_guard);
1390
1391         LASSERT(osd_invariant(obj));
1392         LASSERT(!dt_object_exists(dt));
1393         LASSERT(dof != NULL);
1394
1395         LASSERT(th != NULL);
1396         oh = container_of0(th, struct osd_thandle, ot_super);
1397
1398         /*
1399          * XXX missing: Quote handling.
1400          */
1401
1402         LASSERT(obj->oo_db == NULL);
1403
1404         db = osd_create_type_f(dof->dof_type)(env, osd, attr, oh);
1405         if (IS_ERR(db))
1406                 GOTO(out, rc = PTR_ERR(th));
1407
1408         zde->zde_pad = 0;
1409         zde->zde_dnode = db->db_object;
1410         zde->zde_type = IFTODT(attr->la_mode & S_IFMT);
1411
1412         zapid = osd_get_name_n_idx(env, osd, fid, buf);
1413
1414         rc = -zap_add(osd->od_objset.os, zapid, buf, 8, 1, zde, oh->ot_tx);
1415         if (rc)
1416                 GOTO(out, rc);
1417
1418         /* Add new object to inode accounting.
1419          * Errors are not considered as fatal */
1420         rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
1421                                 (attr->la_valid & LA_UID) ? attr->la_uid : 0, 1,
1422                                 oh->ot_tx);
1423         if (rc)
1424                 CERROR("%s: failed to add "DFID" to accounting ZAP for usr %d "
1425                         "(%d)\n", osd->od_svname, PFID(fid), attr->la_uid, rc);
1426         rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
1427                                 (attr->la_valid & LA_GID) ? attr->la_gid : 0, 1,
1428                                 oh->ot_tx);
1429         if (rc)
1430                 CERROR("%s: failed to add "DFID" to accounting ZAP for grp %d "
1431                         "(%d)\n", osd->od_svname, PFID(fid), attr->la_gid, rc);
1432
1433         /* configure new osd object */
1434         obj->oo_db = db;
1435         rc = osd_object_init0(env, obj);
1436         LASSERT(ergo(rc == 0, dt_object_exists(dt)));
1437         LASSERT(osd_invariant(obj));
1438
1439 out:
1440         cfs_up(&obj->oo_guard);
1441         RETURN(rc);
1442 }
1443
1444 static int osd_declare_object_ref_add(const struct lu_env *env,
1445                                       struct dt_object *dt,
1446                                       struct thandle *th)
1447 {
1448         return osd_declare_attr_set(env, dt, NULL, th);
1449 }
1450
1451 /*
1452  * Concurrency: @dt is write locked.
1453  */
1454 static int osd_object_ref_add(const struct lu_env *env,
1455                               struct dt_object *dt,
1456                               struct thandle *handle)
1457 {
1458         struct osd_object       *obj = osd_dt_obj(dt);
1459         struct osd_thandle      *oh;
1460         struct osd_device       *osd = osd_obj2dev(obj);
1461         udmu_objset_t           *uos = &osd->od_objset;
1462         uint64_t                 nlink;
1463         int rc;
1464
1465         ENTRY;
1466
1467         LASSERT(osd_invariant(obj));
1468         LASSERT(dt_object_exists(dt));
1469         LASSERT(obj->oo_sa_hdl != NULL);
1470
1471         oh = container_of0(handle, struct osd_thandle, ot_super);
1472
1473         cfs_write_lock(&obj->oo_attr_lock);
1474         nlink = ++obj->oo_attr.la_nlink;
1475         cfs_write_unlock(&obj->oo_attr_lock);
1476
1477         rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh);
1478         return rc;
1479 }
1480
1481 static int osd_declare_object_ref_del(const struct lu_env *env,
1482                                       struct dt_object *dt,
1483                                       struct thandle *handle)
1484 {
1485         return osd_declare_attr_set(env, dt, NULL, handle);
1486 }
1487
1488 /*
1489  * Concurrency: @dt is write locked.
1490  */
1491 static int osd_object_ref_del(const struct lu_env *env,
1492                               struct dt_object *dt,
1493                               struct thandle *handle)
1494 {
1495         struct osd_object       *obj = osd_dt_obj(dt);
1496         struct osd_thandle      *oh;
1497         struct osd_device       *osd = osd_obj2dev(obj);
1498         udmu_objset_t           *uos = &osd->od_objset;
1499         uint64_t                 nlink;
1500         int                      rc;
1501
1502         ENTRY;
1503
1504         LASSERT(osd_invariant(obj));
1505         LASSERT(dt_object_exists(dt));
1506         LASSERT(obj->oo_sa_hdl != NULL);
1507
1508         oh = container_of0(handle, struct osd_thandle, ot_super);
1509         LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));
1510
1511         cfs_write_lock(&obj->oo_attr_lock);
1512         nlink = --obj->oo_attr.la_nlink;
1513         cfs_write_unlock(&obj->oo_attr_lock);
1514
1515         rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh);
1516         return rc;
1517 }
1518
1519 static int capa_is_sane(const struct lu_env *env, struct osd_device *dev,
1520                         struct lustre_capa *capa, struct lustre_capa_key *keys)
1521 {
1522         struct osd_thread_info  *oti = osd_oti_get(env);
1523         struct obd_capa         *oc;
1524         int                      i, rc = 0;
1525         ENTRY;
1526
1527         oc = capa_lookup(dev->od_capa_hash, capa, 0);
1528         if (oc) {
1529                 if (capa_is_expired(oc)) {
1530                         DEBUG_CAPA(D_ERROR, capa, "expired");
1531                         rc = -ESTALE;
1532                 }
1533                 capa_put(oc);
1534                 RETURN(rc);
1535         }
1536
1537         cfs_spin_lock(&capa_lock);
1538         for (i = 0; i < 2; i++) {
1539                 if (keys[i].lk_keyid == capa->lc_keyid) {
1540                         oti->oti_capa_key = keys[i];
1541                         break;
1542                 }
1543         }
1544         cfs_spin_unlock(&capa_lock);
1545
1546         if (i == 2) {
1547                 DEBUG_CAPA(D_ERROR, capa, "no matched capa key");
1548                 RETURN(-ESTALE);
1549         }
1550
1551         rc = capa_hmac(oti->oti_capa.lc_hmac, capa, oti->oti_capa_key.lk_key);
1552         if (rc)
1553                 RETURN(rc);
1554         if (memcmp(oti->oti_capa.lc_hmac, capa->lc_hmac, sizeof(capa->lc_hmac)))
1555         {
1556                 DEBUG_CAPA(D_ERROR, capa, "HMAC mismatch");
1557                 RETURN(-EACCES);
1558         }
1559
1560         oc = capa_add(dev->od_capa_hash, capa);
1561         capa_put(oc);
1562
1563         RETURN(0);
1564 }
1565
1566 static int osd_object_auth(const struct lu_env *env, struct dt_object *dt,
1567                            struct lustre_capa *capa, __u64 opc)
1568 {
1569         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1570         struct osd_device       *dev = osd_dev(dt->do_lu.lo_dev);
1571         int                      rc;
1572
1573         if (!dev->od_fl_capa)
1574                 return 0;
1575
1576         if (capa == BYPASS_CAPA)
1577                 return 0;
1578
1579         if (!capa) {
1580                 CERROR("no capability is provided for fid "DFID"\n", PFID(fid));
1581                 return -EACCES;
1582         }
1583
1584         if (!lu_fid_eq(fid, &capa->lc_fid)) {
1585                 DEBUG_CAPA(D_ERROR, capa, "fid "DFID" mismatch with",PFID(fid));
1586                 return -EACCES;
1587         }
1588
1589         if (!capa_opc_supported(capa, opc)) {
1590                 DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
1591                 return -EACCES;
1592         }
1593
1594         if ((rc = capa_is_sane(env, dev, capa, dev->od_capa_keys))) {
1595                 DEBUG_CAPA(D_ERROR, capa, "insane (rc %d)", rc);
1596                 return -EACCES;
1597         }
1598
1599         return 0;
1600 }
1601
1602 static struct obd_capa *osd_capa_get(const struct lu_env *env,
1603                                      struct dt_object *dt,
1604                                      struct lustre_capa *old,
1605                                      __u64 opc)
1606 {
1607         struct osd_thread_info  *info = osd_oti_get(env);
1608         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1609         struct osd_object       *obj = osd_dt_obj(dt);
1610         struct osd_device       *dev = osd_obj2dev(obj);
1611         struct lustre_capa_key  *key = &info->oti_capa_key;
1612         struct lustre_capa      *capa = &info->oti_capa;
1613         struct obd_capa         *oc;
1614         int                      rc;
1615         ENTRY;
1616
1617         if (!dev->od_fl_capa)
1618                 RETURN(ERR_PTR(-ENOENT));
1619
1620         LASSERT(dt_object_exists(dt));
1621         LASSERT(osd_invariant(obj));
1622
1623         /* renewal sanity check */
1624         if (old && osd_object_auth(env, dt, old, opc))
1625                 RETURN(ERR_PTR(-EACCES));
1626
1627         capa->lc_fid = *fid;
1628         capa->lc_opc = opc;
1629         capa->lc_uid = 0;
1630         capa->lc_flags = dev->od_capa_alg << 24;
1631         capa->lc_timeout = dev->od_capa_timeout;
1632         capa->lc_expiry = 0;
1633
1634         oc = capa_lookup(dev->od_capa_hash, capa, 1);
1635         if (oc) {
1636                 LASSERT(!capa_is_expired(oc));
1637                 RETURN(oc);
1638         }
1639
1640         cfs_spin_lock(&capa_lock);
1641         *key = dev->od_capa_keys[1];
1642         cfs_spin_unlock(&capa_lock);
1643
1644         capa->lc_keyid = key->lk_keyid;
1645         capa->lc_expiry = cfs_time_current_sec() + dev->od_capa_timeout;
1646
1647         rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
1648         if (rc) {
1649                 DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
1650                 LBUG();
1651                 RETURN(ERR_PTR(rc));
1652         }
1653
1654         oc = capa_add(dev->od_capa_hash, capa);
1655         RETURN(oc);
1656 }
1657
1658 static int osd_object_sync(const struct lu_env *env, struct dt_object *dt)
1659 {
1660         struct osd_device *osd = osd_obj2dev(osd_dt_obj(dt));
1661         ENTRY;
1662
1663         /* XXX: no other option than syncing the whole filesystem until we
1664          * support ZIL */
1665         txg_wait_synced(dmu_objset_pool(osd->od_objset.os), 0ULL);
1666
1667         RETURN(0);
1668 }
1669
1670 static struct dt_object_operations osd_obj_ops = {
1671         .do_read_lock           = osd_object_read_lock,
1672         .do_write_lock          = osd_object_write_lock,
1673         .do_read_unlock         = osd_object_read_unlock,
1674         .do_write_unlock        = osd_object_write_unlock,
1675         .do_write_locked        = osd_object_write_locked,
1676         .do_attr_get            = osd_attr_get,
1677         .do_declare_attr_set    = osd_declare_attr_set,
1678         .do_attr_set            = osd_attr_set,
1679         .do_ah_init             = osd_ah_init,
1680         .do_declare_create      = osd_declare_object_create,
1681         .do_create              = osd_object_create,
1682         .do_declare_destroy     = osd_declare_object_destroy,
1683         .do_destroy             = osd_object_destroy,
1684         .do_index_try           = osd_index_try,
1685         .do_declare_ref_add     = osd_declare_object_ref_add,
1686         .do_ref_add             = osd_object_ref_add,
1687         .do_declare_ref_del     = osd_declare_object_ref_del,
1688         .do_ref_del             = osd_object_ref_del,
1689         .do_xattr_get           = osd_xattr_get,
1690         .do_declare_xattr_set   = osd_declare_xattr_set,
1691         .do_xattr_set           = osd_xattr_set,
1692         .do_declare_xattr_del   = osd_declare_xattr_del,
1693         .do_xattr_del           = osd_xattr_del,
1694         .do_xattr_list          = osd_xattr_list,
1695         .do_capa_get            = osd_capa_get,
1696         .do_object_sync         = osd_object_sync,
1697 };
1698
1699 static struct lu_object_operations osd_lu_obj_ops = {
1700         .loo_object_init        = osd_object_init,
1701         .loo_object_delete      = osd_object_delete,
1702         .loo_object_release     = osd_object_release,
1703         .loo_object_free        = osd_object_free,
1704         .loo_object_print       = osd_object_print,
1705         .loo_object_invariant   = osd_object_invariant,
1706 };
1707
1708 static int osd_otable_it_attr_get(const struct lu_env *env,
1709                                 struct dt_object *dt,
1710                                 struct lu_attr *attr,
1711                                 struct lustre_capa *capa)
1712 {
1713         attr->la_valid = 0;
1714         return 0;
1715 }
1716
1717 static struct dt_object_operations osd_obj_otable_it_ops = {
1718         .do_attr_get    = osd_otable_it_attr_get,
1719         .do_index_try   = osd_index_try,
1720 };
1721