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