1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * lustre/cmm/cmm_object.c
5 * Lustre Cluster Metadata Manager (cmm)
7 * Copyright (c) 2006 Cluster File Systems, Inc.
8 * Author: Mike Pershin <tappro@clusterfs.com>
10 * This file is part of the Lustre file system, http://www.lustre.org
11 * Lustre is a trademark of Cluster File Systems, Inc.
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.
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.
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.
30 # define EXPORT_SYMTAB
33 #define DEBUG_SUBSYSTEM S_MDS
35 #include <lustre_fid.h>
36 #include "cmm_internal.h"
37 #include "mdc_internal.h"
39 /* XXX: fix later this hack. It exists because OSD produces fids with like this:
40 seq = ROOT_SEQ + 1, etc. */
41 static int cmm_special_fid(const struct lu_fid *fid)
43 struct lu_range *space = (struct lu_range *)&LUSTRE_SEQ_SPACE_RANGE;
44 return !range_within(space, fid_seq(fid));
47 static int cmm_fld_lookup(struct cmm_device *cm,
48 const struct lu_fid *fid)
54 LASSERT(fid_is_sane(fid));
56 /* XXX: is this correct? We need this to prevent FLD lookups while CMM
57 * did not initialized yet all MDCs. */
58 if (cmm_special_fid(fid))
61 rc = fld_client_lookup(&cm->cmm_fld, fid_seq(fid), &mds);
63 CERROR("can't find mds by seq "LPU64", rc %d\n",
68 CWARN("CMM: got MDS "LPU64" for sequence: "LPU64"\n",
74 static struct md_object_operations cml_mo_ops;
75 static struct md_dir_operations cml_dir_ops;
76 static struct lu_object_operations cml_obj_ops;
78 static struct md_object_operations cmr_mo_ops;
79 static struct md_dir_operations cmr_dir_ops;
80 static struct lu_object_operations cmr_obj_ops;
82 struct lu_object *cmm_object_alloc(const struct lu_context *ctx,
83 const struct lu_object_header *loh,
86 struct lu_object *lo = NULL;
87 const struct lu_fid *fid = &loh->loh_fid;
88 struct cmm_device *cd;
93 if (cd->cmm_flags & CMM_INITIALIZED) {
94 /* get object location */
95 mdsnum = cmm_fld_lookup(lu2cmm_dev(ld), fid);
97 RETURN(ERR_PTR(mdsnum));
100 * Device is not yet initialized, cmm_object is being created
101 * as part of early bootstrap procedure (it is /ROOT, or /fld,
102 * etc.). Such object *has* to be local.
104 mdsnum = cd->cmm_local_num;
106 /* select the proper set of operations based on object location */
107 if (mdsnum == cd->cmm_local_num) {
108 struct cml_object *clo;
112 lo = &clo->cmm_obj.cmo_obj.mo_lu;
113 lu_object_init(lo, NULL, ld);
114 clo->cmm_obj.cmo_obj.mo_ops = &cml_mo_ops;
115 clo->cmm_obj.cmo_obj.mo_dir_ops = &cml_dir_ops;
116 lo->lo_ops = &cml_obj_ops;
117 clo->cmm_obj.cmo_local = 1;
120 struct cmr_object *cro;
124 lo = &cro->cmm_obj.cmo_obj.mo_lu;
125 lu_object_init(lo, NULL, ld);
126 cro->cmm_obj.cmo_obj.mo_ops = &cmr_mo_ops;
127 cro->cmm_obj.cmo_obj.mo_dir_ops = &cmr_dir_ops;
128 lo->lo_ops = &cmr_obj_ops;
129 cro->cmo_num = mdsnum;
130 cro->cmm_obj.cmo_local = 0;
137 * CMM has two types of objects - local and remote. They have different set
138 * of operations so we are avoiding multiple checks in code.
142 * local CMM object operations. cml_...
144 static inline struct cml_object *lu2cml_obj(struct lu_object *o)
146 return container_of0(o, struct cml_object, cmm_obj.cmo_obj.mo_lu);
148 static inline struct cml_object *md2cml_obj(struct md_object *mo)
150 return container_of0(mo, struct cml_object, cmm_obj.cmo_obj);
152 static inline struct cml_object *cmm2cml_obj(struct cmm_object *co)
154 return container_of0(co, struct cml_object, cmm_obj);
156 /* get local child device */
157 static struct lu_device *cml_child_dev(struct cmm_device *d)
159 return &d->cmm_child->md_lu_dev;
162 /* lu_object operations */
163 static void cml_object_free(const struct lu_context *ctx,
164 struct lu_object *lo)
166 struct cml_object *clo = lu2cml_obj(lo);
171 static int cml_object_init(const struct lu_context *ctx, struct lu_object *lo)
173 struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
174 struct lu_device *c_dev;
175 struct lu_object *c_obj;
180 c_dev = cml_child_dev(cd);
184 c_obj = c_dev->ld_ops->ldo_object_alloc(ctx,
185 lo->lo_header, c_dev);
187 lu_object_add(lo, c_obj);
197 static int cml_object_exists(const struct lu_context *ctx,
198 struct lu_object *lo)
200 return lu_object_exists(ctx, lu_object_next(lo));
203 static int cml_object_print(const struct lu_context *ctx,
204 struct seq_file *f, const struct lu_object *lo)
206 return seq_printf(f, LUSTRE_CMM0_NAME"-local@%p", lo);
209 static struct lu_object_operations cml_obj_ops = {
210 .loo_object_init = cml_object_init,
211 .loo_object_free = cml_object_free,
212 .loo_object_print = cml_object_print,
213 .loo_object_exists = cml_object_exists
216 /* CMM local md_object operations */
217 static int cml_object_create(const struct lu_context *ctx,
218 struct md_object *mo,
219 struct lu_attr *attr)
223 rc = mo_object_create(ctx, cmm2child_obj(md2cmm_obj(mo)), attr);
227 static int cml_attr_get(const struct lu_context *ctx, struct md_object *mo,
228 struct lu_attr *attr)
232 rc = mo_attr_get(ctx, cmm2child_obj(md2cmm_obj(mo)), attr);
236 static int cml_attr_set(const struct lu_context *ctx, struct md_object *mo,
237 struct lu_attr *attr)
241 rc = mo_attr_set(ctx, cmm2child_obj(md2cmm_obj(mo)), attr);
245 static int cml_xattr_get(const struct lu_context *ctx, struct md_object *mo,
246 void *buf, int buflen, const char *name)
250 rc = mo_xattr_get(ctx, cmm2child_obj(md2cmm_obj(mo)),
255 static int cml_xattr_set(const struct lu_context *ctx, struct md_object *mo,
256 void *buf, int buflen, const char *name)
260 rc = mo_xattr_set(ctx, cmm2child_obj(md2cmm_obj(mo)),
265 static int cml_ref_add(const struct lu_context *ctx, struct md_object *mo)
269 rc = mo_ref_add(ctx, cmm2child_obj(md2cmm_obj(mo)));
273 static int cml_ref_del(const struct lu_context *ctx, struct md_object *mo)
277 rc = mo_ref_del(ctx, cmm2child_obj(md2cmm_obj(mo)));
281 static int cml_open(const struct lu_context *ctx, struct md_object *mo)
285 rc = mo_open(ctx, cmm2child_obj(md2cmm_obj(mo)));
289 static int cml_close(const struct lu_context *ctx, struct md_object *mo)
293 rc = mo_close(ctx, cmm2child_obj(md2cmm_obj(mo)));
297 static struct md_object_operations cml_mo_ops = {
298 .moo_attr_get = cml_attr_get,
299 .moo_attr_set = cml_attr_set,
300 .moo_xattr_get = cml_xattr_get,
301 .moo_xattr_set = cml_xattr_set,
302 .moo_object_create = cml_object_create,
303 .moo_ref_add = cml_ref_add,
304 .moo_ref_del = cml_ref_del,
305 .moo_open = cml_open,
306 .moo_close = cml_close
309 /* md_dir operations */
310 static int cml_lookup(const struct lu_context *ctx, struct md_object *mo_p,
311 const char *name, struct lu_fid *lf)
315 rc = mdo_lookup(ctx, cmm2child_obj(md2cmm_obj(mo_p)), name, lf);
320 static int cml_create(const struct lu_context *ctx,
321 struct md_object *mo_p, const char *name,
322 struct md_object *mo_c, struct lu_attr *attr)
326 rc = mdo_create(ctx, cmm2child_obj(md2cmm_obj(mo_p)), name,
327 cmm2child_obj(md2cmm_obj(mo_c)), attr);
331 static int cml_link(const struct lu_context *ctx, struct md_object *mo_p,
332 struct md_object *mo_s, const char *name)
336 rc = mdo_link(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
337 cmm2child_obj(md2cmm_obj(mo_s)), name);
341 static int cml_unlink(const struct lu_context *ctx, struct md_object *mo_p,
342 struct md_object *mo_c, const char *name)
346 rc = mdo_unlink(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
347 cmm2child_obj(md2cmm_obj(mo_c)), name);
351 static int cml_rename(const struct lu_context *ctx, struct md_object *mo_po,
352 struct md_object *mo_pn, const struct lu_fid *lf,
353 const char *s_name, struct md_object *mo_t,
359 if (mo_t && !cmm_is_local_obj(md2cmm_obj(mo_t))) {
361 rc = mo_ref_del(ctx, cmm2child_obj(md2cmm_obj(mo_t)));
367 rc = mdo_rename(ctx, cmm2child_obj(md2cmm_obj(mo_po)),
368 cmm2child_obj(md2cmm_obj(mo_pn)), lf, s_name,
369 cmm2child_obj(md2cmm_obj(mo_t)), t_name);
374 static int cml_rename_tgt(const struct lu_context *ctx,
375 struct md_object *mo_p, struct md_object *mo_t,
376 const struct lu_fid *lf, const char *name)
381 rc = mdo_rename_tgt(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
382 cmm2child_obj(md2cmm_obj(mo_t)), lf, name);
386 static struct md_dir_operations cml_dir_ops = {
387 .mdo_lookup = cml_lookup,
388 .mdo_create = cml_create,
389 .mdo_link = cml_link,
390 .mdo_unlink = cml_unlink,
391 .mdo_rename = cml_rename,
392 .mdo_rename_tgt = cml_rename_tgt,
395 /* -------------------------------------------------------------------
396 * remote CMM object operations. cmr_...
398 static inline struct cmr_object *lu2cmr_obj(struct lu_object *o)
400 return container_of0(o, struct cmr_object, cmm_obj.cmo_obj.mo_lu);
402 static inline struct cmr_object *md2cmr_obj(struct md_object *mo)
404 return container_of0(mo, struct cmr_object, cmm_obj.cmo_obj);
406 static inline struct cmr_object *cmm2cmr_obj(struct cmm_object *co)
408 return container_of0(co, struct cmr_object, cmm_obj);
411 /* get proper child device from MDCs */
412 static struct lu_device *cmr_child_dev(struct cmm_device *d, __u32 num)
414 struct lu_device *next = NULL;
415 struct mdc_device *mdc;
417 spin_lock(&d->cmm_tgt_guard);
418 list_for_each_entry(mdc, &d->cmm_targets, mc_linkage) {
419 if (mdc->mc_num == num) {
420 next = mdc2lu_dev(mdc);
424 spin_unlock(&d->cmm_tgt_guard);
428 /* lu_object operations */
429 static void cmr_object_free(const struct lu_context *ctx,
430 struct lu_object *lo)
432 struct cmr_object *cro = lu2cmr_obj(lo);
437 static int cmr_object_init(const struct lu_context *ctx, struct lu_object *lo)
439 struct cmm_device *cd = lu2cmm_dev(lo->lo_dev);
440 struct lu_device *c_dev;
441 struct lu_object *c_obj;
446 c_dev = cmr_child_dev(cd, lu2cmr_obj(lo)->cmo_num);
450 c_obj = c_dev->ld_ops->ldo_object_alloc(ctx,
451 lo->lo_header, c_dev);
453 lu_object_add(lo, c_obj);
463 static int cmr_object_exists(const struct lu_context *ctx,
464 struct lu_object *lo)
466 return lu_object_exists(ctx, lu_object_next(lo));
469 static int cmr_object_print(const struct lu_context *ctx,
470 struct seq_file *f, const struct lu_object *lo)
472 return seq_printf(f, LUSTRE_CMM0_NAME"-remote@%p", lo);
475 static struct lu_object_operations cmr_obj_ops = {
476 .loo_object_init = cmr_object_init,
477 .loo_object_free = cmr_object_free,
478 .loo_object_print = cmr_object_print,
479 .loo_object_exists = cmr_object_exists
482 /* CMM remote md_object operations. All are invalid */
483 static int cmr_object_create(const struct lu_context *ctx,
484 struct md_object *mo,
485 struct lu_attr *attr)
490 static int cmr_attr_get(const struct lu_context *ctx, struct md_object *mo,
491 struct lu_attr *attr)
496 static int cmr_attr_set(const struct lu_context *ctx, struct md_object *mo,
497 struct lu_attr *attr)
502 static int cmr_xattr_get(const struct lu_context *ctx, struct md_object *mo,
503 void *buf, int buflen, const char *name)
508 static int cmr_xattr_set(const struct lu_context *ctx, struct md_object *mo,
509 void *buf, int buflen, const char *name)
514 static int cmr_ref_add(const struct lu_context *ctx, struct md_object *mo)
519 static int cmr_ref_del(const struct lu_context *ctx, struct md_object *mo)
524 static int cmr_open(const struct lu_context *ctx, struct md_object *mo)
529 static int cmr_close(const struct lu_context *ctx, struct md_object *mo)
534 static struct md_object_operations cmr_mo_ops = {
535 .moo_attr_get = cmr_attr_get,
536 .moo_attr_set = cmr_attr_set,
537 .moo_xattr_get = cmr_xattr_get,
538 .moo_xattr_set = cmr_xattr_set,
539 .moo_object_create = cmr_object_create,
540 .moo_ref_add = cmr_ref_add,
541 .moo_ref_del = cmr_ref_del,
542 .moo_open = cmr_open,
543 .moo_close = cmr_close
546 /* remote part of md_dir operations */
547 static int cmr_lookup(const struct lu_context *ctx, struct md_object *mo_p,
548 const char *name, struct lu_fid *lf)
550 /*this can happens while rename()
551 * If new parent is remote dir, lookup will happens here */
556 static int cmr_create(const struct lu_context *ctx,
557 struct md_object *mo_p, const char *name,
558 struct md_object *mo_c, struct lu_attr *attr)
564 //XXX: make sure that MDT checks name isn't exist
566 /* remote object creation and local name insert */
567 rc = mo_object_create(ctx, cmm2child_obj(md2cmm_obj(mo_c)), attr);
569 rc = mdo_name_insert(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
570 name, lu_object_fid(&mo_c->mo_lu));
576 static int cmr_link(const struct lu_context *ctx, struct md_object *mo_p,
577 struct md_object *mo_s, const char *name)
582 //XXX: make sure that MDT checks name isn't exist
584 rc = mo_ref_add(ctx, cmm2child_obj(md2cmm_obj(mo_s)));
586 rc = mdo_name_insert(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
587 name, lu_object_fid(&mo_s->mo_lu));
593 static int cmr_unlink(const struct lu_context *ctx, struct md_object *mo_p,
594 struct md_object *mo_c, const char *name)
599 rc = mo_ref_del(ctx, cmm2child_obj(md2cmm_obj(mo_c)));
601 rc = mdo_name_remove(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
608 static int cmr_rename(const struct lu_context *ctx, struct md_object *mo_po,
609 struct md_object *mo_pn, const struct lu_fid *lf,
610 const char *s_name, struct md_object *mo_t,
616 /* the mo_pn is remote directory, so we cannot even know if there is
617 * mo_t or not. Therefore mo_t is NULL here but remote server should do
618 * lookup and process this further */
620 LASSERT(mo_t == NULL);
621 rc = mdo_rename_tgt(ctx, cmm2child_obj(md2cmm_obj(mo_pn)),
622 NULL/* mo_t */, lf, t_name);
623 /* only old name is removed localy */
625 rc = mdo_name_remove(ctx, cmm2child_obj(md2cmm_obj(mo_po)),
631 static int cmr_rename_tgt(const struct lu_context *ctx,
632 struct md_object *mo_p, struct md_object *mo_t,
633 const struct lu_fid *lf, const char *name)
637 /* target object is remote one */
638 rc = mo_ref_del(ctx, cmm2child_obj(md2cmm_obj(mo_t)));
639 /* continue locally with name handling only */
641 rc = mdo_rename_tgt(ctx, cmm2child_obj(md2cmm_obj(mo_p)),
646 static struct md_dir_operations cmr_dir_ops = {
647 .mdo_lookup = cmr_lookup,
648 .mdo_create = cmr_create,
649 .mdo_link = cmr_link,
650 .mdo_unlink = cmr_unlink,
651 .mdo_rename = cmr_rename,
652 .mdo_rename_tgt = cmr_rename_tgt,