Whamcloud - gitweb
LU-2886 obdclass: remove obsoleted md_local_file.c
[fs/lustre-release.git] / lustre / osd-ldiskfs / osd_handler.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 2013, Intel Corporation.
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 #define DEBUG_SUBSYSTEM S_MDS
45
46 #include <linux/module.h>
47
48 /* LUSTRE_VERSION_CODE */
49 #include <lustre_ver.h>
50 /* prerequisite for linux/xattr.h */
51 #include <linux/types.h>
52 /* prerequisite for linux/xattr.h */
53 #include <linux/fs.h>
54 /* XATTR_{REPLACE,CREATE} */
55 #include <linux/xattr.h>
56
57 /*
58  * struct OBD_{ALLOC,FREE}*()
59  * OBD_FAIL_CHECK
60  */
61 #include <obd_support.h>
62 /* struct ptlrpc_thread */
63 #include <lustre_net.h>
64 #include <lustre_fid.h>
65
66 #include "osd_internal.h"
67
68 /* llo_* api support */
69 #include <md_object.h>
70 #include <lustre_quota.h>
71
72 int ldiskfs_pdo = 1;
73 CFS_MODULE_PARM(ldiskfs_pdo, "i", int, 0644,
74                 "ldiskfs with parallel directory operations");
75
76 int ldiskfs_track_declares_assert;
77 CFS_MODULE_PARM(ldiskfs_track_declares_assert, "i", int, 0644,
78                 "LBUG during tracking of declares");
79
80 static const char dot[] = ".";
81 static const char dotdot[] = "..";
82 static const char remote_obj_dir[] = "REM_OBJ_DIR";
83
84 static const struct lu_object_operations      osd_lu_obj_ops;
85 static const struct dt_object_operations      osd_obj_ops;
86 static const struct dt_object_operations      osd_obj_ea_ops;
87 static const struct dt_object_operations      osd_obj_otable_it_ops;
88 static const struct dt_index_operations       osd_index_iam_ops;
89 static const struct dt_index_operations       osd_index_ea_ops;
90
91 int osd_trans_declare_op2rb[] = {
92         [OSD_OT_ATTR_SET]       = OSD_OT_ATTR_SET,
93         [OSD_OT_PUNCH]          = OSD_OT_MAX,
94         [OSD_OT_XATTR_SET]      = OSD_OT_XATTR_SET,
95         [OSD_OT_CREATE]         = OSD_OT_DESTROY,
96         [OSD_OT_DESTROY]        = OSD_OT_CREATE,
97         [OSD_OT_REF_ADD]        = OSD_OT_REF_DEL,
98         [OSD_OT_REF_DEL]        = OSD_OT_REF_ADD,
99         [OSD_OT_WRITE]          = OSD_OT_WRITE,
100         [OSD_OT_INSERT]         = OSD_OT_DELETE,
101         [OSD_OT_DELETE]         = OSD_OT_INSERT,
102         [OSD_OT_QUOTA]          = OSD_OT_MAX,
103 };
104
105 static int osd_has_index(const struct osd_object *obj)
106 {
107         return obj->oo_dt.do_index_ops != NULL;
108 }
109
110 static int osd_object_invariant(const struct lu_object *l)
111 {
112         return osd_invariant(osd_obj(l));
113 }
114
115 /*
116  * Concurrency: doesn't matter
117  */
118 static int osd_read_locked(const struct lu_env *env, struct osd_object *o)
119 {
120         return osd_oti_get(env)->oti_r_locks > 0;
121 }
122
123 /*
124  * Concurrency: doesn't matter
125  */
126 static int osd_write_locked(const struct lu_env *env, struct osd_object *o)
127 {
128         struct osd_thread_info *oti = osd_oti_get(env);
129         return oti->oti_w_locks > 0 && o->oo_owner == env;
130 }
131
132 /*
133  * Concurrency: doesn't access mutable data
134  */
135 static int osd_root_get(const struct lu_env *env,
136                         struct dt_device *dev, struct lu_fid *f)
137 {
138         lu_local_obj_fid(f, OSD_FS_ROOT_OID);
139         return 0;
140 }
141
142 /*
143  * OSD object methods.
144  */
145
146 /*
147  * Concurrency: no concurrent access is possible that early in object
148  * life-cycle.
149  */
150 static struct lu_object *osd_object_alloc(const struct lu_env *env,
151                                           const struct lu_object_header *hdr,
152                                           struct lu_device *d)
153 {
154         struct osd_object *mo;
155
156         OBD_ALLOC_PTR(mo);
157         if (mo != NULL) {
158                 struct lu_object *l;
159
160                 l = &mo->oo_dt.do_lu;
161                 dt_object_init(&mo->oo_dt, NULL, d);
162                 mo->oo_dt.do_ops = &osd_obj_ea_ops;
163                 l->lo_ops = &osd_lu_obj_ops;
164                 init_rwsem(&mo->oo_sem);
165                 init_rwsem(&mo->oo_ext_idx_sem);
166                 spin_lock_init(&mo->oo_guard);
167                 return l;
168         } else {
169                 return NULL;
170         }
171 }
172
173 static inline int __osd_xattr_get(struct inode *inode, struct dentry *dentry,
174                                   const char *name, void *buf, int len)
175 {
176         dentry->d_inode = inode;
177         dentry->d_sb = inode->i_sb;
178         return inode->i_op->getxattr(dentry, name, buf, len);
179 }
180
181 int osd_get_lma(struct osd_thread_info *info, struct inode *inode,
182                 struct dentry *dentry, struct lustre_mdt_attrs *lma)
183 {
184         int rc;
185
186         CLASSERT(LMA_OLD_SIZE >= sizeof(*lma));
187         rc = __osd_xattr_get(inode, dentry, XATTR_NAME_LMA,
188                              info->oti_mdt_attrs_old, LMA_OLD_SIZE);
189         if (rc > 0) {
190                 if ((void *)lma != (void *)info->oti_mdt_attrs_old)
191                         memcpy(lma, info->oti_mdt_attrs_old, sizeof(*lma));
192                 rc = 0;
193                 lustre_lma_swab(lma);
194                 /* Check LMA compatibility */
195                 if (lma->lma_incompat & ~LMA_INCOMPAT_SUPP) {
196                         CWARN("%.16s: unsupported incompat LMA feature(s) %#x "
197                               "for fid = "DFID", ino = %lu\n",
198                               LDISKFS_SB(inode->i_sb)->s_es->s_volume_name,
199                               lma->lma_incompat & ~LMA_INCOMPAT_SUPP,
200                               PFID(&lma->lma_self_fid), inode->i_ino);
201                         rc = -EOPNOTSUPP;
202                 }
203         } else if (rc == 0) {
204                 rc = -ENODATA;
205         }
206
207         return rc;
208 }
209
210 /*
211  * retrieve object from backend ext fs.
212  **/
213 struct inode *osd_iget(struct osd_thread_info *info, struct osd_device *dev,
214                        struct osd_inode_id *id)
215 {
216         struct inode *inode = NULL;
217
218         inode = ldiskfs_iget(osd_sb(dev), id->oii_ino);
219         if (IS_ERR(inode)) {
220                 CDEBUG(D_INODE, "no inode: ino = %u, rc = %ld\n",
221                        id->oii_ino, PTR_ERR(inode));
222         } else if (id->oii_gen != OSD_OII_NOGEN &&
223                    inode->i_generation != id->oii_gen) {
224                 CDEBUG(D_INODE, "unmatched inode: ino = %u, gen0 = %u, "
225                        "gen1 = %u\n",
226                        id->oii_ino, id->oii_gen, inode->i_generation);
227                 iput(inode);
228                 inode = ERR_PTR(-ESTALE);
229         } else if (inode->i_nlink == 0) {
230                 /* due to parallel readdir and unlink,
231                 * we can have dead inode here. */
232                 CDEBUG(D_INODE, "stale inode: ino = %u\n", id->oii_ino);
233                 make_bad_inode(inode);
234                 iput(inode);
235                 inode = ERR_PTR(-ESTALE);
236         } else if (is_bad_inode(inode)) {
237                 CWARN("%.16s: bad inode: ino = %u\n",
238                 LDISKFS_SB(osd_sb(dev))->s_es->s_volume_name, id->oii_ino);
239                 iput(inode);
240                 inode = ERR_PTR(-ENOENT);
241         } else {
242                 if (id->oii_gen == OSD_OII_NOGEN)
243                         osd_id_gen(id, inode->i_ino, inode->i_generation);
244
245                 /* Do not update file c/mtime in ldiskfs.
246                  * NB: we don't have any lock to protect this because we don't
247                  * have reference on osd_object now, but contention with
248                  * another lookup + attr_set can't happen in the tiny window
249                  * between if (...) and set S_NOCMTIME. */
250                 if (!(inode->i_flags & S_NOCMTIME))
251                         inode->i_flags |= S_NOCMTIME;
252         }
253         return inode;
254 }
255
256 static struct inode *
257 osd_iget_fid(struct osd_thread_info *info, struct osd_device *dev,
258              struct osd_inode_id *id, struct lu_fid *fid)
259 {
260         struct lustre_mdt_attrs *lma   = &info->oti_mdt_attrs;
261         struct inode            *inode;
262         int                      rc;
263
264         inode = osd_iget(info, dev, id);
265         if (IS_ERR(inode))
266                 return inode;
267
268         rc = osd_get_lma(info, inode, &info->oti_obj_dentry, lma);
269         if (rc == 0) {
270                 *fid = lma->lma_self_fid;
271         } else if (rc == -ENODATA) {
272                 if (unlikely(inode == osd_sb(dev)->s_root->d_inode))
273                         lu_local_obj_fid(fid, OSD_FS_ROOT_OID);
274                 else
275                         lu_igif_build(fid, inode->i_ino, inode->i_generation);
276         } else {
277                 iput(inode);
278                 inode = ERR_PTR(rc);
279         }
280         return inode;
281 }
282
283 static struct inode *
284 osd_iget_verify(struct osd_thread_info *info, struct osd_device *dev,
285                 struct osd_inode_id *id, const struct lu_fid *fid)
286 {
287         struct lustre_mdt_attrs *lma   = &info->oti_mdt_attrs;
288         struct inode            *inode;
289         int                      rc;
290
291         inode = osd_iget(info, dev, id);
292         if (IS_ERR(inode))
293                 return inode;
294
295         rc = osd_get_lma(info, inode, &info->oti_obj_dentry, lma);
296         if (rc == -ENODATA)
297                 return inode;
298
299         if (rc != 0) {
300                 iput(inode);
301                 return ERR_PTR(rc);
302         }
303
304         if (!lu_fid_eq(fid, &lma->lma_self_fid)) {
305                 CDEBUG(D_LFSCK, "inconsistent obj: "DFID", %lu, "DFID"\n",
306                        PFID(&lma->lma_self_fid), inode->i_ino, PFID(fid));
307                 iput(inode);
308                 return ERR_PTR(-EREMCHG);
309         }
310
311         return inode;
312 }
313
314 static int osd_fid_lookup(const struct lu_env *env, struct osd_object *obj,
315                           const struct lu_fid *fid,
316                           const struct lu_object_conf *conf)
317 {
318         struct osd_thread_info *info;
319         struct lu_device       *ldev   = obj->oo_dt.do_lu.lo_dev;
320         struct osd_device      *dev;
321         struct osd_idmap_cache *oic;
322         struct osd_inode_id    *id;
323         struct inode           *inode;
324         struct osd_scrub       *scrub;
325         struct scrub_file      *sf;
326         int                     result;
327         int                     verify = 0;
328         ENTRY;
329
330         LINVRNT(osd_invariant(obj));
331         LASSERT(obj->oo_inode == NULL);
332         LASSERTF(fid_is_sane(fid) || fid_is_idif(fid), DFID, PFID(fid));
333
334         dev = osd_dev(ldev);
335         scrub = &dev->od_scrub;
336         sf = &scrub->os_file;
337         info = osd_oti_get(env);
338         LASSERT(info);
339         oic = &info->oti_cache;
340
341         if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOENT))
342                 RETURN(-ENOENT);
343
344         /* Search order: 1. per-thread cache. */
345         if (lu_fid_eq(fid, &oic->oic_fid)) {
346                 id = &oic->oic_lid;
347                 goto iget;
348         }
349
350         id = &info->oti_id;
351         if (!cfs_list_empty(&scrub->os_inconsistent_items)) {
352                 /* Search order: 2. OI scrub pending list. */
353                 result = osd_oii_lookup(dev, fid, id);
354                 if (result == 0)
355                         goto iget;
356         }
357
358         if (sf->sf_flags & SF_INCONSISTENT)
359                 verify = 1;
360
361         /*
362          * Objects are created as locking anchors or place holders for objects
363          * yet to be created. No need to osd_oi_lookup() at here because FID
364          * shouldn't never be re-used, if it's really a duplicate FID from
365          * unexpected reason, we should be able to detect it later by calling
366          * do_create->osd_oi_insert()
367          */
368         if (conf != NULL && conf->loc_flags & LOC_F_NEW)
369                 GOTO(out, result = 0);
370
371         /* Search order: 3. OI files. */
372         result = osd_oi_lookup(info, dev, fid, id, true);
373         if (result == -ENOENT) {
374                 if (!fid_is_norm(fid) || fid_is_on_ost(info, dev, fid) ||
375                     !ldiskfs_test_bit(osd_oi_fid2idx(dev,fid),
376                                       sf->sf_oi_bitmap))
377                         GOTO(out, result = 0);
378
379                 goto trigger;
380         }
381
382         if (result != 0)
383                 GOTO(out, result);
384
385 iget:
386         if (verify == 0)
387                 inode = osd_iget(info, dev, id);
388         else
389                 inode = osd_iget_verify(info, dev, id, fid);
390         if (IS_ERR(inode)) {
391                 result = PTR_ERR(inode);
392                 if (result == -ENOENT || result == -ESTALE) {
393                         fid_zero(&oic->oic_fid);
394                         result = 0;
395                 } else if (result == -EREMCHG) {
396
397 trigger:
398                         if (thread_is_running(&scrub->os_thread)) {
399                                 result = -EINPROGRESS;
400                         } else if (!dev->od_noscrub) {
401                                 result = osd_scrub_start(dev);
402                                 LCONSOLE_ERROR("%.16s: trigger OI scrub by RPC "
403                                                "for "DFID", rc = %d [1]\n",
404                                                LDISKFS_SB(osd_sb(dev))->s_es->\
405                                                s_volume_name,PFID(fid), result);
406                                 if (result == 0 || result == -EALREADY)
407                                         result = -EINPROGRESS;
408                                 else
409                                         result = -EREMCHG;
410                         }
411                 }
412
413                 GOTO(out, result);
414         }
415
416         obj->oo_inode = inode;
417         LASSERT(obj->oo_inode->i_sb == osd_sb(dev));
418
419         obj->oo_compat_dot_created = 1;
420         obj->oo_compat_dotdot_created = 1;
421
422         if (!S_ISDIR(inode->i_mode) || !ldiskfs_pdo) /* done */
423                 GOTO(out, result = 0);
424
425         LASSERT(obj->oo_hl_head == NULL);
426         obj->oo_hl_head = ldiskfs_htree_lock_head_alloc(HTREE_HBITS_DEF);
427         if (obj->oo_hl_head == NULL) {
428                 obj->oo_inode = NULL;
429                 iput(inode);
430                 GOTO(out, result = -ENOMEM);
431         }
432         GOTO(out, result = 0);
433
434 out:
435         LINVRNT(osd_invariant(obj));
436         return result;
437 }
438
439 /*
440  * Concurrency: shouldn't matter.
441  */
442 static void osd_object_init0(struct osd_object *obj)
443 {
444         LASSERT(obj->oo_inode != NULL);
445         obj->oo_dt.do_body_ops = &osd_body_ops;
446         obj->oo_dt.do_lu.lo_header->loh_attr |=
447                 (LOHA_EXISTS | (obj->oo_inode->i_mode & S_IFMT));
448 }
449
450 static int osd_check_lma(const struct lu_env *env, struct osd_object *obj)
451 {
452         struct osd_thread_info  *info   = osd_oti_get(env);
453         struct lustre_mdt_attrs *lma    = &info->oti_mdt_attrs;
454         int                     rc;
455         ENTRY;
456
457         CLASSERT(LMA_OLD_SIZE >= sizeof(*lma));
458         rc = __osd_xattr_get(obj->oo_inode, &info->oti_obj_dentry,
459                              XATTR_NAME_LMA, info->oti_mdt_attrs_old,
460                              LMA_OLD_SIZE);
461         if (rc > 0) {
462                 rc = 0;
463                 lustre_lma_swab(lma);
464                 if (unlikely((lma->lma_incompat & ~LMA_INCOMPAT_SUPP) ||
465                              CFS_FAIL_CHECK(OBD_FAIL_OSD_LMA_INCOMPAT))) {
466                         rc = -EOPNOTSUPP;
467                         CWARN("%s: unsupported incompat LMA feature(s) %#x for "
468                               "fid = "DFID", ino = %lu: rc = %d\n",
469                               osd_obj2dev(obj)->od_svname,
470                               lma->lma_incompat & ~LMA_INCOMPAT_SUPP,
471                               PFID(lu_object_fid(&obj->oo_dt.do_lu)),
472                               obj->oo_inode->i_ino, rc);
473                 }
474                 if (unlikely(!lu_fid_eq(lu_object_fid(&obj->oo_dt.do_lu),
475                                         &lma->lma_self_fid))) {
476                         CDEBUG(D_INODE, "%s: FID "DFID" != self_fid "DFID"\n",
477                                osd_obj2dev(obj)->od_svname,
478                                PFID(lu_object_fid(&obj->oo_dt.do_lu)),
479                                PFID(&lma->lma_self_fid));
480                         if (obj->oo_inode != NULL) {
481                                 iput(obj->oo_inode);
482                                 obj->oo_inode = NULL;
483                         }
484                         rc = -ESTALE;
485                 }
486         } else if (rc == -ENODATA) {
487                 /* haven't initialize LMA xattr */
488                 rc = 0;
489         }
490
491         RETURN(rc);
492 }
493
494 /*
495  * Concurrency: no concurrent access is possible that early in object
496  * life-cycle.
497  */
498 static int osd_object_init(const struct lu_env *env, struct lu_object *l,
499                            const struct lu_object_conf *conf)
500 {
501         struct osd_object *obj = osd_obj(l);
502         int result;
503
504         LINVRNT(osd_invariant(obj));
505
506         if (fid_is_otable_it(&l->lo_header->loh_fid)) {
507                 obj->oo_dt.do_ops = &osd_obj_otable_it_ops;
508                 l->lo_header->loh_attr |= LOHA_EXISTS;
509                 return 0;
510         }
511
512         result = osd_fid_lookup(env, obj, lu_object_fid(l), conf);
513         obj->oo_dt.do_body_ops = &osd_body_ops_new;
514         if (result == 0 && obj->oo_inode != NULL) {
515                 result = osd_check_lma(env, obj);
516                 if (result != 0)
517                         return result;
518
519                 osd_object_init0(obj);
520         }
521
522         LINVRNT(osd_invariant(obj));
523         return result;
524 }
525
526 /*
527  * Concurrency: no concurrent access is possible that late in object
528  * life-cycle.
529  */
530 static void osd_object_free(const struct lu_env *env, struct lu_object *l)
531 {
532         struct osd_object *obj = osd_obj(l);
533
534         LINVRNT(osd_invariant(obj));
535
536         dt_object_fini(&obj->oo_dt);
537         if (obj->oo_hl_head != NULL)
538                 ldiskfs_htree_lock_head_free(obj->oo_hl_head);
539         OBD_FREE_PTR(obj);
540 }
541
542 /*
543  * Concurrency: no concurrent access is possible that late in object
544  * life-cycle.
545  */
546 static void osd_index_fini(struct osd_object *o)
547 {
548         struct iam_container *bag;
549
550         if (o->oo_dir != NULL) {
551                 bag = &o->oo_dir->od_container;
552                 if (o->oo_inode != NULL) {
553                         if (bag->ic_object == o->oo_inode)
554                                 iam_container_fini(bag);
555                 }
556                 OBD_FREE_PTR(o->oo_dir);
557                 o->oo_dir = NULL;
558         }
559 }
560
561 /*
562  * Concurrency: no concurrent access is possible that late in object
563  * life-cycle (for all existing callers, that is. New callers have to provide
564  * their own locking.)
565  */
566 static int osd_inode_unlinked(const struct inode *inode)
567 {
568         return inode->i_nlink == 0;
569 }
570
571 enum {
572         OSD_TXN_OI_DELETE_CREDITS    = 20,
573         OSD_TXN_INODE_DELETE_CREDITS = 20
574 };
575
576 /*
577  * Journal
578  */
579
580 #if OSD_THANDLE_STATS
581 /**
582  * Set time when the handle is allocated
583  */
584 static void osd_th_alloced(struct osd_thandle *oth)
585 {
586         oth->oth_alloced = cfs_time_current();
587 }
588
589 /**
590  * Set time when the handle started
591  */
592 static void osd_th_started(struct osd_thandle *oth)
593 {
594         oth->oth_started = cfs_time_current();
595 }
596
597 /**
598  * Helper function to convert time interval to microseconds packed in
599  * long int.
600  */
601 static long interval_to_usec(cfs_time_t start, cfs_time_t end)
602 {
603         struct timeval val;
604
605         cfs_duration_usec(cfs_time_sub(end, start), &val);
606         return val.tv_sec * 1000000 + val.tv_usec;
607 }
608
609 /**
610  * Check whether the we deal with this handle for too long.
611  */
612 static void __osd_th_check_slow(void *oth, struct osd_device *dev,
613                                 cfs_time_t alloced, cfs_time_t started,
614                                 cfs_time_t closed)
615 {
616         cfs_time_t now = cfs_time_current();
617
618         LASSERT(dev != NULL);
619
620         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_STARTING,
621                             interval_to_usec(alloced, started));
622         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_OPEN,
623                             interval_to_usec(started, closed));
624         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_CLOSING,
625                             interval_to_usec(closed, now));
626
627         if (cfs_time_before(cfs_time_add(alloced, cfs_time_seconds(30)), now)) {
628                 CWARN("transaction handle %p was open for too long: "
629                       "now "CFS_TIME_T" ,"
630                       "alloced "CFS_TIME_T" ,"
631                       "started "CFS_TIME_T" ,"
632                       "closed "CFS_TIME_T"\n",
633                       oth, now, alloced, started, closed);
634                 libcfs_debug_dumpstack(NULL);
635         }
636 }
637
638 #define OSD_CHECK_SLOW_TH(oth, dev, expr)                               \
639 {                                                                       \
640         cfs_time_t __closed = cfs_time_current();                       \
641         cfs_time_t __alloced = oth->oth_alloced;                        \
642         cfs_time_t __started = oth->oth_started;                        \
643                                                                         \
644         expr;                                                           \
645         __osd_th_check_slow(oth, dev, __alloced, __started, __closed);  \
646 }
647
648 #else /* OSD_THANDLE_STATS */
649
650 #define osd_th_alloced(h)                  do {} while(0)
651 #define osd_th_started(h)                  do {} while(0)
652 #define OSD_CHECK_SLOW_TH(oth, dev, expr)  expr
653
654 #endif /* OSD_THANDLE_STATS */
655
656 /*
657  * Concurrency: doesn't access mutable data.
658  */
659 static int osd_param_is_not_sane(const struct osd_device *dev,
660                                  const struct thandle *th)
661 {
662         struct osd_thandle *oh = container_of(th, typeof(*oh), ot_super);
663
664         return oh->ot_credits > osd_journal(dev)->j_max_transaction_buffers;
665 }
666
667 /*
668  * Concurrency: shouldn't matter.
669  */
670 static void osd_trans_commit_cb(struct super_block *sb,
671                                 struct ldiskfs_journal_cb_entry *jcb, int error)
672 {
673         struct osd_thandle *oh = container_of0(jcb, struct osd_thandle, ot_jcb);
674         struct thandle     *th  = &oh->ot_super;
675         struct lu_device   *lud = &th->th_dev->dd_lu_dev;
676         struct dt_txn_commit_cb *dcb, *tmp;
677
678         LASSERT(oh->ot_handle == NULL);
679
680         if (error)
681                 CERROR("transaction @0x%p commit error: %d\n", th, error);
682
683         dt_txn_hook_commit(th);
684
685         /* call per-transaction callbacks if any */
686         cfs_list_for_each_entry_safe(dcb, tmp, &oh->ot_dcb_list, dcb_linkage) {
687                 LASSERTF(dcb->dcb_magic == TRANS_COMMIT_CB_MAGIC,
688                          "commit callback entry: magic=%x name='%s'\n",
689                          dcb->dcb_magic, dcb->dcb_name);
690                 cfs_list_del_init(&dcb->dcb_linkage);
691                 dcb->dcb_func(NULL, th, dcb, error);
692         }
693
694         lu_ref_del_at(&lud->ld_reference, oh->ot_dev_link, "osd-tx", th);
695         lu_device_put(lud);
696         th->th_dev = NULL;
697
698         lu_context_exit(&th->th_ctx);
699         lu_context_fini(&th->th_ctx);
700         OBD_FREE_PTR(oh);
701 }
702
703 static struct thandle *osd_trans_create(const struct lu_env *env,
704                                         struct dt_device *d)
705 {
706         struct osd_thread_info *oti = osd_oti_get(env);
707         struct osd_iobuf       *iobuf = &oti->oti_iobuf;
708         struct osd_thandle     *oh;
709         struct thandle         *th;
710         ENTRY;
711
712         /* on pending IO in this thread should left from prev. request */
713         LASSERT(cfs_atomic_read(&iobuf->dr_numreqs) == 0);
714
715         th = ERR_PTR(-ENOMEM);
716         OBD_ALLOC_GFP(oh, sizeof *oh, CFS_ALLOC_IO);
717         if (oh != NULL) {
718                 oh->ot_quota_trans = &oti->oti_quota_trans;
719                 memset(oh->ot_quota_trans, 0, sizeof(*oh->ot_quota_trans));
720                 th = &oh->ot_super;
721                 th->th_dev = d;
722                 th->th_result = 0;
723                 th->th_tags = LCT_TX_HANDLE;
724                 oh->ot_credits = 0;
725                 oti->oti_dev = osd_dt_dev(d);
726                 CFS_INIT_LIST_HEAD(&oh->ot_dcb_list);
727                 osd_th_alloced(oh);
728
729                 memset(oti->oti_declare_ops, 0,
730                                         sizeof(oti->oti_declare_ops));
731                 memset(oti->oti_declare_ops_rb, 0,
732                                         sizeof(oti->oti_declare_ops_rb));
733                 memset(oti->oti_declare_ops_cred, 0,
734                                         sizeof(oti->oti_declare_ops_cred));
735                 oti->oti_rollback = false;
736         }
737         RETURN(th);
738 }
739
740 /*
741  * Concurrency: shouldn't matter.
742  */
743 int osd_trans_start(const struct lu_env *env, struct dt_device *d,
744                     struct thandle *th)
745 {
746         struct osd_thread_info *oti = osd_oti_get(env);
747         struct osd_device  *dev = osd_dt_dev(d);
748         handle_t           *jh;
749         struct osd_thandle *oh;
750         int rc;
751
752         ENTRY;
753
754         LASSERT(current->journal_info == NULL);
755
756         oh = container_of0(th, struct osd_thandle, ot_super);
757         LASSERT(oh != NULL);
758         LASSERT(oh->ot_handle == NULL);
759
760         rc = dt_txn_hook_start(env, d, th);
761         if (rc != 0)
762                 GOTO(out, rc);
763
764         if (unlikely(osd_param_is_not_sane(dev, th))) {
765                 static unsigned long last_printed;
766                 static int last_credits;
767
768                 CWARN("%.16s: too many transaction credits (%d > %d)\n",
769                       LDISKFS_SB(osd_sb(dev))->s_es->s_volume_name,
770                       oh->ot_credits,
771                       osd_journal(dev)->j_max_transaction_buffers);
772                 CWARN("  create: %u/%u, delete: %u/%u, destroy: %u/%u\n",
773                       oti->oti_declare_ops[OSD_OT_CREATE],
774                       oti->oti_declare_ops_cred[OSD_OT_CREATE],
775                       oti->oti_declare_ops[OSD_OT_DELETE],
776                       oti->oti_declare_ops_cred[OSD_OT_DELETE],
777                       oti->oti_declare_ops[OSD_OT_DESTROY],
778                       oti->oti_declare_ops_cred[OSD_OT_DESTROY]);
779                 CWARN("  attr_set: %u/%u, xattr_set: %u/%u\n",
780                       oti->oti_declare_ops[OSD_OT_ATTR_SET],
781                       oti->oti_declare_ops_cred[OSD_OT_ATTR_SET],
782                       oti->oti_declare_ops[OSD_OT_XATTR_SET],
783                       oti->oti_declare_ops_cred[OSD_OT_XATTR_SET]);
784                 CWARN("  write: %u/%u, punch: %u/%u, quota %u/%u\n",
785                       oti->oti_declare_ops[OSD_OT_WRITE],
786                       oti->oti_declare_ops_cred[OSD_OT_WRITE],
787                       oti->oti_declare_ops[OSD_OT_PUNCH],
788                       oti->oti_declare_ops_cred[OSD_OT_PUNCH],
789                       oti->oti_declare_ops[OSD_OT_QUOTA],
790                       oti->oti_declare_ops_cred[OSD_OT_QUOTA]);
791                 CWARN("  insert: %u/%u, delete: %u/%u\n",
792                       oti->oti_declare_ops[OSD_OT_INSERT],
793                       oti->oti_declare_ops_cred[OSD_OT_INSERT],
794                       oti->oti_declare_ops[OSD_OT_DESTROY],
795                       oti->oti_declare_ops_cred[OSD_OT_DESTROY]);
796                 CWARN("  ref_add: %u/%u, ref_del: %u/%u\n",
797                       oti->oti_declare_ops[OSD_OT_REF_ADD],
798                       oti->oti_declare_ops_cred[OSD_OT_REF_ADD],
799                       oti->oti_declare_ops[OSD_OT_REF_DEL],
800                       oti->oti_declare_ops_cred[OSD_OT_REF_DEL]);
801
802                 if (last_credits != oh->ot_credits &&
803                     time_after(jiffies, last_printed + 60 * HZ)) {
804                         libcfs_debug_dumpstack(NULL);
805                         last_credits = oh->ot_credits;
806                         last_printed = jiffies;
807                 }
808                 /* XXX Limit the credits to 'max_transaction_buffers', and
809                  *     let the underlying filesystem to catch the error if
810                  *     we really need so many credits.
811                  *
812                  *     This should be removed when we can calculate the
813                  *     credits precisely. */
814                 oh->ot_credits = osd_journal(dev)->j_max_transaction_buffers;
815         }
816
817         /*
818          * XXX temporary stuff. Some abstraction layer should
819          * be used.
820          */
821         jh = ldiskfs_journal_start_sb(osd_sb(dev), oh->ot_credits);
822         osd_th_started(oh);
823         if (!IS_ERR(jh)) {
824                 oh->ot_handle = jh;
825                 LASSERT(oti->oti_txns == 0);
826                 lu_context_init(&th->th_ctx, th->th_tags);
827                 lu_context_enter(&th->th_ctx);
828
829                 lu_device_get(&d->dd_lu_dev);
830                 oh->ot_dev_link = lu_ref_add(&d->dd_lu_dev.ld_reference,
831                                              "osd-tx", th);
832                 oti->oti_txns++;
833                 rc = 0;
834         } else {
835                 rc = PTR_ERR(jh);
836         }
837 out:
838         RETURN(rc);
839 }
840
841 /*
842  * Concurrency: shouldn't matter.
843  */
844 static int osd_trans_stop(const struct lu_env *env, struct thandle *th)
845 {
846         int                     rc = 0;
847         struct osd_thandle     *oh;
848         struct osd_thread_info *oti = osd_oti_get(env);
849         struct osd_iobuf       *iobuf = &oti->oti_iobuf;
850         struct qsd_instance    *qsd = oti->oti_dev->od_quota_slave;
851         ENTRY;
852
853         oh = container_of0(th, struct osd_thandle, ot_super);
854
855         if (qsd != NULL)
856                 /* inform the quota slave device that the transaction is
857                  * stopping */
858                 qsd_op_end(env, qsd, oh->ot_quota_trans);
859         oh->ot_quota_trans = NULL;
860
861         if (oh->ot_handle != NULL) {
862                 handle_t *hdl = oh->ot_handle;
863
864                 /*
865                  * add commit callback
866                  * notice we don't do this in osd_trans_start()
867                  * as underlying transaction can change during truncate
868                  */
869                 ldiskfs_journal_callback_add(hdl, osd_trans_commit_cb,
870                                          &oh->ot_jcb);
871
872                 LASSERT(oti->oti_txns == 1);
873                 oti->oti_txns--;
874                 rc = dt_txn_hook_stop(env, th);
875                 if (rc != 0)
876                         CERROR("Failure in transaction hook: %d\n", rc);
877
878                 /* hook functions might modify th_sync */
879                 hdl->h_sync = th->th_sync;
880
881                 oh->ot_handle = NULL;
882                 OSD_CHECK_SLOW_TH(oh, oti->oti_dev,
883                                   rc = ldiskfs_journal_stop(hdl));
884                 if (rc != 0)
885                         CERROR("Failure to stop transaction: %d\n", rc);
886         } else {
887                 OBD_FREE_PTR(oh);
888         }
889
890         /* as we want IO to journal and data IO be concurrent, we don't block
891          * awaiting data IO completion in osd_do_bio(), instead we wait here
892          * once transaction is submitted to the journal. all reqular requests
893          * don't do direct IO (except read/write), thus this wait_event becomes
894          * no-op for them.
895          *
896          * IMPORTANT: we have to wait till any IO submited by the thread is
897          * completed otherwise iobuf may be corrupted by different request
898          */
899         cfs_wait_event(iobuf->dr_wait,
900                        cfs_atomic_read(&iobuf->dr_numreqs) == 0);
901         if (!rc)
902                 rc = iobuf->dr_error;
903
904         RETURN(rc);
905 }
906
907 static int osd_trans_cb_add(struct thandle *th, struct dt_txn_commit_cb *dcb)
908 {
909         struct osd_thandle *oh = container_of0(th, struct osd_thandle,
910                                                ot_super);
911
912         LASSERT(dcb->dcb_magic == TRANS_COMMIT_CB_MAGIC);
913         LASSERT(&dcb->dcb_func != NULL);
914         cfs_list_add(&dcb->dcb_linkage, &oh->ot_dcb_list);
915
916         return 0;
917 }
918
919 /*
920  * Called just before object is freed. Releases all resources except for
921  * object itself (that is released by osd_object_free()).
922  *
923  * Concurrency: no concurrent access is possible that late in object
924  * life-cycle.
925  */
926 static void osd_object_delete(const struct lu_env *env, struct lu_object *l)
927 {
928         struct osd_object *obj   = osd_obj(l);
929         struct inode      *inode = obj->oo_inode;
930
931         LINVRNT(osd_invariant(obj));
932
933         /*
934          * If object is unlinked remove fid->ino mapping from object index.
935          */
936
937         osd_index_fini(obj);
938         if (inode != NULL) {
939                 struct qsd_instance     *qsd = osd_obj2dev(obj)->od_quota_slave;
940                 qid_t                    uid = inode->i_uid;
941                 qid_t                    gid = inode->i_gid;
942
943                 iput(inode);
944                 obj->oo_inode = NULL;
945
946                 if (qsd != NULL) {
947                         struct osd_thread_info  *info = osd_oti_get(env);
948                         struct lquota_id_info   *qi = &info->oti_qi;
949
950                         /* Release granted quota to master if necessary */
951                         qi->lqi_id.qid_uid = uid;
952                         qsd_op_adjust(env, qsd, &qi->lqi_id, USRQUOTA);
953
954                         qi->lqi_id.qid_uid = gid;
955                         qsd_op_adjust(env, qsd, &qi->lqi_id, GRPQUOTA);
956                 }
957         }
958 }
959
960 /*
961  * Concurrency: ->loo_object_release() is called under site spin-lock.
962  */
963 static void osd_object_release(const struct lu_env *env,
964                                struct lu_object *l)
965 {
966 }
967
968 /*
969  * Concurrency: shouldn't matter.
970  */
971 static int osd_object_print(const struct lu_env *env, void *cookie,
972                             lu_printer_t p, const struct lu_object *l)
973 {
974         struct osd_object *o = osd_obj(l);
975         struct iam_descr  *d;
976
977         if (o->oo_dir != NULL)
978                 d = o->oo_dir->od_container.ic_descr;
979         else
980                 d = NULL;
981         return (*p)(env, cookie,
982                     LUSTRE_OSD_LDISKFS_NAME"-object@%p(i:%p:%lu/%u)[%s]",
983                     o, o->oo_inode,
984                     o->oo_inode ? o->oo_inode->i_ino : 0UL,
985                     o->oo_inode ? o->oo_inode->i_generation : 0,
986                     d ? d->id_ops->id_name : "plain");
987 }
988
989 /*
990  * Concurrency: shouldn't matter.
991  */
992 int osd_statfs(const struct lu_env *env, struct dt_device *d,
993                struct obd_statfs *sfs)
994 {
995         struct osd_device  *osd = osd_dt_dev(d);
996         struct super_block *sb = osd_sb(osd);
997         struct kstatfs     *ksfs;
998         int result = 0;
999
1000         if (unlikely(osd->od_mnt == NULL))
1001                 return -EINPROGRESS;
1002
1003         /* osd_lproc.c call this without env, allocate ksfs for that case */
1004         if (unlikely(env == NULL)) {
1005                 OBD_ALLOC_PTR(ksfs);
1006                 if (ksfs == NULL)
1007                         return -ENOMEM;
1008         } else {
1009                 ksfs = &osd_oti_get(env)->oti_ksfs;
1010         }
1011
1012         spin_lock(&osd->od_osfs_lock);
1013         /* cache 1 second */
1014         if (cfs_time_before_64(osd->od_osfs_age, cfs_time_shift_64(-1))) {
1015                 result = sb->s_op->statfs(sb->s_root, ksfs);
1016                 if (likely(result == 0)) { /* N.B. statfs can't really fail */
1017                         osd->od_osfs_age = cfs_time_current_64();
1018                         statfs_pack(&osd->od_statfs, ksfs);
1019                         if (sb->s_flags & MS_RDONLY)
1020                                 sfs->os_state = OS_STATE_READONLY;
1021                 }
1022         }
1023
1024         if (likely(result == 0))
1025                 *sfs = osd->od_statfs;
1026         spin_unlock(&osd->od_osfs_lock);
1027
1028         if (unlikely(env == NULL))
1029                 OBD_FREE_PTR(ksfs);
1030
1031         return result;
1032 }
1033
1034 /**
1035  * Estimate space needed for file creations. We assume the largest filename
1036  * which is 2^64 - 1, hence a filename of 20 chars.
1037  * This is 28 bytes per object which is 28MB for 1M objects ... no so bad.
1038  */
1039 #ifdef __LDISKFS_DIR_REC_LEN
1040 #define PER_OBJ_USAGE __LDISKFS_DIR_REC_LEN(20)
1041 #else
1042 #define PER_OBJ_USAGE LDISKFS_DIR_REC_LEN(20)
1043 #endif
1044
1045 /*
1046  * Concurrency: doesn't access mutable data.
1047  */
1048 static void osd_conf_get(const struct lu_env *env,
1049                          const struct dt_device *dev,
1050                          struct dt_device_param *param)
1051 {
1052         struct super_block *sb = osd_sb(osd_dt_dev(dev));
1053
1054         /*
1055          * XXX should be taken from not-yet-existing fs abstraction layer.
1056          */
1057         param->ddp_mnt = osd_dt_dev(dev)->od_mnt;
1058         param->ddp_max_name_len = LDISKFS_NAME_LEN;
1059         param->ddp_max_nlink    = LDISKFS_LINK_MAX;
1060         param->ddp_block_shift  = sb->s_blocksize_bits;
1061         param->ddp_mount_type     = LDD_MT_LDISKFS;
1062         param->ddp_maxbytes       = sb->s_maxbytes;
1063         /* Overhead estimate should be fairly accurate, so we really take a tiny
1064          * error margin which also avoids fragmenting the filesystem too much */
1065         param->ddp_grant_reserved = 2; /* end up to be 1.9% after conversion */
1066         /* inode are statically allocated, so per-inode space consumption
1067          * is the space consumed by the directory entry */
1068         param->ddp_inodespace     = PER_OBJ_USAGE;
1069         /* per-fragment overhead to be used by the client code */
1070         param->ddp_grant_frag     = 6 * LDISKFS_BLOCK_SIZE(sb);
1071         param->ddp_mntopts      = 0;
1072         if (test_opt(sb, XATTR_USER))
1073                 param->ddp_mntopts |= MNTOPT_USERXATTR;
1074         if (test_opt(sb, POSIX_ACL))
1075                 param->ddp_mntopts |= MNTOPT_ACL;
1076
1077 #if defined(LDISKFS_FEATURE_INCOMPAT_EA_INODE)
1078         if (LDISKFS_HAS_INCOMPAT_FEATURE(sb, LDISKFS_FEATURE_INCOMPAT_EA_INODE))
1079                 param->ddp_max_ea_size = LDISKFS_XATTR_MAX_LARGE_EA_SIZE;
1080         else
1081 #endif
1082                 param->ddp_max_ea_size = sb->s_blocksize;
1083
1084 }
1085
1086 /*
1087  * Concurrency: shouldn't matter.
1088  */
1089 static int osd_sync(const struct lu_env *env, struct dt_device *d)
1090 {
1091         CDEBUG(D_HA, "syncing OSD %s\n", LUSTRE_OSD_LDISKFS_NAME);
1092         return ldiskfs_force_commit(osd_sb(osd_dt_dev(d)));
1093 }
1094
1095 /**
1096  * Start commit for OSD device.
1097  *
1098  * An implementation of dt_commit_async method for OSD device.
1099  * Asychronously starts underlayng fs sync and thereby a transaction
1100  * commit.
1101  *
1102  * \param env environment
1103  * \param d dt device
1104  *
1105  * \see dt_device_operations
1106  */
1107 static int osd_commit_async(const struct lu_env *env,
1108                             struct dt_device *d)
1109 {
1110         struct super_block *s = osd_sb(osd_dt_dev(d));
1111         ENTRY;
1112
1113         CDEBUG(D_HA, "async commit OSD %s\n", LUSTRE_OSD_LDISKFS_NAME);
1114         RETURN(s->s_op->sync_fs(s, 0));
1115 }
1116
1117 /*
1118  * Concurrency: shouldn't matter.
1119  */
1120
1121 static int osd_ro(const struct lu_env *env, struct dt_device *d)
1122 {
1123         struct super_block *sb = osd_sb(osd_dt_dev(d));
1124         struct block_device *dev = sb->s_bdev;
1125 #ifdef HAVE_DEV_SET_RDONLY
1126         struct block_device *jdev = LDISKFS_SB(sb)->journal_bdev;
1127         int rc = 0;
1128 #else
1129         int rc = -EOPNOTSUPP;
1130 #endif
1131         ENTRY;
1132
1133 #ifdef HAVE_DEV_SET_RDONLY
1134         CERROR("*** setting %s read-only ***\n", osd_dt_dev(d)->od_svname);
1135
1136         if (jdev && (jdev != dev)) {
1137                 CDEBUG(D_IOCTL | D_HA, "set journal dev %lx rdonly\n",
1138                        (long)jdev);
1139                 dev_set_rdonly(jdev);
1140         }
1141         CDEBUG(D_IOCTL | D_HA, "set dev %lx rdonly\n", (long)dev);
1142         dev_set_rdonly(dev);
1143 #else
1144         CERROR("%s: %lx CANNOT BE SET READONLY: rc = %d\n",
1145                osd_dt_dev(d)->od_svname, (long)dev, rc);
1146 #endif
1147         RETURN(rc);
1148 }
1149
1150 /*
1151  * Concurrency: serialization provided by callers.
1152  */
1153 static int osd_init_capa_ctxt(const struct lu_env *env, struct dt_device *d,
1154                               int mode, unsigned long timeout, __u32 alg,
1155                               struct lustre_capa_key *keys)
1156 {
1157         struct osd_device *dev = osd_dt_dev(d);
1158         ENTRY;
1159
1160         dev->od_fl_capa = mode;
1161         dev->od_capa_timeout = timeout;
1162         dev->od_capa_alg = alg;
1163         dev->od_capa_keys = keys;
1164         RETURN(0);
1165 }
1166
1167 /**
1168  * Note: we do not count into QUOTA here.
1169  * If we mount with --data_journal we may need more.
1170  */
1171 const int osd_dto_credits_noquota[DTO_NR] = {
1172         /**
1173          * Insert/Delete.
1174          * INDEX_EXTRA_TRANS_BLOCKS(8) +
1175          * SINGLEDATA_TRANS_BLOCKS(8)
1176          * XXX Note: maybe iam need more, since iam have more level than
1177          *           EXT3 htree.
1178          */
1179         [DTO_INDEX_INSERT]  = 16,
1180         [DTO_INDEX_DELETE]  = 16,
1181         /**
1182          * Used for OI scrub
1183          */
1184         [DTO_INDEX_UPDATE]  = 16,
1185         /**
1186          * Create a object. The same as create object in EXT3.
1187          * DATA_TRANS_BLOCKS(14) +
1188          * INDEX_EXTRA_BLOCKS(8) +
1189          * 3(inode bits, groups, GDT)
1190          */
1191         [DTO_OBJECT_CREATE] = 25,
1192         /**
1193          * XXX: real credits to be fixed
1194          */
1195         [DTO_OBJECT_DELETE] = 25,
1196         /**
1197          * Attr set credits (inode)
1198          */
1199         [DTO_ATTR_SET_BASE] = 1,
1200         /**
1201          * Xattr set. The same as xattr of EXT3.
1202          * DATA_TRANS_BLOCKS(14)
1203          * XXX Note: in original MDS implmentation INDEX_EXTRA_TRANS_BLOCKS
1204          * are also counted in. Do not know why?
1205          */
1206         [DTO_XATTR_SET]     = 14,
1207         [DTO_LOG_REC]       = 14,
1208         /**
1209          * credits for inode change during write.
1210          */
1211         [DTO_WRITE_BASE]    = 3,
1212         /**
1213          * credits for single block write.
1214          */
1215         [DTO_WRITE_BLOCK]   = 14,
1216         /**
1217          * Attr set credits for chown.
1218          * This is extra credits for setattr, and it is null without quota
1219          */
1220         [DTO_ATTR_SET_CHOWN]= 0
1221 };
1222
1223 static const struct dt_device_operations osd_dt_ops = {
1224         .dt_root_get       = osd_root_get,
1225         .dt_statfs         = osd_statfs,
1226         .dt_trans_create   = osd_trans_create,
1227         .dt_trans_start    = osd_trans_start,
1228         .dt_trans_stop     = osd_trans_stop,
1229         .dt_trans_cb_add   = osd_trans_cb_add,
1230         .dt_conf_get       = osd_conf_get,
1231         .dt_sync           = osd_sync,
1232         .dt_ro             = osd_ro,
1233         .dt_commit_async   = osd_commit_async,
1234         .dt_init_capa_ctxt = osd_init_capa_ctxt,
1235 };
1236
1237 static void osd_object_read_lock(const struct lu_env *env,
1238                                  struct dt_object *dt, unsigned role)
1239 {
1240         struct osd_object *obj = osd_dt_obj(dt);
1241         struct osd_thread_info *oti = osd_oti_get(env);
1242
1243         LINVRNT(osd_invariant(obj));
1244
1245         LASSERT(obj->oo_owner != env);
1246         down_read_nested(&obj->oo_sem, role);
1247
1248         LASSERT(obj->oo_owner == NULL);
1249         oti->oti_r_locks++;
1250 }
1251
1252 static void osd_object_write_lock(const struct lu_env *env,
1253                                   struct dt_object *dt, unsigned role)
1254 {
1255         struct osd_object *obj = osd_dt_obj(dt);
1256         struct osd_thread_info *oti = osd_oti_get(env);
1257
1258         LINVRNT(osd_invariant(obj));
1259
1260         LASSERT(obj->oo_owner != env);
1261         down_write_nested(&obj->oo_sem, role);
1262
1263         LASSERT(obj->oo_owner == NULL);
1264         obj->oo_owner = env;
1265         oti->oti_w_locks++;
1266 }
1267
1268 static void osd_object_read_unlock(const struct lu_env *env,
1269                                    struct dt_object *dt)
1270 {
1271         struct osd_object *obj = osd_dt_obj(dt);
1272         struct osd_thread_info *oti = osd_oti_get(env);
1273
1274         LINVRNT(osd_invariant(obj));
1275
1276         LASSERT(oti->oti_r_locks > 0);
1277         oti->oti_r_locks--;
1278         up_read(&obj->oo_sem);
1279 }
1280
1281 static void osd_object_write_unlock(const struct lu_env *env,
1282                                     struct dt_object *dt)
1283 {
1284         struct osd_object *obj = osd_dt_obj(dt);
1285         struct osd_thread_info *oti = osd_oti_get(env);
1286
1287         LINVRNT(osd_invariant(obj));
1288
1289         LASSERT(obj->oo_owner == env);
1290         LASSERT(oti->oti_w_locks > 0);
1291         oti->oti_w_locks--;
1292         obj->oo_owner = NULL;
1293         up_write(&obj->oo_sem);
1294 }
1295
1296 static int osd_object_write_locked(const struct lu_env *env,
1297                                    struct dt_object *dt)
1298 {
1299         struct osd_object *obj = osd_dt_obj(dt);
1300
1301         LINVRNT(osd_invariant(obj));
1302
1303         return obj->oo_owner == env;
1304 }
1305
1306 static int capa_is_sane(const struct lu_env *env,
1307                         struct osd_device *dev,
1308                         struct lustre_capa *capa,
1309                         struct lustre_capa_key *keys)
1310 {
1311         struct osd_thread_info *oti = osd_oti_get(env);
1312         struct lustre_capa *tcapa = &oti->oti_capa;
1313         struct obd_capa *oc;
1314         int i, rc = 0;
1315         ENTRY;
1316
1317         oc = capa_lookup(dev->od_capa_hash, capa, 0);
1318         if (oc) {
1319                 if (capa_is_expired(oc)) {
1320                         DEBUG_CAPA(D_ERROR, capa, "expired");
1321                         rc = -ESTALE;
1322                 }
1323                 capa_put(oc);
1324                 RETURN(rc);
1325         }
1326
1327         if (capa_is_expired_sec(capa)) {
1328                 DEBUG_CAPA(D_ERROR, capa, "expired");
1329                 RETURN(-ESTALE);
1330         }
1331
1332         spin_lock(&capa_lock);
1333         for (i = 0; i < 2; i++) {
1334                 if (keys[i].lk_keyid == capa->lc_keyid) {
1335                         oti->oti_capa_key = keys[i];
1336                         break;
1337                 }
1338         }
1339         spin_unlock(&capa_lock);
1340
1341         if (i == 2) {
1342                 DEBUG_CAPA(D_ERROR, capa, "no matched capa key");
1343                 RETURN(-ESTALE);
1344         }
1345
1346         rc = capa_hmac(tcapa->lc_hmac, capa, oti->oti_capa_key.lk_key);
1347         if (rc)
1348                 RETURN(rc);
1349
1350         if (memcmp(tcapa->lc_hmac, capa->lc_hmac, sizeof(capa->lc_hmac))) {
1351                 DEBUG_CAPA(D_ERROR, capa, "HMAC mismatch");
1352                 RETURN(-EACCES);
1353         }
1354
1355         oc = capa_add(dev->od_capa_hash, capa);
1356         capa_put(oc);
1357
1358         RETURN(0);
1359 }
1360
1361 int osd_object_auth(const struct lu_env *env, struct dt_object *dt,
1362                     struct lustre_capa *capa, __u64 opc)
1363 {
1364         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
1365         struct osd_device *dev = osd_dev(dt->do_lu.lo_dev);
1366         struct md_capainfo *ci;
1367         int rc;
1368
1369         if (!dev->od_fl_capa)
1370                 return 0;
1371
1372         if (capa == BYPASS_CAPA)
1373                 return 0;
1374
1375         ci = md_capainfo(env);
1376         if (unlikely(!ci))
1377                 return 0;
1378
1379         if (ci->mc_auth == LC_ID_NONE)
1380                 return 0;
1381
1382         if (!capa) {
1383                 CERROR("no capability is provided for fid "DFID"\n", PFID(fid));
1384                 return -EACCES;
1385         }
1386
1387         if (!lu_fid_eq(fid, &capa->lc_fid)) {
1388                 DEBUG_CAPA(D_ERROR, capa, "fid "DFID" mismatch with",
1389                            PFID(fid));
1390                 return -EACCES;
1391         }
1392
1393         if (!capa_opc_supported(capa, opc)) {
1394                 DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
1395                 return -EACCES;
1396         }
1397
1398         if ((rc = capa_is_sane(env, dev, capa, dev->od_capa_keys))) {
1399                 DEBUG_CAPA(D_ERROR, capa, "insane (rc %d)", rc);
1400                 return -EACCES;
1401         }
1402
1403         return 0;
1404 }
1405
1406 static struct timespec *osd_inode_time(const struct lu_env *env,
1407                                        struct inode *inode, __u64 seconds)
1408 {
1409         struct osd_thread_info  *oti = osd_oti_get(env);
1410         struct timespec         *t   = &oti->oti_time;
1411
1412         t->tv_sec = seconds;
1413         t->tv_nsec = 0;
1414         *t = timespec_trunc(*t, inode->i_sb->s_time_gran);
1415         return t;
1416 }
1417
1418
1419 static void osd_inode_getattr(const struct lu_env *env,
1420                               struct inode *inode, struct lu_attr *attr)
1421 {
1422         attr->la_valid      |= LA_ATIME | LA_MTIME | LA_CTIME | LA_MODE |
1423                                LA_SIZE | LA_BLOCKS | LA_UID | LA_GID |
1424                                LA_FLAGS | LA_NLINK | LA_RDEV | LA_BLKSIZE |
1425                                LA_TYPE;
1426
1427         attr->la_atime      = LTIME_S(inode->i_atime);
1428         attr->la_mtime      = LTIME_S(inode->i_mtime);
1429         attr->la_ctime      = LTIME_S(inode->i_ctime);
1430         attr->la_mode       = inode->i_mode;
1431         attr->la_size       = i_size_read(inode);
1432         attr->la_blocks     = inode->i_blocks;
1433         attr->la_uid        = inode->i_uid;
1434         attr->la_gid        = inode->i_gid;
1435         attr->la_flags      = LDISKFS_I(inode)->i_flags;
1436         attr->la_nlink      = inode->i_nlink;
1437         attr->la_rdev       = inode->i_rdev;
1438         attr->la_blksize    = 1 << inode->i_blkbits;
1439         attr->la_blkbits    = inode->i_blkbits;
1440 }
1441
1442 static int osd_attr_get(const struct lu_env *env,
1443                         struct dt_object *dt,
1444                         struct lu_attr *attr,
1445                         struct lustre_capa *capa)
1446 {
1447         struct osd_object *obj = osd_dt_obj(dt);
1448
1449         LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
1450         LINVRNT(osd_invariant(obj));
1451
1452         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
1453                 return -EACCES;
1454
1455         spin_lock(&obj->oo_guard);
1456         osd_inode_getattr(env, obj->oo_inode, attr);
1457         spin_unlock(&obj->oo_guard);
1458         return 0;
1459 }
1460
1461 static int osd_declare_attr_set(const struct lu_env *env,
1462                                 struct dt_object *dt,
1463                                 const struct lu_attr *attr,
1464                                 struct thandle *handle)
1465 {
1466         struct osd_thandle     *oh;
1467         struct osd_object      *obj;
1468         struct osd_thread_info *info = osd_oti_get(env);
1469         struct lquota_id_info  *qi = &info->oti_qi;
1470         long long               bspace;
1471         int                     rc = 0;
1472         bool                    allocated;
1473         ENTRY;
1474
1475         LASSERT(dt != NULL);
1476         LASSERT(handle != NULL);
1477
1478         obj = osd_dt_obj(dt);
1479         LASSERT(osd_invariant(obj));
1480
1481         oh = container_of0(handle, struct osd_thandle, ot_super);
1482         LASSERT(oh->ot_handle == NULL);
1483
1484         osd_trans_declare_op(env, oh, OSD_OT_ATTR_SET,
1485                              osd_dto_credits_noquota[DTO_ATTR_SET_BASE]);
1486
1487         if (attr == NULL || obj->oo_inode == NULL)
1488                 RETURN(rc);
1489
1490         bspace   = obj->oo_inode->i_blocks;
1491         bspace <<= obj->oo_inode->i_sb->s_blocksize_bits;
1492         bspace   = toqb(bspace);
1493
1494         /* Changing ownership is always preformed by super user, it should not
1495          * fail with EDQUOT.
1496          *
1497          * We still need to call the osd_declare_qid() to calculate the journal
1498          * credits for updating quota accounting files and to trigger quota
1499          * space adjustment once the operation is completed.*/
1500         if ((attr->la_valid & LA_UID) != 0 &&
1501              attr->la_uid != obj->oo_inode->i_uid) {
1502                 qi->lqi_type = USRQUOTA;
1503
1504                 /* inode accounting */
1505                 qi->lqi_is_blk = false;
1506
1507                 /* one more inode for the new owner ... */
1508                 qi->lqi_id.qid_uid = attr->la_uid;
1509                 qi->lqi_space      = 1;
1510                 allocated = (attr->la_uid == 0) ? true : false;
1511                 rc = osd_declare_qid(env, oh, qi, allocated, NULL);
1512                 if (rc == -EDQUOT || rc == -EINPROGRESS)
1513                         rc = 0;
1514                 if (rc)
1515                         RETURN(rc);
1516
1517                 /* and one less inode for the current uid */
1518                 qi->lqi_id.qid_uid = obj->oo_inode->i_uid;
1519                 qi->lqi_space      = -1;
1520                 rc = osd_declare_qid(env, oh, qi, true, NULL);
1521                 if (rc == -EDQUOT || rc == -EINPROGRESS)
1522                         rc = 0;
1523                 if (rc)
1524                         RETURN(rc);
1525
1526                 /* block accounting */
1527                 qi->lqi_is_blk = true;
1528
1529                 /* more blocks for the new owner ... */
1530                 qi->lqi_id.qid_uid = attr->la_uid;
1531                 qi->lqi_space      = bspace;
1532                 allocated = (attr->la_uid == 0) ? true : false;
1533                 rc = osd_declare_qid(env, oh, qi, allocated, NULL);
1534                 if (rc == -EDQUOT || rc == -EINPROGRESS)
1535                         rc = 0;
1536                 if (rc)
1537                         RETURN(rc);
1538
1539                 /* and finally less blocks for the current owner */
1540                 qi->lqi_id.qid_uid = obj->oo_inode->i_uid;
1541                 qi->lqi_space      = -bspace;
1542                 rc = osd_declare_qid(env, oh, qi, true, NULL);
1543                 if (rc == -EDQUOT || rc == -EINPROGRESS)
1544                         rc = 0;
1545                 if (rc)
1546                         RETURN(rc);
1547         }
1548
1549         if (attr->la_valid & LA_GID &&
1550             attr->la_gid != obj->oo_inode->i_gid) {
1551                 qi->lqi_type = GRPQUOTA;
1552
1553                 /* inode accounting */
1554                 qi->lqi_is_blk = false;
1555
1556                 /* one more inode for the new group owner ... */
1557                 qi->lqi_id.qid_gid = attr->la_gid;
1558                 qi->lqi_space      = 1;
1559                 allocated = (attr->la_gid == 0) ? true : false;
1560                 rc = osd_declare_qid(env, oh, qi, allocated, NULL);
1561                 if (rc == -EDQUOT || rc == -EINPROGRESS)
1562                         rc = 0;
1563                 if (rc)
1564                         RETURN(rc);
1565
1566                 /* and one less inode for the current gid */
1567                 qi->lqi_id.qid_gid = obj->oo_inode->i_gid;
1568                 qi->lqi_space      = -1;
1569                 rc = osd_declare_qid(env, oh, qi, true, NULL);
1570                 if (rc == -EDQUOT || rc == -EINPROGRESS)
1571                         rc = 0;
1572                 if (rc)
1573                         RETURN(rc);
1574
1575                 /* block accounting */
1576                 qi->lqi_is_blk = true;
1577
1578                 /* more blocks for the new owner ... */
1579                 qi->lqi_id.qid_gid = attr->la_gid;
1580                 qi->lqi_space      = bspace;
1581                 allocated = (attr->la_gid == 0) ? true : false;
1582                 rc = osd_declare_qid(env, oh, qi, allocated, NULL);
1583                 if (rc == -EDQUOT || rc == -EINPROGRESS)
1584                         rc = 0;
1585                 if (rc)
1586                         RETURN(rc);
1587
1588                 /* and finally less blocks for the current owner */
1589                 qi->lqi_id.qid_gid = obj->oo_inode->i_gid;
1590                 qi->lqi_space      = -bspace;
1591                 rc = osd_declare_qid(env, oh, qi, true, NULL);
1592                 if (rc == -EDQUOT || rc == -EINPROGRESS)
1593                         rc = 0;
1594                 if (rc)
1595                         RETURN(rc);
1596         }
1597
1598         RETURN(rc);
1599 }
1600
1601 static int osd_inode_setattr(const struct lu_env *env,
1602                              struct inode *inode, const struct lu_attr *attr)
1603 {
1604         __u64 bits;
1605
1606         bits = attr->la_valid;
1607
1608         if (bits & LA_ATIME)
1609                 inode->i_atime  = *osd_inode_time(env, inode, attr->la_atime);
1610         if (bits & LA_CTIME)
1611                 inode->i_ctime  = *osd_inode_time(env, inode, attr->la_ctime);
1612         if (bits & LA_MTIME)
1613                 inode->i_mtime  = *osd_inode_time(env, inode, attr->la_mtime);
1614         if (bits & LA_SIZE) {
1615                 LDISKFS_I(inode)->i_disksize = attr->la_size;
1616                 i_size_write(inode, attr->la_size);
1617         }
1618
1619 #if 0
1620         /* OSD should not change "i_blocks" which is used by quota.
1621          * "i_blocks" should be changed by ldiskfs only. */
1622         if (bits & LA_BLOCKS)
1623                 inode->i_blocks = attr->la_blocks;
1624 #endif
1625         if (bits & LA_MODE)
1626                 inode->i_mode   = (inode->i_mode & S_IFMT) |
1627                         (attr->la_mode & ~S_IFMT);
1628         if (bits & LA_UID)
1629                 inode->i_uid    = attr->la_uid;
1630         if (bits & LA_GID)
1631                 inode->i_gid    = attr->la_gid;
1632         if (bits & LA_NLINK)
1633                 set_nlink(inode, attr->la_nlink);
1634         if (bits & LA_RDEV)
1635                 inode->i_rdev   = attr->la_rdev;
1636
1637         if (bits & LA_FLAGS) {
1638                 /* always keep S_NOCMTIME */
1639                 inode->i_flags = ll_ext_to_inode_flags(attr->la_flags) |
1640                                  S_NOCMTIME;
1641         }
1642         return 0;
1643 }
1644
1645 static int osd_quota_transfer(struct inode *inode, const struct lu_attr *attr)
1646 {
1647         if ((attr->la_valid & LA_UID && attr->la_uid != inode->i_uid) ||
1648             (attr->la_valid & LA_GID && attr->la_gid != inode->i_gid)) {
1649                 struct iattr    iattr;
1650                 int             rc;
1651
1652                 iattr.ia_valid = 0;
1653                 if (attr->la_valid & LA_UID)
1654                         iattr.ia_valid |= ATTR_UID;
1655                 if (attr->la_valid & LA_GID)
1656                         iattr.ia_valid |= ATTR_GID;
1657                 iattr.ia_uid = attr->la_uid;
1658                 iattr.ia_gid = attr->la_gid;
1659
1660                 rc = ll_vfs_dq_transfer(inode, &iattr);
1661                 if (rc) {
1662                         CERROR("%s: quota transfer failed: rc = %d. Is quota "
1663                                "enforcement enabled on the ldiskfs filesystem?",
1664                                inode->i_sb->s_id, rc);
1665                         return rc;
1666                 }
1667         }
1668         return 0;
1669 }
1670
1671 static int osd_attr_set(const struct lu_env *env,
1672                         struct dt_object *dt,
1673                         const struct lu_attr *attr,
1674                         struct thandle *handle,
1675                         struct lustre_capa *capa)
1676 {
1677         struct osd_object *obj = osd_dt_obj(dt);
1678         struct inode      *inode;
1679         int rc;
1680
1681         LASSERT(handle != NULL);
1682         LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
1683         LASSERT(osd_invariant(obj));
1684
1685         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
1686                 return -EACCES;
1687
1688         osd_trans_exec_op(env, handle, OSD_OT_ATTR_SET);
1689
1690         if (OBD_FAIL_CHECK(OBD_FAIL_OSD_FID_MAPPING)) {
1691                 struct osd_thread_info  *oti  = osd_oti_get(env);
1692                 const struct lu_fid     *fid0 = lu_object_fid(&dt->do_lu);
1693                 struct lu_fid           *fid1 = &oti->oti_fid;
1694                 struct osd_inode_id     *id   = &oti->oti_id;
1695                 struct iam_path_descr   *ipd;
1696                 struct iam_container    *bag;
1697                 struct osd_thandle      *oh;
1698                 int                      rc;
1699
1700                 fid_cpu_to_be(fid1, fid0);
1701                 memset(id, 1, sizeof(*id));
1702                 bag = &osd_fid2oi(osd_dev(dt->do_lu.lo_dev),
1703                                   fid0)->oi_dir.od_container;
1704                 ipd = osd_idx_ipd_get(env, bag);
1705                 if (unlikely(ipd == NULL))
1706                         RETURN(-ENOMEM);
1707
1708                 oh = container_of0(handle, struct osd_thandle, ot_super);
1709                 rc = iam_update(oh->ot_handle, bag, (const struct iam_key *)fid1,
1710                                 (const struct iam_rec *)id, ipd);
1711                 osd_ipd_put(env, bag, ipd);
1712                 return(rc > 0 ? 0 : rc);
1713         }
1714
1715         inode = obj->oo_inode;
1716         ll_vfs_dq_init(inode);
1717
1718         rc = osd_quota_transfer(inode, attr);
1719         if (rc)
1720                 return rc;
1721
1722         spin_lock(&obj->oo_guard);
1723         rc = osd_inode_setattr(env, inode, attr);
1724         spin_unlock(&obj->oo_guard);
1725
1726         if (!rc)
1727                 ll_dirty_inode(inode, I_DIRTY_DATASYNC);
1728         return rc;
1729 }
1730
1731 struct dentry *osd_child_dentry_get(const struct lu_env *env,
1732                                     struct osd_object *obj,
1733                                     const char *name, const int namelen)
1734 {
1735         return osd_child_dentry_by_inode(env, obj->oo_inode, name, namelen);
1736 }
1737
1738 static int osd_mkfile(struct osd_thread_info *info, struct osd_object *obj,
1739                       cfs_umode_t mode,
1740                       struct dt_allocation_hint *hint,
1741                       struct thandle *th)
1742 {
1743         int result;
1744         struct osd_device  *osd = osd_obj2dev(obj);
1745         struct osd_thandle *oth;
1746         struct dt_object   *parent = NULL;
1747         struct inode       *inode;
1748
1749         LINVRNT(osd_invariant(obj));
1750         LASSERT(obj->oo_inode == NULL);
1751         LASSERT(obj->oo_hl_head == NULL);
1752
1753         if (S_ISDIR(mode) && ldiskfs_pdo) {
1754                 obj->oo_hl_head =ldiskfs_htree_lock_head_alloc(HTREE_HBITS_DEF);
1755                 if (obj->oo_hl_head == NULL)
1756                         return -ENOMEM;
1757         }
1758
1759         oth = container_of(th, struct osd_thandle, ot_super);
1760         LASSERT(oth->ot_handle->h_transaction != NULL);
1761
1762         if (hint && hint->dah_parent)
1763                 parent = hint->dah_parent;
1764
1765         inode = ldiskfs_create_inode(oth->ot_handle,
1766                                      parent ? osd_dt_obj(parent)->oo_inode :
1767                                               osd_sb(osd)->s_root->d_inode,
1768                                      mode);
1769         if (!IS_ERR(inode)) {
1770                 /* Do not update file c/mtime in ldiskfs.
1771                  * NB: don't need any lock because no contention at this
1772                  * early stage */
1773                 inode->i_flags |= S_NOCMTIME;
1774
1775                 /* For new created object, it must be consistent,
1776                  * and it is unnecessary to scrub against it. */
1777                 ldiskfs_set_inode_state(inode, LDISKFS_STATE_LUSTRE_NOSCRUB);
1778                 obj->oo_inode = inode;
1779                 result = 0;
1780         } else {
1781                 if (obj->oo_hl_head != NULL) {
1782                         ldiskfs_htree_lock_head_free(obj->oo_hl_head);
1783                         obj->oo_hl_head = NULL;
1784                 }
1785                 result = PTR_ERR(inode);
1786         }
1787         LINVRNT(osd_invariant(obj));
1788         return result;
1789 }
1790
1791 enum {
1792         OSD_NAME_LEN = 255
1793 };
1794
1795 static int osd_mkdir(struct osd_thread_info *info, struct osd_object *obj,
1796                      struct lu_attr *attr,
1797                      struct dt_allocation_hint *hint,
1798                      struct dt_object_format *dof,
1799                      struct thandle *th)
1800 {
1801         int result;
1802         struct osd_thandle *oth;
1803         __u32 mode = (attr->la_mode & (S_IFMT | S_IRWXUGO | S_ISVTX));
1804
1805         LASSERT(S_ISDIR(attr->la_mode));
1806
1807         oth = container_of(th, struct osd_thandle, ot_super);
1808         LASSERT(oth->ot_handle->h_transaction != NULL);
1809         result = osd_mkfile(info, obj, mode, hint, th);
1810
1811         return result;
1812 }
1813
1814 static int osd_mk_index(struct osd_thread_info *info, struct osd_object *obj,
1815                         struct lu_attr *attr,
1816                         struct dt_allocation_hint *hint,
1817                         struct dt_object_format *dof,
1818                         struct thandle *th)
1819 {
1820         int result;
1821         struct osd_thandle *oth;
1822         const struct dt_index_features *feat = dof->u.dof_idx.di_feat;
1823
1824         __u32 mode = (attr->la_mode & (S_IFMT | S_IALLUGO | S_ISVTX));
1825
1826         LASSERT(S_ISREG(attr->la_mode));
1827
1828         oth = container_of(th, struct osd_thandle, ot_super);
1829         LASSERT(oth->ot_handle->h_transaction != NULL);
1830
1831         result = osd_mkfile(info, obj, mode, hint, th);
1832         if (result == 0) {
1833                 LASSERT(obj->oo_inode != NULL);
1834                 if (feat->dif_flags & DT_IND_VARKEY)
1835                         result = iam_lvar_create(obj->oo_inode,
1836                                                  feat->dif_keysize_max,
1837                                                  feat->dif_ptrsize,
1838                                                  feat->dif_recsize_max,
1839                                                  oth->ot_handle);
1840                 else
1841                         result = iam_lfix_create(obj->oo_inode,
1842                                                  feat->dif_keysize_max,
1843                                                  feat->dif_ptrsize,
1844                                                  feat->dif_recsize_max,
1845                                                  oth->ot_handle);
1846
1847         }
1848         return result;
1849 }
1850
1851 static int osd_mkreg(struct osd_thread_info *info, struct osd_object *obj,
1852                      struct lu_attr *attr,
1853                      struct dt_allocation_hint *hint,
1854                      struct dt_object_format *dof,
1855                      struct thandle *th)
1856 {
1857         LASSERT(S_ISREG(attr->la_mode));
1858         return osd_mkfile(info, obj, (attr->la_mode &
1859                                (S_IFMT | S_IALLUGO | S_ISVTX)), hint, th);
1860 }
1861
1862 static int osd_mksym(struct osd_thread_info *info, struct osd_object *obj,
1863                      struct lu_attr *attr,
1864                      struct dt_allocation_hint *hint,
1865                      struct dt_object_format *dof,
1866                      struct thandle *th)
1867 {
1868         LASSERT(S_ISLNK(attr->la_mode));
1869         return osd_mkfile(info, obj, (attr->la_mode &
1870                               (S_IFMT | S_IALLUGO | S_ISVTX)), hint, th);
1871 }
1872
1873 static int osd_mknod(struct osd_thread_info *info, struct osd_object *obj,
1874                      struct lu_attr *attr,
1875                      struct dt_allocation_hint *hint,
1876                      struct dt_object_format *dof,
1877                      struct thandle *th)
1878 {
1879         cfs_umode_t mode = attr->la_mode & (S_IFMT | S_IALLUGO | S_ISVTX);
1880         int result;
1881
1882         LINVRNT(osd_invariant(obj));
1883         LASSERT(obj->oo_inode == NULL);
1884         LASSERT(S_ISCHR(mode) || S_ISBLK(mode) ||
1885                 S_ISFIFO(mode) || S_ISSOCK(mode));
1886
1887         result = osd_mkfile(info, obj, mode, hint, th);
1888         if (result == 0) {
1889                 LASSERT(obj->oo_inode != NULL);
1890                 /*
1891                  * This inode should be marked dirty for i_rdev.  Currently
1892                  * that is done in the osd_attr_init().
1893                  */
1894                 init_special_inode(obj->oo_inode, obj->oo_inode->i_mode,
1895                                    attr->la_rdev);
1896         }
1897         LINVRNT(osd_invariant(obj));
1898         return result;
1899 }
1900
1901 typedef int (*osd_obj_type_f)(struct osd_thread_info *, struct osd_object *,
1902                               struct lu_attr *,
1903                               struct dt_allocation_hint *hint,
1904                               struct dt_object_format *dof,
1905                               struct thandle *);
1906
1907 static osd_obj_type_f osd_create_type_f(enum dt_format_type type)
1908 {
1909         osd_obj_type_f result;
1910
1911         switch (type) {
1912         case DFT_DIR:
1913                 result = osd_mkdir;
1914                 break;
1915         case DFT_REGULAR:
1916                 result = osd_mkreg;
1917                 break;
1918         case DFT_SYM:
1919                 result = osd_mksym;
1920                 break;
1921         case DFT_NODE:
1922                 result = osd_mknod;
1923                 break;
1924         case DFT_INDEX:
1925                 result = osd_mk_index;
1926                 break;
1927
1928         default:
1929                 LBUG();
1930                 break;
1931         }
1932         return result;
1933 }
1934
1935
1936 static void osd_ah_init(const struct lu_env *env, struct dt_allocation_hint *ah,
1937                         struct dt_object *parent, struct dt_object *child,
1938                         cfs_umode_t child_mode)
1939 {
1940         LASSERT(ah);
1941
1942         memset(ah, 0, sizeof(*ah));
1943         ah->dah_parent = parent;
1944         ah->dah_mode = child_mode;
1945 }
1946
1947 static void osd_attr_init(struct osd_thread_info *info, struct osd_object *obj,
1948                           struct lu_attr *attr, struct dt_object_format *dof)
1949 {
1950         struct inode   *inode = obj->oo_inode;
1951         __u64           valid = attr->la_valid;
1952         int             result;
1953
1954         attr->la_valid &= ~(LA_TYPE | LA_MODE);
1955
1956         if (dof->dof_type != DFT_NODE)
1957                 attr->la_valid &= ~LA_RDEV;
1958         if ((valid & LA_ATIME) && (attr->la_atime == LTIME_S(inode->i_atime)))
1959                 attr->la_valid &= ~LA_ATIME;
1960         if ((valid & LA_CTIME) && (attr->la_ctime == LTIME_S(inode->i_ctime)))
1961                 attr->la_valid &= ~LA_CTIME;
1962         if ((valid & LA_MTIME) && (attr->la_mtime == LTIME_S(inode->i_mtime)))
1963                 attr->la_valid &= ~LA_MTIME;
1964
1965         result = osd_quota_transfer(inode, attr);
1966         if (result)
1967                 return;
1968
1969         if (attr->la_valid != 0) {
1970                 result = osd_inode_setattr(info->oti_env, inode, attr);
1971                 /*
1972                  * The osd_inode_setattr() should always succeed here.  The
1973                  * only error that could be returned is EDQUOT when we are
1974                  * trying to change the UID or GID of the inode. However, this
1975                  * should not happen since quota enforcement is no longer
1976                  * enabled on ldiskfs (lquota takes care of it).
1977                  */
1978                 LASSERTF(result == 0, "%d", result);
1979                 ll_dirty_inode(inode, I_DIRTY_DATASYNC);
1980         }
1981
1982         attr->la_valid = valid;
1983 }
1984
1985 /**
1986  * Helper function for osd_object_create()
1987  *
1988  * \retval 0, on success
1989  */
1990 static int __osd_object_create(struct osd_thread_info *info,
1991                                struct osd_object *obj, struct lu_attr *attr,
1992                                struct dt_allocation_hint *hint,
1993                                struct dt_object_format *dof,
1994                                struct thandle *th)
1995 {
1996         int     result;
1997         __u32   umask;
1998
1999         /* we drop umask so that permissions we pass are not affected */
2000         umask = current->fs->umask;
2001         current->fs->umask = 0;
2002
2003         result = osd_create_type_f(dof->dof_type)(info, obj, attr, hint, dof,
2004                                                   th);
2005         if (result == 0) {
2006                 osd_attr_init(info, obj, attr, dof);
2007                 osd_object_init0(obj);
2008                 /* bz 24037 */
2009                 if (obj->oo_inode && (obj->oo_inode->i_state & I_NEW))
2010                         unlock_new_inode(obj->oo_inode);
2011         }
2012
2013         /* restore previous umask value */
2014         current->fs->umask = umask;
2015
2016         return result;
2017 }
2018
2019 /**
2020  * Helper function for osd_object_create()
2021  *
2022  * \retval 0, on success
2023  */
2024 static int __osd_oi_insert(const struct lu_env *env, struct osd_object *obj,
2025                            const struct lu_fid *fid, struct thandle *th)
2026 {
2027         struct osd_thread_info *info = osd_oti_get(env);
2028         struct osd_inode_id    *id   = &info->oti_id;
2029         struct osd_device      *osd  = osd_obj2dev(obj);
2030
2031         LASSERT(obj->oo_inode != NULL);
2032
2033         osd_id_gen(id, obj->oo_inode->i_ino, obj->oo_inode->i_generation);
2034         return osd_oi_insert(info, osd, fid, id, th);
2035 }
2036
2037 int osd_fld_lookup(const struct lu_env *env, struct osd_device *osd,
2038                    const struct lu_fid *fid, struct lu_seq_range *range)
2039 {
2040         struct seq_server_site  *ss = osd_seq_site(osd);
2041         int                     rc;
2042
2043         if (fid_is_idif(fid)) {
2044                 fld_range_set_ost(range);
2045                 range->lsr_index = fid_idif_ost_idx(fid);
2046                 return 0;
2047         }
2048
2049         if (!fid_seq_in_fldb(fid_seq(fid))) {
2050                 fld_range_set_mdt(range);
2051                 if (ss != NULL)
2052                         /* FIXME: If ss is NULL, it suppose not get lsr_index
2053                          * at all */
2054                         range->lsr_index = ss->ss_node_id;
2055                 return 0;
2056         }
2057
2058         LASSERT(ss != NULL);
2059         fld_range_set_any(range);
2060         rc = fld_server_lookup(env, ss->ss_server_fld, fid_seq(fid), range);
2061         if (rc != 0) {
2062                 CERROR("%s: cannot find FLD range for "DFID": rc = %d\n",
2063                        osd_name(osd), PFID(fid), rc);
2064         }
2065         return rc;
2066 }
2067
2068 /*
2069  * Concurrency: no external locking is necessary.
2070  */
2071 static int osd_declare_object_create(const struct lu_env *env,
2072                                      struct dt_object *dt,
2073                                      struct lu_attr *attr,
2074                                      struct dt_allocation_hint *hint,
2075                                      struct dt_object_format *dof,
2076                                      struct thandle *handle)
2077 {
2078         struct lu_seq_range     *range = &osd_oti_get(env)->oti_seq_range;
2079         struct osd_thandle      *oh;
2080         int                      rc;
2081         ENTRY;
2082
2083         LASSERT(handle != NULL);
2084
2085         oh = container_of0(handle, struct osd_thandle, ot_super);
2086         LASSERT(oh->ot_handle == NULL);
2087
2088         osd_trans_declare_op(env, oh, OSD_OT_CREATE,
2089                              osd_dto_credits_noquota[DTO_OBJECT_CREATE]);
2090         if (!fid_is_on_ost(osd_oti_get(env), osd_dt_dev(handle->th_dev),
2091                            lu_object_fid(&dt->do_lu)))
2092                 /* Reuse idle OI block may cause additional one OI block
2093                  * to be changed. */
2094                 osd_trans_declare_op(env, oh, OSD_OT_INSERT,
2095                                 osd_dto_credits_noquota[DTO_INDEX_INSERT] + 1);
2096
2097         /* If this is directory, then we expect . and .. to be inserted as
2098          * well. The one directory block always needs to be created for the
2099          * directory, so we could use DTO_WRITE_BASE here (GDT, block bitmap,
2100          * block), there is no danger of needing a tree for the first block.
2101          */
2102         if (attr && S_ISDIR(attr->la_mode)) {
2103                 osd_trans_declare_op(env, oh, OSD_OT_INSERT,
2104                                      osd_dto_credits_noquota[DTO_WRITE_BASE]);
2105                 osd_trans_declare_op(env, oh, OSD_OT_INSERT, 0);
2106         }
2107
2108         if (!attr)
2109                 RETURN(0);
2110
2111         rc = osd_declare_inode_qid(env, attr->la_uid, attr->la_gid, 1, oh,
2112                                    false, false, NULL, false);
2113         if (rc != 0)
2114                 RETURN(rc);
2115
2116         /* It does fld look up inside declare, and the result will be
2117          * added to fld cache, so the following fld lookup inside insert
2118          * does not need send RPC anymore, so avoid send rpc with holding
2119          * transaction */
2120         if (fid_is_norm(lu_object_fid(&dt->do_lu)) &&
2121                 !fid_is_last_id(lu_object_fid(&dt->do_lu)))
2122                 osd_fld_lookup(env, osd_dt_dev(handle->th_dev),
2123                                lu_object_fid(&dt->do_lu), range);
2124
2125
2126         RETURN(rc);
2127 }
2128
2129 static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
2130                              struct lu_attr *attr,
2131                              struct dt_allocation_hint *hint,
2132                              struct dt_object_format *dof,
2133                              struct thandle *th)
2134 {
2135         const struct lu_fid    *fid    = lu_object_fid(&dt->do_lu);
2136         struct osd_object      *obj    = osd_dt_obj(dt);
2137         struct osd_thread_info *info   = osd_oti_get(env);
2138         int result;
2139
2140         ENTRY;
2141
2142         LINVRNT(osd_invariant(obj));
2143         LASSERT(!dt_object_exists(dt) && !dt_object_remote(dt));
2144         LASSERT(osd_write_locked(env, obj));
2145         LASSERT(th != NULL);
2146
2147         if (unlikely(fid_is_acct(fid)))
2148                 /* Quota files can't be created from the kernel any more,
2149                  * 'tune2fs -O quota' will take care of creating them */
2150                 RETURN(-EPERM);
2151
2152         osd_trans_exec_op(env, th, OSD_OT_CREATE);
2153         osd_trans_declare_rb(env, th, OSD_OT_REF_ADD);
2154
2155         result = __osd_object_create(info, obj, attr, hint, dof, th);
2156         if (result == 0)
2157                 result = __osd_oi_insert(env, obj, fid, th);
2158
2159         LASSERT(ergo(result == 0,
2160                      dt_object_exists(dt) && !dt_object_remote(dt)));
2161
2162         LASSERT(osd_invariant(obj));
2163         RETURN(result);
2164 }
2165
2166 /**
2167  * Called to destroy on-disk representation of the object
2168  *
2169  * Concurrency: must be locked
2170  */
2171 static int osd_declare_object_destroy(const struct lu_env *env,
2172                                       struct dt_object *dt,
2173                                       struct thandle *th)
2174 {
2175         struct osd_object  *obj = osd_dt_obj(dt);
2176         struct inode       *inode = obj->oo_inode;
2177         struct osd_thandle *oh;
2178         int                 rc;
2179         ENTRY;
2180
2181         oh = container_of0(th, struct osd_thandle, ot_super);
2182         LASSERT(oh->ot_handle == NULL);
2183         LASSERT(inode);
2184
2185         osd_trans_declare_op(env, oh, OSD_OT_DESTROY,
2186                              osd_dto_credits_noquota[DTO_OBJECT_DELETE]);
2187         /* Recycle idle OI leaf may cause additional three OI blocks
2188          * to be changed. */
2189         osd_trans_declare_op(env, oh, OSD_OT_DELETE,
2190                              osd_dto_credits_noquota[DTO_INDEX_DELETE] + 3);
2191         /* one less inode */
2192         rc = osd_declare_inode_qid(env, inode->i_uid, inode->i_gid, -1, oh,
2193                                    false, true, NULL, false);
2194         if (rc)
2195                 RETURN(rc);
2196         /* data to be truncated */
2197         rc = osd_declare_inode_qid(env, inode->i_uid, inode->i_gid, 0, oh,
2198                                    true, true, NULL, false);
2199         RETURN(rc);
2200 }
2201
2202 static int osd_object_destroy(const struct lu_env *env,
2203                               struct dt_object *dt,
2204                               struct thandle *th)
2205 {
2206         const struct lu_fid    *fid = lu_object_fid(&dt->do_lu);
2207         struct osd_object      *obj = osd_dt_obj(dt);
2208         struct inode           *inode = obj->oo_inode;
2209         struct osd_device      *osd = osd_obj2dev(obj);
2210         struct osd_thandle     *oh;
2211         int                     result;
2212         ENTRY;
2213
2214         oh = container_of0(th, struct osd_thandle, ot_super);
2215         LASSERT(oh->ot_handle);
2216         LASSERT(inode);
2217         LASSERT(!lu_object_is_dying(dt->do_lu.lo_header));
2218
2219         if (unlikely(fid_is_acct(fid)))
2220                 RETURN(-EPERM);
2221
2222         if (S_ISDIR(inode->i_mode)) {
2223                 LASSERT(osd_inode_unlinked(inode) || inode->i_nlink == 1);
2224                 /* it will check/delete the inode from remote parent,
2225                  * how to optimize it? unlink performance impaction XXX */
2226                 result = osd_delete_from_remote_parent(env, osd, obj, oh);
2227                 if (result != 0 && result != -ENOENT) {
2228                         CERROR("%s: delete inode "DFID": rc = %d\n",
2229                                osd_name(osd), PFID(fid), result);
2230                 }
2231                 spin_lock(&obj->oo_guard);
2232                 clear_nlink(inode);
2233                 spin_unlock(&obj->oo_guard);
2234                 ll_dirty_inode(inode, I_DIRTY_DATASYNC);
2235         }
2236
2237         osd_trans_exec_op(env, th, OSD_OT_DESTROY);
2238
2239         result = osd_oi_delete(osd_oti_get(env), osd, fid, th);
2240
2241         /* XXX: add to ext3 orphan list */
2242         /* rc = ext3_orphan_add(handle_t *handle, struct inode *inode) */
2243
2244         /* not needed in the cache anymore */
2245         set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
2246
2247         RETURN(0);
2248 }
2249
2250 /**
2251  * Put the fid into lustre_mdt_attrs, and then place the structure
2252  * inode's ea. This fid should not be altered during the life time
2253  * of the inode.
2254  *
2255  * \retval +ve, on success
2256  * \retval -ve, on error
2257  *
2258  * FIXME: It is good to have/use ldiskfs_xattr_set_handle() here
2259  */
2260 int osd_ea_fid_set(struct osd_thread_info *info, struct inode *inode,
2261                    const struct lu_fid *fid, __u64 flags)
2262 {
2263         struct lustre_mdt_attrs *lma = &info->oti_mdt_attrs;
2264         int                      rc;
2265
2266         if (OBD_FAIL_CHECK(OBD_FAIL_FID_INLMA))
2267                 return 0;
2268
2269         lustre_lma_init(lma, fid, flags);
2270         lustre_lma_swab(lma);
2271
2272         rc = __osd_xattr_set(info, inode, XATTR_NAME_LMA, lma, sizeof(*lma),
2273                              XATTR_CREATE);
2274         /* Someone may created the EA by race. */
2275         if (unlikely(rc == -EEXIST))
2276                 rc = 0;
2277         return rc;
2278 }
2279
2280 /**
2281  * ldiskfs supports fid in dirent, it is passed in dentry->d_fsdata.
2282  * lustre 1.8 also uses d_fsdata for passing other info to ldiskfs.
2283  * To have compatilibility with 1.8 ldiskfs driver we need to have
2284  * magic number at start of fid data.
2285  * \ldiskfs_dentry_param is used only to pass fid from osd to ldiskfs.
2286  * its inmemory API.
2287  */
2288 void osd_get_ldiskfs_dirent_param(struct ldiskfs_dentry_param *param,
2289                                   const struct dt_rec *fid)
2290 {
2291         if (!fid_is_namespace_visible((const struct lu_fid *)fid) ||
2292             OBD_FAIL_CHECK(OBD_FAIL_FID_IGIF)) {
2293                 param->edp_magic = 0;
2294                 return;
2295         }
2296
2297         param->edp_magic = LDISKFS_LUFID_MAGIC;
2298         param->edp_len =  sizeof(struct lu_fid) + 1;
2299         fid_cpu_to_be((struct lu_fid *)param->edp_data, (struct lu_fid *)fid);
2300 }
2301
2302 /**
2303  * Try to read the fid from inode ea into dt_rec.
2304  *
2305  * \param fid object fid.
2306  *
2307  * \retval 0 on success
2308  */
2309 static int osd_ea_fid_get(const struct lu_env *env, struct osd_object *obj,
2310                           __u32 ino, struct lu_fid *fid,
2311                           struct osd_inode_id *id)
2312 {
2313         struct osd_thread_info *info  = osd_oti_get(env);
2314         struct inode           *inode;
2315         ENTRY;
2316
2317         osd_id_gen(id, ino, OSD_OII_NOGEN);
2318         inode = osd_iget_fid(info, osd_obj2dev(obj), id, fid);
2319         if (IS_ERR(inode))
2320                 RETURN(PTR_ERR(inode));
2321
2322         iput(inode);
2323         RETURN(0);
2324 }
2325
2326 static int osd_add_dot_dotdot_internal(struct osd_thread_info *info,
2327                                         struct inode *dir,
2328                                         struct inode  *parent_dir,
2329                                         const struct dt_rec *dot_fid,
2330                                         const struct dt_rec *dot_dot_fid,
2331                                         struct osd_thandle *oth)
2332 {
2333         struct ldiskfs_dentry_param *dot_ldp;
2334         struct ldiskfs_dentry_param *dot_dot_ldp;
2335
2336         dot_dot_ldp = (struct ldiskfs_dentry_param *)info->oti_ldp2;
2337         osd_get_ldiskfs_dirent_param(dot_dot_ldp, dot_dot_fid);
2338
2339         dot_ldp = (struct ldiskfs_dentry_param *)info->oti_ldp;
2340         dot_ldp->edp_magic = 0;
2341         return ldiskfs_add_dot_dotdot(oth->ot_handle, parent_dir,
2342                                         dir, dot_ldp, dot_dot_ldp);
2343 }
2344
2345 /**
2346  * Create an local agent inode for remote entry
2347  */
2348 static struct inode *osd_create_local_agent_inode(const struct lu_env *env,
2349                                                   struct osd_device *osd,
2350                                                   struct osd_object *pobj,
2351                                                   const struct lu_fid *fid,
2352                                                   struct thandle *th)
2353 {
2354         struct osd_thread_info  *info = osd_oti_get(env);
2355         struct inode            *local;
2356         struct osd_thandle      *oh;
2357         int                     rc;
2358         ENTRY;
2359
2360         LASSERT(th);
2361         oh = container_of(th, struct osd_thandle, ot_super);
2362         LASSERT(oh->ot_handle->h_transaction != NULL);
2363
2364         /* FIXME: Insert index api needs to know the mode of
2365          * the remote object. Just use S_IFDIR for now */
2366         local = ldiskfs_create_inode(oh->ot_handle, pobj->oo_inode, S_IFDIR);
2367         if (IS_ERR(local)) {
2368                 CERROR("%s: create local error %d\n", osd_name(osd),
2369                        (int)PTR_ERR(local));
2370                 RETURN(local);
2371         }
2372
2373         /* Set special LMA flag for local agent inode */
2374         rc = osd_ea_fid_set(info, local, fid, LMAI_AGENT);
2375         if (rc != 0) {
2376                 CERROR("%s: set LMA for "DFID" remote inode failed: rc = %d\n",
2377                        osd_name(osd), PFID(fid), rc);
2378                 RETURN(ERR_PTR(rc));
2379         }
2380
2381         rc = osd_add_dot_dotdot_internal(info, local, pobj->oo_inode,
2382                 (const struct dt_rec *)lu_object_fid(&pobj->oo_dt.do_lu),
2383                 (const struct dt_rec *)fid, oh);
2384         if (rc != 0) {
2385                 CERROR("%s: "DFID" add dot dotdot error: rc = %d\n",
2386                         osd_name(osd), PFID(fid), rc);
2387                 RETURN(ERR_PTR(rc));
2388         }
2389
2390         RETURN(local);
2391 }
2392
2393 /**
2394  * Delete local agent inode for remote entry
2395  */
2396 static int osd_delete_local_agent_inode(const struct lu_env *env,
2397                                         struct osd_device *osd,
2398                                         const struct lu_fid *fid,
2399                                         __u32 ino, struct osd_thandle *oh)
2400 {
2401         struct osd_thread_info  *oti = osd_oti_get(env);
2402         struct osd_inode_id     *id = &oti->oti_id;
2403         struct inode            *inode;
2404         ENTRY;
2405
2406         id->oii_ino = le32_to_cpu(ino);
2407         id->oii_gen = OSD_OII_NOGEN;
2408         inode = osd_iget(oti, osd, id);
2409         if (IS_ERR(inode)) {
2410                 CERROR("%s: iget error "DFID" id %u:%u\n", osd_name(osd),
2411                        PFID(fid), id->oii_ino, id->oii_gen);
2412                 RETURN(PTR_ERR(inode));
2413         }
2414
2415         clear_nlink(inode);
2416         mark_inode_dirty(inode);
2417         CDEBUG(D_INODE, "%s: delete remote inode "DFID" %lu\n",
2418                 osd_name(osd), PFID(fid), inode->i_ino);
2419         iput(inode);
2420         RETURN(0);
2421 }
2422
2423 /**
2424  * OSD layer object create function for interoperability mode (b11826).
2425  * This is mostly similar to osd_object_create(). Only difference being, fid is
2426  * inserted into inode ea here.
2427  *
2428  * \retval   0, on success
2429  * \retval -ve, on error
2430  */
2431 static int osd_object_ea_create(const struct lu_env *env, struct dt_object *dt,
2432                                 struct lu_attr *attr,
2433                                 struct dt_allocation_hint *hint,
2434                                 struct dt_object_format *dof,
2435                                 struct thandle *th)
2436 {
2437         const struct lu_fid    *fid    = lu_object_fid(&dt->do_lu);
2438         struct osd_object      *obj    = osd_dt_obj(dt);
2439         struct osd_thread_info *info   = osd_oti_get(env);
2440         int                     result;
2441
2442         ENTRY;
2443
2444         LASSERT(osd_invariant(obj));
2445         LASSERT(!dt_object_exists(dt) && !dt_object_remote(dt));
2446         LASSERT(osd_write_locked(env, obj));
2447         LASSERT(th != NULL);
2448
2449         if (unlikely(fid_is_acct(fid)))
2450                 /* Quota files can't be created from the kernel any more,
2451                  * 'tune2fs -O quota' will take care of creating them */
2452                 RETURN(-EPERM);
2453
2454         osd_trans_exec_op(env, th, OSD_OT_CREATE);
2455         osd_trans_declare_rb(env, th, OSD_OT_REF_ADD);
2456
2457         result = __osd_object_create(info, obj, attr, hint, dof, th);
2458         if (result == 0)
2459                 result = osd_ea_fid_set(info, obj->oo_inode, fid, 0);
2460
2461         if (result == 0)
2462                 result = __osd_oi_insert(env, obj, fid, th);
2463
2464         LASSERT(ergo(result == 0,
2465                      dt_object_exists(dt) && !dt_object_remote(dt)));
2466         LINVRNT(osd_invariant(obj));
2467         RETURN(result);
2468 }
2469
2470 static int osd_declare_object_ref_add(const struct lu_env *env,
2471                                       struct dt_object *dt,
2472                                       struct thandle *handle)
2473 {
2474         struct osd_thandle       *oh;
2475
2476         /* it's possible that object doesn't exist yet */
2477         LASSERT(handle != NULL);
2478
2479         oh = container_of0(handle, struct osd_thandle, ot_super);
2480         LASSERT(oh->ot_handle == NULL);
2481
2482         osd_trans_declare_op(env, oh, OSD_OT_REF_ADD,
2483                              osd_dto_credits_noquota[DTO_ATTR_SET_BASE]);
2484
2485         return 0;
2486 }
2487
2488 /*
2489  * Concurrency: @dt is write locked.
2490  */
2491 static int osd_object_ref_add(const struct lu_env *env,
2492                               struct dt_object *dt, struct thandle *th)
2493 {
2494         struct osd_object *obj = osd_dt_obj(dt);
2495         struct inode      *inode = obj->oo_inode;
2496         bool               need_dirty = false;
2497         int                rc = 0;
2498
2499         LINVRNT(osd_invariant(obj));
2500         LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
2501         LASSERT(osd_write_locked(env, obj));
2502         LASSERT(th != NULL);
2503
2504         osd_trans_exec_op(env, th, OSD_OT_REF_ADD);
2505
2506         /* This based on ldiskfs_inc_count(), which is not exported.
2507          *
2508          * The DIR_NLINK feature allows directories to exceed LDISKFS_LINK_MAX
2509          * (65000) subdirectories by storing "1" in i_nlink if the link count
2510          * would otherwise overflow. Directory tranversal tools understand
2511          * that (st_nlink == 1) indicates that the filesystem dose not track
2512          * hard links count on the directory, and will not abort subdirectory
2513          * scanning early once (st_nlink - 2) subdirs have been found.
2514          *
2515          * This also has to properly handle the case of inodes with nlink == 0
2516          * in case they are being linked into the PENDING directory
2517          */
2518         spin_lock(&obj->oo_guard);
2519         if (unlikely(!S_ISDIR(inode->i_mode) &&
2520                      inode->i_nlink >= LDISKFS_LINK_MAX)) {
2521                 /* MDD should have checked this, but good to be safe */
2522                 rc = -EMLINK;
2523         } else if (unlikely(inode->i_nlink == 0 ||
2524                             (S_ISDIR(inode->i_mode) &&
2525                              inode->i_nlink >= LDISKFS_LINK_MAX))) {
2526                 /* inc_nlink from 0 may cause WARN_ON */
2527                 set_nlink(inode, 1);
2528                 need_dirty = true;
2529         } else if (!S_ISDIR(inode->i_mode) ||
2530                    (S_ISDIR(inode->i_mode) && inode->i_nlink >= 2)) {
2531                 inc_nlink(inode);
2532                 need_dirty = true;
2533         } /* else (S_ISDIR(inode->i_mode) && inode->i_nlink == 1) { ; } */
2534
2535         LASSERT(inode->i_nlink <= LDISKFS_LINK_MAX);
2536         spin_unlock(&obj->oo_guard);
2537
2538         if (need_dirty)
2539                 ll_dirty_inode(inode, I_DIRTY_DATASYNC);
2540
2541         LINVRNT(osd_invariant(obj));
2542
2543         return rc;
2544 }
2545
2546 static int osd_declare_object_ref_del(const struct lu_env *env,
2547                                       struct dt_object *dt,
2548                                       struct thandle *handle)
2549 {
2550         struct osd_thandle *oh;
2551
2552         LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
2553         LASSERT(handle != NULL);
2554
2555         oh = container_of0(handle, struct osd_thandle, ot_super);
2556         LASSERT(oh->ot_handle == NULL);
2557
2558         osd_trans_declare_op(env, oh, OSD_OT_REF_DEL,
2559                              osd_dto_credits_noquota[DTO_ATTR_SET_BASE]);
2560
2561         return 0;
2562 }
2563
2564 /*
2565  * Concurrency: @dt is write locked.
2566  */
2567 static int osd_object_ref_del(const struct lu_env *env, struct dt_object *dt,
2568                               struct thandle *th)
2569 {
2570         struct osd_object *obj = osd_dt_obj(dt);
2571         struct inode      *inode = obj->oo_inode;
2572
2573         LINVRNT(osd_invariant(obj));
2574         LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
2575         LASSERT(osd_write_locked(env, obj));
2576         LASSERT(th != NULL);
2577
2578         osd_trans_exec_op(env, th, OSD_OT_REF_DEL);
2579
2580         spin_lock(&obj->oo_guard);
2581         LASSERT(inode->i_nlink > 0);
2582
2583         /* This based on ldiskfs_dec_count(), which is not exported.
2584          *
2585          * If a directory already has nlink == 1, then do not drop the nlink
2586          * count to 0, even temporarily, to avoid race conditions with other
2587          * threads not holding oo_guard seeing i_nlink == 0 in rare cases.
2588          *
2589          * nlink == 1 means the directory has/had > EXT4_LINK_MAX subdirs.
2590          * */
2591         if (!S_ISDIR(inode->i_mode) || inode->i_nlink > 1) {
2592                 drop_nlink(inode);
2593
2594                 spin_unlock(&obj->oo_guard);
2595                 ll_dirty_inode(inode, I_DIRTY_DATASYNC);
2596                 LINVRNT(osd_invariant(obj));
2597         } else {
2598                 spin_unlock(&obj->oo_guard);
2599         }
2600
2601         return 0;
2602 }
2603
2604 /*
2605  * Get the 64-bit version for an inode.
2606  */
2607 static int osd_object_version_get(const struct lu_env *env,
2608                                   struct dt_object *dt, dt_obj_version_t *ver)
2609 {
2610         struct inode *inode = osd_dt_obj(dt)->oo_inode;
2611
2612         CDEBUG(D_INODE, "Get version "LPX64" for inode %lu\n",
2613                LDISKFS_I(inode)->i_fs_version, inode->i_ino);
2614         *ver = LDISKFS_I(inode)->i_fs_version;
2615         return 0;
2616 }
2617
2618 /*
2619  * Concurrency: @dt is read locked.
2620  */
2621 static int osd_xattr_get(const struct lu_env *env, struct dt_object *dt,
2622                          struct lu_buf *buf, const char *name,
2623                          struct lustre_capa *capa)
2624 {
2625         struct osd_object      *obj    = osd_dt_obj(dt);
2626         struct inode           *inode  = obj->oo_inode;
2627         struct osd_thread_info *info   = osd_oti_get(env);
2628         struct dentry          *dentry = &info->oti_obj_dentry;
2629
2630         /* version get is not real XATTR but uses xattr API */
2631         if (strcmp(name, XATTR_NAME_VERSION) == 0) {
2632                 /* for version we are just using xattr API but change inode
2633                  * field instead */
2634                 LASSERT(buf->lb_len == sizeof(dt_obj_version_t));
2635                 osd_object_version_get(env, dt, buf->lb_buf);
2636                 return sizeof(dt_obj_version_t);
2637         }
2638
2639         LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
2640         LASSERT(inode->i_op != NULL && inode->i_op->getxattr != NULL);
2641
2642         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
2643                 return -EACCES;
2644
2645         return __osd_xattr_get(inode, dentry, name, buf->lb_buf, buf->lb_len);
2646 }
2647
2648
2649 static int osd_declare_xattr_set(const struct lu_env *env,
2650                                  struct dt_object *dt,
2651                                  const struct lu_buf *buf, const char *name,
2652                                  int fl, struct thandle *handle)
2653 {
2654         struct osd_thandle *oh;
2655
2656         LASSERT(handle != NULL);
2657
2658         oh = container_of0(handle, struct osd_thandle, ot_super);
2659         LASSERT(oh->ot_handle == NULL);
2660
2661         osd_trans_declare_op(env, oh, OSD_OT_XATTR_SET,
2662                              strcmp(name, XATTR_NAME_VERSION) == 0 ?
2663                              osd_dto_credits_noquota[DTO_ATTR_SET_BASE] :
2664                              osd_dto_credits_noquota[DTO_XATTR_SET]);
2665
2666         return 0;
2667 }
2668
2669 /*
2670  * Set the 64-bit version for object
2671  */
2672 static void osd_object_version_set(const struct lu_env *env,
2673                                    struct dt_object *dt,
2674                                    dt_obj_version_t *new_version)
2675 {
2676         struct inode *inode = osd_dt_obj(dt)->oo_inode;
2677
2678         CDEBUG(D_INODE, "Set version "LPX64" (old "LPX64") for inode %lu\n",
2679                *new_version, LDISKFS_I(inode)->i_fs_version, inode->i_ino);
2680
2681         LDISKFS_I(inode)->i_fs_version = *new_version;
2682         /** Version is set after all inode operations are finished,
2683          *  so we should mark it dirty here */
2684         ll_dirty_inode(inode, I_DIRTY_DATASYNC);
2685 }
2686
2687 /*
2688  * Concurrency: @dt is write locked.
2689  */
2690 static int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
2691                          const struct lu_buf *buf, const char *name, int fl,
2692                          struct thandle *handle, struct lustre_capa *capa)
2693 {
2694         struct osd_object      *obj      = osd_dt_obj(dt);
2695         struct inode           *inode    = obj->oo_inode;
2696         struct osd_thread_info *info     = osd_oti_get(env);
2697         int                     fs_flags = 0;
2698         ENTRY;
2699
2700         LASSERT(handle != NULL);
2701
2702         /* version set is not real XATTR */
2703         if (strcmp(name, XATTR_NAME_VERSION) == 0) {
2704                 /* for version we are just using xattr API but change inode
2705                  * field instead */
2706                 LASSERT(buf->lb_len == sizeof(dt_obj_version_t));
2707                 osd_object_version_set(env, dt, buf->lb_buf);
2708                 return sizeof(dt_obj_version_t);
2709         }
2710
2711         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
2712                 return -EACCES;
2713
2714         osd_trans_exec_op(env, handle, OSD_OT_XATTR_SET);
2715         if (fl & LU_XATTR_REPLACE)
2716                 fs_flags |= XATTR_REPLACE;
2717
2718         if (fl & LU_XATTR_CREATE)
2719                 fs_flags |= XATTR_CREATE;
2720
2721         return __osd_xattr_set(info, inode, name, buf->lb_buf, buf->lb_len,
2722                                fs_flags);
2723 }
2724
2725 /*
2726  * Concurrency: @dt is read locked.
2727  */
2728 static int osd_xattr_list(const struct lu_env *env, struct dt_object *dt,
2729                           struct lu_buf *buf, struct lustre_capa *capa)
2730 {
2731         struct osd_object      *obj    = osd_dt_obj(dt);
2732         struct inode           *inode  = obj->oo_inode;
2733         struct osd_thread_info *info   = osd_oti_get(env);
2734         struct dentry          *dentry = &info->oti_obj_dentry;
2735
2736         LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
2737         LASSERT(inode->i_op != NULL && inode->i_op->listxattr != NULL);
2738         LASSERT(osd_read_locked(env, obj) || osd_write_locked(env, obj));
2739
2740         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
2741                 return -EACCES;
2742
2743         dentry->d_inode = inode;
2744         dentry->d_sb = inode->i_sb;
2745         return inode->i_op->listxattr(dentry, buf->lb_buf, buf->lb_len);
2746 }
2747
2748 static int osd_declare_xattr_del(const struct lu_env *env,
2749                                  struct dt_object *dt, const char *name,
2750                                  struct thandle *handle)
2751 {
2752         struct osd_thandle *oh;
2753
2754         LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
2755         LASSERT(handle != NULL);
2756
2757         oh = container_of0(handle, struct osd_thandle, ot_super);
2758         LASSERT(oh->ot_handle == NULL);
2759
2760         osd_trans_declare_op(env, oh, OSD_OT_XATTR_SET,
2761                              osd_dto_credits_noquota[DTO_XATTR_SET]);
2762
2763         return 0;
2764 }
2765
2766 /*
2767  * Concurrency: @dt is write locked.
2768  */
2769 static int osd_xattr_del(const struct lu_env *env, struct dt_object *dt,
2770                          const char *name, struct thandle *handle,
2771                          struct lustre_capa *capa)
2772 {
2773         struct osd_object      *obj    = osd_dt_obj(dt);
2774         struct inode           *inode  = obj->oo_inode;
2775         struct osd_thread_info *info   = osd_oti_get(env);
2776         struct dentry          *dentry = &info->oti_obj_dentry;
2777         int                     rc;
2778
2779         LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
2780         LASSERT(inode->i_op != NULL && inode->i_op->removexattr != NULL);
2781         LASSERT(handle != NULL);
2782
2783         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
2784                 return -EACCES;
2785
2786         osd_trans_exec_op(env, handle, OSD_OT_XATTR_SET);
2787
2788         ll_vfs_dq_init(inode);
2789         dentry->d_inode = inode;
2790         dentry->d_sb = inode->i_sb;
2791         rc = inode->i_op->removexattr(dentry, name);
2792         return rc;
2793 }
2794
2795 static struct obd_capa *osd_capa_get(const struct lu_env *env,
2796                                      struct dt_object *dt,
2797                                      struct lustre_capa *old,
2798                                      __u64 opc)
2799 {
2800         struct osd_thread_info *info = osd_oti_get(env);
2801         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
2802         struct osd_object *obj = osd_dt_obj(dt);
2803         struct osd_device *dev = osd_obj2dev(obj);
2804         struct lustre_capa_key *key = &info->oti_capa_key;
2805         struct lustre_capa *capa = &info->oti_capa;
2806         struct obd_capa *oc;
2807         struct md_capainfo *ci;
2808         int rc;
2809         ENTRY;
2810
2811         if (!dev->od_fl_capa)
2812                 RETURN(ERR_PTR(-ENOENT));
2813
2814         LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
2815         LINVRNT(osd_invariant(obj));
2816
2817         /* renewal sanity check */
2818         if (old && osd_object_auth(env, dt, old, opc))
2819                 RETURN(ERR_PTR(-EACCES));
2820
2821         ci = md_capainfo(env);
2822         if (unlikely(!ci))
2823                 RETURN(ERR_PTR(-ENOENT));
2824
2825         switch (ci->mc_auth) {
2826         case LC_ID_NONE:
2827                 RETURN(NULL);
2828         case LC_ID_PLAIN:
2829                 capa->lc_uid = obj->oo_inode->i_uid;
2830                 capa->lc_gid = obj->oo_inode->i_gid;
2831                 capa->lc_flags = LC_ID_PLAIN;
2832                 break;
2833         case LC_ID_CONVERT: {
2834                 __u32 d[4], s[4];
2835
2836                 s[0] = obj->oo_inode->i_uid;
2837                 cfs_get_random_bytes(&(s[1]), sizeof(__u32));
2838                 s[2] = obj->oo_inode->i_gid;
2839                 cfs_get_random_bytes(&(s[3]), sizeof(__u32));
2840                 rc = capa_encrypt_id(d, s, key->lk_key, CAPA_HMAC_KEY_MAX_LEN);
2841                 if (unlikely(rc))
2842                         RETURN(ERR_PTR(rc));
2843
2844                 capa->lc_uid   = ((__u64)d[1] << 32) | d[0];
2845                 capa->lc_gid   = ((__u64)d[3] << 32) | d[2];
2846                 capa->lc_flags = LC_ID_CONVERT;
2847                 break;
2848         }
2849         default:
2850                 RETURN(ERR_PTR(-EINVAL));
2851         }
2852
2853         capa->lc_fid = *fid;
2854         capa->lc_opc = opc;
2855         capa->lc_flags |= dev->od_capa_alg << 24;
2856         capa->lc_timeout = dev->od_capa_timeout;
2857         capa->lc_expiry = 0;
2858
2859         oc = capa_lookup(dev->od_capa_hash, capa, 1);
2860         if (oc) {
2861                 LASSERT(!capa_is_expired(oc));
2862                 RETURN(oc);
2863         }
2864
2865         spin_lock(&capa_lock);
2866         *key = dev->od_capa_keys[1];
2867         spin_unlock(&capa_lock);
2868
2869         capa->lc_keyid = key->lk_keyid;
2870         capa->lc_expiry = cfs_time_current_sec() + dev->od_capa_timeout;
2871
2872         rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
2873         if (rc) {
2874                 DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
2875                 RETURN(ERR_PTR(rc));
2876         }
2877
2878         oc = capa_add(dev->od_capa_hash, capa);
2879         RETURN(oc);
2880 }
2881
2882 static int osd_object_sync(const struct lu_env *env, struct dt_object *dt)
2883 {
2884         struct osd_object       *obj    = osd_dt_obj(dt);
2885         struct inode            *inode  = obj->oo_inode;
2886         struct osd_thread_info  *info   = osd_oti_get(env);
2887         struct dentry           *dentry = &info->oti_obj_dentry;
2888         struct file             *file   = &info->oti_file;
2889         int                     rc;
2890
2891         ENTRY;
2892
2893         dentry->d_inode = inode;
2894         dentry->d_sb = inode->i_sb;
2895         file->f_dentry = dentry;
2896         file->f_mapping = inode->i_mapping;
2897         file->f_op = inode->i_fop;
2898 #ifndef HAVE_FILE_FSYNC_4ARGS
2899         mutex_lock(&inode->i_mutex);
2900 #endif
2901         rc = do_fsync(file, 0);
2902 #ifndef HAVE_FILE_FSYNC_4ARGS
2903         mutex_unlock(&inode->i_mutex);
2904 #endif
2905         RETURN(rc);
2906 }
2907
2908 static int osd_data_get(const struct lu_env *env, struct dt_object *dt,
2909                         void **data)
2910 {
2911         struct osd_object *obj = osd_dt_obj(dt);
2912         ENTRY;
2913
2914         *data = (void *)obj->oo_inode;
2915         RETURN(0);
2916 }
2917
2918 /*
2919  * Index operations.
2920  */
2921
2922 static int osd_iam_index_probe(const struct lu_env *env, struct osd_object *o,
2923                            const struct dt_index_features *feat)
2924 {
2925         struct iam_descr *descr;
2926
2927         if (osd_object_is_root(o))
2928                 return feat == &dt_directory_features;
2929
2930         LASSERT(o->oo_dir != NULL);
2931
2932         descr = o->oo_dir->od_container.ic_descr;
2933         if (feat == &dt_directory_features) {
2934                 if (descr->id_rec_size == sizeof(struct osd_fid_pack))
2935                         return 1;
2936                 else
2937                         return 0;
2938         } else {
2939                 return
2940                         feat->dif_keysize_min <= descr->id_key_size &&
2941                         descr->id_key_size <= feat->dif_keysize_max &&
2942                         feat->dif_recsize_min <= descr->id_rec_size &&
2943                         descr->id_rec_size <= feat->dif_recsize_max &&
2944                         !(feat->dif_flags & (DT_IND_VARKEY |
2945                                              DT_IND_VARREC | DT_IND_NONUNQ)) &&
2946                         ergo(feat->dif_flags & DT_IND_UPDATE,
2947                              1 /* XXX check that object (and file system) is
2948                                 * writable */);
2949         }
2950 }
2951
2952 static int osd_iam_container_init(const struct lu_env *env,
2953                                   struct osd_object *obj,
2954                                   struct osd_directory *dir)
2955 {
2956         struct iam_container *bag = &dir->od_container;
2957         int result;
2958
2959         result = iam_container_init(bag, &dir->od_descr, obj->oo_inode);
2960         if (result != 0)
2961                 return result;
2962
2963         result = iam_container_setup(bag);
2964         if (result == 0)
2965                 obj->oo_dt.do_index_ops = &osd_index_iam_ops;
2966         else
2967                 iam_container_fini(bag);
2968
2969         return result;
2970 }
2971
2972
2973 /*
2974  * Concurrency: no external locking is necessary.
2975  */
2976 static int osd_index_try(const struct lu_env *env, struct dt_object *dt,
2977                          const struct dt_index_features *feat)
2978 {
2979         int                      result;
2980         int                      skip_iam = 0;
2981         struct osd_object       *obj = osd_dt_obj(dt);
2982
2983         LINVRNT(osd_invariant(obj));
2984
2985         if (osd_object_is_root(obj)) {
2986                 dt->do_index_ops = &osd_index_ea_ops;
2987                 result = 0;
2988         } else if (feat == &dt_directory_features) {
2989                 dt->do_index_ops = &osd_index_ea_ops;
2990                 if (obj->oo_inode != NULL && S_ISDIR(obj->oo_inode->i_mode))
2991                         result = 0;
2992                 else
2993                         result = -ENOTDIR;
2994                 skip_iam = 1;
2995         } else if (unlikely(feat == &dt_otable_features)) {
2996                 dt->do_index_ops = &osd_otable_ops;
2997                 return 0;
2998         } else if (unlikely(feat == &dt_acct_features)) {
2999                 dt->do_index_ops = &osd_acct_index_ops;
3000                 result = 0;
3001                 skip_iam = 1;
3002         } else if (!osd_has_index(obj)) {
3003                 struct osd_directory *dir;
3004
3005                 OBD_ALLOC_PTR(dir);
3006                 if (dir != NULL) {
3007
3008                         spin_lock(&obj->oo_guard);
3009                         if (obj->oo_dir == NULL)
3010                                 obj->oo_dir = dir;
3011                         else
3012                                 /*
3013                                  * Concurrent thread allocated container data.
3014                                  */
3015                                 OBD_FREE_PTR(dir);
3016                         spin_unlock(&obj->oo_guard);
3017                         /*
3018                          * Now, that we have container data, serialize its
3019                          * initialization.
3020                          */
3021                         down_write(&obj->oo_ext_idx_sem);
3022                         /*
3023                          * recheck under lock.
3024                          */
3025                         if (!osd_has_index(obj))
3026                                 result = osd_iam_container_init(env, obj, dir);
3027                         else
3028                                 result = 0;
3029                         up_write(&obj->oo_ext_idx_sem);
3030                 } else {
3031                         result = -ENOMEM;
3032                 }
3033         } else {
3034                 result = 0;
3035         }
3036
3037         if (result == 0 && skip_iam == 0) {
3038                 if (!osd_iam_index_probe(env, obj, feat))
3039                         result = -ENOTDIR;
3040         }
3041         LINVRNT(osd_invariant(obj));
3042
3043         if (result == 0 && is_quota_glb_feat(feat) &&
3044             fid_seq(lu_object_fid(&dt->do_lu)) == FID_SEQ_QUOTA_GLB)
3045                 result = osd_quota_migration(env, dt, feat);
3046
3047         return result;
3048 }
3049
3050 static int osd_otable_it_attr_get(const struct lu_env *env,
3051                                  struct dt_object *dt,
3052                                  struct lu_attr *attr,
3053                                  struct lustre_capa *capa)
3054 {
3055         attr->la_valid = 0;
3056         return 0;
3057 }
3058
3059 static const struct dt_object_operations osd_obj_ops = {
3060         .do_read_lock         = osd_object_read_lock,
3061         .do_write_lock        = osd_object_write_lock,
3062         .do_read_unlock       = osd_object_read_unlock,
3063         .do_write_unlock      = osd_object_write_unlock,
3064         .do_write_locked      = osd_object_write_locked,
3065         .do_attr_get          = osd_attr_get,
3066         .do_declare_attr_set  = osd_declare_attr_set,
3067         .do_attr_set          = osd_attr_set,
3068         .do_ah_init           = osd_ah_init,
3069         .do_declare_create    = osd_declare_object_create,
3070         .do_create            = osd_object_create,
3071         .do_declare_destroy   = osd_declare_object_destroy,
3072         .do_destroy           = osd_object_destroy,
3073         .do_index_try         = osd_index_try,
3074         .do_declare_ref_add   = osd_declare_object_ref_add,
3075         .do_ref_add           = osd_object_ref_add,
3076         .do_declare_ref_del   = osd_declare_object_ref_del,
3077         .do_ref_del           = osd_object_ref_del,
3078         .do_xattr_get         = osd_xattr_get,
3079         .do_declare_xattr_set = osd_declare_xattr_set,
3080         .do_xattr_set         = osd_xattr_set,
3081         .do_declare_xattr_del = osd_declare_xattr_del,
3082         .do_xattr_del         = osd_xattr_del,
3083         .do_xattr_list        = osd_xattr_list,
3084         .do_capa_get          = osd_capa_get,
3085         .do_object_sync       = osd_object_sync,
3086         .do_data_get          = osd_data_get,
3087 };
3088
3089 /**
3090  * dt_object_operations for interoperability mode
3091  * (i.e. to run 2.0 mds on 1.8 disk) (b11826)
3092  */
3093 static const struct dt_object_operations osd_obj_ea_ops = {
3094         .do_read_lock         = osd_object_read_lock,
3095         .do_write_lock        = osd_object_write_lock,
3096         .do_read_unlock       = osd_object_read_unlock,
3097         .do_write_unlock      = osd_object_write_unlock,
3098         .do_write_locked      = osd_object_write_locked,
3099         .do_attr_get          = osd_attr_get,
3100         .do_declare_attr_set  = osd_declare_attr_set,
3101         .do_attr_set          = osd_attr_set,
3102         .do_ah_init           = osd_ah_init,
3103         .do_declare_create    = osd_declare_object_create,
3104         .do_create            = osd_object_ea_create,
3105         .do_declare_destroy   = osd_declare_object_destroy,
3106         .do_destroy           = osd_object_destroy,
3107         .do_index_try         = osd_index_try,
3108         .do_declare_ref_add   = osd_declare_object_ref_add,
3109         .do_ref_add           = osd_object_ref_add,
3110         .do_declare_ref_del   = osd_declare_object_ref_del,
3111         .do_ref_del           = osd_object_ref_del,
3112         .do_xattr_get         = osd_xattr_get,
3113         .do_declare_xattr_set = osd_declare_xattr_set,
3114         .do_xattr_set         = osd_xattr_set,
3115         .do_declare_xattr_del = osd_declare_xattr_del,
3116         .do_xattr_del         = osd_xattr_del,
3117         .do_xattr_list        = osd_xattr_list,
3118         .do_capa_get          = osd_capa_get,
3119         .do_object_sync       = osd_object_sync,
3120         .do_data_get          = osd_data_get,
3121 };
3122
3123 static const struct dt_object_operations osd_obj_otable_it_ops = {
3124         .do_attr_get    = osd_otable_it_attr_get,
3125         .do_index_try   = osd_index_try,
3126 };
3127
3128 static int osd_index_declare_iam_delete(const struct lu_env *env,
3129                                         struct dt_object *dt,
3130                                         const struct dt_key *key,
3131                                         struct thandle *handle)
3132 {
3133         struct osd_thandle    *oh;
3134
3135         oh = container_of0(handle, struct osd_thandle, ot_super);
3136         LASSERT(oh->ot_handle == NULL);
3137
3138         osd_trans_declare_op(env, oh, OSD_OT_DELETE,
3139                              osd_dto_credits_noquota[DTO_INDEX_DELETE]);
3140
3141         return 0;
3142 }
3143
3144 /**
3145  *      delete a (key, value) pair from index \a dt specified by \a key
3146  *
3147  *      \param  dt      osd index object
3148  *      \param  key     key for index
3149  *      \param  rec     record reference
3150  *      \param  handle  transaction handler
3151  *
3152  *      \retval  0  success
3153  *      \retval -ve   failure
3154  */
3155
3156 static int osd_index_iam_delete(const struct lu_env *env, struct dt_object *dt,
3157                                 const struct dt_key *key,
3158                                 struct thandle *handle,
3159                                 struct lustre_capa *capa)
3160 {
3161         struct osd_thread_info *oti = osd_oti_get(env);
3162         struct osd_object      *obj = osd_dt_obj(dt);
3163         struct osd_thandle     *oh;
3164         struct iam_path_descr  *ipd;
3165         struct iam_container   *bag = &obj->oo_dir->od_container;
3166         int                     rc;
3167
3168         ENTRY;
3169
3170         LINVRNT(osd_invariant(obj));
3171         LASSERT(dt_object_exists(dt) && !dt_object_remote(dt));
3172         LASSERT(bag->ic_object == obj->oo_inode);
3173         LASSERT(handle != NULL);
3174
3175         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_DELETE))
3176                 RETURN(-EACCES);
3177
3178         osd_trans_exec_op(env, handle, OSD_OT_DELETE);
3179
3180         ipd = osd_idx_ipd_get(env, bag);
3181         if (unlikely(ipd == NULL))
3182                 RETURN(-ENOMEM);
3183
3184         oh = container_of0(handle, struct osd_thandle, ot_super);
3185         LASSERT(oh->ot_handle != NULL);
3186         LASSERT(oh->ot_handle->h_transaction != NULL);
3187
3188         if (fid_is_quota(lu_object_fid(&dt->do_lu))) {
3189                 /* swab quota uid/gid provided by caller */
3190                 oti->oti_quota_id = cpu_to_le64(*((__u64 *)key));
3191                 key = (const struct dt_key *)&oti->oti_quota_id;
3192         }
3193
3194         rc = iam_delete(oh->ot_handle, bag, (const struct iam_key *)key, ipd);
3195         osd_ipd_put(env, bag, ipd);
3196         LINVRNT(osd_invariant(obj));
3197         RETURN(rc);
3198 }
3199
3200 static int osd_index_declare_ea_delete(const struct lu_env *env,
3201                                        struct dt_object *dt,
3202