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, cfs_umode_t child_mode)
1011 {
1012         LASSERT(ah);
1013
1014         memset(ah, 0, sizeof(*ah));
1015         ah->dah_mode = child_mode;
1016 }
1017
1018 static int osd_declare_object_create(const struct lu_env *env,
1019                                      struct dt_object *dt,
1020                                      struct lu_attr *attr,
1021                                      struct dt_allocation_hint *hint,
1022                                      struct dt_object_format *dof,
1023                                      struct thandle *handle)
1024 {
1025         char                    *buf = osd_oti_get(env)->oti_str;
1026         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1027         struct osd_object       *obj = osd_dt_obj(dt);
1028         struct osd_device       *osd = osd_obj2dev(obj);
1029         struct osd_thandle      *oh;
1030         uint64_t                 zapid;
1031         ENTRY;
1032
1033         LASSERT(dof);
1034
1035         switch (dof->dof_type) {
1036                 case DFT_REGULAR:
1037                 case DFT_SYM:
1038                 case DFT_NODE:
1039                         if (obj->oo_dt.do_body_ops == NULL)
1040                                 obj->oo_dt.do_body_ops = &osd_body_ops;
1041                         break;
1042                 default:
1043                         break;
1044         }
1045
1046         LASSERT(handle != NULL);
1047         oh = container_of0(handle, struct osd_thandle, ot_super);
1048         LASSERT(oh->ot_tx != NULL);
1049
1050         switch (dof->dof_type) {
1051                 case DFT_DIR:
1052                 case DFT_INDEX:
1053                         /* for zap create */
1054                         dmu_tx_hold_zap(oh->ot_tx, DMU_NEW_OBJECT, 1, NULL);
1055                         break;
1056                 case DFT_REGULAR:
1057                 case DFT_SYM:
1058                 case DFT_NODE:
1059                         /* first, we'll create new object */
1060                         dmu_tx_hold_bonus(oh->ot_tx, DMU_NEW_OBJECT);
1061                         break;
1062
1063                 default:
1064                         LBUG();
1065                         break;
1066         }
1067
1068         /* and we'll add it to some mapping */
1069         zapid = osd_get_name_n_idx(env, osd, fid, buf);
1070         dmu_tx_hold_bonus(oh->ot_tx, zapid);
1071         dmu_tx_hold_zap(oh->ot_tx, zapid, TRUE, buf);
1072
1073         /* we will also update inode accounting ZAPs */
1074         dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
1075         dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, TRUE, buf);
1076         dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
1077         dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, TRUE, buf);
1078
1079         dmu_tx_hold_sa_create(oh->ot_tx, ZFS_SA_BASE_ATTR_SIZE);
1080
1081         RETURN(osd_declare_quota(env, osd, attr->la_uid, attr->la_gid, 1, oh,
1082                                  false, NULL, false));
1083 }
1084
1085 int __osd_attr_init(const struct lu_env *env, udmu_objset_t *uos,
1086                     uint64_t oid, dmu_tx_t *tx, struct lu_attr *la)
1087 {
1088         sa_bulk_attr_t  *bulk;
1089         sa_handle_t     *sa_hdl;
1090         struct osa_attr *osa = &osd_oti_get(env)->oti_osa;
1091         uint64_t         gen;
1092         uint64_t         parent;
1093         uint64_t         crtime[2];
1094         timestruc_t      now;
1095         int              cnt;
1096         int              rc;
1097
1098         gethrestime(&now);
1099         gen = dmu_tx_get_txg(tx);
1100
1101         ZFS_TIME_ENCODE(&now, crtime);
1102         /* XXX: this should be real id of parent for ZPL access, but we have no
1103          * such info in OSD, probably it can be part of dt_object_format */
1104         parent = 0;
1105
1106         osa->atime[0] = la->la_atime;
1107         osa->ctime[0] = la->la_ctime;
1108         osa->mtime[0] = la->la_mtime;
1109         osa->mode = la->la_mode;
1110         osa->uid = la->la_uid;
1111         osa->gid = la->la_gid;
1112         osa->rdev = la->la_rdev;
1113         osa->nlink = la->la_nlink;
1114         osa->flags = la->la_flags;
1115         osa->size  = la->la_size;
1116
1117         /* Now add in all of the "SA" attributes */
1118         rc = -sa_handle_get(uos->os, oid, NULL, SA_HDL_PRIVATE, &sa_hdl);
1119         if (rc)
1120                 return rc;
1121
1122         OBD_ALLOC(bulk, sizeof(sa_bulk_attr_t) * 13);
1123         if (bulk == NULL) {
1124                 rc = -ENOMEM;
1125                 goto out;
1126         }
1127         /*
1128          * we need to create all SA below upon object create.
1129          *
1130          * XXX The attribute order matters since the accounting callback relies
1131          * on static offsets (i.e. SA_*_OFFSET, see zfs_space_delta_cb()) to
1132          * look up the UID/GID attributes. Moreover, the callback does not seem
1133          * to support the spill block.
1134          * We define attributes in the same order as SA_*_OFFSET in order to
1135          * work around the problem. See ORI-610.
1136          */
1137         cnt = 0;
1138         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(uos), NULL, &osa->mode, 8);
1139         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(uos), NULL, &osa->size, 8);
1140         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GEN(uos), NULL, &gen, 8);
1141         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(uos), NULL, &osa->uid, 8);
1142         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(uos), NULL, &osa->gid, 8);
1143         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PARENT(uos), NULL, &parent, 8);
1144         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(uos), NULL, &osa->flags, 8);
1145         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(uos), NULL, osa->atime, 16);
1146         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(uos), NULL, osa->mtime, 16);
1147         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(uos), NULL, osa->ctime, 16);
1148         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CRTIME(uos), NULL, crtime, 16);
1149         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(uos), NULL, &osa->nlink, 8);
1150         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(uos), NULL, &osa->rdev, 8);
1151
1152         rc = -sa_replace_all_by_template(sa_hdl, bulk, cnt, tx);
1153
1154         OBD_FREE(bulk, sizeof(sa_bulk_attr_t) * 13);
1155 out:
1156         sa_handle_destroy(sa_hdl);
1157         return rc;
1158 }
1159
1160 /*
1161  * The transaction passed to this routine must have
1162  * dmu_tx_hold_bonus(tx, DMU_NEW_OBJECT) called and then assigned
1163  * to a transaction group.
1164  */
1165 int __osd_object_create(const struct lu_env *env, udmu_objset_t *uos,
1166                         dmu_buf_t **dbp, dmu_tx_t *tx,
1167                         struct lu_attr *la, void *tag)
1168 {
1169         uint64_t oid;
1170         int      rc;
1171
1172         LASSERT(tag);
1173         cfs_spin_lock(&uos->lock);
1174         uos->objects++;
1175         cfs_spin_unlock(&uos->lock);
1176
1177         /* Assert that the transaction has been assigned to a
1178            transaction group. */
1179         LASSERT(tx->tx_txg != 0);
1180
1181         /* Create a new DMU object. */
1182         oid = dmu_object_alloc(uos->os, DMU_OT_PLAIN_FILE_CONTENTS, 0,
1183                                DMU_OT_SA, DN_MAX_BONUSLEN, tx);
1184         rc = -sa_buf_hold(uos->os, oid, tag, dbp);
1185         if (rc)
1186                 return rc;
1187
1188         LASSERT(la->la_valid & LA_MODE);
1189         la->la_size = 0;
1190         la->la_nlink = 1;
1191
1192         return __osd_attr_init(env, uos, oid, tx, la);
1193 }
1194
1195 /*
1196  * The transaction passed to this routine must have
1197  * dmu_tx_hold_zap(tx, DMU_NEW_OBJECT, ...) called and then assigned
1198  * to a transaction group.
1199  *
1200  * Using ZAP_FLAG_HASH64 will force the ZAP to always be a FAT ZAP.
1201  * This is fine for directories today, because storing the FID in the dirent
1202  * will also require a FAT ZAP.  If there is a new type of micro ZAP created
1203  * then we might need to re-evaluate the use of this flag and instead do
1204  * a conversion from the different internal ZAP hash formats being used. */
1205 int __osd_zap_create(const struct lu_env *env, udmu_objset_t *uos,
1206                      dmu_buf_t **zap_dbp, dmu_tx_t *tx,
1207                      struct lu_attr *la, void *tag, zap_flags_t flags)
1208 {
1209         uint64_t oid;
1210         int      rc;
1211
1212         LASSERT(tag);
1213
1214         cfs_spin_lock(&uos->lock);
1215         uos->objects++;
1216         cfs_spin_unlock(&uos->lock);
1217
1218         /* Assert that the transaction has been assigned to a
1219            transaction group. */
1220         LASSERT(tx->tx_txg != 0);
1221
1222         oid = zap_create_flags(uos->os, 0, flags | ZAP_FLAG_HASH64,
1223                                DMU_OT_DIRECTORY_CONTENTS, 12, 12,
1224                                DMU_OT_SA, DN_MAX_BONUSLEN, tx);
1225
1226         rc = -sa_buf_hold(uos->os, oid, tag, zap_dbp);
1227         if (rc)
1228                 return rc;
1229
1230         LASSERT(la->la_valid & LA_MODE);
1231         la->la_size = 2;
1232         la->la_nlink = 1;
1233
1234         return __osd_attr_init(env, uos, oid, tx, la);
1235 }
1236
1237 static dmu_buf_t *osd_mkidx(const struct lu_env *env, struct osd_device *osd,
1238                             struct lu_attr *la, struct osd_thandle *oh)
1239 {
1240         dmu_buf_t *db;
1241         int        rc;
1242
1243         /* Index file should be created as regular file in order not to confuse
1244          * ZPL which could interpret them as directory.
1245          * We set ZAP_FLAG_UINT64_KEY to let ZFS know than we are going to use
1246          * binary keys */
1247         LASSERT(S_ISREG(la->la_mode));
1248         rc = __osd_zap_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1249                               osd_obj_tag, ZAP_FLAG_UINT64_KEY);
1250         if (rc)
1251                 return ERR_PTR(rc);
1252         return db;
1253 }
1254
1255 static dmu_buf_t *osd_mkdir(const struct lu_env *env, struct osd_device *osd,
1256                             struct lu_attr *la, struct osd_thandle *oh)
1257 {
1258         dmu_buf_t *db;
1259         int        rc;
1260
1261         LASSERT(S_ISDIR(la->la_mode));
1262         rc = __osd_zap_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1263                               osd_obj_tag, 0);
1264         if (rc)
1265                 return ERR_PTR(rc);
1266         return db;
1267 }
1268
1269 static dmu_buf_t* osd_mkreg(const struct lu_env *env, struct osd_device *osd,
1270                             struct lu_attr *la, struct osd_thandle *oh)
1271 {
1272         dmu_buf_t *db;
1273         int         rc;
1274
1275         LASSERT(S_ISREG(la->la_mode));
1276         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1277                                  osd_obj_tag);
1278         if (rc)
1279                 return ERR_PTR(rc);
1280
1281         /*
1282          * XXX: a hack, OST to use bigger blocksize. we need
1283          * a method in OSD API to control this from OFD/MDD
1284          */
1285         if (!lu_device_is_md(osd2lu_dev(osd))) {
1286                 rc = -dmu_object_set_blocksize(osd->od_objset.os,
1287                                                db->db_object,
1288                                 128 << 10, 0, oh->ot_tx);
1289                 if (unlikely(rc)) {
1290                         CERROR("%s: can't change blocksize: %d\n",
1291                                osd->od_svname, rc);
1292                         return ERR_PTR(rc);
1293                 }
1294         }
1295
1296         return db;
1297 }
1298
1299 static dmu_buf_t *osd_mksym(const struct lu_env *env, struct osd_device *osd,
1300                             struct lu_attr *la, struct osd_thandle *oh)
1301 {
1302         dmu_buf_t *db;
1303         int        rc;
1304
1305         LASSERT(S_ISLNK(la->la_mode));
1306         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1307                                  osd_obj_tag);
1308         if (rc)
1309                 return ERR_PTR(rc);
1310         return db;
1311 }
1312
1313 static dmu_buf_t *osd_mknod(const struct lu_env *env, struct osd_device *osd,
1314                             struct lu_attr *la, struct osd_thandle *oh)
1315 {
1316         dmu_buf_t *db;
1317         int        rc;
1318
1319         la->la_valid = LA_MODE;
1320         if (S_ISCHR(la->la_mode) || S_ISBLK(la->la_mode))
1321                 la->la_valid |= LA_RDEV;
1322
1323         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1324                                  osd_obj_tag);
1325         if (rc)
1326                 return ERR_PTR(rc);
1327         return db;
1328 }
1329
1330 typedef dmu_buf_t *(*osd_obj_type_f)(const struct lu_env *env, struct osd_device *osd,
1331                                      struct lu_attr *la, struct osd_thandle *oh);
1332
1333 static osd_obj_type_f osd_create_type_f(enum dt_format_type type)
1334 {
1335         osd_obj_type_f result;
1336
1337         switch (type) {
1338         case DFT_DIR:
1339                 result = osd_mkdir;
1340                 break;
1341         case DFT_INDEX:
1342                 result = osd_mkidx;
1343                 break;
1344         case DFT_REGULAR:
1345                 result = osd_mkreg;
1346                 break;
1347         case DFT_SYM:
1348                 result = osd_mksym;
1349                 break;
1350         case DFT_NODE:
1351                 result = osd_mknod;
1352                 break;
1353         default:
1354                 LBUG();
1355                 break;
1356         }
1357         return result;
1358 }
1359
1360 /*
1361  * Primitives for directory (i.e. ZAP) handling
1362  */
1363
1364 /*
1365  * Concurrency: @dt is write locked.
1366  */
1367 static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
1368                              struct lu_attr *attr,
1369                              struct dt_allocation_hint *hint,
1370                              struct dt_object_format *dof,
1371                              struct thandle *th)
1372 {
1373         struct zpl_direntry     *zde = &osd_oti_get(env)->oti_zde.lzd_reg;
1374         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1375         struct osd_object       *obj = osd_dt_obj(dt);
1376         struct osd_device       *osd = osd_obj2dev(obj);
1377         char                    *buf = osd_oti_get(env)->oti_str;
1378         struct osd_thandle      *oh;
1379         dmu_buf_t               *db;
1380         uint64_t                 zapid;
1381         int                      rc;
1382
1383         ENTRY;
1384
1385         /* concurrent create declarations should not see
1386          * the object inconsistent (db, attr, etc).
1387          * in regular cases acquisition should be cheap */
1388         cfs_down(&obj->oo_guard);
1389
1390         LASSERT(osd_invariant(obj));
1391         LASSERT(!dt_object_exists(dt));
1392         LASSERT(dof != NULL);
1393
1394         LASSERT(th != NULL);
1395         oh = container_of0(th, struct osd_thandle, ot_super);
1396
1397         /*
1398          * XXX missing: Quote handling.
1399          */
1400
1401         LASSERT(obj->oo_db == NULL);
1402
1403         db = osd_create_type_f(dof->dof_type)(env, osd, attr, oh);
1404         if (IS_ERR(db))
1405                 GOTO(out, rc = PTR_ERR(th));
1406
1407         zde->zde_pad = 0;
1408         zde->zde_dnode = db->db_object;
1409         zde->zde_type = IFTODT(attr->la_mode & S_IFMT);
1410
1411         zapid = osd_get_name_n_idx(env, osd, fid, buf);
1412
1413         rc = -zap_add(osd->od_objset.os, zapid, buf, 8, 1, zde, oh->ot_tx);
1414         if (rc)
1415                 GOTO(out, rc);
1416
1417         /* Add new object to inode accounting.
1418          * Errors are not considered as fatal */
1419         rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
1420                                 (attr->la_valid & LA_UID) ? attr->la_uid : 0, 1,
1421                                 oh->ot_tx);
1422         if (rc)
1423                 CERROR("%s: failed to add "DFID" to accounting ZAP for usr %d "
1424                         "(%d)\n", osd->od_svname, PFID(fid), attr->la_uid, rc);
1425         rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
1426                                 (attr->la_valid & LA_GID) ? attr->la_gid : 0, 1,
1427                                 oh->ot_tx);
1428         if (rc)
1429                 CERROR("%s: failed to add "DFID" to accounting ZAP for grp %d "
1430                         "(%d)\n", osd->od_svname, PFID(fid), attr->la_gid, rc);
1431
1432         /* configure new osd object */
1433         obj->oo_db = db;
1434         rc = osd_object_init0(env, obj);
1435         LASSERT(ergo(rc == 0, dt_object_exists(dt)));
1436         LASSERT(osd_invariant(obj));
1437
1438 out:
1439         cfs_up(&obj->oo_guard);
1440         RETURN(rc);
1441 }
1442
1443 static int osd_declare_object_ref_add(const struct lu_env *env,
1444                                       struct dt_object *dt,
1445                                       struct thandle *th)
1446 {
1447         return osd_declare_attr_set(env, dt, NULL, th);
1448 }
1449
1450 /*
1451  * Concurrency: @dt is write locked.
1452  */
1453 static int osd_object_ref_add(const struct lu_env *env,
1454                               struct dt_object *dt,
1455                               struct thandle *handle)
1456 {
1457         struct osd_object       *obj = osd_dt_obj(dt);
1458         struct osd_thandle      *oh;
1459         struct osd_device       *osd = osd_obj2dev(obj);
1460         udmu_objset_t           *uos = &osd->od_objset;
1461         uint64_t                 nlink;
1462         int rc;
1463
1464         ENTRY;
1465
1466         LASSERT(osd_invariant(obj));
1467         LASSERT(dt_object_exists(dt));
1468         LASSERT(obj->oo_sa_hdl != NULL);
1469
1470         oh = container_of0(handle, struct osd_thandle, ot_super);
1471
1472         cfs_write_lock(&obj->oo_attr_lock);
1473         nlink = ++obj->oo_attr.la_nlink;
1474         cfs_write_unlock(&obj->oo_attr_lock);
1475
1476         rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh);
1477         return rc;
1478 }
1479
1480 static int osd_declare_object_ref_del(const struct lu_env *env,
1481                                       struct dt_object *dt,
1482                                       struct thandle *handle)
1483 {
1484         return osd_declare_attr_set(env, dt, NULL, handle);
1485 }
1486
1487 /*
1488  * Concurrency: @dt is write locked.
1489  */
1490 static int osd_object_ref_del(const struct lu_env *env,
1491                               struct dt_object *dt,
1492                               struct thandle *handle)
1493 {
1494         struct osd_object       *obj = osd_dt_obj(dt);
1495         struct osd_thandle      *oh;
1496         struct osd_device       *osd = osd_obj2dev(obj);
1497         udmu_objset_t           *uos = &osd->od_objset;
1498         uint64_t                 nlink;
1499         int                      rc;
1500
1501         ENTRY;
1502
1503         LASSERT(osd_invariant(obj));
1504         LASSERT(dt_object_exists(dt));
1505         LASSERT(obj->oo_sa_hdl != NULL);
1506
1507         oh = container_of0(handle, struct osd_thandle, ot_super);
1508         LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));
1509
1510         cfs_write_lock(&obj->oo_attr_lock);
1511         nlink = --obj->oo_attr.la_nlink;
1512         cfs_write_unlock(&obj->oo_attr_lock);
1513
1514         rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh);
1515         return rc;
1516 }
1517
1518 static int capa_is_sane(const struct lu_env *env, struct osd_device *dev,
1519                         struct lustre_capa *capa, struct lustre_capa_key *keys)
1520 {
1521         struct osd_thread_info  *oti = osd_oti_get(env);
1522         struct obd_capa         *oc;
1523         int                      i, rc = 0;
1524         ENTRY;
1525
1526         oc = capa_lookup(dev->od_capa_hash, capa, 0);
1527         if (oc) {
1528                 if (capa_is_expired(oc)) {
1529                         DEBUG_CAPA(D_ERROR, capa, "expired");
1530                         rc = -ESTALE;
1531                 }
1532                 capa_put(oc);
1533                 RETURN(rc);
1534         }
1535
1536         cfs_spin_lock(&capa_lock);
1537         for (i = 0; i < 2; i++) {
1538                 if (keys[i].lk_keyid == capa->lc_keyid) {
1539                         oti->oti_capa_key = keys[i];
1540                         break;
1541                 }
1542         }
1543         cfs_spin_unlock(&capa_lock);
1544
1545         if (i == 2) {
1546                 DEBUG_CAPA(D_ERROR, capa, "no matched capa key");
1547                 RETURN(-ESTALE);
1548         }
1549
1550         rc = capa_hmac(oti->oti_capa.lc_hmac, capa, oti->oti_capa_key.lk_key);
1551         if (rc)
1552                 RETURN(rc);
1553         if (memcmp(oti->oti_capa.lc_hmac, capa->lc_hmac, sizeof(capa->lc_hmac)))
1554         {
1555                 DEBUG_CAPA(D_ERROR, capa, "HMAC mismatch");
1556                 RETURN(-EACCES);
1557         }
1558
1559         oc = capa_add(dev->od_capa_hash, capa);
1560         capa_put(oc);
1561
1562         RETURN(0);
1563 }
1564
1565 static int osd_object_auth(const struct lu_env *env, struct dt_object *dt,
1566                            struct lustre_capa *capa, __u64 opc)
1567 {
1568         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1569         struct osd_device       *dev = osd_dev(dt->do_lu.lo_dev);
1570         int                      rc;
1571
1572         if (!dev->od_fl_capa)
1573                 return 0;
1574
1575         if (capa == BYPASS_CAPA)
1576                 return 0;
1577
1578         if (!capa) {
1579                 CERROR("no capability is provided for fid "DFID"\n", PFID(fid));
1580                 return -EACCES;
1581         }
1582
1583         if (!lu_fid_eq(fid, &capa->lc_fid)) {
1584                 DEBUG_CAPA(D_ERROR, capa, "fid "DFID" mismatch with",PFID(fid));
1585                 return -EACCES;
1586         }
1587
1588         if (!capa_opc_supported(capa, opc)) {
1589                 DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
1590                 return -EACCES;
1591         }
1592
1593         if ((rc = capa_is_sane(env, dev, capa, dev->od_capa_keys))) {
1594                 DEBUG_CAPA(D_ERROR, capa, "insane (rc %d)", rc);
1595                 return -EACCES;
1596         }
1597
1598         return 0;
1599 }
1600
1601 static struct obd_capa *osd_capa_get(const struct lu_env *env,
1602                                      struct dt_object *dt,
1603                                      struct lustre_capa *old,
1604                                      __u64 opc)
1605 {
1606         struct osd_thread_info  *info = osd_oti_get(env);
1607         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1608         struct osd_object       *obj = osd_dt_obj(dt);
1609         struct osd_device       *dev = osd_obj2dev(obj);
1610         struct lustre_capa_key  *key = &info->oti_capa_key;
1611         struct lustre_capa      *capa = &info->oti_capa;
1612         struct obd_capa         *oc;
1613         int                      rc;
1614         ENTRY;
1615
1616         if (!dev->od_fl_capa)
1617                 RETURN(ERR_PTR(-ENOENT));
1618
1619         LASSERT(dt_object_exists(dt));
1620         LASSERT(osd_invariant(obj));
1621
1622         /* renewal sanity check */
1623         if (old && osd_object_auth(env, dt, old, opc))
1624                 RETURN(ERR_PTR(-EACCES));
1625
1626         capa->lc_fid = *fid;
1627         capa->lc_opc = opc;
1628         capa->lc_uid = 0;
1629         capa->lc_flags = dev->od_capa_alg << 24;
1630         capa->lc_timeout = dev->od_capa_timeout;
1631         capa->lc_expiry = 0;
1632
1633         oc = capa_lookup(dev->od_capa_hash, capa, 1);
1634         if (oc) {
1635                 LASSERT(!capa_is_expired(oc));
1636                 RETURN(oc);
1637         }
1638
1639         cfs_spin_lock(&capa_lock);
1640         *key = dev->od_capa_keys[1];
1641         cfs_spin_unlock(&capa_lock);
1642
1643         capa->lc_keyid = key->lk_keyid;
1644         capa->lc_expiry = cfs_time_current_sec() + dev->od_capa_timeout;
1645
1646         rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
1647         if (rc) {
1648                 DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
1649                 LBUG();
1650                 RETURN(ERR_PTR(rc));
1651         }
1652
1653         oc = capa_add(dev->od_capa_hash, capa);
1654         RETURN(oc);
1655 }
1656
1657 static int osd_object_sync(const struct lu_env *env, struct dt_object *dt)
1658 {
1659         struct osd_device *osd = osd_obj2dev(osd_dt_obj(dt));
1660         ENTRY;
1661
1662         /* XXX: no other option than syncing the whole filesystem until we
1663          * support ZIL */
1664         txg_wait_synced(dmu_objset_pool(osd->od_objset.os), 0ULL);
1665
1666         RETURN(0);
1667 }
1668
1669 static struct dt_object_operations osd_obj_ops = {
1670         .do_read_lock           = osd_object_read_lock,
1671         .do_write_lock          = osd_object_write_lock,
1672         .do_read_unlock         = osd_object_read_unlock,
1673         .do_write_unlock        = osd_object_write_unlock,
1674         .do_write_locked        = osd_object_write_locked,
1675         .do_attr_get            = osd_attr_get,
1676         .do_declare_attr_set    = osd_declare_attr_set,
1677         .do_attr_set            = osd_attr_set,
1678         .do_ah_init             = osd_ah_init,
1679         .do_declare_create      = osd_declare_object_create,
1680         .do_create              = osd_object_create,
1681         .do_declare_destroy     = osd_declare_object_destroy,
1682         .do_destroy             = osd_object_destroy,
1683         .do_index_try           = osd_index_try,
1684         .do_declare_ref_add     = osd_declare_object_ref_add,
1685         .do_ref_add             = osd_object_ref_add,
1686         .do_declare_ref_del     = osd_declare_object_ref_del,
1687         .do_ref_del             = osd_object_ref_del,
1688         .do_xattr_get           = osd_xattr_get,
1689         .do_declare_xattr_set   = osd_declare_xattr_set,
1690         .do_xattr_set           = osd_xattr_set,
1691         .do_declare_xattr_del   = osd_declare_xattr_del,
1692         .do_xattr_del           = osd_xattr_del,
1693         .do_xattr_list          = osd_xattr_list,
1694         .do_capa_get            = osd_capa_get,
1695         .do_object_sync         = osd_object_sync,
1696 };
1697
1698 static struct lu_object_operations osd_lu_obj_ops = {
1699         .loo_object_init        = osd_object_init,
1700         .loo_object_delete      = osd_object_delete,
1701         .loo_object_release     = osd_object_release,
1702         .loo_object_free        = osd_object_free,
1703         .loo_object_print       = osd_object_print,
1704         .loo_object_invariant   = osd_object_invariant,
1705 };
1706
1707 static int osd_otable_it_attr_get(const struct lu_env *env,
1708                                 struct dt_object *dt,
1709                                 struct lu_attr *attr,
1710                                 struct lustre_capa *capa)
1711 {
1712         attr->la_valid = 0;
1713         return 0;
1714 }
1715
1716 static struct dt_object_operations osd_obj_otable_it_ops = {
1717         .do_attr_get    = osd_otable_it_attr_get,
1718         .do_index_try   = osd_index_try,
1719 };
1720