Whamcloud - gitweb
LU-5938 mdd: fixed oops when dereferencing structure
[fs/lustre-release.git] / lustre / mdd / mdd_dir.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, 2014, 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/mdd/mdd_dir.c
37  *
38  * Lustre Metadata Server (mdd) routines
39  *
40  * Author: Wang Di <wangdi@intel.com>
41  */
42
43 #define DEBUG_SUBSYSTEM S_MDS
44
45 #include <obd_class.h>
46 #include <obd_support.h>
47 #include <lustre_mds.h>
48 #include <lustre_fid.h>
49
50 #include "mdd_internal.h"
51
52 static const char dot[] = ".";
53 static const char dotdot[] = "..";
54
55 static struct lu_name lname_dotdot = {
56         (char *) dotdot,
57         sizeof(dotdot) - 1
58 };
59
60 static inline int
61 mdd_name_check(struct mdd_device *m, const struct lu_name *ln)
62 {
63         if (!lu_name_is_valid(ln))
64                 return -EINVAL;
65         else if (ln->ln_namelen > m->mdd_dt_conf.ddp_max_name_len)
66                 return -ENAMETOOLONG;
67         else
68                 return 0;
69 }
70
71 /* Get FID from name and parent */
72 static int
73 __mdd_lookup(const struct lu_env *env, struct md_object *pobj,
74              const struct lu_attr *pattr, const struct lu_name *lname,
75              struct lu_fid* fid, int mask)
76 {
77         const char *name                = lname->ln_name;
78         const struct dt_key *key        = (const struct dt_key *)name;
79         struct mdd_object *mdd_obj      = md2mdd_obj(pobj);
80         struct mdd_device *m            = mdo2mdd(pobj);
81         struct dt_object *dir           = mdd_object_child(mdd_obj);
82         int rc;
83         ENTRY;
84
85         if (unlikely(mdd_is_dead_obj(mdd_obj)))
86                 RETURN(-ESTALE);
87
88         if (mdd_object_remote(mdd_obj)) {
89                 CDEBUG(D_INFO, "%s: Object "DFID" locates on remote server\n",
90                        mdd2obd_dev(m)->obd_name, PFID(mdo2fid(mdd_obj)));
91         } else if (!mdd_object_exists(mdd_obj)) {
92                 RETURN(-ESTALE);
93         }
94
95         rc = mdd_permission_internal_locked(env, mdd_obj, pattr, mask,
96                                             MOR_TGT_PARENT);
97         if (rc)
98                 RETURN(rc);
99
100         if (likely(S_ISDIR(mdd_object_type(mdd_obj)) &&
101                    dt_try_as_dir(env, dir)))
102                 rc = dt_lookup(env, dir, (struct dt_rec *)fid, key,
103                                mdd_object_capa(env, mdd_obj));
104         else
105                 rc = -ENOTDIR;
106
107         RETURN(rc);
108 }
109
110 int mdd_lookup(const struct lu_env *env,
111                struct md_object *pobj, const struct lu_name *lname,
112                struct lu_fid *fid, struct md_op_spec *spec)
113 {
114         struct lu_attr *pattr = MDD_ENV_VAR(env, pattr);
115         int rc;
116         ENTRY;
117
118         rc = mdd_la_get(env, md2mdd_obj(pobj), pattr, BYPASS_CAPA);
119         if (rc != 0)
120                 RETURN(rc);
121
122         rc = __mdd_lookup(env, pobj, pattr, lname, fid,
123                           (spec != NULL && spec->sp_permitted) ? 0 : MAY_EXEC);
124         RETURN(rc);
125 }
126
127 static inline int mdd_parent_fid(const struct lu_env *env,
128                                  struct mdd_object *obj,
129                                  const struct lu_attr *attr,
130                                  struct lu_fid *fid)
131 {
132         return __mdd_lookup(env, &obj->mod_obj, attr, &lname_dotdot, fid, 0);
133 }
134
135 /*
136  * For root fid use special function, which does not compare version component
137  * of fid. Version component is different for root fids on all MDTs.
138  */
139 int mdd_is_root(struct mdd_device *mdd, const struct lu_fid *fid)
140 {
141         return fid_seq(&mdd->mdd_root_fid) == fid_seq(fid) &&
142                 fid_oid(&mdd->mdd_root_fid) == fid_oid(fid);
143 }
144
145 /*
146  * return 1: if lf is the fid of the ancestor of p1;
147  * return 0: if not;
148  *
149  * return -EREMOTE: if remote object is found, in this
150  * case fid of remote object is saved to @pf;
151  *
152  * otherwise: values < 0, errors.
153  */
154 static int mdd_is_parent(const struct lu_env *env,
155                         struct mdd_device *mdd,
156                         struct mdd_object *p1,
157                         const struct lu_attr *attr,
158                         const struct lu_fid *lf,
159                         struct lu_fid *pf)
160 {
161         struct mdd_object *parent = NULL;
162         struct lu_fid *pfid;
163         int rc;
164         ENTRY;
165
166         LASSERT(!lu_fid_eq(mdo2fid(p1), lf));
167         pfid = &mdd_env_info(env)->mti_fid;
168
169         /* Check for root first. */
170         if (mdd_is_root(mdd, mdo2fid(p1)))
171                 RETURN(0);
172
173         for(;;) {
174                 /* this is done recursively, bypass capa for each obj */
175                 mdd_set_capainfo(env, 4, p1, BYPASS_CAPA);
176                 rc = mdd_parent_fid(env, p1, attr, pfid);
177                 if (rc)
178                         GOTO(out, rc);
179                 if (mdd_is_root(mdd, pfid))
180                         GOTO(out, rc = 0);
181                 if (lu_fid_eq(pfid, &mdd->mdd_local_root_fid))
182                         GOTO(out, rc = 0);
183                 if (lu_fid_eq(pfid, lf))
184                         GOTO(out, rc = 1);
185                 if (parent)
186                         mdd_object_put(env, parent);
187
188                 parent = mdd_object_find(env, mdd, pfid);
189                 if (IS_ERR(parent)) {
190                         GOTO(out, rc = PTR_ERR(parent));
191                 } else if (mdd_object_remote(parent)) {
192                         /*FIXME: Because of the restriction of rename in Phase I.
193                          * If the parent is remote, we just assumed lf is not the
194                          * parent of P1 for now */
195                         GOTO(out, rc = 0);
196                 }
197                 p1 = parent;
198         }
199         EXIT;
200 out:
201         if (parent && !IS_ERR(parent))
202                 mdd_object_put(env, parent);
203         return rc;
204 }
205
206 /*
207  * No permission check is needed.
208  *
209  * returns 1: if fid is ancestor of @mo;
210  * returns 0: if fid is not a ancestor of @mo;
211  *
212  * returns EREMOTE if remote object is found, fid of remote object is saved to
213  * @fid;
214  *
215  * returns < 0: if error
216  */
217 int mdd_is_subdir(const struct lu_env *env, struct md_object *mo,
218                   const struct lu_fid *fid, struct lu_fid *sfid)
219 {
220         struct mdd_device *mdd = mdo2mdd(mo);
221         struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
222         int rc;
223         ENTRY;
224
225         if (!S_ISDIR(mdd_object_type(md2mdd_obj(mo))))
226                 RETURN(0);
227
228         rc = mdd_la_get(env, md2mdd_obj(mo), attr, BYPASS_CAPA);
229         if (rc != 0)
230                 RETURN(rc);
231
232         rc = mdd_is_parent(env, mdd, md2mdd_obj(mo), attr, fid, sfid);
233         if (rc == 0) {
234                 /* found root */
235                 fid_zero(sfid);
236         } else if (rc == 1) {
237                 /* found @fid is parent */
238                 *sfid = *fid;
239                 rc = 0;
240         }
241         RETURN(rc);
242 }
243
244 /*
245  * Check that @dir contains no entries except (possibly) dot and dotdot.
246  *
247  * Returns:
248  *
249  *             0        empty
250  *      -ENOTDIR        not a directory object
251  *    -ENOTEMPTY        not empty
252  *           -ve        other error
253  *
254  */
255 static int mdd_dir_is_empty(const struct lu_env *env,
256                             struct mdd_object *dir)
257 {
258         struct dt_it     *it;
259         struct dt_object *obj;
260         const struct dt_it_ops *iops;
261         int result;
262         ENTRY;
263
264         obj = mdd_object_child(dir);
265         if (!dt_try_as_dir(env, obj))
266                 RETURN(-ENOTDIR);
267
268         iops = &obj->do_index_ops->dio_it;
269         it = iops->init(env, obj, LUDA_64BITHASH, BYPASS_CAPA);
270         if (!IS_ERR(it)) {
271                 result = iops->get(env, it, (const struct dt_key *)"");
272                 if (result > 0) {
273                         int i;
274                         for (result = 0, i = 0; result == 0 && i < 3; ++i)
275                                 result = iops->next(env, it);
276                         if (result == 0)
277                                 result = -ENOTEMPTY;
278                         else if (result == 1)
279                                 result = 0;
280                 } else if (result == 0)
281                         /*
282                          * Huh? Index contains no zero key?
283                          */
284                         result = -EIO;
285
286                 iops->put(env, it);
287                 iops->fini(env, it);
288         } else
289                 result = PTR_ERR(it);
290         RETURN(result);
291 }
292
293 /**
294  * Determine if the target object can be hard linked, and right now it only
295  * checks if the link count reach the maximum limit. Note: for ldiskfs, the
296  * directory nlink count might exceed the maximum link count(see
297  * osd_object_ref_add), so it only check nlink for non-directories.
298  *
299  * \param[in] env       thread environment
300  * \param[in] obj       object being linked to
301  * \param[in] la        attributes of \a obj
302  *
303  * \retval              0 if \a obj can be hard linked
304  * \retval              negative error if \a obj is a directory or has too
305  *                      many links
306  */
307 static int __mdd_may_link(const struct lu_env *env, struct mdd_object *obj,
308                           const struct lu_attr *la)
309 {
310         struct mdd_device *m = mdd_obj2mdd_dev(obj);
311         ENTRY;
312
313         LASSERT(la != NULL);
314
315         /* Subdir count limitation can be broken through
316          * (see osd_object_ref_add), so only check non-directory here. */
317         if (!S_ISDIR(la->la_mode) &&
318             la->la_nlink >= m->mdd_dt_conf.ddp_max_nlink)
319                 RETURN(-EMLINK);
320
321         RETURN(0);
322 }
323
324 /**
325  * Check whether it may create the cobj under the pobj.
326  *
327  * \param[in] env       execution environment
328  * \param[in] pobj      the parent directory
329  * \param[in] pattr     the attribute of the parent directory
330  * \param[in] cobj      the child to be created
331  * \param[in] check_perm        if check WRITE|EXEC permission for parent
332  *
333  * \retval              = 0 create the child under this dir is allowed
334  * \retval              negative errno create the child under this dir is
335  *                      not allowed
336  */
337 int mdd_may_create(const struct lu_env *env, struct mdd_object *pobj,
338                    const struct lu_attr *pattr, struct mdd_object *cobj,
339                    bool check_perm)
340 {
341         struct mdd_thread_info *info = mdd_env_info(env);
342         struct lu_buf   *xbuf;
343         int rc = 0;
344         ENTRY;
345
346         if (cobj && mdd_object_exists(cobj))
347                 RETURN(-EEXIST);
348
349         if (mdd_is_dead_obj(pobj))
350                 RETURN(-ENOENT);
351
352         /* If the parent is a sub-stripe, check whether it is dead */
353         xbuf = mdd_buf_get(env, info->mti_key, sizeof(info->mti_key));
354         rc = mdo_xattr_get(env, pobj, xbuf, XATTR_NAME_LMV,
355                            mdd_object_capa(env, pobj));
356         if (unlikely(rc > 0)) {
357                 struct lmv_mds_md_v1  *lmv1 = xbuf->lb_buf;
358
359                 if (le32_to_cpu(lmv1->lmv_magic) == LMV_MAGIC_STRIPE &&
360                     le32_to_cpu(lmv1->lmv_hash_type) & LMV_HASH_FLAG_DEAD)
361                         RETURN(-ESTALE);
362         }
363         rc = 0;
364
365         if (check_perm)
366                 rc = mdd_permission_internal_locked(env, pobj, pattr,
367                                                     MAY_WRITE | MAY_EXEC,
368                                                     MOR_TGT_PARENT);
369         RETURN(rc);
370 }
371
372 /*
373  * Check whether can unlink from the pobj in the case of "cobj == NULL".
374  */
375 int mdd_may_unlink(const struct lu_env *env, struct mdd_object *pobj,
376                    const struct lu_attr *pattr, const struct lu_attr *attr)
377 {
378         int rc;
379         ENTRY;
380
381         if (mdd_is_dead_obj(pobj))
382                 RETURN(-ENOENT);
383
384         if (attr->la_flags & (LUSTRE_APPEND_FL | LUSTRE_IMMUTABLE_FL))
385                 RETURN(-EPERM);
386
387         rc = mdd_permission_internal_locked(env, pobj, pattr,
388                                             MAY_WRITE | MAY_EXEC,
389                                             MOR_TGT_PARENT);
390         if (rc != 0)
391                 RETURN(rc);
392
393         if (pattr->la_flags & LUSTRE_APPEND_FL)
394                 RETURN(-EPERM);
395
396         RETURN(rc);
397 }
398
399 /*
400  * pobj == NULL is remote ops case, under such case, pobj's
401  * VTX feature has been checked already, no need check again.
402  */
403 static inline int mdd_is_sticky(const struct lu_env *env,
404                                 struct mdd_object *pobj,
405                                 const struct lu_attr *pattr,
406                                 struct mdd_object *cobj,
407                                 const struct lu_attr *cattr)
408 {
409         struct lu_ucred *uc = lu_ucred_assert(env);
410
411         if (pobj != NULL) {
412                 LASSERT(pattr != NULL);
413                 if (!(pattr->la_mode & S_ISVTX) ||
414                     (pattr->la_uid == uc->uc_fsuid))
415                         return 0;
416         }
417
418         LASSERT(cattr != NULL);
419         if (cattr->la_uid == uc->uc_fsuid)
420                 return 0;
421
422         return !md_capable(uc, CFS_CAP_FOWNER);
423 }
424
425 static int mdd_may_delete_entry(const struct lu_env *env,
426                                 struct mdd_object *pobj,
427                                 const struct lu_attr *pattr,
428                                 int check_perm)
429 {
430         ENTRY;
431
432         LASSERT(pobj != NULL);
433         if (!mdd_object_exists(pobj))
434                 RETURN(-ENOENT);
435
436         if (mdd_is_dead_obj(pobj))
437                 RETURN(-ENOENT);
438
439         if (check_perm) {
440                 int rc;
441                 rc = mdd_permission_internal_locked(env, pobj, pattr,
442                                             MAY_WRITE | MAY_EXEC,
443                                             MOR_TGT_PARENT);
444                 if (rc)
445                         RETURN(rc);
446         }
447
448         if (pattr->la_flags & LUSTRE_APPEND_FL)
449                 RETURN(-EPERM);
450
451         RETURN(0);
452 }
453
454 /*
455  * Check whether it may delete the cobj from the pobj.
456  * pobj maybe NULL
457  */
458 int mdd_may_delete(const struct lu_env *env, struct mdd_object *tpobj,
459                    const struct lu_attr *tpattr, struct mdd_object *tobj,
460                    const struct lu_attr *tattr, const struct lu_attr *cattr,
461                    int check_perm, int check_empty)
462 {
463         int rc = 0;
464         ENTRY;
465
466         if (tpobj) {
467                 LASSERT(tpattr != NULL);
468                 rc = mdd_may_delete_entry(env, tpobj, tpattr, check_perm);
469                 if (rc != 0)
470                         RETURN(rc);
471         }
472
473         if (tobj == NULL)
474                 RETURN(0);
475
476         if (!mdd_object_exists(tobj))
477                 RETURN(-ENOENT);
478
479         if (mdd_is_dead_obj(tobj))
480                 RETURN(-ESTALE);
481
482         if (mdd_is_sticky(env, tpobj, tpattr, tobj, tattr))
483                 RETURN(-EPERM);
484
485         if (tattr->la_flags & (LUSTRE_APPEND_FL | LUSTRE_IMMUTABLE_FL))
486                 RETURN(-EPERM);
487
488         /* additional check the rename case */
489         if (cattr) {
490                 if (S_ISDIR(cattr->la_mode)) {
491                         struct mdd_device *mdd = mdo2mdd(&tobj->mod_obj);
492
493                         if (!S_ISDIR(tattr->la_mode))
494                                 RETURN(-ENOTDIR);
495
496                         if (lu_fid_eq(mdo2fid(tobj), &mdd->mdd_root_fid))
497                                 RETURN(-EBUSY);
498                 } else if (S_ISDIR(tattr->la_mode))
499                         RETURN(-EISDIR);
500         }
501
502         if (S_ISDIR(tattr->la_mode) && check_empty)
503                 rc = mdd_dir_is_empty(env, tobj);
504
505         RETURN(rc);
506 }
507
508 /**
509  * Check whether it can create the link file(linked to @src_obj) under
510  * the target directory(@tgt_obj), and src_obj has been locked by
511  * mdd_write_lock.
512  *
513  * \param[in] env       execution environment
514  * \param[in] tgt_obj   the target directory
515  * \param[in] tattr     attributes of target directory
516  * \param[in] lname     the link name
517  * \param[in] src_obj   source object for link
518  * \param[in] cattr     attributes for source object
519  *
520  * \retval              = 0 it is allowed to create the link file under tgt_obj
521  * \retval              negative error not allowed to create the link file
522  */
523 static int mdd_link_sanity_check(const struct lu_env *env,
524                                  struct mdd_object *tgt_obj,
525                                  const struct lu_attr *tattr,
526                                  const struct lu_name *lname,
527                                  struct mdd_object *src_obj,
528                                  const struct lu_attr *cattr)
529 {
530         struct mdd_device *m = mdd_obj2mdd_dev(src_obj);
531         int rc = 0;
532         ENTRY;
533
534         if (!mdd_object_exists(src_obj))
535                 RETURN(-ENOENT);
536
537         if (mdd_is_dead_obj(src_obj))
538                 RETURN(-ESTALE);
539
540         /* Local ops, no lookup before link, check filename length here. */
541         rc = mdd_name_check(m, lname);
542         if (rc < 0)
543                 RETURN(rc);
544
545         if (cattr->la_flags & (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL))
546                 RETURN(-EPERM);
547
548         if (S_ISDIR(mdd_object_type(src_obj)))
549                 RETURN(-EPERM);
550
551         LASSERT(src_obj != tgt_obj);
552         rc = mdd_may_create(env, tgt_obj, tattr, NULL, true);
553         if (rc != 0)
554                 RETURN(rc);
555
556         rc = __mdd_may_link(env, src_obj, cattr);
557
558         RETURN(rc);
559 }
560
561 static int __mdd_index_delete_only(const struct lu_env *env, struct mdd_object *pobj,
562                                    const char *name, struct thandle *handle,
563                                    struct lustre_capa *capa)
564 {
565         struct dt_object *next = mdd_object_child(pobj);
566         int rc;
567         ENTRY;
568
569         if (dt_try_as_dir(env, next))
570                 rc = dt_delete(env, next, (struct dt_key *)name, handle, capa);
571         else
572                 rc = -ENOTDIR;
573
574         RETURN(rc);
575 }
576
577 static int __mdd_index_insert_only(const struct lu_env *env,
578                                    struct mdd_object *pobj,
579                                    const struct lu_fid *lf, __u32 type,
580                                    const char *name,
581                                    struct thandle *handle,
582                                    struct lustre_capa *capa)
583 {
584         struct dt_object *next = mdd_object_child(pobj);
585         int               rc;
586         ENTRY;
587
588         if (dt_try_as_dir(env, next)) {
589                 struct dt_insert_rec    *rec = &mdd_env_info(env)->mti_dt_rec;
590                 struct lu_ucred         *uc  = lu_ucred_check(env);
591                 int                      ignore_quota;
592
593                 rec->rec_fid = lf;
594                 rec->rec_type = type;
595                 ignore_quota = uc ? uc->uc_cap & CFS_CAP_SYS_RESOURCE_MASK : 1;
596                 rc = dt_insert(env, next, (const struct dt_rec *)rec,
597                                (const struct dt_key *)name, handle, capa,
598                                ignore_quota);
599         } else {
600                 rc = -ENOTDIR;
601         }
602         RETURN(rc);
603 }
604
605 /* insert named index, add reference if isdir */
606 static int __mdd_index_insert(const struct lu_env *env, struct mdd_object *pobj,
607                               const struct lu_fid *lf, __u32 type,
608                               const char *name, struct thandle *handle,
609                               struct lustre_capa *capa)
610 {
611         int rc;
612         ENTRY;
613
614         rc = __mdd_index_insert_only(env, pobj, lf, type, name, handle, capa);
615         if (rc == 0 && S_ISDIR(type)) {
616                 mdd_write_lock(env, pobj, MOR_TGT_PARENT);
617                 mdo_ref_add(env, pobj, handle);
618                 mdd_write_unlock(env, pobj);
619         }
620
621         RETURN(rc);
622 }
623
624 /* delete named index, drop reference if isdir */
625 static int __mdd_index_delete(const struct lu_env *env, struct mdd_object *pobj,
626                               const char *name, int is_dir, struct thandle *handle,
627                               struct lustre_capa *capa)
628 {
629         int               rc;
630         ENTRY;
631
632         rc = __mdd_index_delete_only(env, pobj, name, handle, capa);
633         if (rc == 0 && is_dir) {
634                 mdd_write_lock(env, pobj, MOR_TGT_PARENT);
635                 mdo_ref_del(env, pobj, handle);
636                 mdd_write_unlock(env, pobj);
637         }
638
639         RETURN(rc);
640 }
641
642 static int mdd_llog_record_calc_size(const struct lu_env *env,
643                                      const struct lu_name *tname,
644                                      const struct lu_name *sname)
645 {
646         const struct lu_ucred   *uc = lu_ucred(env);
647         enum changelog_rec_flags crf = 0;
648         size_t                   hdr_size = sizeof(struct llog_changelog_rec) -
649                                             sizeof(struct changelog_rec);
650
651         if (sname != NULL)
652                 crf |= CLF_RENAME;
653
654         if (uc != NULL && uc->uc_jobid[0] != '\0')
655                 crf |= CLF_JOBID;
656
657         return llog_data_len(hdr_size + changelog_rec_offset(crf) +
658                              (tname != NULL ? tname->ln_namelen : 0) +
659                              (sname != NULL ? 1 + sname->ln_namelen : 0));
660 }
661
662 int mdd_declare_changelog_store(const struct lu_env *env,
663                                 struct mdd_device *mdd,
664                                 const struct lu_name *tname,
665                                 const struct lu_name *sname,
666                                 struct thandle *handle)
667 {
668         struct obd_device               *obd = mdd2obd_dev(mdd);
669         struct llog_ctxt                *ctxt;
670         struct llog_changelog_rec       *rec;
671         struct lu_buf                   *buf;
672         int                              reclen;
673         int                              rc;
674
675         /* Not recording */
676         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
677                 return 0;
678
679         reclen = mdd_llog_record_calc_size(env, tname, sname);
680         buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf, reclen);
681         if (buf->lb_buf == NULL)
682                 return -ENOMEM;
683
684         rec = buf->lb_buf;
685         rec->cr_hdr.lrh_len = reclen;
686         rec->cr_hdr.lrh_type = CHANGELOG_REC;
687
688         ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
689         if (ctxt == NULL)
690                 return -ENXIO;
691
692         rc = llog_declare_add(env, ctxt->loc_handle, &rec->cr_hdr, handle);
693         llog_ctxt_put(ctxt);
694
695         return rc;
696 }
697
698 /** Add a changelog entry \a rec to the changelog llog
699  * \param mdd
700  * \param rec
701  * \param handle - currently ignored since llogs start their own transaction;
702  *              this will hopefully be fixed in llog rewrite
703  * \retval 0 ok
704  */
705 int mdd_changelog_store(const struct lu_env *env, struct mdd_device *mdd,
706                         struct llog_changelog_rec *rec, struct thandle *th)
707 {
708         struct obd_device       *obd = mdd2obd_dev(mdd);
709         struct llog_ctxt        *ctxt;
710         int                      rc;
711
712         rec->cr_hdr.lrh_len = llog_data_len(sizeof(*rec) +
713                                             changelog_rec_varsize(&rec->cr));
714
715         /* llog_lvfs_write_rec sets the llog tail len */
716         rec->cr_hdr.lrh_type = CHANGELOG_REC;
717         rec->cr.cr_time = cl_time();
718
719         spin_lock(&mdd->mdd_cl.mc_lock);
720         /* NB: I suppose it's possible llog_add adds out of order wrt cr_index,
721          * but as long as the MDD transactions are ordered correctly for e.g.
722          * rename conflicts, I don't think this should matter. */
723         rec->cr.cr_index = ++mdd->mdd_cl.mc_index;
724         spin_unlock(&mdd->mdd_cl.mc_lock);
725
726         ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
727         if (ctxt == NULL)
728                 return -ENXIO;
729
730         /* nested journal transaction */
731         rc = llog_add(env, ctxt->loc_handle, &rec->cr_hdr, NULL, th);
732         llog_ctxt_put(ctxt);
733         if (rc > 0)
734                 rc = 0;
735
736         return rc;
737 }
738
739 static void mdd_changelog_rec_ext_rename(struct changelog_rec *rec,
740                                          const struct lu_fid *sfid,
741                                          const struct lu_fid *spfid,
742                                          const struct lu_name *sname)
743 {
744         struct changelog_ext_rename     *rnm = changelog_rec_rename(rec);
745         size_t                           extsize = sname->ln_namelen + 1;
746
747         LASSERT(sfid != NULL);
748         LASSERT(spfid != NULL);
749         LASSERT(sname != NULL);
750
751         rnm->cr_sfid = *sfid;
752         rnm->cr_spfid = *spfid;
753
754         changelog_rec_name(rec)[rec->cr_namelen] = '\0';
755         strlcpy(changelog_rec_sname(rec), sname->ln_name, extsize);
756         rec->cr_namelen += extsize;
757 }
758
759 void mdd_changelog_rec_ext_jobid(struct changelog_rec *rec, const char *jobid)
760 {
761         struct changelog_ext_jobid      *jid = changelog_rec_jobid(rec);
762
763         if (jobid == NULL || jobid[0] == '\0')
764                 return;
765
766         strlcpy(jid->cr_jobid, jobid, sizeof(jid->cr_jobid));
767 }
768
769 /** Store a namespace change changelog record
770  * If this fails, we must fail the whole transaction; we don't
771  * want the change to commit without the log entry.
772  * \param target - mdd_object of change
773  * \param tpfid - target parent dir/object fid
774  * \param sfid - source object fid
775  * \param spfid - source parent fid
776  * \param tname - target name string
777  * \param sname - source name string
778  * \param handle - transacion handle
779  */
780 int mdd_changelog_ns_store(const struct lu_env *env,
781                            struct mdd_device *mdd,
782                            enum changelog_rec_type type,
783                            enum changelog_rec_flags crf,
784                            struct mdd_object *target,
785                            const struct lu_fid *tpfid,
786                            const struct lu_fid *sfid,
787                            const struct lu_fid *spfid,
788                            const struct lu_name *tname,
789                            const struct lu_name *sname,
790                            struct thandle *handle)
791 {
792         const struct lu_ucred           *uc = lu_ucred(env);
793         struct llog_changelog_rec       *rec;
794         struct lu_buf                   *buf;
795         int                              reclen;
796         int                              rc;
797         ENTRY;
798
799         /* Not recording */
800         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
801                 RETURN(0);
802
803         if ((mdd->mdd_cl.mc_mask & (1 << type)) == 0)
804                 RETURN(0);
805
806         LASSERT(tpfid != NULL);
807         LASSERT(tname != NULL);
808         LASSERT(handle != NULL);
809
810         reclen = mdd_llog_record_calc_size(env, tname, sname);
811         buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf, reclen);
812         if (buf->lb_buf == NULL)
813                 RETURN(-ENOMEM);
814         rec = buf->lb_buf;
815
816         crf &= CLF_FLAGMASK;
817
818         if (uc != NULL && uc->uc_jobid[0] != '\0')
819                 crf |= CLF_JOBID;
820
821         if (sname != NULL)
822                 crf |= CLF_RENAME;
823         else
824                 crf |= CLF_VERSION;
825
826         rec->cr.cr_flags = crf;
827         rec->cr.cr_type = (__u32)type;
828         rec->cr.cr_pfid = *tpfid;
829         rec->cr.cr_namelen = tname->ln_namelen;
830         memcpy(changelog_rec_name(&rec->cr), tname->ln_name, tname->ln_namelen);
831
832         if (crf & CLF_RENAME)
833                 mdd_changelog_rec_ext_rename(&rec->cr, sfid, spfid, sname);
834
835         if (crf & CLF_JOBID)
836                 mdd_changelog_rec_ext_jobid(&rec->cr, uc->uc_jobid);
837
838         if (likely(target != NULL)) {
839                 rec->cr.cr_tfid = *mdo2fid(target);
840                 target->mod_cltime = cfs_time_current_64();
841         } else {
842                 fid_zero(&rec->cr.cr_tfid);
843         }
844
845         rc = mdd_changelog_store(env, mdd, rec, handle);
846         if (rc < 0) {
847                 CERROR("changelog failed: rc=%d, op%d %s c"DFID" p"DFID"\n",
848                         rc, type, tname->ln_name, PFID(sfid), PFID(tpfid));
849                 return -EFAULT;
850         }
851
852         return 0;
853 }
854
855 static int __mdd_links_add(const struct lu_env *env,
856                            struct mdd_object *mdd_obj,
857                            struct linkea_data *ldata,
858                            const struct lu_name *lname,
859                            const struct lu_fid *pfid,
860                            int first, int check)
861 {
862         int rc;
863
864         if (ldata->ld_leh == NULL) {
865                 rc = first ? -ENODATA : mdd_links_read(env, mdd_obj, ldata);
866                 if (rc) {
867                         if (rc != -ENODATA)
868                                 return rc;
869                         rc = linkea_data_new(ldata,
870                                              &mdd_env_info(env)->mti_link_buf);
871                         if (rc)
872                                 return rc;
873                 }
874         }
875
876         if (check) {
877                 rc = linkea_links_find(ldata, lname, pfid);
878                 if (rc && rc != -ENOENT)
879                         return rc;
880                 if (rc == 0)
881                         return -EEXIST;
882         }
883
884         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LINKEA_MORE)) {
885                 struct lu_fid *tfid = &mdd_env_info(env)->mti_fid2;
886
887                 *tfid = *pfid;
888                 tfid->f_ver = ~0;
889                 linkea_add_buf(ldata, lname, tfid);
890         }
891
892         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LINKEA_MORE2))
893                 linkea_add_buf(ldata, lname, pfid);
894
895         return linkea_add_buf(ldata, lname, pfid);
896 }
897
898 static int __mdd_links_del(const struct lu_env *env,
899                            struct mdd_object *mdd_obj,
900                            struct linkea_data *ldata,
901                            const struct lu_name *lname,
902                            const struct lu_fid *pfid)
903 {
904         int rc;
905
906         if (ldata->ld_leh == NULL) {
907                 rc = mdd_links_read(env, mdd_obj, ldata);
908                 if (rc)
909                         return rc;
910         }
911
912         rc = linkea_links_find(ldata, lname, pfid);
913         if (rc)
914                 return rc;
915
916         linkea_del_buf(ldata, lname);
917         return 0;
918 }
919
920 static int mdd_linkea_prepare(const struct lu_env *env,
921                               struct mdd_object *mdd_obj,
922                               const struct lu_fid *oldpfid,
923                               const struct lu_name *oldlname,
924                               const struct lu_fid *newpfid,
925                               const struct lu_name *newlname,
926                               int first, int check,
927                               struct linkea_data *ldata)
928 {
929         int rc = 0;
930         int rc2 = 0;
931         ENTRY;
932
933         if (OBD_FAIL_CHECK(OBD_FAIL_FID_IGIF))
934                 return 0;
935
936         LASSERT(oldpfid != NULL || newpfid != NULL);
937
938         if (mdd_obj->mod_flags & DEAD_OBJ) {
939                 /* Prevent linkea to be updated which is NOT necessary. */
940                 ldata->ld_reclen = 0;
941                 /* No more links, don't bother */
942                 RETURN(0);
943         }
944
945         if (oldpfid != NULL) {
946                 rc = __mdd_links_del(env, mdd_obj, ldata, oldlname, oldpfid);
947                 if (rc) {
948                         if ((check == 1) ||
949                             (rc != -ENODATA && rc != -ENOENT))
950                                 RETURN(rc);
951                         /* No changes done. */
952                         rc = 0;
953                 }
954         }
955
956         /* If renaming, add the new record */
957         if (newpfid != NULL) {
958                 /* even if the add fails, we still delete the out-of-date
959                  * old link */
960                 rc2 = __mdd_links_add(env, mdd_obj, ldata, newlname, newpfid,
961                                       first, check);
962         }
963
964         rc = rc != 0 ? rc : rc2;
965
966         RETURN(rc);
967 }
968
969 int mdd_links_rename(const struct lu_env *env,
970                      struct mdd_object *mdd_obj,
971                      const struct lu_fid *oldpfid,
972                      const struct lu_name *oldlname,
973                      const struct lu_fid *newpfid,
974                      const struct lu_name *newlname,
975                      struct thandle *handle,
976                      struct linkea_data *ldata,
977                      int first, int check)
978 {
979         int rc = 0;
980         ENTRY;
981
982         if (ldata == NULL) {
983                 ldata = &mdd_env_info(env)->mti_link_data;
984                 memset(ldata, 0, sizeof(*ldata));
985                 rc = mdd_linkea_prepare(env, mdd_obj, oldpfid, oldlname,
986                                         newpfid, newlname, first, check,
987                                         ldata);
988                 if (rc != 0)
989                         GOTO(out, rc);
990         }
991
992         if (ldata->ld_reclen != 0)
993                 rc = mdd_links_write(env, mdd_obj, ldata, handle);
994         EXIT;
995 out:
996         if (rc != 0) {
997                 int error = 1;
998                 if (rc == -EOVERFLOW || rc == -ENOSPC)
999                         error = 0;
1000                 if (oldpfid == NULL)
1001                         CDEBUG(error ? D_ERROR : D_OTHER,
1002                                "link_ea add '%.*s' failed %d "DFID"\n",
1003                                newlname->ln_namelen, newlname->ln_name,
1004                                rc, PFID(mdd_object_fid(mdd_obj)));
1005                 else if (newpfid == NULL)
1006                         CDEBUG(error ? D_ERROR : D_OTHER,
1007                                "link_ea del '%.*s' failed %d "DFID"\n",
1008                                oldlname->ln_namelen, oldlname->ln_name,
1009                                rc, PFID(mdd_object_fid(mdd_obj)));
1010                 else
1011                         CDEBUG(error ? D_ERROR : D_OTHER,
1012                                "link_ea rename '%.*s'->'%.*s' failed %d "
1013                                DFID"\n",
1014                                oldlname->ln_namelen, oldlname->ln_name,
1015                                newlname->ln_namelen, newlname->ln_name,
1016                                rc, PFID(mdd_object_fid(mdd_obj)));
1017         }
1018
1019         if (ldata->ld_buf && ldata->ld_buf->lb_len > OBD_ALLOC_BIG)
1020                 /* if we vmalloced a large buffer drop it */
1021                 lu_buf_free(ldata->ld_buf);
1022
1023         return rc;
1024 }
1025
1026 static inline int mdd_links_add(const struct lu_env *env,
1027                                 struct mdd_object *mdd_obj,
1028                                 const struct lu_fid *pfid,
1029                                 const struct lu_name *lname,
1030                                 struct thandle *handle,
1031                                 struct linkea_data *ldata, int first)
1032 {
1033         return mdd_links_rename(env, mdd_obj, NULL, NULL,
1034                                 pfid, lname, handle, ldata, first, 0);
1035 }
1036
1037 static inline int mdd_links_del(const struct lu_env *env,
1038                                 struct mdd_object *mdd_obj,
1039                                 const struct lu_fid *pfid,
1040                                 const struct lu_name *lname,
1041                                 struct thandle *handle)
1042 {
1043         return mdd_links_rename(env, mdd_obj, pfid, lname,
1044                                 NULL, NULL, handle, NULL, 0, 0);
1045 }
1046
1047 /** Read the link EA into a temp buffer.
1048  * Uses the mdd_thread_info::mti_big_buf since it is generally large.
1049  * A pointer to the buffer is stored in \a ldata::ld_buf.
1050  *
1051  * \retval 0 or error
1052  */
1053 int mdd_links_read(const struct lu_env *env, struct mdd_object *mdd_obj,
1054                    struct linkea_data *ldata)
1055 {
1056         int rc;
1057
1058         if (!mdd_object_exists(mdd_obj))
1059                 return -ENODATA;
1060
1061         /* First try a small buf */
1062         LASSERT(env != NULL);
1063         ldata->ld_buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_link_buf,
1064                                                PAGE_CACHE_SIZE);
1065         if (ldata->ld_buf->lb_buf == NULL)
1066                 return -ENOMEM;
1067
1068         rc = mdo_xattr_get(env, mdd_obj, ldata->ld_buf, XATTR_NAME_LINK,
1069                           BYPASS_CAPA);
1070         if (rc == -ERANGE) {
1071                 /* Buf was too small, figure out what we need. */
1072                 lu_buf_free(ldata->ld_buf);
1073                 rc = mdo_xattr_get(env, mdd_obj, ldata->ld_buf,
1074                                    XATTR_NAME_LINK, BYPASS_CAPA);
1075                 if (rc < 0)
1076                         return rc;
1077                 ldata->ld_buf = lu_buf_check_and_alloc(ldata->ld_buf, rc);
1078                 if (ldata->ld_buf->lb_buf == NULL)
1079                         return -ENOMEM;
1080                 rc = mdo_xattr_get(env, mdd_obj, ldata->ld_buf,
1081                                   XATTR_NAME_LINK, BYPASS_CAPA);
1082         }
1083         if (rc < 0) {
1084                 lu_buf_free(ldata->ld_buf);
1085                 ldata->ld_buf = NULL;
1086                 return rc;
1087         }
1088
1089         return linkea_init(ldata);
1090 }
1091
1092 /** Read the link EA into a temp buffer.
1093  * Uses the name_buf since it is generally large.
1094  * \retval IS_ERR err
1095  * \retval ptr to \a lu_buf (always \a mti_big_buf)
1096  */
1097 struct lu_buf *mdd_links_get(const struct lu_env *env,
1098                              struct mdd_object *mdd_obj)
1099 {
1100         struct linkea_data ldata = { NULL };
1101         int rc;
1102
1103         rc = mdd_links_read(env, mdd_obj, &ldata);
1104         return rc ? ERR_PTR(rc) : ldata.ld_buf;
1105 }
1106
1107 int mdd_links_write(const struct lu_env *env, struct mdd_object *mdd_obj,
1108                     struct linkea_data *ldata, struct thandle *handle)
1109 {
1110         const struct lu_buf *buf = mdd_buf_get_const(env, ldata->ld_buf->lb_buf,
1111                                                      ldata->ld_leh->leh_len);
1112         int                 rc;
1113
1114         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NO_LINKEA))
1115                 return 0;
1116
1117         rc = mdo_xattr_set(env, mdd_obj, buf, XATTR_NAME_LINK, 0, handle,
1118                            mdd_object_capa(env, mdd_obj));
1119         if (unlikely(rc == -ENOSPC) && S_ISREG(mdd_object_type(mdd_obj)) &&
1120             mdd_object_remote(mdd_obj) == 0) {
1121                 struct lfsck_request *lr = &mdd_env_info(env)->mti_lr;
1122
1123                 /* XXX: If the linkEA is overflow, then we need to notify the
1124                  *      namespace LFSCK to skip "nlink" attribute verification
1125                  *      on this object to avoid the "nlink" to be shrinked by
1126                  *      wrong. It may be not good an interaction with LFSCK
1127                  *      like this. We will consider to replace it with other
1128                  *      mechanism in future. LU-5802. */
1129                 lfsck_pack_rfa(lr, mdo2fid(mdd_obj), LE_SKIP_NLINK,
1130                                LFSCK_TYPE_NAMESPACE);
1131                 lfsck_in_notify(env, mdo2mdd(&mdd_obj->mod_obj)->mdd_bottom,
1132                                 lr, handle);
1133         }
1134
1135         return rc;
1136 }
1137
1138 int mdd_declare_links_add(const struct lu_env *env, struct mdd_object *mdd_obj,
1139                           struct thandle *handle, struct linkea_data *ldata,
1140                           enum mdd_links_add_overflow overflow)
1141 {
1142         int     rc;
1143         int     ea_len;
1144         void    *linkea;
1145
1146         if (ldata != NULL && ldata->ld_leh != NULL) {
1147                 ea_len = ldata->ld_leh->leh_len;
1148                 linkea = ldata->ld_buf->lb_buf;
1149         } else {
1150                 ea_len = DEFAULT_LINKEA_SIZE;
1151                 linkea = NULL;
1152         }
1153
1154         /* XXX: max size? */
1155         rc = mdo_declare_xattr_set(env, mdd_obj,
1156                                    mdd_buf_get_const(env, linkea, ea_len),
1157                                    XATTR_NAME_LINK, 0, handle);
1158         if (rc != 0)
1159                 return rc;
1160
1161         if (mdd_object_remote(mdd_obj) == 0 && overflow == MLAO_CHECK) {
1162                 struct lfsck_request *lr = &mdd_env_info(env)->mti_lr;
1163
1164                 /* XXX: If the linkEA is overflow, then we need to notify the
1165                  *      namespace LFSCK to skip "nlink" attribute verification
1166                  *      on this object to avoid the "nlink" to be shrinked by
1167                  *      wrong. It may be not good an interaction with LFSCK
1168                  *      like this. We will consider to replace it with other
1169                  *      mechanism in future. LU-5802. */
1170                 lfsck_pack_rfa(lr, mdo2fid(mdd_obj), LE_SKIP_NLINK_DECLARE,
1171                                LFSCK_TYPE_NAMESPACE);
1172                 rc = lfsck_in_notify(env,
1173                                      mdo2mdd(&mdd_obj->mod_obj)->mdd_bottom,
1174                                      lr, handle);
1175         }
1176
1177         return rc;
1178 }
1179
1180 static inline int mdd_declare_links_del(const struct lu_env *env,
1181                                         struct mdd_object *c,
1182                                         struct thandle *handle)
1183 {
1184         int rc = 0;
1185
1186         /* For directory, the linkEA will be removed together
1187          * with the object. */
1188         if (!S_ISDIR(mdd_object_type(c)))
1189                 rc = mdd_declare_links_add(env, c, handle, NULL, MLAO_IGNORE);
1190
1191         return rc;
1192 }
1193
1194 static int mdd_declare_link(const struct lu_env *env,
1195                             struct mdd_device *mdd,
1196                             struct mdd_object *p,
1197                             struct mdd_object *c,
1198                             const struct lu_name *name,
1199                             struct thandle *handle,
1200                             struct lu_attr *la,
1201                             struct linkea_data *data)
1202 {
1203         int rc;
1204
1205         rc = mdo_declare_index_insert(env, p, mdo2fid(c), mdd_object_type(c),
1206                                       name->ln_name, handle);
1207         if (rc != 0)
1208                 return rc;
1209
1210         rc = mdo_declare_ref_add(env, c, handle);
1211         if (rc != 0)
1212                 return rc;
1213
1214         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_MORE_NLINK)) {
1215                 rc = mdo_declare_ref_add(env, c, handle);
1216                 if (rc != 0)
1217                         return rc;
1218         }
1219
1220         la->la_valid = LA_CTIME | LA_MTIME;
1221         rc = mdo_declare_attr_set(env, p, la, handle);
1222         if (rc != 0)
1223                 return rc;
1224
1225         la->la_valid = LA_CTIME;
1226         rc = mdo_declare_attr_set(env, c, la, handle);
1227         if (rc != 0)
1228                 return rc;
1229
1230         rc = mdd_declare_links_add(env, c, handle, data,
1231                         S_ISREG(mdd_object_type(c)) ? MLAO_CHECK : MLAO_IGNORE);
1232         if (rc != 0)
1233                 return rc;
1234
1235         rc = mdd_declare_changelog_store(env, mdd, name, NULL, handle);
1236
1237         return rc;
1238 }
1239
1240 static int mdd_link(const struct lu_env *env, struct md_object *tgt_obj,
1241                     struct md_object *src_obj, const struct lu_name *lname,
1242                     struct md_attr *ma)
1243 {
1244         const char *name = lname->ln_name;
1245         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
1246         struct mdd_object *mdd_tobj = md2mdd_obj(tgt_obj);
1247         struct mdd_object *mdd_sobj = md2mdd_obj(src_obj);
1248         struct lu_attr    *cattr = MDD_ENV_VAR(env, cattr);
1249         struct lu_attr    *tattr = MDD_ENV_VAR(env, tattr);
1250         struct mdd_device *mdd = mdo2mdd(src_obj);
1251         struct thandle *handle;
1252         struct linkea_data *ldata = &mdd_env_info(env)->mti_link_data;
1253         int rc;
1254         ENTRY;
1255
1256         rc = mdd_la_get(env, mdd_sobj, cattr, BYPASS_CAPA);
1257         if (rc != 0)
1258                 RETURN(rc);
1259
1260         rc = mdd_la_get(env, mdd_tobj, tattr, BYPASS_CAPA);
1261         if (rc != 0)
1262                 RETURN(rc);
1263
1264         handle = mdd_trans_create(env, mdd);
1265         if (IS_ERR(handle))
1266                 GOTO(out_pending, rc = PTR_ERR(handle));
1267
1268         memset(ldata, 0, sizeof(*ldata));
1269
1270         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1271         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1272
1273         rc = mdd_declare_link(env, mdd, mdd_tobj, mdd_sobj, lname, handle,
1274                               la, ldata);
1275         if (rc)
1276                 GOTO(stop, rc);
1277
1278         rc = mdd_trans_start(env, mdd, handle);
1279         if (rc)
1280                 GOTO(stop, rc);
1281
1282         mdd_write_lock(env, mdd_sobj, MOR_TGT_CHILD);
1283         rc = mdd_link_sanity_check(env, mdd_tobj, tattr, lname, mdd_sobj,
1284                                    cattr);
1285         if (rc)
1286                 GOTO(out_unlock, rc);
1287
1288         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LESS_NLINK)) {
1289                 rc = mdo_ref_add(env, mdd_sobj, handle);
1290                 if (rc != 0)
1291                         GOTO(out_unlock, rc);
1292         }
1293
1294         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_MORE_NLINK)) {
1295                 rc = mdo_ref_add(env, mdd_sobj, handle);
1296                 if (rc != 0)
1297                         GOTO(out_unlock, rc);
1298         }
1299
1300         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DANGLING3)) {
1301                 struct lu_fid tfid = *mdo2fid(mdd_sobj);
1302
1303                 tfid.f_oid++;
1304                 rc = __mdd_index_insert_only(env, mdd_tobj, &tfid,
1305                                              mdd_object_type(mdd_sobj),
1306                                              name, handle,
1307                                              mdd_object_capa(env, mdd_tobj));
1308         } else {
1309                 rc = __mdd_index_insert_only(env, mdd_tobj, mdo2fid(mdd_sobj),
1310                                              mdd_object_type(mdd_sobj),
1311                                              name, handle,
1312                                              mdd_object_capa(env, mdd_tobj));
1313         }
1314
1315         if (rc != 0) {
1316                 mdo_ref_del(env, mdd_sobj, handle);
1317                 GOTO(out_unlock, rc);
1318         }
1319
1320         la->la_valid = LA_CTIME | LA_MTIME;
1321         rc = mdd_update_time(env, mdd_tobj, tattr, la, handle);
1322         if (rc)
1323                 GOTO(out_unlock, rc);
1324
1325         la->la_valid = LA_CTIME;
1326         rc = mdd_update_time(env, mdd_sobj, cattr, la, handle);
1327         if (rc == 0) {
1328                 rc = mdd_linkea_prepare(env, mdd_sobj, NULL, NULL,
1329                                         mdo2fid(mdd_tobj), lname, 0, 0,
1330                                         ldata);
1331                 if (rc == 0)
1332                         mdd_links_add(env, mdd_sobj, mdo2fid(mdd_tobj),
1333                                       lname, handle, ldata, 0);
1334                 /* The failure of links_add should not cause the link
1335                  * failure, reset rc here */
1336                 rc = 0;
1337         }
1338         EXIT;
1339 out_unlock:
1340         mdd_write_unlock(env, mdd_sobj);
1341         if (rc == 0)
1342                 rc = mdd_changelog_ns_store(env, mdd, CL_HARDLINK, 0, mdd_sobj,
1343                                             mdo2fid(mdd_tobj), NULL, NULL,
1344                                             lname, NULL, handle);
1345 stop:
1346         mdd_trans_stop(env, mdd, rc, handle);
1347
1348         if (ldata->ld_buf && ldata->ld_buf->lb_len > OBD_ALLOC_BIG)
1349                 /* if we vmalloced a large buffer drop it */
1350                 lu_buf_free(ldata->ld_buf);
1351 out_pending:
1352         return rc;
1353 }
1354
1355 static int mdd_mark_dead_object(const struct lu_env *env,
1356                                 struct mdd_object *obj, struct thandle *handle,
1357                                 bool declare)
1358 {
1359         struct lu_attr *attr = MDD_ENV_VAR(env, la_for_start);
1360         int rc;
1361
1362         if (!declare)
1363                 obj->mod_flags |= DEAD_OBJ;
1364
1365         if (!S_ISDIR(mdd_object_type(obj)))
1366                 return 0;
1367
1368         attr->la_valid = LA_FLAGS;
1369         attr->la_flags = LUSTRE_SLAVE_DEAD_FL;
1370
1371         if (declare)
1372                 rc = mdo_declare_attr_set(env, obj, attr, handle);
1373         else
1374                 rc = mdo_attr_set(env, obj, attr, handle,
1375                                   mdd_object_capa(env, obj));
1376
1377         return rc;
1378 }
1379
1380 static int mdd_declare_finish_unlink(const struct lu_env *env,
1381                                      struct mdd_object *obj,
1382                                      struct thandle *handle)
1383 {
1384         int     rc;
1385
1386         rc = mdd_mark_dead_object(env, obj, handle, true);
1387         if (rc != 0)
1388                 return rc;
1389
1390         rc = orph_declare_index_insert(env, obj, mdd_object_type(obj), handle);
1391         if (rc != 0)
1392                 return rc;
1393
1394         rc = mdo_declare_destroy(env, obj, handle);
1395         if (rc != 0)
1396                 return rc;
1397
1398         return mdd_declare_links_del(env, obj, handle);
1399 }
1400
1401 /* caller should take a lock before calling */
1402 int mdd_finish_unlink(const struct lu_env *env,
1403                       struct mdd_object *obj, struct md_attr *ma,
1404                       const struct mdd_object *pobj,
1405                       const struct lu_name *lname,
1406                       struct thandle *th)
1407 {
1408         int rc = 0;
1409         int is_dir = S_ISDIR(ma->ma_attr.la_mode);
1410         ENTRY;
1411
1412         LASSERT(mdd_write_locked(env, obj) != 0);
1413
1414         if (ma->ma_attr.la_nlink == 0 || is_dir) {
1415                 rc = mdd_mark_dead_object(env, obj, th, false);
1416                 if (rc != 0)
1417                         RETURN(rc);
1418
1419                 /* add new orphan and the object
1420                  * will be deleted during mdd_close() */
1421                 if (obj->mod_count) {
1422                         rc = __mdd_orphan_add(env, obj, th);
1423                         if (rc == 0)
1424                                 CDEBUG(D_HA, "Object "DFID" is inserted into "
1425                                         "orphan list, open count = %d\n",
1426                                         PFID(mdd_object_fid(obj)),
1427                                         obj->mod_count);
1428                         else
1429                                 CERROR("Object "DFID" fail to be an orphan, "
1430                                        "open count = %d, maybe cause failed "
1431                                        "open replay\n",
1432                                         PFID(mdd_object_fid(obj)),
1433                                         obj->mod_count);
1434                 } else {
1435                         rc = mdo_destroy(env, obj, th);
1436                 }
1437         } else if (!is_dir) {
1438                 /* old files may not have link ea; ignore errors */
1439                 mdd_links_del(env, obj, mdo2fid(pobj), lname, th);
1440         }
1441
1442         RETURN(rc);
1443 }
1444
1445 /*
1446  * pobj maybe NULL
1447  * has mdd_write_lock on cobj already, but not on pobj yet
1448  */
1449 int mdd_unlink_sanity_check(const struct lu_env *env, struct mdd_object *pobj,
1450                             const struct lu_attr *pattr,
1451                             struct mdd_object *cobj,
1452                             const struct lu_attr *cattr)
1453 {
1454         int rc;
1455         ENTRY;
1456
1457         rc = mdd_may_delete(env, pobj, pattr, cobj, cattr, NULL, 1, 1);
1458
1459         RETURN(rc);
1460 }
1461
1462 static int mdd_declare_unlink(const struct lu_env *env, struct mdd_device *mdd,
1463                               struct mdd_object *p, struct mdd_object *c,
1464                               const struct lu_name *name, struct md_attr *ma,
1465                               struct thandle *handle, int no_name, int is_dir)
1466 {
1467         struct lu_attr  *la = &mdd_env_info(env)->mti_la_for_fix;
1468         int              rc;
1469
1470         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DANGLING2)) {
1471                 if (likely(no_name == 0)) {
1472                         rc = mdo_declare_index_delete(env, p, name->ln_name,
1473                                                       handle);
1474                         if (rc != 0)
1475                                 return rc;
1476                 }
1477
1478                 if (is_dir != 0) {
1479                         rc = mdo_declare_ref_del(env, p, handle);
1480                         if (rc != 0)
1481                                 return rc;
1482                 }
1483         }
1484
1485         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1486         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1487         la->la_valid = LA_CTIME | LA_MTIME;
1488         rc = mdo_declare_attr_set(env, p, la, handle);
1489         if (rc)
1490                 return rc;
1491
1492         if (c != NULL) {
1493                 rc = mdo_declare_ref_del(env, c, handle);
1494                 if (rc)
1495                         return rc;
1496
1497                 rc = mdo_declare_ref_del(env, c, handle);
1498                 if (rc)
1499                         return rc;
1500
1501                 la->la_valid = LA_CTIME;
1502                 rc = mdo_declare_attr_set(env, c, la, handle);
1503                 if (rc)
1504                         return rc;
1505
1506                 rc = mdd_declare_finish_unlink(env, c, handle);
1507                 if (rc)
1508                         return rc;
1509
1510                 /* FIXME: need changelog for remove entry */
1511                 rc = mdd_declare_changelog_store(env, mdd, name, NULL, handle);
1512         }
1513
1514         return rc;
1515 }
1516
1517 /*
1518  * test if a file has an HSM archive
1519  * if HSM attributes are not found in ma update them from
1520  * HSM xattr
1521  */
1522 static bool mdd_hsm_archive_exists(const struct lu_env *env,
1523                                    struct mdd_object *obj,
1524                                    struct md_attr *ma)
1525 {
1526         ENTRY;
1527
1528         if (!(ma->ma_valid & MA_HSM)) {
1529                 /* no HSM MD provided, read xattr */
1530                 struct lu_buf   *hsm_buf;
1531                 const size_t     buflen = sizeof(struct hsm_attrs);
1532                 int              rc;
1533
1534                 hsm_buf = mdd_buf_get(env, NULL, 0);
1535                 lu_buf_alloc(hsm_buf, buflen);
1536                 rc = mdo_xattr_get(env, obj, hsm_buf, XATTR_NAME_HSM,
1537                                    mdd_object_capa(env, obj));
1538                 rc = lustre_buf2hsm(hsm_buf->lb_buf, rc, &ma->ma_hsm);
1539                 lu_buf_free(hsm_buf);
1540                 if (rc < 0)
1541                         RETURN(false);
1542
1543                 ma->ma_valid = MA_HSM;
1544         }
1545         if (ma->ma_hsm.mh_flags & HS_EXISTS)
1546                 RETURN(true);
1547         RETURN(false);
1548 }
1549
1550 /**
1551  * Delete name entry and the object.
1552  * Note: no_name == 1 means it only destory the object, i.e. name_entry
1553  * does not exist for this object, and it could only happen during resending
1554  * of remote unlink. see the comments in mdt_reint_unlink. Unfortunately, lname
1555  * is also needed in this case(needed by changelog), so we have to add another
1556  * parameter(no_name)here. XXX: this is only needed in DNE phase I, on Phase II,
1557  * the ENOENT failure should be able to be fixed by redo mechanism.
1558  */
1559 static int mdd_unlink(const struct lu_env *env, struct md_object *pobj,
1560                       struct md_object *cobj, const struct lu_name *lname,
1561                       struct md_attr *ma, int no_name)
1562 {
1563         const char *name = lname->ln_name;
1564         struct lu_attr *pattr = MDD_ENV_VAR(env, pattr);
1565         struct lu_attr *cattr = MDD_ENV_VAR(env, cattr);
1566         struct lu_attr *la = &mdd_env_info(env)->mti_la_for_fix;
1567         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1568         struct mdd_object *mdd_cobj = NULL;
1569         struct mdd_device *mdd = mdo2mdd(pobj);
1570         struct thandle    *handle;
1571         int rc, is_dir = 0;
1572         ENTRY;
1573
1574         /* cobj == NULL means only delete name entry */
1575         if (likely(cobj != NULL)) {
1576                 mdd_cobj = md2mdd_obj(cobj);
1577                 if (mdd_object_exists(mdd_cobj) == 0)
1578                         RETURN(-ENOENT);
1579         }
1580
1581         rc = mdd_la_get(env, mdd_pobj, pattr, BYPASS_CAPA);
1582         if (rc)
1583                 RETURN(rc);
1584
1585         if (likely(mdd_cobj != NULL)) {
1586                 /* fetch cattr */
1587                 rc = mdd_la_get(env, mdd_cobj, cattr, BYPASS_CAPA);
1588                 if (rc)
1589                         RETURN(rc);
1590
1591                 is_dir = S_ISDIR(cattr->la_mode);
1592         }
1593
1594         rc = mdd_unlink_sanity_check(env, mdd_pobj, pattr, mdd_cobj, cattr);
1595         if (rc)
1596                 RETURN(rc);
1597
1598         handle = mdd_trans_create(env, mdd);
1599         if (IS_ERR(handle))
1600                 RETURN(PTR_ERR(handle));
1601
1602         rc = mdd_declare_unlink(env, mdd, mdd_pobj, mdd_cobj,
1603                                 lname, ma, handle, no_name, is_dir);
1604         if (rc)
1605                 GOTO(stop, rc);
1606
1607         rc = mdd_trans_start(env, mdd, handle);
1608         if (rc)
1609                 GOTO(stop, rc);
1610
1611         if (likely(mdd_cobj != NULL))
1612                 mdd_write_lock(env, mdd_cobj, MOR_TGT_CHILD);
1613
1614         if (likely(no_name == 0) && !OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DANGLING2)) {
1615                 rc = __mdd_index_delete(env, mdd_pobj, name, is_dir, handle,
1616                                         mdd_object_capa(env, mdd_pobj));
1617                 if (rc)
1618                         GOTO(cleanup, rc);
1619         }
1620
1621         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_MUL_REF) ||
1622             OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NO_NAMEENTRY))
1623                 GOTO(cleanup, rc = 0);
1624
1625         if (likely(mdd_cobj != NULL)) {
1626                 rc = mdo_ref_del(env, mdd_cobj, handle);
1627                 if (rc != 0) {
1628                         __mdd_index_insert_only(env, mdd_pobj,
1629                                                 mdo2fid(mdd_cobj),
1630                                                 mdd_object_type(mdd_cobj),
1631                                                 name, handle,
1632                                                 mdd_object_capa(env, mdd_pobj));
1633                         GOTO(cleanup, rc);
1634                 }
1635
1636                 if (is_dir)
1637                         /* unlink dot */
1638                         mdo_ref_del(env, mdd_cobj, handle);
1639
1640                 /* fetch updated nlink */
1641                 rc = mdd_la_get(env, mdd_cobj, cattr, BYPASS_CAPA);
1642                 if (rc)
1643                         GOTO(cleanup, rc);
1644         }
1645
1646         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1647         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1648
1649         la->la_valid = LA_CTIME | LA_MTIME;
1650         rc = mdd_update_time(env, mdd_pobj, pattr, la, handle);
1651         if (rc)
1652                 GOTO(cleanup, rc);
1653
1654         /* Enough for only unlink the entry */
1655         if (unlikely(mdd_cobj == NULL))
1656                 GOTO(stop, rc);
1657
1658         if (cattr->la_nlink > 0 || mdd_cobj->mod_count > 0) {
1659                 /* update ctime of an unlinked file only if it is still
1660                  * opened or a link still exists */
1661                 la->la_valid = LA_CTIME;
1662                 rc = mdd_update_time(env, mdd_cobj, cattr, la, handle);
1663                 if (rc)
1664                         GOTO(cleanup, rc);
1665         }
1666
1667         /* XXX: this transfer to ma will be removed with LOD/OSP */
1668         ma->ma_attr = *cattr;
1669         ma->ma_valid |= MA_INODE;
1670         rc = mdd_finish_unlink(env, mdd_cobj, ma, mdd_pobj, lname, handle);
1671
1672         /* fetch updated nlink */
1673         if (rc == 0)
1674                 rc = mdd_la_get(env, mdd_cobj, cattr, BYPASS_CAPA);
1675
1676         /* if object is removed then we can't get its attrs, use last get */
1677         if (cattr->la_nlink == 0) {
1678                 ma->ma_attr = *cattr;
1679                 ma->ma_valid |= MA_INODE;
1680         }
1681         EXIT;
1682 cleanup:
1683         if (likely(mdd_cobj != NULL))
1684                 mdd_write_unlock(env, mdd_cobj);
1685
1686         if (rc == 0) {
1687                 int cl_flags = 0;
1688
1689                 if (cattr->la_nlink == 0) {
1690                         cl_flags |= CLF_UNLINK_LAST;
1691                         /* search for an existing archive */
1692                         if (mdd_hsm_archive_exists(env, mdd_cobj, ma))
1693                                 cl_flags |= CLF_UNLINK_HSM_EXISTS;
1694                 }
1695
1696                 rc = mdd_changelog_ns_store(env, mdd,
1697                         is_dir ? CL_RMDIR : CL_UNLINK, cl_flags,
1698                         mdd_cobj, mdo2fid(mdd_pobj), NULL, NULL, lname, NULL,
1699                         handle);
1700         }
1701
1702 stop:
1703         mdd_trans_stop(env, mdd, rc, handle);
1704
1705         return rc;
1706 }
1707
1708 /*
1709  * The permission has been checked when obj created, no need check again.
1710  */
1711 static int mdd_cd_sanity_check(const struct lu_env *env,
1712                                struct mdd_object *obj)
1713 {
1714         ENTRY;
1715
1716         /* EEXIST check */
1717         if (!obj || mdd_is_dead_obj(obj))
1718                 RETURN(-ENOENT);
1719
1720         RETURN(0);
1721 }
1722
1723 static int mdd_create_data(const struct lu_env *env, struct md_object *pobj,
1724                            struct md_object *cobj, const struct md_op_spec *spec,
1725                            struct md_attr *ma)
1726 {
1727         struct mdd_device *mdd = mdo2mdd(cobj);
1728         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1729         struct mdd_object *son = md2mdd_obj(cobj);
1730         struct thandle    *handle;
1731         const struct lu_buf *buf;
1732         struct lu_attr    *attr = MDD_ENV_VAR(env, cattr);
1733         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
1734         int                rc;
1735         ENTRY;
1736
1737         rc = mdd_cd_sanity_check(env, son);
1738         if (rc)
1739                 RETURN(rc);
1740
1741         if (!md_should_create(spec->sp_cr_flags))
1742                 RETURN(0);
1743
1744         /*
1745          * there are following use cases for this function:
1746          * 1) late striping - file was created with MDS_OPEN_DELAY_CREATE
1747          *    striping can be specified or not
1748          * 2) CMD?
1749          */
1750         rc = mdd_la_get(env, son, attr, BYPASS_CAPA);
1751         if (rc)
1752                 RETURN(rc);
1753
1754         /* calling ->ah_make_hint() is used to transfer information from parent */
1755         mdd_object_make_hint(env, mdd_pobj, son, attr, spec, hint);
1756
1757         handle = mdd_trans_create(env, mdd);
1758         if (IS_ERR(handle))
1759                 GOTO(out_free, rc = PTR_ERR(handle));
1760
1761         /*
1762          * XXX: Setting the lov ea is not locked but setting the attr is locked?
1763          * Should this be fixed?
1764          */
1765         CDEBUG(D_OTHER, "ea %p/%u, cr_flags "LPO64", no_create %u\n",
1766                spec->u.sp_ea.eadata, spec->u.sp_ea.eadatalen,
1767                spec->sp_cr_flags, spec->no_create);
1768
1769         if (spec->no_create || (spec->sp_cr_flags & MDS_OPEN_HAS_EA)) {
1770                 buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
1771                                         spec->u.sp_ea.eadatalen);
1772         } else {
1773                 buf = &LU_BUF_NULL;
1774         }
1775
1776         rc = dt_declare_xattr_set(env, mdd_object_child(son), buf,
1777                                   XATTR_NAME_LOV, 0, handle);
1778         if (rc)
1779                 GOTO(stop, rc);
1780
1781         rc = mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
1782         if (rc)
1783                 GOTO(stop, rc);
1784
1785         rc = mdd_trans_start(env, mdd, handle);
1786         if (rc)
1787                 GOTO(stop, rc);
1788
1789         rc = dt_xattr_set(env, mdd_object_child(son), buf, XATTR_NAME_LOV,
1790                           0, handle, mdd_object_capa(env, son));
1791
1792         if (rc)
1793                 GOTO(stop, rc);
1794
1795         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, son, handle);
1796
1797 stop:
1798         mdd_trans_stop(env, mdd, rc, handle);
1799 out_free:
1800         RETURN(rc);
1801 }
1802
1803 static int mdd_declare_object_initialize(const struct lu_env *env,
1804                                          struct mdd_object *parent,
1805                                          struct mdd_object *child,
1806                                          struct lu_attr *attr,
1807                                          struct thandle *handle)
1808 {
1809         int rc;
1810         ENTRY;
1811
1812         /*
1813          * inode mode has been set in creation time, and it's based on umask,
1814          * la_mode and acl, don't set here again! (which will go wrong
1815          * because below function doesn't consider umask).
1816          * I'd suggest set all object attributes in creation time, see above.
1817          */
1818         LASSERT(attr->la_valid & (LA_MODE | LA_TYPE));
1819         attr->la_valid &= ~(LA_MODE | LA_TYPE);
1820         rc = mdo_declare_attr_set(env, child, attr, handle);
1821         attr->la_valid |= LA_MODE | LA_TYPE;
1822         if (rc != 0 || !S_ISDIR(attr->la_mode))
1823                 RETURN(rc);
1824
1825         rc = mdo_declare_index_insert(env, child, mdo2fid(child), S_IFDIR,
1826                                       dot, handle);
1827         if (rc != 0)
1828                 RETURN(rc);
1829
1830         rc = mdo_declare_ref_add(env, child, handle);
1831         if (rc != 0)
1832                 RETURN(rc);
1833
1834         rc = mdo_declare_index_insert(env, child, mdo2fid(parent), S_IFDIR,
1835                                       dotdot, handle);
1836
1837         RETURN(rc);
1838 }
1839
1840 static int mdd_object_initialize(const struct lu_env *env,
1841                                  const struct lu_fid *pfid,
1842                                  struct mdd_object *child,
1843                                  struct lu_attr *attr, struct thandle *handle,
1844                                  const struct md_op_spec *spec)
1845 {
1846         int rc = 0;
1847         ENTRY;
1848
1849         if (S_ISDIR(attr->la_mode)) {
1850                 /* Add "." and ".." for newly created dir */
1851                 mdo_ref_add(env, child, handle);
1852                 rc = __mdd_index_insert_only(env, child, mdo2fid(child),
1853                                              S_IFDIR, dot, handle, BYPASS_CAPA);
1854                 if (rc == 0)
1855                         rc = __mdd_index_insert_only(env, child, pfid, S_IFDIR,
1856                                                      dotdot, handle,
1857                                                      BYPASS_CAPA);
1858                 if (rc != 0)
1859                         mdo_ref_del(env, child, handle);
1860         }
1861
1862         RETURN(rc);
1863 }
1864
1865 /**
1866  * This function checks whether it can create a file/dir under the
1867  * directory(@pobj). The directory(@pobj) is not being locked by
1868  * mdd lock.
1869  *
1870  * \param[in] env       execution environment
1871  * \param[in] pobj      the directory to create files
1872  * \param[in] pattr     the attributes of the directory
1873  * \param[in] lname     the name of the created file/dir
1874  * \param[in] cattr     the attributes of the file/dir
1875  * \param[in] spec      create specification
1876  *
1877  * \retval              = 0 it is allowed to create file/dir under
1878  *                      the directory
1879  * \retval              negative error not allowed to create file/dir
1880  *                      under the directory
1881  */
1882 static int mdd_create_sanity_check(const struct lu_env *env,
1883                                    struct md_object *pobj,
1884                                    const struct lu_attr *pattr,
1885                                    const struct lu_name *lname,
1886                                    struct lu_attr *cattr,
1887                                    struct md_op_spec *spec)
1888 {
1889         struct mdd_thread_info *info = mdd_env_info(env);
1890         struct lu_fid     *fid       = &info->mti_fid;
1891         struct mdd_object *obj       = md2mdd_obj(pobj);
1892         struct mdd_device *m         = mdo2mdd(pobj);
1893         bool            check_perm = true;
1894         int rc;
1895         ENTRY;
1896
1897         /* EEXIST check */
1898         if (mdd_is_dead_obj(obj))
1899                 RETURN(-ENOENT);
1900
1901         /*
1902          * In some cases this lookup is not needed - we know before if name
1903          * exists or not because MDT performs lookup for it.
1904          * name length check is done in lookup.
1905          */
1906         if (spec->sp_cr_lookup) {
1907                 /*
1908                  * Check if the name already exist, though it will be checked in
1909                  * _index_insert also, for avoiding rolling back if exists
1910                  * _index_insert.
1911                  */
1912                 rc = __mdd_lookup(env, pobj, pattr, lname, fid,
1913                                   MAY_WRITE | MAY_EXEC);
1914                 if (rc != -ENOENT)
1915                         RETURN(rc ? : -EEXIST);
1916
1917                 /* Permission is already being checked in mdd_lookup */
1918                 check_perm = false;
1919         }
1920
1921         rc = mdd_may_create(env, obj, pattr, NULL, check_perm);
1922         if (rc != 0)
1923                 RETURN(rc);
1924
1925         /* sgid check */
1926         if (pattr->la_mode & S_ISGID) {
1927                 cattr->la_gid = pattr->la_gid;
1928                 if (S_ISDIR(cattr->la_mode)) {
1929                         cattr->la_mode |= S_ISGID;
1930                         cattr->la_valid |= LA_MODE;
1931                 }
1932         }
1933
1934         rc = mdd_name_check(m, lname);
1935         if (rc < 0)
1936                 RETURN(rc);
1937
1938         switch (cattr->la_mode & S_IFMT) {
1939         case S_IFLNK: {
1940                 unsigned int symlen = strlen(spec->u.sp_symname) + 1;
1941
1942                 if (symlen > (1 << m->mdd_dt_conf.ddp_block_shift))
1943                         RETURN(-ENAMETOOLONG);
1944                 else
1945                         RETURN(0);
1946         }
1947         case S_IFDIR:
1948         case S_IFREG:
1949         case S_IFCHR:
1950         case S_IFBLK:
1951         case S_IFIFO:
1952         case S_IFSOCK:
1953                 rc = 0;
1954                 break;
1955         default:
1956                 rc = -EINVAL;
1957                 break;
1958         }
1959         RETURN(rc);
1960 }
1961
1962 static int mdd_declare_object_create(const struct lu_env *env,
1963                                      struct mdd_device *mdd,
1964                                      struct mdd_object *p, struct mdd_object *c,
1965                                      struct lu_attr *attr,
1966                                      struct thandle *handle,
1967                                      const struct md_op_spec *spec,
1968                                      struct lu_buf *def_acl_buf,
1969                                      struct lu_buf *acl_buf,
1970                                      struct dt_allocation_hint *hint)
1971 {
1972         int rc;
1973
1974         rc = mdd_declare_object_create_internal(env, p, c, attr, handle, spec,
1975                                                 hint);
1976         if (rc)
1977                 GOTO(out, rc);
1978
1979 #ifdef CONFIG_FS_POSIX_ACL
1980         if (def_acl_buf->lb_len > 0 && S_ISDIR(attr->la_mode)) {
1981                 /* if dir, then can inherit default ACl */
1982                 rc = mdo_declare_xattr_set(env, c, def_acl_buf,
1983                                            XATTR_NAME_ACL_DEFAULT,
1984                                            0, handle);
1985                 if (rc)
1986                         GOTO(out, rc);
1987         }
1988
1989         if (acl_buf->lb_len > 0) {
1990                 rc = mdo_declare_attr_set(env, c, attr, handle);
1991                 if (rc)
1992                         GOTO(out, rc);
1993
1994                 rc = mdo_declare_xattr_set(env, c, acl_buf,
1995                                            XATTR_NAME_ACL_ACCESS, 0, handle);
1996                 if (rc)
1997                         GOTO(out, rc);
1998         }
1999 #endif
2000         rc = mdd_declare_object_initialize(env, p, c, attr, handle);
2001         if (rc)
2002                 GOTO(out, rc);
2003
2004         /* replay case, create LOV EA from client data */
2005         if (spec->no_create ||
2006             (spec->sp_cr_flags & MDS_OPEN_HAS_EA && S_ISREG(attr->la_mode))) {
2007                 const struct lu_buf *buf;
2008
2009                 buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
2010                                         spec->u.sp_ea.eadatalen);
2011                 rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_LOV, 0,
2012                                            handle);
2013                 if (rc)
2014                         GOTO(out, rc);
2015         }
2016
2017         if (S_ISLNK(attr->la_mode)) {
2018                 const char *target_name = spec->u.sp_symname;
2019                 int sym_len = strlen(target_name);
2020                 const struct lu_buf *buf;
2021
2022                 buf = mdd_buf_get_const(env, target_name, sym_len);
2023                 rc = dt_declare_record_write(env, mdd_object_child(c),
2024                                              buf, 0, handle);
2025                 if (rc)
2026                         GOTO(out, rc);
2027         }
2028 out:
2029         return rc;
2030 }
2031
2032 static int mdd_declare_create(const struct lu_env *env, struct mdd_device *mdd,
2033                               struct mdd_object *p, struct mdd_object *c,
2034                               const struct lu_name *name,
2035                               struct lu_attr *attr,
2036                               struct thandle *handle,
2037                               const struct md_op_spec *spec,
2038                               struct linkea_data *ldata,
2039                               struct lu_buf *def_acl_buf,
2040                               struct lu_buf *acl_buf,
2041                               struct dt_allocation_hint *hint)
2042 {
2043         int rc;
2044
2045         rc = mdd_declare_object_create(env, mdd, p, c, attr, handle, spec,
2046                                        def_acl_buf, acl_buf, hint);
2047         if (rc)
2048                 GOTO(out, rc);
2049
2050         if (S_ISDIR(attr->la_mode)) {
2051                 rc = mdo_declare_ref_add(env, p, handle);
2052                 if (rc)
2053                         GOTO(out, rc);
2054         }
2055
2056         if (unlikely(spec->sp_cr_flags & MDS_OPEN_VOLATILE)) {
2057                 rc = orph_declare_index_insert(env, c, attr->la_mode, handle);
2058                 if (rc)
2059                         GOTO(out, rc);
2060         } else {
2061                 struct lu_attr  *la = &mdd_env_info(env)->mti_la_for_fix;
2062
2063                 rc = mdo_declare_index_insert(env, p, mdo2fid(c), attr->la_mode,
2064                                               name->ln_name, handle);
2065                 if (rc != 0)
2066                         return rc;
2067
2068                 rc = mdd_declare_links_add(env, c, handle, ldata, MLAO_IGNORE);
2069                 if (rc)
2070                         return rc;
2071
2072                 *la = *attr;
2073                 la->la_valid = LA_CTIME | LA_MTIME;
2074                 rc = mdo_declare_attr_set(env, p, la, handle);
2075                 if (rc)
2076                         return rc;
2077
2078                 rc = mdd_declare_changelog_store(env, mdd, name, NULL, handle);
2079                 if (rc)
2080                         return rc;
2081         }
2082
2083         /* XXX: For remote create, it should indicate the remote RPC
2084          * will be sent after local transaction is finished, which
2085          * is not very nice, but it will be removed once we fully support
2086          * async update */
2087         if (mdd_object_remote(p) && handle->th_update != NULL)
2088                 handle->th_update->tu_sent_after_local_trans = 1;
2089 out:
2090         return rc;
2091 }
2092
2093 static int mdd_acl_init(const struct lu_env *env, struct mdd_object *pobj,
2094                         struct lu_attr *la, struct lu_buf *def_acl_buf,
2095                         struct lu_buf *acl_buf)
2096 {
2097         int     rc;
2098         ENTRY;
2099
2100         if (S_ISLNK(la->la_mode)) {
2101                 acl_buf->lb_len = 0;
2102                 def_acl_buf->lb_len = 0;
2103                 RETURN(0);
2104         }
2105
2106         mdd_read_lock(env, pobj, MOR_TGT_PARENT);
2107         rc = mdo_xattr_get(env, pobj, def_acl_buf,
2108                            XATTR_NAME_ACL_DEFAULT, BYPASS_CAPA);
2109         mdd_read_unlock(env, pobj);
2110         if (rc > 0) {
2111                 /* If there are default ACL, fix mode/ACL by default ACL */
2112                 def_acl_buf->lb_len = rc;
2113                 LASSERT(def_acl_buf->lb_len <= acl_buf->lb_len);
2114                 memcpy(acl_buf->lb_buf, def_acl_buf->lb_buf, rc);
2115                 acl_buf->lb_len = rc;
2116                 rc = __mdd_fix_mode_acl(env, acl_buf, &la->la_mode);
2117                 if (rc < 0)
2118                         RETURN(rc);
2119         } else if (rc == -ENODATA || rc == -EOPNOTSUPP) {
2120                 /* If there are no default ACL, fix mode by mask */
2121                 struct lu_ucred *uc = lu_ucred(env);
2122
2123                 /* The create triggered by MDT internal events, such as
2124                  * LFSCK reset, will not contain valid "uc". */
2125                 if (unlikely(uc != NULL))
2126                         la->la_mode &= ~uc->uc_umask;
2127                 rc = 0;
2128                 acl_buf->lb_len = 0;
2129                 def_acl_buf->lb_len = 0;
2130         }
2131
2132         RETURN(rc);
2133 }
2134
2135 /**
2136  * Create a metadata object and initialize it, set acl, xattr.
2137  **/
2138 static int mdd_object_create(const struct lu_env *env, struct mdd_object *pobj,
2139                              struct mdd_object *son, struct lu_attr *attr,
2140                              struct md_op_spec *spec, struct lu_buf *acl_buf,
2141                              struct lu_buf *def_acl_buf,
2142                              struct dt_allocation_hint *hint,
2143                              struct thandle *handle)
2144 {
2145         int                     rc;
2146
2147         mdd_write_lock(env, son, MOR_TGT_CHILD);
2148         rc = mdd_object_create_internal(env, NULL, son, attr, handle, spec,
2149                                         hint);
2150         if (rc)
2151                 GOTO(unlock, rc);
2152
2153         /* Note: In DNE phase I, for striped dir, though sub-stripes will be
2154          * created in declare phase, they also needs to be added to master
2155          * object as sub-directory entry. So it has to initialize the master
2156          * object, then set dir striped EA.(in mdo_xattr_set) */
2157         rc = mdd_object_initialize(env, mdo2fid(pobj), son, attr, handle,
2158                                    spec);
2159         if (rc != 0)
2160                 GOTO(err_destroy, rc);
2161
2162         /*
2163          * in case of replay we just set LOVEA provided by the client
2164          * XXX: I think it would be interesting to try "old" way where
2165          *      MDT calls this xattr_set(LOV) in a different transaction.
2166          *      probably this way we code can be made better.
2167          */
2168
2169         /* During creation, there are only a few cases we need do xattr_set to
2170          * create stripes.
2171          * 1. regular file: see comments above.
2172          * 2. dir: inherit default striping or pool settings from parent.
2173          * 3. create striped directory with provided stripeEA.
2174          * 4. create striped directory because inherit default layout from the
2175          * parent.
2176          */
2177         if (spec->no_create ||
2178             (S_ISREG(attr->la_mode) && spec->sp_cr_flags & MDS_OPEN_HAS_EA) ||
2179             S_ISDIR(attr->la_mode)) {
2180                 const struct lu_buf *buf;
2181
2182                 buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
2183                                         spec->u.sp_ea.eadatalen);
2184                 rc = mdo_xattr_set(env, son, buf,
2185                                    S_ISDIR(attr->la_mode) ? XATTR_NAME_LMV :
2186                                                             XATTR_NAME_LOV, 0,
2187                                    handle, BYPASS_CAPA);
2188                 if (rc != 0)
2189                         GOTO(err_destroy, rc);
2190         }
2191
2192 #ifdef CONFIG_FS_POSIX_ACL
2193         if (def_acl_buf != NULL && def_acl_buf->lb_len > 0 &&
2194             S_ISDIR(attr->la_mode)) {
2195                 /* set default acl */
2196                 rc = mdo_xattr_set(env, son, def_acl_buf,
2197                                    XATTR_NAME_ACL_DEFAULT, 0,
2198                                    handle, BYPASS_CAPA);
2199                 if (rc)
2200                         GOTO(err_destroy, rc);
2201         }
2202         /* set its own acl */
2203         if (acl_buf != NULL && acl_buf->lb_len > 0) {
2204                 rc = mdo_xattr_set(env, son, acl_buf,
2205                                    XATTR_NAME_ACL_ACCESS,
2206                                    0, handle, BYPASS_CAPA);
2207                 if (rc)
2208                         GOTO(err_destroy, rc);
2209         }
2210 #endif
2211
2212         if (S_ISLNK(attr->la_mode)) {
2213                 struct lu_ucred  *uc = lu_ucred_assert(env);
2214                 struct dt_object *dt = mdd_object_child(son);
2215                 const char *target_name = spec->u.sp_symname;
2216                 int sym_len = strlen(target_name);
2217                 const struct lu_buf *buf;
2218                 loff_t pos = 0;
2219
2220                 buf = mdd_buf_get_const(env, target_name, sym_len);
2221                 rc = dt->do_body_ops->dbo_write(env, dt, buf, &pos, handle,
2222                                                 mdd_object_capa(env, son),
2223                                                 uc->uc_cap &
2224                                                 CFS_CAP_SYS_RESOURCE_MASK);
2225
2226                 if (rc == sym_len)
2227                         rc = 0;
2228                 else
2229                         GOTO(err_initlized, rc = -EFAULT);
2230         }
2231
2232 err_initlized:
2233         if (unlikely(rc != 0)) {
2234                 int rc2;
2235                 if (S_ISDIR(attr->la_mode)) {
2236                         /* Drop the reference, no need to delete "."/"..",
2237                          * because the object to be destroied directly. */
2238                         rc2 = mdo_ref_del(env, son, handle);
2239                         if (rc2 != 0)
2240                                 GOTO(unlock, rc);
2241                 }
2242                 rc2 = mdo_ref_del(env, son, handle);
2243                 if (rc2 != 0)
2244                         GOTO(unlock, rc);
2245 err_destroy:
2246                 mdo_destroy(env, son, handle);
2247         }
2248 unlock:
2249         mdd_write_unlock(env, son);
2250         RETURN(rc);
2251 }
2252
2253 /*
2254  * Create object and insert it into namespace.
2255  */
2256 static int mdd_create(const struct lu_env *env, struct md_object *pobj,
2257                       const struct lu_name *lname, struct md_object *child,
2258                       struct md_op_spec *spec, struct md_attr* ma)
2259 {
2260         struct mdd_thread_info  *info = mdd_env_info(env);
2261         struct lu_attr          *la = &info->mti_la_for_fix;
2262         struct mdd_object       *mdd_pobj = md2mdd_obj(pobj);
2263         struct mdd_object       *son = md2mdd_obj(child);
2264         struct mdd_device       *mdd = mdo2mdd(pobj);
2265         struct lu_attr          *attr = &ma->ma_attr;
2266         struct thandle          *handle;
2267         struct lu_attr          *pattr = &info->mti_pattr;
2268         struct lu_buf           acl_buf;
2269         struct lu_buf           def_acl_buf;
2270         struct linkea_data      *ldata = &info->mti_link_data;
2271         const char              *name = lname->ln_name;
2272         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
2273         int                      rc;
2274         int                      rc2;
2275         ENTRY;
2276
2277         /*
2278          * Two operations have to be performed:
2279          *
2280          *  - an allocation of a new object (->do_create()), and
2281          *
2282          *  - an insertion into a parent index (->dio_insert()).
2283          *
2284          * Due to locking, operation order is not important, when both are
2285          * successful, *but* error handling cases are quite different:
2286          *
2287          *  - if insertion is done first, and following object creation fails,
2288          *  insertion has to be rolled back, but this operation might fail
2289          *  also leaving us with dangling index entry.
2290          *
2291          *  - if creation is done first, is has to be undone if insertion
2292          *  fails, leaving us with leaked space, which is neither good, nor
2293          *  fatal.
2294          *
2295          * It seems that creation-first is simplest solution, but it is
2296          * sub-optimal in the frequent
2297          *
2298          *         $ mkdir foo
2299          *         $ mkdir foo
2300          *
2301          * case, because second mkdir is bound to create object, only to
2302          * destroy it immediately.
2303          *
2304          * To avoid this follow local file systems that do double lookup:
2305          *
2306          *     0. lookup -> -EEXIST (mdd_create_sanity_check())
2307          *
2308          *     1. create            (mdd_object_create_internal())
2309          *
2310          *     2. insert            (__mdd_index_insert(), lookup again)
2311          */
2312
2313         rc = mdd_la_get(env, mdd_pobj, pattr, BYPASS_CAPA);
2314         if (rc != 0)
2315                 RETURN(rc);
2316
2317         /* Sanity checks before big job. */
2318         rc = mdd_create_sanity_check(env, pobj, pattr, lname, attr, spec);
2319         if (rc)
2320                 RETURN(rc);
2321
2322         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_DQACQ_NET))
2323                 GOTO(out_free, rc = -EINPROGRESS);
2324
2325         handle = mdd_trans_create(env, mdd);
2326         if (IS_ERR(handle))
2327                 GOTO(out_free, rc = PTR_ERR(handle));
2328
2329         acl_buf.lb_buf = info->mti_xattr_buf;
2330         acl_buf.lb_len = sizeof(info->mti_xattr_buf);
2331         def_acl_buf.lb_buf = info->mti_key;
2332         def_acl_buf.lb_len = sizeof(info->mti_key);
2333         rc = mdd_acl_init(env, mdd_pobj, attr, &def_acl_buf, &acl_buf);
2334         if (rc < 0)
2335                 GOTO(out_stop, rc);
2336
2337         mdd_object_make_hint(env, mdd_pobj, son, attr, spec, hint);
2338
2339         memset(ldata, 0, sizeof(*ldata));
2340         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PARENT)) {
2341                 struct lu_fid tfid = *mdd_object_fid(mdd_pobj);
2342
2343                 tfid.f_oid--;
2344                 rc = mdd_linkea_prepare(env, son, NULL, NULL,
2345                                         &tfid, lname, 1, 0, ldata);
2346         } else {
2347                 rc = mdd_linkea_prepare(env, son, NULL, NULL,
2348                                         mdd_object_fid(mdd_pobj),
2349                                         lname, 1, 0, ldata);
2350         }
2351
2352         rc = mdd_declare_create(env, mdd, mdd_pobj, son, lname, attr,
2353                                 handle, spec, ldata, &def_acl_buf, &acl_buf,
2354                                 hint);
2355         if (rc)
2356                 GOTO(out_stop, rc);
2357
2358         rc = mdd_trans_start(env, mdd, handle);
2359         if (rc)
2360                 GOTO(out_stop, rc);
2361
2362         rc = mdd_object_create(env, mdd_pobj, son, attr, spec, &acl_buf,
2363                                &def_acl_buf, hint, handle);
2364         if (rc != 0)
2365                 GOTO(out_stop, rc);
2366
2367         if (unlikely(spec->sp_cr_flags & MDS_OPEN_VOLATILE)) {
2368                 mdd_write_lock(env, son, MOR_TGT_CHILD);
2369                 rc = __mdd_orphan_add(env, son, handle);
2370                 GOTO(out_volatile, rc);
2371         } else {
2372                 rc = __mdd_index_insert(env, mdd_pobj, mdo2fid(son),
2373                                         attr->la_mode, name, handle,
2374                                         mdd_object_capa(env, mdd_pobj));
2375                 if (rc != 0)
2376                         GOTO(err_created, rc);
2377
2378                 mdd_links_add(env, son, mdo2fid(mdd_pobj), lname, handle,
2379                               ldata, 1);
2380
2381                 /* update parent directory mtime/ctime */
2382                 *la = *attr;
2383                 la->la_valid = LA_CTIME | LA_MTIME;
2384                 rc = mdd_update_time(env, mdd_pobj, pattr, la, handle);
2385                 if (rc)
2386                         GOTO(err_insert, rc);
2387         }
2388
2389         EXIT;
2390 err_insert:
2391         if (rc != 0) {
2392                 int rc2;
2393
2394                 if (spec->sp_cr_flags & MDS_OPEN_VOLATILE)
2395                         rc2 = __mdd_orphan_del(env, son, handle);
2396                 else
2397                         rc2 = __mdd_index_delete(env, mdd_pobj, name,
2398                                                  S_ISDIR(attr->la_mode),
2399                                                  handle, BYPASS_CAPA);
2400                 if (rc2 != 0)
2401                         goto out_stop;
2402
2403 err_created:
2404                 mdd_write_lock(env, son, MOR_TGT_CHILD);
2405                 if (S_ISDIR(attr->la_mode)) {
2406                         /* Drop the reference, no need to delete "."/"..",
2407                          * because the object is to be destroyed directly. */
2408                         rc2 = mdo_ref_del(env, son, handle);
2409                         if (rc2 != 0) {
2410                                 mdd_write_unlock(env, son);
2411                                 goto out_stop;
2412                         }
2413                 }
2414 out_volatile:
2415                 /* For volatile files drop one link immediately, since there is
2416                  * no filename in the namespace, and save any error returned. */
2417                 rc2 = mdo_ref_del(env, son, handle);
2418                 if (rc2 != 0) {
2419                         mdd_write_unlock(env, son);
2420                         if (unlikely(rc == 0))
2421                                 rc = rc2;
2422                         goto out_stop;
2423                 }
2424
2425                 /* Don't destroy the volatile object on success */
2426                 if (likely(rc != 0))
2427                         mdo_destroy(env, son, handle);
2428                 mdd_write_unlock(env, son);
2429         }
2430
2431         if (rc == 0 && fid_is_namespace_visible(mdo2fid(son)) &&
2432             likely((spec->sp_cr_flags & MDS_OPEN_VOLATILE) == 0))
2433                 rc = mdd_changelog_ns_store(env, mdd,
2434                                 S_ISDIR(attr->la_mode) ? CL_MKDIR :
2435                                 S_ISREG(attr->la_mode) ? CL_CREATE :
2436                                 S_ISLNK(attr->la_mode) ? CL_SOFTLINK : CL_MKNOD,
2437                                 0, son, mdo2fid(mdd_pobj), NULL, NULL, lname,
2438                                 NULL, handle);
2439 out_stop:
2440         rc2 = mdd_trans_stop(env, mdd, rc, handle);
2441         if (rc == 0)
2442                 rc = rc2;
2443 out_free:
2444         if (ldata->ld_buf && ldata->ld_buf->lb_len > OBD_ALLOC_BIG)
2445                 /* if we vmalloced a large buffer drop it */
2446                 lu_buf_free(ldata->ld_buf);
2447
2448         /* The child object shouldn't be cached anymore */
2449         if (rc)
2450                 set_bit(LU_OBJECT_HEARD_BANSHEE,
2451                         &child->mo_lu.lo_header->loh_flags);
2452         return rc;
2453 }
2454
2455 /*
2456  * Get locks on parents in proper order
2457  * RETURN: < 0 - error, rename_order if successful
2458  */
2459 enum rename_order {
2460         MDD_RN_SAME,
2461         MDD_RN_SRCTGT,
2462         MDD_RN_TGTSRC
2463 };
2464
2465 static int mdd_rename_order(const struct lu_env *env,
2466                             struct mdd_device *mdd,
2467                             struct mdd_object *src_pobj,
2468                             const struct lu_attr *pattr,
2469                             struct mdd_object *tgt_pobj)
2470 {
2471         /* order of locking, 1 - tgt-src, 0 - src-tgt*/
2472         int rc;
2473         ENTRY;
2474
2475         if (src_pobj == tgt_pobj)
2476                 RETURN(MDD_RN_SAME);
2477
2478         /* compared the parent child relationship of src_p&tgt_p */
2479         if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(src_pobj))){
2480                 rc = MDD_RN_SRCTGT;
2481         } else if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(tgt_pobj))) {
2482                 rc = MDD_RN_TGTSRC;
2483         } else {
2484                 rc = mdd_is_parent(env, mdd, src_pobj, pattr, mdo2fid(tgt_pobj),
2485                                    NULL);
2486                 if (rc == -EREMOTE)
2487                         rc = 0;
2488
2489                 if (rc == 1)
2490                         rc = MDD_RN_TGTSRC;
2491                 else if (rc == 0)
2492                         rc = MDD_RN_SRCTGT;
2493         }
2494
2495         RETURN(rc);
2496 }
2497
2498 /* has not mdd_write{read}_lock on any obj yet. */
2499 static int mdd_rename_sanity_check(const struct lu_env *env,
2500                                    struct mdd_object *src_pobj,
2501                                    const struct lu_attr *pattr,
2502                                    struct mdd_object *tgt_pobj,
2503                                    const struct lu_attr *tpattr,
2504                                    struct mdd_object *sobj,
2505                                    const struct lu_attr *cattr,
2506                                    struct mdd_object *tobj,
2507                                    const struct lu_attr *tattr)
2508 {
2509         int rc = 0;
2510         ENTRY;
2511
2512         /* XXX: when get here, sobj must NOT be NULL,
2513          * the other case has been processed in cld_rename
2514          * before mdd_rename and enable MDS_PERM_BYPASS. */
2515         LASSERT(sobj);
2516
2517         rc = mdd_may_delete(env, src_pobj, pattr, sobj, cattr, NULL, 1, 0);
2518         if (rc)
2519                 RETURN(rc);
2520
2521         /* XXX: when get here, "tobj == NULL" means tobj must
2522          * NOT exist (neither on remote MDS, such case has been
2523          * processed in cld_rename before mdd_rename and enable
2524          * MDS_PERM_BYPASS).
2525          * So check may_create, but not check may_unlink. */
2526         if (tobj == NULL)
2527                 rc = mdd_may_create(env, tgt_pobj, tpattr, NULL,
2528                                     (src_pobj != tgt_pobj));
2529         else
2530                 rc = mdd_may_delete(env, tgt_pobj, tpattr, tobj, tattr, cattr,
2531                                     (src_pobj != tgt_pobj), 1);
2532
2533         if (!rc && !tobj && (src_pobj != tgt_pobj) && S_ISDIR(cattr->la_mode))
2534                 rc = __mdd_may_link(env, tgt_pobj, tpattr);
2535
2536         RETURN(rc);
2537 }
2538
2539 static int mdd_declare_rename(const struct lu_env *env,
2540                               struct mdd_device *mdd,
2541                               struct mdd_object *mdd_spobj,
2542                               struct mdd_object *mdd_tpobj,
2543                               struct mdd_object *mdd_sobj,
2544                               struct mdd_object *mdd_tobj,
2545                               const struct lu_name *tname,
2546                               const struct lu_name *sname,
2547                               struct md_attr *ma,
2548                               struct linkea_data *ldata,
2549                               struct thandle *handle)
2550 {
2551         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
2552         int rc;
2553
2554         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
2555         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
2556
2557         LASSERT(mdd_spobj);
2558         LASSERT(mdd_tpobj);
2559         LASSERT(mdd_sobj);
2560
2561         /* name from source dir */
2562         rc = mdo_declare_index_delete(env, mdd_spobj, sname->ln_name, handle);
2563         if (rc)
2564                 return rc;
2565
2566         /* .. from source child */
2567         if (S_ISDIR(mdd_object_type(mdd_sobj))) {
2568                 /* source child can be directory,
2569                  * counted by source dir's nlink */
2570                 rc = mdo_declare_ref_del(env, mdd_spobj, handle);
2571                 if (rc)
2572                         return rc;
2573                 if (mdd_spobj != mdd_tpobj) {
2574                         rc = mdo_declare_index_delete(env, mdd_sobj, dotdot,
2575                                                       handle);
2576                         if (rc != 0)
2577                                 return rc;
2578
2579                         rc = mdo_declare_index_insert(env, mdd_sobj,
2580                                                       mdo2fid(mdd_tpobj),
2581                                                       S_IFDIR, dotdot, handle);
2582                         if (rc != 0)
2583                                 return rc;
2584                 }
2585
2586                 /* new target child can be directory,
2587                  * counted by target dir's nlink */
2588                 rc = mdo_declare_ref_add(env, mdd_tpobj, handle);
2589                 if (rc != 0)
2590                         return rc;
2591         }
2592
2593         la->la_valid = LA_CTIME | LA_MTIME;
2594         rc = mdo_declare_attr_set(env, mdd_spobj, la, handle);
2595         if (rc != 0)
2596                 return rc;
2597
2598         rc = mdo_declare_attr_set(env, mdd_tpobj, la, handle);
2599         if (rc != 0)
2600                 return rc;
2601
2602         la->la_valid = LA_CTIME;
2603         rc = mdo_declare_attr_set(env, mdd_sobj, la, handle);
2604         if (rc)
2605                 return rc;
2606
2607         rc = mdd_declare_links_add(env, mdd_sobj, handle, ldata,
2608                 S_ISREG(mdd_object_type(mdd_sobj)) ? MLAO_CHECK : MLAO_IGNORE);
2609         if (rc)
2610                 return rc;
2611
2612         /* new name */
2613         rc = mdo_declare_index_insert(env, mdd_tpobj, mdo2fid(mdd_sobj),
2614                                       mdd_object_type(mdd_sobj),
2615                                       tname->ln_name, handle);
2616         if (rc != 0)
2617                 return rc;
2618
2619         /* name from target dir (old name), we declare it unconditionally
2620          * as mdd_rename() calls delete unconditionally as well. so just
2621          * to balance declarations vs calls to change ... */
2622         rc = mdo_declare_index_delete(env, mdd_tpobj, tname->ln_name, handle);
2623         if (rc)
2624                 return rc;
2625
2626         if (mdd_tobj && mdd_object_exists(mdd_tobj)) {
2627                 /* delete target child in target parent directory */
2628                 rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2629                 if (rc)
2630                         return rc;
2631
2632                 if (S_ISDIR(mdd_object_type(mdd_tobj))) {
2633                         /* target child can be directory,
2634                          * delete "." reference in target child directory */
2635                         rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2636                         if (rc)
2637                                 return rc;
2638
2639                         /* delete ".." reference in target parent directory */
2640                         rc = mdo_declare_ref_del(env, mdd_tpobj, handle);
2641                         if (rc)
2642                                 return rc;
2643                 }
2644
2645                 la->la_valid = LA_CTIME;
2646                 rc = mdo_declare_attr_set(env, mdd_tobj, la, handle);
2647                 if (rc)
2648                         return rc;
2649
2650                 rc = mdd_declare_finish_unlink(env, mdd_tobj, handle);
2651                 if (rc)
2652                         return rc;
2653         }
2654
2655         rc = mdd_declare_changelog_store(env, mdd, tname, sname, handle);
2656         if (rc)
2657                 return rc;
2658
2659         return rc;
2660 }
2661
2662 /* src object can be remote that is why we use only fid and type of object */
2663 static int mdd_rename(const struct lu_env *env,
2664                       struct md_object *src_pobj, struct md_object *tgt_pobj,
2665                       const struct lu_fid *lf, const struct lu_name *lsname,
2666                       struct md_object *tobj, const struct lu_name *ltname,
2667                       struct md_attr *ma)
2668 {
2669         const char *sname = lsname->ln_name;
2670         const char *tname = ltname->ln_name;
2671         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
2672         struct mdd_object *mdd_spobj = md2mdd_obj(src_pobj); /* source parent */
2673         struct mdd_object *mdd_tpobj = md2mdd_obj(tgt_pobj);
2674         struct mdd_device *mdd = mdo2mdd(src_pobj);
2675         struct mdd_object *mdd_sobj = NULL;                  /* source object */
2676         struct mdd_object *mdd_tobj = NULL;
2677         struct lu_attr *cattr = MDD_ENV_VAR(env, cattr);
2678         struct lu_attr *pattr = MDD_ENV_VAR(env, pattr);
2679         struct lu_attr *tattr = MDD_ENV_VAR(env, tattr);
2680         struct lu_attr *tpattr = MDD_ENV_VAR(env, tpattr);
2681         struct thandle *handle;
2682         struct linkea_data  *ldata = &mdd_env_info(env)->mti_link_data;
2683         const struct lu_fid *tpobj_fid = mdo2fid(mdd_tpobj);
2684         const struct lu_fid *spobj_fid = mdo2fid(mdd_spobj);
2685         bool is_dir;
2686         bool tobj_ref = 0;
2687         bool tobj_locked = 0;
2688         unsigned cl_flags = 0;
2689         int rc, rc2;
2690         ENTRY;
2691
2692         if (tobj)
2693                 mdd_tobj = md2mdd_obj(tobj);
2694
2695         mdd_sobj = mdd_object_find(env, mdd, lf);
2696
2697         rc = mdd_la_get(env, mdd_sobj, cattr, BYPASS_CAPA);
2698         if (rc)
2699                 GOTO(out_pending, rc);
2700
2701         rc = mdd_la_get(env, mdd_spobj, pattr, BYPASS_CAPA);
2702         if (rc)
2703                 GOTO(out_pending, rc);
2704
2705         if (mdd_tobj) {
2706                 rc = mdd_la_get(env, mdd_tobj, tattr, BYPASS_CAPA);
2707                 if (rc)
2708                         GOTO(out_pending, rc);
2709         }
2710
2711         rc = mdd_la_get(env, mdd_tpobj, tpattr, BYPASS_CAPA);
2712         if (rc)
2713                 GOTO(out_pending, rc);
2714
2715         rc = mdd_rename_sanity_check(env, mdd_spobj, pattr, mdd_tpobj, tpattr,
2716                                      mdd_sobj, cattr, mdd_tobj, tattr);
2717         if (rc)
2718                 GOTO(out_pending, rc);
2719
2720         rc = mdd_name_check(mdd, ltname);
2721         if (rc < 0)
2722                 GOTO(out_pending, rc);
2723
2724         handle = mdd_trans_create(env, mdd);
2725         if (IS_ERR(handle))
2726                 GOTO(out_pending, rc = PTR_ERR(handle));
2727
2728         memset(ldata, 0, sizeof(*ldata));
2729         mdd_linkea_prepare(env, mdd_sobj, NULL, NULL, mdd_object_fid(mdd_tpobj),
2730                            ltname, 1, 0, ldata);
2731         rc = mdd_declare_rename(env, mdd, mdd_spobj, mdd_tpobj, mdd_sobj,
2732                                 mdd_tobj, lsname, ltname, ma, ldata, handle);
2733         if (rc)
2734                 GOTO(stop, rc);
2735
2736         rc = mdd_trans_start(env, mdd, handle);
2737         if (rc)
2738                 GOTO(stop, rc);
2739
2740         /* FIXME: Should consider tobj and sobj too in rename_lock. */
2741         rc = mdd_rename_order(env, mdd, mdd_spobj, pattr, mdd_tpobj);
2742         if (rc < 0)
2743                 GOTO(cleanup_unlocked, rc);
2744
2745         is_dir = S_ISDIR(cattr->la_mode);
2746
2747         /* Remove source name from source directory */
2748         rc = __mdd_index_delete(env, mdd_spobj, sname, is_dir, handle,
2749                                 mdd_object_capa(env, mdd_spobj));
2750         if (rc)
2751                 GOTO(cleanup, rc);
2752
2753         /* "mv dir1 dir2" needs "dir1/.." link update */
2754         if (is_dir && mdd_sobj && !lu_fid_eq(spobj_fid, tpobj_fid)) {
2755                 rc = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle,
2756                                         mdd_object_capa(env, mdd_sobj));
2757                 if (rc != 0)
2758                         GOTO(fixup_spobj2, rc);
2759
2760                 rc = __mdd_index_insert_only(env, mdd_sobj, tpobj_fid, S_IFDIR,
2761                                              dotdot, handle,
2762                                              mdd_object_capa(env, mdd_sobj));
2763                 if (rc != 0)
2764                         GOTO(fixup_spobj, rc);
2765         }
2766
2767         /* Remove target name from target directory
2768          * Here tobj can be remote one, so we do index_delete unconditionally
2769          * and -ENOENT is allowed.
2770          */
2771         rc = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle,
2772                                 mdd_object_capa(env, mdd_tpobj));
2773         if (rc != 0) {
2774                 if (mdd_tobj) {
2775                         /* tname might been renamed to something else */
2776                         GOTO(fixup_spobj, rc);
2777                 }
2778                 if (rc != -ENOENT)
2779                         GOTO(fixup_spobj, rc);
2780         }
2781
2782         /* Insert new fid with target name into target dir */
2783         rc = __mdd_index_insert(env, mdd_tpobj, lf, cattr->la_mode,
2784                                 tname, handle, mdd_object_capa(env, mdd_tpobj));
2785         if (rc != 0)
2786                 GOTO(fixup_tpobj, rc);
2787
2788         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
2789         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
2790
2791         /* XXX: mdd_sobj must be local one if it is NOT NULL. */
2792         if (mdd_sobj) {
2793                 la->la_valid = LA_CTIME;
2794                 rc = mdd_update_time(env, mdd_sobj, cattr, la, handle);
2795                 if (rc)
2796                         GOTO(fixup_tpobj, rc);
2797         }
2798
2799         /* Remove old target object
2800          * For tobj is remote case cmm layer has processed
2801          * and set tobj to NULL then. So when tobj is NOT NULL,
2802          * it must be local one.
2803          */
2804         if (tobj && mdd_object_exists(mdd_tobj)) {
2805                 mdd_write_lock(env, mdd_tobj, MOR_TGT_CHILD);
2806                 tobj_locked = 1;
2807                 if (mdd_is_dead_obj(mdd_tobj)) {
2808                         /* shld not be dead, something is wrong */
2809                         CERROR("tobj is dead, something is wrong\n");
2810                         rc = -EINVAL;
2811                         goto cleanup;
2812                 }
2813                 mdo_ref_del(env, mdd_tobj, handle);
2814
2815                 /* Remove dot reference. */
2816                 if (S_ISDIR(tattr->la_mode))
2817                         mdo_ref_del(env, mdd_tobj, handle);
2818                 tobj_ref = 1;
2819
2820                 /* fetch updated nlink */
2821                 rc = mdd_la_get(env, mdd_tobj, tattr, BYPASS_CAPA);
2822                 if (rc != 0) {
2823                         CERROR("%s: Failed to get nlink for tobj "
2824                                 DFID": rc = %d\n",
2825                                 mdd2obd_dev(mdd)->obd_name,
2826                                 PFID(tpobj_fid), rc);
2827                         GOTO(fixup_tpobj, rc);
2828                 }
2829
2830                 la->la_valid = LA_CTIME;
2831                 rc = mdd_update_time(env, mdd_tobj, tattr, la, handle);
2832                 if (rc != 0) {
2833                         CERROR("%s: Failed to set ctime for tobj "
2834                                 DFID": rc = %d\n",
2835                                 mdd2obd_dev(mdd)->obd_name,
2836                                 PFID(tpobj_fid), rc);
2837                         GOTO(fixup_tpobj, rc);
2838                 }
2839
2840                 /* XXX: this transfer to ma will be removed with LOD/OSP */
2841                 ma->ma_attr = *tattr;
2842                 ma->ma_valid |= MA_INODE;
2843                 rc = mdd_finish_unlink(env, mdd_tobj, ma, mdd_tpobj, ltname,
2844                                        handle);
2845                 if (rc != 0) {
2846                         CERROR("%s: Failed to unlink tobj "
2847                                 DFID": rc = %d\n",
2848                                 mdd2obd_dev(mdd)->obd_name,
2849                                 PFID(tpobj_fid), rc);
2850                         GOTO(fixup_tpobj, rc);
2851                 }
2852
2853                 /* fetch updated nlink */
2854                 rc = mdd_la_get(env, mdd_tobj, tattr, BYPASS_CAPA);
2855                 if (rc != 0) {
2856                         CERROR("%s: Failed to get nlink for tobj "
2857                                 DFID": rc = %d\n",
2858                                 mdd2obd_dev(mdd)->obd_name,
2859                                 PFID(tpobj_fid), rc);
2860                         GOTO(fixup_tpobj, rc);
2861                 }
2862                 /* XXX: this transfer to ma will be removed with LOD/OSP */
2863                 ma->ma_attr = *tattr;
2864                 ma->ma_valid |= MA_INODE;
2865
2866                 if (tattr->la_nlink == 0) {
2867                         cl_flags |= CLF_RENAME_LAST;
2868                         if (mdd_hsm_archive_exists(env, mdd_tobj, ma))
2869                                 cl_flags |= CLF_RENAME_LAST_EXISTS;
2870                 }
2871         }
2872
2873         la->la_valid = LA_CTIME | LA_MTIME;
2874         rc = mdd_update_time(env, mdd_spobj, pattr, la, handle);
2875         if (rc)
2876                 GOTO(fixup_tpobj, rc);
2877
2878         if (mdd_spobj != mdd_tpobj) {
2879                 la->la_valid = LA_CTIME | LA_MTIME;
2880                 rc = mdd_update_time(env, mdd_tpobj, tpattr, la, handle);
2881         }
2882
2883         if (rc == 0 && mdd_sobj) {
2884                 mdd_write_lock(env, mdd_sobj, MOR_SRC_CHILD);
2885                 rc = mdd_links_rename(env, mdd_sobj, mdo2fid(mdd_spobj), lsname,
2886                                       mdo2fid(mdd_tpobj), ltname, handle, NULL,
2887                                       0, 0);
2888                 if (rc == -ENOENT)
2889                         /* Old files might not have EA entry */
2890                         mdd_links_add(env, mdd_sobj, mdo2fid(mdd_spobj),
2891                                       lsname, handle, NULL, 0);
2892                 mdd_write_unlock(env, mdd_sobj);
2893                 /* We don't fail the transaction if the link ea can't be
2894                    updated -- fid2path will use alternate lookup method. */
2895                 rc = 0;
2896         }
2897
2898         EXIT;
2899
2900 fixup_tpobj:
2901         if (rc) {
2902                 rc2 = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle,
2903                                          BYPASS_CAPA);
2904                 if (rc2)
2905                         CWARN("tp obj fix error %d\n",rc2);
2906
2907                 if (mdd_tobj && mdd_object_exists(mdd_tobj) &&
2908                     !mdd_is_dead_obj(mdd_tobj)) {
2909                         if (tobj_ref) {
2910                                 mdo_ref_add(env, mdd_tobj, handle);
2911                                 if (is_dir)
2912                                         mdo_ref_add(env, mdd_tobj, handle);
2913                         }
2914
2915                         rc2 = __mdd_index_insert(env, mdd_tpobj,
2916                                                   mdo2fid(mdd_tobj),
2917                                                   mdd_object_type(mdd_tobj),
2918                                                   tname, handle, BYPASS_CAPA);
2919                         if (rc2 != 0)
2920                                 CWARN("tp obj fix error: rc = %d\n", rc2);
2921                 }
2922         }
2923
2924 fixup_spobj:
2925         if (rc && is_dir && mdd_sobj && mdd_spobj != mdd_tpobj) {
2926                 rc2 = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle,
2927                                               BYPASS_CAPA);
2928
2929                 if (rc2)
2930                         CWARN("%s: sp obj dotdot delete error: rc = %d\n",
2931                                mdd2obd_dev(mdd)->obd_name, rc2);
2932
2933
2934                 rc2 = __mdd_index_insert_only(env, mdd_sobj, spobj_fid, S_IFDIR,
2935                                               dotdot, handle, BYPASS_CAPA);
2936                 if (rc2 != 0)
2937                         CWARN("%s: sp obj dotdot insert error: rc = %d\n",
2938                               mdd2obd_dev(mdd)->obd_name, rc2);
2939         }
2940
2941 fixup_spobj2:
2942         if (rc != 0) {
2943                 rc2 = __mdd_index_insert(env, mdd_spobj, lf,
2944                                          mdd_object_type(mdd_sobj), sname,
2945                                          handle, BYPASS_CAPA);
2946                 if (rc2 != 0)
2947                         CWARN("sp obj fix error: rc = %d\n", rc2);
2948         }
2949
2950 cleanup:
2951         if (tobj_locked)
2952                 mdd_write_unlock(env, mdd_tobj);
2953
2954 cleanup_unlocked:
2955         if (rc == 0)
2956                 rc = mdd_changelog_ns_store(env, mdd, CL_RENAME, cl_flags,
2957                                             mdd_tobj, tpobj_fid, lf, spobj_fid,
2958                                             ltname, lsname, handle);
2959
2960 stop:
2961         mdd_trans_stop(env, mdd, rc, handle);
2962
2963 out_pending:
2964         mdd_object_put(env, mdd_sobj);
2965         return rc;
2966 }
2967
2968 /**
2969  * During migration once the parent FID has been changed,
2970  * we need update the parent FID in linkea.
2971  **/
2972 static int mdd_linkea_update_child_internal(const struct lu_env *env,
2973                                             struct mdd_object *parent,
2974                                             struct mdd_object *child,
2975                                             const char *name, int namelen,
2976                                             struct thandle *handle,
2977                                             bool declare)
2978 {
2979         struct mdd_thread_info  *info = mdd_env_info(env);
2980         struct linkea_data      ldata = { NULL };
2981         struct lu_buf           *buf = &info->mti_link_buf;
2982         int                     count;
2983         int                     rc = 0;
2984
2985         ENTRY;
2986
2987         buf = lu_buf_check_and_alloc(buf, PATH_MAX);
2988         if (buf->lb_buf == NULL)
2989                 RETURN(-ENOMEM);
2990
2991         ldata.ld_buf = buf;
2992         rc = mdd_links_read(env, child, &ldata);
2993         if (rc != 0) {
2994                 if (rc == -ENOENT || rc == -ENODATA)
2995                         rc = 0;
2996                 RETURN(rc);
2997         }
2998
2999         LASSERT(ldata.ld_leh != NULL);
3000         ldata.ld_lee = (struct link_ea_entry *)(ldata.ld_leh + 1);
3001         for (count = 0; count < ldata.ld_leh->leh_reccount; count++) {
3002                 struct mdd_device *mdd = mdo2mdd(&child->mod_obj);
3003                 struct lu_name lname;
3004                 struct lu_fid  fid;
3005
3006                 linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen,
3007                                     &lname, &fid);
3008
3009                 if (strncmp(lname.ln_name, name, namelen) != 0 ||
3010                     lu_fid_eq(&fid, mdd_object_fid(parent))) {
3011                         ldata.ld_lee = (struct link_ea_entry *)
3012                                        ((char *)ldata.ld_lee +
3013                                         ldata.ld_reclen);
3014                         continue;
3015                 }
3016
3017                 CDEBUG(D_INFO, "%s: update "DFID" with %.*s:"DFID"\n",
3018                        mdd2obd_dev(mdd)->obd_name, PFID(mdd_object_fid(child)),
3019                        lname.ln_namelen, lname.ln_name,
3020                        PFID(mdd_object_fid(parent)));
3021                 /* update to the new parent fid */
3022                 linkea_entry_pack(ldata.ld_lee, &lname,
3023                                   mdd_object_fid(parent));
3024                 if (declare)
3025                         rc = mdd_declare_links_add(env, child, handle, &ldata,
3026                                                    MLAO_IGNORE);
3027                 else
3028                         rc = mdd_links_write(env, child, &ldata, handle);
3029                 break;
3030         }
3031         RETURN(rc);
3032 }
3033
3034 static int mdd_linkea_declare_update_child(const struct lu_env *env,
3035                                            struct mdd_object *parent,
3036                                            struct mdd_object *child,
3037                                            const char *name, int namelen,
3038                                            struct thandle *handle)
3039 {
3040         return mdd_linkea_update_child_internal(env, parent, child, name,
3041                                                 namelen, handle, true);
3042 }
3043
3044 static int mdd_linkea_update_child(const struct lu_env *env,
3045                                    struct mdd_object *parent,
3046                                    struct mdd_object *child,
3047                                    const char *name, int namelen,
3048                                    struct thandle *handle)
3049 {
3050         return mdd_linkea_update_child_internal(env, parent, child, name,
3051                                                 namelen, handle, false);
3052 }
3053
3054 static int mdd_update_linkea_internal(const struct lu_env *env,
3055                                       struct mdd_object *mdd_pobj,
3056                                       struct mdd_object *mdd_sobj,
3057                                       struct mdd_object *mdd_tobj,
3058                                       const struct lu_name *child_name,
3059                                       struct thandle *handle,
3060                                       int declare)
3061 {
3062         struct mdd_thread_info  *info = mdd_env_info(env);
3063         struct linkea_data      *ldata = &info->mti_link_data;
3064         int                     count;
3065         int                     rc = 0;
3066         ENTRY;
3067
3068         rc = mdd_links_read(env, mdd_sobj, ldata);
3069         if (rc != 0) {
3070                 if (rc == -ENOENT || rc == -ENODATA)
3071                         rc = 0;
3072                 RETURN(rc);
3073         }
3074
3075         /* If it is mulitple links file, we need update the name entry for
3076          * all parent */
3077         LASSERT(ldata->ld_leh != NULL);
3078         ldata->ld_lee = (struct link_ea_entry *)(ldata->ld_leh + 1);
3079         for (count = 0; count < ldata->ld_leh->leh_reccount; count++) {
3080                 struct mdd_device       *mdd = mdo2mdd(&mdd_sobj->mod_obj);
3081                 struct mdd_object       *pobj;
3082                 struct lu_name          lname;
3083                 struct lu_fid           fid;
3084
3085                 linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen,
3086                                     &lname, &fid);
3087                 ldata->ld_lee = (struct link_ea_entry *)((char *)ldata->ld_lee +
3088                                                          ldata->ld_reclen);
3089                 pobj = mdd_object_find(env, mdd, &fid);
3090                 if (IS_ERR(pobj)) {
3091                         CWARN("%s: cannot find obj "DFID": rc = %ld\n",
3092                               mdd2obd_dev(mdd)->obd_name, PFID(&fid),
3093                               PTR_ERR(pobj));
3094                         continue;
3095                 }
3096
3097                 if (!mdd_object_exists(pobj)) {
3098                         CDEBUG(D_INFO, "%s: obj "DFID" does not exist\n",
3099                               mdd2obd_dev(mdd)->obd_name, PFID(&fid));
3100                         GOTO(next_put, rc);
3101                 }
3102
3103                 if (pobj == mdd_pobj &&
3104                     lname.ln_namelen == child_name->ln_namelen &&
3105                     strncmp(lname.ln_name, child_name->ln_name,
3106                             lname.ln_namelen) == 0) {
3107                         CDEBUG(D_INFO, "%s: skip its own %s: "DFID"\n",
3108                               mdd2obd_dev(mdd)->obd_name, child_name->ln_name,
3109                               PFID(&fid));
3110                         GOTO(next_put, rc);
3111                 }
3112
3113                 CDEBUG(D_INFO, "%s: update "DFID" with "DNAME":"DFID"\n",
3114                        mdd2obd_dev(mdd)->obd_name, PFID(mdd_object_fid(pobj)),
3115                        PNAME(&lname), PFID(mdd_object_fid(mdd_tobj)));
3116
3117                 if (declare) {
3118                         /* Remove source name from source directory */
3119                         /* Insert new fid with target name into target dir */
3120                         rc = mdo_declare_index_delete(env, pobj, lname.ln_name,
3121                                                       handle);
3122                         if (rc != 0)
3123                                 GOTO(next_put, rc);
3124
3125                         rc = mdo_declare_index_insert(env, pobj,
3126                                         mdd_object_fid(mdd_tobj),
3127                                         mdd_object_type(mdd_tobj),
3128                                         lname.ln_name, handle);
3129                         if (rc != 0)
3130                                 GOTO(next_put, rc);
3131
3132                         rc = mdo_declare_ref_add(env, mdd_tobj, handle);
3133                         if (rc)
3134                                 GOTO(next_put, rc);
3135
3136                         rc = mdo_declare_ref_del(env, mdd_sobj, handle);
3137                         if (rc)
3138                                 GOTO(next_put, rc);
3139                 } else {
3140                         rc = __mdd_index_delete(env, pobj, lname.ln_name,
3141                                                 0, handle,
3142                                                 mdd_object_capa(env, pobj));
3143                         if (rc)
3144                                 GOTO(next_put, rc);
3145
3146                         rc = __mdd_index_insert(env, pobj,
3147                                         mdd_object_fid(mdd_tobj),
3148                                         mdd_object_type(mdd_tobj),
3149                                         lname.ln_name, handle,
3150                                         mdd_object_capa(env, pobj));
3151                         if (rc != 0)
3152                                 GOTO(next_put, rc);
3153
3154                         mdd_write_lock(env, mdd_tobj, MOR_SRC_CHILD);
3155                         rc = mdo_ref_add(env, mdd_tobj, handle);
3156                         mdd_write_unlock(env, mdd_tobj);
3157                         if (rc)
3158                                 GOTO(next_put, rc);
3159
3160                         mdd_write_lock(env, mdd_sobj, MOR_TGT_CHILD);
3161                         mdo_ref_del(env, mdd_sobj, handle);
3162                         mdd_write_unlock(env, mdd_sobj);
3163                 }
3164 next_put:
3165                 mdd_object_put(env, pobj);
3166                 if (rc != 0)
3167                         break;
3168         }
3169
3170         RETURN(rc);
3171 }
3172
3173 static int mdd_migrate_xattrs(const struct lu_env *env,
3174                               struct mdd_object *mdd_sobj,
3175                               struct mdd_object *mdd_tobj)
3176 {
3177         struct mdd_thread_info  *info = mdd_env_info(env);
3178         struct mdd_device       *mdd = mdo2mdd(&mdd_sobj->mod_obj);
3179         char                    *xname;
3180         struct thandle          *handle;
3181         struct lu_buf           xbuf;
3182         int                     xlen;
3183         int                     rem;
3184         int                     xsize;
3185         int                     list_xsize;
3186         struct lu_buf           list_xbuf;
3187         int                     rc;
3188
3189         /* retrieve xattr list from the old object */
3190         list_xsize = mdo_xattr_list(env, mdd_sobj, &LU_BUF_NULL,
3191                                     mdd_object_capa(env, mdd_sobj));
3192         if (list_xsize == -ENODATA)
3193                 return 0;
3194
3195         if (list_xsize < 0)
3196                 return list_xsize;
3197
3198         lu_buf_check_and_alloc(&info->mti_big_buf, list_xsize);
3199         if (info->mti_big_buf.lb_buf == NULL)
3200                 return -ENOMEM;
3201
3202         list_xbuf.lb_buf = info->mti_big_buf.lb_buf;
3203         list_xbuf.lb_len = list_xsize;
3204         rc = mdo_xattr_list(env, mdd_sobj, &list_xbuf,
3205                             mdd_object_capa(env, mdd_sobj));
3206         if (rc < 0)
3207                 return rc;
3208         rc = 0;
3209         rem = list_xsize;
3210         xname = list_xbuf.lb_buf;
3211         while (rem > 0) {
3212                 xlen = strnlen(xname, rem - 1) + 1;
3213                 if (strcmp(XATTR_NAME_LINK, xname) == 0 ||
3214                     strcmp(XATTR_NAME_LMA, xname) == 0 ||
3215                     strcmp(XATTR_NAME_LMV, xname) == 0)
3216                         goto next;
3217
3218                 /* For directory, if there are default layout, migrate here */
3219                 if (strcmp(XATTR_NAME_LOV, xname) == 0 &&
3220                     !S_ISDIR(lu_object_attr(&mdd_sobj->mod_obj.mo_lu)))
3221                         goto next;
3222
3223                 xsize = mdo_xattr_get(env, mdd_sobj, &LU_BUF_NULL,
3224                                       xname,
3225                                       mdd_object_capa(env, mdd_sobj));
3226                 if (xsize == -ENODATA)
3227                         goto next;
3228                 if (xsize < 0)
3229                         GOTO(out, rc);
3230
3231                 lu_buf_check_and_alloc(&info->mti_link_buf, xsize);
3232                 if (info->mti_link_buf.lb_buf == NULL)
3233                         GOTO(out, rc = -ENOMEM);
3234
3235                 xbuf.lb_len = xsize;
3236                 xbuf.lb_buf = info->mti_link_buf.lb_buf;
3237                 rc = mdo_xattr_get(env, mdd_sobj, &xbuf, xname,
3238                                    mdd_object_capa(env, mdd_sobj));
3239                 if (rc == -ENODATA)
3240                         goto next;
3241                 if (rc < 0)
3242                         GOTO(out, rc);
3243
3244                 handle = mdd_trans_create(env, mdd);
3245                 if (IS_ERR(handle))
3246                         GOTO(out, rc = PTR_ERR(handle));
3247
3248                 rc = mdo_declare_xattr_set(env, mdd_tobj, &xbuf, xname, 0,
3249                                            handle);
3250                 if (rc != 0)
3251                         GOTO(stop_trans, rc);
3252                 /* Note: this transaction is part of migration, and it is not
3253                  * the last step of migration, so we set th_local = 1 to avoid
3254                  * update last rcvd for this transaction */
3255                 handle->th_local = 1;
3256                 rc = mdd_trans_start(env, mdd, handle);
3257                 if (rc != 0)
3258                         GOTO(stop_trans, rc);
3259
3260                 rc = mdo_xattr_set(env, mdd_tobj, &xbuf, xname, 0, handle,
3261                                    mdd_object_capa(env, mdd_sobj));
3262                 if (rc == -EEXIST)
3263                         GOTO(stop_trans, rc = 0);
3264
3265                 if (rc != 0)
3266                         GOTO(stop_trans, rc);
3267 stop_trans:
3268                 mdd_trans_stop(env, mdd, rc, handle);
3269                 if (rc != 0)
3270                         GOTO(out, rc);
3271 next:
3272                 rem -= xlen;
3273                 memmove(xname, xname + xlen, rem);
3274         }
3275 out:
3276         return rc;
3277 }
3278
3279 static int mdd_declare_migrate_create(const struct lu_env *env,
3280                                       struct mdd_object *mdd_pobj,
3281                                       struct mdd_object *mdd_sobj,
3282                                       struct mdd_object *mdd_tobj,
3283                                       struct md_op_spec *spec,
3284                                       struct lu_attr *la,
3285                                       union lmv_mds_md *mgr_ea,
3286                                       struct linkea_data *ldata,
3287                                       struct thandle *handle)
3288 {
3289         struct lu_attr          *la_flag = MDD_ENV_VAR(env, la_for_fix);
3290         const struct lu_buf     *buf;
3291         int                     rc;
3292         int                     mgr_easize;
3293
3294         rc = mdd_declare_object_create_internal(env, mdd_pobj, mdd_tobj, la,
3295                                                 handle, spec, NULL);
3296         if (rc != 0)
3297                 return rc;
3298
3299         rc = mdd_declare_object_initialize(env, mdd_pobj, mdd_tobj, la,
3300                                            handle);
3301         if (rc != 0)
3302                 return rc;
3303
3304         if (S_ISLNK(la->la_mode)) {
3305                 const char *target_name = spec->u.sp_symname;
3306                 int sym_len = strlen(target_name);
3307                 const struct lu_buf *buf;
3308
3309                 buf = mdd_buf_get_const(env, target_name, sym_len);
3310                 rc = dt_declare_record_write(env, mdd_object_child(mdd_tobj),
3311                                              buf, 0, handle);
3312                 if (rc != 0)
3313                         return rc;
3314         } else if (S_ISDIR(la->la_mode)) {
3315                 rc = mdd_declare_links_add(env, mdd_tobj, handle, ldata,
3316                                            MLAO_IGNORE);
3317                 if (rc != 0)
3318                         return rc;
3319         }
3320
3321         if (spec->u.sp_ea.eadata != NULL && spec->u.sp_ea.eadatalen != 0) {
3322                 buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
3323                                         spec->u.sp_ea.eadatalen);
3324                 rc = mdo_declare_xattr_set(env, mdd_tobj, buf, XATTR_NAME_LOV,
3325                                            0, handle);
3326                 if (rc)
3327                         return rc;
3328         }
3329
3330         mgr_easize = lmv_mds_md_size(2, LMV_MAGIC_V1);
3331         buf = mdd_buf_get_const(env, mgr_ea, mgr_easize);
3332         rc = mdo_declare_xattr_set(env, mdd_sobj, buf, XATTR_NAME_LMV,
3333                                    0, handle);
3334         if (rc)
3335                 return rc;
3336
3337         la_flag->la_valid = LA_FLAGS;
3338         la_flag->la_flags = la->la_flags | LUSTRE_IMMUTABLE_FL;
3339         rc = mdo_declare_attr_set(env, mdd_sobj, la_flag, handle);
3340
3341         return rc;
3342 }
3343
3344 static int mdd_migrate_create(const struct lu_env *env,
3345                               struct mdd_object *mdd_pobj,
3346                               struct mdd_object *mdd_sobj,
3347                               struct mdd_object *mdd_tobj,
3348                               const struct lu_name *lname,
3349                               struct lu_attr *la)
3350 {
3351         struct mdd_thread_info  *info = mdd_env_info(env);
3352         struct mdd_device       *mdd = mdo2mdd(&mdd_sobj->mod_obj);
3353         struct md_op_spec       *spec = &info->mti_spec;
3354         struct lu_buf           lmm_buf = { NULL };
3355         struct lu_buf           link_buf = { NULL };
3356         const struct lu_buf     *buf;
3357         struct thandle          *handle;
3358         struct lmv_mds_md_v1    *mgr_ea;
3359         struct lu_attr          *la_flag = MDD_ENV_VAR(env, la_for_fix);
3360         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
3361         int                     mgr_easize;
3362         struct linkea_data      *ldata = &mdd_env_info(env)->mti_link_data;
3363         int                     rc;
3364         ENTRY;
3365
3366         /* prepare spec for create */
3367         memset(spec, 0, sizeof(*spec));
3368         spec->sp_cr_lookup = 0;
3369         spec->sp_feat = &dt_directory_features;
3370         if (S_ISLNK(la->la_mode)) {
3371                 buf = lu_buf_check_and_alloc(
3372                                 &mdd_env_info(env)->mti_big_buf,
3373                                 la->la_size + 1);
3374                 link_buf = *buf;
3375                 link_buf.lb_len = la->la_size + 1;
3376                 memset(link_buf.lb_buf, 0, link_buf.lb_len);
3377                 rc = mdd_readlink(env, &mdd_sobj->mod_obj, &link_buf);
3378                 if (rc <= 0) {
3379                         rc = rc != 0 ? rc : -EFAULT;
3380                         CERROR("%s: "DFID" readlink failed: rc = %d\n",
3381                                mdd2obd_dev(mdd)->obd_name,
3382                                PFID(mdd_object_fid(mdd_sobj)), rc);
3383                         RETURN(rc);
3384                 }
3385                 spec->u.sp_symname = link_buf.lb_buf;
3386         } else if S_ISREG(la->la_mode) {
3387                 /* retrieve lov of the old object */
3388                 rc = mdd_get_lov_ea(env, mdd_sobj, &lmm_buf);
3389                 if (rc != 0 && rc != -ENODATA)
3390                         RETURN(rc);
3391                 if (lmm_buf.lb_buf != NULL && lmm_buf.lb_len != 0) {
3392                         spec->u.sp_ea.eadata = lmm_buf.lb_buf;
3393                         spec->u.sp_ea.eadatalen = lmm_buf.lb_len;
3394                         spec->sp_cr_flags |= MDS_OPEN_HAS_EA;
3395                 }
3396         } else if (S_ISDIR(la->la_mode)) {
3397                 rc = mdd_links_read(env, mdd_sobj, ldata);
3398                 if (rc < 0 && rc != -ENODATA)
3399                         RETURN(rc);
3400         }
3401
3402         mgr_ea = (struct lmv_mds_md_v1 *)info->mti_xattr_buf;
3403         memset(mgr_ea, 0, sizeof(*mgr_ea));
3404         mgr_ea->lmv_magic = cpu_to_le32(LMV_MAGIC_V1);
3405         mgr_ea->lmv_stripe_count = cpu_to_le32(2);
3406         mgr_ea->lmv_master_mdt_index = mdd_seq_site(mdd)->ss_node_id;
3407         mgr_ea->lmv_hash_type = cpu_to_le32(LMV_HASH_FLAG_MIGRATION);
3408         fid_cpu_to_le(&mgr_ea->lmv_stripe_fids[0], mdd_object_fid(mdd_sobj));
3409         fid_cpu_to_le(&mgr_ea->lmv_stripe_fids[1], mdd_object_fid(mdd_tobj));
3410
3411         mdd_object_make_hint(env, mdd_pobj, mdd_tobj, la, spec, hint);
3412
3413         handle = mdd_trans_create(env, mdd);
3414         if (IS_ERR(handle))
3415                 GOTO(out_free, rc = PTR_ERR(handle));
3416
3417         /* Note: this transaction is part of migration, and it is not
3418          * the last step of migration, so we set th_local = 1 to avoid
3419          * update last rcvd for this transaction */
3420         handle->th_local = 1;
3421         rc = mdd_declare_migrate_create(env, mdd_pobj, mdd_sobj, mdd_tobj,
3422                                         spec, la,
3423                                         (union lmv_mds_md *)info->mti_xattr_buf,
3424                                         ldata, handle);
3425         if (rc != 0)
3426                 GOTO(stop_trans, rc);
3427
3428         rc = mdd_trans_start(env, mdd, handle);
3429         if (rc != 0)
3430                 GOTO(stop_trans, rc);
3431
3432         /* create the target object */
3433         rc = mdd_object_create(env, mdd_pobj, mdd_tobj, la, spec, NULL, NULL,
3434                                hint, handle);
3435         if (rc != 0)
3436                 GOTO(stop_trans, rc);
3437
3438         if (S_ISDIR(la->la_mode)) {
3439                 rc = mdd_links_write(env, mdd_tobj, ldata, handle);
3440                 if (rc != 0)
3441                         GOTO(stop_trans, rc);
3442         }
3443
3444         /* Set MIGRATE EA on the source inode, so once the migration needs
3445          * to be re-done during failover, the re-do process can locate the
3446          * target object which is already being created. */
3447         mgr_easize = lmv_mds_md_size(2, LMV_MAGIC_V1);
3448         buf = mdd_buf_get_const(env, mgr_ea, mgr_easize);
3449         rc = mdo_xattr_set(env, mdd_sobj, buf, XATTR_NAME_LMV, 0,
3450                            handle, mdd_object_capa(env, mdd_sobj));
3451         if (rc != 0)
3452                 GOTO(stop_trans, rc);
3453
3454         /* Set immutable flag, so any modification is disabled until
3455          * the migration is done. Once the migration is interrupted,
3456          * if the resume process find the migrating object has both
3457          * IMMUTALBE flag and MIGRATE EA, it need to clear IMMUTABLE
3458          * flag and approve the migration */
3459         la_flag->la_valid = LA_FLAGS;
3460         la_flag->la_flags = la->la_flags | LUSTRE_IMMUTABLE_FL;
3461         rc = mdo_attr_set(env, mdd_sobj, la_flag, handle,
3462                           mdd_object_capa(env, mdd_sobj));
3463 stop_trans:
3464         if (handle != NULL)
3465                 mdd_trans_stop(env, mdd, rc, handle);
3466 out_free:
3467         if (lmm_buf.lb_buf != NULL)
3468                 OBD_FREE(lmm_buf.lb_buf, lmm_buf.lb_len);
3469         RETURN(rc);
3470 }
3471
3472 static int mdd_migrate_entries(const struct lu_env *env,
3473                                struct mdd_object *mdd_sobj,
3474                                struct mdd_object *mdd_tobj)
3475 {
3476         struct dt_object        *next = mdd_object_child(mdd_sobj);
3477         struct mdd_device       *mdd = mdo2mdd(&mdd_sobj->mod_obj);
3478         struct dt_object        *dt_tobj = mdd_object_child(mdd_tobj);
3479         struct thandle          *handle;
3480         struct dt_it            *it;
3481         const struct dt_it_ops  *iops;
3482         int                      rc;
3483         int                      result;
3484         struct lu_dirent        *ent;
3485         ENTRY;
3486
3487         OBD_ALLOC(ent, NAME_MAX + sizeof(*ent) + 1);
3488         if (ent == NULL)
3489                 RETURN(-ENOMEM);
3490
3491         if (!dt_try_as_dir(env, next))
3492                 GOTO(out_ent, rc = -ENOTDIR);
3493         /*
3494          * iterate directories
3495          */
3496         iops = &next->do_index_ops->dio_it;
3497         it = iops->init(env, next, LUDA_FID | LUDA_TYPE,
3498                         mdd_object_capa(env, mdd_sobj));
3499         if (IS_ERR(it))
3500                 GOTO(out_ent, rc = PTR_ERR(it));
3501
3502         rc = iops->load(env, it, 0);
3503         if (rc == 0)
3504                 rc = iops->next(env, it);
3505         else if (rc > 0)
3506                 rc = 0;
3507         /*
3508          * At this point and across for-loop:
3509          *
3510          *  rc == 0 -> ok, proceed.
3511          *  rc >  0 -> end of directory.
3512          *  rc <  0 -> error.
3513          */
3514         do {
3515                 struct mdd_object       *child;
3516                 char                    *name = mdd_env_info(env)->mti_key;
3517                 int                     len;
3518                 int                     recsize;
3519                 int                     is_dir;
3520                 bool                    target_exist = false;
3521
3522                 len = iops->key_size(env, it);
3523                 if (len == 0)
3524                         goto next;
3525
3526                 result = iops->rec(env, it, (struct dt_rec *)ent,
3527                                    LUDA_FID | LUDA_TYPE);
3528                 if (result == -ESTALE)
3529                         goto next;
3530                 if (result != 0) {
3531                         rc = result;
3532                         goto out;
3533                 }
3534
3535                 fid_le_to_cpu(&ent->lde_fid, &ent->lde_fid);
3536                 recsize = le16_to_cpu(ent->lde_reclen);
3537
3538                 /* Insert new fid with target name into target dir */
3539                 if ((ent->lde_namelen == 1 && ent->lde_name[0] == '.') ||
3540                     (ent->lde_namelen == 2 && ent->lde_name[0] == '.' &&
3541                      ent->lde_name[1] == '.'))
3542                         goto next;
3543
3544                 child = mdd_object_find(env, mdd, &ent->lde_fid);
3545                 if (IS_ERR(child))
3546                         GOTO(out, rc = PTR_ERR(child));
3547
3548                 is_dir = S_ISDIR(mdd_object_type(child));
3549
3550                 snprintf(name, ent->lde_namelen + 1, "%s", ent->lde_name);
3551
3552                 /* Check whether the name has been inserted to the target */
3553                 if (dt_try_as_dir(env, dt_tobj)) {
3554                         struct lu_fid *fid = &mdd_env_info(env)->mti_fid2;
3555
3556                         rc = dt_lookup(env, dt_tobj, (struct dt_rec *)fid,
3557                                        (struct dt_key *)name,
3558                                        mdd_object_capa(env, mdd_tobj));
3559                         if (unlikely(rc == 0))
3560                                 target_exist = true;
3561                 }
3562
3563                 handle = mdd_trans_create(env, mdd);
3564                 if (IS_ERR(handle))
3565                         GOTO(out, rc = PTR_ERR(handle));
3566
3567                 /* Note: this transaction is part of migration, and it is not
3568                  * the last step of migration, so we set th_local = 1 to avoid
3569                  * updating last rcvd for this transaction */
3570                 handle->th_local = 1;
3571                 if (likely(!target_exist)) {
3572                         rc = mdo_declare_index_insert(env, mdd_tobj,
3573                                                       &ent->lde_fid,
3574                                                       mdd_object_type(child),
3575                                                       name, handle);
3576                         if (rc != 0)
3577                                 GOTO(out_put, rc);
3578
3579                         if (is_dir) {
3580                                 rc = mdo_declare_ref_add(env, mdd_tobj, handle);
3581                                 if (rc != 0)
3582                                         GOTO(out_put, rc);
3583                         }
3584                 }
3585
3586                 rc = mdo_declare_index_delete(env, mdd_sobj, name, handle);
3587                 if (rc != 0)
3588                         GOTO(out_put, rc);
3589
3590                 if (is_dir) {
3591                         rc = mdo_declare_ref_del(env, mdd_sobj, handle);
3592                         if (rc != 0)
3593                                 GOTO(out_put, rc);
3594
3595                         /* Update .. for child */
3596                         rc = mdo_declare_index_delete(env, child, dotdot,
3597                                                       handle);
3598                         if (rc != 0)
3599                                 GOTO(out_put, rc);
3600
3601                         rc = mdo_declare_index_insert(env, child,
3602                                                       mdd_object_fid(mdd_tobj),
3603                                                       S_IFDIR, dotdot, handle);
3604                         if (rc != 0)
3605                                 GOTO(out_put, rc);
3606                 }
3607
3608                 rc = mdd_linkea_declare_update_child(env, mdd_tobj,
3609                                                      child, name,
3610                                                      strlen(name),
3611