Whamcloud - gitweb
LU-909 osd: changes to osd api
[fs/lustre-release.git] / lustre / cmm / cmm_object.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
30  * Use is subject to license terms.
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/cmm/cmm_object.c
37  *
38  * Lustre Cluster Metadata Manager (cmm)
39  *
40  * Author: Mike Pershin <tappro@clusterfs.com>
41  */
42
43 #ifndef EXPORT_SYMTAB
44 # define EXPORT_SYMTAB
45 #endif
46
47 #define DEBUG_SUBSYSTEM S_MDS
48
49 #include <lustre_fid.h>
50 #include "cmm_internal.h"
51 #include "mdc_internal.h"
52 /**
53  * \ingroup cmm
54  * Lookup MDS number \a mds by FID \a fid.
55  *
56  * \param fid FID of object to find MDS
57  * \param mds mds number to return.
58  */
59 int cmm_fld_lookup(struct cmm_device *cm, const struct lu_fid *fid,
60                    mdsno_t *mds, const struct lu_env *env)
61 {
62         int rc = 0;
63         ENTRY;
64
65         LASSERT(fid_is_sane(fid));
66
67         rc = fld_client_lookup(cm->cmm_fld, fid_seq(fid), mds,
68                                LU_SEQ_RANGE_MDT, env);
69         if (rc) {
70                 CERROR("Can't find mds by seq "LPX64", rc %d\n",
71                        fid_seq(fid), rc);
72                 RETURN(rc);
73         }
74
75         if (*mds > cm->cmm_tgt_count) {
76                 CERROR("Got invalid mdsno: %x (max: %x)\n",
77                        *mds, cm->cmm_tgt_count);
78                 rc = -EINVAL;
79         } else {
80                 CDEBUG(D_INFO, "CMM: got MDS %x for sequence: "
81                        LPX64"\n", *mds, fid_seq(fid));
82         }
83
84         RETURN (rc);
85 }
86
87 /**
88  * \addtogroup cml
89  * @{
90  */
91 static const struct md_object_operations cml_mo_ops;
92 static const struct md_dir_operations    cml_dir_ops;
93 static const struct lu_object_operations cml_obj_ops;
94
95 static const struct md_object_operations cmr_mo_ops;
96 static const struct md_dir_operations    cmr_dir_ops;
97 static const struct lu_object_operations cmr_obj_ops;
98
99 /**
100  * \ingroup cmm
101  * Allocate CMM object.
102  */
103 struct lu_object *cmm_object_alloc(const struct lu_env *env,
104                                    const struct lu_object_header *loh,
105                                    struct lu_device *ld)
106 {
107         const struct lu_fid *fid = &loh->loh_fid;
108         struct lu_object  *lo = NULL;
109         struct cmm_device *cd;
110         mdsno_t mds;
111         int rc = 0;
112
113         ENTRY;
114
115         cd = lu2cmm_dev(ld);
116         if (cd->cmm_flags & CMM_INITIALIZED) {
117                 /* get object location */
118                 rc = cmm_fld_lookup(lu2cmm_dev(ld), fid, &mds, env);
119                 if (rc)
120                         RETURN(NULL);
121         } else
122                 /*
123                  * Device is not yet initialized, cmm_object is being created
124                  * as part of early bootstrap procedure (it is /ROOT, or /fld,
125                  * etc.). Such object *has* to be local.
126                  */
127                 mds = cd->cmm_local_num;
128
129         /* select the proper set of operations based on object location */
130         if (mds == cd->cmm_local_num) {
131                 struct cml_object *clo;
132
133                 OBD_ALLOC_PTR(clo);
134                 if (clo != NULL) {
135                         lo = &clo->cmm_obj.cmo_obj.mo_lu;
136                         lu_object_init(lo, NULL, ld);
137                         clo->cmm_obj.cmo_obj.mo_ops = &cml_mo_ops;
138                         clo->cmm_obj.cmo_obj.mo_dir_ops = &cml_dir_ops;
139                         lo->lo_ops = &cml_obj_ops;
140                 }
141         } else {
142                 struct cmr_object *cro;
143
144                 OBD_ALLOC_PTR(cro);
145                 if (cro != NULL) {
146                         lo = &cro->cmm_obj.cmo_obj.mo_lu;
147                         lu_object_init(lo, NULL, ld);
148                         cro->cmm_obj.cmo_obj.mo_ops = &cmr_mo_ops;
149                         cro->cmm_obj.cmo_obj.mo_dir_ops = &cmr_dir_ops;
150                         lo->lo_ops = &cmr_obj_ops;
151                         cro->cmo_num = mds;
152                 }
153         }
154         RETURN(lo);
155 }
156
157 /**
158  * Get local child device.
159  */
160 static struct lu_device *cml_child_dev(struct cmm_device *d)
161 {
162         return &d->cmm_child->md_lu_dev;
163 }
164
165 /**
166  * Free cml_object.
167  */
168 static void cml_object_free(const struct lu_env *env,
169                             struct lu_object *lo)
170 {
171         struct cml_object *clo = lu2cml_obj(lo);
172         lu_object_fini(lo);
173         OBD_FREE_PTR(clo);
174 }
175
176 /**
177  * Initialize cml_object.
178  */
179 static int cml_object_init(const struct lu_env *env, struct lu_object *lo,
180                            const struct lu_object_conf *unused)
181 {
182         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
183         struct lu_device  *c_dev;
184         struct lu_object  *c_obj;
185         int rc;
186
187         ENTRY;
188
189 #ifdef HAVE_SPLIT_SUPPORT
190         if (cd->cmm_tgt_count == 0)
191                 lu2cml_obj(lo)->clo_split = CMM_SPLIT_DENIED;
192         else
193                 lu2cml_obj(lo)->clo_split = CMM_SPLIT_UNKNOWN;
194 #endif
195         c_dev = cml_child_dev(cd);
196         if (c_dev == NULL) {
197                 rc = -ENOENT;
198         } else {
199                 c_obj = c_dev->ld_ops->ldo_object_alloc(env,
200                                                         lo->lo_header, c_dev);
201                 if (c_obj != NULL) {
202                         lu_object_add(lo, c_obj);
203                         rc = 0;
204                 } else {
205                         rc = -ENOMEM;
206                 }
207         }
208
209         RETURN(rc);
210 }
211
212 static int cml_object_print(const struct lu_env *env, void *cookie,
213                             lu_printer_t p, const struct lu_object *lo)
214 {
215         return (*p)(env, cookie, "[local]");
216 }
217
218 static const struct lu_object_operations cml_obj_ops = {
219         .loo_object_init    = cml_object_init,
220         .loo_object_free    = cml_object_free,
221         .loo_object_print   = cml_object_print
222 };
223
224 /**
225  * \name CMM local md_object operations.
226  * All of them call just corresponding operations on next layer.
227  * @{
228  */
229 static int cml_object_create(const struct lu_env *env,
230                              struct md_object *mo,
231                              const struct md_op_spec *spec,
232                              struct md_attr *attr)
233 {
234         int rc;
235         ENTRY;
236         rc = mo_object_create(env, md_object_next(mo), spec, attr);
237         RETURN(rc);
238 }
239
240 static int cml_permission(const struct lu_env *env,
241                           struct md_object *p, struct md_object *c,
242                           struct md_attr *attr, int mask)
243 {
244         int rc;
245         ENTRY;
246         rc = mo_permission(env, md_object_next(p), md_object_next(c),
247                            attr, mask);
248         RETURN(rc);
249 }
250
251 static int cml_attr_get(const struct lu_env *env, struct md_object *mo,
252                         struct md_attr *attr)
253 {
254         int rc;
255         ENTRY;
256         rc = mo_attr_get(env, md_object_next(mo), attr);
257         RETURN(rc);
258 }
259
260 static int cml_attr_set(const struct lu_env *env, struct md_object *mo,
261                         const struct md_attr *attr)
262 {
263         int rc;
264         ENTRY;
265         rc = mo_attr_set(env, md_object_next(mo), attr);
266         RETURN(rc);
267 }
268
269 static int cml_xattr_get(const struct lu_env *env, struct md_object *mo,
270                          struct lu_buf *buf, const char *name)
271 {
272         int rc;
273         ENTRY;
274         rc = mo_xattr_get(env, md_object_next(mo), buf, name);
275         RETURN(rc);
276 }
277
278 static int cml_readlink(const struct lu_env *env, struct md_object *mo,
279                         struct lu_buf *buf)
280 {
281         int rc;
282         ENTRY;
283         rc = mo_readlink(env, md_object_next(mo), buf);
284         RETURN(rc);
285 }
286
287 static int cml_changelog(const struct lu_env *env, enum changelog_rec_type type,
288                          int flags, struct md_object *mo)
289 {
290         int rc;
291         ENTRY;
292         rc = mo_changelog(env, type, flags, md_object_next(mo));
293         RETURN(rc);
294 }
295
296 static int cml_xattr_list(const struct lu_env *env, struct md_object *mo,
297                           struct lu_buf *buf)
298 {
299         int rc;
300         ENTRY;
301         rc = mo_xattr_list(env, md_object_next(mo), buf);
302         RETURN(rc);
303 }
304
305 static int cml_xattr_set(const struct lu_env *env, struct md_object *mo,
306                          const struct lu_buf *buf, const char *name,
307                          int fl)
308 {
309         int rc;
310         ENTRY;
311         rc = mo_xattr_set(env, md_object_next(mo), buf, name, fl);
312         RETURN(rc);
313 }
314
315 static int cml_xattr_del(const struct lu_env *env, struct md_object *mo,
316                          const char *name)
317 {
318         int rc;
319         ENTRY;
320         rc = mo_xattr_del(env, md_object_next(mo), name);
321         RETURN(rc);
322 }
323
324 static int cml_ref_add(const struct lu_env *env, struct md_object *mo,
325                        const struct md_attr *ma)
326 {
327         int rc;
328         ENTRY;
329         rc = mo_ref_add(env, md_object_next(mo), ma);
330         RETURN(rc);
331 }
332
333 static int cml_ref_del(const struct lu_env *env, struct md_object *mo,
334                        struct md_attr *ma)
335 {
336         int rc;
337         ENTRY;
338         rc = mo_ref_del(env, md_object_next(mo), ma);
339         RETURN(rc);
340 }
341
342 static int cml_open(const struct lu_env *env, struct md_object *mo,
343                     int flags)
344 {
345         int rc;
346         ENTRY;
347         rc = mo_open(env, md_object_next(mo), flags);
348         RETURN(rc);
349 }
350
351 static int cml_close(const struct lu_env *env, struct md_object *mo,
352                      struct md_attr *ma, int mode)
353 {
354         int rc;
355         ENTRY;
356         rc = mo_close(env, md_object_next(mo), ma, mode);
357         RETURN(rc);
358 }
359
360 static int cml_readpage(const struct lu_env *env, struct md_object *mo,
361                         const struct lu_rdpg *rdpg)
362 {
363         int rc;
364         ENTRY;
365         rc = mo_readpage(env, md_object_next(mo), rdpg);
366         RETURN(rc);
367 }
368
369 static int cml_capa_get(const struct lu_env *env, struct md_object *mo,
370                         struct lustre_capa *capa, int renewal)
371 {
372         int rc;
373         ENTRY;
374         rc = mo_capa_get(env, md_object_next(mo), capa, renewal);
375         RETURN(rc);
376 }
377
378 static int cml_path(const struct lu_env *env, struct md_object *mo,
379                     char *path, int pathlen, __u64 *recno, int *linkno)
380 {
381         int rc;
382         ENTRY;
383         rc = mo_path(env, md_object_next(mo), path, pathlen, recno, linkno);
384         RETURN(rc);
385 }
386
387 static int cml_file_lock(const struct lu_env *env, struct md_object *mo,
388                          struct lov_mds_md *lmm, struct ldlm_extent *extent,
389                          struct lustre_handle *lockh)
390 {
391         int rc;
392         ENTRY;
393         rc = mo_file_lock(env, md_object_next(mo), lmm, extent, lockh);
394         RETURN(rc);
395 }
396
397 static int cml_file_unlock(const struct lu_env *env, struct md_object *mo,
398                            struct lov_mds_md *lmm, struct lustre_handle *lockh)
399 {
400         int rc;
401         ENTRY;
402         rc = mo_file_unlock(env, md_object_next(mo), lmm, lockh);
403         RETURN(rc);
404 }
405
406 static int cml_object_sync(const struct lu_env *env, struct md_object *mo)
407 {
408         int rc;
409         ENTRY;
410         rc = mo_object_sync(env, md_object_next(mo));
411         RETURN(rc);
412 }
413
414 static const struct md_object_operations cml_mo_ops = {
415         .moo_permission    = cml_permission,
416         .moo_attr_get      = cml_attr_get,
417         .moo_attr_set      = cml_attr_set,
418         .moo_xattr_get     = cml_xattr_get,
419         .moo_xattr_list    = cml_xattr_list,
420         .moo_xattr_set     = cml_xattr_set,
421         .moo_xattr_del     = cml_xattr_del,
422         .moo_object_create = cml_object_create,
423         .moo_ref_add       = cml_ref_add,
424         .moo_ref_del       = cml_ref_del,
425         .moo_open          = cml_open,
426         .moo_close         = cml_close,
427         .moo_readpage      = cml_readpage,
428         .moo_readlink      = cml_readlink,
429         .moo_changelog     = cml_changelog,
430         .moo_capa_get      = cml_capa_get,
431         .moo_object_sync   = cml_object_sync,
432         .moo_path          = cml_path,
433         .moo_file_lock     = cml_file_lock,
434         .moo_file_unlock   = cml_file_unlock,
435 };
436 /** @} */
437
438 /**
439  * \name CMM local md_dir_operations.
440  * @{
441  */
442 /**
443  * cml lookup object fid by name.
444  * This returns only FID by name.
445  */
446 static int cml_lookup(const struct lu_env *env, struct md_object *mo_p,
447                       const struct lu_name *lname, struct lu_fid *lf,
448                       struct md_op_spec *spec)
449 {
450         int rc;
451         ENTRY;
452
453 #ifdef HAVE_SPLIT_SUPPORT
454         if (spec != NULL && spec->sp_ck_split) {
455                 rc = cmm_split_check(env, mo_p, lname->ln_name);
456                 if (rc)
457                         RETURN(rc);
458         }
459 #endif
460         rc = mdo_lookup(env, md_object_next(mo_p), lname, lf, spec);
461         RETURN(rc);
462
463 }
464
465 /**
466  * Helper to return lock mode. Used in split cases only.
467  */
468 static mdl_mode_t cml_lock_mode(const struct lu_env *env,
469                                 struct md_object *mo, mdl_mode_t lm)
470 {
471         int rc = MDL_MINMODE;
472         ENTRY;
473
474 #ifdef HAVE_SPLIT_SUPPORT
475         rc = cmm_split_access(env, mo, lm);
476 #endif
477
478         RETURN(rc);
479 }
480
481 /**
482  * Create operation for cml.
483  * Objects are local, but split can happen.
484  * If split is not needed this will call next layer mdo_create().
485  *
486  * \param mo_p Parent directory. Local object.
487  * \param lname name of file to create.
488  * \param mo_c Child object. It has no real inode yet.
489  * \param spec creation specification.
490  * \param ma child object attributes.
491  */
492 static int cml_create(const struct lu_env *env, struct md_object *mo_p,
493                       const struct lu_name *lname, struct md_object *mo_c,
494                       struct md_op_spec *spec, struct md_attr *ma)
495 {
496         int rc;
497         ENTRY;
498
499 #ifdef HAVE_SPLIT_SUPPORT
500         /* Lock mode always should be sane. */
501         LASSERT(spec->sp_cr_mode != MDL_MINMODE);
502
503         /*
504          * Sigh... This is long story. MDT may have race with detecting if split
505          * is possible in cmm. We know this race and let it live, because
506          * getting it rid (with some sem or spinlock) will also mean that
507          * PDIROPS for create will not work because we kill parallel work, what
508          * is really bad for performance and makes no sense having PDIROPS. So,
509          * we better allow the race to live, but split dir only if some of
510          * concurrent threads takes EX lock, not matter which one. So that, say,
511          * two concurrent threads may have different lock modes on directory (CW
512          * and EX) and not first one which comes here and see that split is
513          * possible should split the dir, but only that one which has EX
514          * lock. And we do not care that in this case, split may happen a bit
515          * later (when dir size will not be necessarily 64K, but may be a bit
516          * larger). So that, we allow concurrent creates and protect split by EX
517          * lock.
518          */
519         if (spec->sp_cr_mode == MDL_EX) {
520                 /**
521                  * Split cases:
522                  * - Try to split \a mo_p upon each create operation.
523                  *   If split is ok, -ERESTART is returned and current thread
524                  *   will not peoceed with create. Instead it sends -ERESTART
525                  *   to client to let it know that correct MDT must be chosen.
526                  * \see cmm_split_dir()
527                  */
528                 rc = cmm_split_dir(env, mo_p);
529                 if (rc)
530                         /*
531                          * -ERESTART or some split error is returned, we can't
532                          * proceed with create.
533                          */
534                         GOTO(out, rc);
535         }
536
537         if (spec != NULL && spec->sp_ck_split) {
538                 /**
539                  * - Directory is split already. Let the caller know that
540                  * it should tell client that directory is split and operation
541                  * should repeat to correct MDT.
542                  * \see cmm_split_check()
543                  */
544                 rc = cmm_split_check(env, mo_p, lname->ln_name);
545                 if (rc)
546                         GOTO(out, rc);
547         }
548 #endif
549
550         rc = mdo_create(env, md_object_next(mo_p), lname, md_object_next(mo_c),
551                         spec, ma);
552
553         EXIT;
554 #ifdef HAVE_SPLIT_SUPPORT
555 out:
556 #endif
557         return rc;
558 }
559
560 /** Call mdo_create_data() on next layer. All objects are local. */
561 static int cml_create_data(const struct lu_env *env, struct md_object *p,
562                            struct md_object *o,
563                            const struct md_op_spec *spec,
564                            struct md_attr *ma)
565 {
566         int rc;
567         ENTRY;
568         rc = mdo_create_data(env, md_object_next(p), md_object_next(o),
569                              spec, ma);
570         RETURN(rc);
571 }
572
573 /** Call mdo_link() on next layer. All objects are local. */
574 static int cml_link(const struct lu_env *env, struct md_object *mo_p,
575                     struct md_object *mo_s, const struct lu_name *lname,
576                     struct md_attr *ma)
577 {
578         int rc;
579         ENTRY;
580         rc = mdo_link(env, md_object_next(mo_p), md_object_next(mo_s),
581                       lname, ma);
582         RETURN(rc);
583 }
584
585 /** Call mdo_unlink() on next layer. All objects are local. */
586 static int cml_unlink(const struct lu_env *env, struct md_object *mo_p,
587                       struct md_object *mo_c, const struct lu_name *lname,
588                       struct md_attr *ma)
589 {
590         int rc;
591         ENTRY;
592         rc = mdo_unlink(env, md_object_next(mo_p), md_object_next(mo_c),
593                         lname, ma);
594         RETURN(rc);
595 }
596
597 /**
598  * \ingroup cmm
599  * Get mode of object.
600  * Used in both cml and cmr hence can produce RPC to another server.
601  */
602 static int cmm_mode_get(const struct lu_env *env, struct md_device *md,
603                         const struct lu_fid *lf, struct md_attr *ma,
604                         int *remote)
605 {
606         struct md_object *mo_s = md_object_find_slice(env, md, lf);
607         struct cmm_thread_info *cmi;
608         struct md_attr *tmp_ma;
609         int rc;
610         ENTRY;
611
612         if (IS_ERR(mo_s))
613                 RETURN(PTR_ERR(mo_s));
614
615         if (remote && (lu_object_exists(&mo_s->mo_lu) < 0))
616                 *remote = 1;
617
618         cmi = cmm_env_info(env);
619         tmp_ma = &cmi->cmi_ma;
620         tmp_ma->ma_need = MA_INODE;
621         tmp_ma->ma_valid = 0;
622         /* get type from src, can be remote req */
623         rc = mo_attr_get(env, md_object_next(mo_s), tmp_ma);
624         if (rc == 0) {
625                 ma->ma_attr.la_mode = tmp_ma->ma_attr.la_mode;
626                 ma->ma_attr.la_uid = tmp_ma->ma_attr.la_uid;
627                 ma->ma_attr.la_gid = tmp_ma->ma_attr.la_gid;
628                 ma->ma_attr.la_flags = tmp_ma->ma_attr.la_flags;
629                 ma->ma_attr.la_valid |= LA_MODE | LA_UID | LA_GID | LA_FLAGS;
630         }
631         lu_object_put(env, &mo_s->mo_lu);
632         RETURN(rc);
633 }
634
635 /**
636  * \ingroup cmm
637  * Set ctime for object.
638  * Used in both cml and cmr hence can produce RPC to another server.
639  */
640 static int cmm_rename_ctime(const struct lu_env *env, struct md_device *md,
641                             const struct lu_fid *lf, struct md_attr *ma)
642 {
643         struct md_object *mo_s = md_object_find_slice(env, md, lf);
644         int rc;
645         ENTRY;
646
647         if (IS_ERR(mo_s))
648                 RETURN(PTR_ERR(mo_s));
649
650         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
651         /* set ctime to obj, can be remote req */
652         rc = mo_attr_set(env, md_object_next(mo_s), ma);
653         lu_object_put(env, &mo_s->mo_lu);
654         RETURN(rc);
655 }
656
657 /** Helper to output debug information about rename operation. */
658 static inline void cml_rename_warn(const char *fname,
659                                   struct md_object *mo_po,
660                                   struct md_object *mo_pn,
661                                   const struct lu_fid *lf,
662                                   const char *s_name,
663                                   struct md_object *mo_t,
664                                   const char *t_name,
665                                   int err)
666 {
667         if (mo_t)
668                 CWARN("cml_rename failed for %s, should revoke: [mo_po "DFID"] "
669                       "[mo_pn "DFID"] [lf "DFID"] [sname %s] [mo_t "DFID"] "
670                       "[tname %s] [err %d]\n", fname,
671                       PFID(lu_object_fid(&mo_po->mo_lu)),
672                       PFID(lu_object_fid(&mo_pn->mo_lu)),
673                       PFID(lf), s_name,
674                       PFID(lu_object_fid(&mo_t->mo_lu)),
675                       t_name, err);
676         else
677                 CWARN("cml_rename failed for %s, should revoke: [mo_po "DFID"] "
678                       "[mo_pn "DFID"] [lf "DFID"] [sname %s] [mo_t NULL] "
679                       "[tname %s] [err %d]\n", fname,
680                       PFID(lu_object_fid(&mo_po->mo_lu)),
681                       PFID(lu_object_fid(&mo_pn->mo_lu)),
682                       PFID(lf), s_name,
683                       t_name, err);
684 }
685
686 /**
687  * Rename operation for cml.
688  *
689  * This is the most complex cross-reference operation. It may consist of up to 4
690  * MDS server and require several RPCs to be sent.
691  *
692  * \param mo_po Old parent object.
693  * \param mo_pn New parent object.
694  * \param lf FID of object to rename.
695  * \param ls_name Source file name.
696  * \param mo_t target object. Should be NULL here.
697  * \param lt_name Name of target file.
698  * \param ma object attributes.
699  */
700 static int cml_rename(const struct lu_env *env, struct md_object *mo_po,
701                       struct md_object *mo_pn, const struct lu_fid *lf,
702                       const struct lu_name *ls_name, struct md_object *mo_t,
703                       const struct lu_name *lt_name, struct md_attr *ma)
704 {
705         struct cmm_thread_info *cmi;
706         struct md_attr *tmp_ma = NULL;
707         struct md_object *tmp_t = mo_t;
708         int remote = 0, rc;
709         ENTRY;
710
711         rc = cmm_mode_get(env, md_obj2dev(mo_po), lf, ma, &remote);
712         if (rc)
713                 RETURN(rc);
714
715         if (mo_t && lu_object_exists(&mo_t->mo_lu) < 0) {
716                 /**
717                  * \note \a mo_t is remote object and there is RPC to unlink it.
718                  * Before that, do local sanity check for rename first.
719                  */
720                 if (!remote) {
721                         struct md_object *mo_s = md_object_find_slice(env,
722                                                         md_obj2dev(mo_po), lf);
723                         if (IS_ERR(mo_s))
724                                 RETURN(PTR_ERR(mo_s));
725
726                         LASSERT(lu_object_exists(&mo_s->mo_lu) > 0);
727                         rc = mo_permission(env, md_object_next(mo_po),
728                                            md_object_next(mo_s),
729                                            ma, MAY_RENAME_SRC);
730                         lu_object_put(env, &mo_s->mo_lu);
731                         if (rc)
732                                 RETURN(rc);
733                 } else {
734                         rc = mo_permission(env, NULL, md_object_next(mo_po),
735                                            ma, MAY_UNLINK | MAY_VTX_FULL);
736                         if (rc)
737                                 RETURN(rc);
738                 }
739
740                 rc = mo_permission(env, NULL, md_object_next(mo_pn), ma,
741                                    MAY_UNLINK | MAY_VTX_PART);
742                 if (rc)
743                         RETURN(rc);
744
745                 /*
746                  * /note \a ma will be changed after mo_ref_del(), but we will use
747                  * it for mdo_rename() later, so save it before mo_ref_del().
748                  */
749                 cmi = cmm_env_info(env);
750                 tmp_ma = &cmi->cmi_ma;
751                 *tmp_ma = *ma;
752                 rc = mo_ref_del(env, md_object_next(mo_t), ma);
753                 if (rc)
754                         RETURN(rc);
755
756                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
757                 mo_t = NULL;
758         }
759
760         /**
761          * \note for src on remote MDS case, change its ctime before local
762          * rename. Firstly, do local sanity check for rename if necessary.
763          */
764         if (remote) {
765                 if (!tmp_ma) {
766                         rc = mo_permission(env, NULL, md_object_next(mo_po),
767                                            ma, MAY_UNLINK | MAY_VTX_FULL);
768                         if (rc)
769                                 RETURN(rc);
770
771                         if (mo_t) {
772                                 LASSERT(lu_object_exists(&mo_t->mo_lu) > 0);
773                                 rc = mo_permission(env, md_object_next(mo_pn),
774                                                    md_object_next(mo_t),
775                                                    ma, MAY_RENAME_TAR);
776                                 if (rc)
777                                         RETURN(rc);
778                         } else {
779                                 int mask;
780
781                                 if (mo_po != mo_pn)
782                                         mask = (S_ISDIR(ma->ma_attr.la_mode) ?
783                                                 MAY_LINK : MAY_CREATE);
784                                 else
785                                         mask = MAY_CREATE;
786                                 rc = mo_permission(env, NULL,
787                                                    md_object_next(mo_pn),
788                                                    NULL, mask);
789                                 if (rc)
790                                         RETURN(rc);
791                         }
792
793                         ma->ma_attr_flags |= MDS_PERM_BYPASS;
794                 } else {
795                         LASSERT(tmp_ma->ma_attr_flags & MDS_PERM_BYPASS);
796                 }
797
798                 rc = cmm_rename_ctime(env, md_obj2dev(mo_po), lf,
799                                       tmp_ma ? tmp_ma : ma);
800                 if (rc) {
801                         /* TODO: revoke mo_t if necessary. */
802                         cml_rename_warn("cmm_rename_ctime", mo_po,
803                                         mo_pn, lf, ls_name->ln_name,
804                                         tmp_t, lt_name->ln_name, rc);
805                         RETURN(rc);
806                 }
807         }
808
809         /* local rename, mo_t can be NULL */
810         rc = mdo_rename(env, md_object_next(mo_po),
811                         md_object_next(mo_pn), lf, ls_name,
812                         md_object_next(mo_t), lt_name, tmp_ma ? tmp_ma : ma);
813         if (rc)
814                 /* TODO: revoke all cml_rename */
815                 cml_rename_warn("mdo_rename", mo_po, mo_pn, lf,
816                                 ls_name->ln_name, tmp_t, lt_name->ln_name, rc);
817
818         RETURN(rc);
819 }
820
821 /**
822  * Rename target partial operation.
823  * Used for cross-ref rename.
824  */
825 static int cml_rename_tgt(const struct lu_env *env, struct md_object *mo_p,
826                           struct md_object *mo_t, const struct lu_fid *lf,
827                           const struct lu_name *lname, struct md_attr *ma)
828 {
829         int rc;
830         ENTRY;
831
832         rc = mdo_rename_tgt(env, md_object_next(mo_p),
833                             md_object_next(mo_t), lf, lname, ma);
834         RETURN(rc);
835 }
836
837 /**
838  * Name insert only operation.
839  * used only in case of rename_tgt() when target doesn't exist.
840  */
841 static int cml_name_insert(const struct lu_env *env, struct md_object *p,
842                            const struct lu_name *lname, const struct lu_fid *lf,
843                            const struct md_attr *ma)
844 {
845         int rc;
846         ENTRY;
847
848         rc = mdo_name_insert(env, md_object_next(p), lname, lf, ma);
849
850         RETURN(rc);
851 }
852
853 /**
854  * \ingroup cmm
855  * Check two fids are not subdirectories.
856  */
857 static int cmm_is_subdir(const struct lu_env *env, struct md_object *mo,
858                          const struct lu_fid *fid, struct lu_fid *sfid)
859 {
860         struct cmm_thread_info *cmi;
861         int rc;
862         ENTRY;
863
864         cmi = cmm_env_info(env);
865         rc = cmm_mode_get(env, md_obj2dev(mo), fid, &cmi->cmi_ma, NULL);
866         if (rc)
867                 RETURN(rc);
868
869         if (!S_ISDIR(cmi->cmi_ma.ma_attr.la_mode))
870                 RETURN(0);
871
872         rc = mdo_is_subdir(env, md_object_next(mo), fid, sfid);
873         RETURN(rc);
874 }
875
876 static const struct md_dir_operations cml_dir_ops = {
877         .mdo_is_subdir   = cmm_is_subdir,
878         .mdo_lookup      = cml_lookup,
879         .mdo_lock_mode   = cml_lock_mode,
880         .mdo_create      = cml_create,
881         .mdo_link        = cml_link,
882         .mdo_unlink      = cml_unlink,
883         .mdo_name_insert = cml_name_insert,
884         .mdo_rename      = cml_rename,
885         .mdo_rename_tgt  = cml_rename_tgt,
886         .mdo_create_data = cml_create_data,
887 };
888 /** @} */
889 /** @} */
890
891 /**
892  * \addtogroup cmr
893  * @{
894  */
895 /**
896  * \name cmr helpers
897  * @{
898  */
899 /** Get cmr_object from lu_object. */
900 static inline struct cmr_object *lu2cmr_obj(struct lu_object *o)
901 {
902         return container_of0(o, struct cmr_object, cmm_obj.cmo_obj.mo_lu);
903 }
904 /** Get cmr_object from md_object. */
905 static inline struct cmr_object *md2cmr_obj(struct md_object *mo)
906 {
907         return container_of0(mo, struct cmr_object, cmm_obj.cmo_obj);
908 }
909 /** Get cmr_object from cmm_object. */
910 static inline struct cmr_object *cmm2cmr_obj(struct cmm_object *co)
911 {
912         return container_of0(co, struct cmr_object, cmm_obj);
913 }
914 /** @} */
915
916 /**
917  * Get proper child device from MDCs.
918  */
919 static struct lu_device *cmr_child_dev(struct cmm_device *d, __u32 num)
920 {
921         struct lu_device *next = NULL;
922         struct mdc_device *mdc;
923
924         cfs_spin_lock(&d->cmm_tgt_guard);
925         cfs_list_for_each_entry(mdc, &d->cmm_targets, mc_linkage) {
926                 if (mdc->mc_num == num) {
927                         next = mdc2lu_dev(mdc);
928                         break;
929                 }
930         }
931         cfs_spin_unlock(&d->cmm_tgt_guard);
932         return next;
933 }
934
935 /**
936  * Free cmr_object.
937  */
938 static void cmr_object_free(const struct lu_env *env,
939                             struct lu_object *lo)
940 {
941         struct cmr_object *cro = lu2cmr_obj(lo);
942         lu_object_fini(lo);
943         OBD_FREE_PTR(cro);
944 }
945
946 /**
947  * Initialize cmr object.
948  */
949 static int cmr_object_init(const struct lu_env *env, struct lu_object *lo,
950                            const struct lu_object_conf *unused)
951 {
952         struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
953         struct lu_device  *c_dev;
954         struct lu_object  *c_obj;
955         int rc;
956
957         ENTRY;
958
959         c_dev = cmr_child_dev(cd, lu2cmr_obj(lo)->cmo_num);
960         if (c_dev == NULL) {
961                 rc = -ENOENT;
962         } else {
963                 c_obj = c_dev->ld_ops->ldo_object_alloc(env,
964                                                         lo->lo_header, c_dev);
965                 if (c_obj != NULL) {
966                         lu_object_add(lo, c_obj);
967                         rc = 0;
968                 } else {
969                         rc = -ENOMEM;
970                 }
971         }
972
973         RETURN(rc);
974 }
975
976 /**
977  * Output lu_object data.
978  */
979 static int cmr_object_print(const struct lu_env *env, void *cookie,
980                             lu_printer_t p, const struct lu_object *lo)
981 {
982         const struct cmr_object *cro = lu2cmr_obj((struct lu_object *)lo);
983         return (*p)(env, cookie, "[remote](mds_num=%d)", cro->cmo_num);
984 }
985
986 /**
987  * Cmr instance of lu_object_operations.
988  */
989 static const struct lu_object_operations cmr_obj_ops = {
990         .loo_object_init    = cmr_object_init,
991         .loo_object_free    = cmr_object_free,
992         .loo_object_print   = cmr_object_print
993 };
994
995 /**
996  * \name cmr remote md_object operations.
997  * All operations here are invalid and return errors. There is no local object
998  * so these operations return two kinds of error:
999  * -# -EFAULT if operation is prohibited.
1000  * -# -EREMOTE if operation can be done just to notify upper level about remote
1001  *  object.
1002  *
1003  * @{
1004  */
1005 static int cmr_object_create(const struct lu_env *env,
1006                              struct md_object *mo,
1007                              const struct md_op_spec *spec,
1008                              struct md_attr *ma)
1009 {
1010         return -EFAULT;
1011 }
1012
1013 static int cmr_permission(const struct lu_env *env,
1014                           struct md_object *p, struct md_object *c,
1015                           struct md_attr *attr, int mask)
1016 {
1017         return -EREMOTE;
1018 }
1019
1020 static int cmr_attr_get(const struct lu_env *env, struct md_object *mo,
1021                         struct md_attr *attr)
1022 {
1023         return -EREMOTE;
1024 }
1025
1026 static int cmr_attr_set(const struct lu_env *env, struct md_object *mo,
1027                         const struct md_attr *attr)
1028 {
1029         return -EFAULT;
1030 }
1031
1032 static int cmr_xattr_get(const struct lu_env *env, struct md_object *mo,
1033                          struct lu_buf *buf, const char *name)
1034 {
1035         return -EFAULT;
1036 }
1037
1038 static int cmr_readlink(const struct lu_env *env, struct md_object *mo,
1039                         struct lu_buf *buf)
1040 {
1041         return -EFAULT;
1042 }
1043
1044 static int cmr_changelog(const struct lu_env *env, enum changelog_rec_type type,
1045                          int flags, struct md_object *mo)
1046 {
1047         return -EFAULT;
1048 }
1049
1050 static int cmr_xattr_list(const struct lu_env *env, struct md_object *mo,
1051                           struct lu_buf *buf)
1052 {
1053         return -EFAULT;
1054 }
1055
1056 static int cmr_xattr_set(const struct lu_env *env, struct md_object *mo,
1057                          const struct lu_buf *buf, const char *name,
1058                          int fl)
1059 {
1060         return -EFAULT;
1061 }
1062
1063 static int cmr_xattr_del(const struct lu_env *env, struct md_object *mo,
1064                          const char *name)
1065 {
1066         return -EFAULT;
1067 }
1068
1069 static int cmr_ref_add(const struct lu_env *env, struct md_object *mo,
1070                        const struct md_attr *ma)
1071 {
1072         return -EFAULT;
1073 }
1074
1075 static int cmr_ref_del(const struct lu_env *env, struct md_object *mo,
1076                        struct md_attr *ma)
1077 {
1078         return -EFAULT;
1079 }
1080
1081 static int cmr_open(const struct lu_env *env, struct md_object *mo,
1082                     int flags)
1083 {
1084         return -EREMOTE;
1085 }
1086
1087 static int cmr_close(const struct lu_env *env, struct md_object *mo,
1088                      struct md_attr *ma, int mode)
1089 {
1090         return -EFAULT;
1091 }
1092
1093 static int cmr_readpage(const struct lu_env *env, struct md_object *mo,
1094                         const struct lu_rdpg *rdpg)
1095 {
1096         return -EREMOTE;
1097 }
1098
1099 static int cmr_capa_get(const struct lu_env *env, struct md_object *mo,
1100                         struct lustre_capa *capa, int renewal)
1101 {
1102         return -EFAULT;
1103 }
1104
1105 static int cmr_path(const struct lu_env *env, struct md_object *obj,
1106                     char *path, int pathlen, __u64 *recno, int *linkno)
1107 {
1108         return -EREMOTE;
1109 }
1110
1111 static int cmr_object_sync(const struct lu_env *env, struct md_object *mo)
1112 {
1113         return -EFAULT;
1114 }
1115
1116 static int cmr_file_lock(const struct lu_env *env, struct md_object *mo,
1117                          struct lov_mds_md *lmm, struct ldlm_extent *extent,
1118                          struct lustre_handle *lockh)
1119 {
1120         return -EREMOTE;
1121 }
1122
1123 static int cmr_file_unlock(const struct lu_env *env, struct md_object *mo,
1124                            struct lov_mds_md *lmm, struct lustre_handle *lockh)
1125 {
1126         return -EREMOTE;
1127 }
1128
1129 /** Set of md_object_operations for cmr. */
1130 static const struct md_object_operations cmr_mo_ops = {
1131         .moo_permission    = cmr_permission,
1132         .moo_attr_get      = cmr_attr_get,
1133         .moo_attr_set      = cmr_attr_set,
1134         .moo_xattr_get     = cmr_xattr_get,
1135         .moo_xattr_set     = cmr_xattr_set,
1136         .moo_xattr_list    = cmr_xattr_list,
1137         .moo_xattr_del     = cmr_xattr_del,
1138         .moo_object_create = cmr_object_create,
1139         .moo_ref_add       = cmr_ref_add,
1140         .moo_ref_del       = cmr_ref_del,
1141         .moo_open          = cmr_open,
1142         .moo_close         = cmr_close,
1143         .moo_readpage      = cmr_readpage,
1144         .moo_readlink      = cmr_readlink,
1145         .moo_changelog     = cmr_changelog,
1146         .moo_capa_get      = cmr_capa_get,
1147         .moo_object_sync   = cmr_object_sync,
1148         .moo_path          = cmr_path,
1149         .moo_file_lock     = cmr_file_lock,
1150         .moo_file_unlock   = cmr_file_unlock,
1151 };
1152 /** @} */
1153
1154 /**
1155  * \name cmr md_dir operations.
1156  *
1157  * All methods below are cross-ref by nature. They consist of remote call and
1158  * local operation. Due to future rollback functionality there are several
1159  * limitations for such methods:
1160  * -# remote call should be done at first to do epoch negotiation between all
1161  * MDS involved and to avoid the RPC inside transaction.
1162  * -# only one RPC can be sent - also due to epoch negotiation.
1163  * For more details see rollback HLD/DLD.
1164  * @{
1165  */
1166 static int cmr_lookup(const struct lu_env *env, struct md_object *mo_p,
1167                       const struct lu_name *lname, struct lu_fid *lf,
1168                       struct md_op_spec *spec)
1169 {
1170         /*
1171          * This can happens while rename() If new parent is remote dir, lookup
1172          * will happen here.
1173          */
1174
1175         return -EREMOTE;
1176 }
1177
1178 /** Return lock mode. */
1179 static mdl_mode_t cmr_lock_mode(const struct lu_env *env,
1180                                 struct md_object *mo, mdl_mode_t lm)
1181 {
1182         return MDL_MINMODE;
1183 }
1184
1185 /**
1186  * Create operation for cmr.
1187  * Remote object creation and local name insert.
1188  *
1189  * \param mo_p Parent directory. Local object.
1190  * \param lchild_name name of file to create.
1191  * \param mo_c Child object. It has no real inode yet.
1192  * \param spec creation specification.
1193  * \param ma child object attributes.
1194  */
1195 static int cmr_create(const struct lu_env *env, struct md_object *mo_p,
1196                       const struct lu_name *lchild_name, struct md_object *mo_c,
1197                       struct md_op_spec *spec,
1198                       struct md_attr *ma)
1199 {
1200         struct cmm_thread_info *cmi;
1201         struct md_attr *tmp_ma;
1202         int rc;
1203         ENTRY;
1204
1205         /* Make sure that name isn't exist before doing remote call. */
1206         rc = mdo_lookup(env, md_object_next(mo_p), lchild_name,
1207                         &cmm_env_info(env)->cmi_fid, NULL);
1208         if (rc == 0)
1209                 RETURN(-EEXIST);
1210         else if (rc != -ENOENT)
1211                 RETURN(rc);
1212
1213         /* check the SGID attr */
1214         cmi = cmm_env_info(env);
1215         LASSERT(cmi);
1216         tmp_ma = &cmi->cmi_ma;
1217         tmp_ma->ma_valid = 0;
1218         tmp_ma->ma_need = MA_INODE;
1219
1220 #ifdef CONFIG_FS_POSIX_ACL
1221         if (!S_ISLNK(ma->ma_attr.la_mode)) {
1222                 tmp_ma->ma_acl = cmi->cmi_xattr_buf;
1223                 tmp_ma->ma_acl_size = sizeof(cmi->cmi_xattr_buf);
1224                 tmp_ma->ma_need |= MA_ACL_DEF;
1225         }
1226 #endif
1227         rc = mo_attr_get(env, md_object_next(mo_p), tmp_ma);
1228         if (rc)
1229                 RETURN(rc);
1230
1231         if (tmp_ma->ma_attr.la_mode & S_ISGID) {
1232                 ma->ma_attr.la_gid = tmp_ma->ma_attr.la_gid;
1233                 if (S_ISDIR(ma->ma_attr.la_mode)) {
1234                         ma->ma_attr.la_mode |= S_ISGID;
1235                         ma->ma_attr.la_valid |= LA_MODE;
1236                 }
1237         }
1238
1239 #ifdef CONFIG_FS_POSIX_ACL
1240         if (tmp_ma->ma_valid & MA_ACL_DEF) {
1241                 spec->u.sp_ea.fid = spec->u.sp_pfid;
1242                 spec->u.sp_ea.eadata = tmp_ma->ma_acl;
1243                 spec->u.sp_ea.eadatalen = tmp_ma->ma_acl_size;
1244                 spec->sp_cr_flags |= MDS_CREATE_RMT_ACL;
1245         }
1246 #endif
1247
1248         /* Local permission check for name_insert before remote ops. */
1249         rc = mo_permission(env, NULL, md_object_next(mo_p), NULL,
1250                            (S_ISDIR(ma->ma_attr.la_mode) ?
1251                            MAY_LINK : MAY_CREATE));
1252         if (rc)
1253                 RETURN(rc);
1254
1255         /**
1256          * \note \a ma will be changed after mo_object_create(), but we will use
1257          * it for mdo_name_insert() later, so save it before mo_object_create().
1258          */
1259         *tmp_ma = *ma;
1260         rc = mo_object_create(env, md_object_next(mo_c), spec, ma);
1261         if (rc == 0) {
1262                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1263                 rc = mdo_name_insert(env, md_object_next(mo_p), lchild_name,
1264                                      lu_object_fid(&mo_c->mo_lu), tmp_ma);
1265                 if (unlikely(rc)) {
1266                         /* TODO: remove object mo_c on remote MDS */
1267                         CWARN("cmr_create failed, should revoke: [mo_p "DFID"]"
1268                               " [name %s] [mo_c "DFID"] [err %d]\n",
1269                               PFID(lu_object_fid(&mo_p->mo_lu)),
1270                               lchild_name->ln_name,
1271                               PFID(lu_object_fid(&mo_c->mo_lu)), rc);
1272                 }
1273         }
1274
1275         RETURN(rc);
1276 }
1277
1278 /**
1279  * Link operations for cmr.
1280  *
1281  * The link RPC is always issued to the server where source parent is living.
1282  * The first operation to do is object nlink increment on remote server.
1283  * Second one is local mdo_name_insert().
1284  *
1285  * \param mo_p parent directory. It is local.
1286  * \param mo_s source object to link. It is remote.
1287  * \param lname Name of link file.
1288  * \param ma object attributes.
1289  */
1290 static int cmr_link(const struct lu_env *env, struct md_object *mo_p,
1291                     struct md_object *mo_s, const struct lu_name *lname,
1292                     struct md_attr *ma)
1293 {
1294         int rc;
1295         ENTRY;
1296
1297         /* Make sure that name isn't exist before doing remote call. */
1298         rc = mdo_lookup(env, md_object_next(mo_p), lname,
1299                         &cmm_env_info(env)->cmi_fid, NULL);
1300         if (rc == 0) {
1301                 rc = -EEXIST;
1302         } else if (rc == -ENOENT) {
1303                 /* Local permission check for name_insert before remote ops. */
1304                 rc = mo_permission(env, NULL, md_object_next(mo_p), NULL,
1305                                    MAY_CREATE);
1306                 if (rc)
1307                         RETURN(rc);
1308
1309                 rc = mo_ref_add(env, md_object_next(mo_s), ma);
1310                 if (rc == 0) {
1311                         ma->ma_attr_flags |= MDS_PERM_BYPASS;
1312                         rc = mdo_name_insert(env, md_object_next(mo_p), lname,
1313                                              lu_object_fid(&mo_s->mo_lu), ma);
1314                         if (unlikely(rc)) {
1315                                 /* TODO: ref_del from mo_s on remote MDS */
1316                                 CWARN("cmr_link failed, should revoke: "
1317                                       "[mo_p "DFID"] [mo_s "DFID"] "
1318                                       "[name %s] [err %d]\n",
1319                                       PFID(lu_object_fid(&mo_p->mo_lu)),
1320                                       PFID(lu_object_fid(&mo_s->mo_lu)),
1321                                       lname->ln_name, rc);
1322                         }
1323                 }
1324         }
1325         RETURN(rc);
1326 }
1327
1328 /**
1329  * Unlink operations for cmr.
1330  *
1331  * The unlink RPC is always issued to the server where parent is living. Hence
1332  * the first operation to do is object unlink on remote server. Second one is
1333  * local mdo_name_remove().
1334  *
1335  * \param mo_p parent md_object. It is local.
1336  * \param mo_c child object to be unlinked. It is remote.
1337  * \param lname Name of file to unlink.
1338  * \param ma object attributes.
1339  */
1340 static int cmr_unlink(const struct lu_env *env, struct md_object *mo_p,
1341                       struct md_object *mo_c, const struct lu_name *lname,
1342                       struct md_attr *ma)
1343 {
1344         struct cmm_thread_info *cmi;
1345         struct md_attr *tmp_ma;
1346         int rc;
1347         ENTRY;
1348
1349         /* Local permission check for name_remove before remote ops. */
1350         rc = mo_permission(env, NULL, md_object_next(mo_p), ma,
1351                            MAY_UNLINK | MAY_VTX_PART);
1352         if (rc)
1353                 RETURN(rc);
1354
1355         /*
1356          * \note \a ma will be changed after mo_ref_del, but we will use
1357          * it for mdo_name_remove() later, so save it before mo_ref_del().
1358          */
1359         cmi = cmm_env_info(env);
1360         tmp_ma = &cmi->cmi_ma;
1361         *tmp_ma = *ma;
1362         rc = mo_ref_del(env, md_object_next(mo_c), ma);
1363         if (rc == 0) {
1364                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1365                 rc = mdo_name_remove(env, md_object_next(mo_p), lname, tmp_ma);
1366                 if (unlikely(rc)) {
1367                         /* TODO: ref_add to mo_c on remote MDS */
1368                         CWARN("cmr_unlink failed, should revoke: [mo_p "DFID"]"
1369                               " [mo_c "DFID"] [name %s] [err %d]\n",
1370                               PFID(lu_object_fid(&mo_p->mo_lu)),
1371                               PFID(lu_object_fid(&mo_c->mo_lu)),
1372                               lname->ln_name, rc);
1373                 }
1374         }
1375
1376         RETURN(rc);
1377 }
1378
1379 /** Helper which outputs error message during cmr_rename() */
1380 static inline void cmr_rename_warn(const char *fname,
1381                                   struct md_object *mo_po,
1382                                   struct md_object *mo_pn,
1383                                   const struct lu_fid *lf,
1384                                   const char *s_name,
1385                                   const char *t_name,
1386                                   int err)
1387 {
1388         CWARN("cmr_rename failed for %s, should revoke: "
1389               "[mo_po "DFID"] [mo_pn "DFID"] [lf "DFID"] "
1390               "[sname %s] [tname %s] [err %d]\n", fname,
1391               PFID(lu_object_fid(&mo_po->mo_lu)),
1392               PFID(lu_object_fid(&mo_pn->mo_lu)),
1393               PFID(lf), s_name, t_name, err);
1394 }
1395
1396 /**
1397  * Rename operation for cmr.
1398  *
1399  * This is the most complex cross-reference operation. It may consist of up to 4
1400  * MDS server and require several RPCs to be sent.
1401  *
1402  * \param mo_po Old parent object.
1403  * \param mo_pn New parent object.
1404  * \param lf FID of object to rename.
1405  * \param ls_name Source file name.
1406  * \param mo_t target object. Should be NULL here.
1407  * \param lt_name Name of target file.
1408  * \param ma object attributes.
1409  */
1410 static int cmr_rename(const struct lu_env *env,
1411                       struct md_object *mo_po, struct md_object *mo_pn,
1412                       const struct lu_fid *lf, const struct lu_name *ls_name,
1413                       struct md_object *mo_t, const struct lu_name *lt_name,
1414                       struct md_attr *ma)
1415 {
1416         struct cmm_thread_info *cmi;
1417         struct md_attr *tmp_ma;
1418         int rc;
1419         ENTRY;
1420
1421         LASSERT(mo_t == NULL);
1422
1423         /* get real type of src */
1424         rc = cmm_mode_get(env, md_obj2dev(mo_po), lf, ma, NULL);
1425         if (rc)
1426                 RETURN(rc);
1427
1428         /* Local permission check for name_remove before remote ops. */
1429         rc = mo_permission(env, NULL, md_object_next(mo_po), ma,
1430                            MAY_UNLINK | MAY_VTX_FULL);
1431         if (rc)
1432                 RETURN(rc);
1433
1434         /**
1435          * \todo \a ma maybe changed after mdo_rename_tgt(), but we will use it
1436          * for mdo_name_remove() later, so save it before mdo_rename_tgt.
1437          */
1438         cmi = cmm_env_info(env);
1439         tmp_ma = &cmi->cmi_ma;
1440         *tmp_ma = *ma;
1441         /**
1442          * \note The \a mo_pn is remote directory, so we cannot even know if there is
1443          * \a mo_t or not. Therefore \a mo_t is NULL here but remote server should do
1444          * lookup and process this further.
1445          */
1446         rc = mdo_rename_tgt(env, md_object_next(mo_pn),
1447                             NULL/* mo_t */, lf, lt_name, ma);
1448         if (rc)
1449                 RETURN(rc);
1450
1451         tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1452
1453         /* src object maybe on remote MDS, do remote ops first. */
1454         rc = cmm_rename_ctime(env, md_obj2dev(mo_po), lf, tmp_ma);
1455         if (unlikely(rc)) {
1456                 /* TODO: revoke mdo_rename_tgt */
1457                 cmr_rename_warn("cmm_rename_ctime", mo_po, mo_pn, lf,
1458                                 ls_name->ln_name, lt_name->ln_name, rc);
1459                 RETURN(rc);
1460         }
1461
1462         /* only old name is removed localy */
1463         rc = mdo_name_remove(env, md_object_next(mo_po), ls_name, tmp_ma);
1464         if (unlikely(rc))
1465                 /* TODO: revoke all cmr_rename */
1466                 cmr_rename_warn("mdo_name_remove", mo_po, mo_pn, lf,
1467                                 ls_name->ln_name, lt_name->ln_name, rc);
1468
1469         RETURN(rc);
1470 }
1471
1472 /**
1473  * Part of cross-ref rename().
1474  * Used to insert new name in new parent and unlink target.
1475  */
1476 static int cmr_rename_tgt(const struct lu_env *env,
1477                           struct md_object *mo_p, struct md_object *mo_t,
1478                           const struct lu_fid *lf, const struct lu_name *lname,
1479                           struct md_attr *ma)
1480 {
1481         struct cmm_thread_info *cmi;
1482         struct md_attr *tmp_ma;
1483         int rc;
1484         ENTRY;
1485
1486         /* target object is remote one */
1487         /* Local permission check for rename_tgt before remote ops. */
1488         rc = mo_permission(env, NULL, md_object_next(mo_p), ma,
1489                            MAY_UNLINK | MAY_VTX_PART);
1490         if (rc)
1491                 RETURN(rc);
1492
1493         /*
1494          * XXX: @ma maybe changed after mo_ref_del, but we will use
1495          * it for mdo_rename_tgt later, so save it before mo_ref_del.
1496          */
1497         cmi = cmm_env_info(env);
1498         tmp_ma = &cmi->cmi_ma;
1499         *tmp_ma = *ma;
1500         rc = mo_ref_del(env, md_object_next(mo_t), ma);
1501         /* continue locally with name handling only */
1502         if (rc == 0) {
1503                 tmp_ma->ma_attr_flags |= MDS_PERM_BYPASS;
1504                 rc = mdo_rename_tgt(env, md_object_next(mo_p),
1505                                     NULL, lf, lname, tmp_ma);
1506                 if (unlikely(rc)) {
1507                         /* TODO: ref_add to mo_t on remote MDS */
1508                         CWARN("cmr_rename_tgt failed, should revoke: "
1509                               "[mo_p "DFID"] [mo_t "DFID"] [lf "DFID"] "
1510                               "[name %s] [err %d]\n",
1511                               PFID(lu_object_fid(&mo_p->mo_lu)),
1512                               PFID(lu_object_fid(&mo_t->mo_lu)),
1513                               PFID(lf),
1514                               lname->ln_name, rc);
1515                 }
1516         }
1517         RETURN(rc);
1518 }
1519 /** @} */
1520 /**
1521  * The md_dir_operations for cmr.
1522  */
1523 static const struct md_dir_operations cmr_dir_ops = {
1524         .mdo_is_subdir   = cmm_is_subdir,
1525         .mdo_lookup      = cmr_lookup,
1526         .mdo_lock_mode   = cmr_lock_mode,
1527         .mdo_create      = cmr_create,
1528         .mdo_link        = cmr_link,
1529         .mdo_unlink      = cmr_unlink,
1530         .mdo_rename      = cmr_rename,
1531         .mdo_rename_tgt  = cmr_rename_tgt
1532 };
1533 /** @} */