Whamcloud - gitweb
b=19427 Correct lmm_object_id
[fs/lustre-release.git] / lustre / osd-ldiskfs / osd_handler.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
30  * Use is subject to license terms.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  *
36  * lustre/osd/osd_handler.c
37  *
38  * Top-level entry points into osd module
39  *
40  * Author: Nikita Danilov <nikita@clusterfs.com>
41  *         Pravin Shelar <pravin.shelar@sun.com> : Added fid in dirent
42  */
43
44 #ifndef EXPORT_SYMTAB
45 # define EXPORT_SYMTAB
46 #endif
47 #define DEBUG_SUBSYSTEM S_MDS
48
49 #include <linux/module.h>
50
51 /* LUSTRE_VERSION_CODE */
52 #include <lustre_ver.h>
53 /* prerequisite for linux/xattr.h */
54 #include <linux/types.h>
55 /* prerequisite for linux/xattr.h */
56 #include <linux/fs.h>
57 /* XATTR_{REPLACE,CREATE} */
58 #include <linux/xattr.h>
59 /* simple_mkdir() */
60 #include <lvfs.h>
61
62 /*
63  * struct OBD_{ALLOC,FREE}*()
64  * OBD_FAIL_CHECK
65  */
66 #include <obd_support.h>
67 /* struct ptlrpc_thread */
68 #include <lustre_net.h>
69
70 /* fid_is_local() */
71 #include <lustre_fid.h>
72
73 #include "osd_internal.h"
74 #include "osd_igif.h"
75
76 /* llo_* api support */
77 #include <md_object.h>
78
79 static const char dot[] = ".";
80 static const char dotdot[] = "..";
81 static const char remote_obj_dir[] = "REM_OBJ_DIR";
82
83 struct osd_directory {
84         struct iam_container od_container;
85         struct iam_descr     od_descr;
86 };
87
88 struct osd_object {
89         struct dt_object       oo_dt;
90         /**
91          * Inode for file system object represented by this osd_object. This
92          * inode is pinned for the whole duration of lu_object life.
93          *
94          * Not modified concurrently (either setup early during object
95          * creation, or assigned by osd_object_create() under write lock).
96          */
97         struct inode          *oo_inode;
98         /**
99          * to protect index ops.
100          */
101         cfs_rw_semaphore_t     oo_ext_idx_sem;
102         cfs_rw_semaphore_t     oo_sem;
103         struct osd_directory  *oo_dir;
104         /** protects inode attributes. */
105         cfs_spinlock_t         oo_guard;
106         /**
107          * Following two members are used to indicate the presence of dot and
108          * dotdot in the given directory. This is required for interop mode
109          * (b11826).
110          */
111         int                    oo_compat_dot_created;
112         int                    oo_compat_dotdot_created;
113
114         const struct lu_env   *oo_owner;
115 #ifdef CONFIG_LOCKDEP
116         struct lockdep_map     oo_dep_map;
117 #endif
118 };
119
120 static const struct lu_object_operations      osd_lu_obj_ops;
121 static const struct lu_device_operations      osd_lu_ops;
122 static       struct lu_context_key            osd_key;
123 static const struct dt_object_operations      osd_obj_ops;
124 static const struct dt_object_operations      osd_obj_ea_ops;
125 static const struct dt_body_operations        osd_body_ops;
126 static const struct dt_index_operations       osd_index_iam_ops;
127 static const struct dt_index_operations       osd_index_ea_ops;
128
129 struct osd_thandle {
130         struct thandle          ot_super;
131         handle_t               *ot_handle;
132         struct journal_callback ot_jcb;
133         /* Link to the device, for debugging. */
134         struct lu_ref_link     *ot_dev_link;
135
136 #if OSD_THANDLE_STATS
137         /** time when this handle was allocated */
138         cfs_time_t oth_alloced;
139
140         /** time when this thanle was started */
141         cfs_time_t oth_started;
142 #endif
143 };
144
145 /*
146  * Helpers.
147  */
148 static int lu_device_is_osd(const struct lu_device *d)
149 {
150         return ergo(d != NULL && d->ld_ops != NULL, d->ld_ops == &osd_lu_ops);
151 }
152
153 static struct osd_device *osd_dt_dev(const struct dt_device *d)
154 {
155         LASSERT(lu_device_is_osd(&d->dd_lu_dev));
156         return container_of0(d, struct osd_device, od_dt_dev);
157 }
158
159 static struct osd_device *osd_dev(const struct lu_device *d)
160 {
161         LASSERT(lu_device_is_osd(d));
162         return osd_dt_dev(container_of0(d, struct dt_device, dd_lu_dev));
163 }
164
165 static struct osd_device *osd_obj2dev(const struct osd_object *o)
166 {
167         return osd_dev(o->oo_dt.do_lu.lo_dev);
168 }
169
170 static struct super_block *osd_sb(const struct osd_device *dev)
171 {
172         return dev->od_mount->lmi_mnt->mnt_sb;
173 }
174
175 static int osd_object_is_root(const struct osd_object *obj)
176 {
177         return osd_sb(osd_obj2dev(obj))->s_root->d_inode == obj->oo_inode;
178 }
179
180 static struct osd_object *osd_obj(const struct lu_object *o)
181 {
182         LASSERT(lu_device_is_osd(o->lo_dev));
183         return container_of0(o, struct osd_object, oo_dt.do_lu);
184 }
185
186 static struct osd_object *osd_dt_obj(const struct dt_object *d)
187 {
188         return osd_obj(&d->do_lu);
189 }
190
191 static struct lu_device *osd2lu_dev(struct osd_device *osd)
192 {
193         return &osd->od_dt_dev.dd_lu_dev;
194 }
195
196 static journal_t *osd_journal(const struct osd_device *dev)
197 {
198         return LDISKFS_SB(osd_sb(dev))->s_journal;
199 }
200
201 static int osd_has_index(const struct osd_object *obj)
202 {
203         return obj->oo_dt.do_index_ops != NULL;
204 }
205
206 static int osd_object_invariant(const struct lu_object *l)
207 {
208         return osd_invariant(osd_obj(l));
209 }
210
211 #ifdef HAVE_QUOTA_SUPPORT
212 static inline void
213 osd_push_ctxt(const struct lu_env *env, struct osd_ctxt *save)
214 {
215         struct md_ucred    *uc = md_ucred(env);
216
217         LASSERT(uc != NULL);
218
219         save->oc_uid = current->fsuid;
220         save->oc_gid = current->fsgid;
221         save->oc_cap = current->cap_effective;
222         current->fsuid         = uc->mu_fsuid;
223         current->fsgid         = uc->mu_fsgid;
224         current->cap_effective = uc->mu_cap;
225 }
226
227 static inline void
228 osd_pop_ctxt(struct osd_ctxt *save)
229 {
230         current->fsuid         = save->oc_uid;
231         current->fsgid         = save->oc_gid;
232         current->cap_effective = save->oc_cap;
233 }
234 #endif
235
236 static inline struct osd_thread_info *osd_oti_get(const struct lu_env *env)
237 {
238         return lu_context_key_get(&env->le_ctx, &osd_key);
239 }
240
241 /*
242  * Concurrency: doesn't matter
243  */
244 static int osd_read_locked(const struct lu_env *env, struct osd_object *o)
245 {
246         return osd_oti_get(env)->oti_r_locks > 0;
247 }
248
249 /*
250  * Concurrency: doesn't matter
251  */
252 static int osd_write_locked(const struct lu_env *env, struct osd_object *o)
253 {
254         struct osd_thread_info *oti = osd_oti_get(env);
255         return oti->oti_w_locks > 0 && o->oo_owner == env;
256 }
257
258 /*
259  * Concurrency: doesn't access mutable data
260  */
261 static int osd_root_get(const struct lu_env *env,
262                         struct dt_device *dev, struct lu_fid *f)
263 {
264         struct inode *inode;
265
266         inode = osd_sb(osd_dt_dev(dev))->s_root->d_inode;
267         LU_IGIF_BUILD(f, inode->i_ino, inode->i_generation);
268         return 0;
269 }
270
271 /*
272  * OSD object methods.
273  */
274
275 /*
276  * Concurrency: no concurrent access is possible that early in object
277  * life-cycle.
278  */
279 static struct lu_object *osd_object_alloc(const struct lu_env *env,
280                                           const struct lu_object_header *hdr,
281                                           struct lu_device *d)
282 {
283         struct osd_object *mo;
284
285         OBD_ALLOC_PTR(mo);
286         if (mo != NULL) {
287                 struct lu_object *l;
288
289                 l = &mo->oo_dt.do_lu;
290                 dt_object_init(&mo->oo_dt, NULL, d);
291                 if (osd_dev(d)->od_iop_mode)
292                         mo->oo_dt.do_ops = &osd_obj_ea_ops;
293                 else
294                         mo->oo_dt.do_ops = &osd_obj_ops;
295
296                 l->lo_ops = &osd_lu_obj_ops;
297                 cfs_init_rwsem(&mo->oo_sem);
298                 cfs_init_rwsem(&mo->oo_ext_idx_sem);
299                 cfs_spin_lock_init(&mo->oo_guard);
300                 return l;
301         } else
302                 return NULL;
303 }
304
305 /*
306  * retrieve object from backend ext fs.
307  **/
308 static struct inode *osd_iget(struct osd_thread_info *info,
309                               struct osd_device *dev,
310                               const struct osd_inode_id *id)
311 {
312         struct inode *inode = NULL;
313
314 #ifdef HAVE_EXT4_LDISKFS
315         inode = ldiskfs_iget(osd_sb(dev), id->oii_ino);
316         if (IS_ERR(inode))
317         /* Newer kernels return an error instead of a NULL pointer */
318                 inode = NULL;
319 #else
320         inode = iget(osd_sb(dev), id->oii_ino);
321 #endif
322         if (inode == NULL) {
323                 CERROR("no inode\n");
324                 inode = ERR_PTR(-EACCES);
325         } else if (id->oii_gen != OSD_OII_NOGEN &&
326                    inode->i_generation != id->oii_gen) {
327                 iput(inode);
328                 inode = ERR_PTR(-ESTALE);
329         } else if (inode->i_nlink == 0) {
330                 /* due to parallel readdir and unlink,
331                 * we can have dead inode here. */
332                 CWARN("stale inode\n");
333                 make_bad_inode(inode);
334                 iput(inode);
335                 inode = ERR_PTR(-ESTALE);
336         } else if (is_bad_inode(inode)) {
337                 CERROR("bad inode %lx\n",inode->i_ino);
338                 iput(inode);
339                 inode = ERR_PTR(-ENOENT);
340         }
341         return inode;
342 }
343
344 static int osd_fid_lookup(const struct lu_env *env,
345                           struct osd_object *obj, const struct lu_fid *fid)
346 {
347         struct osd_thread_info *info;
348         struct lu_device       *ldev = obj->oo_dt.do_lu.lo_dev;
349         struct osd_device      *dev;
350         struct osd_inode_id    *id;
351         struct osd_oi          *oi;
352         struct inode           *inode;
353         int                     result;
354
355         LINVRNT(osd_invariant(obj));
356         LASSERT(obj->oo_inode == NULL);
357         LASSERT(fid_is_sane(fid) || osd_fid_is_root(fid));
358         /*
359          * This assertion checks that osd layer sees only local
360          * fids. Unfortunately it is somewhat expensive (does a
361          * cache-lookup). Disabling it for production/acceptance-testing.
362          */
363         LASSERT(1 || fid_is_local(env, ldev->ld_site, fid));
364
365         ENTRY;
366
367         info = osd_oti_get(env);
368         dev  = osd_dev(ldev);
369         id   = &info->oti_id;
370         oi   = &dev->od_oi;
371
372         if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOENT))
373                 RETURN(-ENOENT);
374
375         result = osd_oi_lookup(info, oi, fid, id);
376         if (result == 0) {
377                 inode = osd_iget(info, dev, id);
378                 if (!IS_ERR(inode)) {
379                         obj->oo_inode = inode;
380                         LASSERT(obj->oo_inode->i_sb == osd_sb(dev));
381                         if (dev->od_iop_mode) {
382                                 obj->oo_compat_dot_created = 1;
383                                 obj->oo_compat_dotdot_created = 1;
384                         }
385                         result = 0;
386                 } else
387                         /*
388                          * If fid wasn't found in oi, inode-less object is
389                          * created, for which lu_object_exists() returns
390                          * false. This is used in a (frequent) case when
391                          * objects are created as locking anchors or
392                          * place holders for objects yet to be created.
393                          */
394                         result = PTR_ERR(inode);
395         } else if (result == -ENOENT)
396                 result = 0;
397         LINVRNT(osd_invariant(obj));
398
399         RETURN(result);
400 }
401
402 /*
403  * Concurrency: shouldn't matter.
404  */
405 static void osd_object_init0(struct osd_object *obj)
406 {
407         LASSERT(obj->oo_inode != NULL);
408         obj->oo_dt.do_body_ops = &osd_body_ops;
409         obj->oo_dt.do_lu.lo_header->loh_attr |=
410                 (LOHA_EXISTS | (obj->oo_inode->i_mode & S_IFMT));
411 }
412
413 /*
414  * Concurrency: no concurrent access is possible that early in object
415  * life-cycle.
416  */
417 static int osd_object_init(const struct lu_env *env, struct lu_object *l,
418                            const struct lu_object_conf *unused)
419 {
420         struct osd_object *obj = osd_obj(l);
421         int result;
422
423         LINVRNT(osd_invariant(obj));
424
425         result = osd_fid_lookup(env, obj, lu_object_fid(l));
426         if (result == 0) {
427                 if (obj->oo_inode != NULL)
428                         osd_object_init0(obj);
429         }
430         LINVRNT(osd_invariant(obj));
431         return result;
432 }
433
434 /*
435  * Concurrency: no concurrent access is possible that late in object
436  * life-cycle.
437  */
438 static void osd_object_free(const struct lu_env *env, struct lu_object *l)
439 {
440         struct osd_object *obj = osd_obj(l);
441
442         LINVRNT(osd_invariant(obj));
443
444         dt_object_fini(&obj->oo_dt);
445         OBD_FREE_PTR(obj);
446 }
447
448 /**
449  * IAM Iterator
450  */
451 static struct iam_path_descr *osd_it_ipd_get(const struct lu_env *env,
452                                              const struct iam_container *bag)
453 {
454         return bag->ic_descr->id_ops->id_ipd_alloc(bag,
455                                            osd_oti_get(env)->oti_it_ipd);
456 }
457
458 static struct iam_path_descr *osd_idx_ipd_get(const struct lu_env *env,
459                                               const struct iam_container *bag)
460 {
461         return bag->ic_descr->id_ops->id_ipd_alloc(bag,
462                                            osd_oti_get(env)->oti_idx_ipd);
463 }
464
465 static void osd_ipd_put(const struct lu_env *env,
466                         const struct iam_container *bag,
467                         struct iam_path_descr *ipd)
468 {
469         bag->ic_descr->id_ops->id_ipd_free(ipd);
470 }
471
472 /*
473  * Concurrency: no concurrent access is possible that late in object
474  * life-cycle.
475  */
476 static void osd_index_fini(struct osd_object *o)
477 {
478         struct iam_container *bag;
479
480         if (o->oo_dir != NULL) {
481                 bag = &o->oo_dir->od_container;
482                 if (o->oo_inode != NULL) {
483                         if (bag->ic_object == o->oo_inode)
484                                 iam_container_fini(bag);
485                 }
486                 OBD_FREE_PTR(o->oo_dir);
487                 o->oo_dir = NULL;
488         }
489 }
490
491 /*
492  * Concurrency: no concurrent access is possible that late in object
493  * life-cycle (for all existing callers, that is. New callers have to provide
494  * their own locking.)
495  */
496 static int osd_inode_unlinked(const struct inode *inode)
497 {
498         return inode->i_nlink == 0;
499 }
500
501 enum {
502         OSD_TXN_OI_DELETE_CREDITS    = 20,
503         OSD_TXN_INODE_DELETE_CREDITS = 20
504 };
505
506 /*
507  * Journal
508  */
509
510 #if OSD_THANDLE_STATS
511 /**
512  * Set time when the handle is allocated
513  */
514 static void osd_th_alloced(struct osd_thandle *oth)
515 {
516         oth->oth_alloced = cfs_time_current();
517 }
518
519 /**
520  * Set time when the handle started
521  */
522 static void osd_th_started(struct osd_thandle *oth)
523 {
524         oth->oth_started = cfs_time_current();
525 }
526
527 /**
528  * Helper function to convert time interval to microseconds packed in
529  * long int (default time units for the counter in "stats" initialized
530  * by lu_time_init() )
531  */
532 static long interval_to_usec(cfs_time_t start, cfs_time_t end)
533 {
534         struct timeval val;
535
536         cfs_duration_usec(cfs_time_sub(end, start), &val);
537         return val.tv_sec * 1000000 + val.tv_usec;
538 }
539
540 /**
541  * Check whether the we deal with this handle for too long.
542  */
543 static void __osd_th_check_slow(void *oth, struct osd_device *dev,
544                                 cfs_time_t alloced, cfs_time_t started,
545                                 cfs_time_t closed)
546 {
547         cfs_time_t now = cfs_time_current();
548
549         LASSERT(dev != NULL);
550
551         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_STARTING,
552                             interval_to_usec(alloced, started));
553         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_OPEN,
554                             interval_to_usec(started, closed));
555         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_CLOSING,
556                             interval_to_usec(closed, now));
557
558         if (cfs_time_before(cfs_time_add(alloced, cfs_time_seconds(30)), now)) {
559                 CWARN("transaction handle %p was open for too long: "
560                       "now "CFS_TIME_T" ,"
561                       "alloced "CFS_TIME_T" ,"
562                       "started "CFS_TIME_T" ,"
563                       "closed "CFS_TIME_T"\n",
564                       oth, now, alloced, started, closed);
565                 libcfs_debug_dumpstack(NULL);
566         }
567 }
568
569 #define OSD_CHECK_SLOW_TH(oth, dev, expr)                               \
570 {                                                                       \
571         cfs_time_t __closed = cfs_time_current();                       \
572         cfs_time_t __alloced = oth->oth_alloced;                        \
573         cfs_time_t __started = oth->oth_started;                        \
574                                                                         \
575         expr;                                                           \
576         __osd_th_check_slow(oth, dev, __alloced, __started, __closed);  \
577 }
578
579 #else /* OSD_THANDLE_STATS */
580
581 #define osd_th_alloced(h)                  do {} while(0)
582 #define osd_th_started(h)                  do {} while(0)
583 #define OSD_CHECK_SLOW_TH(oth, dev, expr)  expr
584
585 #endif /* OSD_THANDLE_STATS */
586
587 /*
588  * Concurrency: doesn't access mutable data.
589  */
590 static int osd_param_is_sane(const struct osd_device *dev,
591                              const struct txn_param *param)
592 {
593         return param->tp_credits <= osd_journal(dev)->j_max_transaction_buffers;
594 }
595
596 /*
597  * Concurrency: shouldn't matter.
598  */
599 static void osd_trans_commit_cb(struct journal_callback *jcb, int error)
600 {
601         struct osd_thandle *oh = container_of0(jcb, struct osd_thandle, ot_jcb);
602         struct thandle     *th  = &oh->ot_super;
603         struct dt_device   *dev = th->th_dev;
604         struct lu_device   *lud = &dev->dd_lu_dev;
605
606         LASSERT(dev != NULL);
607         LASSERT(oh->ot_handle == NULL);
608
609         if (error) {
610                 CERROR("transaction @0x%p commit error: %d\n", th, error);
611         } else {
612                 struct lu_env *env = &osd_dt_dev(dev)->od_env_for_commit;
613                 /*
614                  * This od_env_for_commit is only for commit usage.  see
615                  * "struct dt_device"
616                  */
617                 lu_context_enter(&env->le_ctx);
618                 dt_txn_hook_commit(env, th);
619                 lu_context_exit(&env->le_ctx);
620         }
621
622         lu_ref_del_at(&lud->ld_reference, oh->ot_dev_link, "osd-tx", th);
623         lu_device_put(lud);
624         th->th_dev = NULL;
625
626         lu_context_exit(&th->th_ctx);
627         lu_context_fini(&th->th_ctx);
628         OBD_FREE_PTR(oh);
629 }
630
631 /*
632  * Concurrency: shouldn't matter.
633  */
634 static struct thandle *osd_trans_start(const struct lu_env *env,
635                                        struct dt_device *d,
636                                        struct txn_param *p)
637 {
638         struct osd_device  *dev = osd_dt_dev(d);
639         handle_t           *jh;
640         struct osd_thandle *oh;
641         struct thandle     *th;
642         int hook_res;
643
644         ENTRY;
645
646         hook_res = dt_txn_hook_start(env, d, p);
647         if (hook_res != 0)
648                 RETURN(ERR_PTR(hook_res));
649
650         if (osd_param_is_sane(dev, p)) {
651                 OBD_ALLOC_GFP(oh, sizeof *oh, CFS_ALLOC_IO);
652                 if (oh != NULL) {
653                         struct osd_thread_info *oti = osd_oti_get(env);
654
655                         /*
656                          * XXX temporary stuff. Some abstraction layer should
657                          * be used.
658                          */
659                         oti->oti_dev = dev;
660                         osd_th_alloced(oh);
661                         jh = ldiskfs_journal_start_sb(osd_sb(dev), p->tp_credits);
662                         osd_th_started(oh);
663                         if (!IS_ERR(jh)) {
664                                 oh->ot_handle = jh;
665                                 th = &oh->ot_super;
666                                 th->th_dev = d;
667                                 th->th_result = 0;
668                                 jh->h_sync = p->tp_sync;
669                                 lu_device_get(&d->dd_lu_dev);
670                                 oh->ot_dev_link = lu_ref_add
671                                         (&d->dd_lu_dev.ld_reference,
672                                          "osd-tx", th);
673                                 /* add commit callback */
674                                 lu_context_init(&th->th_ctx, LCT_TX_HANDLE);
675                                 lu_context_enter(&th->th_ctx);
676                                 osd_journal_callback_set(jh, osd_trans_commit_cb,
677                                                          (struct journal_callback *)&oh->ot_jcb);
678                                         LASSERT(oti->oti_txns == 0);
679                                         LASSERT(oti->oti_r_locks == 0);
680                                         LASSERT(oti->oti_w_locks == 0);
681                                         oti->oti_txns++;
682                         } else {
683                                 OBD_FREE_PTR(oh);
684                                 th = (void *)jh;
685                         }
686                 } else
687                         th = ERR_PTR(-ENOMEM);
688         } else {
689                 CERROR("Invalid transaction parameters\n");
690                 th = ERR_PTR(-EINVAL);
691         }
692
693         RETURN(th);
694 }
695
696 /*
697  * Concurrency: shouldn't matter.
698  */
699 static void osd_trans_stop(const struct lu_env *env, struct thandle *th)
700 {
701         int result;
702         struct osd_thandle *oh;
703         struct osd_thread_info *oti = osd_oti_get(env);
704
705         ENTRY;
706
707         oh = container_of0(th, struct osd_thandle, ot_super);
708         if (oh->ot_handle != NULL) {
709                 handle_t *hdl = oh->ot_handle;
710
711                 LASSERT(oti->oti_txns == 1);
712                 oti->oti_txns--;
713                 LASSERT(oti->oti_r_locks == 0);
714                 LASSERT(oti->oti_w_locks == 0);
715                 result = dt_txn_hook_stop(env, th);
716                 if (result != 0)
717                         CERROR("Failure in transaction hook: %d\n", result);
718                 oh->ot_handle = NULL;
719                 OSD_CHECK_SLOW_TH(oh, oti->oti_dev,
720                                   result = ldiskfs_journal_stop(hdl));
721                 if (result != 0)
722                         CERROR("Failure to stop transaction: %d\n", result);
723         }
724         EXIT;
725 }
726
727 /*
728  * Concurrency: no concurrent access is possible that late in object
729  * life-cycle.
730  */
731 static int osd_inode_remove(const struct lu_env *env, struct osd_object *obj)
732 {
733         const struct lu_fid    *fid = lu_object_fid(&obj->oo_dt.do_lu);
734         struct osd_device      *osd = osd_obj2dev(obj);
735         struct osd_thread_info *oti = osd_oti_get(env);
736         struct txn_param       *prm = &oti->oti_txn;
737         struct lu_env          *env_del_obj = &oti->oti_obj_delete_tx_env;
738         struct thandle         *th;
739         int result;
740
741         lu_env_init(env_del_obj, LCT_DT_THREAD);
742         txn_param_init(prm, OSD_TXN_OI_DELETE_CREDITS +
743                             OSD_TXN_INODE_DELETE_CREDITS);
744         th = osd_trans_start(env_del_obj, &osd->od_dt_dev, prm);
745         if (!IS_ERR(th)) {
746                 result = osd_oi_delete(osd_oti_get(env_del_obj),
747                                        &osd->od_oi, fid, th);
748                 osd_trans_stop(env_del_obj, th);
749         } else
750                 result = PTR_ERR(th);
751
752         lu_env_fini(env_del_obj);
753         return result;
754 }
755
756 /*
757  * Called just before object is freed. Releases all resources except for
758  * object itself (that is released by osd_object_free()).
759  *
760  * Concurrency: no concurrent access is possible that late in object
761  * life-cycle.
762  */
763 static void osd_object_delete(const struct lu_env *env, struct lu_object *l)
764 {
765         struct osd_object *obj   = osd_obj(l);
766         struct inode      *inode = obj->oo_inode;
767
768         LINVRNT(osd_invariant(obj));
769
770         /*
771          * If object is unlinked remove fid->ino mapping from object index.
772          */
773
774         osd_index_fini(obj);
775         if (inode != NULL) {
776                 int result;
777
778                 if (osd_inode_unlinked(inode)) {
779                         result = osd_inode_remove(env, obj);
780                         if (result != 0)
781                                 LU_OBJECT_DEBUG(D_ERROR, env, l,
782                                                 "Failed to cleanup: %d\n",
783                                                 result);
784                 }
785
786                 iput(inode);
787                 obj->oo_inode = NULL;
788         }
789 }
790
791 /*
792  * Concurrency: ->loo_object_release() is called under site spin-lock.
793  */
794 static void osd_object_release(const struct lu_env *env,
795                                struct lu_object *l)
796 {
797         struct osd_object *o = osd_obj(l);
798
799         LASSERT(!lu_object_is_dying(l->lo_header));
800         if (o->oo_inode != NULL && osd_inode_unlinked(o->oo_inode))
801                 cfs_set_bit(LU_OBJECT_HEARD_BANSHEE, &l->lo_header->loh_flags);
802 }
803
804 /*
805  * Concurrency: shouldn't matter.
806  */
807 static int osd_object_print(const struct lu_env *env, void *cookie,
808                             lu_printer_t p, const struct lu_object *l)
809 {
810         struct osd_object *o = osd_obj(l);
811         struct iam_descr  *d;
812
813         if (o->oo_dir != NULL)
814                 d = o->oo_dir->od_container.ic_descr;
815         else
816                 d = NULL;
817         return (*p)(env, cookie, LUSTRE_OSD_NAME"-object@%p(i:%p:%lu/%u)[%s]",
818                     o, o->oo_inode,
819                     o->oo_inode ? o->oo_inode->i_ino : 0UL,
820                     o->oo_inode ? o->oo_inode->i_generation : 0,
821                     d ? d->id_ops->id_name : "plain");
822 }
823
824 /*
825  * Concurrency: shouldn't matter.
826  */
827 int osd_statfs(const struct lu_env *env, struct dt_device *d,
828                cfs_kstatfs_t *sfs)
829 {
830         struct osd_device *osd = osd_dt_dev(d);
831         struct super_block *sb = osd_sb(osd);
832         int result = 0;
833
834         cfs_spin_lock(&osd->od_osfs_lock);
835         /* cache 1 second */
836         if (cfs_time_before_64(osd->od_osfs_age, cfs_time_shift_64(-1))) {
837                 result = ll_do_statfs(sb, &osd->od_kstatfs);
838                 if (likely(result == 0)) /* N.B. statfs can't really fail */
839                         osd->od_osfs_age = cfs_time_current_64();
840         }
841
842         if (likely(result == 0))
843                 *sfs = osd->od_kstatfs;
844         cfs_spin_unlock(&osd->od_osfs_lock);
845
846         return result;
847 }
848
849 /*
850  * Concurrency: doesn't access mutable data.
851  */
852 static void osd_conf_get(const struct lu_env *env,
853                          const struct dt_device *dev,
854                          struct dt_device_param *param)
855 {
856         /*
857          * XXX should be taken from not-yet-existing fs abstraction layer.
858          */
859         param->ddp_max_name_len  = LDISKFS_NAME_LEN;
860         param->ddp_max_nlink     = LDISKFS_LINK_MAX;
861         param->ddp_block_shift   = osd_sb(osd_dt_dev(dev))->s_blocksize_bits;
862 }
863
864 /**
865  * Helper function to get and fill the buffer with input values.
866  */
867 static struct lu_buf *osd_buf_get(const struct lu_env *env, void *area, ssize_t len)
868 {
869         struct lu_buf *buf;
870
871         buf = &osd_oti_get(env)->oti_buf;
872         buf->lb_buf = area;
873         buf->lb_len = len;
874         return buf;
875 }
876
877 /*
878  * Concurrency: shouldn't matter.
879  */
880 static int osd_sync(const struct lu_env *env, struct dt_device *d)
881 {
882         CDEBUG(D_HA, "syncing OSD %s\n", LUSTRE_OSD_NAME);
883         return ldiskfs_force_commit(osd_sb(osd_dt_dev(d)));
884 }
885
886 /**
887  * Start commit for OSD device.
888  *
889  * An implementation of dt_commit_async method for OSD device.
890  * Asychronously starts underlayng fs sync and thereby a transaction
891  * commit.
892  *
893  * \param env environment
894  * \param d dt device
895  *
896  * \see dt_device_operations
897  */
898 static int osd_commit_async(const struct lu_env *env,
899                             struct dt_device *d)
900 {
901         struct super_block *s = osd_sb(osd_dt_dev(d));
902         ENTRY;
903
904         CDEBUG(D_HA, "async commit OSD %s\n", LUSTRE_OSD_NAME);
905         RETURN(s->s_op->sync_fs(s, 0));
906 }
907
908 /*
909  * Concurrency: shouldn't matter.
910  */
911 lvfs_sbdev_type fsfilt_ldiskfs_journal_sbdev(struct super_block *);
912
913 static void osd_ro(const struct lu_env *env, struct dt_device *d)
914 {
915         ENTRY;
916
917         CERROR("*** setting device %s read-only ***\n", LUSTRE_OSD_NAME);
918
919         __lvfs_set_rdonly(lvfs_sbdev(osd_sb(osd_dt_dev(d))),
920                           fsfilt_ldiskfs_journal_sbdev(osd_sb(osd_dt_dev(d))));
921         EXIT;
922 }
923
924
925 /*
926  * Concurrency: serialization provided by callers.
927  */
928 static int osd_init_capa_ctxt(const struct lu_env *env, struct dt_device *d,
929                               int mode, unsigned long timeout, __u32 alg,
930                               struct lustre_capa_key *keys)
931 {
932         struct osd_device *dev = osd_dt_dev(d);
933         ENTRY;
934
935         dev->od_fl_capa = mode;
936         dev->od_capa_timeout = timeout;
937         dev->od_capa_alg = alg;
938         dev->od_capa_keys = keys;
939         RETURN(0);
940 }
941
942 /**
943  * Concurrency: serialization provided by callers.
944  */
945 static void osd_init_quota_ctxt(const struct lu_env *env, struct dt_device *d,
946                                struct dt_quota_ctxt *ctxt, void *data)
947 {
948         struct obd_device *obd = (void *)ctxt;
949         struct vfsmount *mnt = (struct vfsmount *)data;
950         ENTRY;
951
952         obd->u.obt.obt_sb = mnt->mnt_root->d_inode->i_sb;
953         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
954         obd->obd_lvfs_ctxt.pwdmnt = mnt;
955         obd->obd_lvfs_ctxt.pwd = mnt->mnt_root;
956         obd->obd_lvfs_ctxt.fs = get_ds();
957
958         EXIT;
959 }
960
961 /**
962  * Note: we do not count into QUOTA here.
963  * If we mount with --data_journal we may need more.
964  */
965 static const int osd_dto_credits_noquota[DTO_NR] = {
966         /**
967          * Insert/Delete.
968          * INDEX_EXTRA_TRANS_BLOCKS(8) +
969          * SINGLEDATA_TRANS_BLOCKS(8)
970          * XXX Note: maybe iam need more, since iam have more level than
971          *           EXT3 htree.
972          */
973         [DTO_INDEX_INSERT]  = 16,
974         [DTO_INDEX_DELETE]  = 16,
975         /**
976          * Unused now
977          */
978         [DTO_IDNEX_UPDATE]  = 16,
979         /**
980          * Create a object. The same as create object in EXT3.
981          * DATA_TRANS_BLOCKS(14) +
982          * INDEX_EXTRA_BLOCKS(8) +
983          * 3(inode bits, groups, GDT)
984          */
985         [DTO_OBJECT_CREATE] = 25,
986         /**
987          * Unused now
988          */
989         [DTO_OBJECT_DELETE] = 25,
990         /**
991          * Attr set credits.
992          * 3(inode bits, group, GDT)
993          */
994         [DTO_ATTR_SET_BASE] = 3,
995         /**
996          * Xattr set. The same as xattr of EXT3.
997          * DATA_TRANS_BLOCKS(14)
998          * XXX Note: in original MDS implmentation INDEX_EXTRA_TRANS_BLOCKS
999          * are also counted in. Do not know why?
1000          */
1001         [DTO_XATTR_SET]     = 14,
1002         [DTO_LOG_REC]       = 14,
1003         /**
1004          * creadits for inode change during write.
1005          */
1006         [DTO_WRITE_BASE]    = 3,
1007         /**
1008          * credits for single block write.
1009          */
1010         [DTO_WRITE_BLOCK]   = 14,
1011         /**
1012          * Attr set credits for chown.
1013          * This is extra credits for setattr, and it is null without quota
1014          */
1015         [DTO_ATTR_SET_CHOWN]= 0
1016 };
1017
1018 /**
1019  * Note: we count into QUOTA here.
1020  * If we mount with --data_journal we may need more.
1021  */
1022 static const int osd_dto_credits_quota[DTO_NR] = {
1023         /**
1024          * INDEX_EXTRA_TRANS_BLOCKS(8) +
1025          * SINGLEDATA_TRANS_BLOCKS(8) +
1026          * 2 * QUOTA_TRANS_BLOCKS(2)
1027          */
1028         [DTO_INDEX_INSERT]  = 20,
1029         /**
1030          * INDEX_EXTRA_TRANS_BLOCKS(8) +
1031          * SINGLEDATA_TRANS_BLOCKS(8) +
1032          * 2 * QUOTA_TRANS_BLOCKS(2)
1033          */
1034         [DTO_INDEX_DELETE]  = 20,
1035         /**
1036          * Unused now.
1037          */
1038         [DTO_IDNEX_UPDATE]  = 16,
1039         /*
1040          * Create a object. Same as create object in EXT3 filesystem.
1041          * DATA_TRANS_BLOCKS(16) +
1042          * INDEX_EXTRA_BLOCKS(8) +
1043          * 3(inode bits, groups, GDT) +
1044          * 2 * QUOTA_INIT_BLOCKS(25)
1045          */
1046         [DTO_OBJECT_CREATE] = 77,
1047         /*
1048          * Unused now.
1049          * DATA_TRANS_BLOCKS(16) +
1050          * INDEX_EXTRA_BLOCKS(8) +
1051          * 3(inode bits, groups, GDT) +
1052          * QUOTA(?)
1053          */
1054         [DTO_OBJECT_DELETE] = 27,
1055         /**
1056          * Attr set credits.
1057          * 3 (inode bit, group, GDT) +
1058          */
1059         [DTO_ATTR_SET_BASE] = 3,
1060         /**
1061          * Xattr set. The same as xattr of EXT3.
1062          * DATA_TRANS_BLOCKS(16)
1063          * XXX Note: in original MDS implmentation INDEX_EXTRA_TRANS_BLOCKS are
1064          *           also counted in. Do not know why?
1065          */
1066         [DTO_XATTR_SET]     = 16,
1067         [DTO_LOG_REC]       = 16,
1068         /**
1069          * creadits for inode change during write.
1070          */
1071         [DTO_WRITE_BASE]    = 3,
1072         /**
1073          * credits for single block write.
1074          */
1075         [DTO_WRITE_BLOCK]   = 16,
1076         /**
1077          * Attr set credits for chown.
1078          * It is added to already set setattr credits
1079          * 2 * QUOTA_INIT_BLOCKS(25) +
1080          * 2 * QUOTA_DEL_BLOCKS(9)
1081          */
1082         [DTO_ATTR_SET_CHOWN]= 68,
1083 };
1084
1085 static int osd_credit_get(const struct lu_env *env, struct dt_device *d,
1086                           enum dt_txn_op op)
1087 {
1088         LASSERT(ARRAY_SIZE(osd_dto_credits_noquota) ==
1089                 ARRAY_SIZE(osd_dto_credits_quota));
1090         LASSERT(0 <= op && op < ARRAY_SIZE(osd_dto_credits_noquota));
1091 #ifdef HAVE_QUOTA_SUPPORT
1092         if (test_opt(osd_sb(osd_dt_dev(d)), QUOTA))
1093                 return osd_dto_credits_quota[op];
1094         else
1095 #endif
1096                 return osd_dto_credits_noquota[op];
1097 }
1098
1099 static const struct dt_device_operations osd_dt_ops = {
1100         .dt_root_get       = osd_root_get,
1101         .dt_statfs         = osd_statfs,
1102         .dt_trans_start    = osd_trans_start,
1103         .dt_trans_stop     = osd_trans_stop,
1104         .dt_conf_get       = osd_conf_get,
1105         .dt_sync           = osd_sync,
1106         .dt_ro             = osd_ro,
1107         .dt_commit_async   = osd_commit_async,
1108         .dt_credit_get     = osd_credit_get,
1109         .dt_init_capa_ctxt = osd_init_capa_ctxt,
1110         .dt_init_quota_ctxt= osd_init_quota_ctxt,
1111 };
1112
1113 static void osd_object_read_lock(const struct lu_env *env,
1114                                  struct dt_object *dt, unsigned role)
1115 {
1116         struct osd_object *obj = osd_dt_obj(dt);
1117         struct osd_thread_info *oti = osd_oti_get(env);
1118
1119         LINVRNT(osd_invariant(obj));
1120
1121         LASSERT(obj->oo_owner != env);
1122         cfs_down_read_nested(&obj->oo_sem, role);
1123
1124         LASSERT(obj->oo_owner == NULL);
1125         oti->oti_r_locks++;
1126 }
1127
1128 static void osd_object_write_lock(const struct lu_env *env,
1129                                   struct dt_object *dt, unsigned role)
1130 {
1131         struct osd_object *obj = osd_dt_obj(dt);
1132         struct osd_thread_info *oti = osd_oti_get(env);
1133
1134         LINVRNT(osd_invariant(obj));
1135
1136         LASSERT(obj->oo_owner != env);
1137         cfs_down_write_nested(&obj->oo_sem, role);
1138
1139         LASSERT(obj->oo_owner == NULL);
1140         obj->oo_owner = env;
1141         oti->oti_w_locks++;
1142 }
1143
1144 static void osd_object_read_unlock(const struct lu_env *env,
1145                                    struct dt_object *dt)
1146 {
1147         struct osd_object *obj = osd_dt_obj(dt);
1148         struct osd_thread_info *oti = osd_oti_get(env);
1149
1150         LINVRNT(osd_invariant(obj));
1151
1152         LASSERT(oti->oti_r_locks > 0);
1153         oti->oti_r_locks--;
1154         cfs_up_read(&obj->oo_sem);
1155 }
1156
1157 static void osd_object_write_unlock(const struct lu_env *env,
1158                                     struct dt_object *dt)
1159 {
1160         struct osd_object *obj = osd_dt_obj(dt);
1161         struct osd_thread_info *oti = osd_oti_get(env);
1162
1163         LINVRNT(osd_invariant(obj));
1164
1165         LASSERT(obj->oo_owner == env);
1166         LASSERT(oti->oti_w_locks > 0);
1167         oti->oti_w_locks--;
1168         obj->oo_owner = NULL;
1169         cfs_up_write(&obj->oo_sem);
1170 }
1171
1172 static int osd_object_write_locked(const struct lu_env *env,
1173                                    struct dt_object *dt)
1174 {
1175         struct osd_object *obj = osd_dt_obj(dt);
1176
1177         LINVRNT(osd_invariant(obj));
1178
1179         return obj->oo_owner == env;
1180 }
1181
1182 static int capa_is_sane(const struct lu_env *env,
1183                         struct osd_device *dev,
1184                         struct lustre_capa *capa,
1185                         struct lustre_capa_key *keys)
1186 {
1187         struct osd_thread_info *oti = osd_oti_get(env);
1188         struct lustre_capa *tcapa = &oti->oti_capa;
1189         struct obd_capa *oc;
1190         int i, rc = 0;
1191         ENTRY;
1192
1193         oc = capa_lookup(dev->od_capa_hash, capa, 0);
1194         if (oc) {
1195                 if (capa_is_expired(oc)) {
1196                         DEBUG_CAPA(D_ERROR, capa, "expired");
1197                         rc = -ESTALE;
1198                 }
1199                 capa_put(oc);
1200                 RETURN(rc);
1201         }
1202
1203         if (capa_is_expired_sec(capa)) {
1204                 DEBUG_CAPA(D_ERROR, capa, "expired");
1205                 RETURN(-ESTALE);
1206         }
1207
1208         cfs_spin_lock(&capa_lock);
1209         for (i = 0; i < 2; i++) {
1210                 if (keys[i].lk_keyid == capa->lc_keyid) {
1211                         oti->oti_capa_key = keys[i];
1212                         break;
1213                 }
1214         }
1215         cfs_spin_unlock(&capa_lock);
1216
1217         if (i == 2) {
1218                 DEBUG_CAPA(D_ERROR, capa, "no matched capa key");
1219                 RETURN(-ESTALE);
1220         }
1221
1222         rc = capa_hmac(tcapa->lc_hmac, capa, oti->oti_capa_key.lk_key);
1223         if (rc)
1224                 RETURN(rc);
1225
1226         if (memcmp(tcapa->lc_hmac, capa->lc_hmac, sizeof(capa->lc_hmac))) {
1227                 DEBUG_CAPA(D_ERROR, capa, "HMAC mismatch");
1228                 RETURN(-EACCES);
1229         }
1230
1231         oc = capa_add(dev->od_capa_hash, capa);
1232         capa_put(oc);
1233
1234         RETURN(0);
1235 }
1236
1237 static int osd_object_auth(const struct lu_env *env, struct dt_object *dt,
1238                            struct lustre_capa *capa, __u64 opc)
1239 {
1240         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
1241         struct osd_device *dev = osd_dev(dt->do_lu.lo_dev);
1242         struct md_capainfo *ci;
1243         int rc;
1244
1245         if (!dev->od_fl_capa)
1246                 return 0;
1247
1248         if (capa == BYPASS_CAPA)
1249                 return 0;
1250
1251         ci = md_capainfo(env);
1252         if (unlikely(!ci))
1253                 return 0;
1254
1255         if (ci->mc_auth == LC_ID_NONE)
1256                 return 0;
1257
1258         if (!capa) {
1259                 CERROR("no capability is provided for fid "DFID"\n", PFID(fid));
1260                 return -EACCES;
1261         }
1262
1263         if (!lu_fid_eq(fid, &capa->lc_fid)) {
1264                 DEBUG_CAPA(D_ERROR, capa, "fid "DFID" mismatch with",
1265                            PFID(fid));
1266                 return -EACCES;
1267         }
1268
1269         if (!capa_opc_supported(capa, opc)) {
1270                 DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
1271                 return -EACCES;
1272         }
1273
1274         if ((rc = capa_is_sane(env, dev, capa, dev->od_capa_keys))) {
1275                 DEBUG_CAPA(D_ERROR, capa, "insane (rc %d)", rc);
1276                 return -EACCES;
1277         }
1278
1279         return 0;
1280 }
1281
1282 static struct timespec *osd_inode_time(const struct lu_env *env,
1283                                        struct inode *inode, __u64 seconds)
1284 {
1285         struct osd_thread_info *oti = osd_oti_get(env);
1286         struct timespec        *t   = &oti->oti_time;
1287
1288         t->tv_sec  = seconds;
1289         t->tv_nsec = 0;
1290         *t = timespec_trunc(*t, get_sb_time_gran(inode->i_sb));
1291         return t;
1292 }
1293
1294
1295 static void osd_inode_getattr(const struct lu_env *env,
1296                               struct inode *inode, struct lu_attr *attr)
1297 {
1298         attr->la_valid      |= LA_ATIME | LA_MTIME | LA_CTIME | LA_MODE |
1299                                LA_SIZE | LA_BLOCKS | LA_UID | LA_GID |
1300                                LA_FLAGS | LA_NLINK | LA_RDEV | LA_BLKSIZE;
1301
1302         attr->la_atime      = LTIME_S(inode->i_atime);
1303         attr->la_mtime      = LTIME_S(inode->i_mtime);
1304         attr->la_ctime      = LTIME_S(inode->i_ctime);
1305         attr->la_mode       = inode->i_mode;
1306         attr->la_size       = i_size_read(inode);
1307         attr->la_blocks     = inode->i_blocks;
1308         attr->la_uid        = inode->i_uid;
1309         attr->la_gid        = inode->i_gid;
1310         attr->la_flags      = LDISKFS_I(inode)->i_flags;
1311         attr->la_nlink      = inode->i_nlink;
1312         attr->la_rdev       = inode->i_rdev;
1313         attr->la_blksize    = ll_inode_blksize(inode);
1314         attr->la_blkbits    = inode->i_blkbits;
1315 }
1316
1317 static int osd_attr_get(const struct lu_env *env,
1318                         struct dt_object *dt,
1319                         struct lu_attr *attr,
1320                         struct lustre_capa *capa)
1321 {
1322         struct osd_object *obj = osd_dt_obj(dt);
1323
1324         LASSERT(dt_object_exists(dt));
1325         LINVRNT(osd_invariant(obj));
1326
1327         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
1328                 return -EACCES;
1329
1330         cfs_spin_lock(&obj->oo_guard);
1331         osd_inode_getattr(env, obj->oo_inode, attr);
1332         cfs_spin_unlock(&obj->oo_guard);
1333         return 0;
1334 }
1335
1336 static int osd_inode_setattr(const struct lu_env *env,
1337                              struct inode *inode, const struct lu_attr *attr)
1338 {
1339         __u64 bits;
1340
1341         bits = attr->la_valid;
1342
1343         LASSERT(!(bits & LA_TYPE)); /* Huh? You want too much. */
1344
1345 #ifdef HAVE_QUOTA_SUPPORT
1346         if ((bits & LA_UID && attr->la_uid != inode->i_uid) ||
1347             (bits & LA_GID && attr->la_gid != inode->i_gid)) {
1348                 struct osd_ctxt *save = &osd_oti_get(env)->oti_ctxt;
1349                 struct iattr iattr;
1350                 int rc;
1351
1352                 iattr.ia_valid = 0;
1353                 if (bits & LA_UID)
1354                         iattr.ia_valid |= ATTR_UID;
1355                 if (bits & LA_GID)
1356                         iattr.ia_valid |= ATTR_GID;
1357                 iattr.ia_uid = attr->la_uid;
1358                 iattr.ia_gid = attr->la_gid;
1359                 osd_push_ctxt(env, save);
1360                 rc = DQUOT_TRANSFER(inode, &iattr) ? -EDQUOT : 0;
1361                 osd_pop_ctxt(save);
1362                 if (rc != 0)
1363                         return rc;
1364         }
1365 #endif
1366
1367         if (bits & LA_ATIME)
1368                 inode->i_atime  = *osd_inode_time(env, inode, attr->la_atime);
1369         if (bits & LA_CTIME)
1370                 inode->i_ctime  = *osd_inode_time(env, inode, attr->la_ctime);
1371         if (bits & LA_MTIME)
1372                 inode->i_mtime  = *osd_inode_time(env, inode, attr->la_mtime);
1373         if (bits & LA_SIZE) {
1374                 LDISKFS_I(inode)->i_disksize = attr->la_size;
1375                 i_size_write(inode, attr->la_size);
1376         }
1377
1378 #if 0
1379         /* OSD should not change "i_blocks" which is used by quota.
1380          * "i_blocks" should be changed by ldiskfs only. */
1381         if (bits & LA_BLOCKS)
1382                 inode->i_blocks = attr->la_blocks;
1383 #endif
1384         if (bits & LA_MODE)
1385                 inode->i_mode   = (inode->i_mode & S_IFMT) |
1386                         (attr->la_mode & ~S_IFMT);
1387         if (bits & LA_UID)
1388                 inode->i_uid    = attr->la_uid;
1389         if (bits & LA_GID)
1390                 inode->i_gid    = attr->la_gid;
1391         if (bits & LA_NLINK)
1392                 inode->i_nlink  = attr->la_nlink;
1393         if (bits & LA_RDEV)
1394                 inode->i_rdev   = attr->la_rdev;
1395
1396         if (bits & LA_FLAGS)
1397                 inode->i_flags = ll_ext_to_inode_flags(attr->la_flags);
1398         return 0;
1399 }
1400
1401 static int osd_attr_set(const struct lu_env *env,
1402                         struct dt_object *dt,
1403                         const struct lu_attr *attr,
1404                         struct thandle *handle,
1405                         struct lustre_capa *capa)
1406 {
1407         struct osd_object *obj = osd_dt_obj(dt);
1408         int rc;
1409
1410         LASSERT(handle != NULL);
1411         LASSERT(dt_object_exists(dt));
1412         LASSERT(osd_invariant(obj));
1413
1414         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
1415                 return -EACCES;
1416
1417         cfs_spin_lock(&obj->oo_guard);
1418         rc = osd_inode_setattr(env, obj->oo_inode, attr);
1419         cfs_spin_unlock(&obj->oo_guard);
1420
1421         if (!rc)
1422                 mark_inode_dirty(obj->oo_inode);
1423         return rc;
1424 }
1425
1426 /*
1427  * Object creation.
1428  *
1429  * XXX temporary solution.
1430  */
1431 static int osd_create_pre(struct osd_thread_info *info, struct osd_object *obj,
1432                           struct lu_attr *attr, struct thandle *th)
1433 {
1434         return 0;
1435 }
1436
1437 static int osd_create_post(struct osd_thread_info *info, struct osd_object *obj,
1438                            struct lu_attr *attr, struct thandle *th)
1439 {
1440         osd_object_init0(obj);
1441         return 0;
1442 }
1443
1444 static struct dentry * osd_child_dentry_get(const struct lu_env *env,
1445                                             struct osd_object *obj,
1446                                             const char *name,
1447                                             const int namelen)
1448 {
1449         struct osd_thread_info *info   = osd_oti_get(env);
1450         struct dentry *child_dentry = &info->oti_child_dentry;
1451         struct dentry *obj_dentry = &info->oti_obj_dentry;
1452
1453         obj_dentry->d_inode = obj->oo_inode;
1454         obj_dentry->d_sb = osd_sb(osd_obj2dev(obj));
1455         obj_dentry->d_name.hash = 0;
1456
1457         child_dentry->d_name.hash = 0;
1458         child_dentry->d_parent = obj_dentry;
1459         child_dentry->d_name.name = name;
1460         child_dentry->d_name.len = namelen;
1461         return child_dentry;
1462 }
1463
1464
1465 static int osd_mkfile(struct osd_thread_info *info, struct osd_object *obj,
1466                       cfs_umode_t mode,
1467                       struct dt_allocation_hint *hint,
1468                       struct thandle *th)
1469 {
1470         int result;
1471         struct osd_device  *osd = osd_obj2dev(obj);
1472         struct osd_thandle *oth;
1473         struct dt_object   *parent;
1474         struct inode       *inode;
1475 #ifdef HAVE_QUOTA_SUPPORT
1476         struct osd_ctxt    *save = &info->oti_ctxt;
1477 #endif
1478
1479         LINVRNT(osd_invariant(obj));
1480         LASSERT(obj->oo_inode == NULL);
1481
1482         oth = container_of(th, struct osd_thandle, ot_super);
1483         LASSERT(oth->ot_handle->h_transaction != NULL);
1484
1485         if (hint && hint->dah_parent)
1486                 parent = hint->dah_parent;
1487         else
1488                 parent = osd->od_obj_area;
1489
1490         LASSERT(parent != NULL);
1491         LASSERT(osd_dt_obj(parent)->oo_inode->i_op != NULL);
1492
1493 #ifdef HAVE_QUOTA_SUPPORT
1494         osd_push_ctxt(info->oti_env, save);
1495 #endif
1496         inode = ldiskfs_create_inode(oth->ot_handle,
1497                                      osd_dt_obj(parent)->oo_inode, mode);
1498 #ifdef HAVE_QUOTA_SUPPORT
1499         osd_pop_ctxt(save);
1500 #endif
1501         if (!IS_ERR(inode)) {
1502                 obj->oo_inode = inode;
1503                 result = 0;
1504         } else
1505                 result = PTR_ERR(inode);
1506         LINVRNT(osd_invariant(obj));
1507         return result;
1508 }
1509
1510 enum {
1511         OSD_NAME_LEN = 255
1512 };
1513
1514 static int osd_mkdir(struct osd_thread_info *info, struct osd_object *obj,
1515                      struct lu_attr *attr,
1516                      struct dt_allocation_hint *hint,
1517                      struct dt_object_format *dof,
1518                      struct thandle *th)
1519 {
1520         int result;
1521         struct osd_thandle *oth;
1522         struct osd_device *osd = osd_obj2dev(obj);
1523         __u32 mode = (attr->la_mode & (S_IFMT | S_IRWXUGO | S_ISVTX));
1524
1525         LASSERT(S_ISDIR(attr->la_mode));
1526
1527         oth = container_of(th, struct osd_thandle, ot_super);
1528         LASSERT(oth->ot_handle->h_transaction != NULL);
1529         result = osd_mkfile(info, obj, mode, hint, th);
1530         if (result == 0 && osd->od_iop_mode == 0) {
1531                 LASSERT(obj->oo_inode != NULL);
1532                 /*
1533                  * XXX uh-oh... call low-level iam function directly.
1534                  */
1535
1536                 result = iam_lvar_create(obj->oo_inode, OSD_NAME_LEN, 4,
1537                                          sizeof (struct osd_fid_pack),
1538                                          oth->ot_handle);
1539         }
1540         return result;
1541 }
1542
1543 static int osd_mk_index(struct osd_thread_info *info, struct osd_object *obj,
1544                         struct lu_attr *attr,
1545                         struct dt_allocation_hint *hint,
1546                         struct dt_object_format *dof,
1547                         struct thandle *th)
1548 {
1549         int result;
1550         struct osd_thandle *oth;
1551         const struct dt_index_features *feat = dof->u.dof_idx.di_feat;
1552
1553         __u32 mode = (attr->la_mode & (S_IFMT | S_IRWXUGO | S_ISVTX));
1554
1555         LASSERT(S_ISREG(attr->la_mode));
1556
1557         oth = container_of(th, struct osd_thandle, ot_super);
1558         LASSERT(oth->ot_handle->h_transaction != NULL);
1559
1560         result = osd_mkfile(info, obj, mode, hint, th);
1561         if (result == 0) {
1562                 LASSERT(obj->oo_inode != NULL);
1563                 if (feat->dif_flags & DT_IND_VARKEY)
1564                         result = iam_lvar_create(obj->oo_inode,
1565                                                  feat->dif_keysize_max,
1566                                                  feat->dif_ptrsize,
1567                                                  feat->dif_recsize_max,
1568                                                  oth->ot_handle);
1569                 else
1570                         result = iam_lfix_create(obj->oo_inode,
1571                                                  feat->dif_keysize_max,
1572                                                  feat->dif_ptrsize,
1573                                                  feat->dif_recsize_max,
1574                                                  oth->ot_handle);
1575
1576         }
1577         return result;
1578 }
1579
1580 static int osd_mkreg(struct osd_thread_info *info, struct osd_object *obj,
1581                      struct lu_attr *attr,
1582                      struct dt_allocation_hint *hint,
1583                      struct dt_object_format *dof,
1584                      struct thandle *th)
1585 {
1586         LASSERT(S_ISREG(attr->la_mode));
1587         return osd_mkfile(info, obj, (attr->la_mode &
1588                                (S_IFMT | S_IRWXUGO | S_ISVTX)), hint, th);
1589 }
1590
1591 static int osd_mksym(struct osd_thread_info *info, struct osd_object *obj,
1592                      struct lu_attr *attr,
1593                      struct dt_allocation_hint *hint,
1594                      struct dt_object_format *dof,
1595                      struct thandle *th)
1596 {
1597         LASSERT(S_ISLNK(attr->la_mode));
1598         return osd_mkfile(info, obj, (attr->la_mode &
1599                               (S_IFMT | S_IRWXUGO | S_ISVTX)), hint, th);
1600 }
1601
1602 static int osd_mknod(struct osd_thread_info *info, struct osd_object *obj,
1603                      struct lu_attr *attr,
1604                      struct dt_allocation_hint *hint,
1605                      struct dt_object_format *dof,
1606                      struct thandle *th)
1607 {
1608         cfs_umode_t mode = attr->la_mode & (S_IFMT | S_IRWXUGO | S_ISVTX);
1609         int result;
1610
1611         LINVRNT(osd_invariant(obj));
1612         LASSERT(obj->oo_inode == NULL);
1613         LASSERT(S_ISCHR(mode) || S_ISBLK(mode) ||
1614                 S_ISFIFO(mode) || S_ISSOCK(mode));
1615
1616         result = osd_mkfile(info, obj, mode, hint, th);
1617         if (result == 0) {
1618                 LASSERT(obj->oo_inode != NULL);
1619                 init_special_inode(obj->oo_inode, mode, attr->la_rdev);
1620         }
1621         LINVRNT(osd_invariant(obj));
1622         return result;
1623 }
1624
1625 typedef int (*osd_obj_type_f)(struct osd_thread_info *, struct osd_object *,
1626                               struct lu_attr *,
1627                               struct dt_allocation_hint *hint,
1628                               struct dt_object_format *dof,
1629                               struct thandle *);
1630
1631 static osd_obj_type_f osd_create_type_f(enum dt_format_type type)
1632 {
1633         osd_obj_type_f result;
1634
1635         switch (type) {
1636         case DFT_DIR:
1637                 result = osd_mkdir;
1638                 break;
1639         case DFT_REGULAR:
1640                 result = osd_mkreg;
1641                 break;
1642         case DFT_SYM:
1643                 result = osd_mksym;
1644                 break;
1645         case DFT_NODE:
1646                 result = osd_mknod;
1647                 break;
1648         case DFT_INDEX:
1649                 result = osd_mk_index;
1650                 break;
1651
1652         default:
1653                 LBUG();
1654                 break;
1655         }
1656         return result;
1657 }
1658
1659
1660 static void osd_ah_init(const struct lu_env *env, struct dt_allocation_hint *ah,
1661                         struct dt_object *parent, cfs_umode_t child_mode)
1662 {
1663         LASSERT(ah);
1664
1665         memset(ah, 0, sizeof(*ah));
1666         ah->dah_parent = parent;
1667         ah->dah_mode = child_mode;
1668 }
1669
1670 /**
1671  * Helper function for osd_object_create()
1672  *
1673  * \retval 0, on success
1674  */
1675 static int __osd_object_create(struct osd_thread_info *info,
1676                                struct osd_object *obj, struct lu_attr *attr,
1677                                struct dt_allocation_hint *hint,
1678                                struct dt_object_format *dof,
1679                                struct thandle *th)
1680 {
1681
1682         int result;
1683
1684         result = osd_create_pre(info, obj, attr, th);
1685         if (result == 0) {
1686                 result = osd_create_type_f(dof->dof_type)(info, obj,
1687                                            attr, hint, dof, th);
1688                 if (result == 0)
1689                         result = osd_create_post(info, obj, attr, th);
1690         }
1691         return result;
1692 }
1693
1694 /**
1695  * Helper function for osd_object_create()
1696  *
1697  * \retval 0, on success
1698  */
1699 static int __osd_oi_insert(const struct lu_env *env, struct osd_object *obj,
1700                            const struct lu_fid *fid, struct thandle *th)
1701 {
1702         struct osd_thread_info *info = osd_oti_get(env);
1703         struct osd_inode_id    *id   = &info->oti_id;
1704         struct osd_device      *osd  = osd_obj2dev(obj);
1705         struct md_ucred        *uc   = md_ucred(env);
1706
1707         LASSERT(obj->oo_inode != NULL);
1708         LASSERT(uc != NULL);
1709
1710         id->oii_ino = obj->oo_inode->i_ino;
1711         id->oii_gen = obj->oo_inode->i_generation;
1712
1713         return osd_oi_insert(info, &osd->od_oi, fid, id, th,
1714                              uc->mu_cap & CFS_CAP_SYS_RESOURCE_MASK);
1715 }
1716
1717 static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
1718                              struct lu_attr *attr,
1719                              struct dt_allocation_hint *hint,
1720                              struct dt_object_format *dof,
1721                              struct thandle *th)
1722 {
1723         const struct lu_fid    *fid    = lu_object_fid(&dt->do_lu);
1724         struct osd_object      *obj    = osd_dt_obj(dt);
1725         struct osd_thread_info *info   = osd_oti_get(env);
1726         int result;
1727
1728         ENTRY;
1729
1730         LINVRNT(osd_invariant(obj));
1731         LASSERT(!dt_object_exists(dt));
1732         LASSERT(osd_write_locked(env, obj));
1733         LASSERT(th != NULL);
1734
1735         result = __osd_object_create(info, obj, attr, hint, dof, th);
1736         if (result == 0)
1737                 result = __osd_oi_insert(env, obj, fid, th);
1738
1739         LASSERT(ergo(result == 0, dt_object_exists(dt)));
1740         LASSERT(osd_invariant(obj));
1741         RETURN(result);
1742 }
1743
1744 /**
1745  * Helper function for osd_xattr_set()
1746  */
1747 static int __osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
1748                            const struct lu_buf *buf, const char *name, int fl)
1749 {
1750         struct osd_object      *obj      = osd_dt_obj(dt);
1751         struct inode           *inode    = obj->oo_inode;
1752         struct osd_thread_info *info     = osd_oti_get(env);
1753         struct dentry          *dentry   = &info->oti_child_dentry;
1754         struct timespec        *t        = &info->oti_time;
1755         int                     fs_flags = 0;
1756         int  rc;
1757
1758         LASSERT(dt_object_exists(dt));
1759         LASSERT(inode->i_op != NULL && inode->i_op->setxattr != NULL);
1760         LASSERT(osd_write_locked(env, obj));
1761
1762         if (fl & LU_XATTR_REPLACE)
1763                 fs_flags |= XATTR_REPLACE;
1764
1765         if (fl & LU_XATTR_CREATE)
1766                 fs_flags |= XATTR_CREATE;
1767
1768         dentry->d_inode = inode;
1769         *t = inode->i_ctime;
1770         rc = inode->i_op->setxattr(dentry, name, buf->lb_buf,
1771                                    buf->lb_len, fs_flags);
1772         /* ctime should not be updated with server-side time. */
1773         cfs_spin_lock(&obj->oo_guard);
1774         inode->i_ctime = *t;
1775         cfs_spin_unlock(&obj->oo_guard);
1776         mark_inode_dirty(inode);
1777         return rc;
1778 }
1779
1780 /**
1781  * Put the fid into lustre_mdt_attrs, and then place the structure
1782  * inode's ea. This fid should not be altered during the life time
1783  * of the inode.
1784  *
1785  * \retval +ve, on success
1786  * \retval -ve, on error
1787  *
1788  * FIXME: It is good to have/use ldiskfs_xattr_set_handle() here
1789  */
1790 static int osd_ea_fid_set(const struct lu_env *env, struct dt_object *dt,
1791                           const struct lu_fid *fid)
1792 {
1793         struct osd_thread_info  *info      = osd_oti_get(env);
1794         struct lustre_mdt_attrs *mdt_attrs = &info->oti_mdt_attrs;
1795
1796         lustre_lma_init(mdt_attrs, fid);
1797         lustre_lma_swab(mdt_attrs);
1798         return __osd_xattr_set(env, dt,
1799                                osd_buf_get(env, mdt_attrs, sizeof *mdt_attrs),
1800                                XATTR_NAME_LMA, LU_XATTR_CREATE);
1801
1802 }
1803
1804 /**
1805  * Helper function to form igif
1806  */
1807 static inline void osd_igif_get(const struct lu_env *env, struct inode  *inode,
1808                                 struct lu_fid *fid)
1809 {
1810         LU_IGIF_BUILD(fid, inode->i_ino, inode->i_generation);
1811 }
1812
1813 /**
1814  * Helper function to pack the fid, ldiskfs stores fid in packed format.
1815  */
1816 void osd_fid_pack(struct osd_fid_pack *pack, const struct dt_rec *fid,
1817                   struct lu_fid *befider)
1818 {
1819         fid_cpu_to_be(befider, (struct lu_fid *)fid);
1820         memcpy(pack->fp_area, befider, sizeof(*befider));
1821         pack->fp_len =  sizeof(*befider) + 1;
1822 }
1823
1824 /**
1825  * ldiskfs supports fid in dirent, it is passed in dentry->d_fsdata.
1826  * lustre 1.8 also uses d_fsdata for passing other info to ldiskfs.
1827  * To have compatilibility with 1.8 ldiskfs driver we need to have
1828  * magic number at start of fid data.
1829  * \ldiskfs_dentry_param is used only to pass fid from osd to ldiskfs.
1830  * its inmemory API.
1831  */
1832 void osd_get_ldiskfs_dirent_param(struct ldiskfs_dentry_param *param,
1833                                   const struct dt_rec *fid)
1834 {
1835         param->edp_magic = LDISKFS_LUFID_MAGIC;
1836         param->edp_len =  sizeof(struct lu_fid) + 1;
1837
1838         fid_cpu_to_be((struct lu_fid *)param->edp_data,
1839                       (struct lu_fid *)fid);
1840 }
1841
1842 int osd_fid_unpack(struct lu_fid *fid, const struct osd_fid_pack *pack)
1843 {
1844         int result;
1845
1846         result = 0;
1847         switch (pack->fp_len) {
1848         case sizeof *fid + 1:
1849                 memcpy(fid, pack->fp_area, sizeof *fid);
1850                 fid_be_to_cpu(fid, fid);
1851                 break;
1852         default:
1853                 CERROR("Unexpected packed fid size: %d\n", pack->fp_len);
1854                 result = -EIO;
1855         }
1856         return result;
1857 }
1858
1859 /**
1860  * Try to read the fid from inode ea into dt_rec, if return value
1861  * i.e. rc is +ve, then we got fid, otherwise we will have to form igif
1862  *
1863  * \param fid object fid.
1864  *
1865  * \retval 0 on success
1866  */
1867 static int osd_ea_fid_get(const struct lu_env *env, struct osd_object *obj,
1868                           __u32 ino, struct lu_fid *fid)
1869 {
1870         struct osd_thread_info  *info      = osd_oti_get(env);
1871         struct lustre_mdt_attrs *mdt_attrs = &info->oti_mdt_attrs;
1872         struct lu_device        *ldev   = obj->oo_dt.do_lu.lo_dev;
1873         struct dentry           *dentry = &info->oti_child_dentry;
1874         struct osd_inode_id     *id     = &info->oti_id;
1875         struct osd_device       *dev;
1876         struct inode            *inode;
1877         int                      rc;
1878
1879         ENTRY;
1880         dev  = osd_dev(ldev);
1881
1882         id->oii_ino = ino;
1883         id->oii_gen = OSD_OII_NOGEN;
1884
1885         inode = osd_iget(info, dev, id);
1886         if (IS_ERR(inode)) {
1887                 rc = PTR_ERR(inode);
1888                 GOTO(out,rc);
1889         }
1890         dentry->d_inode = inode;
1891
1892         LASSERT(inode->i_op != NULL && inode->i_op->getxattr != NULL);
1893         rc = inode->i_op->getxattr(dentry, XATTR_NAME_LMA, (void *)mdt_attrs,
1894                                    sizeof *mdt_attrs);
1895
1896         /* Check LMA compatibility */
1897         if (rc > 0 &&
1898             (mdt_attrs->lma_incompat & ~cpu_to_le32(LMA_INCOMPAT_SUPP))) {
1899                 CWARN("Inode %lx: Unsupported incompat LMA feature(s) %#x\n",
1900                       inode->i_ino, le32_to_cpu(mdt_attrs->lma_incompat) &
1901                       ~LMA_INCOMPAT_SUPP);
1902                 return -ENOSYS;
1903         }
1904
1905         if (rc > 0) {
1906                 lustre_lma_swab(mdt_attrs);
1907                 memcpy(fid, &mdt_attrs->lma_self_fid, sizeof(*fid));
1908                 rc = 0;
1909         } else if (rc == -ENODATA) {
1910                 osd_igif_get(env, inode, fid);
1911                 rc = 0;
1912         }
1913         iput(inode);
1914 out:
1915         RETURN(rc);
1916 }
1917
1918 /**
1919  * OSD layer object create function for interoperability mode (b11826).
1920  * This is mostly similar to osd_object_create(). Only difference being, fid is
1921  * inserted into inode ea here.
1922  *
1923  * \retval   0, on success
1924  * \retval -ve, on error
1925  */
1926 static int osd_object_ea_create(const struct lu_env *env, struct dt_object *dt,
1927                              struct lu_attr *attr,
1928                              struct dt_allocation_hint *hint,
1929                              struct dt_object_format *dof,
1930                              struct thandle *th)
1931 {
1932         const struct lu_fid    *fid    = lu_object_fid(&dt->do_lu);
1933         struct osd_object      *obj    = osd_dt_obj(dt);
1934         struct osd_thread_info *info   = osd_oti_get(env);
1935         int result;
1936
1937         ENTRY;
1938
1939         LASSERT(osd_invariant(obj));
1940         LASSERT(!dt_object_exists(dt));
1941         LASSERT(osd_write_locked(env, obj));
1942         LASSERT(th != NULL);
1943
1944         result = __osd_object_create(info, obj, attr, hint, dof, th);
1945
1946         /* objects under osd root shld have igif fid, so dont add fid EA */
1947         if (result == 0 && fid_seq(fid) >= FID_SEQ_NORMAL)
1948                 result = osd_ea_fid_set(env, dt, fid);
1949
1950         if (result == 0)
1951                 result = __osd_oi_insert(env, obj, fid, th);
1952
1953         LASSERT(ergo(result == 0, dt_object_exists(dt)));
1954         LINVRNT(osd_invariant(obj));
1955         RETURN(result);
1956 }
1957
1958 /*
1959  * Concurrency: @dt is write locked.
1960  */
1961 static void osd_object_ref_add(const struct lu_env *env,
1962                                struct dt_object *dt,
1963                                struct thandle *th)
1964 {
1965         struct osd_object *obj = osd_dt_obj(dt);
1966         struct inode *inode = obj->oo_inode;
1967
1968         LINVRNT(osd_invariant(obj));
1969         LASSERT(dt_object_exists(dt));
1970         LASSERT(osd_write_locked(env, obj));
1971         LASSERT(th != NULL);
1972
1973         cfs_spin_lock(&obj->oo_guard);
1974         LASSERT(inode->i_nlink < LDISKFS_LINK_MAX);
1975         inode->i_nlink++;
1976         cfs_spin_unlock(&obj->oo_guard);
1977         mark_inode_dirty(inode);
1978         LINVRNT(osd_invariant(obj));
1979 }
1980
1981 /*
1982  * Concurrency: @dt is write locked.
1983  */
1984 static void osd_object_ref_del(const struct lu_env *env,
1985                                struct dt_object *dt,
1986                                struct thandle *th)
1987 {
1988         struct osd_object *obj = osd_dt_obj(dt);
1989         struct inode *inode = obj->oo_inode;
1990
1991         LINVRNT(osd_invariant(obj));
1992         LASSERT(dt_object_exists(dt));
1993         LASSERT(osd_write_locked(env, obj));
1994         LASSERT(th != NULL);
1995
1996         cfs_spin_lock(&obj->oo_guard);
1997         LASSERT(inode->i_nlink > 0);
1998         inode->i_nlink--;
1999         cfs_spin_unlock(&obj->oo_guard);
2000         mark_inode_dirty(inode);
2001         LINVRNT(osd_invariant(obj));
2002 }
2003
2004 /*
2005  * Concurrency: @dt is read locked.
2006  */
2007 static int osd_xattr_get(const struct lu_env *env,
2008                          struct dt_object *dt,
2009                          struct lu_buf *buf,
2010                          const char *name,
2011                          struct lustre_capa *capa)
2012 {
2013         struct osd_object      *obj    = osd_dt_obj(dt);
2014         struct inode           *inode  = obj->oo_inode;
2015         struct osd_thread_info *info   = osd_oti_get(env);
2016         struct dentry          *dentry = &info->oti_obj_dentry;
2017
2018         LASSERT(dt_object_exists(dt));
2019         LASSERT(inode->i_op != NULL && inode->i_op->getxattr != NULL);
2020         LASSERT(osd_read_locked(env, obj) || osd_write_locked(env, obj));
2021
2022         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
2023                 return -EACCES;
2024
2025         dentry->d_inode = inode;
2026         return inode->i_op->getxattr(dentry, name, buf->lb_buf, buf->lb_len);
2027 }
2028
2029 /*
2030  * Concurrency: @dt is write locked.
2031  */
2032 static int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
2033                          const struct lu_buf *buf, const char *name, int fl,
2034                          struct thandle *handle, struct lustre_capa *capa)
2035 {
2036         LASSERT(handle != NULL);
2037
2038         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
2039                 return -EACCES;
2040
2041         return __osd_xattr_set(env, dt, buf, name, fl);
2042 }
2043
2044 /*
2045  * Concurrency: @dt is read locked.
2046  */
2047 static int osd_xattr_list(const struct lu_env *env,
2048                           struct dt_object *dt,
2049                           struct lu_buf *buf,
2050                           struct lustre_capa *capa)
2051 {
2052         struct osd_object      *obj    = osd_dt_obj(dt);
2053         struct inode           *inode  = obj->oo_inode;
2054         struct osd_thread_info *info   = osd_oti_get(env);
2055         struct dentry          *dentry = &info->oti_obj_dentry;
2056
2057         LASSERT(dt_object_exists(dt));
2058         LASSERT(inode->i_op != NULL && inode->i_op->listxattr != NULL);
2059         LASSERT(osd_read_locked(env, obj) || osd_write_locked(env, obj));
2060
2061         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
2062                 return -EACCES;
2063
2064         dentry->d_inode = inode;
2065         return inode->i_op->listxattr(dentry, buf->lb_buf, buf->lb_len);
2066 }
2067
2068 /*
2069  * Concurrency: @dt is write locked.
2070  */
2071 static int osd_xattr_del(const struct lu_env *env,
2072                          struct dt_object *dt,
2073                          const char *name,
2074                          struct thandle *handle,
2075                          struct lustre_capa *capa)
2076 {
2077         struct osd_object      *obj    = osd_dt_obj(dt);
2078         struct inode           *inode  = obj->oo_inode;
2079         struct osd_thread_info *info   = osd_oti_get(env);
2080         struct dentry          *dentry = &info->oti_obj_dentry;
2081         struct timespec        *t      = &info->oti_time;
2082         int                     rc;
2083
2084         LASSERT(dt_object_exists(dt));
2085         LASSERT(inode->i_op != NULL && inode->i_op->removexattr != NULL);
2086         LASSERT(osd_write_locked(env, obj));
2087         LASSERT(handle != NULL);
2088
2089         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
2090                 return -EACCES;
2091
2092         dentry->d_inode = inode;
2093         *t = inode->i_ctime;
2094         rc = inode->i_op->removexattr(dentry, name);
2095         /* ctime should not be updated with server-side time. */
2096         cfs_spin_lock(&obj->oo_guard);
2097         inode->i_ctime = *t;
2098         cfs_spin_unlock(&obj->oo_guard);
2099         mark_inode_dirty(inode);
2100         return rc;
2101 }
2102
2103 static struct obd_capa *osd_capa_get(const struct lu_env *env,
2104                                      struct dt_object *dt,
2105                                      struct lustre_capa *old,
2106                                      __u64 opc)
2107 {
2108         struct osd_thread_info *info = osd_oti_get(env);
2109         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
2110         struct osd_object *obj = osd_dt_obj(dt);
2111         struct osd_device *dev = osd_obj2dev(obj);
2112         struct lustre_capa_key *key = &info->oti_capa_key;
2113         struct lustre_capa *capa = &info->oti_capa;
2114         struct obd_capa *oc;
2115         struct md_capainfo *ci;
2116         int rc;
2117         ENTRY;
2118
2119         if (!dev->od_fl_capa)
2120                 RETURN(ERR_PTR(-ENOENT));
2121
2122         LASSERT(dt_object_exists(dt));
2123         LINVRNT(osd_invariant(obj));
2124
2125         /* renewal sanity check */
2126         if (old && osd_object_auth(env, dt, old, opc))
2127                 RETURN(ERR_PTR(-EACCES));
2128
2129         ci = md_capainfo(env);
2130         if (unlikely(!ci))
2131                 RETURN(ERR_PTR(-ENOENT));
2132
2133         switch (ci->mc_auth) {
2134         case LC_ID_NONE:
2135                 RETURN(NULL);
2136         case LC_ID_PLAIN:
2137                 capa->lc_uid = obj->oo_inode->i_uid;
2138                 capa->lc_gid = obj->oo_inode->i_gid;
2139                 capa->lc_flags = LC_ID_PLAIN;
2140                 break;
2141         case LC_ID_CONVERT: {
2142                 __u32 d[4], s[4];
2143
2144                 s[0] = obj->oo_inode->i_uid;
2145                 cfs_get_random_bytes(&(s[1]), sizeof(__u32));
2146                 s[2] = obj->oo_inode->i_gid;
2147                 cfs_get_random_bytes(&(s[3]), sizeof(__u32));
2148                 rc = capa_encrypt_id(d, s, key->lk_key, CAPA_HMAC_KEY_MAX_LEN);
2149                 if (unlikely(rc))
2150                         RETURN(ERR_PTR(rc));
2151
2152                 capa->lc_uid   = ((__u64)d[1] << 32) | d[0];
2153                 capa->lc_gid   = ((__u64)d[3] << 32) | d[2];
2154                 capa->lc_flags = LC_ID_CONVERT;
2155                 break;
2156         }
2157         default:
2158                 RETURN(ERR_PTR(-EINVAL));
2159         }
2160
2161         capa->lc_fid = *fid;
2162         capa->lc_opc = opc;
2163         capa->lc_flags |= dev->od_capa_alg << 24;
2164         capa->lc_timeout = dev->od_capa_timeout;
2165         capa->lc_expiry = 0;
2166
2167         oc = capa_lookup(dev->od_capa_hash, capa, 1);
2168         if (oc) {
2169                 LASSERT(!capa_is_expired(oc));
2170                 RETURN(oc);
2171         }
2172
2173         cfs_spin_lock(&capa_lock);
2174         *key = dev->od_capa_keys[1];
2175         cfs_spin_unlock(&capa_lock);
2176
2177         capa->lc_keyid = key->lk_keyid;
2178         capa->lc_expiry = cfs_time_current_sec() + dev->od_capa_timeout;
2179
2180         rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
2181         if (rc) {
2182                 DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
2183                 RETURN(ERR_PTR(rc));
2184         }
2185
2186         oc = capa_add(dev->od_capa_hash, capa);
2187         RETURN(oc);
2188 }
2189
2190 static int osd_object_sync(const struct lu_env *env, struct dt_object *dt)
2191 {
2192         int rc;
2193         struct osd_object      *obj    = osd_dt_obj(dt);
2194         struct inode           *inode  = obj->oo_inode;
2195         struct osd_thread_info *info   = osd_oti_get(env);
2196         struct dentry          *dentry = &info->oti_obj_dentry;
2197         struct file            *file   = &info->oti_file;
2198         ENTRY;
2199
2200         dentry->d_inode = inode;
2201         file->f_dentry = dentry;
2202         file->f_mapping = inode->i_mapping;
2203         file->f_op = inode->i_fop;
2204         LOCK_INODE_MUTEX(inode);
2205         rc = file->f_op->fsync(file, dentry, 0);
2206         UNLOCK_INODE_MUTEX(inode);
2207         RETURN(rc);
2208 }
2209
2210 /*
2211  * Get the 64-bit version for an inode.
2212  */
2213 static dt_obj_version_t osd_object_version_get(const struct lu_env *env,
2214                                                struct dt_object *dt)
2215 {
2216         struct inode *inode = osd_dt_obj(dt)->oo_inode;
2217
2218         CDEBUG(D_INFO, "Get version "LPX64" for inode %lu\n",
2219                LDISKFS_I(inode)->i_fs_version, inode->i_ino);
2220         return LDISKFS_I(inode)->i_fs_version;
2221 }
2222
2223 /*
2224  * Set the 64-bit version and return the old version.
2225  */
2226 static void osd_object_version_set(const struct lu_env *env, struct dt_object *dt,
2227                                    dt_obj_version_t new_version)
2228 {
2229         struct inode *inode = osd_dt_obj(dt)->oo_inode;
2230
2231         CDEBUG(D_INFO, "Set version "LPX64" (old "LPX64") for inode %lu\n",
2232                new_version, LDISKFS_I(inode)->i_fs_version, inode->i_ino);
2233         LDISKFS_I(inode)->i_fs_version = new_version;
2234         /** Version is set after all inode operations are finished,
2235          *  so we should mark it dirty here */
2236         inode->i_sb->s_op->dirty_inode(inode);
2237 }
2238
2239 static int osd_data_get(const struct lu_env *env, struct dt_object *dt,
2240                         void **data)
2241 {
2242         struct osd_object *obj = osd_dt_obj(dt);
2243         ENTRY;
2244
2245         *data = (void *)obj->oo_inode;
2246         RETURN(0);
2247 }
2248
2249 /*
2250  * Index operations.
2251  */
2252
2253 static int osd_iam_index_probe(const struct lu_env *env, struct osd_object *o,
2254                            const struct dt_index_features *feat)
2255 {
2256         struct iam_descr *descr;
2257
2258         if (osd_object_is_root(o))
2259                 return feat == &dt_directory_features;
2260
2261         LASSERT(o->oo_dir != NULL);
2262
2263         descr = o->oo_dir->od_container.ic_descr;
2264         if (feat == &dt_directory_features) {
2265                 if (descr->id_rec_size == sizeof(struct osd_fid_pack))
2266                         return 1;
2267                 else
2268                         return 0;
2269         } else {
2270                 return
2271                         feat->dif_keysize_min <= descr->id_key_size &&
2272                         descr->id_key_size <= feat->dif_keysize_max &&
2273                         feat->dif_recsize_min <= descr->id_rec_size &&
2274                         descr->id_rec_size <= feat->dif_recsize_max &&
2275                         !(feat->dif_flags & (DT_IND_VARKEY |
2276                                              DT_IND_VARREC | DT_IND_NONUNQ)) &&
2277                         ergo(feat->dif_flags & DT_IND_UPDATE,
2278                              1 /* XXX check that object (and file system) is
2279                                 * writable */);
2280         }
2281 }
2282
2283 static int osd_iam_container_init(const struct lu_env *env,
2284                                   struct osd_object *obj,
2285                                   struct osd_directory *dir)
2286 {
2287         int result;
2288         struct iam_container *bag;
2289
2290         bag    = &dir->od_container;
2291         result = iam_container_init(bag, &dir->od_descr, obj->oo_inode);
2292         if (result == 0) {
2293                 result = iam_container_setup(bag);
2294                 if (result == 0)
2295                         obj->oo_dt.do_index_ops = &osd_index_iam_ops;
2296                 else
2297                         iam_container_fini(bag);
2298         }
2299         return result;
2300 }
2301
2302
2303 /*
2304  * Concurrency: no external locking is necessary.
2305  */
2306 static int osd_index_try(const struct lu_env *env, struct dt_object *dt,
2307                          const struct dt_index_features *feat)
2308 {
2309         int result;
2310         int ea_dir = 0;
2311         struct osd_object *obj = osd_dt_obj(dt);
2312         struct osd_device *osd = osd_obj2dev(obj);
2313
2314         LINVRNT(osd_invariant(obj));
2315         LASSERT(dt_object_exists(dt));
2316
2317         if (osd_object_is_root(obj)) {
2318                 dt->do_index_ops = &osd_index_ea_ops;
2319                 result = 0;
2320         } else if (feat == &dt_directory_features && osd->od_iop_mode) {
2321                 dt->do_index_ops = &osd_index_ea_ops;
2322                 if (S_ISDIR(obj->oo_inode->i_mode))
2323                         result = 0;
2324                 else
2325                         result = -ENOTDIR;
2326                 ea_dir = 1;
2327         } else if (!osd_has_index(obj)) {
2328                 struct osd_directory *dir;
2329
2330                 OBD_ALLOC_PTR(dir);
2331                 if (dir != NULL) {
2332
2333                         cfs_spin_lock(&obj->oo_guard);
2334                         if (obj->oo_dir == NULL)
2335                                 obj->oo_dir = dir;
2336                         else
2337                                 /*
2338                                  * Concurrent thread allocated container data.
2339                                  */
2340                                 OBD_FREE_PTR(dir);
2341                         cfs_spin_unlock(&obj->oo_guard);
2342                         /*
2343                          * Now, that we have container data, serialize its
2344                          * initialization.
2345                          */
2346                         cfs_down_write(&obj->oo_ext_idx_sem);
2347                         /*
2348                          * recheck under lock.
2349                          */
2350                         if (!osd_has_index(obj))
2351                                 result = osd_iam_container_init(env, obj, dir);
2352                         else
2353                                 result = 0;
2354                         cfs_up_write(&obj->oo_ext_idx_sem);
2355                 } else
2356                         result = -ENOMEM;
2357         } else
2358                 result = 0;
2359
2360         if (result == 0 && ea_dir == 0) {
2361                 if (!osd_iam_index_probe(env, obj, feat))
2362                         result = -ENOTDIR;
2363         }
2364         LINVRNT(osd_invariant(obj));
2365
2366         return result;
2367 }
2368
2369 static const struct dt_object_operations osd_obj_ops = {
2370         .do_read_lock    = osd_object_read_lock,
2371         .do_write_lock   = osd_object_write_lock,
2372         .do_read_unlock  = osd_object_read_unlock,
2373         .do_write_unlock = osd_object_write_unlock,
2374         .do_write_locked = osd_object_write_locked,
2375         .do_attr_get     = osd_attr_get,
2376         .do_attr_set     = osd_attr_set,
2377         .do_ah_init      = osd_ah_init,
2378         .do_create       = osd_object_create,
2379         .do_index_try    = osd_index_try,
2380         .do_ref_add      = osd_object_ref_add,
2381         .do_ref_del      = osd_object_ref_del,
2382         .do_xattr_get    = osd_xattr_get,
2383         .do_xattr_set    = osd_xattr_set,
2384         .do_xattr_del    = osd_xattr_del,
2385         .do_xattr_list   = osd_xattr_list,
2386         .do_capa_get     = osd_capa_get,
2387         .do_object_sync  = osd_object_sync,
2388         .do_version_get  = osd_object_version_get,
2389         .do_version_set  = osd_object_version_set,
2390         .do_data_get     = osd_data_get,
2391 };
2392
2393 /**
2394  * dt_object_operations for interoperability mode
2395  * (i.e. to run 2.0 mds on 1.8 disk) (b11826)
2396  */
2397 static const struct dt_object_operations osd_obj_ea_ops = {
2398         .do_read_lock    = osd_object_read_lock,
2399         .do_write_lock   = osd_object_write_lock,
2400         .do_read_unlock  = osd_object_read_unlock,
2401         .do_write_unlock = osd_object_write_unlock,
2402         .do_write_locked = osd_object_write_locked,
2403         .do_attr_get     = osd_attr_get,
2404         .do_attr_set     = osd_attr_set,
2405         .do_ah_init      = osd_ah_init,
2406         .do_create       = osd_object_ea_create,
2407         .do_index_try    = osd_index_try,
2408         .do_ref_add      = osd_object_ref_add,
2409         .do_ref_del      = osd_object_ref_del,
2410         .do_xattr_get    = osd_xattr_get,
2411         .do_xattr_set    = osd_xattr_set,
2412         .do_xattr_del    = osd_xattr_del,
2413         .do_xattr_list   = osd_xattr_list,
2414         .do_capa_get     = osd_capa_get,
2415         .do_object_sync  = osd_object_sync,
2416         .do_version_get  = osd_object_version_get,
2417         .do_version_set  = osd_object_version_set,
2418         .do_data_get     = osd_data_get,
2419 };
2420
2421 /*
2422  * Body operations.
2423  */
2424
2425 /*
2426  * XXX: Another layering violation for now.
2427  *
2428  * We don't want to use ->f_op->read methods, because generic file write
2429  *
2430  *         - serializes on ->i_sem, and
2431  *
2432  *         - does a lot of extra work like balance_dirty_pages(),
2433  *
2434  * which doesn't work for globally shared files like /last-received.
2435  */
2436 static int osd_ldiskfs_readlink(struct inode *inode, char *buffer, int buflen)
2437 {
2438         struct ldiskfs_inode_info *ei = LDISKFS_I(inode);
2439
2440         memcpy(buffer, (char*)ei->i_data, buflen);
2441
2442         return  buflen;
2443 }
2444
2445 static int osd_ldiskfs_read(struct inode *inode, void *buf, int size,
2446                             loff_t *offs)
2447 {
2448         struct buffer_head *bh;
2449         unsigned long block;
2450         int osize = size;
2451         int blocksize;
2452         int csize;
2453         int boffs;
2454         int err;
2455
2456         /* prevent reading after eof */
2457         spin_lock(&inode->i_lock);
2458         if (i_size_read(inode) < *offs + size) {
2459                 size = i_size_read(inode) - *offs;
2460                 spin_unlock(&inode->i_lock);
2461                 if (size < 0) {
2462                         CDEBUG(D_EXT2, "size %llu is too short to read @%llu\n",
2463                                i_size_read(inode), *offs);
2464                         return -EBADR;
2465                 } else if (size == 0) {
2466                         return 0;
2467                 }
2468         } else {
2469                 spin_unlock(&inode->i_lock);
2470         }
2471
2472         blocksize = 1 << inode->i_blkbits;
2473
2474         while (size > 0) {
2475                 block = *offs >> inode->i_blkbits;
2476                 boffs = *offs & (blocksize - 1);
2477                 csize = min(blocksize - boffs, size);
2478                 bh = ldiskfs_bread(NULL, inode, block, 0, &err);
2479                 if (!bh) {
2480                         CERROR("can't read block: %d\n", err);
2481                         return err;
2482                 }
2483
2484                 memcpy(buf, bh->b_data + boffs, csize);
2485                 brelse(bh);
2486
2487                 *offs += csize;
2488                 buf += csize;
2489                 size -= csize;
2490         }
2491         return osize;
2492 }
2493
2494 static ssize_t osd_read(const struct lu_env *env, struct dt_object *dt,
2495                         struct lu_buf *buf, loff_t *pos,
2496                         struct lustre_capa *capa)
2497 {
2498         struct osd_object      *obj    = osd_dt_obj(dt);
2499         struct inode           *inode  = obj->oo_inode;
2500         int rc;
2501
2502         if (osd_object_auth(env, dt, capa, CAPA_OPC_BODY_READ))
2503                 RETURN(-EACCES);
2504
2505         /* Read small symlink from inode body as we need to maintain correct
2506          * on-disk symlinks for ldiskfs.
2507          */
2508         if (S_ISLNK(obj->oo_dt.do_lu.lo_header->loh_attr) &&
2509             (buf->lb_len <= sizeof (LDISKFS_I(inode)->i_data)))
2510                 rc = osd_ldiskfs_readlink(inode, buf->lb_buf, buf->lb_len);
2511         else
2512                 rc = osd_ldiskfs_read(inode, buf->lb_buf, buf->lb_len, pos);
2513
2514         return rc;
2515 }
2516
2517 static int osd_ldiskfs_writelink(struct inode *inode, char *buffer, int buflen)
2518 {
2519
2520         memcpy((char*)&LDISKFS_I(inode)->i_data, (char *)buffer,
2521                buflen);
2522         LDISKFS_I(inode)->i_disksize = buflen;
2523         i_size_write(inode, buflen);
2524         inode->i_sb->s_op->dirty_inode(inode);
2525
2526         return 0;
2527 }
2528
2529 static int osd_ldiskfs_write_record(struct inode *inode, void *buf, int bufsize,
2530                                     loff_t *offs, handle_t *handle)
2531 {
2532         struct buffer_head *bh = NULL;
2533         loff_t offset = *offs;
2534         loff_t new_size = i_size_read(inode);
2535         unsigned long block;
2536         int blocksize = 1 << inode->i_blkbits;
2537         int err = 0;
2538         int size;
2539         int boffs;
2540         int dirty_inode = 0;
2541
2542         while (bufsize > 0) {
2543                 if (bh != NULL)
2544                         brelse(bh);
2545
2546                 block = offset >> inode->i_blkbits;
2547                 boffs = offset & (blocksize - 1);
2548                 size = min(blocksize - boffs, bufsize);
2549                 bh = ldiskfs_bread(handle, inode, block, 1, &err);
2550                 if (!bh) {
2551                         CERROR("can't read/create block: %d\n", err);
2552                         break;
2553                 }
2554
2555                 err = ldiskfs_journal_get_write_access(handle, bh);
2556                 if (err) {
2557                         CERROR("journal_get_write_access() returned error %d\n",
2558                                err);
2559                         break;
2560                 }
2561                 LASSERTF(boffs + size <= bh->b_size,
2562                          "boffs %d size %d bh->b_size %lu",
2563                          boffs, size, (unsigned long)bh->b_size);
2564                 memcpy(bh->b_data + boffs, buf, size);
2565                 err = ldiskfs_journal_dirty_metadata(handle, bh);
2566                 if (err)
2567                         break;
2568
2569                 if (offset + size > new_size)
2570                         new_size = offset + size;
2571                 offset += size;
2572                 bufsize -= size;
2573                 buf += size;
2574         }
2575         if (bh)
2576                 brelse(bh);
2577
2578         /* correct in-core and on-disk sizes */
2579         if (new_size > i_size_read(inode)) {
2580                 spin_lock(&inode->i_lock);
2581                 if (new_size > i_size_read(inode))
2582                         i_size_write(inode, new_size);
2583                 if (i_size_read(inode) > LDISKFS_I(inode)->i_disksize) {
2584                         LDISKFS_I(inode)->i_disksize = i_size_read(inode);
2585                         dirty_inode = 1;
2586                 }
2587                 spin_unlock(&inode->i_lock);
2588                 if (dirty_inode)
2589                         inode->i_sb->s_op->dirty_inode(inode);
2590         }
2591
2592         if (err == 0)
2593                 *offs = offset;
2594         return err;
2595 }
2596
2597 static ssize_t osd_write(const struct lu_env *env, struct dt_object *dt,
2598                          const struct lu_buf *buf, loff_t *pos,
2599                          struct thandle *handle, struct lustre_capa *capa,
2600                          int ignore_quota)
2601 {
2602         struct osd_object  *obj   = osd_dt_obj(dt);
2603         struct inode       *inode = obj->oo_inode;
2604         struct osd_thandle *oh;
2605         ssize_t            result = 0;
2606 #ifdef HAVE_QUOTA_SUPPORT
2607         cfs_cap_t           save = current->cap_effective;
2608 #endif
2609
2610         LASSERT(handle != NULL);
2611
2612         if (osd_object_auth(env, dt, capa, CAPA_OPC_BODY_WRITE))
2613                 RETURN(-EACCES);
2614
2615         oh = container_of(handle, struct osd_thandle, ot_super);
2616         LASSERT(oh->ot_handle->h_transaction != NULL);
2617 #ifdef HAVE_QUOTA_SUPPORT
2618         if (ignore_quota)
2619                 current->cap_effective |= CFS_CAP_SYS_RESOURCE_MASK;
2620         else
2621                 current->cap_effective &= ~CFS_CAP_SYS_RESOURCE_MASK;
2622 #endif
2623         /* Write small symlink to inode body as we need to maintain correct
2624          * on-disk symlinks for ldiskfs.
2625          */
2626         if(S_ISLNK(obj->oo_dt.do_lu.lo_header->loh_attr) &&
2627            (buf->lb_len < sizeof (LDISKFS_I(inode)->i_data)))
2628                 result = osd_ldiskfs_writelink(inode, buf->lb_buf, buf->lb_len);
2629         else
2630                 result = osd_ldiskfs_write_record(inode, buf->lb_buf,
2631                                                   buf->lb_len, pos,
2632                                                   oh->ot_handle);
2633 #ifdef HAVE_QUOTA_SUPPORT
2634         current->cap_effective = save;
2635 #endif
2636         if (result == 0)
2637                 result = buf->lb_len;
2638         return result;
2639 }
2640
2641 static const struct dt_body_operations osd_body_ops = {
2642         .dbo_read  = osd_read,
2643         .dbo_write = osd_write
2644 };
2645
2646
2647 /**
2648  *      delete a (key, value) pair from index \a dt specified by \a key
2649  *
2650  *      \param  dt      osd index object
2651  *      \param  key     key for index
2652  *      \param  rec     record reference
2653  *      \param  handle  transaction handler
2654  *
2655  *      \retval  0  success
2656  *      \retval -ve   failure
2657  */
2658
2659 static int osd_index_iam_delete(const struct lu_env *env, struct dt_object *dt,
2660                                 const struct dt_key *key, struct thandle *handle,
2661                                 struct lustre_capa *capa)
2662 {
2663         struct osd_object     *obj = osd_dt_obj(dt);
2664         struct osd_thandle    *oh;
2665         struct iam_path_descr *ipd;
2666         struct iam_container  *bag = &obj->oo_dir->od_container;
2667         int rc;
2668
2669         ENTRY;
2670
2671         LINVRNT(osd_invariant(obj));
2672         LASSERT(dt_object_exists(dt));
2673         LASSERT(bag->ic_object == obj->oo_inode);
2674         LASSERT(handle != NULL);
2675
2676         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_DELETE))
2677                 RETURN(-EACCES);
2678
2679         ipd = osd_idx_ipd_get(env, bag);
2680         if (unlikely(ipd == NULL))
2681                 RETURN(-ENOMEM);
2682
2683         oh = container_of0(handle, struct osd_thandle, ot_super);
2684         LASSERT(oh->ot_handle != NULL);
2685         LASSERT(oh->ot_handle->h_transaction != NULL);
2686
2687         rc = iam_delete(oh->ot_handle, bag, (const struct iam_key *)key, ipd);
2688         osd_ipd_put(env, bag, ipd);
2689         LINVRNT(osd_invariant(obj));
2690         RETURN(rc);
2691 }
2692
2693 static inline int osd_get_fid_from_dentry(struct ldiskfs_dir_entry_2 *de,
2694                                           struct dt_rec *fid)
2695 {
2696         struct osd_fid_pack *rec;
2697         int rc = -ENODATA;
2698
2699         if (de->file_type & LDISKFS_DIRENT_LUFID) {
2700                 rec = (struct osd_fid_pack *) (de->name + de->name_len + 1);
2701                 rc = osd_fid_unpack((struct lu_fid *)fid, rec);
2702         }
2703         RETURN(rc);
2704 }
2705
2706 /**
2707  * Index delete function for interoperability mode (b11826).
2708  * It will remove the directory entry added by osd_index_ea_insert().
2709  * This entry is needed to maintain name->fid mapping.
2710  *
2711  * \param key,  key i.e. file entry to be deleted
2712  *
2713  * \retval   0, on success
2714  * \retval -ve, on error
2715  */
2716 static int osd_index_ea_delete(const struct lu_env *env, struct dt_object *dt,
2717                                const struct dt_key *key, struct thandle *handle,
2718                                struct lustre_capa *capa)
2719 {
2720         struct osd_object          *obj    = osd_dt_obj(dt);
2721         struct inode               *dir    = obj->oo_inode;
2722         struct dentry              *dentry;
2723         struct osd_thandle         *oh;
2724         struct ldiskfs_dir_entry_2 *de;
2725         struct buffer_head         *bh;
2726
2727         int rc;
2728
2729         ENTRY;
2730
2731         LINVRNT(osd_invariant(obj));
2732         LASSERT(dt_object_exists(dt));
2733         LASSERT(handle != NULL);
2734
2735         oh = container_of(handle, struct osd_thandle, ot_super);
2736         LASSERT(oh->ot_handle != NULL);
2737         LASSERT(oh->ot_handle->h_transaction != NULL);
2738
2739         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_DELETE))
2740                 RETURN(-EACCES);
2741
2742         dentry = osd_child_dentry_get(env, obj,
2743                                       (char *)key, strlen((char *)key));
2744
2745         cfs_down_write(&obj->oo_ext_idx_sem);
2746         bh = ll_ldiskfs_find_entry(dir, dentry, &de);
2747         if (bh) {
2748                 struct osd_thread_info *oti = osd_oti_get(env);
2749                 struct timespec *ctime = &oti->oti_time;
2750                 struct timespec *mtime = &oti->oti_time2;
2751
2752                 *ctime = dir->i_ctime;
2753                 *mtime = dir->i_mtime;
2754                 rc = ldiskfs_delete_entry(oh->ot_handle,
2755                                 dir, de, bh);
2756                 /* xtime should not be updated with server-side time. */
2757                 cfs_spin_lock(&obj->oo_guard);
2758                 dir->i_ctime = *ctime;
2759                 dir->i_mtime = *mtime;
2760                 cfs_spin_unlock(&obj->oo_guard);
2761                 mark_inode_dirty(dir);
2762                 brelse(bh);
2763         } else
2764                 rc = -ENOENT;
2765
2766         cfs_up_write(&obj->oo_ext_idx_sem);
2767         LASSERT(osd_invariant(obj));
2768         RETURN(rc);
2769 }
2770
2771 /**
2772  *      Lookup index for \a key and copy record to \a rec.
2773  *
2774  *      \param  dt      osd index object
2775  *      \param  key     key for index
2776  *      \param  rec     record reference
2777  *
2778  *      \retval  +ve  success : exact mach
2779  *      \retval  0    return record with key not greater than \a key
2780  *      \retval -ve   failure
2781  */
2782 static int osd_index_iam_lookup(const struct lu_env *env, struct dt_object *dt,
2783                                 struct dt_rec *rec, const struct dt_key *key,
2784                                 struct lustre_capa *capa)
2785 {
2786         struct osd_object     *obj = osd_dt_obj(dt);
2787         struct iam_path_descr *ipd;
2788         struct iam_container  *bag = &obj->oo_dir->od_container;
2789         struct osd_thread_info *oti = osd_oti_get(env);
2790         struct iam_iterator    *it = &oti->oti_idx_it;
2791         struct iam_rec *iam_rec;
2792         int rc;
2793         ENTRY;
2794
2795         LASSERT(osd_invariant(obj));
2796         LASSERT(dt_object_exists(dt));
2797         LASSERT(bag->ic_object == obj->oo_inode);
2798
2799         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_LOOKUP))
2800                 RETURN(-EACCES);
2801
2802         ipd = osd_idx_ipd_get(env, bag);
2803         if (IS_ERR(ipd))
2804                 RETURN(-ENOMEM);
2805
2806         /* got ipd now we can start iterator. */
2807         iam_it_init(it, bag, 0, ipd);
2808
2809         rc = iam_it_get(it, (struct iam_key *)key);
2810         if (rc >= 0) {
2811                 if (S_ISDIR(obj->oo_inode->i_mode))
2812                         iam_rec = (struct iam_rec *)oti->oti_ldp;
2813                 else
2814                         iam_rec = (struct iam_rec *) rec;
2815
2816                 iam_reccpy(&it->ii_path.ip_leaf, (struct iam_rec *)iam_rec);
2817                 if (S_ISDIR(obj->oo_inode->i_mode))
2818                         osd_fid_unpack((struct lu_fid *) rec,
2819                                        (struct osd_fid_pack *)iam_rec);
2820         }
2821         iam_it_put(it);
2822         iam_it_fini(it);
2823         osd_ipd_put(env, bag, ipd);
2824
2825         LINVRNT(osd_invariant(obj));
2826
2827         RETURN(rc);
2828 }
2829
2830 /**
2831  *      Inserts (key, value) pair in \a dt index object.
2832  *
2833  *      \param  dt      osd index object
2834  *      \param  key     key for index
2835  *      \param  rec     record reference
2836  *      \param  th      transaction handler
2837  *
2838  *      \retval  0  success
2839  *      \retval -ve failure
2840  */
2841 static int osd_index_iam_insert(const struct lu_env *env, struct dt_object *dt,
2842                                 const struct dt_rec *rec, const struct dt_key *key,
2843                                 struct thandle *th, struct lustre_capa *capa,
2844                                 int ignore_quota)
2845 {
2846         struct osd_object     *obj = osd_dt_obj(dt);
2847         struct iam_path_descr *ipd;
2848         struct osd_thandle    *oh;
2849         struct iam_container  *bag = &obj->oo_dir->od_container;
2850 #ifdef HAVE_QUOTA_SUPPORT
2851         cfs_cap_t              save = current->cap_effective;
2852 #endif
2853         struct osd_thread_info *oti = osd_oti_get(env);
2854         struct iam_rec *iam_rec = (struct iam_rec *)oti->oti_ldp;
2855         int rc;
2856
2857         ENTRY;
2858
2859         LINVRNT(osd_invariant(obj));
2860         LASSERT(dt_object_exists(dt));
2861         LASSERT(bag->ic_object == obj->oo_inode);
2862         LASSERT(th != NULL);
2863
2864         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_INSERT))
2865                 return -EACCES;
2866
2867         ipd = osd_idx_ipd_get(env, bag);
2868         if (unlikely(ipd == NULL))
2869                 RETURN(-ENOMEM);
2870
2871         oh = container_of0(th, struct osd_thandle, ot_super);
2872         LASSERT(oh->ot_handle != NULL);
2873         LASSERT(oh->ot_handle->h_transaction != NULL);
2874 #ifdef HAVE_QUOTA_SUPPORT
2875         if (ignore_quota)
2876                 current->cap_effective |= CFS_CAP_SYS_RESOURCE_MASK;
2877         else
2878                 current->cap_effective &= ~CFS_CAP_SYS_RESOURCE_MASK;
2879 #endif
2880         if (S_ISDIR(obj->oo_inode->i_mode))
2881                 osd_fid_pack((struct osd_fid_pack *)iam_rec, rec, &oti->oti_fid);
2882         else
2883                 iam_rec = (struct iam_rec *) rec;
2884         rc = iam_insert(oh->ot_handle, bag, (const struct iam_key *)key,
2885                         iam_rec, ipd);
2886 #ifdef HAVE_QUOTA_SUPPORT
2887         current->cap_effective = save;
2888 #endif
2889         osd_ipd_put(env, bag, ipd);
2890         LINVRNT(osd_invariant(obj));
2891         RETURN(rc);
2892 }
2893
2894 /**
2895  * Calls ldiskfs_add_entry() to add directory entry
2896  * into the directory. This is required for
2897  * interoperability mode (b11826)
2898  *
2899  * \retval   0, on success
2900  * \retval -ve, on error
2901  */
2902 static int __osd_ea_add_rec(struct osd_thread_info *info,
2903                             struct osd_object *pobj,
2904                             struct inode  *cinode,
2905                             const char *name,
2906                             const struct dt_rec *fid,
2907                             struct thandle *th)
2908 {
2909         struct ldiskfs_dentry_param *ldp;
2910         struct dentry      *child;
2911         struct osd_thandle *oth;
2912         int rc;
2913
2914         oth = container_of(th, struct osd_thandle, ot_super);
2915         LASSERT(oth->ot_handle != NULL);
2916         LASSERT(oth->ot_handle->h_transaction != NULL);
2917
2918         child = osd_child_dentry_get(info->oti_env, pobj, name, strlen(name));
2919
2920         if (fid_is_igif((struct lu_fid *)fid) ||
2921             fid_seq((struct lu_fid *)fid) >= FID_SEQ_NORMAL) {
2922                 ldp = (struct ldiskfs_dentry_param *)info->oti_ldp;
2923                 osd_get_ldiskfs_dirent_param(ldp, fid);
2924                 child->d_fsdata = (void*) ldp;
2925         } else
2926                 child->d_fsdata = NULL;
2927         rc = ldiskfs_add_entry(oth->ot_handle, child, cinode);
2928
2929         RETURN(rc);
2930 }
2931
2932 /**
2933  * Calls ldiskfs_add_dot_dotdot() to add dot and dotdot entries
2934  * into the directory.Also sets flags into osd object to
2935  * indicate dot and dotdot are created. This is required for
2936  * interoperability mode (b11826)
2937  *
2938  * \param dir   directory for dot and dotdot fixup.
2939  * \param obj   child object for linking
2940  *
2941  * \retval   0, on success
2942  * \retval -ve, on error
2943  */
2944 static int osd_add_dot_dotdot(struct osd_thread_info *info,
2945                               struct osd_object *dir,
2946                               struct inode  *parent_dir, const char *name,
2947                               const struct dt_rec *dot_fid,
2948                               const struct dt_rec *dot_dot_fid,
2949                               struct thandle *th)
2950 {
2951         struct inode            *inode  = dir->oo_inode;
2952         struct ldiskfs_dentry_param *dot_ldp;
2953         struct ldiskfs_dentry_param *dot_dot_ldp;
2954         struct osd_thandle      *oth;
2955         int result = 0;
2956
2957         oth = container_of(th, struct osd_thandle, ot_super);
2958         LASSERT(oth->ot_handle->h_transaction != NULL);
2959         LASSERT(S_ISDIR(dir->oo_inode->i_mode));
2960
2961         if (strcmp(name, dot) == 0) {
2962                 if (dir->oo_compat_dot_created) {
2963                         result = -EEXIST;
2964                 } else {
2965                         LASSERT(inode == parent_dir);
2966                         dir->oo_compat_dot_created = 1;
2967                         result = 0;
2968                 }
2969         } else if(strcmp(name, dotdot) == 0) {
2970                 dot_ldp = (struct ldiskfs_dentry_param *)info->oti_ldp;
2971                 dot_dot_ldp = (struct ldiskfs_dentry_param *)info->oti_ldp2;
2972
2973                 if (!dir->oo_compat_dot_created)
2974                         return -EINVAL;
2975                 if (fid_seq((struct lu_fid *)dot_fid) >= FID_SEQ_NORMAL) {
2976                         osd_get_ldiskfs_dirent_param(dot_ldp, dot_fid);
2977                         osd_get_ldiskfs_dirent_param(dot_dot_ldp, dot_dot_fid);
2978                 } else {
2979                         dot_ldp = NULL;
2980                         dot_dot_ldp = NULL;
2981                 }
2982                 /* in case of rename, dotdot is already created */
2983                 if (dir->oo_compat_dotdot_created) {
2984                         return __osd_ea_add_rec(info, dir, parent_dir, name,
2985                                                 dot_dot_fid, th);
2986                 }
2987
2988                 result = ldiskfs_add_dot_dotdot(oth->ot_handle, parent_dir, inode,
2989                                                 dot_ldp, dot_dot_ldp);
2990                 if (result == 0)
2991                        dir->oo_compat_dotdot_created = 1;
2992         }
2993
2994         return result;
2995 }
2996
2997
2998 /**
2999  * It will call the appropriate osd_add* function and return the
3000  * value, return by respective functions.
3001  */
3002 static int osd_ea_add_rec(const struct lu_env *env,
3003                           struct osd_object *pobj,
3004                           struct inode *cinode,
3005                           const char *name,
3006                           const struct dt_rec *fid,
3007                           struct thandle *th)
3008 {
3009         struct osd_thread_info    *info   = osd_oti_get(env);
3010         int rc;
3011
3012         if (name[0] == '.' && (name[1] == '\0' || (name[1] == '.' &&
3013                                                    name[2] =='\0')))
3014                 rc = osd_add_dot_dotdot(info, pobj, cinode, name,
3015                      (struct dt_rec *)lu_object_fid(&pobj->oo_dt.do_lu),
3016                                         fid, th);
3017         else
3018                 rc = __osd_ea_add_rec(info, pobj, cinode, name, fid, th);
3019
3020         return rc;
3021 }
3022
3023 /**
3024  * Calls ->lookup() to find dentry. From dentry get inode and
3025  * read inode's ea to get fid. This is required for  interoperability
3026  * mode (b11826)
3027  *
3028  * \retval   0, on success
3029  * \retval -ve, on error
3030  */
3031 static int osd_ea_lookup_rec(const struct lu_env *env, struct osd_object *obj,
3032                              struct dt_rec *rec, const struct dt_key *key)
3033 {
3034         struct inode               *dir    = obj->oo_inode;
3035         struct dentry              *dentry;
3036         struct ldiskfs_dir_entry_2 *de;
3037         struct buffer_head         *bh;
3038         struct lu_fid              *fid = (struct lu_fid *) rec;
3039         int ino;
3040         int rc;
3041
3042         LASSERT(dir->i_op != NULL && dir->i_op->lookup != NULL);
3043
3044         dentry = osd_child_dentry_get(env, obj,
3045                                       (char *)key, strlen((char *)key));
3046
3047         cfs_down_read(&obj->oo_ext_idx_sem);
3048         bh = ll_ldiskfs_find_entry(dir, dentry, &de);
3049         if (bh) {
3050                 ino = le32_to_cpu(de->inode);
3051                 rc = osd_get_fid_from_dentry(de, rec);
3052
3053                 /* done with de, release bh */
3054                 brelse(bh);
3055                 if (rc != 0)
3056                         rc = osd_ea_fid_get(env, obj, ino, fid);
3057         } else
3058                 rc = -ENOENT;
3059
3060         cfs_up_read(&obj->oo_ext_idx_sem);
3061         RETURN (rc);
3062 }
3063
3064 /**
3065  * Find the osd object for given fid.
3066  *
3067  * \param fid need to find the osd object having this fid
3068  *
3069  * \retval osd_object on success
3070  * \retval        -ve on error
3071  */
3072 struct osd_object *osd_object_find(const struct lu_env *env,
3073                                    struct dt_object *dt,
3074                                    const struct lu_fid *fid)
3075 {
3076         struct lu_device         *ludev = dt->do_lu.lo_dev;
3077         struct osd_object        *child = NULL;
3078         struct lu_object         *luch;
3079         struct lu_object         *lo;
3080
3081         luch = lu_object_find(env, ludev, fid, NULL);
3082         if (!IS_ERR(luch)) {
3083                 if (lu_object_exists(luch)) {
3084                         lo = lu_object_locate(luch->lo_header, ludev->ld_type);
3085                         if (lo != NULL)
3086                                 child = osd_obj(lo);
3087                         else
3088                                 LU_OBJECT_DEBUG(D_ERROR, env, luch,
3089                                                 "lu_object can't be located"
3090                                                 ""DFID"\n", PFID(fid));
3091
3092                         if (child == NULL) {
3093                                 lu_object_put(env, luch);
3094                                 CERROR("Unable to get osd_object\n");
3095                                 child = ERR_PTR(-ENOENT);
3096                         }
3097                 } else {
3098                         LU_OBJECT_DEBUG(D_ERROR, env, luch,
3099                                         "lu_object does not exists "DFID"\n",
3100                                         PFID(fid));
3101                         child = ERR_PTR(-ENOENT);
3102                 }
3103         } else
3104                 child = (void *)luch;
3105
3106         return child;
3107 }
3108
3109 /**
3110  * Put the osd object once done with it.
3111  *
3112  * \param obj osd object that needs to be put
3113  */
3114 static inline void osd_object_put(const struct lu_env *env,
3115                                   struct osd_object *obj)
3116 {
3117         lu_object_put(env, &obj->oo_dt.do_lu);
3118 }
3119
3120 /**
3121  * Index add function for interoperability mode (b11826).
3122  * It will add the directory entry.This entry is needed to
3123  * maintain name->fid mapping.
3124  *
3125  * \param key it is key i.e. file entry to be inserted
3126  * \param rec it is value of given key i.e. fid
3127  *
3128  * \retval   0, on success
3129  * \retval -ve, on error
3130  */
3131 static int osd_index_ea_insert(const struct lu_env *env, struct dt_object *dt,
3132                                const struct dt_rec *rec,
3133                                const struct dt_key *key, struct thandle *th,
3134                                struct lustre_capa *capa, int ignore_quota)
3135 {
3136         struct osd_object        *obj   = osd_dt_obj(dt);
3137         struct lu_fid            *fid   = (struct lu_fid *) rec;
3138         const char               *name  = (const char *)key;
3139         struct osd_object        *child;
3140 #ifdef HAVE_QUOTA_SUPPORT
3141         cfs_cap_t                 save  = current->cap_effective;
3142 #endif
3143         int rc;
3144
3145         ENTRY;
3146
3147         LASSERT(osd_invariant(obj));
3148         LASSERT(dt_object_exists(dt));
3149         LASSERT(th != NULL);
3150
3151         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_INSERT))
3152                 RETURN(-EACCES);
3153
3154         child = osd_object_find(env, dt, fid);
3155         if (!IS_ERR(child)) {
3156                 struct inode *inode = obj->oo_inode;
3157                 struct osd_thread_info *oti = osd_oti_get(env);
3158                 struct timespec *ctime = &oti->oti_time;
3159                 struct timespec *mtime = &oti->oti_time2;
3160
3161                 *ctime = inode->i_ctime;
3162                 *mtime = inode->i_mtime;
3163 #ifdef HAVE_QUOTA_SUPPORT
3164                 if (ignore_quota)
3165                         current->cap_effective |= CFS_CAP_SYS_RESOURCE_MASK;
3166                 else
3167                         current->cap_effective &= ~CFS_CAP_SYS_RESOURCE_MASK;
3168 #endif
3169                 cfs_down_write(&obj->oo_ext_idx_sem);
3170                 rc = osd_ea_add_rec(env, obj, child->oo_inode, name, rec, th);
3171                 cfs_up_write(&obj->oo_ext_idx_sem);
3172 #ifdef HAVE_QUOTA_SUPPORT
3173                 current->cap_effective = save;
3174 #endif
3175                 osd_object_put(env, child);
3176                 /* xtime should not be updated with server-side time. */
3177                 cfs_spin_lock(&obj->oo_guard);
3178                 inode->i_ctime = *ctime;
3179                 inode->i_mtime = *mtime;
3180                 cfs_spin_unlock(&obj->oo_guard);
3181                 mark_inode_dirty(inode);
3182         } else {
3183                 rc = PTR_ERR(child);
3184         }
3185
3186         LASSERT(osd_invariant(obj));
3187         RETURN(rc);
3188 }
3189
3190 /**
3191  *  Initialize osd Iterator for given osd index object.
3192  *
3193  *  \param  dt      osd index object
3194  */
3195
3196 static struct dt_it *osd_it_iam_init(const struct lu_env *env,
3197                                  struct dt_object *dt,
3198                                  struct lustre_capa *capa)
3199 {
3200         struct osd_it_iam         *it;
3201         struct osd_thread_info *oti = osd_oti_get(env);
3202         struct osd_object     *obj = osd_dt_obj(dt);
3203         struct lu_object      *lo  = &dt->do_lu;
3204         struct iam_path_descr *ipd;
3205         struct iam_container  *bag = &obj->oo_dir->od_container;
3206
3207         LASSERT(lu_object_exists(lo));
3208
3209         if (osd_object_auth(env, dt, capa, CAPA_OPC_BODY_READ))
3210                 return ERR_PTR(-EACCES);
3211
3212         it = &oti->oti_it;
3213         ipd = osd_it_ipd_get(env, bag);
3214         if (likely(ipd != NULL)) {
3215                 it->oi_obj = obj;
3216                 it->oi_ipd = ipd;
3217                 lu_object_get(lo);
3218                 iam_it_init(&it->oi_it, bag, IAM_IT_MOVE, ipd);
3219                 return (struct dt_it *)it;
3220         }
3221         return ERR_PTR(-ENOMEM);
3222 }
3223
3224 /**
3225  * free given Iterator.
3226  */
3227
3228 static void osd_it_iam_fini(const struct lu_env *env, struct dt_it *di)
3229 {
3230         struct osd_it_iam     *it = (struct osd_it_iam *)di;
3231         struct osd_object *obj = it->oi_obj;
3232
3233         iam_it_fini(&it->oi_it);
3234         osd_ipd_put(env, &obj->oo_dir->od_container, it->oi_ipd);
3235         lu_object_put(env, &obj->oo_dt.do_lu);
3236 }
3237
3238 /**
3239  *  Move Iterator to record specified by \a key
3240  *
3241  *  \param  di      osd iterator
3242  *  \param  key     key for index
3243  *
3244  *  \retval +ve  di points to record with least key not larger than key
3245  *  \retval  0   di points to exact matched key
3246  *  \retval -ve  failure
3247  */
3248
3249 static int osd_it_iam_get(const struct lu_env *env,
3250                       struct dt_it *di, const struct dt_key *key)
3251 {
3252         struct osd_it_iam *it = (struct osd_it_iam *)di;
3253
3254         return iam_it_get(&it->oi_it, (const struct iam_key *)key);
3255 }
3256
3257 /**
3258  *  Release Iterator
3259  *
3260  *  \param  di      osd iterator
3261  */
3262
3263 static void osd_it_iam_put(const struct lu_env *env, struct dt_it *di)
3264 {
3265         struct osd_it_iam *it = (struct osd_it_iam *)di;
3266
3267         iam_it_put(&it->oi_it);
3268 }
3269
3270 /**
3271  *  Move iterator by one record
3272  *
3273  *  \param  di      osd iterator
3274  *
3275  *  \retval +1   end of container reached
3276  *  \retval  0   success
3277  *  \retval -ve  failure
3278  */
3279
3280 static int osd_it_iam_next(const struct lu_env *env, struct dt_it *di)
3281 {
3282         struct osd_it_iam *it = (struct osd_it_iam *)di;
3283
3284         return iam_it_next(&it->oi_it);
3285 }
3286
3287 /**
3288  * Return pointer to the key under iterator.
3289  */
3290
3291 static struct dt_key *osd_it_iam_key(const struct lu_env *env,
3292                                  const struct dt_it *di)
3293 {