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