1 /* -*- mode: c; c-basic-offset: 8; -*- 2 * vim: noexpandtab sw=8 ts=8 sts=0: 3 * 4 * dlmlock.c 5 * 6 * underlying calls for lock creation 7 * 8 * Copyright (C) 2004 Oracle. All rights reserved. 9 * 10 * This program is free software; you can redistribute it and/or 11 * modify it under the terms of the GNU General Public 12 * License as published by the Free Software Foundation; either 13 * version 2 of the License, or (at your option) any later version. 14 * 15 * This program is distributed in the hope that it will be useful, 16 * but WITHOUT ANY WARRANTY; without even the implied warranty of 17 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 18 * General Public License for more details. 19 * 20 * You should have received a copy of the GNU General Public 21 * License along with this program; if not, write to the 22 * Free Software Foundation, Inc., 59 Temple Place - Suite 330, 23 * Boston, MA 021110-1307, USA. 24 * 25 */ 26 27 28 #include <linux/module.h> 29 #include <linux/fs.h> 30 #include <linux/types.h> 31 #include <linux/slab.h> 32 #include <linux/highmem.h> 33 #include <linux/init.h> 34 #include <linux/sysctl.h> 35 #include <linux/random.h> 36 #include <linux/blkdev.h> 37 #include <linux/socket.h> 38 #include <linux/inet.h> 39 #include <linux/spinlock.h> 40 #include <linux/delay.h> 41 42 43 #include "cluster/heartbeat.h" 44 #include "cluster/nodemanager.h" 45 #include "cluster/tcp.h" 46 47 #include "dlmapi.h" 48 #include "dlmcommon.h" 49 50 #include "dlmconvert.h" 51 52 #define MLOG_MASK_PREFIX ML_DLM 53 #include "cluster/masklog.h" 54 55 static struct kmem_cache *dlm_lock_cache = NULL; 56 57 static DEFINE_SPINLOCK(dlm_cookie_lock); 58 static u64 dlm_next_cookie = 1; 59 60 static enum dlm_status dlm_send_remote_lock_request(struct dlm_ctxt *dlm, 61 struct dlm_lock_resource *res, 62 struct dlm_lock *lock, int flags); 63 static void dlm_init_lock(struct dlm_lock *newlock, int type, 64 u8 node, u64 cookie); 65 static void dlm_lock_release(struct kref *kref); 66 static void dlm_lock_detach_lockres(struct dlm_lock *lock); 67 68 int dlm_init_lock_cache(void) 69 { 70 dlm_lock_cache = kmem_cache_create("o2dlm_lock", 71 sizeof(struct dlm_lock), 72 0, SLAB_HWCACHE_ALIGN, NULL); 73 if (dlm_lock_cache == NULL) 74 return -ENOMEM; 75 return 0; 76 } 77 78 void dlm_destroy_lock_cache(void) 79 { 80 if (dlm_lock_cache) 81 kmem_cache_destroy(dlm_lock_cache); 82 } 83 84 /* Tell us whether we can grant a new lock request. 85 * locking: 86 * caller needs: res->spinlock 87 * taken: none 88 * held on exit: none 89 * returns: 1 if the lock can be granted, 0 otherwise. 90 */ 91 static int dlm_can_grant_new_lock(struct dlm_lock_resource *res, 92 struct dlm_lock *lock) 93 { 94 struct list_head *iter; 95 struct dlm_lock *tmplock; 96 97 list_for_each(iter, &res->granted) { 98 tmplock = list_entry(iter, struct dlm_lock, list); 99 100 if (!dlm_lock_compatible(tmplock->ml.type, lock->ml.type)) 101 return 0; 102 } 103 104 list_for_each(iter, &res->converting) { 105 tmplock = list_entry(iter, struct dlm_lock, list); 106 107 if (!dlm_lock_compatible(tmplock->ml.type, lock->ml.type)) 108 return 0; 109 } 110 111 return 1; 112 } 113 114 /* performs lock creation at the lockres master site 115 * locking: 116 * caller needs: none 117 * taken: takes and drops res->spinlock 118 * held on exit: none 119 * returns: DLM_NORMAL, DLM_NOTQUEUED 120 */ 121 static enum dlm_status dlmlock_master(struct dlm_ctxt *dlm, 122 struct dlm_lock_resource *res, 123 struct dlm_lock *lock, int flags) 124 { 125 int call_ast = 0, kick_thread = 0; 126 enum dlm_status status = DLM_NORMAL; 127 128 mlog_entry("type=%d\n", lock->ml.type); 129 130 spin_lock(&res->spinlock); 131 /* if called from dlm_create_lock_handler, need to 132 * ensure it will not sleep in dlm_wait_on_lockres */ 133 status = __dlm_lockres_state_to_status(res); 134 if (status != DLM_NORMAL && 135 lock->ml.node != dlm->node_num) { 136 /* erf. state changed after lock was dropped. */ 137 spin_unlock(&res->spinlock); 138 dlm_error(status); 139 return status; 140 } 141 __dlm_wait_on_lockres(res); 142 __dlm_lockres_reserve_ast(res); 143 144 if (dlm_can_grant_new_lock(res, lock)) { 145 mlog(0, "I can grant this lock right away\n"); 146 /* got it right away */ 147 lock->lksb->status = DLM_NORMAL; 148 status = DLM_NORMAL; 149 dlm_lock_get(lock); 150 list_add_tail(&lock->list, &res->granted); 151 152 /* for the recovery lock, we can't allow the ast 153 * to be queued since the dlmthread is already 154 * frozen. but the recovery lock is always locked 155 * with LKM_NOQUEUE so we do not need the ast in 156 * this special case */ 157 if (!dlm_is_recovery_lock(res->lockname.name, 158 res->lockname.len)) { 159 kick_thread = 1; 160 call_ast = 1; 161 } else { 162 mlog(0, "%s: returning DLM_NORMAL to " 163 "node %u for reco lock\n", dlm->name, 164 lock->ml.node); 165 } 166 } else { 167 /* for NOQUEUE request, unless we get the 168 * lock right away, return DLM_NOTQUEUED */ 169 if (flags & LKM_NOQUEUE) { 170 status = DLM_NOTQUEUED; 171 if (dlm_is_recovery_lock(res->lockname.name, 172 res->lockname.len)) { 173 mlog(0, "%s: returning NOTQUEUED to " 174 "node %u for reco lock\n", dlm->name, 175 lock->ml.node); 176 } 177 } else { 178 dlm_lock_get(lock); 179 list_add_tail(&lock->list, &res->blocked); 180 kick_thread = 1; 181 } 182 } 183 /* reduce the inflight count, this may result in the lockres 184 * being purged below during calc_usage */ 185 if (lock->ml.node == dlm->node_num) 186 dlm_lockres_drop_inflight_ref(dlm, res); 187 188 spin_unlock(&res->spinlock); 189 wake_up(&res->wq); 190 191 /* either queue the ast or release it */ 192 if (call_ast) 193 dlm_queue_ast(dlm, lock); 194 else 195 dlm_lockres_release_ast(dlm, res); 196 197 dlm_lockres_calc_usage(dlm, res); 198 if (kick_thread) 199 dlm_kick_thread(dlm, res); 200 201 return status; 202 } 203 204 void dlm_revert_pending_lock(struct dlm_lock_resource *res, 205 struct dlm_lock *lock) 206 { 207 /* remove from local queue if it failed */ 208 list_del_init(&lock->list); 209 lock->lksb->flags &= ~DLM_LKSB_GET_LVB; 210 } 211 212 213 /* 214 * locking: 215 * caller needs: none 216 * taken: takes and drops res->spinlock 217 * held on exit: none 218 * returns: DLM_DENIED, DLM_RECOVERING, or net status 219 */ 220 static enum dlm_status dlmlock_remote(struct dlm_ctxt *dlm, 221 struct dlm_lock_resource *res, 222 struct dlm_lock *lock, int flags) 223 { 224 enum dlm_status status = DLM_DENIED; 225 int lockres_changed = 1; 226 227 mlog_entry("type=%d\n", lock->ml.type); 228 mlog(0, "lockres %.*s, flags = 0x%x\n", res->lockname.len, 229 res->lockname.name, flags); 230 231 spin_lock(&res->spinlock); 232 233 /* will exit this call with spinlock held */ 234 __dlm_wait_on_lockres(res); 235 res->state |= DLM_LOCK_RES_IN_PROGRESS; 236 237 /* add lock to local (secondary) queue */ 238 dlm_lock_get(lock); 239 list_add_tail(&lock->list, &res->blocked); 240 lock->lock_pending = 1; 241 spin_unlock(&res->spinlock); 242 243 /* spec seems to say that you will get DLM_NORMAL when the lock 244 * has been queued, meaning we need to wait for a reply here. */ 245 status = dlm_send_remote_lock_request(dlm, res, lock, flags); 246 247 spin_lock(&res->spinlock); 248 res->state &= ~DLM_LOCK_RES_IN_PROGRESS; 249 lock->lock_pending = 0; 250 if (status != DLM_NORMAL) { 251 if (status == DLM_RECOVERING && 252 dlm_is_recovery_lock(res->lockname.name, 253 res->lockname.len)) { 254 /* recovery lock was mastered by dead node. 255 * we need to have calc_usage shoot down this 256 * lockres and completely remaster it. */ 257 mlog(0, "%s: recovery lock was owned by " 258 "dead node %u, remaster it now.\n", 259 dlm->name, res->owner); 260 } else if (status != DLM_NOTQUEUED) { 261 /* 262 * DO NOT call calc_usage, as this would unhash 263 * the remote lockres before we ever get to use 264 * it. treat as if we never made any change to 265 * the lockres. 266 */ 267 lockres_changed = 0; 268 dlm_error(status); 269 } 270 dlm_revert_pending_lock(res, lock); 271 dlm_lock_put(lock); 272 } else if (dlm_is_recovery_lock(res->lockname.name, 273 res->lockname.len)) { 274 /* special case for the $RECOVERY lock. 275 * there will never be an AST delivered to put 276 * this lock on the proper secondary queue 277 * (granted), so do it manually. */ 278 mlog(0, "%s: $RECOVERY lock for this node (%u) is " 279 "mastered by %u; got lock, manually granting (no ast)\n", 280 dlm->name, dlm->node_num, res->owner); 281 list_move_tail(&lock->list, &res->granted); 282 } 283 spin_unlock(&res->spinlock); 284 285 if (lockres_changed) 286 dlm_lockres_calc_usage(dlm, res); 287 288 wake_up(&res->wq); 289 return status; 290 } 291 292 293 /* for remote lock creation. 294 * locking: 295 * caller needs: none, but need res->state & DLM_LOCK_RES_IN_PROGRESS 296 * taken: none 297 * held on exit: none 298 * returns: DLM_NOLOCKMGR, or net status 299 */ 300 static enum dlm_status dlm_send_remote_lock_request(struct dlm_ctxt *dlm, 301 struct dlm_lock_resource *res, 302 struct dlm_lock *lock, int flags) 303 { 304 struct dlm_create_lock create; 305 int tmpret, status = 0; 306 enum dlm_status ret; 307 308 mlog_entry_void(); 309 310 memset(&create, 0, sizeof(create)); 311 create.node_idx = dlm->node_num; 312 create.requested_type = lock->ml.type; 313 create.cookie = lock->ml.cookie; 314 create.namelen = res->lockname.len; 315 create.flags = cpu_to_be32(flags); 316 memcpy(create.name, res->lockname.name, create.namelen); 317 318 tmpret = o2net_send_message(DLM_CREATE_LOCK_MSG, dlm->key, &create, 319 sizeof(create), res->owner, &status); 320 if (tmpret >= 0) { 321 // successfully sent and received 322 ret = status; // this is already a dlm_status 323 if (ret == DLM_REJECTED) { 324 mlog(ML_ERROR, "%s:%.*s: BUG. this is a stale lockres " 325 "no longer owned by %u. that node is coming back " 326 "up currently.\n", dlm->name, create.namelen, 327 create.name, res->owner); 328 dlm_print_one_lock_resource(res); 329 BUG(); 330 } 331 } else { 332 mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to " 333 "node %u\n", tmpret, DLM_CREATE_LOCK_MSG, dlm->key, 334 res->owner); 335 if (dlm_is_host_down(tmpret)) { 336 ret = DLM_RECOVERING; 337 mlog(0, "node %u died so returning DLM_RECOVERING " 338 "from lock message!\n", res->owner); 339 } else { 340 ret = dlm_err_to_dlm_status(tmpret); 341 } 342 } 343 344 return ret; 345 } 346 347 void dlm_lock_get(struct dlm_lock *lock) 348 { 349 kref_get(&lock->lock_refs); 350 } 351 352 void dlm_lock_put(struct dlm_lock *lock) 353 { 354 kref_put(&lock->lock_refs, dlm_lock_release); 355 } 356 357 static void dlm_lock_release(struct kref *kref) 358 { 359 struct dlm_lock *lock; 360 361 lock = container_of(kref, struct dlm_lock, lock_refs); 362 363 BUG_ON(!list_empty(&lock->list)); 364 BUG_ON(!list_empty(&lock->ast_list)); 365 BUG_ON(!list_empty(&lock->bast_list)); 366 BUG_ON(lock->ast_pending); 367 BUG_ON(lock->bast_pending); 368 369 dlm_lock_detach_lockres(lock); 370 371 if (lock->lksb_kernel_allocated) { 372 mlog(0, "freeing kernel-allocated lksb\n"); 373 kfree(lock->lksb); 374 } 375 kmem_cache_free(dlm_lock_cache, lock); 376 } 377 378 /* associate a lock with it's lockres, getting a ref on the lockres */ 379 void dlm_lock_attach_lockres(struct dlm_lock *lock, 380 struct dlm_lock_resource *res) 381 { 382 dlm_lockres_get(res); 383 lock->lockres = res; 384 } 385 386 /* drop ref on lockres, if there is still one associated with lock */ 387 static void dlm_lock_detach_lockres(struct dlm_lock *lock) 388 { 389 struct dlm_lock_resource *res; 390 391 res = lock->lockres; 392 if (res) { 393 lock->lockres = NULL; 394 mlog(0, "removing lock's lockres reference\n"); 395 dlm_lockres_put(res); 396 } 397 } 398 399 static void dlm_init_lock(struct dlm_lock *newlock, int type, 400 u8 node, u64 cookie) 401 { 402 INIT_LIST_HEAD(&newlock->list); 403 INIT_LIST_HEAD(&newlock->ast_list); 404 INIT_LIST_HEAD(&newlock->bast_list); 405 spin_lock_init(&newlock->spinlock); 406 newlock->ml.type = type; 407 newlock->ml.convert_type = LKM_IVMODE; 408 newlock->ml.highest_blocked = LKM_IVMODE; 409 newlock->ml.node = node; 410 newlock->ml.pad1 = 0; 411 newlock->ml.list = 0; 412 newlock->ml.flags = 0; 413 newlock->ast = NULL; 414 newlock->bast = NULL; 415 newlock->astdata = NULL; 416 newlock->ml.cookie = cpu_to_be64(cookie); 417 newlock->ast_pending = 0; 418 newlock->bast_pending = 0; 419 newlock->convert_pending = 0; 420 newlock->lock_pending = 0; 421 newlock->unlock_pending = 0; 422 newlock->cancel_pending = 0; 423 newlock->lksb_kernel_allocated = 0; 424 425 kref_init(&newlock->lock_refs); 426 } 427 428 struct dlm_lock * dlm_new_lock(int type, u8 node, u64 cookie, 429 struct dlm_lockstatus *lksb) 430 { 431 struct dlm_lock *lock; 432 int kernel_allocated = 0; 433 434 lock = kmem_cache_zalloc(dlm_lock_cache, GFP_NOFS); 435 if (!lock) 436 return NULL; 437 438 if (!lksb) { 439 /* zero memory only if kernel-allocated */ 440 lksb = kzalloc(sizeof(*lksb), GFP_NOFS); 441 if (!lksb) { 442 kfree(lock); 443 return NULL; 444 } 445 kernel_allocated = 1; 446 } 447 448 dlm_init_lock(lock, type, node, cookie); 449 if (kernel_allocated) 450 lock->lksb_kernel_allocated = 1; 451 lock->lksb = lksb; 452 lksb->lockid = lock; 453 return lock; 454 } 455 456 /* handler for lock creation net message 457 * locking: 458 * caller needs: none 459 * taken: takes and drops res->spinlock 460 * held on exit: none 461 * returns: DLM_NORMAL, DLM_SYSERR, DLM_IVLOCKID, DLM_NOTQUEUED 462 */ 463 int dlm_create_lock_handler(struct o2net_msg *msg, u32 len, void *data, 464 void **ret_data) 465 { 466 struct dlm_ctxt *dlm = data; 467 struct dlm_create_lock *create = (struct dlm_create_lock *)msg->buf; 468 struct dlm_lock_resource *res = NULL; 469 struct dlm_lock *newlock = NULL; 470 struct dlm_lockstatus *lksb = NULL; 471 enum dlm_status status = DLM_NORMAL; 472 char *name; 473 unsigned int namelen; 474 475 BUG_ON(!dlm); 476 477 mlog_entry_void(); 478 479 if (!dlm_grab(dlm)) 480 return DLM_REJECTED; 481 482 name = create->name; 483 namelen = create->namelen; 484 status = DLM_REJECTED; 485 if (!dlm_domain_fully_joined(dlm)) { 486 mlog(ML_ERROR, "Domain %s not fully joined, but node %u is " 487 "sending a create_lock message for lock %.*s!\n", 488 dlm->name, create->node_idx, namelen, name); 489 dlm_error(status); 490 goto leave; 491 } 492 493 status = DLM_IVBUFLEN; 494 if (namelen > DLM_LOCKID_NAME_MAX) { 495 dlm_error(status); 496 goto leave; 497 } 498 499 status = DLM_SYSERR; 500 newlock = dlm_new_lock(create->requested_type, 501 create->node_idx, 502 be64_to_cpu(create->cookie), NULL); 503 if (!newlock) { 504 dlm_error(status); 505 goto leave; 506 } 507 508 lksb = newlock->lksb; 509 510 if (be32_to_cpu(create->flags) & LKM_GET_LVB) { 511 lksb->flags |= DLM_LKSB_GET_LVB; 512 mlog(0, "set DLM_LKSB_GET_LVB flag\n"); 513 } 514 515 status = DLM_IVLOCKID; 516 res = dlm_lookup_lockres(dlm, name, namelen); 517 if (!res) { 518 dlm_error(status); 519 goto leave; 520 } 521 522 spin_lock(&res->spinlock); 523 status = __dlm_lockres_state_to_status(res); 524 spin_unlock(&res->spinlock); 525 526 if (status != DLM_NORMAL) { 527 mlog(0, "lockres recovering/migrating/in-progress\n"); 528 goto leave; 529 } 530 531 dlm_lock_attach_lockres(newlock, res); 532 533 status = dlmlock_master(dlm, res, newlock, be32_to_cpu(create->flags)); 534 leave: 535 if (status != DLM_NORMAL) 536 if (newlock) 537 dlm_lock_put(newlock); 538 539 if (res) 540 dlm_lockres_put(res); 541 542 dlm_put(dlm); 543 544 return status; 545 } 546 547 548 /* fetch next node-local (u8 nodenum + u56 cookie) into u64 */ 549 static inline void dlm_get_next_cookie(u8 node_num, u64 *cookie) 550 { 551 u64 tmpnode = node_num; 552 553 /* shift single byte of node num into top 8 bits */ 554 tmpnode <<= 56; 555 556 spin_lock(&dlm_cookie_lock); 557 *cookie = (dlm_next_cookie | tmpnode); 558 if (++dlm_next_cookie & 0xff00000000000000ull) { 559 mlog(0, "This node's cookie will now wrap!\n"); 560 dlm_next_cookie = 1; 561 } 562 spin_unlock(&dlm_cookie_lock); 563 } 564 565 enum dlm_status dlmlock(struct dlm_ctxt *dlm, int mode, 566 struct dlm_lockstatus *lksb, int flags, 567 const char *name, int namelen, dlm_astlockfunc_t *ast, 568 void *data, dlm_bastlockfunc_t *bast) 569 { 570 enum dlm_status status; 571 struct dlm_lock_resource *res = NULL; 572 struct dlm_lock *lock = NULL; 573 int convert = 0, recovery = 0; 574 575 /* yes this function is a mess. 576 * TODO: clean this up. lots of common code in the 577 * lock and convert paths, especially in the retry blocks */ 578 if (!lksb) { 579 dlm_error(DLM_BADARGS); 580 return DLM_BADARGS; 581 } 582 583 status = DLM_BADPARAM; 584 if (mode != LKM_EXMODE && mode != LKM_PRMODE && mode != LKM_NLMODE) { 585 dlm_error(status); 586 goto error; 587 } 588 589 if (flags & ~LKM_VALID_FLAGS) { 590 dlm_error(status); 591 goto error; 592 } 593 594 convert = (flags & LKM_CONVERT); 595 recovery = (flags & LKM_RECOVERY); 596 597 if (recovery && 598 (!dlm_is_recovery_lock(name, namelen) || convert) ) { 599 dlm_error(status); 600 goto error; 601 } 602 if (convert && (flags & LKM_LOCAL)) { 603 mlog(ML_ERROR, "strange LOCAL convert request!\n"); 604 goto error; 605 } 606 607 if (convert) { 608 /* CONVERT request */ 609 610 /* if converting, must pass in a valid dlm_lock */ 611 lock = lksb->lockid; 612 if (!lock) { 613 mlog(ML_ERROR, "NULL lock pointer in convert " 614 "request\n"); 615 goto error; 616 } 617 618 res = lock->lockres; 619 if (!res) { 620 mlog(ML_ERROR, "NULL lockres pointer in convert " 621 "request\n"); 622 goto error; 623 } 624 dlm_lockres_get(res); 625 626 /* XXX: for ocfs2 purposes, the ast/bast/astdata/lksb are 627 * static after the original lock call. convert requests will 628 * ensure that everything is the same, or return DLM_BADARGS. 629 * this means that DLM_DENIED_NOASTS will never be returned. 630 */ 631 if (lock->lksb != lksb || lock->ast != ast || 632 lock->bast != bast || lock->astdata != data) { 633 status = DLM_BADARGS; 634 mlog(ML_ERROR, "new args: lksb=%p, ast=%p, bast=%p, " 635 "astdata=%p\n", lksb, ast, bast, data); 636 mlog(ML_ERROR, "orig args: lksb=%p, ast=%p, bast=%p, " 637 "astdata=%p\n", lock->lksb, lock->ast, 638 lock->bast, lock->astdata); 639 goto error; 640 } 641 retry_convert: 642 dlm_wait_for_recovery(dlm); 643 644 if (res->owner == dlm->node_num) 645 status = dlmconvert_master(dlm, res, lock, flags, mode); 646 else 647 status = dlmconvert_remote(dlm, res, lock, flags, mode); 648 if (status == DLM_RECOVERING || status == DLM_MIGRATING || 649 status == DLM_FORWARD) { 650 /* for now, see how this works without sleeping 651 * and just retry right away. I suspect the reco 652 * or migration will complete fast enough that 653 * no waiting will be necessary */ 654 mlog(0, "retrying convert with migration/recovery/" 655 "in-progress\n"); 656 msleep(100); 657 goto retry_convert; 658 } 659 } else { 660 u64 tmpcookie; 661 662 /* LOCK request */ 663 status = DLM_BADARGS; 664 if (!name) { 665 dlm_error(status); 666 goto error; 667 } 668 669 status = DLM_IVBUFLEN; 670 if (namelen > DLM_LOCKID_NAME_MAX || namelen < 1) { 671 dlm_error(status); 672 goto error; 673 } 674 675 dlm_get_next_cookie(dlm->node_num, &tmpcookie); 676 lock = dlm_new_lock(mode, dlm->node_num, tmpcookie, lksb); 677 if (!lock) { 678 dlm_error(status); 679 goto error; 680 } 681 682 if (!recovery) 683 dlm_wait_for_recovery(dlm); 684 685 /* find or create the lock resource */ 686 res = dlm_get_lock_resource(dlm, name, namelen, flags); 687 if (!res) { 688 status = DLM_IVLOCKID; 689 dlm_error(status); 690 goto error; 691 } 692 693 mlog(0, "type=%d, flags = 0x%x\n", mode, flags); 694 mlog(0, "creating lock: lock=%p res=%p\n", lock, res); 695 696 dlm_lock_attach_lockres(lock, res); 697 lock->ast = ast; 698 lock->bast = bast; 699 lock->astdata = data; 700 701 retry_lock: 702 if (flags & LKM_VALBLK) { 703 mlog(0, "LKM_VALBLK passed by caller\n"); 704 705 /* LVB requests for non PR, PW or EX locks are 706 * ignored. */ 707 if (mode < LKM_PRMODE) 708 flags &= ~LKM_VALBLK; 709 else { 710 flags |= LKM_GET_LVB; 711 lock->lksb->flags |= DLM_LKSB_GET_LVB; 712 } 713 } 714 715 if (res->owner == dlm->node_num) 716 status = dlmlock_master(dlm, res, lock, flags); 717 else 718 status = dlmlock_remote(dlm, res, lock, flags); 719 720 if (status == DLM_RECOVERING || status == DLM_MIGRATING || 721 status == DLM_FORWARD) { 722 mlog(0, "retrying lock with migration/" 723 "recovery/in progress\n"); 724 msleep(100); 725 /* no waiting for dlm_reco_thread */ 726 if (recovery) { 727 if (status != DLM_RECOVERING) 728 goto retry_lock; 729 730 mlog(0, "%s: got RECOVERING " 731 "for $RECOVERY lock, master " 732 "was %u\n", dlm->name, 733 res->owner); 734 /* wait to see the node go down, then 735 * drop down and allow the lockres to 736 * get cleaned up. need to remaster. */ 737 dlm_wait_for_node_death(dlm, res->owner, 738 DLM_NODE_DEATH_WAIT_MAX); 739 } else { 740 dlm_wait_for_recovery(dlm); 741 goto retry_lock; 742 } 743 } 744 745 if (status != DLM_NORMAL) { 746 lock->lksb->flags &= ~DLM_LKSB_GET_LVB; 747 if (status != DLM_NOTQUEUED) 748 dlm_error(status); 749 goto error; 750 } 751 } 752 753 error: 754 if (status != DLM_NORMAL) { 755 if (lock && !convert) 756 dlm_lock_put(lock); 757 // this is kind of unnecessary 758 lksb->status = status; 759 } 760 761 /* put lockres ref from the convert path 762 * or from dlm_get_lock_resource */ 763 if (res) 764 dlm_lockres_put(res); 765 766 return status; 767 } 768 EXPORT_SYMBOL_GPL(dlmlock); 769