Whamcloud - gitweb
LU-4833 osd-zfs: object leak in __osd_object_create
[fs/lustre-release.git] / 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) 2012, 2013, Intel Corporation.
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 #define DEBUG_SUBSYSTEM S_OSD
46
47 #include <lustre_ver.h>
48 #include <libcfs/libcfs.h>
49 #include <obd_support.h>
50 #include <lustre_net.h>
51 #include <obd.h>
52 #include <obd_class.h>
53 #include <lustre_disk.h>
54 #include <lustre_fid.h>
55
56 #include "osd_internal.h"
57
58 #include <sys/dnode.h>
59 #include <sys/dbuf.h>
60 #include <sys/spa.h>
61 #include <sys/stat.h>
62 #include <sys/zap.h>
63 #include <sys/spa_impl.h>
64 #include <sys/zfs_znode.h>
65 #include <sys/dmu_tx.h>
66 #include <sys/dmu_objset.h>
67 #include <sys/dsl_prop.h>
68 #include <sys/sa_impl.h>
69 #include <sys/txg.h>
70
71 static char *osd_obj_tag = "osd_object";
72
73 static struct dt_object_operations osd_obj_ops;
74 static struct lu_object_operations osd_lu_obj_ops;
75 extern struct dt_body_operations osd_body_ops;
76 static struct dt_object_operations osd_obj_otable_it_ops;
77
78 extern struct kmem_cache *osd_object_kmem;
79
80 static void
81 osd_object_sa_fini(struct osd_object *obj)
82 {
83         if (obj->oo_sa_hdl) {
84                 sa_handle_destroy(obj->oo_sa_hdl);
85                 obj->oo_sa_hdl = NULL;
86         }
87 }
88
89 static int
90 osd_object_sa_init(struct osd_object *obj, udmu_objset_t *uos)
91 {
92         int rc;
93
94         LASSERT(obj->oo_sa_hdl == NULL);
95         LASSERT(obj->oo_db != NULL);
96
97         rc = -sa_handle_get(uos->os, obj->oo_db->db_object, obj,
98                             SA_HDL_PRIVATE, &obj->oo_sa_hdl);
99         if (rc)
100                 return rc;
101
102         /* Cache the xattr object id, valid for the life of the object */
103         rc = -sa_lookup(obj->oo_sa_hdl, SA_ZPL_XATTR(uos), &obj->oo_xattr, 8);
104         if (rc == -ENOENT) {
105                 obj->oo_xattr = ZFS_NO_OBJECT;
106                 rc = 0;
107         } else if (rc) {
108                 osd_object_sa_fini(obj);
109         }
110
111         return rc;
112 }
113
114 /*
115  * Add object to list of dirty objects in tx handle.
116  */
117 static void
118 osd_object_sa_dirty_add(struct osd_object *obj, struct osd_thandle *oh)
119 {
120         if (!cfs_list_empty(&obj->oo_sa_linkage))
121                 return;
122
123         down(&oh->ot_sa_lock);
124         write_lock(&obj->oo_attr_lock);
125         if (likely(cfs_list_empty(&obj->oo_sa_linkage)))
126                 cfs_list_add(&obj->oo_sa_linkage, &oh->ot_sa_list);
127         write_unlock(&obj->oo_attr_lock);
128         up(&oh->ot_sa_lock);
129 }
130
131 /*
132  * Release spill block dbuf hold for all dirty SAs.
133  */
134 void osd_object_sa_dirty_rele(struct osd_thandle *oh)
135 {
136         struct osd_object *obj;
137
138         down(&oh->ot_sa_lock);
139         while (!cfs_list_empty(&oh->ot_sa_list)) {
140                 obj = cfs_list_entry(oh->ot_sa_list.next,
141                                      struct osd_object, oo_sa_linkage);
142                 sa_spill_rele(obj->oo_sa_hdl);
143                 write_lock(&obj->oo_attr_lock);
144                 cfs_list_del_init(&obj->oo_sa_linkage);
145                 write_unlock(&obj->oo_attr_lock);
146         }
147         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 = attrs_zfs2fs(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, GFP_NOFS);
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                 init_rwsem(&mo->oo_sem);
299                 sema_init(&mo->oo_guard, 1);
300                 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         smp_mb();
343         obj->oo_dt.do_lu.lo_header->loh_attr |= LOHA_EXISTS;
344
345         RETURN(0);
346 }
347
348 static int osd_check_lma(const struct lu_env *env, struct osd_object *obj)
349 {
350         struct osd_thread_info  *info = osd_oti_get(env);
351         struct lu_buf           buf;
352         int                     rc;
353         struct lustre_mdt_attrs *lma;
354         ENTRY;
355
356         CLASSERT(sizeof(info->oti_buf) >= sizeof(*lma));
357         lma = (struct lustre_mdt_attrs *)info->oti_buf;
358         buf.lb_buf = lma;
359         buf.lb_len = sizeof(info->oti_buf);
360
361         rc = osd_xattr_get(env, &obj->oo_dt, &buf, XATTR_NAME_LMA, BYPASS_CAPA);
362         if (rc > 0) {
363                 rc = 0;
364                 lustre_lma_swab(lma);
365                 if (unlikely((lma->lma_incompat & ~LMA_INCOMPAT_SUPP) ||
366                              CFS_FAIL_CHECK(OBD_FAIL_OSD_LMA_INCOMPAT))) {
367                         CWARN("%s: unsupported incompat LMA feature(s) %#x for "
368                               "fid = "DFID"\n", osd_obj2dev(obj)->od_svname,
369                               lma->lma_incompat & ~LMA_INCOMPAT_SUPP,
370                               PFID(lu_object_fid(&obj->oo_dt.do_lu)));
371                         rc = -EOPNOTSUPP;
372                 }
373         } else if (rc == -ENODATA) {
374                 /* haven't initialize LMA xattr */
375                 rc = 0;
376         }
377
378         RETURN(rc);
379 }
380
381 /*
382  * Concurrency: no concurrent access is possible that early in object
383  * life-cycle.
384  */
385 static int osd_object_init(const struct lu_env *env, struct lu_object *l,
386                            const struct lu_object_conf *conf)
387 {
388         struct osd_object       *obj = osd_obj(l);
389         struct osd_device       *osd = osd_obj2dev(obj);
390         uint64_t                 oid;
391         int                      rc;
392         ENTRY;
393
394         LASSERT(osd_invariant(obj));
395
396         if (fid_is_otable_it(&l->lo_header->loh_fid)) {
397                 obj->oo_dt.do_ops = &osd_obj_otable_it_ops;
398                 l->lo_header->loh_attr |= LOHA_EXISTS;
399                 RETURN(0);
400         }
401
402         rc = osd_fid_lookup(env, osd, lu_object_fid(l), &oid);
403         if (rc == 0) {
404                 LASSERT(obj->oo_db == NULL);
405                 rc = __osd_obj2dbuf(env, osd->od_objset.os, oid,
406                                         &obj->oo_db, osd_obj_tag);
407                 if (rc != 0) {
408                         CERROR("%s: lookup "DFID"/"LPX64" failed: rc = %d\n",
409                                osd->od_svname, PFID(lu_object_fid(l)), oid, rc);
410                         GOTO(out, rc);
411                 }
412                 LASSERT(obj->oo_db);
413                 rc = osd_object_init0(env, obj);
414                 if (rc != 0)
415                         GOTO(out, rc);
416
417                 rc = osd_check_lma(env, obj);
418                 if (rc != 0)
419                         GOTO(out, rc);
420         } else if (rc == -ENOENT) {
421                 rc = 0;
422         }
423         LASSERT(osd_invariant(obj));
424 out:
425         RETURN(rc);
426 }
427
428 /*
429  * Concurrency: no concurrent access is possible that late in object
430  * life-cycle.
431  */
432 static void osd_object_free(const struct lu_env *env, struct lu_object *l)
433 {
434         struct osd_object *obj = osd_obj(l);
435
436         LASSERT(osd_invariant(obj));
437
438         dt_object_fini(&obj->oo_dt);
439         OBD_SLAB_FREE_PTR(obj, osd_object_kmem);
440 }
441
442 static void __osd_declare_object_destroy(const struct lu_env *env,
443                                          struct osd_object *obj,
444                                          struct osd_thandle *oh)
445 {
446         struct osd_device       *osd = osd_obj2dev(obj);
447         udmu_objset_t           *uos = &osd->od_objset;
448         dmu_buf_t               *db = obj->oo_db;
449         zap_attribute_t         *za = &osd_oti_get(env)->oti_za;
450         uint64_t                 oid = db->db_object, xid;
451         dmu_tx_t                *tx = oh->ot_tx;
452         zap_cursor_t            *zc;
453         int                      rc = 0;
454
455         dmu_tx_hold_free(tx, oid, 0, DMU_OBJECT_END);
456
457         /* zap holding xattrs */
458         if (obj->oo_xattr != ZFS_NO_OBJECT) {
459                 oid = obj->oo_xattr;
460
461                 dmu_tx_hold_free(tx, oid, 0, DMU_OBJECT_END);
462
463                 rc = -udmu_zap_cursor_init(&zc, uos, oid, 0);
464                 if (rc)
465                         goto out;
466
467                 while ((rc = -zap_cursor_retrieve(zc, za)) == 0) {
468                         BUG_ON(za->za_integer_length != sizeof(uint64_t));
469                         BUG_ON(za->za_num_integers != 1);
470
471                         rc = -zap_lookup(uos->os, obj->oo_xattr, za->za_name,
472                                          sizeof(uint64_t), 1, &xid);
473                         if (rc) {
474                                 CERROR("%s: xattr lookup failed: rc = %d\n",
475                                        osd->od_svname, rc);
476                                 goto out_err;
477                         }
478                         dmu_tx_hold_free(tx, xid, 0, DMU_OBJECT_END);
479
480                         zap_cursor_advance(zc);
481                 }
482                 if (rc == -ENOENT)
483                         rc = 0;
484 out_err:
485                 udmu_zap_cursor_fini(zc);
486         }
487 out:
488         if (rc && tx->tx_err == 0)
489                 tx->tx_err = -rc;
490 }
491
492 static int osd_declare_object_destroy(const struct lu_env *env,
493                                       struct dt_object *dt,
494                                       struct thandle *th)
495 {
496         char                    *buf = osd_oti_get(env)->oti_str;
497         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
498         struct osd_object       *obj = osd_dt_obj(dt);
499         struct osd_device       *osd = osd_obj2dev(obj);
500         struct osd_thandle      *oh;
501         uint64_t                 zapid;
502         int                      rc;
503         ENTRY;
504
505         LASSERT(th != NULL);
506         LASSERT(dt_object_exists(dt));
507
508         oh = container_of0(th, struct osd_thandle, ot_super);
509         LASSERT(oh->ot_tx != NULL);
510
511         /* declare that we'll destroy the object */
512         __osd_declare_object_destroy(env, obj, oh);
513
514         /* declare that we'll remove object from fid-dnode mapping */
515         zapid = osd_get_name_n_idx(env, osd, fid, buf);
516         dmu_tx_hold_bonus(oh->ot_tx, zapid);
517         dmu_tx_hold_zap(oh->ot_tx, zapid, 0, buf);
518
519         /* declare that we'll remove object from inode accounting ZAPs */
520         dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
521         dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, 0, buf);
522         dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
523         dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, 0, buf);
524
525         /* one less inode */
526         rc = osd_declare_quota(env, osd, obj->oo_attr.la_uid,
527                                obj->oo_attr.la_gid, -1, oh, false, NULL, false);
528         if (rc)
529                 RETURN(rc);
530
531         /* data to be truncated */
532         rc = osd_declare_quota(env, osd, obj->oo_attr.la_uid,
533                                obj->oo_attr.la_gid, 0, oh, true, NULL, false);
534         RETURN(rc);
535 }
536
537 int __osd_object_free(udmu_objset_t *uos, uint64_t oid, dmu_tx_t *tx)
538 {
539         LASSERT(uos->objects != 0);
540         spin_lock(&uos->lock);
541         uos->objects--;
542         spin_unlock(&uos->lock);
543
544         return -dmu_object_free(uos->os, oid, tx);
545 }
546
547 /*
548  * Delete a DMU object
549  *
550  * The transaction passed to this routine must have
551  * dmu_tx_hold_free(tx, oid, 0, DMU_OBJECT_END) called
552  * and then assigned to a transaction group.
553  *
554  * This will release db and set it to NULL to prevent further dbuf releases.
555  */
556 static int __osd_object_destroy(const struct lu_env *env,
557                                 struct osd_object *obj,
558                                 dmu_tx_t *tx, void *tag)
559 {
560         struct osd_device       *osd = osd_obj2dev(obj);
561         udmu_objset_t           *uos = &osd->od_objset;
562         uint64_t                 xid;
563         zap_attribute_t         *za = &osd_oti_get(env)->oti_za;
564         zap_cursor_t            *zc;
565         int                      rc;
566
567         /* Assert that the transaction has been assigned to a
568            transaction group. */
569         LASSERT(tx->tx_txg != 0);
570
571         /* zap holding xattrs */
572         if (obj->oo_xattr != ZFS_NO_OBJECT) {
573                 rc = -udmu_zap_cursor_init(&zc, uos, obj->oo_xattr, 0);
574                 if (rc)
575                         return rc;
576                 while ((rc = -zap_cursor_retrieve(zc, za)) == 0) {
577                         BUG_ON(za->za_integer_length != sizeof(uint64_t));
578                         BUG_ON(za->za_num_integers != 1);
579
580                         rc = -zap_lookup(uos->os, obj->oo_xattr, za->za_name,
581                                          sizeof(uint64_t), 1, &xid);
582                         if (rc) {
583                                 CERROR("%s: lookup xattr %s failed: rc = %d\n",
584                                        osd->od_svname, za->za_name, rc);
585                                 continue;
586                         }
587                         rc = __osd_object_free(uos, xid, tx);
588                         if (rc)
589                                 CERROR("%s: fetch xattr %s failed: rc = %d\n",
590                                        osd->od_svname, za->za_name, rc);
591                         zap_cursor_advance(zc);
592                 }
593                 udmu_zap_cursor_fini(zc);
594
595                 rc = __osd_object_free(uos, obj->oo_xattr, tx);
596                 if (rc)
597                         CERROR("%s: freeing xattr failed: rc = %d\n",
598                                osd->od_svname, rc);
599         }
600
601         return __osd_object_free(uos, obj->oo_db->db_object, tx);
602 }
603
604 static int osd_object_destroy(const struct lu_env *env,
605                               struct dt_object *dt,
606                               struct thandle *th)
607 {
608         char                    *buf = osd_oti_get(env)->oti_str;
609         struct osd_object       *obj = osd_dt_obj(dt);
610         struct osd_device       *osd = osd_obj2dev(obj);
611         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
612         struct osd_thandle      *oh;
613         int                      rc;
614         uint64_t                 zapid;
615         ENTRY;
616
617         LASSERT(obj->oo_db != NULL);
618         LASSERT(dt_object_exists(dt));
619         LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));
620
621         oh = container_of0(th, struct osd_thandle, ot_super);
622         LASSERT(oh != NULL);
623         LASSERT(oh->ot_tx != NULL);
624
625         zapid = osd_get_name_n_idx(env, osd, fid, buf);
626
627         /* remove obj ref from index dir (it depends) */
628         rc = -zap_remove(osd->od_objset.os, zapid, buf, oh->ot_tx);
629         if (rc) {
630                 CERROR("%s: zap_remove() failed: rc = %d\n",
631                        osd->od_svname, rc);
632                 GOTO(out, rc);
633         }
634
635         /* Remove object from inode accounting. It is not fatal for the destroy
636          * operation if something goes wrong while updating accounting, but we
637          * still log an error message to notify the administrator */
638         rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
639                         obj->oo_attr.la_uid, -1, oh->ot_tx);
640         if (rc)
641                 CERROR("%s: failed to remove "DFID" from accounting ZAP for usr"
642                         " %d: rc = %d\n", osd->od_svname, PFID(fid),
643                         obj->oo_attr.la_uid, rc);
644         rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
645                                 obj->oo_attr.la_gid, -1, oh->ot_tx);
646         if (rc)
647                 CERROR("%s: failed to remove "DFID" from accounting ZAP for grp"
648                         " %d: rc = %d\n", osd->od_svname, PFID(fid),
649                         obj->oo_attr.la_gid, rc);
650
651         /* kill object */
652         rc = __osd_object_destroy(env, obj, oh->ot_tx, osd_obj_tag);
653         if (rc) {
654                 CERROR("%s: __osd_object_destroy() failed: rc = %d\n",
655                        osd->od_svname, rc);
656                 GOTO(out, rc);
657         }
658
659 out:
660         /* not needed in the cache anymore */
661         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
662
663         RETURN (0);
664 }
665
666 static void osd_object_delete(const struct lu_env *env, struct lu_object *l)
667 {
668         struct osd_object *obj = osd_obj(l);
669
670         if (obj->oo_db != NULL) {
671                 osd_object_sa_fini(obj);
672                 if (obj->oo_sa_xattr) {
673                         nvlist_free(obj->oo_sa_xattr);
674                         obj->oo_sa_xattr = NULL;
675                 }
676                 sa_buf_rele(obj->oo_db, osd_obj_tag);
677                 cfs_list_del(&obj->oo_sa_linkage);
678                 obj->oo_db = NULL;
679         }
680 }
681
682 /*
683  * Concurrency: ->loo_object_release() is called under site spin-lock.
684  */
685 static void osd_object_release(const struct lu_env *env,
686                                struct lu_object *l)
687 {
688 }
689
690 /*
691  * Concurrency: shouldn't matter.
692  */
693 static int osd_object_print(const struct lu_env *env, void *cookie,
694                             lu_printer_t p, const struct lu_object *l)
695 {
696         struct osd_object *o = osd_obj(l);
697
698         return (*p)(env, cookie, LUSTRE_OSD_ZFS_NAME"-object@%p", o);
699 }
700
701 static void osd_object_read_lock(const struct lu_env *env,
702                                  struct dt_object *dt, unsigned role)
703 {
704         struct osd_object *obj = osd_dt_obj(dt);
705
706         LASSERT(osd_invariant(obj));
707
708         down_read(&obj->oo_sem);
709 }
710
711 static void osd_object_write_lock(const struct lu_env *env,
712                                   struct dt_object *dt, unsigned role)
713 {
714         struct osd_object *obj = osd_dt_obj(dt);
715
716         LASSERT(osd_invariant(obj));
717
718         down_write(&obj->oo_sem);
719 }
720
721 static void osd_object_read_unlock(const struct lu_env *env,
722                                    struct dt_object *dt)
723 {
724         struct osd_object *obj = osd_dt_obj(dt);
725
726         LASSERT(osd_invariant(obj));
727         up_read(&obj->oo_sem);
728 }
729
730 static void osd_object_write_unlock(const struct lu_env *env,
731                                     struct dt_object *dt)
732 {
733         struct osd_object *obj = osd_dt_obj(dt);
734
735         LASSERT(osd_invariant(obj));
736         up_write(&obj->oo_sem);
737 }
738
739 static int osd_object_write_locked(const struct lu_env *env,
740                                    struct dt_object *dt)
741 {
742         struct osd_object *obj = osd_dt_obj(dt);
743         int rc = 1;
744
745         LASSERT(osd_invariant(obj));
746
747         if (down_write_trylock(&obj->oo_sem)) {
748                 rc = 0;
749                 up_write(&obj->oo_sem);
750         }
751         return rc;
752 }
753
754 static int osd_attr_get(const struct lu_env *env,
755                         struct dt_object *dt,
756                         struct lu_attr *attr,
757                         struct lustre_capa *capa)
758 {
759         struct osd_object       *obj = osd_dt_obj(dt);
760         uint64_t                 blocks;
761         uint32_t                 blksize;
762
763         LASSERT(dt_object_exists(dt));
764         LASSERT(osd_invariant(obj));
765         LASSERT(obj->oo_db);
766
767         read_lock(&obj->oo_attr_lock);
768         *attr = obj->oo_attr;
769         read_unlock(&obj->oo_attr_lock);
770
771         /* with ZFS_DEBUG zrl_add_debug() called by DB_DNODE_ENTER()
772          * from within sa_object_size() can block on a mutex, so
773          * we can't call sa_object_size() holding rwlock */
774         sa_object_size(obj->oo_sa_hdl, &blksize, &blocks);
775         /* we do not control size of indices, so always calculate
776          * it from number of blocks reported by DMU */
777         if (S_ISDIR(attr->la_mode))
778                 attr->la_size = 512 * blocks;
779         /* Block size may be not set; suggest maximal I/O transfers. */
780         if (blksize == 0)
781                 blksize = 1ULL << SPA_MAXBLOCKSHIFT;
782
783         attr->la_blksize = blksize;
784         attr->la_blocks = blocks;
785         attr->la_valid |= LA_BLOCKS | LA_BLKSIZE;
786
787         return 0;
788 }
789
790 /* Simple wrapper on top of qsd API which implement quota transfer for osd
791  * setattr needs. As a reminder, only the root user can change ownership of
792  * a file, that's why EDQUOT & EINPROGRESS errors are discarded */
793 static inline int qsd_transfer(const struct lu_env *env,
794                                struct qsd_instance *qsd,
795                                struct lquota_trans *trans, int qtype,
796                                __u64 orig_id, __u64 new_id, __u64 bspace,
797                                struct lquota_id_info *qi)
798 {
799         int     rc;
800
801         if (unlikely(qsd == NULL))
802                 return 0;
803
804         LASSERT(qtype >= 0 && qtype < MAXQUOTAS);
805         qi->lqi_type = qtype;
806
807         /* inode accounting */
808         qi->lqi_is_blk = false;
809
810         /* one more inode for the new owner ... */
811         qi->lqi_id.qid_uid = new_id;
812         qi->lqi_space      = 1;
813         rc = qsd_op_begin(env, qsd, trans, qi, NULL);
814         if (rc == -EDQUOT || rc == -EINPROGRESS)
815                 rc = 0;
816         if (rc)
817                 return rc;
818
819         /* and one less inode for the current id */
820         qi->lqi_id.qid_uid = orig_id;;
821         qi->lqi_space      = -1;
822         /* can't get EDQUOT when reducing usage */
823         rc = qsd_op_begin(env, qsd, trans, qi, NULL);
824         if (rc == -EINPROGRESS)
825                 rc = 0;
826         if (rc)
827                 return rc;
828
829         /* block accounting */
830         qi->lqi_is_blk = true;
831
832         /* more blocks for the new owner ... */
833         qi->lqi_id.qid_uid = new_id;
834         qi->lqi_space      = bspace;
835         rc = qsd_op_begin(env, qsd, trans, qi, NULL);
836         if (rc == -EDQUOT || rc == -EINPROGRESS)
837                 rc = 0;
838         if (rc)
839                 return rc;
840
841         /* and finally less blocks for the current owner */
842         qi->lqi_id.qid_uid = orig_id;
843         qi->lqi_space      = -bspace;
844         rc = qsd_op_begin(env, qsd, trans, qi, NULL);
845         /* can't get EDQUOT when reducing usage */
846         if (rc == -EINPROGRESS)
847                 rc = 0;
848         return rc;
849 }
850
851 static int osd_declare_attr_set(const struct lu_env *env,
852                                 struct dt_object *dt,
853                                 const struct lu_attr *attr,
854                                 struct thandle *handle)
855 {
856         struct osd_thread_info  *info = osd_oti_get(env);
857         char                    *buf = osd_oti_get(env)->oti_str;
858         struct osd_object       *obj = osd_dt_obj(dt);
859         struct osd_device       *osd = osd_obj2dev(obj);
860         struct osd_thandle      *oh;
861         uint64_t                 bspace;
862         uint32_t                 blksize;
863         int                      rc;
864         ENTRY;
865
866         if (!dt_object_exists(dt)) {
867                 /* XXX: sanity check that object creation is declared */
868                 RETURN(0);
869         }
870
871         LASSERT(handle != NULL);
872         LASSERT(osd_invariant(obj));
873
874         oh = container_of0(handle, struct osd_thandle, ot_super);
875
876         LASSERT(obj->oo_sa_hdl != NULL);
877         LASSERT(oh->ot_tx != NULL);
878         dmu_tx_hold_sa(oh->ot_tx, obj->oo_sa_hdl, 0);
879         if (oh->ot_tx->tx_err != 0)
880                 RETURN(-oh->ot_tx->tx_err);
881
882         sa_object_size(obj->oo_sa_hdl, &blksize, &bspace);
883         bspace = toqb(bspace * blksize);
884
885         if (attr && attr->la_valid & LA_UID) {
886                 /* account for user inode tracking ZAP update */
887                 dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
888                 dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, TRUE, buf);
889
890                 /* quota enforcement for user */
891                 if (attr->la_uid != obj->oo_attr.la_uid) {
892                         rc = qsd_transfer(env, osd->od_quota_slave,
893                                           &oh->ot_quota_trans, USRQUOTA,
894                                           obj->oo_attr.la_uid, attr->la_uid,
895                                           bspace, &info->oti_qi);
896                         if (rc)
897                                 RETURN(rc);
898                 }
899         }
900         if (attr && attr->la_valid & LA_GID) {
901                 /* account for user inode tracking ZAP update */
902                 dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
903                 dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, TRUE, buf);
904
905                 /* quota enforcement for group */
906                 if (attr->la_gid != obj->oo_attr.la_gid) {
907                         rc = qsd_transfer(env, osd->od_quota_slave,
908                                           &oh->ot_quota_trans, GRPQUOTA,
909                                           obj->oo_attr.la_gid, attr->la_gid,
910                                           bspace, &info->oti_qi);
911                         if (rc)
912                                 RETURN(rc);
913                 }
914         }
915
916         RETURN(0);
917 }
918
919 /*
920  * Set the attributes of an object
921  *
922  * The transaction passed to this routine must have
923  * dmu_tx_hold_bonus(tx, oid) called and then assigned
924  * to a transaction group.
925  */
926 static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
927                         const struct lu_attr *la, struct thandle *handle,
928                         struct lustre_capa *capa)
929 {
930         struct osd_object       *obj = osd_dt_obj(dt);
931         struct osd_device       *osd = osd_obj2dev(obj);
932         udmu_objset_t           *uos = &osd->od_objset;
933         struct osd_thandle      *oh;
934         struct osa_attr         *osa = &osd_oti_get(env)->oti_osa;
935         sa_bulk_attr_t          *bulk;
936         int                      cnt;
937         int                      rc = 0;
938
939         ENTRY;
940         LASSERT(handle != NULL);
941         LASSERT(dt_object_exists(dt));
942         LASSERT(osd_invariant(obj));
943         LASSERT(obj->oo_sa_hdl);
944
945         oh = container_of0(handle, struct osd_thandle, ot_super);
946         /* Assert that the transaction has been assigned to a
947            transaction group. */
948         LASSERT(oh->ot_tx->tx_txg != 0);
949
950         if (la->la_valid == 0)
951                 RETURN(0);
952
953         OBD_ALLOC(bulk, sizeof(sa_bulk_attr_t) * 10);
954         if (bulk == NULL)
955                 RETURN(-ENOMEM);
956
957         /* do both accounting updates outside oo_attr_lock below */
958         if ((la->la_valid & LA_UID) && (la->la_uid != obj->oo_attr.la_uid)) {
959                 /* Update user accounting. Failure isn't fatal, but we still
960                  * log an error message */
961                 rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
962                                         la->la_uid, 1, oh->ot_tx);
963                 if (rc)
964                         CERROR("%s: failed to update accounting ZAP for user "
965                                 "%d (%d)\n", osd->od_svname, la->la_uid, rc);
966                 rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
967                                         obj->oo_attr.la_uid, -1, oh->ot_tx);
968                 if (rc)
969                         CERROR("%s: failed to update accounting ZAP for user "
970                                 "%d (%d)\n", osd->od_svname,
971                                 obj->oo_attr.la_uid, rc);
972         }
973         if ((la->la_valid & LA_GID) && (la->la_gid != obj->oo_attr.la_gid)) {
974                 /* Update group accounting. Failure isn't fatal, but we still
975                  * log an error message */
976                 rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
977                                         la->la_gid, 1, oh->ot_tx);
978                 if (rc)
979                         CERROR("%s: failed to update accounting ZAP for user "
980                                 "%d (%d)\n", osd->od_svname, la->la_gid, rc);
981                 rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
982                                         obj->oo_attr.la_gid, -1, oh->ot_tx);
983                 if (rc)
984                         CERROR("%s: failed to update accounting ZAP for user "
985                                 "%d (%d)\n", osd->od_svname,
986                                 obj->oo_attr.la_gid, rc);
987         }
988
989         write_lock(&obj->oo_attr_lock);
990         cnt = 0;
991         if (la->la_valid & LA_ATIME) {
992                 osa->atime[0] = obj->oo_attr.la_atime = la->la_atime;
993                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(uos), NULL,
994                                  osa->atime, 16);
995         }
996         if (la->la_valid & LA_MTIME) {
997                 osa->mtime[0] = obj->oo_attr.la_mtime = la->la_mtime;
998                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(uos), NULL,
999                                  osa->mtime, 16);
1000         }
1001         if (la->la_valid & LA_CTIME) {
1002                 osa->ctime[0] = obj->oo_attr.la_ctime = la->la_ctime;
1003                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(uos), NULL,
1004                                  osa->ctime, 16);
1005         }
1006         if (la->la_valid & LA_MODE) {
1007                 /* mode is stored along with type, so read it first */
1008                 obj->oo_attr.la_mode = (obj->oo_attr.la_mode & S_IFMT) |
1009                         (la->la_mode & ~S_IFMT);
1010                 osa->mode = obj->oo_attr.la_mode;
1011                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(uos), NULL,
1012                                  &osa->mode, 8);
1013         }
1014         if (la->la_valid & LA_SIZE) {
1015                 osa->size = obj->oo_attr.la_size = la->la_size;
1016                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(uos), NULL,
1017                                  &osa->size, 8);
1018         }
1019         if (la->la_valid & LA_NLINK) {
1020                 osa->nlink = obj->oo_attr.la_nlink = la->la_nlink;
1021                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(uos), NULL,
1022                                  &osa->nlink, 8);
1023         }
1024         if (la->la_valid & LA_RDEV) {
1025                 osa->rdev = obj->oo_attr.la_rdev = la->la_rdev;
1026                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(uos), NULL,
1027                                  &osa->rdev, 8);
1028         }
1029         if (la->la_valid & LA_FLAGS) {
1030                 osa->flags = attrs_fs2zfs(la->la_flags);
1031                 /* many flags are not supported by zfs, so ensure a good cached
1032                  * copy */
1033                 obj->oo_attr.la_flags = attrs_zfs2fs(osa->flags);
1034                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(uos), NULL,
1035                                  &osa->flags, 8);
1036         }
1037         if (la->la_valid & LA_UID) {
1038                 osa->uid = obj->oo_attr.la_uid = la->la_uid;
1039                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(uos), NULL,
1040                                  &osa->uid, 8);
1041         }
1042         if (la->la_valid & LA_GID) {
1043                 osa->gid = obj->oo_attr.la_gid = la->la_gid;
1044                 SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(uos), NULL,
1045                                  &osa->gid, 8);
1046         }
1047         obj->oo_attr.la_valid |= la->la_valid;
1048         write_unlock(&obj->oo_attr_lock);
1049
1050         rc = osd_object_sa_bulk_update(obj, bulk, cnt, oh);
1051
1052         OBD_FREE(bulk, sizeof(sa_bulk_attr_t) * 10);
1053         RETURN(rc);
1054 }
1055
1056 /*
1057  * Object creation.
1058  *
1059  * XXX temporary solution.
1060  */
1061
1062 static void osd_ah_init(const struct lu_env *env, struct dt_allocation_hint *ah,
1063                         struct dt_object *parent, struct dt_object *child,
1064                         umode_t child_mode)
1065 {
1066         LASSERT(ah);
1067
1068         memset(ah, 0, sizeof(*ah));
1069         ah->dah_parent = parent;
1070         ah->dah_mode = child_mode;
1071 }
1072
1073 static int osd_declare_object_create(const struct lu_env *env,
1074                                      struct dt_object *dt,
1075                                      struct lu_attr *attr,
1076                                      struct dt_allocation_hint *hint,
1077                                      struct dt_object_format *dof,
1078                                      struct thandle *handle)
1079 {
1080         char                    *buf = osd_oti_get(env)->oti_str;
1081         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1082         struct osd_object       *obj = osd_dt_obj(dt);
1083         struct osd_device       *osd = osd_obj2dev(obj);
1084         struct osd_thandle      *oh;
1085         uint64_t                 zapid;
1086         int                      rc;
1087         ENTRY;
1088
1089         LASSERT(dof);
1090
1091         switch (dof->dof_type) {
1092                 case DFT_REGULAR:
1093                 case DFT_SYM:
1094                 case DFT_NODE:
1095                         if (obj->oo_dt.do_body_ops == NULL)
1096                                 obj->oo_dt.do_body_ops = &osd_body_ops;
1097                         break;
1098                 default:
1099                         break;
1100         }
1101
1102         LASSERT(handle != NULL);
1103         oh = container_of0(handle, struct osd_thandle, ot_super);
1104         LASSERT(oh->ot_tx != NULL);
1105
1106         switch (dof->dof_type) {
1107                 case DFT_DIR:
1108                 case DFT_INDEX:
1109                         /* for zap create */
1110                         dmu_tx_hold_zap(oh->ot_tx, DMU_NEW_OBJECT, 1, NULL);
1111                         break;
1112                 case DFT_REGULAR:
1113                 case DFT_SYM:
1114                 case DFT_NODE:
1115                         /* first, we'll create new object */
1116                         dmu_tx_hold_bonus(oh->ot_tx, DMU_NEW_OBJECT);
1117                         break;
1118
1119                 default:
1120                         LBUG();
1121                         break;
1122         }
1123
1124         /* and we'll add it to some mapping */
1125         zapid = osd_get_name_n_idx(env, osd, fid, buf);
1126         dmu_tx_hold_bonus(oh->ot_tx, zapid);
1127         dmu_tx_hold_zap(oh->ot_tx, zapid, TRUE, buf);
1128
1129         /* we will also update inode accounting ZAPs */
1130         dmu_tx_hold_bonus(oh->ot_tx, osd->od_iusr_oid);
1131         dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, TRUE, buf);
1132         dmu_tx_hold_bonus(oh->ot_tx, osd->od_igrp_oid);
1133         dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, TRUE, buf);
1134
1135         dmu_tx_hold_sa_create(oh->ot_tx, ZFS_SA_BASE_ATTR_SIZE);
1136
1137         __osd_xattr_declare_set(env, obj, sizeof(struct lustre_mdt_attrs),
1138                                 XATTR_NAME_LMA, oh);
1139
1140         rc = osd_declare_quota(env, osd, attr->la_uid, attr->la_gid, 1, oh,
1141                                false, NULL, false);
1142         RETURN(rc);
1143 }
1144
1145 int __osd_attr_init(const struct lu_env *env, udmu_objset_t *uos, uint64_t oid,
1146                     dmu_tx_t *tx, struct lu_attr *la, uint64_t parent)
1147 {
1148         sa_bulk_attr_t  *bulk;
1149         sa_handle_t     *sa_hdl;
1150         struct osa_attr *osa = &osd_oti_get(env)->oti_osa;
1151         uint64_t         gen;
1152         uint64_t         crtime[2];
1153         timestruc_t      now;
1154         int              cnt;
1155         int              rc;
1156
1157         gethrestime(&now);
1158         gen = dmu_tx_get_txg(tx);
1159
1160         ZFS_TIME_ENCODE(&now, crtime);
1161
1162         osa->atime[0] = la->la_atime;
1163         osa->ctime[0] = la->la_ctime;
1164         osa->mtime[0] = la->la_mtime;
1165         osa->mode = la->la_mode;
1166         osa->uid = la->la_uid;
1167         osa->gid = la->la_gid;
1168         osa->rdev = la->la_rdev;
1169         osa->nlink = la->la_nlink;
1170         osa->flags = attrs_fs2zfs(la->la_flags);
1171         osa->size  = la->la_size;
1172
1173         /* Now add in all of the "SA" attributes */
1174         rc = -sa_handle_get(uos->os, oid, NULL, SA_HDL_PRIVATE, &sa_hdl);
1175         if (rc)
1176                 return rc;
1177
1178         OBD_ALLOC(bulk, sizeof(sa_bulk_attr_t) * 13);
1179         if (bulk == NULL) {
1180                 rc = -ENOMEM;
1181                 goto out;
1182         }
1183         /*
1184          * we need to create all SA below upon object create.
1185          *
1186          * XXX The attribute order matters since the accounting callback relies
1187          * on static offsets (i.e. SA_*_OFFSET, see zfs_space_delta_cb()) to
1188          * look up the UID/GID attributes. Moreover, the callback does not seem
1189          * to support the spill block.
1190          * We define attributes in the same order as SA_*_OFFSET in order to
1191          * work around the problem. See ORI-610.
1192          */
1193         cnt = 0;
1194         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(uos), NULL, &osa->mode, 8);
1195         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(uos), NULL, &osa->size, 8);
1196         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GEN(uos), NULL, &gen, 8);
1197         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(uos), NULL, &osa->uid, 8);
1198         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(uos), NULL, &osa->gid, 8);
1199         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_PARENT(uos), NULL, &parent, 8);
1200         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(uos), NULL, &osa->flags, 8);
1201         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(uos), NULL, osa->atime, 16);
1202         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(uos), NULL, osa->mtime, 16);
1203         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(uos), NULL, osa->ctime, 16);
1204         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CRTIME(uos), NULL, crtime, 16);
1205         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(uos), NULL, &osa->nlink, 8);
1206         SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_RDEV(uos), NULL, &osa->rdev, 8);
1207
1208         rc = -sa_replace_all_by_template(sa_hdl, bulk, cnt, tx);
1209
1210         OBD_FREE(bulk, sizeof(sa_bulk_attr_t) * 13);
1211 out:
1212         sa_handle_destroy(sa_hdl);
1213         return rc;
1214 }
1215
1216 /*
1217  * The transaction passed to this routine must have
1218  * dmu_tx_hold_bonus(tx, DMU_NEW_OBJECT) called and then assigned
1219  * to a transaction group.
1220  */
1221 int __osd_object_create(const struct lu_env *env, udmu_objset_t *uos,
1222                         dmu_buf_t **dbp, dmu_tx_t *tx, struct lu_attr *la,
1223                         uint64_t parent, void *tag)
1224 {
1225         uint64_t oid;
1226         int      rc;
1227
1228         LASSERT(tag);
1229         /* Assert that the transaction has been assigned to a
1230            transaction group. */
1231         LASSERT(tx->tx_txg != 0);
1232
1233         /* Create a new DMU object. */
1234         oid = dmu_object_alloc(uos->os, DMU_OT_PLAIN_FILE_CONTENTS, 0,
1235                                DMU_OT_SA, DN_MAX_BONUSLEN, tx);
1236         rc = -sa_buf_hold(uos->os, oid, tag, dbp);
1237         LASSERTF(rc == 0, "sa_buf_hold "LPU64" failed: %d\n", oid, rc);
1238
1239         LASSERT(la->la_valid & LA_MODE);
1240         la->la_size = 0;
1241         la->la_nlink = 1;
1242
1243         rc = __osd_attr_init(env, uos, oid, tx, la, parent);
1244         if (rc != 0) {
1245                 sa_buf_rele(*dbp, tag);
1246                 *dbp = NULL;
1247                 dmu_object_free(uos->os, oid, tx);
1248                 return rc;
1249         }
1250
1251         spin_lock(&uos->lock);
1252         uos->objects++;
1253         spin_unlock(&uos->lock);
1254         return 0;
1255 }
1256
1257 /*
1258  * The transaction passed to this routine must have
1259  * dmu_tx_hold_zap(tx, DMU_NEW_OBJECT, ...) called and then assigned
1260  * to a transaction group.
1261  *
1262  * Using ZAP_FLAG_HASH64 will force the ZAP to always be a FAT ZAP.
1263  * This is fine for directories today, because storing the FID in the dirent
1264  * will also require a FAT ZAP.  If there is a new type of micro ZAP created
1265  * then we might need to re-evaluate the use of this flag and instead do
1266  * a conversion from the different internal ZAP hash formats being used. */
1267 int __osd_zap_create(const struct lu_env *env, udmu_objset_t *uos,
1268                      dmu_buf_t **zap_dbp, dmu_tx_t *tx,
1269                      struct lu_attr *la, uint64_t parent,
1270                      void *tag, zap_flags_t flags)
1271 {
1272         uint64_t oid;
1273         int      rc;
1274
1275         LASSERT(tag);
1276
1277         spin_lock(&uos->lock);
1278         uos->objects++;
1279         spin_unlock(&uos->lock);
1280
1281         /* Assert that the transaction has been assigned to a
1282            transaction group. */
1283         LASSERT(tx->tx_txg != 0);
1284
1285         oid = zap_create_flags(uos->os, 0, flags | ZAP_FLAG_HASH64,
1286                                DMU_OT_DIRECTORY_CONTENTS, 12, 12,
1287                                DMU_OT_SA, DN_MAX_BONUSLEN, tx);
1288
1289         rc = -sa_buf_hold(uos->os, oid, tag, zap_dbp);
1290         if (rc)
1291                 return rc;
1292
1293         LASSERT(la->la_valid & LA_MODE);
1294         la->la_size = 2;
1295         la->la_nlink = 1;
1296
1297         return __osd_attr_init(env, uos, oid, tx, la, parent);
1298 }
1299
1300 static dmu_buf_t *osd_mkidx(const struct lu_env *env, struct osd_device *osd,
1301                             struct lu_attr *la, uint64_t parent,
1302                             struct osd_thandle *oh)
1303 {
1304         dmu_buf_t *db;
1305         int        rc;
1306
1307         /* Index file should be created as regular file in order not to confuse
1308          * ZPL which could interpret them as directory.
1309          * We set ZAP_FLAG_UINT64_KEY to let ZFS know than we are going to use
1310          * binary keys */
1311         LASSERT(S_ISREG(la->la_mode));
1312         rc = __osd_zap_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1313                               parent, osd_obj_tag, ZAP_FLAG_UINT64_KEY);
1314         if (rc)
1315                 return ERR_PTR(rc);
1316         return db;
1317 }
1318
1319 static dmu_buf_t *osd_mkdir(const struct lu_env *env, struct osd_device *osd,
1320                             struct lu_attr *la, uint64_t parent,
1321                             struct osd_thandle *oh)
1322 {
1323         dmu_buf_t *db;
1324         int        rc;
1325
1326         LASSERT(S_ISDIR(la->la_mode));
1327         rc = __osd_zap_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1328                               parent, osd_obj_tag, 0);
1329         if (rc)
1330                 return ERR_PTR(rc);
1331         return db;
1332 }
1333
1334 static dmu_buf_t* osd_mkreg(const struct lu_env *env, struct osd_device *osd,
1335                             struct lu_attr *la, uint64_t parent,
1336                             struct osd_thandle *oh)
1337 {
1338         dmu_buf_t *db;
1339         int         rc;
1340
1341         LASSERT(S_ISREG(la->la_mode));
1342         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1343                                  parent, osd_obj_tag);
1344         if (rc)
1345                 return ERR_PTR(rc);
1346
1347         /*
1348          * XXX: a hack, OST to use bigger blocksize. we need
1349          * a method in OSD API to control this from OFD/MDD
1350          */
1351         if (!lu_device_is_md(osd2lu_dev(osd))) {
1352                 rc = -dmu_object_set_blocksize(osd->od_objset.os,
1353                                                db->db_object,
1354                                 128 << 10, 0, oh->ot_tx);
1355                 if (unlikely(rc)) {
1356                         CERROR("%s: can't change blocksize: %d\n",
1357                                osd->od_svname, rc);
1358                         return ERR_PTR(rc);
1359                 }
1360         }
1361
1362         return db;
1363 }
1364
1365 static dmu_buf_t *osd_mksym(const struct lu_env *env, struct osd_device *osd,
1366                             struct lu_attr *la, uint64_t parent,
1367                             struct osd_thandle *oh)
1368 {
1369         dmu_buf_t *db;
1370         int        rc;
1371
1372         LASSERT(S_ISLNK(la->la_mode));
1373         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1374                                  parent, osd_obj_tag);
1375         if (rc)
1376                 return ERR_PTR(rc);
1377         return db;
1378 }
1379
1380 static dmu_buf_t *osd_mknod(const struct lu_env *env, struct osd_device *osd,
1381                             struct lu_attr *la, uint64_t parent,
1382                             struct osd_thandle *oh)
1383 {
1384         dmu_buf_t *db;
1385         int        rc;
1386
1387         la->la_valid = LA_MODE;
1388         if (S_ISCHR(la->la_mode) || S_ISBLK(la->la_mode))
1389                 la->la_valid |= LA_RDEV;
1390
1391         rc = __osd_object_create(env, &osd->od_objset, &db, oh->ot_tx, la,
1392                                  parent, osd_obj_tag);
1393         if (rc)
1394                 return ERR_PTR(rc);
1395         return db;
1396 }
1397
1398 typedef dmu_buf_t *(*osd_obj_type_f)(const struct lu_env *env,
1399                                      struct osd_device *osd,
1400                                      struct lu_attr *la,
1401                                      uint64_t parent,
1402                                      struct osd_thandle *oh);
1403
1404 static osd_obj_type_f osd_create_type_f(enum dt_format_type type)
1405 {
1406         osd_obj_type_f result;
1407
1408         switch (type) {
1409         case DFT_DIR:
1410                 result = osd_mkdir;
1411                 break;
1412         case DFT_INDEX:
1413                 result = osd_mkidx;
1414                 break;
1415         case DFT_REGULAR:
1416                 result = osd_mkreg;
1417                 break;
1418         case DFT_SYM:
1419                 result = osd_mksym;
1420                 break;
1421         case DFT_NODE:
1422                 result = osd_mknod;
1423                 break;
1424         default:
1425                 LBUG();
1426                 break;
1427         }
1428         return result;
1429 }
1430
1431 /*
1432  * Primitives for directory (i.e. ZAP) handling
1433  */
1434 static inline int osd_init_lma(const struct lu_env *env, struct osd_object *obj,
1435                                const struct lu_fid *fid, struct osd_thandle *oh)
1436 {
1437         struct osd_thread_info  *info = osd_oti_get(env);
1438         struct lustre_mdt_attrs *lma = &info->oti_mdt_attrs;
1439         struct lu_buf            buf;
1440         int rc;
1441
1442         lustre_lma_init(lma, fid, 0, 0);
1443         lustre_lma_swab(lma);
1444         buf.lb_buf = lma;
1445         buf.lb_len = sizeof(*lma);
1446
1447         rc = osd_xattr_set_internal(env, obj, &buf, XATTR_NAME_LMA,
1448                                     LU_XATTR_CREATE, oh, BYPASS_CAPA);
1449
1450         return rc;
1451 }
1452
1453 /*
1454  * Concurrency: @dt is write locked.
1455  */
1456 static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
1457                              struct lu_attr *attr,
1458                              struct dt_allocation_hint *hint,
1459                              struct dt_object_format *dof,
1460                              struct thandle *th)
1461 {
1462         struct zpl_direntry     *zde = &osd_oti_get(env)->oti_zde.lzd_reg;
1463         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1464         struct osd_object       *obj = osd_dt_obj(dt);
1465         struct osd_device       *osd = osd_obj2dev(obj);
1466         char                    *buf = osd_oti_get(env)->oti_str;
1467         struct osd_thandle      *oh;
1468         dmu_buf_t               *db;
1469         uint64_t                 zapid;
1470         int                      rc;
1471
1472         ENTRY;
1473
1474         /* concurrent create declarations should not see
1475          * the object inconsistent (db, attr, etc).
1476          * in regular cases acquisition should be cheap */
1477         down(&obj->oo_guard);
1478
1479         LASSERT(osd_invariant(obj));
1480         LASSERT(!dt_object_exists(dt));
1481         LASSERT(dof != NULL);
1482
1483         LASSERT(th != NULL);
1484         oh = container_of0(th, struct osd_thandle, ot_super);
1485
1486         /*
1487          * XXX missing: Quote handling.
1488          */
1489
1490         LASSERT(obj->oo_db == NULL);
1491
1492         /* to follow ZFS on-disk format we need
1493          * to initialize parent dnode properly */
1494         zapid = 0;
1495         if (hint && hint->dah_parent)
1496                 zapid = osd_dt_obj(hint->dah_parent)->oo_db->db_object;
1497
1498         db = osd_create_type_f(dof->dof_type)(env, osd, attr, zapid, oh);
1499         if (IS_ERR(db))
1500                 GOTO(out, rc = PTR_ERR(db));
1501
1502         zde->zde_pad = 0;
1503         zde->zde_dnode = db->db_object;
1504         zde->zde_type = IFTODT(attr->la_mode & S_IFMT);
1505
1506         zapid = osd_get_name_n_idx(env, osd, fid, buf);
1507
1508         rc = -zap_add(osd->od_objset.os, zapid, buf, 8, 1, zde, oh->ot_tx);
1509         if (rc)
1510                 GOTO(out, rc);
1511
1512         /* Add new object to inode accounting.
1513          * Errors are not considered as fatal */
1514         rc = -zap_increment_int(osd->od_objset.os, osd->od_iusr_oid,
1515                                 (attr->la_valid & LA_UID) ? attr->la_uid : 0, 1,
1516                                 oh->ot_tx);
1517         if (rc)
1518                 CERROR("%s: failed to add "DFID" to accounting ZAP for usr %d "
1519                         "(%d)\n", osd->od_svname, PFID(fid), attr->la_uid, rc);
1520         rc = -zap_increment_int(osd->od_objset.os, osd->od_igrp_oid,
1521                                 (attr->la_valid & LA_GID) ? attr->la_gid : 0, 1,
1522                                 oh->ot_tx);
1523         if (rc)
1524                 CERROR("%s: failed to add "DFID" to accounting ZAP for grp %d "
1525                         "(%d)\n", osd->od_svname, PFID(fid), attr->la_gid, rc);
1526
1527         /* configure new osd object */
1528         obj->oo_db = db;
1529         rc = osd_object_init0(env, obj);
1530         LASSERT(ergo(rc == 0, dt_object_exists(dt)));
1531         LASSERT(osd_invariant(obj));
1532
1533         rc = osd_init_lma(env, obj, fid, oh);
1534         if (rc) {
1535                 CERROR("%s: can not set LMA on "DFID": rc = %d\n",
1536                        osd->od_svname, PFID(fid), rc);
1537                 /* ignore errors during LMA initialization */
1538                 rc = 0;
1539         }
1540
1541 out:
1542         up(&obj->oo_guard);
1543         RETURN(rc);
1544 }
1545
1546 static int osd_declare_object_ref_add(const struct lu_env *env,
1547                                       struct dt_object *dt,
1548                                       struct thandle *th)
1549 {
1550         return osd_declare_attr_set(env, dt, NULL, th);
1551 }
1552
1553 /*
1554  * Concurrency: @dt is write locked.
1555  */
1556 static int osd_object_ref_add(const struct lu_env *env,
1557                               struct dt_object *dt,
1558                               struct thandle *handle)
1559 {
1560         struct osd_object       *obj = osd_dt_obj(dt);
1561         struct osd_thandle      *oh;
1562         struct osd_device       *osd = osd_obj2dev(obj);
1563         udmu_objset_t           *uos = &osd->od_objset;
1564         uint64_t                 nlink;
1565         int rc;
1566
1567         ENTRY;
1568
1569         LASSERT(osd_invariant(obj));
1570         LASSERT(dt_object_exists(dt));
1571         LASSERT(obj->oo_sa_hdl != NULL);
1572
1573         oh = container_of0(handle, struct osd_thandle, ot_super);
1574
1575         write_lock(&obj->oo_attr_lock);
1576         nlink = ++obj->oo_attr.la_nlink;
1577         write_unlock(&obj->oo_attr_lock);
1578
1579         rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh);
1580         return rc;
1581 }
1582
1583 static int osd_declare_object_ref_del(const struct lu_env *env,
1584                                       struct dt_object *dt,
1585                                       struct thandle *handle)
1586 {
1587         return osd_declare_attr_set(env, dt, NULL, handle);
1588 }
1589
1590 /*
1591  * Concurrency: @dt is write locked.
1592  */
1593 static int osd_object_ref_del(const struct lu_env *env,
1594                               struct dt_object *dt,
1595                               struct thandle *handle)
1596 {
1597         struct osd_object       *obj = osd_dt_obj(dt);
1598         struct osd_thandle      *oh;
1599         struct osd_device       *osd = osd_obj2dev(obj);
1600         udmu_objset_t           *uos = &osd->od_objset;
1601         uint64_t                 nlink;
1602         int                      rc;
1603
1604         ENTRY;
1605
1606         LASSERT(osd_invariant(obj));
1607         LASSERT(dt_object_exists(dt));
1608         LASSERT(obj->oo_sa_hdl != NULL);
1609
1610         oh = container_of0(handle, struct osd_thandle, ot_super);
1611         LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));
1612
1613         write_lock(&obj->oo_attr_lock);
1614         nlink = --obj->oo_attr.la_nlink;
1615         write_unlock(&obj->oo_attr_lock);
1616
1617         rc = osd_object_sa_update(obj, SA_ZPL_LINKS(uos), &nlink, 8, oh);
1618         return rc;
1619 }
1620
1621 static int capa_is_sane(const struct lu_env *env, struct osd_device *dev,
1622                         struct lustre_capa *capa, struct lustre_capa_key *keys)
1623 {
1624         struct osd_thread_info  *oti = osd_oti_get(env);
1625         struct obd_capa         *oc;
1626         int                      i, rc = 0;
1627         ENTRY;
1628
1629         oc = capa_lookup(dev->od_capa_hash, capa, 0);
1630         if (oc) {
1631                 if (capa_is_expired(oc)) {
1632                         DEBUG_CAPA(D_ERROR, capa, "expired");
1633                         rc = -ESTALE;
1634                 }
1635                 capa_put(oc);
1636                 RETURN(rc);
1637         }
1638
1639         spin_lock(&capa_lock);
1640         for (i = 0; i < 2; i++) {
1641                 if (keys[i].lk_keyid == capa->lc_keyid) {
1642                         oti->oti_capa_key = keys[i];
1643                         break;
1644                 }
1645         }
1646         spin_unlock(&capa_lock);
1647
1648         if (i == 2) {
1649                 DEBUG_CAPA(D_ERROR, capa, "no matched capa key");
1650                 RETURN(-ESTALE);
1651         }
1652
1653         rc = capa_hmac(oti->oti_capa.lc_hmac, capa, oti->oti_capa_key.lk_key);
1654         if (rc)
1655                 RETURN(rc);
1656         if (memcmp(oti->oti_capa.lc_hmac, capa->lc_hmac, sizeof(capa->lc_hmac)))
1657         {
1658                 DEBUG_CAPA(D_ERROR, capa, "HMAC mismatch");
1659                 RETURN(-EACCES);
1660         }
1661
1662         oc = capa_add(dev->od_capa_hash, capa);
1663         capa_put(oc);
1664
1665         RETURN(0);
1666 }
1667
1668 static int osd_object_auth(const struct lu_env *env, struct dt_object *dt,
1669                            struct lustre_capa *capa, __u64 opc)
1670 {
1671         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1672         struct osd_device       *dev = osd_dev(dt->do_lu.lo_dev);
1673         int                      rc;
1674
1675         if (!dev->od_fl_capa)
1676                 return 0;
1677
1678         if (capa == BYPASS_CAPA)
1679                 return 0;
1680
1681         if (!capa) {
1682                 CERROR("no capability is provided for fid "DFID"\n", PFID(fid));
1683                 return -EACCES;
1684         }
1685
1686         if (!lu_fid_eq(fid, &capa->lc_fid)) {
1687                 DEBUG_CAPA(D_ERROR, capa, "fid "DFID" mismatch with",PFID(fid));
1688                 return -EACCES;
1689         }
1690
1691         if (!capa_opc_supported(capa, opc)) {
1692                 DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
1693                 return -EACCES;
1694         }
1695
1696         if ((rc = capa_is_sane(env, dev, capa, dev->od_capa_keys))) {
1697                 DEBUG_CAPA(D_ERROR, capa, "insane (rc %d)", rc);
1698                 return -EACCES;
1699         }
1700
1701         return 0;
1702 }
1703
1704 static struct obd_capa *osd_capa_get(const struct lu_env *env,
1705                                      struct dt_object *dt,
1706                                      struct lustre_capa *old,
1707                                      __u64 opc)
1708 {
1709         struct osd_thread_info  *info = osd_oti_get(env);
1710         const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
1711         struct osd_object       *obj = osd_dt_obj(dt);
1712         struct osd_device       *dev = osd_obj2dev(obj);
1713         struct lustre_capa_key  *key = &info->oti_capa_key;
1714         struct lustre_capa      *capa = &info->oti_capa;
1715         struct obd_capa         *oc;
1716         int                      rc;
1717         ENTRY;
1718
1719         if (!dev->od_fl_capa)
1720                 RETURN(ERR_PTR(-ENOENT));
1721
1722         LASSERT(dt_object_exists(dt));
1723         LASSERT(osd_invariant(obj));
1724
1725         /* renewal sanity check */
1726         if (old && osd_object_auth(env, dt, old, opc))
1727                 RETURN(ERR_PTR(-EACCES));
1728
1729         capa->lc_fid = *fid;
1730         capa->lc_opc = opc;
1731         capa->lc_uid = 0;
1732         capa->lc_flags = dev->od_capa_alg << 24;
1733         capa->lc_timeout = dev->od_capa_timeout;
1734         capa->lc_expiry = 0;
1735
1736         oc = capa_lookup(dev->od_capa_hash, capa, 1);
1737         if (oc) {
1738                 LASSERT(!capa_is_expired(oc));
1739                 RETURN(oc);
1740         }
1741
1742         spin_lock(&capa_lock);
1743         *key = dev->od_capa_keys[1];
1744         spin_unlock(&capa_lock);
1745
1746         capa->lc_keyid = key->lk_keyid;
1747         capa->lc_expiry = cfs_time_current_sec() + dev->od_capa_timeout;
1748
1749         rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
1750         if (rc) {
1751                 DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
1752                 LBUG();
1753                 RETURN(ERR_PTR(rc));
1754         }
1755
1756         oc = capa_add(dev->od_capa_hash, capa);
1757         RETURN(oc);
1758 }
1759
1760 static int osd_object_sync(const struct lu_env *env, struct dt_object *dt,
1761                            __u64 start, __u64 end)
1762 {
1763         struct osd_device *osd = osd_obj2dev(osd_dt_obj(dt));
1764         ENTRY;
1765
1766         /* XXX: no other option than syncing the whole filesystem until we
1767          * support ZIL.  If the object tracked the txg that it was last
1768          * modified in, it could pass that txg here instead of "0".  Maybe
1769          * the changes are already committed, so no wait is needed at all? */
1770         txg_wait_synced(dmu_objset_pool(osd->od_objset.os), 0ULL);
1771
1772         RETURN(0);
1773 }
1774
1775 static struct dt_object_operations osd_obj_ops = {
1776         .do_read_lock           = osd_object_read_lock,
1777         .do_write_lock          = osd_object_write_lock,
1778         .do_read_unlock         = osd_object_read_unlock,
1779         .do_write_unlock        = osd_object_write_unlock,
1780         .do_write_locked        = osd_object_write_locked,
1781         .do_attr_get            = osd_attr_get,
1782         .do_declare_attr_set    = osd_declare_attr_set,
1783         .do_attr_set            = osd_attr_set,
1784         .do_ah_init             = osd_ah_init,
1785         .do_declare_create      = osd_declare_object_create,
1786         .do_create              = osd_object_create,
1787         .do_declare_destroy     = osd_declare_object_destroy,
1788         .do_destroy             = osd_object_destroy,
1789         .do_index_try           = osd_index_try,
1790         .do_declare_ref_add     = osd_declare_object_ref_add,
1791         .do_ref_add             = osd_object_ref_add,
1792         .do_declare_ref_del     = osd_declare_object_ref_del,
1793         .do_ref_del             = osd_object_ref_del,
1794         .do_xattr_get           = osd_xattr_get,
1795         .do_declare_xattr_set   = osd_declare_xattr_set,
1796         .do_xattr_set           = osd_xattr_set,
1797         .do_declare_xattr_del   = osd_declare_xattr_del,
1798         .do_xattr_del           = osd_xattr_del,
1799         .do_xattr_list          = osd_xattr_list,
1800         .do_capa_get            = osd_capa_get,
1801         .do_object_sync         = osd_object_sync,
1802 };
1803
1804 static struct lu_object_operations osd_lu_obj_ops = {
1805         .loo_object_init        = osd_object_init,
1806         .loo_object_delete      = osd_object_delete,
1807         .loo_object_release     = osd_object_release,
1808         .loo_object_free        = osd_object_free,
1809         .loo_object_print       = osd_object_print,
1810         .loo_object_invariant   = osd_object_invariant,
1811 };
1812
1813 static int osd_otable_it_attr_get(const struct lu_env *env,
1814                                 struct dt_object *dt,
1815                                 struct lu_attr *attr,
1816                                 struct lustre_capa *capa)
1817 {
1818         attr->la_valid = 0;
1819         return 0;
1820 }
1821
1822 static struct dt_object_operations osd_obj_otable_it_ops = {
1823         .do_attr_get    = osd_otable_it_attr_get,
1824         .do_index_try   = osd_index_try,
1825 };
1826