fld_request.c revision 114acca8ef16f21c5d50f16d154d05ffddb20049
1/* 2 * GPL HEADER START 3 * 4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. 5 * 6 * This program is free software; you can redistribute it and/or modify 7 * it under the terms of the GNU General Public License version 2 only, 8 * as published by the Free Software Foundation. 9 * 10 * This program is distributed in the hope that it will be useful, but 11 * WITHOUT ANY WARRANTY; without even the implied warranty of 12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 13 * General Public License version 2 for more details (a copy is included 14 * in the LICENSE file that accompanied this code). 15 * 16 * You should have received a copy of the GNU General Public License 17 * version 2 along with this program; If not, see 18 * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf 19 * 20 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara, 21 * CA 95054 USA or visit www.sun.com if you need additional information or 22 * have any questions. 23 * 24 * GPL HEADER END 25 */ 26/* 27 * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved. 28 * Use is subject to license terms. 29 * 30 * Copyright (c) 2011, 2013, Intel Corporation. 31 */ 32/* 33 * This file is part of Lustre, http://www.lustre.org/ 34 * Lustre is a trademark of Sun Microsystems, Inc. 35 * 36 * lustre/fld/fld_request.c 37 * 38 * FLD (Fids Location Database) 39 * 40 * Author: Yury Umanets <umka@clusterfs.com> 41 */ 42 43#define DEBUG_SUBSYSTEM S_FLD 44 45#include "../../include/linux/libcfs/libcfs.h" 46#include <linux/module.h> 47#include <asm/div64.h> 48 49#include "../include/obd.h" 50#include "../include/obd_class.h" 51#include "../include/lustre_ver.h" 52#include "../include/obd_support.h" 53#include "../include/lprocfs_status.h" 54 55#include "../include/dt_object.h" 56#include "../include/md_object.h" 57#include "../include/lustre_req_layout.h" 58#include "../include/lustre_fld.h" 59#include "../include/lustre_mdc.h" 60#include "fld_internal.h" 61 62/* TODO: these 3 functions are copies of flow-control code from mdc_lib.c 63 * It should be common thing. The same about mdc RPC lock */ 64static int fld_req_avail(struct client_obd *cli, struct mdc_cache_waiter *mcw) 65{ 66 int rc; 67 68 client_obd_list_lock(&cli->cl_loi_list_lock); 69 rc = list_empty(&mcw->mcw_entry); 70 client_obd_list_unlock(&cli->cl_loi_list_lock); 71 return rc; 72}; 73 74static void fld_enter_request(struct client_obd *cli) 75{ 76 struct mdc_cache_waiter mcw; 77 struct l_wait_info lwi = { 0 }; 78 79 client_obd_list_lock(&cli->cl_loi_list_lock); 80 if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) { 81 list_add_tail(&mcw.mcw_entry, &cli->cl_cache_waiters); 82 init_waitqueue_head(&mcw.mcw_waitq); 83 client_obd_list_unlock(&cli->cl_loi_list_lock); 84 l_wait_event(mcw.mcw_waitq, fld_req_avail(cli, &mcw), &lwi); 85 } else { 86 cli->cl_r_in_flight++; 87 client_obd_list_unlock(&cli->cl_loi_list_lock); 88 } 89} 90 91static void fld_exit_request(struct client_obd *cli) 92{ 93 struct list_head *l, *tmp; 94 struct mdc_cache_waiter *mcw; 95 96 client_obd_list_lock(&cli->cl_loi_list_lock); 97 cli->cl_r_in_flight--; 98 list_for_each_safe(l, tmp, &cli->cl_cache_waiters) { 99 100 if (cli->cl_r_in_flight >= cli->cl_max_rpcs_in_flight) { 101 /* No free request slots anymore */ 102 break; 103 } 104 105 mcw = list_entry(l, struct mdc_cache_waiter, mcw_entry); 106 list_del_init(&mcw->mcw_entry); 107 cli->cl_r_in_flight++; 108 wake_up(&mcw->mcw_waitq); 109 } 110 client_obd_list_unlock(&cli->cl_loi_list_lock); 111} 112 113static int fld_rrb_hash(struct lu_client_fld *fld, u64 seq) 114{ 115 LASSERT(fld->lcf_count > 0); 116 return do_div(seq, fld->lcf_count); 117} 118 119static struct lu_fld_target * 120fld_rrb_scan(struct lu_client_fld *fld, u64 seq) 121{ 122 struct lu_fld_target *target; 123 int hash; 124 125 /* Because almost all of special sequence located in MDT0, 126 * it should go to index 0 directly, instead of calculating 127 * hash again, and also if other MDTs is not being connected, 128 * the fld lookup requests(for seq on MDT0) should not be 129 * blocked because of other MDTs */ 130 if (fid_seq_is_norm(seq)) 131 hash = fld_rrb_hash(fld, seq); 132 else 133 hash = 0; 134 135 list_for_each_entry(target, &fld->lcf_targets, ft_chain) { 136 if (target->ft_idx == hash) 137 return target; 138 } 139 140 CERROR("%s: Can't find target by hash %d (seq %#llx). Targets (%d):\n", 141 fld->lcf_name, hash, seq, fld->lcf_count); 142 143 list_for_each_entry(target, &fld->lcf_targets, ft_chain) { 144 const char *srv_name = target->ft_srv != NULL ? 145 target->ft_srv->lsf_name : "<null>"; 146 const char *exp_name = target->ft_exp != NULL ? 147 (char *)target->ft_exp->exp_obd->obd_uuid.uuid : 148 "<null>"; 149 150 CERROR(" exp: 0x%p (%s), srv: 0x%p (%s), idx: %llu\n", 151 target->ft_exp, exp_name, target->ft_srv, 152 srv_name, target->ft_idx); 153 } 154 155 /* 156 * If target is not found, there is logical error anyway, so here is 157 * LBUG() to catch this situation. 158 */ 159 LBUG(); 160 return NULL; 161} 162 163struct lu_fld_hash fld_hash[] = { 164 { 165 .fh_name = "RRB", 166 .fh_hash_func = fld_rrb_hash, 167 .fh_scan_func = fld_rrb_scan 168 }, 169 { 170 NULL, 171 } 172}; 173 174static struct lu_fld_target * 175fld_client_get_target(struct lu_client_fld *fld, u64 seq) 176{ 177 struct lu_fld_target *target; 178 179 LASSERT(fld->lcf_hash != NULL); 180 181 spin_lock(&fld->lcf_lock); 182 target = fld->lcf_hash->fh_scan_func(fld, seq); 183 spin_unlock(&fld->lcf_lock); 184 185 if (target != NULL) { 186 CDEBUG(D_INFO, "%s: Found target (idx %llu) by seq %#llx\n", 187 fld->lcf_name, target->ft_idx, seq); 188 } 189 190 return target; 191} 192 193/* 194 * Add export to FLD. This is usually done by CMM and LMV as they are main users 195 * of FLD module. 196 */ 197int fld_client_add_target(struct lu_client_fld *fld, 198 struct lu_fld_target *tar) 199{ 200 const char *name; 201 struct lu_fld_target *target, *tmp; 202 203 LASSERT(tar != NULL); 204 name = fld_target_name(tar); 205 LASSERT(name != NULL); 206 LASSERT(tar->ft_srv != NULL || tar->ft_exp != NULL); 207 208 if (fld->lcf_flags != LUSTRE_FLD_INIT) { 209 CERROR("%s: Attempt to add target %s (idx %llu) on fly - skip it\n", 210 fld->lcf_name, name, tar->ft_idx); 211 return 0; 212 } else { 213 CDEBUG(D_INFO, "%s: Adding target %s (idx %llu)\n", 214 fld->lcf_name, name, tar->ft_idx); 215 } 216 217 OBD_ALLOC_PTR(target); 218 if (target == NULL) 219 return -ENOMEM; 220 221 spin_lock(&fld->lcf_lock); 222 list_for_each_entry(tmp, &fld->lcf_targets, ft_chain) { 223 if (tmp->ft_idx == tar->ft_idx) { 224 spin_unlock(&fld->lcf_lock); 225 OBD_FREE_PTR(target); 226 CERROR("Target %s exists in FLD and known as %s:#%llu\n", 227 name, fld_target_name(tmp), tmp->ft_idx); 228 return -EEXIST; 229 } 230 } 231 232 target->ft_exp = tar->ft_exp; 233 if (target->ft_exp != NULL) 234 class_export_get(target->ft_exp); 235 target->ft_srv = tar->ft_srv; 236 target->ft_idx = tar->ft_idx; 237 238 list_add_tail(&target->ft_chain, 239 &fld->lcf_targets); 240 241 fld->lcf_count++; 242 spin_unlock(&fld->lcf_lock); 243 244 return 0; 245} 246EXPORT_SYMBOL(fld_client_add_target); 247 248/* Remove export from FLD */ 249int fld_client_del_target(struct lu_client_fld *fld, __u64 idx) 250{ 251 struct lu_fld_target *target, *tmp; 252 253 spin_lock(&fld->lcf_lock); 254 list_for_each_entry_safe(target, tmp, 255 &fld->lcf_targets, ft_chain) { 256 if (target->ft_idx == idx) { 257 fld->lcf_count--; 258 list_del(&target->ft_chain); 259 spin_unlock(&fld->lcf_lock); 260 261 if (target->ft_exp != NULL) 262 class_export_put(target->ft_exp); 263 264 OBD_FREE_PTR(target); 265 return 0; 266 } 267 } 268 spin_unlock(&fld->lcf_lock); 269 return -ENOENT; 270} 271EXPORT_SYMBOL(fld_client_del_target); 272 273struct proc_dir_entry *fld_type_proc_dir = NULL; 274 275#if defined (CONFIG_PROC_FS) 276static int fld_client_proc_init(struct lu_client_fld *fld) 277{ 278 int rc; 279 280 fld->lcf_proc_dir = lprocfs_register(fld->lcf_name, 281 fld_type_proc_dir, 282 NULL, NULL); 283 284 if (IS_ERR(fld->lcf_proc_dir)) { 285 CERROR("%s: LProcFS failed in fld-init\n", 286 fld->lcf_name); 287 rc = PTR_ERR(fld->lcf_proc_dir); 288 return rc; 289 } 290 291 rc = lprocfs_add_vars(fld->lcf_proc_dir, 292 fld_client_proc_list, fld); 293 if (rc) { 294 CERROR("%s: Can't init FLD proc, rc %d\n", 295 fld->lcf_name, rc); 296 GOTO(out_cleanup, rc); 297 } 298 299 return 0; 300 301out_cleanup: 302 fld_client_proc_fini(fld); 303 return rc; 304} 305 306void fld_client_proc_fini(struct lu_client_fld *fld) 307{ 308 if (fld->lcf_proc_dir) { 309 if (!IS_ERR(fld->lcf_proc_dir)) 310 lprocfs_remove(&fld->lcf_proc_dir); 311 fld->lcf_proc_dir = NULL; 312 } 313} 314#else 315static int fld_client_proc_init(struct lu_client_fld *fld) 316{ 317 return 0; 318} 319 320void fld_client_proc_fini(struct lu_client_fld *fld) 321{ 322 return; 323} 324#endif 325EXPORT_SYMBOL(fld_client_proc_fini); 326 327static inline int hash_is_sane(int hash) 328{ 329 return (hash >= 0 && hash < ARRAY_SIZE(fld_hash)); 330} 331 332int fld_client_init(struct lu_client_fld *fld, 333 const char *prefix, int hash) 334{ 335 int cache_size, cache_threshold; 336 int rc; 337 338 LASSERT(fld != NULL); 339 340 snprintf(fld->lcf_name, sizeof(fld->lcf_name), 341 "cli-%s", prefix); 342 343 if (!hash_is_sane(hash)) { 344 CERROR("%s: Wrong hash function %#x\n", 345 fld->lcf_name, hash); 346 return -EINVAL; 347 } 348 349 fld->lcf_count = 0; 350 spin_lock_init(&fld->lcf_lock); 351 fld->lcf_hash = &fld_hash[hash]; 352 fld->lcf_flags = LUSTRE_FLD_INIT; 353 INIT_LIST_HEAD(&fld->lcf_targets); 354 355 cache_size = FLD_CLIENT_CACHE_SIZE / 356 sizeof(struct fld_cache_entry); 357 358 cache_threshold = cache_size * 359 FLD_CLIENT_CACHE_THRESHOLD / 100; 360 361 fld->lcf_cache = fld_cache_init(fld->lcf_name, 362 cache_size, cache_threshold); 363 if (IS_ERR(fld->lcf_cache)) { 364 rc = PTR_ERR(fld->lcf_cache); 365 fld->lcf_cache = NULL; 366 GOTO(out, rc); 367 } 368 369 rc = fld_client_proc_init(fld); 370 if (rc) 371 GOTO(out, rc); 372out: 373 if (rc) 374 fld_client_fini(fld); 375 else 376 CDEBUG(D_INFO, "%s: Using \"%s\" hash\n", 377 fld->lcf_name, fld->lcf_hash->fh_name); 378 return rc; 379} 380EXPORT_SYMBOL(fld_client_init); 381 382void fld_client_fini(struct lu_client_fld *fld) 383{ 384 struct lu_fld_target *target, *tmp; 385 386 spin_lock(&fld->lcf_lock); 387 list_for_each_entry_safe(target, tmp, 388 &fld->lcf_targets, ft_chain) { 389 fld->lcf_count--; 390 list_del(&target->ft_chain); 391 if (target->ft_exp != NULL) 392 class_export_put(target->ft_exp); 393 OBD_FREE_PTR(target); 394 } 395 spin_unlock(&fld->lcf_lock); 396 397 if (fld->lcf_cache != NULL) { 398 if (!IS_ERR(fld->lcf_cache)) 399 fld_cache_fini(fld->lcf_cache); 400 fld->lcf_cache = NULL; 401 } 402} 403EXPORT_SYMBOL(fld_client_fini); 404 405int fld_client_rpc(struct obd_export *exp, 406 struct lu_seq_range *range, __u32 fld_op) 407{ 408 struct ptlrpc_request *req; 409 struct lu_seq_range *prange; 410 __u32 *op; 411 int rc; 412 struct obd_import *imp; 413 414 LASSERT(exp != NULL); 415 416 imp = class_exp2cliimp(exp); 417 req = ptlrpc_request_alloc_pack(imp, &RQF_FLD_QUERY, LUSTRE_MDS_VERSION, 418 FLD_QUERY); 419 if (req == NULL) 420 return -ENOMEM; 421 422 op = req_capsule_client_get(&req->rq_pill, &RMF_FLD_OPC); 423 *op = fld_op; 424 425 prange = req_capsule_client_get(&req->rq_pill, &RMF_FLD_MDFLD); 426 *prange = *range; 427 428 ptlrpc_request_set_replen(req); 429 req->rq_request_portal = FLD_REQUEST_PORTAL; 430 ptlrpc_at_set_req_timeout(req); 431 432 if (fld_op == FLD_LOOKUP && 433 imp->imp_connect_flags_orig & OBD_CONNECT_MDS_MDS) 434 req->rq_allow_replay = 1; 435 436 if (fld_op != FLD_LOOKUP) 437 mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL); 438 fld_enter_request(&exp->exp_obd->u.cli); 439 rc = ptlrpc_queue_wait(req); 440 fld_exit_request(&exp->exp_obd->u.cli); 441 if (fld_op != FLD_LOOKUP) 442 mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL); 443 if (rc) 444 GOTO(out_req, rc); 445 446 prange = req_capsule_server_get(&req->rq_pill, &RMF_FLD_MDFLD); 447 if (prange == NULL) 448 GOTO(out_req, rc = -EFAULT); 449 *range = *prange; 450out_req: 451 ptlrpc_req_finished(req); 452 return rc; 453} 454 455int fld_client_lookup(struct lu_client_fld *fld, u64 seq, u32 *mds, 456 __u32 flags, const struct lu_env *env) 457{ 458 struct lu_seq_range res = { 0 }; 459 struct lu_fld_target *target; 460 int rc; 461 462 fld->lcf_flags |= LUSTRE_FLD_RUN; 463 464 rc = fld_cache_lookup(fld->lcf_cache, seq, &res); 465 if (rc == 0) { 466 *mds = res.lsr_index; 467 return 0; 468 } 469 470 /* Can not find it in the cache */ 471 target = fld_client_get_target(fld, seq); 472 LASSERT(target != NULL); 473 474 CDEBUG(D_INFO, "%s: Lookup fld entry (seq: %#llx) on target %s (idx %llu)\n", 475 fld->lcf_name, seq, fld_target_name(target), target->ft_idx); 476 477 res.lsr_start = seq; 478 fld_range_set_type(&res, flags); 479 rc = fld_client_rpc(target->ft_exp, &res, FLD_LOOKUP); 480 481 if (rc == 0) { 482 *mds = res.lsr_index; 483 484 fld_cache_insert(fld->lcf_cache, &res); 485 } 486 return rc; 487} 488EXPORT_SYMBOL(fld_client_lookup); 489 490void fld_client_flush(struct lu_client_fld *fld) 491{ 492 fld_cache_flush(fld->lcf_cache); 493} 494EXPORT_SYMBOL(fld_client_flush); 495 496static int __init fld_mod_init(void) 497{ 498 fld_type_proc_dir = lprocfs_register(LUSTRE_FLD_NAME, 499 proc_lustre_root, 500 NULL, NULL); 501 return PTR_ERR_OR_ZERO(fld_type_proc_dir); 502} 503 504static void __exit fld_mod_exit(void) 505{ 506 if (fld_type_proc_dir != NULL && !IS_ERR(fld_type_proc_dir)) { 507 lprocfs_remove(&fld_type_proc_dir); 508 fld_type_proc_dir = NULL; 509 } 510} 511 512MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>"); 513MODULE_DESCRIPTION("Lustre FLD"); 514MODULE_LICENSE("GPL"); 515 516module_init(fld_mod_init) 517module_exit(fld_mod_exit) 518