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