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