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