fid_request.c revision 29aaf4962a3bce337d37176858ef1025b9f29cc4
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/fid/fid_request.c 37 * 38 * Lustre Sequence Manager 39 * 40 * Author: Yury Umanets <umka@clusterfs.com> 41 */ 42 43#define DEBUG_SUBSYSTEM S_FID 44 45#include <linux/libcfs/libcfs.h> 46#include <linux/module.h> 47 48#include <obd.h> 49#include <obd_class.h> 50#include <obd_support.h> 51#include <lustre_fid.h> 52/* mdc RPC locks */ 53#include <lustre_mdc.h> 54#include "fid_internal.h" 55 56static int seq_client_rpc(struct lu_client_seq *seq, 57 struct lu_seq_range *output, __u32 opc, 58 const char *opcname) 59{ 60 struct obd_export *exp = seq->lcs_exp; 61 struct ptlrpc_request *req; 62 struct lu_seq_range *out, *in; 63 __u32 *op; 64 unsigned int debug_mask; 65 int rc; 66 67 req = ptlrpc_request_alloc_pack(class_exp2cliimp(exp), &RQF_SEQ_QUERY, 68 LUSTRE_MDS_VERSION, SEQ_QUERY); 69 if (req == NULL) 70 RETURN(-ENOMEM); 71 72 /* Init operation code */ 73 op = req_capsule_client_get(&req->rq_pill, &RMF_SEQ_OPC); 74 *op = opc; 75 76 /* Zero out input range, this is not recovery yet. */ 77 in = req_capsule_client_get(&req->rq_pill, &RMF_SEQ_RANGE); 78 range_init(in); 79 80 ptlrpc_request_set_replen(req); 81 82 in->lsr_index = seq->lcs_space.lsr_index; 83 if (seq->lcs_type == LUSTRE_SEQ_METADATA) 84 fld_range_set_mdt(in); 85 else 86 fld_range_set_ost(in); 87 88 if (opc == SEQ_ALLOC_SUPER) { 89 req->rq_request_portal = SEQ_CONTROLLER_PORTAL; 90 req->rq_reply_portal = MDC_REPLY_PORTAL; 91 /* During allocating super sequence for data object, 92 * the current thread might hold the export of MDT0(MDT0 93 * precreating objects on this OST), and it will send the 94 * request to MDT0 here, so we can not keep resending the 95 * request here, otherwise if MDT0 is failed(umounted), 96 * it can not release the export of MDT0 */ 97 if (seq->lcs_type == LUSTRE_SEQ_DATA) 98 req->rq_no_delay = req->rq_no_resend = 1; 99 debug_mask = D_CONSOLE; 100 } else { 101 if (seq->lcs_type == LUSTRE_SEQ_METADATA) 102 req->rq_request_portal = SEQ_METADATA_PORTAL; 103 else 104 req->rq_request_portal = SEQ_DATA_PORTAL; 105 debug_mask = D_INFO; 106 } 107 108 ptlrpc_at_set_req_timeout(req); 109 110 if (seq->lcs_type == LUSTRE_SEQ_METADATA) 111 mdc_get_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL); 112 rc = ptlrpc_queue_wait(req); 113 if (seq->lcs_type == LUSTRE_SEQ_METADATA) 114 mdc_put_rpc_lock(exp->exp_obd->u.cli.cl_rpc_lock, NULL); 115 if (rc) 116 GOTO(out_req, rc); 117 118 out = req_capsule_server_get(&req->rq_pill, &RMF_SEQ_RANGE); 119 *output = *out; 120 121 if (!range_is_sane(output)) { 122 CERROR("%s: Invalid range received from server: " 123 DRANGE"\n", seq->lcs_name, PRANGE(output)); 124 GOTO(out_req, rc = -EINVAL); 125 } 126 127 if (range_is_exhausted(output)) { 128 CERROR("%s: Range received from server is exhausted: " 129 DRANGE"]\n", seq->lcs_name, PRANGE(output)); 130 GOTO(out_req, rc = -EINVAL); 131 } 132 133 CDEBUG_LIMIT(debug_mask, "%s: Allocated %s-sequence "DRANGE"]\n", 134 seq->lcs_name, opcname, PRANGE(output)); 135 136 EXIT; 137out_req: 138 ptlrpc_req_finished(req); 139 return rc; 140} 141 142/* Request sequence-controller node to allocate new super-sequence. */ 143int seq_client_alloc_super(struct lu_client_seq *seq, 144 const struct lu_env *env) 145{ 146 int rc; 147 148 mutex_lock(&seq->lcs_mutex); 149 150 if (seq->lcs_srv) { 151 rc = 0; 152 } else { 153 /* Check whether the connection to seq controller has been 154 * setup (lcs_exp != NULL) */ 155 if (seq->lcs_exp == NULL) { 156 mutex_unlock(&seq->lcs_mutex); 157 RETURN(-EINPROGRESS); 158 } 159 160 rc = seq_client_rpc(seq, &seq->lcs_space, 161 SEQ_ALLOC_SUPER, "super"); 162 } 163 mutex_unlock(&seq->lcs_mutex); 164 RETURN(rc); 165} 166 167/* Request sequence-controller node to allocate new meta-sequence. */ 168static int seq_client_alloc_meta(const struct lu_env *env, 169 struct lu_client_seq *seq) 170{ 171 int rc; 172 173 if (seq->lcs_srv) { 174 rc = 0; 175 } else { 176 do { 177 /* If meta server return -EINPROGRESS or EAGAIN, 178 * it means meta server might not be ready to 179 * allocate super sequence from sequence controller 180 * (MDT0)yet */ 181 rc = seq_client_rpc(seq, &seq->lcs_space, 182 SEQ_ALLOC_META, "meta"); 183 } while (rc == -EINPROGRESS || rc == -EAGAIN); 184 } 185 186 RETURN(rc); 187} 188 189/* Allocate new sequence for client. */ 190static int seq_client_alloc_seq(const struct lu_env *env, 191 struct lu_client_seq *seq, seqno_t *seqnr) 192{ 193 int rc; 194 195 LASSERT(range_is_sane(&seq->lcs_space)); 196 197 if (range_is_exhausted(&seq->lcs_space)) { 198 rc = seq_client_alloc_meta(env, seq); 199 if (rc) { 200 CERROR("%s: Can't allocate new meta-sequence," 201 "rc %d\n", seq->lcs_name, rc); 202 RETURN(rc); 203 } else { 204 CDEBUG(D_INFO, "%s: New range - "DRANGE"\n", 205 seq->lcs_name, PRANGE(&seq->lcs_space)); 206 } 207 } else { 208 rc = 0; 209 } 210 211 LASSERT(!range_is_exhausted(&seq->lcs_space)); 212 *seqnr = seq->lcs_space.lsr_start; 213 seq->lcs_space.lsr_start += 1; 214 215 CDEBUG(D_INFO, "%s: Allocated sequence ["LPX64"]\n", seq->lcs_name, 216 *seqnr); 217 218 RETURN(rc); 219} 220 221static int seq_fid_alloc_prep(struct lu_client_seq *seq, 222 wait_queue_t *link) 223{ 224 if (seq->lcs_update) { 225 add_wait_queue(&seq->lcs_waitq, link); 226 set_current_state(TASK_UNINTERRUPTIBLE); 227 mutex_unlock(&seq->lcs_mutex); 228 229 waitq_wait(link, TASK_UNINTERRUPTIBLE); 230 231 mutex_lock(&seq->lcs_mutex); 232 remove_wait_queue(&seq->lcs_waitq, link); 233 set_current_state(TASK_RUNNING); 234 return -EAGAIN; 235 } 236 ++seq->lcs_update; 237 mutex_unlock(&seq->lcs_mutex); 238 return 0; 239} 240 241static void seq_fid_alloc_fini(struct lu_client_seq *seq) 242{ 243 LASSERT(seq->lcs_update == 1); 244 mutex_lock(&seq->lcs_mutex); 245 --seq->lcs_update; 246 wake_up(&seq->lcs_waitq); 247} 248 249/** 250 * Allocate the whole seq to the caller. 251 **/ 252int seq_client_get_seq(const struct lu_env *env, 253 struct lu_client_seq *seq, seqno_t *seqnr) 254{ 255 wait_queue_t link; 256 int rc; 257 258 LASSERT(seqnr != NULL); 259 mutex_lock(&seq->lcs_mutex); 260 init_waitqueue_entry_current(&link); 261 262 while (1) { 263 rc = seq_fid_alloc_prep(seq, &link); 264 if (rc == 0) 265 break; 266 } 267 268 rc = seq_client_alloc_seq(env, seq, seqnr); 269 if (rc) { 270 CERROR("%s: Can't allocate new sequence, " 271 "rc %d\n", seq->lcs_name, rc); 272 seq_fid_alloc_fini(seq); 273 mutex_unlock(&seq->lcs_mutex); 274 return rc; 275 } 276 277 CDEBUG(D_INFO, "%s: allocate sequence " 278 "[0x%16.16"LPF64"x]\n", seq->lcs_name, *seqnr); 279 280 /* Since the caller require the whole seq, 281 * so marked this seq to be used */ 282 if (seq->lcs_type == LUSTRE_SEQ_METADATA) 283 seq->lcs_fid.f_oid = LUSTRE_METADATA_SEQ_MAX_WIDTH; 284 else 285 seq->lcs_fid.f_oid = LUSTRE_DATA_SEQ_MAX_WIDTH; 286 287 seq->lcs_fid.f_seq = *seqnr; 288 seq->lcs_fid.f_ver = 0; 289 /* 290 * Inform caller that sequence switch is performed to allow it 291 * to setup FLD for it. 292 */ 293 seq_fid_alloc_fini(seq); 294 mutex_unlock(&seq->lcs_mutex); 295 296 return rc; 297} 298EXPORT_SYMBOL(seq_client_get_seq); 299 300/* Allocate new fid on passed client @seq and save it to @fid. */ 301int seq_client_alloc_fid(const struct lu_env *env, 302 struct lu_client_seq *seq, struct lu_fid *fid) 303{ 304 wait_queue_t link; 305 int rc; 306 307 LASSERT(seq != NULL); 308 LASSERT(fid != NULL); 309 310 init_waitqueue_entry_current(&link); 311 mutex_lock(&seq->lcs_mutex); 312 313 if (OBD_FAIL_CHECK(OBD_FAIL_SEQ_EXHAUST)) 314 seq->lcs_fid.f_oid = seq->lcs_width; 315 316 while (1) { 317 seqno_t seqnr; 318 319 if (!fid_is_zero(&seq->lcs_fid) && 320 fid_oid(&seq->lcs_fid) < seq->lcs_width) { 321 /* Just bump last allocated fid and return to caller. */ 322 seq->lcs_fid.f_oid += 1; 323 rc = 0; 324 break; 325 } 326 327 rc = seq_fid_alloc_prep(seq, &link); 328 if (rc) 329 continue; 330 331 rc = seq_client_alloc_seq(env, seq, &seqnr); 332 if (rc) { 333 CERROR("%s: Can't allocate new sequence, " 334 "rc %d\n", seq->lcs_name, rc); 335 seq_fid_alloc_fini(seq); 336 mutex_unlock(&seq->lcs_mutex); 337 RETURN(rc); 338 } 339 340 CDEBUG(D_INFO, "%s: Switch to sequence " 341 "[0x%16.16"LPF64"x]\n", seq->lcs_name, seqnr); 342 343 seq->lcs_fid.f_oid = LUSTRE_FID_INIT_OID; 344 seq->lcs_fid.f_seq = seqnr; 345 seq->lcs_fid.f_ver = 0; 346 347 /* 348 * Inform caller that sequence switch is performed to allow it 349 * to setup FLD for it. 350 */ 351 rc = 1; 352 353 seq_fid_alloc_fini(seq); 354 break; 355 } 356 357 *fid = seq->lcs_fid; 358 mutex_unlock(&seq->lcs_mutex); 359 360 CDEBUG(D_INFO, "%s: Allocated FID "DFID"\n", seq->lcs_name, PFID(fid)); 361 RETURN(rc); 362} 363EXPORT_SYMBOL(seq_client_alloc_fid); 364 365/* 366 * Finish the current sequence due to disconnect. 367 * See mdc_import_event() 368 */ 369void seq_client_flush(struct lu_client_seq *seq) 370{ 371 wait_queue_t link; 372 373 LASSERT(seq != NULL); 374 init_waitqueue_entry_current(&link); 375 mutex_lock(&seq->lcs_mutex); 376 377 while (seq->lcs_update) { 378 add_wait_queue(&seq->lcs_waitq, &link); 379 set_current_state(TASK_UNINTERRUPTIBLE); 380 mutex_unlock(&seq->lcs_mutex); 381 382 waitq_wait(&link, TASK_UNINTERRUPTIBLE); 383 384 mutex_lock(&seq->lcs_mutex); 385 remove_wait_queue(&seq->lcs_waitq, &link); 386 set_current_state(TASK_RUNNING); 387 } 388 389 fid_zero(&seq->lcs_fid); 390 /** 391 * this id shld not be used for seq range allocation. 392 * set to -1 for dgb check. 393 */ 394 395 seq->lcs_space.lsr_index = -1; 396 397 range_init(&seq->lcs_space); 398 mutex_unlock(&seq->lcs_mutex); 399} 400EXPORT_SYMBOL(seq_client_flush); 401 402static void seq_client_proc_fini(struct lu_client_seq *seq) 403{ 404#ifdef LPROCFS 405 if (seq->lcs_proc_dir) { 406 if (!IS_ERR(seq->lcs_proc_dir)) 407 lprocfs_remove(&seq->lcs_proc_dir); 408 seq->lcs_proc_dir = NULL; 409 } 410 EXIT; 411#endif /* LPROCFS */ 412} 413 414static int seq_client_proc_init(struct lu_client_seq *seq) 415{ 416#ifdef LPROCFS 417 int rc; 418 419 seq->lcs_proc_dir = lprocfs_register(seq->lcs_name, 420 seq_type_proc_dir, 421 NULL, NULL); 422 423 if (IS_ERR(seq->lcs_proc_dir)) { 424 CERROR("%s: LProcFS failed in seq-init\n", 425 seq->lcs_name); 426 rc = PTR_ERR(seq->lcs_proc_dir); 427 RETURN(rc); 428 } 429 430 rc = lprocfs_add_vars(seq->lcs_proc_dir, 431 seq_client_proc_list, seq); 432 if (rc) { 433 CERROR("%s: Can't init sequence manager " 434 "proc, rc %d\n", seq->lcs_name, rc); 435 GOTO(out_cleanup, rc); 436 } 437 438 RETURN(0); 439 440out_cleanup: 441 seq_client_proc_fini(seq); 442 return rc; 443 444#else /* LPROCFS */ 445 return 0; 446#endif 447} 448 449int seq_client_init(struct lu_client_seq *seq, 450 struct obd_export *exp, 451 enum lu_cli_type type, 452 const char *prefix, 453 struct lu_server_seq *srv) 454{ 455 int rc; 456 457 LASSERT(seq != NULL); 458 LASSERT(prefix != NULL); 459 460 seq->lcs_srv = srv; 461 seq->lcs_type = type; 462 463 mutex_init(&seq->lcs_mutex); 464 if (type == LUSTRE_SEQ_METADATA) 465 seq->lcs_width = LUSTRE_METADATA_SEQ_MAX_WIDTH; 466 else 467 seq->lcs_width = LUSTRE_DATA_SEQ_MAX_WIDTH; 468 469 init_waitqueue_head(&seq->lcs_waitq); 470 /* Make sure that things are clear before work is started. */ 471 seq_client_flush(seq); 472 473 if (exp != NULL) 474 seq->lcs_exp = class_export_get(exp); 475 else if (type == LUSTRE_SEQ_METADATA) 476 LASSERT(seq->lcs_srv != NULL); 477 478 snprintf(seq->lcs_name, sizeof(seq->lcs_name), 479 "cli-%s", prefix); 480 481 rc = seq_client_proc_init(seq); 482 if (rc) 483 seq_client_fini(seq); 484 RETURN(rc); 485} 486EXPORT_SYMBOL(seq_client_init); 487 488void seq_client_fini(struct lu_client_seq *seq) 489{ 490 seq_client_proc_fini(seq); 491 492 if (seq->lcs_exp != NULL) { 493 class_export_put(seq->lcs_exp); 494 seq->lcs_exp = NULL; 495 } 496 497 seq->lcs_srv = NULL; 498 EXIT; 499} 500EXPORT_SYMBOL(seq_client_fini); 501 502int client_fid_init(struct obd_device *obd, 503 struct obd_export *exp, enum lu_cli_type type) 504{ 505 struct client_obd *cli = &obd->u.cli; 506 char *prefix; 507 int rc; 508 509 OBD_ALLOC_PTR(cli->cl_seq); 510 if (cli->cl_seq == NULL) 511 RETURN(-ENOMEM); 512 513 OBD_ALLOC(prefix, MAX_OBD_NAME + 5); 514 if (prefix == NULL) 515 GOTO(out_free_seq, rc = -ENOMEM); 516 517 snprintf(prefix, MAX_OBD_NAME + 5, "cli-%s", obd->obd_name); 518 519 /* Init client side sequence-manager */ 520 rc = seq_client_init(cli->cl_seq, exp, type, prefix, NULL); 521 OBD_FREE(prefix, MAX_OBD_NAME + 5); 522 if (rc) 523 GOTO(out_free_seq, rc); 524 525 RETURN(rc); 526out_free_seq: 527 OBD_FREE_PTR(cli->cl_seq); 528 cli->cl_seq = NULL; 529 return rc; 530} 531EXPORT_SYMBOL(client_fid_init); 532 533int client_fid_fini(struct obd_device *obd) 534{ 535 struct client_obd *cli = &obd->u.cli; 536 537 if (cli->cl_seq != NULL) { 538 seq_client_fini(cli->cl_seq); 539 OBD_FREE_PTR(cli->cl_seq); 540 cli->cl_seq = NULL; 541 } 542 543 RETURN(0); 544} 545EXPORT_SYMBOL(client_fid_fini); 546 547struct proc_dir_entry *seq_type_proc_dir; 548 549static int __init fid_mod_init(void) 550{ 551 seq_type_proc_dir = lprocfs_register(LUSTRE_SEQ_NAME, 552 proc_lustre_root, 553 NULL, NULL); 554 if (IS_ERR(seq_type_proc_dir)) 555 return PTR_ERR(seq_type_proc_dir); 556 return 0; 557} 558 559static void __exit fid_mod_exit(void) 560{ 561 if (seq_type_proc_dir != NULL && !IS_ERR(seq_type_proc_dir)) { 562 lprocfs_remove(&seq_type_proc_dir); 563 seq_type_proc_dir = NULL; 564 } 565} 566 567MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>"); 568MODULE_DESCRIPTION("Lustre FID Module"); 569MODULE_LICENSE("GPL"); 570 571cfs_module(fid, "0.1.0", fid_mod_init, fid_mod_exit); 572