1 /* -*- MODE: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * lustre/fld/fld_handler.c
5 * FLD (Fids Location Database)
7 * Copyright (C) 2006 Cluster File Systems, Inc.
8 * Author: Yury Umanets <umka@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.
29 # define EXPORT_SYMTAB
31 #define DEBUG_SUBSYSTEM S_FLD
34 # include <libcfs/libcfs.h>
35 # include <linux/module.h>
36 # include <linux/jbd.h>
37 # include <asm/div64.h>
38 #else /* __KERNEL__ */
39 # include <liblustre.h>
40 # include <libcfs/list.h>
44 #include <obd_class.h>
45 #include <lustre_ver.h>
46 #include <obd_support.h>
47 #include <lprocfs_status.h>
49 #include <dt_object.h>
50 #include <md_object.h>
51 #include <lustre_req_layout.h>
52 #include <lustre_fld.h>
53 #include "fld_internal.h"
56 fld_rrb_hash(struct lu_client_fld *fld, seqno_t seq)
58 if (fld->fld_count == 0)
61 return do_div(seq, fld->fld_count);
65 fld_dht_hash(struct lu_client_fld *fld, seqno_t seq)
67 /* XXX: here should DHT hash */
68 return fld_rrb_hash(fld, seq);
71 struct lu_fld_hash fld_hash[3] = {
74 .fh_func = fld_dht_hash
77 .fh_name = "Round Robin",
78 .fh_func = fld_rrb_hash
85 /* this function makes decision if passed @target appropriate acoordingly to
86 * passed @hash. In case of usual round-robin hash, this is decided by comparing
87 * hash and target's index. In the case of DHT, algorithm is a bit more
90 fld_client_apt_target(struct fld_target *target, int hash)
92 /* XXX: DHT case should be worked out. */
93 return (target->fldt_idx == hash);
96 static struct fld_target *
97 fld_client_get_target(struct lu_client_fld *fld, seqno_t seq)
99 struct fld_target *target;
103 LASSERT(fld->fld_hash != NULL);
105 spin_lock(&fld->fld_lock);
106 hash = fld->fld_hash->fh_func(fld, seq);
108 list_for_each_entry(target,
109 &fld->fld_targets, fldt_chain) {
110 if (fld_client_apt_target(target, hash)) {
111 spin_unlock(&fld->fld_lock);
115 spin_unlock(&fld->fld_lock);
117 /* if target is not found, there is logical error anyway, so here is
118 * LBUG() to catch that situation. */
123 /* add export to FLD. This is usually done by CMM and LMV as they are main users
126 fld_client_add_target(struct lu_client_fld *fld,
127 struct obd_export *exp)
129 struct client_obd *cli = &exp->exp_obd->u.cli;
130 struct fld_target *target, *tmp;
133 LASSERT(exp != NULL);
135 CDEBUG(D_INFO|D_WARNING, "FLD(cli): adding export %s\n",
136 cli->cl_target_uuid.uuid);
138 OBD_ALLOC_PTR(target);
142 spin_lock(&fld->fld_lock);
143 list_for_each_entry(tmp, &fld->fld_targets, fldt_chain) {
144 if (obd_uuid_equals(&tmp->fldt_exp->exp_client_uuid,
145 &exp->exp_client_uuid))
147 spin_unlock(&fld->fld_lock);
148 OBD_FREE_PTR(target);
153 target->fldt_exp = class_export_get(exp);
154 target->fldt_idx = fld->fld_count;
156 list_add_tail(&target->fldt_chain,
159 spin_unlock(&fld->fld_lock);
163 EXPORT_SYMBOL(fld_client_add_target);
165 /* remove export from FLD */
167 fld_client_del_target(struct lu_client_fld *fld,
168 struct obd_export *exp)
170 struct fld_target *target, *tmp;
173 spin_lock(&fld->fld_lock);
174 list_for_each_entry_safe(target, tmp,
175 &fld->fld_targets, fldt_chain) {
176 if (obd_uuid_equals(&target->fldt_exp->exp_client_uuid,
177 &exp->exp_client_uuid))
180 list_del(&target->fldt_chain);
181 spin_unlock(&fld->fld_lock);
182 class_export_put(target->fldt_exp);
183 OBD_FREE_PTR(target);
187 spin_unlock(&fld->fld_lock);
190 EXPORT_SYMBOL(fld_client_del_target);
194 fld_client_proc_init(struct lu_client_fld *fld)
199 fld->fld_proc_dir = lprocfs_register(fld->fld_name,
203 if (IS_ERR(fld->fld_proc_dir)) {
204 CERROR("LProcFS failed in fld-init\n");
205 rc = PTR_ERR(fld->fld_proc_dir);
209 rc = lprocfs_add_vars(fld->fld_proc_dir,
210 fld_client_proc_list, fld);
212 CERROR("can't init FLD "
213 "proc, rc %d\n", rc);
220 fld->fld_proc_dir = NULL;
225 fld_client_proc_fini(struct lu_client_fld *fld)
228 if (fld->fld_proc_dir) {
229 lprocfs_remove(fld->fld_proc_dir);
230 fld->fld_proc_dir = NULL;
236 static inline int hash_is_sane(int hash)
238 return (hash >= 0 && hash < ARRAY_SIZE(fld_hash));
242 fld_client_init(struct lu_client_fld *fld,
243 const char *uuid, int hash)
248 LASSERT(fld != NULL);
250 if (!hash_is_sane(hash)) {
251 CERROR("wrong hash function 0x%x\n", hash);
255 INIT_LIST_HEAD(&fld->fld_targets);
256 spin_lock_init(&fld->fld_lock);
257 fld->fld_hash = &fld_hash[hash];
260 snprintf(fld->fld_name, sizeof(fld->fld_name),
261 "%s-%s", LUSTRE_FLD_NAME, uuid);
264 fld->fld_cache = fld_cache_init(FLD_HTABLE_SIZE);
265 if (IS_ERR(fld->fld_cache)) {
266 rc = PTR_ERR(fld->fld_cache);
267 fld->fld_cache = NULL;
273 rc = fld_client_proc_init(fld);
280 fld_client_fini(fld);
282 CDEBUG(D_INFO|D_WARNING,
283 "Client FLD, using \"%s\" hash\n",
284 fld->fld_hash->fh_name);
287 EXPORT_SYMBOL(fld_client_init);
290 fld_client_fini(struct lu_client_fld *fld)
292 struct fld_target *target, *tmp;
296 fld_client_proc_fini(fld);
299 spin_lock(&fld->fld_lock);
300 list_for_each_entry_safe(target, tmp,
301 &fld->fld_targets, fldt_chain) {
303 list_del(&target->fldt_chain);
304 class_export_put(target->fldt_exp);
305 OBD_FREE_PTR(target);
307 spin_unlock(&fld->fld_lock);
310 if (fld->fld_cache != NULL) {
311 fld_cache_fini(fld->fld_cache);
312 fld->fld_cache = NULL;
316 CDEBUG(D_INFO|D_WARNING, "Client FLD finalized\n");
319 EXPORT_SYMBOL(fld_client_fini);
322 fld_client_rpc(struct obd_export *exp,
323 struct md_fld *mf, __u32 fld_op)
325 int size[2] = {sizeof(__u32), sizeof(struct md_fld)}, rc;
326 int mf_size = sizeof(struct md_fld);
327 struct ptlrpc_request *req;
332 LASSERT(exp != NULL);
334 req = ptlrpc_prep_req(class_exp2cliimp(exp),
335 LUSTRE_MDS_VERSION, FLD_QUERY,
340 op = lustre_msg_buf(req->rq_reqmsg, 0, sizeof (*op));
343 pmf = lustre_msg_buf(req->rq_reqmsg, 1, sizeof (*pmf));
346 req->rq_replen = lustre_msg_size(1, &mf_size);
347 req->rq_request_portal = FLD_REQUEST_PORTAL;
349 rc = ptlrpc_queue_wait(req);
353 pmf = lustre_swab_repbuf(req, 0, sizeof(*pmf),
357 ptlrpc_req_finished(req);
362 __fld_client_create(struct lu_client_fld *fld,
363 seqno_t seq, mdsno_t mds,
364 struct md_fld *md_fld)
366 struct fld_target *target;
370 target = fld_client_get_target(fld, seq);
374 rc = fld_client_rpc(target->fldt_exp, md_fld, FLD_CREATE);
377 /* do not return result of calling fld_cache_insert()
378 * here. First of all because it may return -EEXISTS. Another
379 * reason is that, we do not want to stop proceeding because of
380 * cache errors. --umka */
381 fld_cache_insert(fld->fld_cache, seq, mds);
388 fld_client_create(struct lu_client_fld *fld,
389 seqno_t seq, mdsno_t mds)
391 struct md_fld md_fld = { .mf_seq = seq, .mf_mds = mds };
395 rc = __fld_client_create(fld, seq, mds, &md_fld);
398 EXPORT_SYMBOL(fld_client_create);
401 __fld_client_delete(struct lu_client_fld *fld,
402 seqno_t seq, struct md_fld *md_fld)
404 struct fld_target *target;
407 fld_cache_delete(fld->fld_cache, seq);
409 target = fld_client_get_target(fld, seq);
413 rc = fld_client_rpc(target->fldt_exp,
419 fld_client_delete(struct lu_client_fld *fld,
422 struct md_fld md_fld = { .mf_seq = seq, .mf_mds = 0 };
425 rc = __fld_client_delete(fld, seq, &md_fld);
428 EXPORT_SYMBOL(fld_client_delete);
431 __fld_client_lookup(struct lu_client_fld *fld,
432 seqno_t seq, mdsno_t *mds,
433 struct md_fld *md_fld)
435 struct fld_target *target;
439 /* lookup it in the cache */
440 rc = fld_cache_lookup(fld->fld_cache, seq, mds);
444 /* can not find it in the cache */
445 target = fld_client_get_target(fld, seq);
449 rc = fld_client_rpc(target->fldt_exp,
452 *mds = md_fld->mf_mds;
454 /* do not return error here as well. See previous comment in same
455 * situation in function fld_client_create(). --umka */
456 fld_cache_insert(fld->fld_cache, seq, *mds);
462 fld_client_lookup(struct lu_client_fld *fld,
463 seqno_t seq, mdsno_t *mds)
465 struct md_fld md_fld = { .mf_seq = seq, .mf_mds = 0 };
469 rc = __fld_client_lookup(fld, seq, mds, &md_fld);
472 EXPORT_SYMBOL(fld_client_lookup);