Whamcloud - gitweb
dc240035262c77d0b93ed64f9ba6ce0bb6db614a
[fs/lustre-release.git] / lustre / mdd / mdd_dir.c
1 /* -*- MODE: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  *  mdd/mdd_handler.c
5  *  Lustre Metadata Server (mdd) routines
6  *
7  *  Copyright (C) 2006 Cluster File Systems, Inc.
8  *   Author: Wang Di <wangdi@clusterfs.com>
9  *
10  *   This file is part of the Lustre file system, http://www.lustre.org
11  *   Lustre is a trademark of Cluster File Systems, Inc.
12  *
13  *   You may have signed or agreed to another license before downloading
14  *   this software.  If so, you are bound by the terms and conditions
15  *   of that agreement, and the following does not apply to you.  See the
16  *   LICENSE file included with this distribution for more information.
17  *
18  *   If you did not agree to a different license, then this copy of Lustre
19  *   is open source software; you can redistribute it and/or modify it
20  *   under the terms of version 2 of the GNU General Public License as
21  *   published by the Free Software Foundation.
22  *
23  *   In either case, Lustre is distributed in the hope that it will be
24  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
25  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
26  *   license text for more details.
27  */
28 #ifndef EXPORT_SYMTAB
29 # define EXPORT_SYMTAB
30 #endif
31 #define DEBUG_SUBSYSTEM S_MDS
32
33 #include <linux/module.h>
34 #include <linux/jbd.h>
35 #include <obd.h>
36 #include <obd_class.h>
37 #include <lustre_ver.h>
38 #include <obd_support.h>
39 #include <lprocfs_status.h>
40
41 #include <linux/ldiskfs_fs.h>
42 #include <lustre_mds.h>
43 #include <lustre/lustre_idl.h>
44 #include <lustre_fid.h>
45
46 #include "mdd_internal.h"
47
48 static const char dot[] = ".";
49 static const char dotdot[] = "..";
50
51 static struct lu_name lname_dotdot = {
52         (char *) dotdot,
53         sizeof(dotdot) - 1
54 };
55
56 static int __mdd_lookup(const struct lu_env *env, struct md_object *pobj,
57                         const struct lu_name *lname, struct lu_fid* fid,
58                         int mask);
59 static int
60 __mdd_lookup_locked(const struct lu_env *env, struct md_object *pobj,
61                     const struct lu_name *lname, struct lu_fid* fid, int mask)
62 {
63         char *name = lname->ln_name;
64         struct mdd_object *mdd_obj = md2mdd_obj(pobj);
65         struct dynlock_handle *dlh;
66         int rc;
67
68         dlh = mdd_pdo_read_lock(env, mdd_obj, name);
69         if (unlikely(dlh == NULL))
70                 return -ENOMEM;
71         rc = __mdd_lookup(env, pobj, lname, fid, mask);
72         mdd_pdo_read_unlock(env, mdd_obj, dlh);
73
74         return rc;
75 }
76
77 static int mdd_lookup(const struct lu_env *env,
78                       struct md_object *pobj, const struct lu_name *lname,
79                       struct lu_fid* fid, struct md_op_spec *spec)
80 {
81         int rc;
82         ENTRY;
83         rc = __mdd_lookup_locked(env, pobj, lname, fid, MAY_EXEC);
84         RETURN(rc);
85 }
86
87
88 static int mdd_parent_fid(const struct lu_env *env, struct mdd_object *obj,
89                           struct lu_fid *fid)
90 {
91         return __mdd_lookup_locked(env, &obj->mod_obj, &lname_dotdot, fid, 0);
92 }
93
94 /*
95  * For root fid use special function, whcih does not compare version component
96  * of fid. Vresion component is different for root fids on all MDTs.
97  */
98 static int mdd_is_root(struct mdd_device *mdd, const struct lu_fid *fid)
99 {
100         return fid_seq(&mdd->mdd_root_fid) == fid_seq(fid) &&
101                 fid_oid(&mdd->mdd_root_fid) == fid_oid(fid);
102 }
103
104 /*
105  * return 1: if lf is the fid of the ancestor of p1;
106  * return 0: if not;
107  *
108  * return -EREMOTE: if remote object is found, in this
109  * case fid of remote object is saved to @pf;
110  *
111  * otherwise: values < 0, errors.
112  */
113 static int mdd_is_parent(const struct lu_env *env,
114                          struct mdd_device *mdd,
115                          struct mdd_object *p1,
116                          const struct lu_fid *lf,
117                          struct lu_fid *pf)
118 {
119         struct mdd_object *parent = NULL;
120         struct lu_fid *pfid;
121         int rc;
122         ENTRY;
123
124         LASSERT(!lu_fid_eq(mdo2fid(p1), lf));
125         pfid = &mdd_env_info(env)->mti_fid;
126
127         /* Check for root first. */
128         if (mdd_is_root(mdd, mdo2fid(p1)))
129                 RETURN(0);
130
131         for(;;) {
132                 /* this is done recursively, bypass capa for each obj */
133                 mdd_set_capainfo(env, 4, p1, BYPASS_CAPA);
134                 rc = mdd_parent_fid(env, p1, pfid);
135                 if (rc)
136                         GOTO(out, rc);
137                 if (mdd_is_root(mdd, pfid))
138                         GOTO(out, rc = 0);
139                 if (lu_fid_eq(pfid, lf))
140                         GOTO(out, rc = 1);
141                 if (parent)
142                         mdd_object_put(env, parent);
143                 parent = mdd_object_find(env, mdd, pfid);
144
145                 /* cross-ref parent */
146                 if (parent == NULL) {
147                         if (pf != NULL)
148                                 *pf = *pfid;
149                         GOTO(out, rc = -EREMOTE);
150                 } else if (IS_ERR(parent))
151                         GOTO(out, rc = PTR_ERR(parent));
152                 p1 = parent;
153         }
154         EXIT;
155 out:
156         if (parent && !IS_ERR(parent))
157                 mdd_object_put(env, parent);
158         return rc;
159 }
160
161 /*
162  * No permission check is needed.
163  *
164  * returns 1: if fid is ancestor of @mo;
165  * returns 0: if fid is not a ancestor of @mo;
166  *
167  * returns EREMOTE if remote object is found, fid of remote object is saved to
168  * @fid;
169  *
170  * returns < 0: if error
171  */
172 static int mdd_is_subdir(const struct lu_env *env,
173                          struct md_object *mo, const struct lu_fid *fid,
174                          struct lu_fid *sfid)
175 {
176         struct mdd_device *mdd = mdo2mdd(mo);
177         int rc;
178         ENTRY;
179
180         if (!S_ISDIR(mdd_object_type(md2mdd_obj(mo))))
181                 RETURN(0);
182
183         rc = mdd_is_parent(env, mdd, md2mdd_obj(mo), fid, sfid);
184         if (rc == 0) {
185                 /* found root */
186                 fid_zero(sfid);
187         } else if (rc == 1) {
188                 /* found @fid is parent */
189                 *sfid = *fid;
190                 rc = 0;
191         }
192         RETURN(rc);
193 }
194
195 /*
196  * Check that @dir contains no entries except (possibly) dot and dotdot.
197  *
198  * Returns:
199  *
200  *             0        empty
201  *      -ENOTDIR        not a directory object
202  *    -ENOTEMPTY        not empty
203  *           -ve        other error
204  *
205  */
206 static int mdd_dir_is_empty(const struct lu_env *env,
207                             struct mdd_object *dir)
208 {
209         struct dt_it     *it;
210         struct dt_object *obj;
211         struct dt_it_ops *iops;
212         int result;
213         ENTRY;
214
215         obj = mdd_object_child(dir);
216         if (!dt_try_as_dir(env, obj))
217                 RETURN(-ENOTDIR);
218
219         iops = &obj->do_index_ops->dio_it;
220         it = iops->init(env, obj, 0, BYPASS_CAPA);
221         if (it != NULL) {
222                 result = iops->get(env, it, (const void *)"");
223                 if (result > 0) {
224                         int i;
225                         for (result = 0, i = 0; result == 0 && i < 3; ++i)
226                                 result = iops->next(env, it);
227                         if (result == 0)
228                                 result = -ENOTEMPTY;
229                         else if (result == +1)
230                                 result = 0;
231                 } else if (result == 0)
232                         /*
233                          * Huh? Index contains no zero key?
234                          */
235                         result = -EIO;
236
237                 iops->put(env, it);
238                 iops->fini(env, it);
239         } else
240                 result = -ENOMEM;
241         RETURN(result);
242 }
243
244 static int __mdd_may_link(const struct lu_env *env, struct mdd_object *obj)
245 {
246         struct mdd_device *m = mdd_obj2mdd_dev(obj);
247         struct lu_attr *la = &mdd_env_info(env)->mti_la;
248         int rc;
249         ENTRY;
250
251         rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
252         if (rc)
253                 RETURN(rc);
254
255         if (la->la_nlink >= m->mdd_dt_conf.ddp_max_nlink)
256                 RETURN(-EMLINK);
257         else
258                 RETURN(0);
259 }
260
261 /*
262  * Check whether it may create the cobj under the pobj.
263  * cobj maybe NULL
264  */
265 int mdd_may_create(const struct lu_env *env, struct mdd_object *pobj,
266                    struct mdd_object *cobj, int check_perm, int check_nlink)
267 {
268         int rc = 0;
269         ENTRY;
270
271         if (cobj && mdd_object_exists(cobj))
272                 RETURN(-EEXIST);
273
274         if (mdd_is_dead_obj(pobj))
275                 RETURN(-ENOENT);
276
277         if (check_perm)
278                 rc = mdd_permission_internal_locked(env, pobj, NULL,
279                                                     MAY_WRITE | MAY_EXEC);
280
281         if (!rc && check_nlink)
282                 rc = __mdd_may_link(env, pobj);
283
284         RETURN(rc);
285 }
286
287 /*
288  * Check whether can unlink from the pobj in the case of "cobj == NULL".
289  */
290 int mdd_may_unlink(const struct lu_env *env, struct mdd_object *pobj,
291                    const struct md_attr *ma)
292 {
293         int rc;
294         ENTRY;
295
296         if (mdd_is_dead_obj(pobj))
297                 RETURN(-ENOENT);
298
299         if ((ma->ma_attr.la_valid & LA_FLAGS) &&
300             (ma->ma_attr.la_flags & (LUSTRE_APPEND_FL | LUSTRE_IMMUTABLE_FL)))
301                 RETURN(-EPERM);
302
303         rc = mdd_permission_internal_locked(env, pobj, NULL,
304                                             MAY_WRITE | MAY_EXEC);
305         if (rc)
306                 RETURN(rc);
307
308         if (mdd_is_append(pobj))
309                 RETURN(-EPERM);
310
311         RETURN(rc);
312 }
313
314 /*
315  * pobj == NULL is remote ops case, under such case, pobj's
316  * VTX feature has been checked already, no need check again.
317  */
318 static inline int mdd_is_sticky(const struct lu_env *env,
319                                 struct mdd_object *pobj,
320                                 struct mdd_object *cobj)
321 {
322         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
323         struct md_ucred *uc = md_ucred(env);
324         int rc;
325
326         if (pobj) {
327                 rc = mdd_la_get(env, pobj, tmp_la, BYPASS_CAPA);
328                 if (rc)
329                         return rc;
330         
331                 if (!(tmp_la->la_mode & S_ISVTX) ||
332                      (tmp_la->la_uid == uc->mu_fsuid))
333                         return 0;
334         }
335
336         rc = mdd_la_get(env, cobj, tmp_la, BYPASS_CAPA);
337         if (rc) 
338                 return rc;
339         
340         if (tmp_la->la_uid == uc->mu_fsuid)
341                 return 0;
342         
343         return !mdd_capable(uc, CAP_FOWNER);
344 }
345
346 /*
347  * Check whether it may delete the cobj from the pobj.
348  * pobj maybe NULL
349  */
350 int mdd_may_delete(const struct lu_env *env, struct mdd_object *pobj,
351                    struct mdd_object *cobj, struct md_attr *ma,
352                    int check_perm, int check_empty)
353 {
354         int rc = 0;
355         ENTRY;
356
357         LASSERT(cobj);
358         if (!mdd_object_exists(cobj))
359                 RETURN(-ENOENT);
360
361         if (pobj) {
362                 if (mdd_is_dead_obj(pobj))
363                         RETURN(-ENOENT);
364
365                 if (check_perm) {
366                         rc = mdd_permission_internal_locked(env, pobj, NULL,
367                                                     MAY_WRITE | MAY_EXEC);
368                         if (rc)
369                                 RETURN(rc);
370                 }
371
372                 if (mdd_is_append(pobj))
373                         RETURN(-EPERM);
374         }
375
376         if (!(ma->ma_attr_flags & MDS_VTX_BYPASS) &&
377             mdd_is_sticky(env, pobj, cobj))
378                 RETURN(-EPERM);
379
380         if (mdd_is_immutable(cobj) || mdd_is_append(cobj))
381                 RETURN(-EPERM);
382
383         if ((ma->ma_attr.la_valid & LA_FLAGS) &&
384             (ma->ma_attr.la_flags & (LUSTRE_APPEND_FL | LUSTRE_IMMUTABLE_FL)))
385                 RETURN(-EPERM);
386
387         if (S_ISDIR(ma->ma_attr.la_mode)) {
388                 struct mdd_device *mdd = mdo2mdd(&cobj->mod_obj);
389
390                 if (!S_ISDIR(mdd_object_type(cobj)))
391                         RETURN(-ENOTDIR);
392
393                 if (lu_fid_eq(mdo2fid(cobj), &mdd->mdd_root_fid))
394                         RETURN(-EBUSY);
395         } else if (S_ISDIR(mdd_object_type(cobj)))
396                 RETURN(-EISDIR);
397
398         if (S_ISDIR(ma->ma_attr.la_mode) && check_empty)
399                 rc = mdd_dir_is_empty(env, cobj);
400
401         RETURN(rc);
402 }
403
404 /*
405  * tgt maybe NULL
406  * has mdd_write_lock on src already, but not on tgt yet
407  */
408 int mdd_link_sanity_check(const struct lu_env *env,
409                           struct mdd_object *tgt_obj,
410                           const struct lu_name *lname,
411                           struct mdd_object *src_obj)
412 {
413         struct mdd_device *m = mdd_obj2mdd_dev(src_obj);
414         int rc = 0;
415         ENTRY;
416
417         /* Local ops, no lookup before link, check filename length here. */
418         if (lname && (lname->ln_namelen > m->mdd_dt_conf.ddp_max_name_len))
419                 RETURN(-ENAMETOOLONG);
420
421         if (mdd_is_immutable(src_obj) || mdd_is_append(src_obj))
422                 RETURN(-EPERM);
423
424         if (S_ISDIR(mdd_object_type(src_obj)))
425                 RETURN(-EPERM);
426
427         LASSERT(src_obj != tgt_obj);
428         if (tgt_obj) {
429                 rc = mdd_may_create(env, tgt_obj, NULL, 1, 0);
430                 if (rc)
431                         RETURN(rc);
432         }
433
434         rc = __mdd_may_link(env, src_obj);
435
436         RETURN(rc);
437 }
438
439 const struct dt_rec *__mdd_fid_rec(const struct lu_env *env,
440                                    const struct lu_fid *fid)
441 {
442         struct lu_fid_pack *pack = &mdd_env_info(env)->mti_pack;
443
444         fid_pack(pack, fid, &mdd_env_info(env)->mti_fid2);
445         return (const struct dt_rec *)pack;
446 }
447
448
449 /* insert named index, add reference if isdir */
450 static int __mdd_index_insert(const struct lu_env *env, struct mdd_object *pobj,
451                               const struct lu_fid *lf, const char *name, int is_dir,
452                               struct thandle *handle, struct lustre_capa *capa)
453 {
454         struct dt_object *next = mdd_object_child(pobj);
455         int               rc;
456         ENTRY;
457
458         if (dt_try_as_dir(env, next)) {
459                 rc = next->do_index_ops->dio_insert(env, next,
460                                                     __mdd_fid_rec(env, lf),
461                                                     (const struct dt_key *)name,
462                                                     handle, capa);
463         } else {
464                 rc = -ENOTDIR;
465         }
466
467         if (rc == 0) {
468                 if (is_dir) {
469                         mdd_write_lock(env, pobj);
470                         mdo_ref_add(env, pobj, handle);
471                         mdd_write_unlock(env, pobj);
472                 }
473         }
474         RETURN(rc);
475 }
476
477 /* delete named index, drop reference if isdir */
478 static int __mdd_index_delete(const struct lu_env *env, struct mdd_object *pobj,
479                               const char *name, int is_dir, struct thandle *handle,
480                               struct lustre_capa *capa)
481 {
482         struct dt_object *next = mdd_object_child(pobj);
483         int               rc;
484         ENTRY;
485
486         if (dt_try_as_dir(env, next)) {
487                 rc = next->do_index_ops->dio_delete(env, next,
488                                                     (struct dt_key *)name,
489                                                     handle, capa);
490                 if (rc == 0 && is_dir) {
491                         mdd_write_lock(env, pobj);
492                         mdo_ref_del(env, pobj, handle);
493                         mdd_write_unlock(env, pobj);
494                 }
495         } else
496                 rc = -ENOTDIR;
497
498         RETURN(rc);
499 }
500
501 static int
502 __mdd_index_insert_only(const struct lu_env *env, struct mdd_object *pobj,
503                         const struct lu_fid *lf, const char *name,
504                         struct thandle *handle, struct lustre_capa *capa)
505 {
506         struct dt_object *next = mdd_object_child(pobj);
507         int               rc;
508         ENTRY;
509
510         if (dt_try_as_dir(env, next)) {
511                 rc = next->do_index_ops->dio_insert(env, next,
512                                                     __mdd_fid_rec(env, lf),
513                                                     (const struct dt_key *)name,
514                                                     handle, capa);
515         } else {
516                 rc = -ENOTDIR;
517         }
518         RETURN(rc);
519 }
520
521 static int mdd_link(const struct lu_env *env, struct md_object *tgt_obj,
522                     struct md_object *src_obj, const struct lu_name *lname,
523                     struct md_attr *ma)
524 {
525         char *name = lname->ln_name;
526         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
527         struct mdd_object *mdd_tobj = md2mdd_obj(tgt_obj);
528         struct mdd_object *mdd_sobj = md2mdd_obj(src_obj);
529         struct mdd_device *mdd = mdo2mdd(src_obj);
530         struct dynlock_handle *dlh;
531         struct thandle *handle;
532         int rc;
533         ENTRY;
534
535         mdd_txn_param_build(env, mdd, MDD_TXN_LINK_OP);
536         handle = mdd_trans_start(env, mdd);
537         if (IS_ERR(handle))
538                 RETURN(PTR_ERR(handle));
539
540         dlh = mdd_pdo_write_lock(env, mdd_tobj, name);
541         if (dlh == NULL)
542                 GOTO(out_trans, rc = -ENOMEM);
543         mdd_write_lock(env, mdd_sobj);
544
545         rc = mdd_link_sanity_check(env, mdd_tobj, lname, mdd_sobj);
546         if (rc)
547                 GOTO(out_unlock, rc);
548
549         rc = __mdd_index_insert_only(env, mdd_tobj, mdo2fid(mdd_sobj),
550                                      name, handle,
551                                      mdd_object_capa(env, mdd_tobj));
552         if (rc)
553                 GOTO(out_unlock, rc);
554
555         mdo_ref_add(env, mdd_sobj, handle);
556
557         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
558         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
559
560         la->la_valid = LA_CTIME | LA_MTIME;
561         rc = mdd_attr_set_internal_locked(env, mdd_tobj, la, handle, 0);
562         if (rc)
563                 GOTO(out_unlock, rc);
564
565         la->la_valid = LA_CTIME;
566         rc = mdd_attr_set_internal(env, mdd_sobj, la, handle, 0);
567         EXIT;
568 out_unlock:
569         mdd_write_unlock(env, mdd_sobj);
570         mdd_pdo_write_unlock(env, mdd_tobj, dlh);
571 out_trans:
572         mdd_trans_stop(env, mdd, rc, handle);
573         return rc;
574 }
575
576 /* caller should take a lock before calling */
577 int mdd_finish_unlink(const struct lu_env *env,
578                       struct mdd_object *obj, struct md_attr *ma,
579                       struct thandle *th)
580 {
581         int rc;
582         ENTRY;
583
584         rc = mdd_iattr_get(env, obj, ma);
585         if (rc == 0 && ma->ma_attr.la_nlink == 0) {
586                 /* add new orphan and the object
587                  * will be deleted during the object_put() */
588                 if (__mdd_orphan_add(env, obj, th) == 0)
589                         obj->mod_flags |= ORPHAN_OBJ;
590
591                 obj->mod_flags |= DEAD_OBJ;
592                 if (obj->mod_count == 0)
593                         rc = mdd_object_kill(env, obj, ma);
594                 else
595                         /* clear MA_LOV | MA_COOKIE, if we do not
596                          * unlink it in case we get it somewhere */
597                         ma->ma_valid &= ~(MA_LOV | MA_COOKIE);
598         } else
599                 ma->ma_valid &= ~(MA_LOV | MA_COOKIE);
600
601         RETURN(rc);
602 }
603
604 /*
605  * pobj maybe NULL
606  * has mdd_write_lock on cobj already, but not on pobj yet
607  */
608 int mdd_unlink_sanity_check(const struct lu_env *env, struct mdd_object *pobj,
609                             struct mdd_object *cobj, struct md_attr *ma)
610 {
611         int rc;
612         ENTRY;
613
614         rc = mdd_may_delete(env, pobj, cobj, ma, 1, 1);
615
616         RETURN(rc);
617 }
618
619 static int mdd_unlink(const struct lu_env *env, struct md_object *pobj,
620                       struct md_object *cobj, const struct lu_name *lname,
621                       struct md_attr *ma)
622 {
623         char *name = lname->ln_name;
624         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
625         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
626         struct mdd_object *mdd_cobj = md2mdd_obj(cobj);
627         struct mdd_device *mdd = mdo2mdd(pobj);
628         struct dynlock_handle *dlh;
629         struct thandle    *handle;
630         int rc, is_dir;
631         ENTRY;
632
633         LASSERTF(mdd_object_exists(mdd_cobj) > 0, "FID is "DFID"\n",
634                  PFID(mdd_object_fid(mdd_cobj)));
635
636         rc = mdd_log_txn_param_build(env, cobj, ma, MDD_TXN_UNLINK_OP);
637         if (rc)
638                 RETURN(rc);
639
640         handle = mdd_trans_start(env, mdd);
641         if (IS_ERR(handle))
642                 RETURN(PTR_ERR(handle));
643
644
645         dlh = mdd_pdo_write_lock(env, mdd_pobj, name);
646         if (dlh == NULL)
647                 GOTO(out_trans, rc = -ENOMEM);
648         mdd_write_lock(env, mdd_cobj);
649
650         is_dir = S_ISDIR(ma->ma_attr.la_mode);
651         rc = mdd_unlink_sanity_check(env, mdd_pobj, mdd_cobj, ma);
652         if (rc)
653                 GOTO(cleanup, rc);
654
655         rc = __mdd_index_delete(env, mdd_pobj, name, is_dir, handle,
656                                 mdd_object_capa(env, mdd_pobj));
657         if (rc)
658                 GOTO(cleanup, rc);
659
660         mdo_ref_del(env, mdd_cobj, handle);
661         if (is_dir)
662                 /* unlink dot */
663                 mdo_ref_del(env, mdd_cobj, handle);
664
665         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
666         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
667
668         la->la_valid = LA_CTIME | LA_MTIME;
669         rc = mdd_attr_set_internal_locked(env, mdd_pobj, la, handle, 0);
670         if (rc)
671                 GOTO(cleanup, rc);
672
673         la->la_valid = LA_CTIME;
674         rc = mdd_attr_set_internal(env, mdd_cobj, la, handle, 0);
675         if (rc)
676                 GOTO(cleanup, rc);
677
678         rc = mdd_finish_unlink(env, mdd_cobj, ma, handle);
679
680         if (rc == 0)
681                 obd_set_info_async(mdd2obd_dev(mdd)->u.mds.mds_osc_exp,
682                                    strlen("unlinked"), "unlinked", 0,
683                                    NULL, NULL);
684         EXIT;
685 cleanup:
686         mdd_write_unlock(env, mdd_cobj);
687         mdd_pdo_write_unlock(env, mdd_pobj, dlh);
688 out_trans:
689         mdd_trans_stop(env, mdd, rc, handle);
690         return rc;
691 }
692
693 /* has not lock on pobj yet */
694 static int mdd_ni_sanity_check(const struct lu_env *env,
695                                struct md_object *pobj,
696                                const struct md_attr *ma)
697 {
698         struct mdd_object *obj = md2mdd_obj(pobj);
699         int rc;
700         ENTRY;
701
702         if (ma->ma_attr_flags & MDS_PERM_BYPASS)
703                 RETURN(0);
704
705         rc = mdd_may_create(env, obj, NULL, 1, S_ISDIR(ma->ma_attr.la_mode));
706
707         RETURN(rc);
708 }
709
710 /*
711  * Partial operation.
712  */
713 static int mdd_name_insert(const struct lu_env *env,
714                            struct md_object *pobj,
715                            const struct lu_name *lname,
716                            const struct lu_fid *fid,
717                            const struct md_attr *ma)
718 {
719         char *name = lname->ln_name;
720         struct lu_attr   *la = &mdd_env_info(env)->mti_la_for_fix;
721         struct mdd_object *mdd_obj = md2mdd_obj(pobj);
722         struct mdd_device *mdd = mdo2mdd(pobj);
723         struct dynlock_handle *dlh;
724         struct thandle *handle;
725         int is_dir = S_ISDIR(ma->ma_attr.la_mode);
726         int rc;
727         ENTRY;
728
729         mdd_txn_param_build(env, mdd, MDD_TXN_INDEX_INSERT_OP);
730         handle = mdd_trans_start(env, mdo2mdd(pobj));
731         if (IS_ERR(handle))
732                 RETURN(PTR_ERR(handle));
733
734         dlh = mdd_pdo_write_lock(env, mdd_obj, name);
735         if (dlh == NULL)
736                 GOTO(out_trans, rc = -ENOMEM);
737
738         rc = mdd_ni_sanity_check(env, pobj, ma);
739         if (rc)
740                 GOTO(out_unlock, rc);
741
742         rc = __mdd_index_insert(env, mdd_obj, fid, name, is_dir,
743                                 handle, BYPASS_CAPA);
744         if (rc)
745                 GOTO(out_unlock, rc);
746
747         /*
748          * For some case, no need update obj's ctime (LA_CTIME is not set),
749          * e.g. split_dir.
750          * For other cases, update obj's ctime (LA_CTIME is set),
751          * e.g. cmr_link.
752          */
753         if (ma->ma_attr.la_valid & LA_CTIME) {
754                 la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
755                 la->la_valid = LA_CTIME | LA_MTIME;
756                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la, handle, 0);
757         }
758         EXIT;
759 out_unlock:
760         mdd_pdo_write_unlock(env, mdd_obj, dlh);
761 out_trans:
762         mdd_trans_stop(env, mdo2mdd(pobj), rc, handle);
763         return rc;
764 }
765
766 /* has not lock on pobj yet */
767 static int mdd_nr_sanity_check(const struct lu_env *env,
768                                struct md_object *pobj,
769                                const struct md_attr *ma)
770 {
771         struct mdd_object *obj = md2mdd_obj(pobj);
772         int rc;
773         ENTRY;
774
775         if (ma->ma_attr_flags & MDS_PERM_BYPASS)
776                 RETURN(0);
777
778         rc = mdd_may_unlink(env, obj, ma);
779
780         RETURN(rc);
781 }
782
783 /*
784  * Partial operation.
785  */
786 static int mdd_name_remove(const struct lu_env *env,
787                            struct md_object *pobj,
788                            const struct lu_name *lname,
789                            const struct md_attr *ma)
790 {
791         char *name = lname->ln_name;
792         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
793         struct mdd_object *mdd_obj = md2mdd_obj(pobj);
794         struct mdd_device *mdd = mdo2mdd(pobj);
795         struct dynlock_handle *dlh;
796         struct thandle *handle;
797         int is_dir = S_ISDIR(ma->ma_attr.la_mode);
798         int rc;
799         ENTRY;
800
801         mdd_txn_param_build(env, mdd, MDD_TXN_INDEX_DELETE_OP);
802         handle = mdd_trans_start(env, mdd);
803         if (IS_ERR(handle))
804                 RETURN(PTR_ERR(handle));
805
806         dlh = mdd_pdo_write_lock(env, mdd_obj, name);
807         if (dlh == NULL)
808                 GOTO(out_trans, rc = -ENOMEM);
809
810         rc = mdd_nr_sanity_check(env, pobj, ma);
811         if (rc)
812                 GOTO(out_unlock, rc);
813
814         rc = __mdd_index_delete(env, mdd_obj, name, is_dir,
815                                 handle, BYPASS_CAPA);
816         if (rc)
817                 GOTO(out_unlock, rc);
818
819         /*
820          * For some case, no need update obj's ctime (LA_CTIME is not set),
821          * e.g. split_dir.
822          * For other cases, update obj's ctime (LA_CTIME is set),
823          * e.g. cmr_unlink.
824          */
825         if (ma->ma_attr.la_valid & LA_CTIME) {
826                 la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
827                 la->la_valid = LA_CTIME | LA_MTIME;
828                 rc = mdd_attr_set_internal_locked(env, mdd_obj, la, handle, 0);
829         }
830         EXIT;
831 out_unlock:
832         mdd_pdo_write_unlock(env, mdd_obj, dlh);
833 out_trans:
834         mdd_trans_stop(env, mdd, rc, handle);
835         return rc;
836 }
837
838 /*
839  * tobj maybe NULL
840  * has mdd_write_lock on tobj alreay, but not on tgt_pobj yet
841  */
842 static int mdd_rt_sanity_check(const struct lu_env *env,
843                                struct mdd_object *tgt_pobj,
844                                struct mdd_object *tobj,
845                                struct md_attr *ma)
846 {
847         int rc;
848         ENTRY;
849
850         if (unlikely(ma->ma_attr_flags & MDS_PERM_BYPASS))
851                 RETURN(0);
852
853         /* XXX: for mdd_rename_tgt, "tobj == NULL" does not mean tobj not
854          * exist. In fact, tobj must exist, otherwise the call trace will be:
855          * mdt_reint_rename_tgt -> mdo_name_insert -> ... -> mdd_name_insert.
856          * When get here, tobj must be NOT NULL, the other case has been
857          * processed in cmr_rename_tgt before mdd_rename_tgt and enable
858          * MDS_PERM_BYPASS.
859          * So check may_delete, but not check nlink of tgt_pobj. */
860         LASSERT(tobj);
861         rc = mdd_may_delete(env, tgt_pobj, tobj, ma, 1, 1);
862
863         RETURN(rc);
864 }
865
866 static int mdd_rename_tgt(const struct lu_env *env,
867                           struct md_object *pobj, struct md_object *tobj,
868                           const struct lu_fid *lf, const struct lu_name *lname,
869                           struct md_attr *ma)
870 {
871         char *name = lname->ln_name;
872         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
873         struct mdd_object *mdd_tpobj = md2mdd_obj(pobj);
874         struct mdd_object *mdd_tobj = md2mdd_obj(tobj);
875         struct mdd_device *mdd = mdo2mdd(pobj);
876         struct dynlock_handle *dlh;
877         struct thandle *handle;
878         int rc;
879         ENTRY;
880
881         mdd_txn_param_build(env, mdd, MDD_TXN_RENAME_TGT_OP);
882         handle = mdd_trans_start(env, mdd);
883         if (IS_ERR(handle))
884                 RETURN(PTR_ERR(handle));
885
886         dlh = mdd_pdo_write_lock(env, mdd_tpobj, name);
887         if (dlh == NULL)
888                 GOTO(out_trans, rc = -ENOMEM);
889         if (tobj)
890                 mdd_write_lock(env, mdd_tobj);
891
892         rc = mdd_rt_sanity_check(env, mdd_tpobj, mdd_tobj, ma);
893         if (rc)
894                 GOTO(cleanup, rc);
895
896         /*
897          * If rename_tgt is called then we should just re-insert name with
898          * correct fid, no need to dec/inc parent nlink if obj is dir.
899          */
900         rc = __mdd_index_delete(env, mdd_tpobj, name, 0, handle, BYPASS_CAPA);
901         if (rc)
902                 GOTO(cleanup, rc);
903
904         rc = __mdd_index_insert_only(env, mdd_tpobj, lf, name, handle,
905                                      BYPASS_CAPA);
906         if (rc)
907                 GOTO(cleanup, rc);
908
909         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
910         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
911
912         la->la_valid = LA_CTIME | LA_MTIME;
913         rc = mdd_attr_set_internal_locked(env, mdd_tpobj, la, handle, 0);
914         if (rc)
915                 GOTO(cleanup, rc);
916
917         /* 
918          * For tobj is remote case cmm layer has processed
919          * and pass NULL tobj to here. So when tobj is NOT NULL,
920          * it must be local one.
921          */
922         if (tobj && mdd_object_exists(mdd_tobj)) {
923                 mdo_ref_del(env, mdd_tobj, handle);
924
925                 /* Remove dot reference. */
926                 if (S_ISDIR(ma->ma_attr.la_mode))
927                         mdo_ref_del(env, mdd_tobj, handle);
928
929                 la->la_valid = LA_CTIME;
930                 rc = mdd_attr_set_internal(env, mdd_tobj, la, handle, 0);
931                 if (rc)
932                         GOTO(cleanup, rc);
933
934                 rc = mdd_finish_unlink(env, mdd_tobj, ma, handle);
935                 if (rc)
936                         GOTO(cleanup, rc);
937         }
938         EXIT;
939 cleanup:
940         if (tobj)
941                 mdd_write_unlock(env, mdd_tobj);
942         mdd_pdo_write_unlock(env, mdd_tpobj, dlh);
943 out_trans:
944         mdd_trans_stop(env, mdd, rc, handle);
945         return rc;
946 }
947
948 /*
949  * The permission has been checked when obj created, no need check again.
950  */
951 static int mdd_cd_sanity_check(const struct lu_env *env,
952                                struct mdd_object *obj)
953 {
954         ENTRY;
955
956         /* EEXIST check */
957         if (!obj || mdd_is_dead_obj(obj))
958                 RETURN(-ENOENT);
959
960         RETURN(0);
961
962 }
963
964 static int mdd_create_data(const struct lu_env *env, struct md_object *pobj,
965                            struct md_object *cobj, const struct md_op_spec *spec,
966                            struct md_attr *ma)
967 {
968         struct mdd_device *mdd = mdo2mdd(cobj);
969         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
970         struct mdd_object *son = md2mdd_obj(cobj);
971         struct lu_attr    *attr = &ma->ma_attr;
972         struct lov_mds_md *lmm = NULL;
973         int                lmm_size = 0;
974         struct thandle    *handle;
975         int                rc;
976         ENTRY;
977
978         rc = mdd_cd_sanity_check(env, son);
979         if (rc)
980                 RETURN(rc);
981
982         if (!md_should_create(spec->sp_cr_flags))
983                 RETURN(0);
984
985         rc = mdd_lov_create(env, mdd, mdd_pobj, son, &lmm, &lmm_size,
986                             spec, attr);
987         if (rc)
988                 RETURN(rc);
989
990         mdd_txn_param_build(env, mdd, MDD_TXN_CREATE_DATA_OP);
991         handle = mdd_trans_start(env, mdd);
992         if (IS_ERR(handle))
993                 GOTO(out_free, rc = PTR_ERR(handle));
994
995         /*
996          * XXX: Setting the lov ea is not locked but setting the attr is locked?
997          * Should this be fixed?
998          */
999
1000         /* Replay creates has objects already */
1001 #if 0
1002         if (spec->u.sp_ea.no_lov_create) {
1003                 CDEBUG(D_INFO, "we already have lov ea\n");
1004                 rc = mdd_lov_set_md(env, mdd_pobj, son,
1005                                     (struct lov_mds_md *)spec->u.sp_ea.eadata,
1006                                     spec->u.sp_ea.eadatalen, handle, 0);
1007         } else
1008 #endif
1009                 /* No need mdd_lsm_sanity_check here */
1010                 rc = mdd_lov_set_md(env, mdd_pobj, son, lmm,
1011                                     lmm_size, handle, 0);
1012
1013         if (rc == 0)
1014                rc = mdd_attr_get_internal_locked(env, son, ma);
1015
1016         /* update lov_objid data, must be before transaction stop! */
1017         if (rc == 0)
1018                 mdd_lov_objid_update(env, mdd);
1019
1020         mdd_trans_stop(env, mdd, rc, handle);
1021 out_free:
1022         /* Finish mdd_lov_create() stuff. */
1023         mdd_lov_create_finish(env, mdd, lmm, lmm_size, spec);
1024         RETURN(rc);
1025 }
1026
1027 static int
1028 __mdd_lookup(const struct lu_env *env, struct md_object *pobj,
1029              const struct lu_name *lname, struct lu_fid* fid, int mask)
1030 {
1031         char                *name = lname->ln_name;
1032         const struct dt_key *key = (const struct dt_key *)name;
1033         struct mdd_object   *mdd_obj = md2mdd_obj(pobj);
1034         struct mdd_device   *m = mdo2mdd(pobj);
1035         struct dt_object    *dir = mdd_object_child(mdd_obj);
1036         struct lu_fid_pack  *pack = &mdd_env_info(env)->mti_pack;
1037         int rc;
1038         ENTRY;
1039
1040         if (unlikely(mdd_is_dead_obj(mdd_obj)))
1041                 RETURN(-ESTALE);
1042
1043         rc = mdd_object_exists(mdd_obj);
1044         if (unlikely(rc == 0))
1045                 RETURN(-ESTALE);
1046         else if (unlikely(rc < 0)) {
1047                 CERROR("Object "DFID" locates on remote server\n",
1048                         PFID(mdo2fid(mdd_obj)));
1049                 LBUG();
1050         }
1051
1052         /* The common filename length check. */
1053         if (unlikely(lname->ln_namelen > m->mdd_dt_conf.ddp_max_name_len))
1054                 RETURN(-ENAMETOOLONG);
1055
1056         rc = mdd_permission_internal_locked(env, mdd_obj, NULL, mask);
1057         if (rc)
1058                 RETURN(rc);
1059
1060         if (likely(S_ISDIR(mdd_object_type(mdd_obj)) &&
1061                    dt_try_as_dir(env, dir))) {
1062                 rc = dir->do_index_ops->dio_lookup(env, dir,
1063                                                  (struct dt_rec *)pack, key,
1064                                                  mdd_object_capa(env, mdd_obj));
1065                 if (rc == 0)
1066                         rc = fid_unpack(pack, fid);
1067         } else
1068                 rc = -ENOTDIR;
1069
1070         RETURN(rc);
1071 }
1072
1073 int mdd_object_initialize(const struct lu_env *env, const struct lu_fid *pfid,
1074                           struct mdd_object *child, struct md_attr *ma,
1075                           struct thandle *handle)
1076 {
1077         int rc;
1078         ENTRY;
1079
1080         /*
1081          * Update attributes for child.
1082          *
1083          * FIXME:
1084          *  (1) the valid bits should be converted between Lustre and Linux;
1085          *  (2) maybe, the child attributes should be set in OSD when creation.
1086          */
1087
1088         rc = mdd_attr_set_internal(env, child, &ma->ma_attr, handle, 0);
1089         if (rc != 0)
1090                 RETURN(rc);
1091
1092         if (S_ISDIR(ma->ma_attr.la_mode)) {
1093                 /* Add "." and ".." for newly created dir */
1094                 mdo_ref_add(env, child, handle);
1095                 rc = __mdd_index_insert_only(env, child, mdo2fid(child),
1096                                              dot, handle, BYPASS_CAPA);
1097                 if (rc == 0) {
1098                         rc = __mdd_index_insert_only(env, child, pfid,
1099                                                      dotdot, handle,
1100                                                      BYPASS_CAPA);
1101                         if (rc != 0) {
1102                                 int rc2;
1103
1104                                 rc2 = __mdd_index_delete(env, child, dot, 0,
1105                                                          handle, BYPASS_CAPA);
1106                                 if (rc2 != 0)
1107                                         CERROR("Failure to cleanup after dotdot"
1108                                                " creation: %d (%d)\n", rc2, rc);
1109                                 else
1110                                         mdo_ref_del(env, child, handle);
1111                         }
1112                 }
1113         }
1114         RETURN(rc);
1115 }
1116
1117 /* has not lock on pobj yet */
1118 static int mdd_create_sanity_check(const struct lu_env *env,
1119                                    struct md_object *pobj,
1120                                    const struct lu_name *lname,
1121                                    struct md_attr *ma,
1122                                    struct md_op_spec *spec)
1123 {
1124         struct mdd_thread_info *info = mdd_env_info(env);
1125         struct lu_attr    *la        = &info->mti_la;
1126         struct lu_fid     *fid       = &info->mti_fid;
1127         struct mdd_object *obj       = md2mdd_obj(pobj);
1128         struct mdd_device *m         = mdo2mdd(pobj);
1129         int lookup                   = spec->sp_cr_lookup;
1130         int rc;
1131         ENTRY;
1132
1133         /* EEXIST check */
1134         if (mdd_is_dead_obj(obj))
1135                 RETURN(-ENOENT);
1136
1137         /*
1138          * In some cases this lookup is not needed - we know before if name
1139          * exists or not because MDT performs lookup for it.
1140          * name length check is done in lookup.
1141          */
1142         if (lookup) {
1143                 /*
1144                  * Check if the name already exist, though it will be checked in
1145                  * _index_insert also, for avoiding rolling back if exists
1146                  * _index_insert.
1147                  */
1148                 rc = __mdd_lookup_locked(env, pobj, lname, fid,
1149                                          MAY_WRITE | MAY_EXEC);
1150                 if (rc != -ENOENT)
1151                         RETURN(rc ? : -EEXIST);
1152         } else {
1153                 /*
1154                  * Check WRITE permission for the parent.
1155                  * EXEC permission have been checked
1156                  * when lookup before create already.
1157                  */
1158                 rc = mdd_permission_internal_locked(env, obj, NULL, MAY_WRITE);
1159                 if (rc)
1160                         RETURN(rc);
1161         }
1162
1163         /* sgid check */
1164         rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
1165         if (rc != 0)
1166                 RETURN(rc);
1167
1168         if (la->la_mode & S_ISGID) {
1169                 ma->ma_attr.la_gid = la->la_gid;
1170                 if (S_ISDIR(ma->ma_attr.la_mode)) {
1171                         ma->ma_attr.la_mode |= S_ISGID;
1172                         ma->ma_attr.la_valid |= LA_MODE;
1173                 }
1174         }
1175
1176         switch (ma->ma_attr.la_mode & S_IFMT) {
1177         case S_IFDIR: {
1178                 if (la->la_nlink >= m->mdd_dt_conf.ddp_max_nlink)
1179                         RETURN(-EMLINK);
1180                 else
1181                         RETURN(0);
1182         }
1183         case S_IFLNK: {
1184                 unsigned int symlen = strlen(spec->u.sp_symname) + 1;
1185
1186                 if (symlen > (1 << m->mdd_dt_conf.ddp_block_shift))
1187                         RETURN(-ENAMETOOLONG);
1188                 else
1189                         RETURN(0);
1190         }
1191         case S_IFREG:
1192         case S_IFCHR:
1193         case S_IFBLK:
1194         case S_IFIFO:
1195         case S_IFSOCK:
1196                 rc = 0;
1197                 break;
1198         default:
1199                 rc = -EINVAL;
1200                 break;
1201         }
1202         RETURN(rc);
1203 }
1204
1205 /*
1206  * Create object and insert it into namespace.
1207  */
1208 static int mdd_create(const struct lu_env *env,
1209                       struct md_object *pobj,
1210                       const struct lu_name *lname,
1211                       struct md_object *child,
1212                       struct md_op_spec *spec,
1213                       struct md_attr* ma)
1214 {
1215         char *name = lname->ln_name;
1216         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
1217         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1218         struct mdd_object *son = md2mdd_obj(child);
1219         struct mdd_device *mdd = mdo2mdd(pobj);
1220         struct lu_attr    *attr = &ma->ma_attr;
1221         struct lov_mds_md *lmm = NULL;
1222         struct thandle    *handle;
1223         int rc, created = 0, inserted = 0, lmm_size = 0;
1224         struct dynlock_handle *dlh;
1225         ENTRY;
1226
1227         /*
1228          * Two operations have to be performed:
1229          *
1230          *  - allocation of new object (->do_create()), and
1231          *
1232          *  - insertion into parent index (->dio_insert()).
1233          *
1234          * Due to locking, operation order is not important, when both are
1235          * successful, *but* error handling cases are quite different:
1236          *
1237          *  - if insertion is done first, and following object creation fails,
1238          *  insertion has to be rolled back, but this operation might fail
1239          *  also leaving us with dangling index entry.
1240          *
1241          *  - if creation is done first, is has to be undone if insertion
1242          *  fails, leaving us with leaked space, which is neither good, nor
1243          *  fatal.
1244          *
1245          * It seems that creation-first is simplest solution, but it is
1246          * sub-optimal in the frequent
1247          *
1248          *         $ mkdir foo
1249          *         $ mkdir foo
1250          *
1251          * case, because second mkdir is bound to create object, only to
1252          * destroy it immediately.
1253          *
1254          * To avoid this follow local file systems that do double lookup:
1255          *
1256          *     0. lookup -> -EEXIST (mdd_create_sanity_check())
1257          *
1258          *     1. create            (mdd_object_create_internal())
1259          *
1260          *     2. insert            (__mdd_index_insert(), lookup again)
1261          */
1262
1263         /* Sanity checks before big job. */
1264         rc = mdd_create_sanity_check(env, pobj, lname, ma, spec);
1265         if (rc)
1266                 RETURN(rc);
1267
1268         /*
1269          * No RPC inside the transaction, so OST objects should be created at
1270          * first.
1271          */
1272         if (S_ISREG(attr->la_mode)) {
1273                 rc = mdd_lov_create(env, mdd, mdd_pobj, son, &lmm, &lmm_size,
1274                                     spec, attr);
1275                 if (rc)
1276                         RETURN(rc);
1277         }
1278
1279         mdd_txn_param_build(env, mdd, MDD_TXN_MKDIR_OP);
1280         handle = mdd_trans_start(env, mdd);
1281         if (IS_ERR(handle))
1282                 GOTO(out_free, rc = PTR_ERR(handle));
1283
1284         dlh = mdd_pdo_write_lock(env, mdd_pobj, name);
1285         if (dlh == NULL)
1286                 GOTO(out_trans, rc = -ENOMEM);
1287
1288         /*
1289          * XXX: Check that link can be added to the parent in mkdir case.
1290          */
1291
1292         mdd_write_lock(env, son);
1293         rc = mdd_object_create_internal(env, mdd_pobj, son, ma, handle);
1294         if (rc) {
1295                 mdd_write_unlock(env, son);
1296                 GOTO(cleanup, rc);
1297         }
1298
1299         created = 1;
1300
1301 #ifdef CONFIG_FS_POSIX_ACL
1302         mdd_read_lock(env, mdd_pobj);
1303         rc = mdd_acl_init(env, mdd_pobj, son, &ma->ma_attr.la_mode, handle);
1304         mdd_read_unlock(env, mdd_pobj);
1305         if (rc) {
1306                 mdd_write_unlock(env, son);
1307                 GOTO(cleanup, rc);
1308         } else {
1309                 ma->ma_attr.la_valid |= LA_MODE;
1310         }
1311 #endif
1312
1313         rc = mdd_object_initialize(env, mdo2fid(mdd_pobj),
1314                                    son, ma, handle);
1315         mdd_write_unlock(env, son);
1316         if (rc)
1317                 /*
1318                  * Object has no links, so it will be destroyed when last
1319                  * reference is released. (XXX not now.)
1320                  */
1321                 GOTO(cleanup, rc);
1322
1323         rc = __mdd_index_insert(env, mdd_pobj, mdo2fid(son),
1324                                 name, S_ISDIR(attr->la_mode), handle,
1325                                 mdd_object_capa(env, mdd_pobj));
1326
1327         if (rc)
1328                 GOTO(cleanup, rc);
1329
1330         inserted = 1;
1331
1332         /* No need mdd_lsm_sanity_check here */
1333         rc = mdd_lov_set_md(env, mdd_pobj, son, lmm, lmm_size, handle, 0);
1334         if (rc) {
1335                 CERROR("error on stripe info copy %d \n", rc);
1336                 GOTO(cleanup, rc);
1337         }
1338         if (lmm && lmm_size > 0) {
1339                 /* Set Lov here, do not get lmm again later */
1340                 memcpy(ma->ma_lmm, lmm, lmm_size);
1341                 ma->ma_lmm_size = lmm_size;
1342                 ma->ma_valid |= MA_LOV;
1343         }
1344
1345         if (S_ISLNK(attr->la_mode)) {
1346                 struct dt_object *dt = mdd_object_child(son);
1347                 const char *target_name = spec->u.sp_symname;
1348                 int sym_len = strlen(target_name);
1349                 const struct lu_buf *buf;
1350                 loff_t pos = 0;
1351
1352                 buf = mdd_buf_get_const(env, target_name, sym_len);
1353                 rc = dt->do_body_ops->dbo_write(env, dt, buf, &pos, handle,
1354                                                 mdd_object_capa(env, son));
1355
1356                 if (rc == sym_len)
1357                         rc = 0;
1358                 else
1359                         GOTO(cleanup, rc = -EFAULT);
1360         }
1361
1362         *la = ma->ma_attr;
1363         la->la_valid = LA_CTIME | LA_MTIME;
1364         rc = mdd_attr_set_internal_locked(env, mdd_pobj, la, handle, 0);
1365         if (rc)
1366                 GOTO(cleanup, rc);
1367
1368         /* Return attr back. */
1369         rc = mdd_attr_get_internal_locked(env, son, ma);
1370         EXIT;
1371 cleanup:
1372         if (rc && created) {
1373                 int rc2 = 0;
1374
1375                 if (inserted) {
1376                         rc2 = __mdd_index_delete(env, mdd_pobj, name,
1377                                                  S_ISDIR(attr->la_mode),
1378                                                  handle, BYPASS_CAPA);
1379                         if (rc2)
1380                                 CERROR("error can not cleanup destroy %d\n",
1381                                        rc2);
1382                 }
1383                 if (rc2 == 0) {
1384                         mdd_write_lock(env, son);
1385                         mdo_ref_del(env, son, handle);
1386                         mdd_write_unlock(env, son);
1387                 }
1388         }
1389
1390         /* update lov_objid data, must be before transaction stop! */
1391         if (rc == 0)
1392                 mdd_lov_objid_update(env, mdd);
1393
1394         mdd_pdo_write_unlock(env, mdd_pobj, dlh);
1395 out_trans:
1396         mdd_trans_stop(env, mdd, rc, handle);
1397 out_free:
1398         /* finis lov_create stuff, free all temporary data */
1399         mdd_lov_create_finish(env, mdd, lmm, lmm_size, spec);
1400         return rc;
1401 }
1402
1403 /*
1404  * Get locks on parents in proper order
1405  * RETURN: < 0 - error, rename_order if successful
1406  */
1407 enum rename_order {
1408         MDD_RN_SAME,
1409         MDD_RN_SRCTGT,
1410         MDD_RN_TGTSRC
1411 };
1412
1413 static int mdd_rename_order(const struct lu_env *env,
1414                             struct mdd_device *mdd,
1415                             struct mdd_object *src_pobj,
1416                             struct mdd_object *tgt_pobj)
1417 {
1418         /* order of locking, 1 - tgt-src, 0 - src-tgt*/
1419         int rc;
1420         ENTRY;
1421
1422         if (src_pobj == tgt_pobj)
1423                 RETURN(MDD_RN_SAME);
1424
1425         /* compared the parent child relationship of src_p&tgt_p */
1426         if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(src_pobj))){
1427                 rc = MDD_RN_SRCTGT;
1428         } else if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(tgt_pobj))) {
1429                 rc = MDD_RN_TGTSRC;
1430         } else {
1431                 rc = mdd_is_parent(env, mdd, src_pobj, mdo2fid(tgt_pobj), NULL);
1432                 if (rc == -EREMOTE)
1433                         rc = 0;
1434
1435                 if (rc == 1)
1436                         rc = MDD_RN_TGTSRC;
1437                 else if (rc == 0)
1438                         rc = MDD_RN_SRCTGT;
1439         }
1440
1441         RETURN(rc);
1442 }
1443
1444 /* has not mdd_write{read}_lock on any obj yet. */
1445 static int mdd_rename_sanity_check(const struct lu_env *env,
1446                                    struct mdd_object *src_pobj,
1447                                    struct mdd_object *tgt_pobj,
1448                                    struct mdd_object *sobj,
1449                                    struct mdd_object *tobj,
1450                                    struct md_attr *ma)
1451 {
1452         int rc = 0;
1453         ENTRY;
1454
1455         if (unlikely(ma->ma_attr_flags & MDS_PERM_BYPASS))
1456                 RETURN(0);
1457
1458         /* XXX: when get here, sobj must NOT be NULL,
1459          * the other case has been processed in cml_rename
1460          * before mdd_rename and enable MDS_PERM_BYPASS. */
1461         LASSERT(sobj);
1462         rc = mdd_may_delete(env, src_pobj, sobj, ma, 1, 0);
1463         if (rc)
1464                 RETURN(rc);
1465
1466         /* XXX: when get here, "tobj == NULL" means tobj must
1467          * NOT exist (neither on remote MDS, such case has been
1468          * processed in cml_rename before mdd_rename and enable
1469          * MDS_PERM_BYPASS).
1470          * So check may_create, but not check may_unlink. */
1471         if (!tobj)
1472                 rc = mdd_may_create(env, tgt_pobj, NULL,
1473                                     (src_pobj != tgt_pobj), 0);
1474         else
1475                 rc = mdd_may_delete(env, tgt_pobj, tobj, ma,
1476                                     (src_pobj != tgt_pobj), 1);
1477
1478         if (!rc && !tobj && (src_pobj != tgt_pobj) &&
1479             S_ISDIR(ma->ma_attr.la_mode))
1480                 rc = __mdd_may_link(env, tgt_pobj);
1481
1482         RETURN(rc);
1483 }
1484
1485 /* src object can be remote that is why we use only fid and type of object */
1486 static int mdd_rename(const struct lu_env *env,
1487                       struct md_object *src_pobj, struct md_object *tgt_pobj,
1488                       const struct lu_fid *lf, const struct lu_name *lsname,
1489                       struct md_object *tobj, const struct lu_name *ltname,
1490                       struct md_attr *ma)
1491 {
1492         char *sname = lsname->ln_name;
1493         char *tname = ltname->ln_name;
1494         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
1495         struct mdd_object *mdd_spobj = md2mdd_obj(src_pobj);
1496         struct mdd_object *mdd_tpobj = md2mdd_obj(tgt_pobj);
1497         struct mdd_device *mdd = mdo2mdd(src_pobj);
1498         struct mdd_object *mdd_sobj = NULL;
1499         struct mdd_object *mdd_tobj = NULL;
1500         struct dynlock_handle *sdlh, *tdlh;
1501         struct thandle *handle;
1502         int is_dir;
1503         int rc;
1504         ENTRY;
1505
1506         LASSERT(ma->ma_attr.la_mode & S_IFMT);
1507         is_dir = S_ISDIR(ma->ma_attr.la_mode);
1508
1509         if (tobj)
1510                 mdd_tobj = md2mdd_obj(tobj);
1511
1512         mdd_txn_param_build(env, mdd, MDD_TXN_RENAME_OP);
1513         handle = mdd_trans_start(env, mdd);
1514         if (IS_ERR(handle))
1515                 RETURN(PTR_ERR(handle));
1516
1517         /* FIXME: Should consider tobj and sobj too in rename_lock. */
1518         rc = mdd_rename_order(env, mdd, mdd_spobj, mdd_tpobj);
1519         if (rc < 0)
1520                 GOTO(cleanup_unlocked, rc);
1521
1522         /* Get locks in determined order */
1523         if (rc == MDD_RN_SAME) {
1524                 sdlh = mdd_pdo_write_lock(env, mdd_spobj, sname);
1525                 /* check hashes to determine do we need one lock or two */
1526                 if (mdd_name2hash(sname) != mdd_name2hash(tname))
1527                         tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname);
1528                 else
1529                         tdlh = sdlh;
1530         } else if (rc == MDD_RN_SRCTGT) {
1531                 sdlh = mdd_pdo_write_lock(env, mdd_spobj, sname);
1532                 tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname);
1533         } else {
1534                 tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname);
1535                 sdlh = mdd_pdo_write_lock(env, mdd_spobj, sname);
1536         }
1537         if (sdlh == NULL || tdlh == NULL)
1538                 GOTO(cleanup, rc = -ENOMEM);
1539
1540         mdd_sobj = mdd_object_find(env, mdd, lf);
1541         rc = mdd_rename_sanity_check(env, mdd_spobj, mdd_tpobj,
1542                                      mdd_sobj, mdd_tobj, ma);
1543         if (rc)
1544                 GOTO(cleanup, rc);
1545
1546         rc = __mdd_index_delete(env, mdd_spobj, sname, is_dir, handle,
1547                                 mdd_object_capa(env, mdd_spobj));
1548         if (rc)
1549                 GOTO(cleanup, rc);
1550
1551         /*
1552          * Here tobj can be remote one, so we do index_delete unconditionally
1553          * and -ENOENT is allowed.
1554          */
1555         rc = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle,
1556                                 mdd_object_capa(env, mdd_tpobj));
1557         if (rc != 0 && rc != -ENOENT)
1558                 GOTO(cleanup, rc);
1559
1560         rc = __mdd_index_insert(env, mdd_tpobj, lf, tname, is_dir, handle,
1561                                 mdd_object_capa(env, mdd_tpobj));
1562         if (rc)
1563                 GOTO(cleanup, rc);
1564
1565         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1566         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1567
1568         /* XXX: mdd_sobj must be local one if it is NOT NULL. */
1569         if (mdd_sobj) {
1570                 la->la_valid = LA_CTIME;
1571                 rc = mdd_attr_set_internal_locked(env, mdd_sobj, la, handle, 0);
1572                 if (rc)
1573                         GOTO(cleanup, rc);
1574         }
1575
1576         /* 
1577          * For tobj is remote case cmm layer has processed
1578          * and set tobj to NULL then. So when tobj is NOT NULL,
1579          * it must be local one.
1580          */
1581         if (tobj && mdd_object_exists(mdd_tobj)) {
1582                 mdd_write_lock(env, mdd_tobj);
1583                 mdo_ref_del(env, mdd_tobj, handle);
1584
1585                 /* Remove dot reference. */
1586                 if (is_dir)
1587                         mdo_ref_del(env, mdd_tobj, handle);
1588
1589                 la->la_valid = LA_CTIME;
1590                 rc = mdd_attr_set_internal(env, mdd_tobj, la, handle, 0);
1591                 if (rc)
1592                         GOTO(cleanup, rc);
1593
1594                 rc = mdd_finish_unlink(env, mdd_tobj, ma, handle);
1595                 mdd_write_unlock(env, mdd_tobj);
1596                 if (rc)
1597                         GOTO(cleanup, rc);
1598         }
1599
1600         la->la_valid = LA_CTIME | LA_MTIME;
1601         rc = mdd_attr_set_internal_locked(env, mdd_spobj, la, handle, 0);
1602         if (rc)
1603                 GOTO(cleanup, rc);
1604
1605         if (mdd_spobj != mdd_tpobj) {
1606                 la->la_valid = LA_CTIME | LA_MTIME;
1607                 rc = mdd_attr_set_internal_locked(env, mdd_tpobj, la,
1608                                                   handle, 0);
1609         }
1610
1611         EXIT;
1612 cleanup:
1613         if (likely(tdlh) && sdlh != tdlh)
1614                 mdd_pdo_write_unlock(env, mdd_tpobj, tdlh);
1615         if (likely(sdlh))
1616                 mdd_pdo_write_unlock(env, mdd_spobj, sdlh);
1617 cleanup_unlocked:
1618         mdd_trans_stop(env, mdd, rc, handle);
1619         if (mdd_sobj)
1620                 mdd_object_put(env, mdd_sobj);
1621         return rc;
1622 }
1623
1624 struct md_dir_operations mdd_dir_ops = {
1625         .mdo_is_subdir     = mdd_is_subdir,
1626         .mdo_lookup        = mdd_lookup,
1627         .mdo_create        = mdd_create,
1628         .mdo_rename        = mdd_rename,
1629         .mdo_link          = mdd_link,
1630         .mdo_unlink        = mdd_unlink,
1631         .mdo_name_insert   = mdd_name_insert,
1632         .mdo_name_remove   = mdd_name_remove,
1633         .mdo_rename_tgt    = mdd_rename_tgt,
1634         .mdo_create_data   = mdd_create_data
1635 };