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