1 /****************************************************************************** 2 ******************************************************************************* 3 ** 4 ** Copyright (C) 2005-2010 Red Hat, Inc. All rights reserved. 5 ** 6 ** This copyrighted material is made available to anyone wishing to use, 7 ** modify, copy, or redistribute it subject to the terms and conditions 8 ** of the GNU General Public License v.2. 9 ** 10 ******************************************************************************* 11 ******************************************************************************/ 12 13 /* Central locking logic has four stages: 14 15 dlm_lock() 16 dlm_unlock() 17 18 request_lock(ls, lkb) 19 convert_lock(ls, lkb) 20 unlock_lock(ls, lkb) 21 cancel_lock(ls, lkb) 22 23 _request_lock(r, lkb) 24 _convert_lock(r, lkb) 25 _unlock_lock(r, lkb) 26 _cancel_lock(r, lkb) 27 28 do_request(r, lkb) 29 do_convert(r, lkb) 30 do_unlock(r, lkb) 31 do_cancel(r, lkb) 32 33 Stage 1 (lock, unlock) is mainly about checking input args and 34 splitting into one of the four main operations: 35 36 dlm_lock = request_lock 37 dlm_lock+CONVERT = convert_lock 38 dlm_unlock = unlock_lock 39 dlm_unlock+CANCEL = cancel_lock 40 41 Stage 2, xxxx_lock(), just finds and locks the relevant rsb which is 42 provided to the next stage. 43 44 Stage 3, _xxxx_lock(), determines if the operation is local or remote. 45 When remote, it calls send_xxxx(), when local it calls do_xxxx(). 46 47 Stage 4, do_xxxx(), is the guts of the operation. It manipulates the 48 given rsb and lkb and queues callbacks. 49 50 For remote operations, send_xxxx() results in the corresponding do_xxxx() 51 function being executed on the remote node. The connecting send/receive 52 calls on local (L) and remote (R) nodes: 53 54 L: send_xxxx() -> R: receive_xxxx() 55 R: do_xxxx() 56 L: receive_xxxx_reply() <- R: send_xxxx_reply() 57 */ 58 #include <linux/types.h> 59 #include <linux/slab.h> 60 #include "dlm_internal.h" 61 #include <linux/dlm_device.h> 62 #include "memory.h" 63 #include "lowcomms.h" 64 #include "requestqueue.h" 65 #include "util.h" 66 #include "dir.h" 67 #include "member.h" 68 #include "lockspace.h" 69 #include "ast.h" 70 #include "lock.h" 71 #include "rcom.h" 72 #include "recover.h" 73 #include "lvb_table.h" 74 #include "user.h" 75 #include "config.h" 76 77 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb); 78 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb); 79 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb); 80 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb); 81 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb); 82 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode); 83 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb); 84 static int send_remove(struct dlm_rsb *r); 85 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb); 86 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb); 87 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, 88 struct dlm_message *ms); 89 static int receive_extralen(struct dlm_message *ms); 90 static void do_purge(struct dlm_ls *ls, int nodeid, int pid); 91 static void del_timeout(struct dlm_lkb *lkb); 92 93 /* 94 * Lock compatibilty matrix - thanks Steve 95 * UN = Unlocked state. Not really a state, used as a flag 96 * PD = Padding. Used to make the matrix a nice power of two in size 97 * Other states are the same as the VMS DLM. 98 * Usage: matrix[grmode+1][rqmode+1] (although m[rq+1][gr+1] is the same) 99 */ 100 101 static const int __dlm_compat_matrix[8][8] = { 102 /* UN NL CR CW PR PW EX PD */ 103 {1, 1, 1, 1, 1, 1, 1, 0}, /* UN */ 104 {1, 1, 1, 1, 1, 1, 1, 0}, /* NL */ 105 {1, 1, 1, 1, 1, 1, 0, 0}, /* CR */ 106 {1, 1, 1, 1, 0, 0, 0, 0}, /* CW */ 107 {1, 1, 1, 0, 1, 0, 0, 0}, /* PR */ 108 {1, 1, 1, 0, 0, 0, 0, 0}, /* PW */ 109 {1, 1, 0, 0, 0, 0, 0, 0}, /* EX */ 110 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */ 111 }; 112 113 /* 114 * This defines the direction of transfer of LVB data. 115 * Granted mode is the row; requested mode is the column. 116 * Usage: matrix[grmode+1][rqmode+1] 117 * 1 = LVB is returned to the caller 118 * 0 = LVB is written to the resource 119 * -1 = nothing happens to the LVB 120 */ 121 122 const int dlm_lvb_operations[8][8] = { 123 /* UN NL CR CW PR PW EX PD*/ 124 { -1, 1, 1, 1, 1, 1, 1, -1 }, /* UN */ 125 { -1, 1, 1, 1, 1, 1, 1, 0 }, /* NL */ 126 { -1, -1, 1, 1, 1, 1, 1, 0 }, /* CR */ 127 { -1, -1, -1, 1, 1, 1, 1, 0 }, /* CW */ 128 { -1, -1, -1, -1, 1, 1, 1, 0 }, /* PR */ 129 { -1, 0, 0, 0, 0, 0, 1, 0 }, /* PW */ 130 { -1, 0, 0, 0, 0, 0, 0, 0 }, /* EX */ 131 { -1, 0, 0, 0, 0, 0, 0, 0 } /* PD */ 132 }; 133 134 #define modes_compat(gr, rq) \ 135 __dlm_compat_matrix[(gr)->lkb_grmode + 1][(rq)->lkb_rqmode + 1] 136 137 int dlm_modes_compat(int mode1, int mode2) 138 { 139 return __dlm_compat_matrix[mode1 + 1][mode2 + 1]; 140 } 141 142 /* 143 * Compatibility matrix for conversions with QUECVT set. 144 * Granted mode is the row; requested mode is the column. 145 * Usage: matrix[grmode+1][rqmode+1] 146 */ 147 148 static const int __quecvt_compat_matrix[8][8] = { 149 /* UN NL CR CW PR PW EX PD */ 150 {0, 0, 0, 0, 0, 0, 0, 0}, /* UN */ 151 {0, 0, 1, 1, 1, 1, 1, 0}, /* NL */ 152 {0, 0, 0, 1, 1, 1, 1, 0}, /* CR */ 153 {0, 0, 0, 0, 1, 1, 1, 0}, /* CW */ 154 {0, 0, 0, 1, 0, 1, 1, 0}, /* PR */ 155 {0, 0, 0, 0, 0, 0, 1, 0}, /* PW */ 156 {0, 0, 0, 0, 0, 0, 0, 0}, /* EX */ 157 {0, 0, 0, 0, 0, 0, 0, 0} /* PD */ 158 }; 159 160 void dlm_print_lkb(struct dlm_lkb *lkb) 161 { 162 printk(KERN_ERR "lkb: nodeid %d id %x remid %x exflags %x flags %x\n" 163 " status %d rqmode %d grmode %d wait_type %d\n", 164 lkb->lkb_nodeid, lkb->lkb_id, lkb->lkb_remid, lkb->lkb_exflags, 165 lkb->lkb_flags, lkb->lkb_status, lkb->lkb_rqmode, 166 lkb->lkb_grmode, lkb->lkb_wait_type); 167 } 168 169 static void dlm_print_rsb(struct dlm_rsb *r) 170 { 171 printk(KERN_ERR "rsb: nodeid %d flags %lx first %x rlc %d name %s\n", 172 r->res_nodeid, r->res_flags, r->res_first_lkid, 173 r->res_recover_locks_count, r->res_name); 174 } 175 176 void dlm_dump_rsb(struct dlm_rsb *r) 177 { 178 struct dlm_lkb *lkb; 179 180 dlm_print_rsb(r); 181 182 printk(KERN_ERR "rsb: root_list empty %d recover_list empty %d\n", 183 list_empty(&r->res_root_list), list_empty(&r->res_recover_list)); 184 printk(KERN_ERR "rsb lookup list\n"); 185 list_for_each_entry(lkb, &r->res_lookup, lkb_rsb_lookup) 186 dlm_print_lkb(lkb); 187 printk(KERN_ERR "rsb grant queue:\n"); 188 list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue) 189 dlm_print_lkb(lkb); 190 printk(KERN_ERR "rsb convert queue:\n"); 191 list_for_each_entry(lkb, &r->res_convertqueue, lkb_statequeue) 192 dlm_print_lkb(lkb); 193 printk(KERN_ERR "rsb wait queue:\n"); 194 list_for_each_entry(lkb, &r->res_waitqueue, lkb_statequeue) 195 dlm_print_lkb(lkb); 196 } 197 198 /* Threads cannot use the lockspace while it's being recovered */ 199 200 static inline void dlm_lock_recovery(struct dlm_ls *ls) 201 { 202 down_read(&ls->ls_in_recovery); 203 } 204 205 void dlm_unlock_recovery(struct dlm_ls *ls) 206 { 207 up_read(&ls->ls_in_recovery); 208 } 209 210 int dlm_lock_recovery_try(struct dlm_ls *ls) 211 { 212 return down_read_trylock(&ls->ls_in_recovery); 213 } 214 215 static inline int can_be_queued(struct dlm_lkb *lkb) 216 { 217 return !(lkb->lkb_exflags & DLM_LKF_NOQUEUE); 218 } 219 220 static inline int force_blocking_asts(struct dlm_lkb *lkb) 221 { 222 return (lkb->lkb_exflags & DLM_LKF_NOQUEUEBAST); 223 } 224 225 static inline int is_demoted(struct dlm_lkb *lkb) 226 { 227 return (lkb->lkb_sbflags & DLM_SBF_DEMOTED); 228 } 229 230 static inline int is_altmode(struct dlm_lkb *lkb) 231 { 232 return (lkb->lkb_sbflags & DLM_SBF_ALTMODE); 233 } 234 235 static inline int is_granted(struct dlm_lkb *lkb) 236 { 237 return (lkb->lkb_status == DLM_LKSTS_GRANTED); 238 } 239 240 static inline int is_remote(struct dlm_rsb *r) 241 { 242 DLM_ASSERT(r->res_nodeid >= 0, dlm_print_rsb(r);); 243 return !!r->res_nodeid; 244 } 245 246 static inline int is_process_copy(struct dlm_lkb *lkb) 247 { 248 return (lkb->lkb_nodeid && !(lkb->lkb_flags & DLM_IFL_MSTCPY)); 249 } 250 251 static inline int is_master_copy(struct dlm_lkb *lkb) 252 { 253 if (lkb->lkb_flags & DLM_IFL_MSTCPY) 254 DLM_ASSERT(lkb->lkb_nodeid, dlm_print_lkb(lkb);); 255 return (lkb->lkb_flags & DLM_IFL_MSTCPY) ? 1 : 0; 256 } 257 258 static inline int middle_conversion(struct dlm_lkb *lkb) 259 { 260 if ((lkb->lkb_grmode==DLM_LOCK_PR && lkb->lkb_rqmode==DLM_LOCK_CW) || 261 (lkb->lkb_rqmode==DLM_LOCK_PR && lkb->lkb_grmode==DLM_LOCK_CW)) 262 return 1; 263 return 0; 264 } 265 266 static inline int down_conversion(struct dlm_lkb *lkb) 267 { 268 return (!middle_conversion(lkb) && lkb->lkb_rqmode < lkb->lkb_grmode); 269 } 270 271 static inline int is_overlap_unlock(struct dlm_lkb *lkb) 272 { 273 return lkb->lkb_flags & DLM_IFL_OVERLAP_UNLOCK; 274 } 275 276 static inline int is_overlap_cancel(struct dlm_lkb *lkb) 277 { 278 return lkb->lkb_flags & DLM_IFL_OVERLAP_CANCEL; 279 } 280 281 static inline int is_overlap(struct dlm_lkb *lkb) 282 { 283 return (lkb->lkb_flags & (DLM_IFL_OVERLAP_UNLOCK | 284 DLM_IFL_OVERLAP_CANCEL)); 285 } 286 287 static void queue_cast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv) 288 { 289 if (is_master_copy(lkb)) 290 return; 291 292 del_timeout(lkb); 293 294 DLM_ASSERT(lkb->lkb_lksb, dlm_print_lkb(lkb);); 295 296 /* if the operation was a cancel, then return -DLM_ECANCEL, if a 297 timeout caused the cancel then return -ETIMEDOUT */ 298 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_TIMEOUT_CANCEL)) { 299 lkb->lkb_flags &= ~DLM_IFL_TIMEOUT_CANCEL; 300 rv = -ETIMEDOUT; 301 } 302 303 if (rv == -DLM_ECANCEL && (lkb->lkb_flags & DLM_IFL_DEADLOCK_CANCEL)) { 304 lkb->lkb_flags &= ~DLM_IFL_DEADLOCK_CANCEL; 305 rv = -EDEADLK; 306 } 307 308 dlm_add_cb(lkb, DLM_CB_CAST, lkb->lkb_grmode, rv, lkb->lkb_sbflags); 309 } 310 311 static inline void queue_cast_overlap(struct dlm_rsb *r, struct dlm_lkb *lkb) 312 { 313 queue_cast(r, lkb, 314 is_overlap_unlock(lkb) ? -DLM_EUNLOCK : -DLM_ECANCEL); 315 } 316 317 static void queue_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int rqmode) 318 { 319 if (is_master_copy(lkb)) { 320 send_bast(r, lkb, rqmode); 321 } else { 322 dlm_add_cb(lkb, DLM_CB_BAST, rqmode, 0, 0); 323 } 324 } 325 326 /* 327 * Basic operations on rsb's and lkb's 328 */ 329 330 static int pre_rsb_struct(struct dlm_ls *ls) 331 { 332 struct dlm_rsb *r1, *r2; 333 int count = 0; 334 335 spin_lock(&ls->ls_new_rsb_spin); 336 if (ls->ls_new_rsb_count > dlm_config.ci_new_rsb_count / 2) { 337 spin_unlock(&ls->ls_new_rsb_spin); 338 return 0; 339 } 340 spin_unlock(&ls->ls_new_rsb_spin); 341 342 r1 = dlm_allocate_rsb(ls); 343 r2 = dlm_allocate_rsb(ls); 344 345 spin_lock(&ls->ls_new_rsb_spin); 346 if (r1) { 347 list_add(&r1->res_hashchain, &ls->ls_new_rsb); 348 ls->ls_new_rsb_count++; 349 } 350 if (r2) { 351 list_add(&r2->res_hashchain, &ls->ls_new_rsb); 352 ls->ls_new_rsb_count++; 353 } 354 count = ls->ls_new_rsb_count; 355 spin_unlock(&ls->ls_new_rsb_spin); 356 357 if (!count) 358 return -ENOMEM; 359 return 0; 360 } 361 362 /* If ls->ls_new_rsb is empty, return -EAGAIN, so the caller can 363 unlock any spinlocks, go back and call pre_rsb_struct again. 364 Otherwise, take an rsb off the list and return it. */ 365 366 static int get_rsb_struct(struct dlm_ls *ls, char *name, int len, 367 struct dlm_rsb **r_ret) 368 { 369 struct dlm_rsb *r; 370 int count; 371 372 spin_lock(&ls->ls_new_rsb_spin); 373 if (list_empty(&ls->ls_new_rsb)) { 374 count = ls->ls_new_rsb_count; 375 spin_unlock(&ls->ls_new_rsb_spin); 376 log_debug(ls, "find_rsb retry %d %d %s", 377 count, dlm_config.ci_new_rsb_count, name); 378 return -EAGAIN; 379 } 380 381 r = list_first_entry(&ls->ls_new_rsb, struct dlm_rsb, res_hashchain); 382 list_del(&r->res_hashchain); 383 ls->ls_new_rsb_count--; 384 spin_unlock(&ls->ls_new_rsb_spin); 385 386 r->res_ls = ls; 387 r->res_length = len; 388 memcpy(r->res_name, name, len); 389 mutex_init(&r->res_mutex); 390 391 INIT_LIST_HEAD(&r->res_hashchain); 392 INIT_LIST_HEAD(&r->res_lookup); 393 INIT_LIST_HEAD(&r->res_grantqueue); 394 INIT_LIST_HEAD(&r->res_convertqueue); 395 INIT_LIST_HEAD(&r->res_waitqueue); 396 INIT_LIST_HEAD(&r->res_root_list); 397 INIT_LIST_HEAD(&r->res_recover_list); 398 399 *r_ret = r; 400 return 0; 401 } 402 403 static int search_rsb_list(struct list_head *head, char *name, int len, 404 unsigned int flags, struct dlm_rsb **r_ret) 405 { 406 struct dlm_rsb *r; 407 int error = 0; 408 409 list_for_each_entry(r, head, res_hashchain) { 410 if (len == r->res_length && !memcmp(name, r->res_name, len)) 411 goto found; 412 } 413 *r_ret = NULL; 414 return -EBADR; 415 416 found: 417 if (r->res_nodeid && (flags & R_MASTER)) 418 error = -ENOTBLK; 419 *r_ret = r; 420 return error; 421 } 422 423 static int _search_rsb(struct dlm_ls *ls, char *name, int len, int b, 424 unsigned int flags, struct dlm_rsb **r_ret) 425 { 426 struct dlm_rsb *r; 427 int error; 428 429 error = search_rsb_list(&ls->ls_rsbtbl[b].list, name, len, flags, &r); 430 if (!error) { 431 kref_get(&r->res_ref); 432 goto out; 433 } 434 error = search_rsb_list(&ls->ls_rsbtbl[b].toss, name, len, flags, &r); 435 if (error) 436 goto out; 437 438 list_move(&r->res_hashchain, &ls->ls_rsbtbl[b].list); 439 440 if (dlm_no_directory(ls)) 441 goto out; 442 443 if (r->res_nodeid == -1) { 444 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN); 445 r->res_first_lkid = 0; 446 } else if (r->res_nodeid > 0) { 447 rsb_set_flag(r, RSB_MASTER_UNCERTAIN); 448 r->res_first_lkid = 0; 449 } else { 450 DLM_ASSERT(r->res_nodeid == 0, dlm_print_rsb(r);); 451 DLM_ASSERT(!rsb_flag(r, RSB_MASTER_UNCERTAIN),); 452 } 453 out: 454 *r_ret = r; 455 return error; 456 } 457 458 /* 459 * Find rsb in rsbtbl and potentially create/add one 460 * 461 * Delaying the release of rsb's has a similar benefit to applications keeping 462 * NL locks on an rsb, but without the guarantee that the cached master value 463 * will still be valid when the rsb is reused. Apps aren't always smart enough 464 * to keep NL locks on an rsb that they may lock again shortly; this can lead 465 * to excessive master lookups and removals if we don't delay the release. 466 * 467 * Searching for an rsb means looking through both the normal list and toss 468 * list. When found on the toss list the rsb is moved to the normal list with 469 * ref count of 1; when found on normal list the ref count is incremented. 470 */ 471 472 static int find_rsb(struct dlm_ls *ls, char *name, int namelen, 473 unsigned int flags, struct dlm_rsb **r_ret) 474 { 475 struct dlm_rsb *r = NULL; 476 uint32_t hash, bucket; 477 int error; 478 479 if (namelen > DLM_RESNAME_MAXLEN) { 480 error = -EINVAL; 481 goto out; 482 } 483 484 if (dlm_no_directory(ls)) 485 flags |= R_CREATE; 486 487 hash = jhash(name, namelen, 0); 488 bucket = hash & (ls->ls_rsbtbl_size - 1); 489 490 retry: 491 if (flags & R_CREATE) { 492 error = pre_rsb_struct(ls); 493 if (error < 0) 494 goto out; 495 } 496 497 spin_lock(&ls->ls_rsbtbl[bucket].lock); 498 499 error = _search_rsb(ls, name, namelen, bucket, flags, &r); 500 if (!error) 501 goto out_unlock; 502 503 if (error == -EBADR && !(flags & R_CREATE)) 504 goto out_unlock; 505 506 /* the rsb was found but wasn't a master copy */ 507 if (error == -ENOTBLK) 508 goto out_unlock; 509 510 error = get_rsb_struct(ls, name, namelen, &r); 511 if (error == -EAGAIN) { 512 spin_unlock(&ls->ls_rsbtbl[bucket].lock); 513 goto retry; 514 } 515 if (error) 516 goto out_unlock; 517 518 r->res_hash = hash; 519 r->res_bucket = bucket; 520 r->res_nodeid = -1; 521 kref_init(&r->res_ref); 522 523 /* With no directory, the master can be set immediately */ 524 if (dlm_no_directory(ls)) { 525 int nodeid = dlm_dir_nodeid(r); 526 if (nodeid == dlm_our_nodeid()) 527 nodeid = 0; 528 r->res_nodeid = nodeid; 529 } 530 list_add(&r->res_hashchain, &ls->ls_rsbtbl[bucket].list); 531 error = 0; 532 out_unlock: 533 spin_unlock(&ls->ls_rsbtbl[bucket].lock); 534 out: 535 *r_ret = r; 536 return error; 537 } 538 539 /* This is only called to add a reference when the code already holds 540 a valid reference to the rsb, so there's no need for locking. */ 541 542 static inline void hold_rsb(struct dlm_rsb *r) 543 { 544 kref_get(&r->res_ref); 545 } 546 547 void dlm_hold_rsb(struct dlm_rsb *r) 548 { 549 hold_rsb(r); 550 } 551 552 static void toss_rsb(struct kref *kref) 553 { 554 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref); 555 struct dlm_ls *ls = r->res_ls; 556 557 DLM_ASSERT(list_empty(&r->res_root_list), dlm_print_rsb(r);); 558 kref_init(&r->res_ref); 559 list_move(&r->res_hashchain, &ls->ls_rsbtbl[r->res_bucket].toss); 560 r->res_toss_time = jiffies; 561 if (r->res_lvbptr) { 562 dlm_free_lvb(r->res_lvbptr); 563 r->res_lvbptr = NULL; 564 } 565 } 566 567 /* When all references to the rsb are gone it's transferred to 568 the tossed list for later disposal. */ 569 570 static void put_rsb(struct dlm_rsb *r) 571 { 572 struct dlm_ls *ls = r->res_ls; 573 uint32_t bucket = r->res_bucket; 574 575 spin_lock(&ls->ls_rsbtbl[bucket].lock); 576 kref_put(&r->res_ref, toss_rsb); 577 spin_unlock(&ls->ls_rsbtbl[bucket].lock); 578 } 579 580 void dlm_put_rsb(struct dlm_rsb *r) 581 { 582 put_rsb(r); 583 } 584 585 /* See comment for unhold_lkb */ 586 587 static void unhold_rsb(struct dlm_rsb *r) 588 { 589 int rv; 590 rv = kref_put(&r->res_ref, toss_rsb); 591 DLM_ASSERT(!rv, dlm_dump_rsb(r);); 592 } 593 594 static void kill_rsb(struct kref *kref) 595 { 596 struct dlm_rsb *r = container_of(kref, struct dlm_rsb, res_ref); 597 598 /* All work is done after the return from kref_put() so we 599 can release the write_lock before the remove and free. */ 600 601 DLM_ASSERT(list_empty(&r->res_lookup), dlm_dump_rsb(r);); 602 DLM_ASSERT(list_empty(&r->res_grantqueue), dlm_dump_rsb(r);); 603 DLM_ASSERT(list_empty(&r->res_convertqueue), dlm_dump_rsb(r);); 604 DLM_ASSERT(list_empty(&r->res_waitqueue), dlm_dump_rsb(r);); 605 DLM_ASSERT(list_empty(&r->res_root_list), dlm_dump_rsb(r);); 606 DLM_ASSERT(list_empty(&r->res_recover_list), dlm_dump_rsb(r);); 607 } 608 609 /* Attaching/detaching lkb's from rsb's is for rsb reference counting. 610 The rsb must exist as long as any lkb's for it do. */ 611 612 static void attach_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb) 613 { 614 hold_rsb(r); 615 lkb->lkb_resource = r; 616 } 617 618 static void detach_lkb(struct dlm_lkb *lkb) 619 { 620 if (lkb->lkb_resource) { 621 put_rsb(lkb->lkb_resource); 622 lkb->lkb_resource = NULL; 623 } 624 } 625 626 static int create_lkb(struct dlm_ls *ls, struct dlm_lkb **lkb_ret) 627 { 628 struct dlm_lkb *lkb; 629 int rv, id; 630 631 lkb = dlm_allocate_lkb(ls); 632 if (!lkb) 633 return -ENOMEM; 634 635 lkb->lkb_nodeid = -1; 636 lkb->lkb_grmode = DLM_LOCK_IV; 637 kref_init(&lkb->lkb_ref); 638 INIT_LIST_HEAD(&lkb->lkb_ownqueue); 639 INIT_LIST_HEAD(&lkb->lkb_rsb_lookup); 640 INIT_LIST_HEAD(&lkb->lkb_time_list); 641 INIT_LIST_HEAD(&lkb->lkb_cb_list); 642 mutex_init(&lkb->lkb_cb_mutex); 643 INIT_WORK(&lkb->lkb_cb_work, dlm_callback_work); 644 645 retry: 646 rv = idr_pre_get(&ls->ls_lkbidr, GFP_NOFS); 647 if (!rv) 648 return -ENOMEM; 649 650 spin_lock(&ls->ls_lkbidr_spin); 651 rv = idr_get_new_above(&ls->ls_lkbidr, lkb, 1, &id); 652 if (!rv) 653 lkb->lkb_id = id; 654 spin_unlock(&ls->ls_lkbidr_spin); 655 656 if (rv == -EAGAIN) 657 goto retry; 658 659 if (rv < 0) { 660 log_error(ls, "create_lkb idr error %d", rv); 661 return rv; 662 } 663 664 *lkb_ret = lkb; 665 return 0; 666 } 667 668 static int find_lkb(struct dlm_ls *ls, uint32_t lkid, struct dlm_lkb **lkb_ret) 669 { 670 struct dlm_lkb *lkb; 671 672 spin_lock(&ls->ls_lkbidr_spin); 673 lkb = idr_find(&ls->ls_lkbidr, lkid); 674 if (lkb) 675 kref_get(&lkb->lkb_ref); 676 spin_unlock(&ls->ls_lkbidr_spin); 677 678 *lkb_ret = lkb; 679 return lkb ? 0 : -ENOENT; 680 } 681 682 static void kill_lkb(struct kref *kref) 683 { 684 struct dlm_lkb *lkb = container_of(kref, struct dlm_lkb, lkb_ref); 685 686 /* All work is done after the return from kref_put() so we 687 can release the write_lock before the detach_lkb */ 688 689 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb);); 690 } 691 692 /* __put_lkb() is used when an lkb may not have an rsb attached to 693 it so we need to provide the lockspace explicitly */ 694 695 static int __put_lkb(struct dlm_ls *ls, struct dlm_lkb *lkb) 696 { 697 uint32_t lkid = lkb->lkb_id; 698 699 spin_lock(&ls->ls_lkbidr_spin); 700 if (kref_put(&lkb->lkb_ref, kill_lkb)) { 701 idr_remove(&ls->ls_lkbidr, lkid); 702 spin_unlock(&ls->ls_lkbidr_spin); 703 704 detach_lkb(lkb); 705 706 /* for local/process lkbs, lvbptr points to caller's lksb */ 707 if (lkb->lkb_lvbptr && is_master_copy(lkb)) 708 dlm_free_lvb(lkb->lkb_lvbptr); 709 dlm_free_lkb(lkb); 710 return 1; 711 } else { 712 spin_unlock(&ls->ls_lkbidr_spin); 713 return 0; 714 } 715 } 716 717 int dlm_put_lkb(struct dlm_lkb *lkb) 718 { 719 struct dlm_ls *ls; 720 721 DLM_ASSERT(lkb->lkb_resource, dlm_print_lkb(lkb);); 722 DLM_ASSERT(lkb->lkb_resource->res_ls, dlm_print_lkb(lkb);); 723 724 ls = lkb->lkb_resource->res_ls; 725 return __put_lkb(ls, lkb); 726 } 727 728 /* This is only called to add a reference when the code already holds 729 a valid reference to the lkb, so there's no need for locking. */ 730 731 static inline void hold_lkb(struct dlm_lkb *lkb) 732 { 733 kref_get(&lkb->lkb_ref); 734 } 735 736 /* This is called when we need to remove a reference and are certain 737 it's not the last ref. e.g. del_lkb is always called between a 738 find_lkb/put_lkb and is always the inverse of a previous add_lkb. 739 put_lkb would work fine, but would involve unnecessary locking */ 740 741 static inline void unhold_lkb(struct dlm_lkb *lkb) 742 { 743 int rv; 744 rv = kref_put(&lkb->lkb_ref, kill_lkb); 745 DLM_ASSERT(!rv, dlm_print_lkb(lkb);); 746 } 747 748 static void lkb_add_ordered(struct list_head *new, struct list_head *head, 749 int mode) 750 { 751 struct dlm_lkb *lkb = NULL; 752 753 list_for_each_entry(lkb, head, lkb_statequeue) 754 if (lkb->lkb_rqmode < mode) 755 break; 756 757 __list_add(new, lkb->lkb_statequeue.prev, &lkb->lkb_statequeue); 758 } 759 760 /* add/remove lkb to rsb's grant/convert/wait queue */ 761 762 static void add_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int status) 763 { 764 kref_get(&lkb->lkb_ref); 765 766 DLM_ASSERT(!lkb->lkb_status, dlm_print_lkb(lkb);); 767 768 lkb->lkb_timestamp = ktime_get(); 769 770 lkb->lkb_status = status; 771 772 switch (status) { 773 case DLM_LKSTS_WAITING: 774 if (lkb->lkb_exflags & DLM_LKF_HEADQUE) 775 list_add(&lkb->lkb_statequeue, &r->res_waitqueue); 776 else 777 list_add_tail(&lkb->lkb_statequeue, &r->res_waitqueue); 778 break; 779 case DLM_LKSTS_GRANTED: 780 /* convention says granted locks kept in order of grmode */ 781 lkb_add_ordered(&lkb->lkb_statequeue, &r->res_grantqueue, 782 lkb->lkb_grmode); 783 break; 784 case DLM_LKSTS_CONVERT: 785 if (lkb->lkb_exflags & DLM_LKF_HEADQUE) 786 list_add(&lkb->lkb_statequeue, &r->res_convertqueue); 787 else 788 list_add_tail(&lkb->lkb_statequeue, 789 &r->res_convertqueue); 790 break; 791 default: 792 DLM_ASSERT(0, dlm_print_lkb(lkb); printk("sts=%d\n", status);); 793 } 794 } 795 796 static void del_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb) 797 { 798 lkb->lkb_status = 0; 799 list_del(&lkb->lkb_statequeue); 800 unhold_lkb(lkb); 801 } 802 803 static void move_lkb(struct dlm_rsb *r, struct dlm_lkb *lkb, int sts) 804 { 805 hold_lkb(lkb); 806 del_lkb(r, lkb); 807 add_lkb(r, lkb, sts); 808 unhold_lkb(lkb); 809 } 810 811 static int msg_reply_type(int mstype) 812 { 813 switch (mstype) { 814 case DLM_MSG_REQUEST: 815 return DLM_MSG_REQUEST_REPLY; 816 case DLM_MSG_CONVERT: 817 return DLM_MSG_CONVERT_REPLY; 818 case DLM_MSG_UNLOCK: 819 return DLM_MSG_UNLOCK_REPLY; 820 case DLM_MSG_CANCEL: 821 return DLM_MSG_CANCEL_REPLY; 822 case DLM_MSG_LOOKUP: 823 return DLM_MSG_LOOKUP_REPLY; 824 } 825 return -1; 826 } 827 828 static int nodeid_warned(int nodeid, int num_nodes, int *warned) 829 { 830 int i; 831 832 for (i = 0; i < num_nodes; i++) { 833 if (!warned[i]) { 834 warned[i] = nodeid; 835 return 0; 836 } 837 if (warned[i] == nodeid) 838 return 1; 839 } 840 return 0; 841 } 842 843 void dlm_scan_waiters(struct dlm_ls *ls) 844 { 845 struct dlm_lkb *lkb; 846 ktime_t zero = ktime_set(0, 0); 847 s64 us; 848 s64 debug_maxus = 0; 849 u32 debug_scanned = 0; 850 u32 debug_expired = 0; 851 int num_nodes = 0; 852 int *warned = NULL; 853 854 if (!dlm_config.ci_waitwarn_us) 855 return; 856 857 mutex_lock(&ls->ls_waiters_mutex); 858 859 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) { 860 if (ktime_equal(lkb->lkb_wait_time, zero)) 861 continue; 862 863 debug_scanned++; 864 865 us = ktime_to_us(ktime_sub(ktime_get(), lkb->lkb_wait_time)); 866 867 if (us < dlm_config.ci_waitwarn_us) 868 continue; 869 870 lkb->lkb_wait_time = zero; 871 872 debug_expired++; 873 if (us > debug_maxus) 874 debug_maxus = us; 875 876 if (!num_nodes) { 877 num_nodes = ls->ls_num_nodes; 878 warned = kzalloc(num_nodes * sizeof(int), GFP_KERNEL); 879 } 880 if (!warned) 881 continue; 882 if (nodeid_warned(lkb->lkb_wait_nodeid, num_nodes, warned)) 883 continue; 884 885 log_error(ls, "waitwarn %x %lld %d us check connection to " 886 "node %d", lkb->lkb_id, (long long)us, 887 dlm_config.ci_waitwarn_us, lkb->lkb_wait_nodeid); 888 } 889 mutex_unlock(&ls->ls_waiters_mutex); 890 kfree(warned); 891 892 if (debug_expired) 893 log_debug(ls, "scan_waiters %u warn %u over %d us max %lld us", 894 debug_scanned, debug_expired, 895 dlm_config.ci_waitwarn_us, (long long)debug_maxus); 896 } 897 898 /* add/remove lkb from global waiters list of lkb's waiting for 899 a reply from a remote node */ 900 901 static int add_to_waiters(struct dlm_lkb *lkb, int mstype, int to_nodeid) 902 { 903 struct dlm_ls *ls = lkb->lkb_resource->res_ls; 904 int error = 0; 905 906 mutex_lock(&ls->ls_waiters_mutex); 907 908 if (is_overlap_unlock(lkb) || 909 (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL))) { 910 error = -EINVAL; 911 goto out; 912 } 913 914 if (lkb->lkb_wait_type || is_overlap_cancel(lkb)) { 915 switch (mstype) { 916 case DLM_MSG_UNLOCK: 917 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK; 918 break; 919 case DLM_MSG_CANCEL: 920 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL; 921 break; 922 default: 923 error = -EBUSY; 924 goto out; 925 } 926 lkb->lkb_wait_count++; 927 hold_lkb(lkb); 928 929 log_debug(ls, "addwait %x cur %d overlap %d count %d f %x", 930 lkb->lkb_id, lkb->lkb_wait_type, mstype, 931 lkb->lkb_wait_count, lkb->lkb_flags); 932 goto out; 933 } 934 935 DLM_ASSERT(!lkb->lkb_wait_count, 936 dlm_print_lkb(lkb); 937 printk("wait_count %d\n", lkb->lkb_wait_count);); 938 939 lkb->lkb_wait_count++; 940 lkb->lkb_wait_type = mstype; 941 lkb->lkb_wait_time = ktime_get(); 942 lkb->lkb_wait_nodeid = to_nodeid; /* for debugging */ 943 hold_lkb(lkb); 944 list_add(&lkb->lkb_wait_reply, &ls->ls_waiters); 945 out: 946 if (error) 947 log_error(ls, "addwait error %x %d flags %x %d %d %s", 948 lkb->lkb_id, error, lkb->lkb_flags, mstype, 949 lkb->lkb_wait_type, lkb->lkb_resource->res_name); 950 mutex_unlock(&ls->ls_waiters_mutex); 951 return error; 952 } 953 954 /* We clear the RESEND flag because we might be taking an lkb off the waiters 955 list as part of process_requestqueue (e.g. a lookup that has an optimized 956 request reply on the requestqueue) between dlm_recover_waiters_pre() which 957 set RESEND and dlm_recover_waiters_post() */ 958 959 static int _remove_from_waiters(struct dlm_lkb *lkb, int mstype, 960 struct dlm_message *ms) 961 { 962 struct dlm_ls *ls = lkb->lkb_resource->res_ls; 963 int overlap_done = 0; 964 965 if (is_overlap_unlock(lkb) && (mstype == DLM_MSG_UNLOCK_REPLY)) { 966 log_debug(ls, "remwait %x unlock_reply overlap", lkb->lkb_id); 967 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK; 968 overlap_done = 1; 969 goto out_del; 970 } 971 972 if (is_overlap_cancel(lkb) && (mstype == DLM_MSG_CANCEL_REPLY)) { 973 log_debug(ls, "remwait %x cancel_reply overlap", lkb->lkb_id); 974 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL; 975 overlap_done = 1; 976 goto out_del; 977 } 978 979 /* Cancel state was preemptively cleared by a successful convert, 980 see next comment, nothing to do. */ 981 982 if ((mstype == DLM_MSG_CANCEL_REPLY) && 983 (lkb->lkb_wait_type != DLM_MSG_CANCEL)) { 984 log_debug(ls, "remwait %x cancel_reply wait_type %d", 985 lkb->lkb_id, lkb->lkb_wait_type); 986 return -1; 987 } 988 989 /* Remove for the convert reply, and premptively remove for the 990 cancel reply. A convert has been granted while there's still 991 an outstanding cancel on it (the cancel is moot and the result 992 in the cancel reply should be 0). We preempt the cancel reply 993 because the app gets the convert result and then can follow up 994 with another op, like convert. This subsequent op would see the 995 lingering state of the cancel and fail with -EBUSY. */ 996 997 if ((mstype == DLM_MSG_CONVERT_REPLY) && 998 (lkb->lkb_wait_type == DLM_MSG_CONVERT) && 999 is_overlap_cancel(lkb) && ms && !ms->m_result) { 1000 log_debug(ls, "remwait %x convert_reply zap overlap_cancel", 1001 lkb->lkb_id); 1002 lkb->lkb_wait_type = 0; 1003 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL; 1004 lkb->lkb_wait_count--; 1005 goto out_del; 1006 } 1007 1008 /* N.B. type of reply may not always correspond to type of original 1009 msg due to lookup->request optimization, verify others? */ 1010 1011 if (lkb->lkb_wait_type) { 1012 lkb->lkb_wait_type = 0; 1013 goto out_del; 1014 } 1015 1016 log_error(ls, "remwait error %x reply %d flags %x no wait_type", 1017 lkb->lkb_id, mstype, lkb->lkb_flags); 1018 return -1; 1019 1020 out_del: 1021 /* the force-unlock/cancel has completed and we haven't recvd a reply 1022 to the op that was in progress prior to the unlock/cancel; we 1023 give up on any reply to the earlier op. FIXME: not sure when/how 1024 this would happen */ 1025 1026 if (overlap_done && lkb->lkb_wait_type) { 1027 log_error(ls, "remwait error %x reply %d wait_type %d overlap", 1028 lkb->lkb_id, mstype, lkb->lkb_wait_type); 1029 lkb->lkb_wait_count--; 1030 lkb->lkb_wait_type = 0; 1031 } 1032 1033 DLM_ASSERT(lkb->lkb_wait_count, dlm_print_lkb(lkb);); 1034 1035 lkb->lkb_flags &= ~DLM_IFL_RESEND; 1036 lkb->lkb_wait_count--; 1037 if (!lkb->lkb_wait_count) 1038 list_del_init(&lkb->lkb_wait_reply); 1039 unhold_lkb(lkb); 1040 return 0; 1041 } 1042 1043 static int remove_from_waiters(struct dlm_lkb *lkb, int mstype) 1044 { 1045 struct dlm_ls *ls = lkb->lkb_resource->res_ls; 1046 int error; 1047 1048 mutex_lock(&ls->ls_waiters_mutex); 1049 error = _remove_from_waiters(lkb, mstype, NULL); 1050 mutex_unlock(&ls->ls_waiters_mutex); 1051 return error; 1052 } 1053 1054 /* Handles situations where we might be processing a "fake" or "stub" reply in 1055 which we can't try to take waiters_mutex again. */ 1056 1057 static int remove_from_waiters_ms(struct dlm_lkb *lkb, struct dlm_message *ms) 1058 { 1059 struct dlm_ls *ls = lkb->lkb_resource->res_ls; 1060 int error; 1061 1062 if (ms->m_flags != DLM_IFL_STUB_MS) 1063 mutex_lock(&ls->ls_waiters_mutex); 1064 error = _remove_from_waiters(lkb, ms->m_type, ms); 1065 if (ms->m_flags != DLM_IFL_STUB_MS) 1066 mutex_unlock(&ls->ls_waiters_mutex); 1067 return error; 1068 } 1069 1070 static void dir_remove(struct dlm_rsb *r) 1071 { 1072 int to_nodeid; 1073 1074 if (dlm_no_directory(r->res_ls)) 1075 return; 1076 1077 to_nodeid = dlm_dir_nodeid(r); 1078 if (to_nodeid != dlm_our_nodeid()) 1079 send_remove(r); 1080 else 1081 dlm_dir_remove_entry(r->res_ls, to_nodeid, 1082 r->res_name, r->res_length); 1083 } 1084 1085 /* FIXME: shouldn't this be able to exit as soon as one non-due rsb is 1086 found since they are in order of newest to oldest? */ 1087 1088 static int shrink_bucket(struct dlm_ls *ls, int b) 1089 { 1090 struct dlm_rsb *r; 1091 int count = 0, found; 1092 1093 for (;;) { 1094 found = 0; 1095 spin_lock(&ls->ls_rsbtbl[b].lock); 1096 list_for_each_entry_reverse(r, &ls->ls_rsbtbl[b].toss, 1097 res_hashchain) { 1098 if (!time_after_eq(jiffies, r->res_toss_time + 1099 dlm_config.ci_toss_secs * HZ)) 1100 continue; 1101 found = 1; 1102 break; 1103 } 1104 1105 if (!found) { 1106 spin_unlock(&ls->ls_rsbtbl[b].lock); 1107 break; 1108 } 1109 1110 if (kref_put(&r->res_ref, kill_rsb)) { 1111 list_del(&r->res_hashchain); 1112 spin_unlock(&ls->ls_rsbtbl[b].lock); 1113 1114 if (is_master(r)) 1115 dir_remove(r); 1116 dlm_free_rsb(r); 1117 count++; 1118 } else { 1119 spin_unlock(&ls->ls_rsbtbl[b].lock); 1120 log_error(ls, "tossed rsb in use %s", r->res_name); 1121 } 1122 } 1123 1124 return count; 1125 } 1126 1127 void dlm_scan_rsbs(struct dlm_ls *ls) 1128 { 1129 int i; 1130 1131 for (i = 0; i < ls->ls_rsbtbl_size; i++) { 1132 shrink_bucket(ls, i); 1133 if (dlm_locking_stopped(ls)) 1134 break; 1135 cond_resched(); 1136 } 1137 } 1138 1139 static void add_timeout(struct dlm_lkb *lkb) 1140 { 1141 struct dlm_ls *ls = lkb->lkb_resource->res_ls; 1142 1143 if (is_master_copy(lkb)) 1144 return; 1145 1146 if (test_bit(LSFL_TIMEWARN, &ls->ls_flags) && 1147 !(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) { 1148 lkb->lkb_flags |= DLM_IFL_WATCH_TIMEWARN; 1149 goto add_it; 1150 } 1151 if (lkb->lkb_exflags & DLM_LKF_TIMEOUT) 1152 goto add_it; 1153 return; 1154 1155 add_it: 1156 DLM_ASSERT(list_empty(&lkb->lkb_time_list), dlm_print_lkb(lkb);); 1157 mutex_lock(&ls->ls_timeout_mutex); 1158 hold_lkb(lkb); 1159 list_add_tail(&lkb->lkb_time_list, &ls->ls_timeout); 1160 mutex_unlock(&ls->ls_timeout_mutex); 1161 } 1162 1163 static void del_timeout(struct dlm_lkb *lkb) 1164 { 1165 struct dlm_ls *ls = lkb->lkb_resource->res_ls; 1166 1167 mutex_lock(&ls->ls_timeout_mutex); 1168 if (!list_empty(&lkb->lkb_time_list)) { 1169 list_del_init(&lkb->lkb_time_list); 1170 unhold_lkb(lkb); 1171 } 1172 mutex_unlock(&ls->ls_timeout_mutex); 1173 } 1174 1175 /* FIXME: is it safe to look at lkb_exflags, lkb_flags, lkb_timestamp, and 1176 lkb_lksb_timeout without lock_rsb? Note: we can't lock timeout_mutex 1177 and then lock rsb because of lock ordering in add_timeout. We may need 1178 to specify some special timeout-related bits in the lkb that are just to 1179 be accessed under the timeout_mutex. */ 1180 1181 void dlm_scan_timeout(struct dlm_ls *ls) 1182 { 1183 struct dlm_rsb *r; 1184 struct dlm_lkb *lkb; 1185 int do_cancel, do_warn; 1186 s64 wait_us; 1187 1188 for (;;) { 1189 if (dlm_locking_stopped(ls)) 1190 break; 1191 1192 do_cancel = 0; 1193 do_warn = 0; 1194 mutex_lock(&ls->ls_timeout_mutex); 1195 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) { 1196 1197 wait_us = ktime_to_us(ktime_sub(ktime_get(), 1198 lkb->lkb_timestamp)); 1199 1200 if ((lkb->lkb_exflags & DLM_LKF_TIMEOUT) && 1201 wait_us >= (lkb->lkb_timeout_cs * 10000)) 1202 do_cancel = 1; 1203 1204 if ((lkb->lkb_flags & DLM_IFL_WATCH_TIMEWARN) && 1205 wait_us >= dlm_config.ci_timewarn_cs * 10000) 1206 do_warn = 1; 1207 1208 if (!do_cancel && !do_warn) 1209 continue; 1210 hold_lkb(lkb); 1211 break; 1212 } 1213 mutex_unlock(&ls->ls_timeout_mutex); 1214 1215 if (!do_cancel && !do_warn) 1216 break; 1217 1218 r = lkb->lkb_resource; 1219 hold_rsb(r); 1220 lock_rsb(r); 1221 1222 if (do_warn) { 1223 /* clear flag so we only warn once */ 1224 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN; 1225 if (!(lkb->lkb_exflags & DLM_LKF_TIMEOUT)) 1226 del_timeout(lkb); 1227 dlm_timeout_warn(lkb); 1228 } 1229 1230 if (do_cancel) { 1231 log_debug(ls, "timeout cancel %x node %d %s", 1232 lkb->lkb_id, lkb->lkb_nodeid, r->res_name); 1233 lkb->lkb_flags &= ~DLM_IFL_WATCH_TIMEWARN; 1234 lkb->lkb_flags |= DLM_IFL_TIMEOUT_CANCEL; 1235 del_timeout(lkb); 1236 _cancel_lock(r, lkb); 1237 } 1238 1239 unlock_rsb(r); 1240 unhold_rsb(r); 1241 dlm_put_lkb(lkb); 1242 } 1243 } 1244 1245 /* This is only called by dlm_recoverd, and we rely on dlm_ls_stop() stopping 1246 dlm_recoverd before checking/setting ls_recover_begin. */ 1247 1248 void dlm_adjust_timeouts(struct dlm_ls *ls) 1249 { 1250 struct dlm_lkb *lkb; 1251 u64 adj_us = jiffies_to_usecs(jiffies - ls->ls_recover_begin); 1252 1253 ls->ls_recover_begin = 0; 1254 mutex_lock(&ls->ls_timeout_mutex); 1255 list_for_each_entry(lkb, &ls->ls_timeout, lkb_time_list) 1256 lkb->lkb_timestamp = ktime_add_us(lkb->lkb_timestamp, adj_us); 1257 mutex_unlock(&ls->ls_timeout_mutex); 1258 1259 if (!dlm_config.ci_waitwarn_us) 1260 return; 1261 1262 mutex_lock(&ls->ls_waiters_mutex); 1263 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) { 1264 if (ktime_to_us(lkb->lkb_wait_time)) 1265 lkb->lkb_wait_time = ktime_get(); 1266 } 1267 mutex_unlock(&ls->ls_waiters_mutex); 1268 } 1269 1270 /* lkb is master or local copy */ 1271 1272 static void set_lvb_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) 1273 { 1274 int b, len = r->res_ls->ls_lvblen; 1275 1276 /* b=1 lvb returned to caller 1277 b=0 lvb written to rsb or invalidated 1278 b=-1 do nothing */ 1279 1280 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1]; 1281 1282 if (b == 1) { 1283 if (!lkb->lkb_lvbptr) 1284 return; 1285 1286 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK)) 1287 return; 1288 1289 if (!r->res_lvbptr) 1290 return; 1291 1292 memcpy(lkb->lkb_lvbptr, r->res_lvbptr, len); 1293 lkb->lkb_lvbseq = r->res_lvbseq; 1294 1295 } else if (b == 0) { 1296 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) { 1297 rsb_set_flag(r, RSB_VALNOTVALID); 1298 return; 1299 } 1300 1301 if (!lkb->lkb_lvbptr) 1302 return; 1303 1304 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK)) 1305 return; 1306 1307 if (!r->res_lvbptr) 1308 r->res_lvbptr = dlm_allocate_lvb(r->res_ls); 1309 1310 if (!r->res_lvbptr) 1311 return; 1312 1313 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, len); 1314 r->res_lvbseq++; 1315 lkb->lkb_lvbseq = r->res_lvbseq; 1316 rsb_clear_flag(r, RSB_VALNOTVALID); 1317 } 1318 1319 if (rsb_flag(r, RSB_VALNOTVALID)) 1320 lkb->lkb_sbflags |= DLM_SBF_VALNOTVALID; 1321 } 1322 1323 static void set_lvb_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb) 1324 { 1325 if (lkb->lkb_grmode < DLM_LOCK_PW) 1326 return; 1327 1328 if (lkb->lkb_exflags & DLM_LKF_IVVALBLK) { 1329 rsb_set_flag(r, RSB_VALNOTVALID); 1330 return; 1331 } 1332 1333 if (!lkb->lkb_lvbptr) 1334 return; 1335 1336 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK)) 1337 return; 1338 1339 if (!r->res_lvbptr) 1340 r->res_lvbptr = dlm_allocate_lvb(r->res_ls); 1341 1342 if (!r->res_lvbptr) 1343 return; 1344 1345 memcpy(r->res_lvbptr, lkb->lkb_lvbptr, r->res_ls->ls_lvblen); 1346 r->res_lvbseq++; 1347 rsb_clear_flag(r, RSB_VALNOTVALID); 1348 } 1349 1350 /* lkb is process copy (pc) */ 1351 1352 static void set_lvb_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb, 1353 struct dlm_message *ms) 1354 { 1355 int b; 1356 1357 if (!lkb->lkb_lvbptr) 1358 return; 1359 1360 if (!(lkb->lkb_exflags & DLM_LKF_VALBLK)) 1361 return; 1362 1363 b = dlm_lvb_operations[lkb->lkb_grmode + 1][lkb->lkb_rqmode + 1]; 1364 if (b == 1) { 1365 int len = receive_extralen(ms); 1366 if (len > DLM_RESNAME_MAXLEN) 1367 len = DLM_RESNAME_MAXLEN; 1368 memcpy(lkb->lkb_lvbptr, ms->m_extra, len); 1369 lkb->lkb_lvbseq = ms->m_lvbseq; 1370 } 1371 } 1372 1373 /* Manipulate lkb's on rsb's convert/granted/waiting queues 1374 remove_lock -- used for unlock, removes lkb from granted 1375 revert_lock -- used for cancel, moves lkb from convert to granted 1376 grant_lock -- used for request and convert, adds lkb to granted or 1377 moves lkb from convert or waiting to granted 1378 1379 Each of these is used for master or local copy lkb's. There is 1380 also a _pc() variation used to make the corresponding change on 1381 a process copy (pc) lkb. */ 1382 1383 static void _remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) 1384 { 1385 del_lkb(r, lkb); 1386 lkb->lkb_grmode = DLM_LOCK_IV; 1387 /* this unhold undoes the original ref from create_lkb() 1388 so this leads to the lkb being freed */ 1389 unhold_lkb(lkb); 1390 } 1391 1392 static void remove_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) 1393 { 1394 set_lvb_unlock(r, lkb); 1395 _remove_lock(r, lkb); 1396 } 1397 1398 static void remove_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb) 1399 { 1400 _remove_lock(r, lkb); 1401 } 1402 1403 /* returns: 0 did nothing 1404 1 moved lock to granted 1405 -1 removed lock */ 1406 1407 static int revert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) 1408 { 1409 int rv = 0; 1410 1411 lkb->lkb_rqmode = DLM_LOCK_IV; 1412 1413 switch (lkb->lkb_status) { 1414 case DLM_LKSTS_GRANTED: 1415 break; 1416 case DLM_LKSTS_CONVERT: 1417 move_lkb(r, lkb, DLM_LKSTS_GRANTED); 1418 rv = 1; 1419 break; 1420 case DLM_LKSTS_WAITING: 1421 del_lkb(r, lkb); 1422 lkb->lkb_grmode = DLM_LOCK_IV; 1423 /* this unhold undoes the original ref from create_lkb() 1424 so this leads to the lkb being freed */ 1425 unhold_lkb(lkb); 1426 rv = -1; 1427 break; 1428 default: 1429 log_print("invalid status for revert %d", lkb->lkb_status); 1430 } 1431 return rv; 1432 } 1433 1434 static int revert_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb) 1435 { 1436 return revert_lock(r, lkb); 1437 } 1438 1439 static void _grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) 1440 { 1441 if (lkb->lkb_grmode != lkb->lkb_rqmode) { 1442 lkb->lkb_grmode = lkb->lkb_rqmode; 1443 if (lkb->lkb_status) 1444 move_lkb(r, lkb, DLM_LKSTS_GRANTED); 1445 else 1446 add_lkb(r, lkb, DLM_LKSTS_GRANTED); 1447 } 1448 1449 lkb->lkb_rqmode = DLM_LOCK_IV; 1450 } 1451 1452 static void grant_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) 1453 { 1454 set_lvb_lock(r, lkb); 1455 _grant_lock(r, lkb); 1456 lkb->lkb_highbast = 0; 1457 } 1458 1459 static void grant_lock_pc(struct dlm_rsb *r, struct dlm_lkb *lkb, 1460 struct dlm_message *ms) 1461 { 1462 set_lvb_lock_pc(r, lkb, ms); 1463 _grant_lock(r, lkb); 1464 } 1465 1466 /* called by grant_pending_locks() which means an async grant message must 1467 be sent to the requesting node in addition to granting the lock if the 1468 lkb belongs to a remote node. */ 1469 1470 static void grant_lock_pending(struct dlm_rsb *r, struct dlm_lkb *lkb) 1471 { 1472 grant_lock(r, lkb); 1473 if (is_master_copy(lkb)) 1474 send_grant(r, lkb); 1475 else 1476 queue_cast(r, lkb, 0); 1477 } 1478 1479 /* The special CONVDEADLK, ALTPR and ALTCW flags allow the master to 1480 change the granted/requested modes. We're munging things accordingly in 1481 the process copy. 1482 CONVDEADLK: our grmode may have been forced down to NL to resolve a 1483 conversion deadlock 1484 ALTPR/ALTCW: our rqmode may have been changed to PR or CW to become 1485 compatible with other granted locks */ 1486 1487 static void munge_demoted(struct dlm_lkb *lkb) 1488 { 1489 if (lkb->lkb_rqmode == DLM_LOCK_IV || lkb->lkb_grmode == DLM_LOCK_IV) { 1490 log_print("munge_demoted %x invalid modes gr %d rq %d", 1491 lkb->lkb_id, lkb->lkb_grmode, lkb->lkb_rqmode); 1492 return; 1493 } 1494 1495 lkb->lkb_grmode = DLM_LOCK_NL; 1496 } 1497 1498 static void munge_altmode(struct dlm_lkb *lkb, struct dlm_message *ms) 1499 { 1500 if (ms->m_type != DLM_MSG_REQUEST_REPLY && 1501 ms->m_type != DLM_MSG_GRANT) { 1502 log_print("munge_altmode %x invalid reply type %d", 1503 lkb->lkb_id, ms->m_type); 1504 return; 1505 } 1506 1507 if (lkb->lkb_exflags & DLM_LKF_ALTPR) 1508 lkb->lkb_rqmode = DLM_LOCK_PR; 1509 else if (lkb->lkb_exflags & DLM_LKF_ALTCW) 1510 lkb->lkb_rqmode = DLM_LOCK_CW; 1511 else { 1512 log_print("munge_altmode invalid exflags %x", lkb->lkb_exflags); 1513 dlm_print_lkb(lkb); 1514 } 1515 } 1516 1517 static inline int first_in_list(struct dlm_lkb *lkb, struct list_head *head) 1518 { 1519 struct dlm_lkb *first = list_entry(head->next, struct dlm_lkb, 1520 lkb_statequeue); 1521 if (lkb->lkb_id == first->lkb_id) 1522 return 1; 1523 1524 return 0; 1525 } 1526 1527 /* Check if the given lkb conflicts with another lkb on the queue. */ 1528 1529 static int queue_conflict(struct list_head *head, struct dlm_lkb *lkb) 1530 { 1531 struct dlm_lkb *this; 1532 1533 list_for_each_entry(this, head, lkb_statequeue) { 1534 if (this == lkb) 1535 continue; 1536 if (!modes_compat(this, lkb)) 1537 return 1; 1538 } 1539 return 0; 1540 } 1541 1542 /* 1543 * "A conversion deadlock arises with a pair of lock requests in the converting 1544 * queue for one resource. The granted mode of each lock blocks the requested 1545 * mode of the other lock." 1546 * 1547 * Part 2: if the granted mode of lkb is preventing an earlier lkb in the 1548 * convert queue from being granted, then deadlk/demote lkb. 1549 * 1550 * Example: 1551 * Granted Queue: empty 1552 * Convert Queue: NL->EX (first lock) 1553 * PR->EX (second lock) 1554 * 1555 * The first lock can't be granted because of the granted mode of the second 1556 * lock and the second lock can't be granted because it's not first in the 1557 * list. We either cancel lkb's conversion (PR->EX) and return EDEADLK, or we 1558 * demote the granted mode of lkb (from PR to NL) if it has the CONVDEADLK 1559 * flag set and return DEMOTED in the lksb flags. 1560 * 1561 * Originally, this function detected conv-deadlk in a more limited scope: 1562 * - if !modes_compat(lkb1, lkb2) && !modes_compat(lkb2, lkb1), or 1563 * - if lkb1 was the first entry in the queue (not just earlier), and was 1564 * blocked by the granted mode of lkb2, and there was nothing on the 1565 * granted queue preventing lkb1 from being granted immediately, i.e. 1566 * lkb2 was the only thing preventing lkb1 from being granted. 1567 * 1568 * That second condition meant we'd only say there was conv-deadlk if 1569 * resolving it (by demotion) would lead to the first lock on the convert 1570 * queue being granted right away. It allowed conversion deadlocks to exist 1571 * between locks on the convert queue while they couldn't be granted anyway. 1572 * 1573 * Now, we detect and take action on conversion deadlocks immediately when 1574 * they're created, even if they may not be immediately consequential. If 1575 * lkb1 exists anywhere in the convert queue and lkb2 comes in with a granted 1576 * mode that would prevent lkb1's conversion from being granted, we do a 1577 * deadlk/demote on lkb2 right away and don't let it onto the convert queue. 1578 * I think this means that the lkb_is_ahead condition below should always 1579 * be zero, i.e. there will never be conv-deadlk between two locks that are 1580 * both already on the convert queue. 1581 */ 1582 1583 static int conversion_deadlock_detect(struct dlm_rsb *r, struct dlm_lkb *lkb2) 1584 { 1585 struct dlm_lkb *lkb1; 1586 int lkb_is_ahead = 0; 1587 1588 list_for_each_entry(lkb1, &r->res_convertqueue, lkb_statequeue) { 1589 if (lkb1 == lkb2) { 1590 lkb_is_ahead = 1; 1591 continue; 1592 } 1593 1594 if (!lkb_is_ahead) { 1595 if (!modes_compat(lkb2, lkb1)) 1596 return 1; 1597 } else { 1598 if (!modes_compat(lkb2, lkb1) && 1599 !modes_compat(lkb1, lkb2)) 1600 return 1; 1601 } 1602 } 1603 return 0; 1604 } 1605 1606 /* 1607 * Return 1 if the lock can be granted, 0 otherwise. 1608 * Also detect and resolve conversion deadlocks. 1609 * 1610 * lkb is the lock to be granted 1611 * 1612 * now is 1 if the function is being called in the context of the 1613 * immediate request, it is 0 if called later, after the lock has been 1614 * queued. 1615 * 1616 * References are from chapter 6 of "VAXcluster Principles" by Roy Davis 1617 */ 1618 1619 static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now) 1620 { 1621 int8_t conv = (lkb->lkb_grmode != DLM_LOCK_IV); 1622 1623 /* 1624 * 6-10: Version 5.4 introduced an option to address the phenomenon of 1625 * a new request for a NL mode lock being blocked. 1626 * 1627 * 6-11: If the optional EXPEDITE flag is used with the new NL mode 1628 * request, then it would be granted. In essence, the use of this flag 1629 * tells the Lock Manager to expedite theis request by not considering 1630 * what may be in the CONVERTING or WAITING queues... As of this 1631 * writing, the EXPEDITE flag can be used only with new requests for NL 1632 * mode locks. This flag is not valid for conversion requests. 1633 * 1634 * A shortcut. Earlier checks return an error if EXPEDITE is used in a 1635 * conversion or used with a non-NL requested mode. We also know an 1636 * EXPEDITE request is always granted immediately, so now must always 1637 * be 1. The full condition to grant an expedite request: (now && 1638 * !conv && lkb->rqmode == DLM_LOCK_NL && (flags & EXPEDITE)) can 1639 * therefore be shortened to just checking the flag. 1640 */ 1641 1642 if (lkb->lkb_exflags & DLM_LKF_EXPEDITE) 1643 return 1; 1644 1645 /* 1646 * A shortcut. Without this, !queue_conflict(grantqueue, lkb) would be 1647 * added to the remaining conditions. 1648 */ 1649 1650 if (queue_conflict(&r->res_grantqueue, lkb)) 1651 goto out; 1652 1653 /* 1654 * 6-3: By default, a conversion request is immediately granted if the 1655 * requested mode is compatible with the modes of all other granted 1656 * locks 1657 */ 1658 1659 if (queue_conflict(&r->res_convertqueue, lkb)) 1660 goto out; 1661 1662 /* 1663 * 6-5: But the default algorithm for deciding whether to grant or 1664 * queue conversion requests does not by itself guarantee that such 1665 * requests are serviced on a "first come first serve" basis. This, in 1666 * turn, can lead to a phenomenon known as "indefinate postponement". 1667 * 1668 * 6-7: This issue is dealt with by using the optional QUECVT flag with 1669 * the system service employed to request a lock conversion. This flag 1670 * forces certain conversion requests to be queued, even if they are 1671 * compatible with the granted modes of other locks on the same 1672 * resource. Thus, the use of this flag results in conversion requests 1673 * being ordered on a "first come first servce" basis. 1674 * 1675 * DCT: This condition is all about new conversions being able to occur 1676 * "in place" while the lock remains on the granted queue (assuming 1677 * nothing else conflicts.) IOW if QUECVT isn't set, a conversion 1678 * doesn't _have_ to go onto the convert queue where it's processed in 1679 * order. The "now" variable is necessary to distinguish converts 1680 * being received and processed for the first time now, because once a 1681 * convert is moved to the conversion queue the condition below applies 1682 * requiring fifo granting. 1683 */ 1684 1685 if (now && conv && !(lkb->lkb_exflags & DLM_LKF_QUECVT)) 1686 return 1; 1687 1688 /* 1689 * The NOORDER flag is set to avoid the standard vms rules on grant 1690 * order. 1691 */ 1692 1693 if (lkb->lkb_exflags & DLM_LKF_NOORDER) 1694 return 1; 1695 1696 /* 1697 * 6-3: Once in that queue [CONVERTING], a conversion request cannot be 1698 * granted until all other conversion requests ahead of it are granted 1699 * and/or canceled. 1700 */ 1701 1702 if (!now && conv && first_in_list(lkb, &r->res_convertqueue)) 1703 return 1; 1704 1705 /* 1706 * 6-4: By default, a new request is immediately granted only if all 1707 * three of the following conditions are satisfied when the request is 1708 * issued: 1709 * - The queue of ungranted conversion requests for the resource is 1710 * empty. 1711 * - The queue of ungranted new requests for the resource is empty. 1712 * - The mode of the new request is compatible with the most 1713 * restrictive mode of all granted locks on the resource. 1714 */ 1715 1716 if (now && !conv && list_empty(&r->res_convertqueue) && 1717 list_empty(&r->res_waitqueue)) 1718 return 1; 1719 1720 /* 1721 * 6-4: Once a lock request is in the queue of ungranted new requests, 1722 * it cannot be granted until the queue of ungranted conversion 1723 * requests is empty, all ungranted new requests ahead of it are 1724 * granted and/or canceled, and it is compatible with the granted mode 1725 * of the most restrictive lock granted on the resource. 1726 */ 1727 1728 if (!now && !conv && list_empty(&r->res_convertqueue) && 1729 first_in_list(lkb, &r->res_waitqueue)) 1730 return 1; 1731 out: 1732 return 0; 1733 } 1734 1735 static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now, 1736 int *err) 1737 { 1738 int rv; 1739 int8_t alt = 0, rqmode = lkb->lkb_rqmode; 1740 int8_t is_convert = (lkb->lkb_grmode != DLM_LOCK_IV); 1741 1742 if (err) 1743 *err = 0; 1744 1745 rv = _can_be_granted(r, lkb, now); 1746 if (rv) 1747 goto out; 1748 1749 /* 1750 * The CONVDEADLK flag is non-standard and tells the dlm to resolve 1751 * conversion deadlocks by demoting grmode to NL, otherwise the dlm 1752 * cancels one of the locks. 1753 */ 1754 1755 if (is_convert && can_be_queued(lkb) && 1756 conversion_deadlock_detect(r, lkb)) { 1757 if (lkb->lkb_exflags & DLM_LKF_CONVDEADLK) { 1758 lkb->lkb_grmode = DLM_LOCK_NL; 1759 lkb->lkb_sbflags |= DLM_SBF_DEMOTED; 1760 } else if (!(lkb->lkb_exflags & DLM_LKF_NODLCKWT)) { 1761 if (err) 1762 *err = -EDEADLK; 1763 else { 1764 log_print("can_be_granted deadlock %x now %d", 1765 lkb->lkb_id, now); 1766 dlm_dump_rsb(r); 1767 } 1768 } 1769 goto out; 1770 } 1771 1772 /* 1773 * The ALTPR and ALTCW flags are non-standard and tell the dlm to try 1774 * to grant a request in a mode other than the normal rqmode. It's a 1775 * simple way to provide a big optimization to applications that can 1776 * use them. 1777 */ 1778 1779 if (rqmode != DLM_LOCK_PR && (lkb->lkb_exflags & DLM_LKF_ALTPR)) 1780 alt = DLM_LOCK_PR; 1781 else if (rqmode != DLM_LOCK_CW && (lkb->lkb_exflags & DLM_LKF_ALTCW)) 1782 alt = DLM_LOCK_CW; 1783 1784 if (alt) { 1785 lkb->lkb_rqmode = alt; 1786 rv = _can_be_granted(r, lkb, now); 1787 if (rv) 1788 lkb->lkb_sbflags |= DLM_SBF_ALTMODE; 1789 else 1790 lkb->lkb_rqmode = rqmode; 1791 } 1792 out: 1793 return rv; 1794 } 1795 1796 /* FIXME: I don't think that can_be_granted() can/will demote or find deadlock 1797 for locks pending on the convert list. Once verified (watch for these 1798 log_prints), we should be able to just call _can_be_granted() and not 1799 bother with the demote/deadlk cases here (and there's no easy way to deal 1800 with a deadlk here, we'd have to generate something like grant_lock with 1801 the deadlk error.) */ 1802 1803 /* Returns the highest requested mode of all blocked conversions; sets 1804 cw if there's a blocked conversion to DLM_LOCK_CW. */ 1805 1806 static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw) 1807 { 1808 struct dlm_lkb *lkb, *s; 1809 int hi, demoted, quit, grant_restart, demote_restart; 1810 int deadlk; 1811 1812 quit = 0; 1813 restart: 1814 grant_restart = 0; 1815 demote_restart = 0; 1816 hi = DLM_LOCK_IV; 1817 1818 list_for_each_entry_safe(lkb, s, &r->res_convertqueue, lkb_statequeue) { 1819 demoted = is_demoted(lkb); 1820 deadlk = 0; 1821 1822 if (can_be_granted(r, lkb, 0, &deadlk)) { 1823 grant_lock_pending(r, lkb); 1824 grant_restart = 1; 1825 continue; 1826 } 1827 1828 if (!demoted && is_demoted(lkb)) { 1829 log_print("WARN: pending demoted %x node %d %s", 1830 lkb->lkb_id, lkb->lkb_nodeid, r->res_name); 1831 demote_restart = 1; 1832 continue; 1833 } 1834 1835 if (deadlk) { 1836 log_print("WARN: pending deadlock %x node %d %s", 1837 lkb->lkb_id, lkb->lkb_nodeid, r->res_name); 1838 dlm_dump_rsb(r); 1839 continue; 1840 } 1841 1842 hi = max_t(int, lkb->lkb_rqmode, hi); 1843 1844 if (cw && lkb->lkb_rqmode == DLM_LOCK_CW) 1845 *cw = 1; 1846 } 1847 1848 if (grant_restart) 1849 goto restart; 1850 if (demote_restart && !quit) { 1851 quit = 1; 1852 goto restart; 1853 } 1854 1855 return max_t(int, high, hi); 1856 } 1857 1858 static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw) 1859 { 1860 struct dlm_lkb *lkb, *s; 1861 1862 list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) { 1863 if (can_be_granted(r, lkb, 0, NULL)) 1864 grant_lock_pending(r, lkb); 1865 else { 1866 high = max_t(int, lkb->lkb_rqmode, high); 1867 if (lkb->lkb_rqmode == DLM_LOCK_CW) 1868 *cw = 1; 1869 } 1870 } 1871 1872 return high; 1873 } 1874 1875 /* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked 1876 on either the convert or waiting queue. 1877 high is the largest rqmode of all locks blocked on the convert or 1878 waiting queue. */ 1879 1880 static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw) 1881 { 1882 if (gr->lkb_grmode == DLM_LOCK_PR && cw) { 1883 if (gr->lkb_highbast < DLM_LOCK_EX) 1884 return 1; 1885 return 0; 1886 } 1887 1888 if (gr->lkb_highbast < high && 1889 !__dlm_compat_matrix[gr->lkb_grmode+1][high+1]) 1890 return 1; 1891 return 0; 1892 } 1893 1894 static void grant_pending_locks(struct dlm_rsb *r) 1895 { 1896 struct dlm_lkb *lkb, *s; 1897 int high = DLM_LOCK_IV; 1898 int cw = 0; 1899 1900 DLM_ASSERT(is_master(r), dlm_dump_rsb(r);); 1901 1902 high = grant_pending_convert(r, high, &cw); 1903 high = grant_pending_wait(r, high, &cw); 1904 1905 if (high == DLM_LOCK_IV) 1906 return; 1907 1908 /* 1909 * If there are locks left on the wait/convert queue then send blocking 1910 * ASTs to granted locks based on the largest requested mode (high) 1911 * found above. 1912 */ 1913 1914 list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) { 1915 if (lkb->lkb_bastfn && lock_requires_bast(lkb, high, cw)) { 1916 if (cw && high == DLM_LOCK_PR && 1917 lkb->lkb_grmode == DLM_LOCK_PR) 1918 queue_bast(r, lkb, DLM_LOCK_CW); 1919 else 1920 queue_bast(r, lkb, high); 1921 lkb->lkb_highbast = high; 1922 } 1923 } 1924 } 1925 1926 static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq) 1927 { 1928 if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) || 1929 (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) { 1930 if (gr->lkb_highbast < DLM_LOCK_EX) 1931 return 1; 1932 return 0; 1933 } 1934 1935 if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq)) 1936 return 1; 1937 return 0; 1938 } 1939 1940 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head, 1941 struct dlm_lkb *lkb) 1942 { 1943 struct dlm_lkb *gr; 1944 1945 list_for_each_entry(gr, head, lkb_statequeue) { 1946 /* skip self when sending basts to convertqueue */ 1947 if (gr == lkb) 1948 continue; 1949 if (gr->lkb_bastfn && modes_require_bast(gr, lkb)) { 1950 queue_bast(r, gr, lkb->lkb_rqmode); 1951 gr->lkb_highbast = lkb->lkb_rqmode; 1952 } 1953 } 1954 } 1955 1956 static void send_blocking_asts(struct dlm_rsb *r, struct dlm_lkb *lkb) 1957 { 1958 send_bast_queue(r, &r->res_grantqueue, lkb); 1959 } 1960 1961 static void send_blocking_asts_all(struct dlm_rsb *r, struct dlm_lkb *lkb) 1962 { 1963 send_bast_queue(r, &r->res_grantqueue, lkb); 1964 send_bast_queue(r, &r->res_convertqueue, lkb); 1965 } 1966 1967 /* set_master(r, lkb) -- set the master nodeid of a resource 1968 1969 The purpose of this function is to set the nodeid field in the given 1970 lkb using the nodeid field in the given rsb. If the rsb's nodeid is 1971 known, it can just be copied to the lkb and the function will return 1972 0. If the rsb's nodeid is _not_ known, it needs to be looked up 1973 before it can be copied to the lkb. 1974 1975 When the rsb nodeid is being looked up remotely, the initial lkb 1976 causing the lookup is kept on the ls_waiters list waiting for the 1977 lookup reply. Other lkb's waiting for the same rsb lookup are kept 1978 on the rsb's res_lookup list until the master is verified. 1979 1980 Return values: 1981 0: nodeid is set in rsb/lkb and the caller should go ahead and use it 1982 1: the rsb master is not available and the lkb has been placed on 1983 a wait queue 1984 */ 1985 1986 static int set_master(struct dlm_rsb *r, struct dlm_lkb *lkb) 1987 { 1988 struct dlm_ls *ls = r->res_ls; 1989 int i, error, dir_nodeid, ret_nodeid, our_nodeid = dlm_our_nodeid(); 1990 1991 if (rsb_flag(r, RSB_MASTER_UNCERTAIN)) { 1992 rsb_clear_flag(r, RSB_MASTER_UNCERTAIN); 1993 r->res_first_lkid = lkb->lkb_id; 1994 lkb->lkb_nodeid = r->res_nodeid; 1995 return 0; 1996 } 1997 1998 if (r->res_first_lkid && r->res_first_lkid != lkb->lkb_id) { 1999 list_add_tail(&lkb->lkb_rsb_lookup, &r->res_lookup); 2000 return 1; 2001 } 2002 2003 if (r->res_nodeid == 0) { 2004 lkb->lkb_nodeid = 0; 2005 return 0; 2006 } 2007 2008 if (r->res_nodeid > 0) { 2009 lkb->lkb_nodeid = r->res_nodeid; 2010 return 0; 2011 } 2012 2013 DLM_ASSERT(r->res_nodeid == -1, dlm_dump_rsb(r);); 2014 2015 dir_nodeid = dlm_dir_nodeid(r); 2016 2017 if (dir_nodeid != our_nodeid) { 2018 r->res_first_lkid = lkb->lkb_id; 2019 send_lookup(r, lkb); 2020 return 1; 2021 } 2022 2023 for (i = 0; i < 2; i++) { 2024 /* It's possible for dlm_scand to remove an old rsb for 2025 this same resource from the toss list, us to create 2026 a new one, look up the master locally, and find it 2027 already exists just before dlm_scand does the 2028 dir_remove() on the previous rsb. */ 2029 2030 error = dlm_dir_lookup(ls, our_nodeid, r->res_name, 2031 r->res_length, &ret_nodeid); 2032 if (!error) 2033 break; 2034 log_debug(ls, "dir_lookup error %d %s", error, r->res_name); 2035 schedule(); 2036 } 2037 if (error && error != -EEXIST) 2038 return error; 2039 2040 if (ret_nodeid == our_nodeid) { 2041 r->res_first_lkid = 0; 2042 r->res_nodeid = 0; 2043 lkb->lkb_nodeid = 0; 2044 } else { 2045 r->res_first_lkid = lkb->lkb_id; 2046 r->res_nodeid = ret_nodeid; 2047 lkb->lkb_nodeid = ret_nodeid; 2048 } 2049 return 0; 2050 } 2051 2052 static void process_lookup_list(struct dlm_rsb *r) 2053 { 2054 struct dlm_lkb *lkb, *safe; 2055 2056 list_for_each_entry_safe(lkb, safe, &r->res_lookup, lkb_rsb_lookup) { 2057 list_del_init(&lkb->lkb_rsb_lookup); 2058 _request_lock(r, lkb); 2059 schedule(); 2060 } 2061 } 2062 2063 /* confirm_master -- confirm (or deny) an rsb's master nodeid */ 2064 2065 static void confirm_master(struct dlm_rsb *r, int error) 2066 { 2067 struct dlm_lkb *lkb; 2068 2069 if (!r->res_first_lkid) 2070 return; 2071 2072 switch (error) { 2073 case 0: 2074 case -EINPROGRESS: 2075 r->res_first_lkid = 0; 2076 process_lookup_list(r); 2077 break; 2078 2079 case -EAGAIN: 2080 case -EBADR: 2081 case -ENOTBLK: 2082 /* the remote request failed and won't be retried (it was 2083 a NOQUEUE, or has been canceled/unlocked); make a waiting 2084 lkb the first_lkid */ 2085 2086 r->res_first_lkid = 0; 2087 2088 if (!list_empty(&r->res_lookup)) { 2089 lkb = list_entry(r->res_lookup.next, struct dlm_lkb, 2090 lkb_rsb_lookup); 2091 list_del_init(&lkb->lkb_rsb_lookup); 2092 r->res_first_lkid = lkb->lkb_id; 2093 _request_lock(r, lkb); 2094 } 2095 break; 2096 2097 default: 2098 log_error(r->res_ls, "confirm_master unknown error %d", error); 2099 } 2100 } 2101 2102 static int set_lock_args(int mode, struct dlm_lksb *lksb, uint32_t flags, 2103 int namelen, unsigned long timeout_cs, 2104 void (*ast) (void *astparam), 2105 void *astparam, 2106 void (*bast) (void *astparam, int mode), 2107 struct dlm_args *args) 2108 { 2109 int rv = -EINVAL; 2110 2111 /* check for invalid arg usage */ 2112 2113 if (mode < 0 || mode > DLM_LOCK_EX) 2114 goto out; 2115 2116 if (!(flags & DLM_LKF_CONVERT) && (namelen > DLM_RESNAME_MAXLEN)) 2117 goto out; 2118 2119 if (flags & DLM_LKF_CANCEL) 2120 goto out; 2121 2122 if (flags & DLM_LKF_QUECVT && !(flags & DLM_LKF_CONVERT)) 2123 goto out; 2124 2125 if (flags & DLM_LKF_CONVDEADLK && !(flags & DLM_LKF_CONVERT)) 2126 goto out; 2127 2128 if (flags & DLM_LKF_CONVDEADLK && flags & DLM_LKF_NOQUEUE) 2129 goto out; 2130 2131 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_CONVERT) 2132 goto out; 2133 2134 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_QUECVT) 2135 goto out; 2136 2137 if (flags & DLM_LKF_EXPEDITE && flags & DLM_LKF_NOQUEUE) 2138 goto out; 2139 2140 if (flags & DLM_LKF_EXPEDITE && mode != DLM_LOCK_NL) 2141 goto out; 2142 2143 if (!ast || !lksb) 2144 goto out; 2145 2146 if (flags & DLM_LKF_VALBLK && !lksb->sb_lvbptr) 2147 goto out; 2148 2149 if (flags & DLM_LKF_CONVERT && !lksb->sb_lkid) 2150 goto out; 2151 2152 /* these args will be copied to the lkb in validate_lock_args, 2153 it cannot be done now because when converting locks, fields in 2154 an active lkb cannot be modified before locking the rsb */ 2155 2156 args->flags = flags; 2157 args->astfn = ast; 2158 args->astparam = astparam; 2159 args->bastfn = bast; 2160 args->timeout = timeout_cs; 2161 args->mode = mode; 2162 args->lksb = lksb; 2163 rv = 0; 2164 out: 2165 return rv; 2166 } 2167 2168 static int set_unlock_args(uint32_t flags, void *astarg, struct dlm_args *args) 2169 { 2170 if (flags & ~(DLM_LKF_CANCEL | DLM_LKF_VALBLK | DLM_LKF_IVVALBLK | 2171 DLM_LKF_FORCEUNLOCK)) 2172 return -EINVAL; 2173 2174 if (flags & DLM_LKF_CANCEL && flags & DLM_LKF_FORCEUNLOCK) 2175 return -EINVAL; 2176 2177 args->flags = flags; 2178 args->astparam = astarg; 2179 return 0; 2180 } 2181 2182 static int validate_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb, 2183 struct dlm_args *args) 2184 { 2185 int rv = -EINVAL; 2186 2187 if (args->flags & DLM_LKF_CONVERT) { 2188 if (lkb->lkb_flags & DLM_IFL_MSTCPY) 2189 goto out; 2190 2191 if (args->flags & DLM_LKF_QUECVT && 2192 !__quecvt_compat_matrix[lkb->lkb_grmode+1][args->mode+1]) 2193 goto out; 2194 2195 rv = -EBUSY; 2196 if (lkb->lkb_status != DLM_LKSTS_GRANTED) 2197 goto out; 2198 2199 if (lkb->lkb_wait_type) 2200 goto out; 2201 2202 if (is_overlap(lkb)) 2203 goto out; 2204 } 2205 2206 lkb->lkb_exflags = args->flags; 2207 lkb->lkb_sbflags = 0; 2208 lkb->lkb_astfn = args->astfn; 2209 lkb->lkb_astparam = args->astparam; 2210 lkb->lkb_bastfn = args->bastfn; 2211 lkb->lkb_rqmode = args->mode; 2212 lkb->lkb_lksb = args->lksb; 2213 lkb->lkb_lvbptr = args->lksb->sb_lvbptr; 2214 lkb->lkb_ownpid = (int) current->pid; 2215 lkb->lkb_timeout_cs = args->timeout; 2216 rv = 0; 2217 out: 2218 if (rv) 2219 log_debug(ls, "validate_lock_args %d %x %x %x %d %d %s", 2220 rv, lkb->lkb_id, lkb->lkb_flags, args->flags, 2221 lkb->lkb_status, lkb->lkb_wait_type, 2222 lkb->lkb_resource->res_name); 2223 return rv; 2224 } 2225 2226 /* when dlm_unlock() sees -EBUSY with CANCEL/FORCEUNLOCK it returns 0 2227 for success */ 2228 2229 /* note: it's valid for lkb_nodeid/res_nodeid to be -1 when we get here 2230 because there may be a lookup in progress and it's valid to do 2231 cancel/unlockf on it */ 2232 2233 static int validate_unlock_args(struct dlm_lkb *lkb, struct dlm_args *args) 2234 { 2235 struct dlm_ls *ls = lkb->lkb_resource->res_ls; 2236 int rv = -EINVAL; 2237 2238 if (lkb->lkb_flags & DLM_IFL_MSTCPY) { 2239 log_error(ls, "unlock on MSTCPY %x", lkb->lkb_id); 2240 dlm_print_lkb(lkb); 2241 goto out; 2242 } 2243 2244 /* an lkb may still exist even though the lock is EOL'ed due to a 2245 cancel, unlock or failed noqueue request; an app can't use these 2246 locks; return same error as if the lkid had not been found at all */ 2247 2248 if (lkb->lkb_flags & DLM_IFL_ENDOFLIFE) { 2249 log_debug(ls, "unlock on ENDOFLIFE %x", lkb->lkb_id); 2250 rv = -ENOENT; 2251 goto out; 2252 } 2253 2254 /* an lkb may be waiting for an rsb lookup to complete where the 2255 lookup was initiated by another lock */ 2256 2257 if (!list_empty(&lkb->lkb_rsb_lookup)) { 2258 if (args->flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK)) { 2259 log_debug(ls, "unlock on rsb_lookup %x", lkb->lkb_id); 2260 list_del_init(&lkb->lkb_rsb_lookup); 2261 queue_cast(lkb->lkb_resource, lkb, 2262 args->flags & DLM_LKF_CANCEL ? 2263 -DLM_ECANCEL : -DLM_EUNLOCK); 2264 unhold_lkb(lkb); /* undoes create_lkb() */ 2265 } 2266 /* caller changes -EBUSY to 0 for CANCEL and FORCEUNLOCK */ 2267 rv = -EBUSY; 2268 goto out; 2269 } 2270 2271 /* cancel not allowed with another cancel/unlock in progress */ 2272 2273 if (args->flags & DLM_LKF_CANCEL) { 2274 if (lkb->lkb_exflags & DLM_LKF_CANCEL) 2275 goto out; 2276 2277 if (is_overlap(lkb)) 2278 goto out; 2279 2280 /* don't let scand try to do a cancel */ 2281 del_timeout(lkb); 2282 2283 if (lkb->lkb_flags & DLM_IFL_RESEND) { 2284 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL; 2285 rv = -EBUSY; 2286 goto out; 2287 } 2288 2289 /* there's nothing to cancel */ 2290 if (lkb->lkb_status == DLM_LKSTS_GRANTED && 2291 !lkb->lkb_wait_type) { 2292 rv = -EBUSY; 2293 goto out; 2294 } 2295 2296 switch (lkb->lkb_wait_type) { 2297 case DLM_MSG_LOOKUP: 2298 case DLM_MSG_REQUEST: 2299 lkb->lkb_flags |= DLM_IFL_OVERLAP_CANCEL; 2300 rv = -EBUSY; 2301 goto out; 2302 case DLM_MSG_UNLOCK: 2303 case DLM_MSG_CANCEL: 2304 goto out; 2305 } 2306 /* add_to_waiters() will set OVERLAP_CANCEL */ 2307 goto out_ok; 2308 } 2309 2310 /* do we need to allow a force-unlock if there's a normal unlock 2311 already in progress? in what conditions could the normal unlock 2312 fail such that we'd want to send a force-unlock to be sure? */ 2313 2314 if (args->flags & DLM_LKF_FORCEUNLOCK) { 2315 if (lkb->lkb_exflags & DLM_LKF_FORCEUNLOCK) 2316 goto out; 2317 2318 if (is_overlap_unlock(lkb)) 2319 goto out; 2320 2321 /* don't let scand try to do a cancel */ 2322 del_timeout(lkb); 2323 2324 if (lkb->lkb_flags & DLM_IFL_RESEND) { 2325 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK; 2326 rv = -EBUSY; 2327 goto out; 2328 } 2329 2330 switch (lkb->lkb_wait_type) { 2331 case DLM_MSG_LOOKUP: 2332 case DLM_MSG_REQUEST: 2333 lkb->lkb_flags |= DLM_IFL_OVERLAP_UNLOCK; 2334 rv = -EBUSY; 2335 goto out; 2336 case DLM_MSG_UNLOCK: 2337 goto out; 2338 } 2339 /* add_to_waiters() will set OVERLAP_UNLOCK */ 2340 goto out_ok; 2341 } 2342 2343 /* normal unlock not allowed if there's any op in progress */ 2344 rv = -EBUSY; 2345 if (lkb->lkb_wait_type || lkb->lkb_wait_count) 2346 goto out; 2347 2348 out_ok: 2349 /* an overlapping op shouldn't blow away exflags from other op */ 2350 lkb->lkb_exflags |= args->flags; 2351 lkb->lkb_sbflags = 0; 2352 lkb->lkb_astparam = args->astparam; 2353 rv = 0; 2354 out: 2355 if (rv) 2356 log_debug(ls, "validate_unlock_args %d %x %x %x %x %d %s", rv, 2357 lkb->lkb_id, lkb->lkb_flags, lkb->lkb_exflags, 2358 args->flags, lkb->lkb_wait_type, 2359 lkb->lkb_resource->res_name); 2360 return rv; 2361 } 2362 2363 /* 2364 * Four stage 4 varieties: 2365 * do_request(), do_convert(), do_unlock(), do_cancel() 2366 * These are called on the master node for the given lock and 2367 * from the central locking logic. 2368 */ 2369 2370 static int do_request(struct dlm_rsb *r, struct dlm_lkb *lkb) 2371 { 2372 int error = 0; 2373 2374 if (can_be_granted(r, lkb, 1, NULL)) { 2375 grant_lock(r, lkb); 2376 queue_cast(r, lkb, 0); 2377 goto out; 2378 } 2379 2380 if (can_be_queued(lkb)) { 2381 error = -EINPROGRESS; 2382 add_lkb(r, lkb, DLM_LKSTS_WAITING); 2383 add_timeout(lkb); 2384 goto out; 2385 } 2386 2387 error = -EAGAIN; 2388 queue_cast(r, lkb, -EAGAIN); 2389 out: 2390 return error; 2391 } 2392 2393 static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb, 2394 int error) 2395 { 2396 switch (error) { 2397 case -EAGAIN: 2398 if (force_blocking_asts(lkb)) 2399 send_blocking_asts_all(r, lkb); 2400 break; 2401 case -EINPROGRESS: 2402 send_blocking_asts(r, lkb); 2403 break; 2404 } 2405 } 2406 2407 static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb) 2408 { 2409 int error = 0; 2410 int deadlk = 0; 2411 2412 /* changing an existing lock may allow others to be granted */ 2413 2414 if (can_be_granted(r, lkb, 1, &deadlk)) { 2415 grant_lock(r, lkb); 2416 queue_cast(r, lkb, 0); 2417 goto out; 2418 } 2419 2420 /* can_be_granted() detected that this lock would block in a conversion 2421 deadlock, so we leave it on the granted queue and return EDEADLK in 2422 the ast for the convert. */ 2423 2424 if (deadlk) { 2425 /* it's left on the granted queue */ 2426 revert_lock(r, lkb); 2427 queue_cast(r, lkb, -EDEADLK); 2428 error = -EDEADLK; 2429 goto out; 2430 } 2431 2432 /* is_demoted() means the can_be_granted() above set the grmode 2433 to NL, and left us on the granted queue. This auto-demotion 2434 (due to CONVDEADLK) might mean other locks, and/or this lock, are 2435 now grantable. We have to try to grant other converting locks 2436 before we try again to grant this one. */ 2437 2438 if (is_demoted(lkb)) { 2439 grant_pending_convert(r, DLM_LOCK_IV, NULL); 2440 if (_can_be_granted(r, lkb, 1)) { 2441 grant_lock(r, lkb); 2442 queue_cast(r, lkb, 0); 2443 goto out; 2444 } 2445 /* else fall through and move to convert queue */ 2446 } 2447 2448 if (can_be_queued(lkb)) { 2449 error = -EINPROGRESS; 2450 del_lkb(r, lkb); 2451 add_lkb(r, lkb, DLM_LKSTS_CONVERT); 2452 add_timeout(lkb); 2453 goto out; 2454 } 2455 2456 error = -EAGAIN; 2457 queue_cast(r, lkb, -EAGAIN); 2458 out: 2459 return error; 2460 } 2461 2462 static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb, 2463 int error) 2464 { 2465 switch (error) { 2466 case 0: 2467 grant_pending_locks(r); 2468 /* grant_pending_locks also sends basts */ 2469 break; 2470 case -EAGAIN: 2471 if (force_blocking_asts(lkb)) 2472 send_blocking_asts_all(r, lkb); 2473 break; 2474 case -EINPROGRESS: 2475 send_blocking_asts(r, lkb); 2476 break; 2477 } 2478 } 2479 2480 static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb) 2481 { 2482 remove_lock(r, lkb); 2483 queue_cast(r, lkb, -DLM_EUNLOCK); 2484 return -DLM_EUNLOCK; 2485 } 2486 2487 static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb, 2488 int error) 2489 { 2490 grant_pending_locks(r); 2491 } 2492 2493 /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */ 2494 2495 static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb) 2496 { 2497 int error; 2498 2499 error = revert_lock(r, lkb); 2500 if (error) { 2501 queue_cast(r, lkb, -DLM_ECANCEL); 2502 return -DLM_ECANCEL; 2503 } 2504 return 0; 2505 } 2506 2507 static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb, 2508 int error) 2509 { 2510 if (error) 2511 grant_pending_locks(r); 2512 } 2513 2514 /* 2515 * Four stage 3 varieties: 2516 * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock() 2517 */ 2518 2519 /* add a new lkb to a possibly new rsb, called by requesting process */ 2520 2521 static int _request_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) 2522 { 2523 int error; 2524 2525 /* set_master: sets lkb nodeid from r */ 2526 2527 error = set_master(r, lkb); 2528 if (error < 0) 2529 goto out; 2530 if (error) { 2531 error = 0; 2532 goto out; 2533 } 2534 2535 if (is_remote(r)) { 2536 /* receive_request() calls do_request() on remote node */ 2537 error = send_request(r, lkb); 2538 } else { 2539 error = do_request(r, lkb); 2540 /* for remote locks the request_reply is sent 2541 between do_request and do_request_effects */ 2542 do_request_effects(r, lkb, error); 2543 } 2544 out: 2545 return error; 2546 } 2547 2548 /* change some property of an existing lkb, e.g. mode */ 2549 2550 static int _convert_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) 2551 { 2552 int error; 2553 2554 if (is_remote(r)) { 2555 /* receive_convert() calls do_convert() on remote node */ 2556 error = send_convert(r, lkb); 2557 } else { 2558 error = do_convert(r, lkb); 2559 /* for remote locks the convert_reply is sent 2560 between do_convert and do_convert_effects */ 2561 do_convert_effects(r, lkb, error); 2562 } 2563 2564 return error; 2565 } 2566 2567 /* remove an existing lkb from the granted queue */ 2568 2569 static int _unlock_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) 2570 { 2571 int error; 2572 2573 if (is_remote(r)) { 2574 /* receive_unlock() calls do_unlock() on remote node */ 2575 error = send_unlock(r, lkb); 2576 } else { 2577 error = do_unlock(r, lkb); 2578 /* for remote locks the unlock_reply is sent 2579 between do_unlock and do_unlock_effects */ 2580 do_unlock_effects(r, lkb, error); 2581 } 2582 2583 return error; 2584 } 2585 2586 /* remove an existing lkb from the convert or wait queue */ 2587 2588 static int _cancel_lock(struct dlm_rsb *r, struct dlm_lkb *lkb) 2589 { 2590 int error; 2591 2592 if (is_remote(r)) { 2593 /* receive_cancel() calls do_cancel() on remote node */ 2594 error = send_cancel(r, lkb); 2595 } else { 2596 error = do_cancel(r, lkb); 2597 /* for remote locks the cancel_reply is sent 2598 between do_cancel and do_cancel_effects */ 2599 do_cancel_effects(r, lkb, error); 2600 } 2601 2602 return error; 2603 } 2604 2605 /* 2606 * Four stage 2 varieties: 2607 * request_lock(), convert_lock(), unlock_lock(), cancel_lock() 2608 */ 2609 2610 static int request_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, char *name, 2611 int len, struct dlm_args *args) 2612 { 2613 struct dlm_rsb *r; 2614 int error; 2615 2616 error = validate_lock_args(ls, lkb, args); 2617 if (error) 2618 goto out; 2619 2620 error = find_rsb(ls, name, len, R_CREATE, &r); 2621 if (error) 2622 goto out; 2623 2624 lock_rsb(r); 2625 2626 attach_lkb(r, lkb); 2627 lkb->lkb_lksb->sb_lkid = lkb->lkb_id; 2628 2629 error = _request_lock(r, lkb); 2630 2631 unlock_rsb(r); 2632 put_rsb(r); 2633 2634 out: 2635 return error; 2636 } 2637 2638 static int convert_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, 2639 struct dlm_args *args) 2640 { 2641 struct dlm_rsb *r; 2642 int error; 2643 2644 r = lkb->lkb_resource; 2645 2646 hold_rsb(r); 2647 lock_rsb(r); 2648 2649 error = validate_lock_args(ls, lkb, args); 2650 if (error) 2651 goto out; 2652 2653 error = _convert_lock(r, lkb); 2654 out: 2655 unlock_rsb(r); 2656 put_rsb(r); 2657 return error; 2658 } 2659 2660 static int unlock_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, 2661 struct dlm_args *args) 2662 { 2663 struct dlm_rsb *r; 2664 int error; 2665 2666 r = lkb->lkb_resource; 2667 2668 hold_rsb(r); 2669 lock_rsb(r); 2670 2671 error = validate_unlock_args(lkb, args); 2672 if (error) 2673 goto out; 2674 2675 error = _unlock_lock(r, lkb); 2676 out: 2677 unlock_rsb(r); 2678 put_rsb(r); 2679 return error; 2680 } 2681 2682 static int cancel_lock(struct dlm_ls *ls, struct dlm_lkb *lkb, 2683 struct dlm_args *args) 2684 { 2685 struct dlm_rsb *r; 2686 int error; 2687 2688 r = lkb->lkb_resource; 2689 2690 hold_rsb(r); 2691 lock_rsb(r); 2692 2693 error = validate_unlock_args(lkb, args); 2694 if (error) 2695 goto out; 2696 2697 error = _cancel_lock(r, lkb); 2698 out: 2699 unlock_rsb(r); 2700 put_rsb(r); 2701 return error; 2702 } 2703 2704 /* 2705 * Two stage 1 varieties: dlm_lock() and dlm_unlock() 2706 */ 2707 2708 int dlm_lock(dlm_lockspace_t *lockspace, 2709 int mode, 2710 struct dlm_lksb *lksb, 2711 uint32_t flags, 2712 void *name, 2713 unsigned int namelen, 2714 uint32_t parent_lkid, 2715 void (*ast) (void *astarg), 2716 void *astarg, 2717 void (*bast) (void *astarg, int mode)) 2718 { 2719 struct dlm_ls *ls; 2720 struct dlm_lkb *lkb; 2721 struct dlm_args args; 2722 int error, convert = flags & DLM_LKF_CONVERT; 2723 2724 ls = dlm_find_lockspace_local(lockspace); 2725 if (!ls) 2726 return -EINVAL; 2727 2728 dlm_lock_recovery(ls); 2729 2730 if (convert) 2731 error = find_lkb(ls, lksb->sb_lkid, &lkb); 2732 else 2733 error = create_lkb(ls, &lkb); 2734 2735 if (error) 2736 goto out; 2737 2738 error = set_lock_args(mode, lksb, flags, namelen, 0, ast, 2739 astarg, bast, &args); 2740 if (error) 2741 goto out_put; 2742 2743 if (convert) 2744 error = convert_lock(ls, lkb, &args); 2745 else 2746 error = request_lock(ls, lkb, name, namelen, &args); 2747 2748 if (error == -EINPROGRESS) 2749 error = 0; 2750 out_put: 2751 if (convert || error) 2752 __put_lkb(ls, lkb); 2753 if (error == -EAGAIN || error == -EDEADLK) 2754 error = 0; 2755 out: 2756 dlm_unlock_recovery(ls); 2757 dlm_put_lockspace(ls); 2758 return error; 2759 } 2760 2761 int dlm_unlock(dlm_lockspace_t *lockspace, 2762 uint32_t lkid, 2763 uint32_t flags, 2764 struct dlm_lksb *lksb, 2765 void *astarg) 2766 { 2767 struct dlm_ls *ls; 2768 struct dlm_lkb *lkb; 2769 struct dlm_args args; 2770 int error; 2771 2772 ls = dlm_find_lockspace_local(lockspace); 2773 if (!ls) 2774 return -EINVAL; 2775 2776 dlm_lock_recovery(ls); 2777 2778 error = find_lkb(ls, lkid, &lkb); 2779 if (error) 2780 goto out; 2781 2782 error = set_unlock_args(flags, astarg, &args); 2783 if (error) 2784 goto out_put; 2785 2786 if (flags & DLM_LKF_CANCEL) 2787 error = cancel_lock(ls, lkb, &args); 2788 else 2789 error = unlock_lock(ls, lkb, &args); 2790 2791 if (error == -DLM_EUNLOCK || error == -DLM_ECANCEL) 2792 error = 0; 2793 if (error == -EBUSY && (flags & (DLM_LKF_CANCEL | DLM_LKF_FORCEUNLOCK))) 2794 error = 0; 2795 out_put: 2796 dlm_put_lkb(lkb); 2797 out: 2798 dlm_unlock_recovery(ls); 2799 dlm_put_lockspace(ls); 2800 return error; 2801 } 2802 2803 /* 2804 * send/receive routines for remote operations and replies 2805 * 2806 * send_args 2807 * send_common 2808 * send_request receive_request 2809 * send_convert receive_convert 2810 * send_unlock receive_unlock 2811 * send_cancel receive_cancel 2812 * send_grant receive_grant 2813 * send_bast receive_bast 2814 * send_lookup receive_lookup 2815 * send_remove receive_remove 2816 * 2817 * send_common_reply 2818 * receive_request_reply send_request_reply 2819 * receive_convert_reply send_convert_reply 2820 * receive_unlock_reply send_unlock_reply 2821 * receive_cancel_reply send_cancel_reply 2822 * receive_lookup_reply send_lookup_reply 2823 */ 2824 2825 static int _create_message(struct dlm_ls *ls, int mb_len, 2826 int to_nodeid, int mstype, 2827 struct dlm_message **ms_ret, 2828 struct dlm_mhandle **mh_ret) 2829 { 2830 struct dlm_message *ms; 2831 struct dlm_mhandle *mh; 2832 char *mb; 2833 2834 /* get_buffer gives us a message handle (mh) that we need to 2835 pass into lowcomms_commit and a message buffer (mb) that we 2836 write our data into */ 2837 2838 mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb); 2839 if (!mh) 2840 return -ENOBUFS; 2841 2842 memset(mb, 0, mb_len); 2843 2844 ms = (struct dlm_message *) mb; 2845 2846 ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); 2847 ms->m_header.h_lockspace = ls->ls_global_id; 2848 ms->m_header.h_nodeid = dlm_our_nodeid(); 2849 ms->m_header.h_length = mb_len; 2850 ms->m_header.h_cmd = DLM_MSG; 2851 2852 ms->m_type = mstype; 2853 2854 *mh_ret = mh; 2855 *ms_ret = ms; 2856 return 0; 2857 } 2858 2859 static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb, 2860 int to_nodeid, int mstype, 2861 struct dlm_message **ms_ret, 2862 struct dlm_mhandle **mh_ret) 2863 { 2864 int mb_len = sizeof(struct dlm_message); 2865 2866 switch (mstype) { 2867 case DLM_MSG_REQUEST: 2868 case DLM_MSG_LOOKUP: 2869 case DLM_MSG_REMOVE: 2870 mb_len += r->res_length; 2871 break; 2872 case DLM_MSG_CONVERT: 2873 case DLM_MSG_UNLOCK: 2874 case DLM_MSG_REQUEST_REPLY: 2875 case DLM_MSG_CONVERT_REPLY: 2876 case DLM_MSG_GRANT: 2877 if (lkb && lkb->lkb_lvbptr) 2878 mb_len += r->res_ls->ls_lvblen; 2879 break; 2880 } 2881 2882 return _create_message(r->res_ls, mb_len, to_nodeid, mstype, 2883 ms_ret, mh_ret); 2884 } 2885 2886 /* further lowcomms enhancements or alternate implementations may make 2887 the return value from this function useful at some point */ 2888 2889 static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms) 2890 { 2891 dlm_message_out(ms); 2892 dlm_lowcomms_commit_buffer(mh); 2893 return 0; 2894 } 2895 2896 static void send_args(struct dlm_rsb *r, struct dlm_lkb *lkb, 2897 struct dlm_message *ms) 2898 { 2899 ms->m_nodeid = lkb->lkb_nodeid; 2900 ms->m_pid = lkb->lkb_ownpid; 2901 ms->m_lkid = lkb->lkb_id; 2902 ms->m_remid = lkb->lkb_remid; 2903 ms->m_exflags = lkb->lkb_exflags; 2904 ms->m_sbflags = lkb->lkb_sbflags; 2905 ms->m_flags = lkb->lkb_flags; 2906 ms->m_lvbseq = lkb->lkb_lvbseq; 2907 ms->m_status = lkb->lkb_status; 2908 ms->m_grmode = lkb->lkb_grmode; 2909 ms->m_rqmode = lkb->lkb_rqmode; 2910 ms->m_hash = r->res_hash; 2911 2912 /* m_result and m_bastmode are set from function args, 2913 not from lkb fields */ 2914 2915 if (lkb->lkb_bastfn) 2916 ms->m_asts |= DLM_CB_BAST; 2917 if (lkb->lkb_astfn) 2918 ms->m_asts |= DLM_CB_CAST; 2919 2920 /* compare with switch in create_message; send_remove() doesn't 2921 use send_args() */ 2922 2923 switch (ms->m_type) { 2924 case DLM_MSG_REQUEST: 2925 case DLM_MSG_LOOKUP: 2926 memcpy(ms->m_extra, r->res_name, r->res_length); 2927 break; 2928 case DLM_MSG_CONVERT: 2929 case DLM_MSG_UNLOCK: 2930 case DLM_MSG_REQUEST_REPLY: 2931 case DLM_MSG_CONVERT_REPLY: 2932 case DLM_MSG_GRANT: 2933 if (!lkb->lkb_lvbptr) 2934 break; 2935 memcpy(ms->m_extra, lkb->lkb_lvbptr, r->res_ls->ls_lvblen); 2936 break; 2937 } 2938 } 2939 2940 static int send_common(struct dlm_rsb *r, struct dlm_lkb *lkb, int mstype) 2941 { 2942 struct dlm_message *ms; 2943 struct dlm_mhandle *mh; 2944 int to_nodeid, error; 2945 2946 to_nodeid = r->res_nodeid; 2947 2948 error = add_to_waiters(lkb, mstype, to_nodeid); 2949 if (error) 2950 return error; 2951 2952 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh); 2953 if (error) 2954 goto fail; 2955 2956 send_args(r, lkb, ms); 2957 2958 error = send_message(mh, ms); 2959 if (error) 2960 goto fail; 2961 return 0; 2962 2963 fail: 2964 remove_from_waiters(lkb, msg_reply_type(mstype)); 2965 return error; 2966 } 2967 2968 static int send_request(struct dlm_rsb *r, struct dlm_lkb *lkb) 2969 { 2970 return send_common(r, lkb, DLM_MSG_REQUEST); 2971 } 2972 2973 static int send_convert(struct dlm_rsb *r, struct dlm_lkb *lkb) 2974 { 2975 int error; 2976 2977 error = send_common(r, lkb, DLM_MSG_CONVERT); 2978 2979 /* down conversions go without a reply from the master */ 2980 if (!error && down_conversion(lkb)) { 2981 remove_from_waiters(lkb, DLM_MSG_CONVERT_REPLY); 2982 r->res_ls->ls_stub_ms.m_flags = DLM_IFL_STUB_MS; 2983 r->res_ls->ls_stub_ms.m_type = DLM_MSG_CONVERT_REPLY; 2984 r->res_ls->ls_stub_ms.m_result = 0; 2985 __receive_convert_reply(r, lkb, &r->res_ls->ls_stub_ms); 2986 } 2987 2988 return error; 2989 } 2990 2991 /* FIXME: if this lkb is the only lock we hold on the rsb, then set 2992 MASTER_UNCERTAIN to force the next request on the rsb to confirm 2993 that the master is still correct. */ 2994 2995 static int send_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb) 2996 { 2997 return send_common(r, lkb, DLM_MSG_UNLOCK); 2998 } 2999 3000 static int send_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb) 3001 { 3002 return send_common(r, lkb, DLM_MSG_CANCEL); 3003 } 3004 3005 static int send_grant(struct dlm_rsb *r, struct dlm_lkb *lkb) 3006 { 3007 struct dlm_message *ms; 3008 struct dlm_mhandle *mh; 3009 int to_nodeid, error; 3010 3011 to_nodeid = lkb->lkb_nodeid; 3012 3013 error = create_message(r, lkb, to_nodeid, DLM_MSG_GRANT, &ms, &mh); 3014 if (error) 3015 goto out; 3016 3017 send_args(r, lkb, ms); 3018 3019 ms->m_result = 0; 3020 3021 error = send_message(mh, ms); 3022 out: 3023 return error; 3024 } 3025 3026 static int send_bast(struct dlm_rsb *r, struct dlm_lkb *lkb, int mode) 3027 { 3028 struct dlm_message *ms; 3029 struct dlm_mhandle *mh; 3030 int to_nodeid, error; 3031 3032 to_nodeid = lkb->lkb_nodeid; 3033 3034 error = create_message(r, NULL, to_nodeid, DLM_MSG_BAST, &ms, &mh); 3035 if (error) 3036 goto out; 3037 3038 send_args(r, lkb, ms); 3039 3040 ms->m_bastmode = mode; 3041 3042 error = send_message(mh, ms); 3043 out: 3044 return error; 3045 } 3046 3047 static int send_lookup(struct dlm_rsb *r, struct dlm_lkb *lkb) 3048 { 3049 struct dlm_message *ms; 3050 struct dlm_mhandle *mh; 3051 int to_nodeid, error; 3052 3053 to_nodeid = dlm_dir_nodeid(r); 3054 3055 error = add_to_waiters(lkb, DLM_MSG_LOOKUP, to_nodeid); 3056 if (error) 3057 return error; 3058 3059 error = create_message(r, NULL, to_nodeid, DLM_MSG_LOOKUP, &ms, &mh); 3060 if (error) 3061 goto fail; 3062 3063 send_args(r, lkb, ms); 3064 3065 error = send_message(mh, ms); 3066 if (error) 3067 goto fail; 3068 return 0; 3069 3070 fail: 3071 remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY); 3072 return error; 3073 } 3074 3075 static int send_remove(struct dlm_rsb *r) 3076 { 3077 struct dlm_message *ms; 3078 struct dlm_mhandle *mh; 3079 int to_nodeid, error; 3080 3081 to_nodeid = dlm_dir_nodeid(r); 3082 3083 error = create_message(r, NULL, to_nodeid, DLM_MSG_REMOVE, &ms, &mh); 3084 if (error) 3085 goto out; 3086 3087 memcpy(ms->m_extra, r->res_name, r->res_length); 3088 ms->m_hash = r->res_hash; 3089 3090 error = send_message(mh, ms); 3091 out: 3092 return error; 3093 } 3094 3095 static int send_common_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, 3096 int mstype, int rv) 3097 { 3098 struct dlm_message *ms; 3099 struct dlm_mhandle *mh; 3100 int to_nodeid, error; 3101 3102 to_nodeid = lkb->lkb_nodeid; 3103 3104 error = create_message(r, lkb, to_nodeid, mstype, &ms, &mh); 3105 if (error) 3106 goto out; 3107 3108 send_args(r, lkb, ms); 3109 3110 ms->m_result = rv; 3111 3112 error = send_message(mh, ms); 3113 out: 3114 return error; 3115 } 3116 3117 static int send_request_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv) 3118 { 3119 return send_common_reply(r, lkb, DLM_MSG_REQUEST_REPLY, rv); 3120 } 3121 3122 static int send_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv) 3123 { 3124 return send_common_reply(r, lkb, DLM_MSG_CONVERT_REPLY, rv); 3125 } 3126 3127 static int send_unlock_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv) 3128 { 3129 return send_common_reply(r, lkb, DLM_MSG_UNLOCK_REPLY, rv); 3130 } 3131 3132 static int send_cancel_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, int rv) 3133 { 3134 return send_common_reply(r, lkb, DLM_MSG_CANCEL_REPLY, rv); 3135 } 3136 3137 static int send_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms_in, 3138 int ret_nodeid, int rv) 3139 { 3140 struct dlm_rsb *r = &ls->ls_stub_rsb; 3141 struct dlm_message *ms; 3142 struct dlm_mhandle *mh; 3143 int error, nodeid = ms_in->m_header.h_nodeid; 3144 3145 error = create_message(r, NULL, nodeid, DLM_MSG_LOOKUP_REPLY, &ms, &mh); 3146 if (error) 3147 goto out; 3148 3149 ms->m_lkid = ms_in->m_lkid; 3150 ms->m_result = rv; 3151 ms->m_nodeid = ret_nodeid; 3152 3153 error = send_message(mh, ms); 3154 out: 3155 return error; 3156 } 3157 3158 /* which args we save from a received message depends heavily on the type 3159 of message, unlike the send side where we can safely send everything about 3160 the lkb for any type of message */ 3161 3162 static void receive_flags(struct dlm_lkb *lkb, struct dlm_message *ms) 3163 { 3164 lkb->lkb_exflags = ms->m_exflags; 3165 lkb->lkb_sbflags = ms->m_sbflags; 3166 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) | 3167 (ms->m_flags & 0x0000FFFF); 3168 } 3169 3170 static void receive_flags_reply(struct dlm_lkb *lkb, struct dlm_message *ms) 3171 { 3172 if (ms->m_flags == DLM_IFL_STUB_MS) 3173 return; 3174 3175 lkb->lkb_sbflags = ms->m_sbflags; 3176 lkb->lkb_flags = (lkb->lkb_flags & 0xFFFF0000) | 3177 (ms->m_flags & 0x0000FFFF); 3178 } 3179 3180 static int receive_extralen(struct dlm_message *ms) 3181 { 3182 return (ms->m_header.h_length - sizeof(struct dlm_message)); 3183 } 3184 3185 static int receive_lvb(struct dlm_ls *ls, struct dlm_lkb *lkb, 3186 struct dlm_message *ms) 3187 { 3188 int len; 3189 3190 if (lkb->lkb_exflags & DLM_LKF_VALBLK) { 3191 if (!lkb->lkb_lvbptr) 3192 lkb->lkb_lvbptr = dlm_allocate_lvb(ls); 3193 if (!lkb->lkb_lvbptr) 3194 return -ENOMEM; 3195 len = receive_extralen(ms); 3196 if (len > DLM_RESNAME_MAXLEN) 3197 len = DLM_RESNAME_MAXLEN; 3198 memcpy(lkb->lkb_lvbptr, ms->m_extra, len); 3199 } 3200 return 0; 3201 } 3202 3203 static void fake_bastfn(void *astparam, int mode) 3204 { 3205 log_print("fake_bastfn should not be called"); 3206 } 3207 3208 static void fake_astfn(void *astparam) 3209 { 3210 log_print("fake_astfn should not be called"); 3211 } 3212 3213 static int receive_request_args(struct dlm_ls *ls, struct dlm_lkb *lkb, 3214 struct dlm_message *ms) 3215 { 3216 lkb->lkb_nodeid = ms->m_header.h_nodeid; 3217 lkb->lkb_ownpid = ms->m_pid; 3218 lkb->lkb_remid = ms->m_lkid; 3219 lkb->lkb_grmode = DLM_LOCK_IV; 3220 lkb->lkb_rqmode = ms->m_rqmode; 3221 3222 lkb->lkb_bastfn = (ms->m_asts & DLM_CB_BAST) ? &fake_bastfn : NULL; 3223 lkb->lkb_astfn = (ms->m_asts & DLM_CB_CAST) ? &fake_astfn : NULL; 3224 3225 if (lkb->lkb_exflags & DLM_LKF_VALBLK) { 3226 /* lkb was just created so there won't be an lvb yet */ 3227 lkb->lkb_lvbptr = dlm_allocate_lvb(ls); 3228 if (!lkb->lkb_lvbptr) 3229 return -ENOMEM; 3230 } 3231 3232 return 0; 3233 } 3234 3235 static int receive_convert_args(struct dlm_ls *ls, struct dlm_lkb *lkb, 3236 struct dlm_message *ms) 3237 { 3238 if (lkb->lkb_status != DLM_LKSTS_GRANTED) 3239 return -EBUSY; 3240 3241 if (receive_lvb(ls, lkb, ms)) 3242 return -ENOMEM; 3243 3244 lkb->lkb_rqmode = ms->m_rqmode; 3245 lkb->lkb_lvbseq = ms->m_lvbseq; 3246 3247 return 0; 3248 } 3249 3250 static int receive_unlock_args(struct dlm_ls *ls, struct dlm_lkb *lkb, 3251 struct dlm_message *ms) 3252 { 3253 if (receive_lvb(ls, lkb, ms)) 3254 return -ENOMEM; 3255 return 0; 3256 } 3257 3258 /* We fill in the stub-lkb fields with the info that send_xxxx_reply() 3259 uses to send a reply and that the remote end uses to process the reply. */ 3260 3261 static void setup_stub_lkb(struct dlm_ls *ls, struct dlm_message *ms) 3262 { 3263 struct dlm_lkb *lkb = &ls->ls_stub_lkb; 3264 lkb->lkb_nodeid = ms->m_header.h_nodeid; 3265 lkb->lkb_remid = ms->m_lkid; 3266 } 3267 3268 /* This is called after the rsb is locked so that we can safely inspect 3269 fields in the lkb. */ 3270 3271 static int validate_message(struct dlm_lkb *lkb, struct dlm_message *ms) 3272 { 3273 int from = ms->m_header.h_nodeid; 3274 int error = 0; 3275 3276 switch (ms->m_type) { 3277 case DLM_MSG_CONVERT: 3278 case DLM_MSG_UNLOCK: 3279 case DLM_MSG_CANCEL: 3280 if (!is_master_copy(lkb) || lkb->lkb_nodeid != from) 3281 error = -EINVAL; 3282 break; 3283 3284 case DLM_MSG_CONVERT_REPLY: 3285 case DLM_MSG_UNLOCK_REPLY: 3286 case DLM_MSG_CANCEL_REPLY: 3287 case DLM_MSG_GRANT: 3288 case DLM_MSG_BAST: 3289 if (!is_process_copy(lkb) || lkb->lkb_nodeid != from) 3290 error = -EINVAL; 3291 break; 3292 3293 case DLM_MSG_REQUEST_REPLY: 3294 if (!is_process_copy(lkb)) 3295 error = -EINVAL; 3296 else if (lkb->lkb_nodeid != -1 && lkb->lkb_nodeid != from) 3297 error = -EINVAL; 3298 break; 3299 3300 default: 3301 error = -EINVAL; 3302 } 3303 3304 if (error) 3305 log_error(lkb->lkb_resource->res_ls, 3306 "ignore invalid message %d from %d %x %x %x %d", 3307 ms->m_type, from, lkb->lkb_id, lkb->lkb_remid, 3308 lkb->lkb_flags, lkb->lkb_nodeid); 3309 return error; 3310 } 3311 3312 static void receive_request(struct dlm_ls *ls, struct dlm_message *ms) 3313 { 3314 struct dlm_lkb *lkb; 3315 struct dlm_rsb *r; 3316 int error, namelen; 3317 3318 error = create_lkb(ls, &lkb); 3319 if (error) 3320 goto fail; 3321 3322 receive_flags(lkb, ms); 3323 lkb->lkb_flags |= DLM_IFL_MSTCPY; 3324 error = receive_request_args(ls, lkb, ms); 3325 if (error) { 3326 __put_lkb(ls, lkb); 3327 goto fail; 3328 } 3329 3330 namelen = receive_extralen(ms); 3331 3332 error = find_rsb(ls, ms->m_extra, namelen, R_MASTER, &r); 3333 if (error) { 3334 __put_lkb(ls, lkb); 3335 goto fail; 3336 } 3337 3338 lock_rsb(r); 3339 3340 attach_lkb(r, lkb); 3341 error = do_request(r, lkb); 3342 send_request_reply(r, lkb, error); 3343 do_request_effects(r, lkb, error); 3344 3345 unlock_rsb(r); 3346 put_rsb(r); 3347 3348 if (error == -EINPROGRESS) 3349 error = 0; 3350 if (error) 3351 dlm_put_lkb(lkb); 3352 return; 3353 3354 fail: 3355 setup_stub_lkb(ls, ms); 3356 send_request_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error); 3357 } 3358 3359 static void receive_convert(struct dlm_ls *ls, struct dlm_message *ms) 3360 { 3361 struct dlm_lkb *lkb; 3362 struct dlm_rsb *r; 3363 int error, reply = 1; 3364 3365 error = find_lkb(ls, ms->m_remid, &lkb); 3366 if (error) 3367 goto fail; 3368 3369 r = lkb->lkb_resource; 3370 3371 hold_rsb(r); 3372 lock_rsb(r); 3373 3374 error = validate_message(lkb, ms); 3375 if (error) 3376 goto out; 3377 3378 receive_flags(lkb, ms); 3379 3380 error = receive_convert_args(ls, lkb, ms); 3381 if (error) { 3382 send_convert_reply(r, lkb, error); 3383 goto out; 3384 } 3385 3386 reply = !down_conversion(lkb); 3387 3388 error = do_convert(r, lkb); 3389 if (reply) 3390 send_convert_reply(r, lkb, error); 3391 do_convert_effects(r, lkb, error); 3392 out: 3393 unlock_rsb(r); 3394 put_rsb(r); 3395 dlm_put_lkb(lkb); 3396 return; 3397 3398 fail: 3399 setup_stub_lkb(ls, ms); 3400 send_convert_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error); 3401 } 3402 3403 static void receive_unlock(struct dlm_ls *ls, struct dlm_message *ms) 3404 { 3405 struct dlm_lkb *lkb; 3406 struct dlm_rsb *r; 3407 int error; 3408 3409 error = find_lkb(ls, ms->m_remid, &lkb); 3410 if (error) 3411 goto fail; 3412 3413 r = lkb->lkb_resource; 3414 3415 hold_rsb(r); 3416 lock_rsb(r); 3417 3418 error = validate_message(lkb, ms); 3419 if (error) 3420 goto out; 3421 3422 receive_flags(lkb, ms); 3423 3424 error = receive_unlock_args(ls, lkb, ms); 3425 if (error) { 3426 send_unlock_reply(r, lkb, error); 3427 goto out; 3428 } 3429 3430 error = do_unlock(r, lkb); 3431 send_unlock_reply(r, lkb, error); 3432 do_unlock_effects(r, lkb, error); 3433 out: 3434 unlock_rsb(r); 3435 put_rsb(r); 3436 dlm_put_lkb(lkb); 3437 return; 3438 3439 fail: 3440 setup_stub_lkb(ls, ms); 3441 send_unlock_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error); 3442 } 3443 3444 static void receive_cancel(struct dlm_ls *ls, struct dlm_message *ms) 3445 { 3446 struct dlm_lkb *lkb; 3447 struct dlm_rsb *r; 3448 int error; 3449 3450 error = find_lkb(ls, ms->m_remid, &lkb); 3451 if (error) 3452 goto fail; 3453 3454 receive_flags(lkb, ms); 3455 3456 r = lkb->lkb_resource; 3457 3458 hold_rsb(r); 3459 lock_rsb(r); 3460 3461 error = validate_message(lkb, ms); 3462 if (error) 3463 goto out; 3464 3465 error = do_cancel(r, lkb); 3466 send_cancel_reply(r, lkb, error); 3467 do_cancel_effects(r, lkb, error); 3468 out: 3469 unlock_rsb(r); 3470 put_rsb(r); 3471 dlm_put_lkb(lkb); 3472 return; 3473 3474 fail: 3475 setup_stub_lkb(ls, ms); 3476 send_cancel_reply(&ls->ls_stub_rsb, &ls->ls_stub_lkb, error); 3477 } 3478 3479 static void receive_grant(struct dlm_ls *ls, struct dlm_message *ms) 3480 { 3481 struct dlm_lkb *lkb; 3482 struct dlm_rsb *r; 3483 int error; 3484 3485 error = find_lkb(ls, ms->m_remid, &lkb); 3486 if (error) { 3487 log_debug(ls, "receive_grant from %d no lkb %x", 3488 ms->m_header.h_nodeid, ms->m_remid); 3489 return; 3490 } 3491 3492 r = lkb->lkb_resource; 3493 3494 hold_rsb(r); 3495 lock_rsb(r); 3496 3497 error = validate_message(lkb, ms); 3498 if (error) 3499 goto out; 3500 3501 receive_flags_reply(lkb, ms); 3502 if (is_altmode(lkb)) 3503 munge_altmode(lkb, ms); 3504 grant_lock_pc(r, lkb, ms); 3505 queue_cast(r, lkb, 0); 3506 out: 3507 unlock_rsb(r); 3508 put_rsb(r); 3509 dlm_put_lkb(lkb); 3510 } 3511 3512 static void receive_bast(struct dlm_ls *ls, struct dlm_message *ms) 3513 { 3514 struct dlm_lkb *lkb; 3515 struct dlm_rsb *r; 3516 int error; 3517 3518 error = find_lkb(ls, ms->m_remid, &lkb); 3519 if (error) { 3520 log_debug(ls, "receive_bast from %d no lkb %x", 3521 ms->m_header.h_nodeid, ms->m_remid); 3522 return; 3523 } 3524 3525 r = lkb->lkb_resource; 3526 3527 hold_rsb(r); 3528 lock_rsb(r); 3529 3530 error = validate_message(lkb, ms); 3531 if (error) 3532 goto out; 3533 3534 queue_bast(r, lkb, ms->m_bastmode); 3535 out: 3536 unlock_rsb(r); 3537 put_rsb(r); 3538 dlm_put_lkb(lkb); 3539 } 3540 3541 static void receive_lookup(struct dlm_ls *ls, struct dlm_message *ms) 3542 { 3543 int len, error, ret_nodeid, dir_nodeid, from_nodeid, our_nodeid; 3544 3545 from_nodeid = ms->m_header.h_nodeid; 3546 our_nodeid = dlm_our_nodeid(); 3547 3548 len = receive_extralen(ms); 3549 3550 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash); 3551 if (dir_nodeid != our_nodeid) { 3552 log_error(ls, "lookup dir_nodeid %d from %d", 3553 dir_nodeid, from_nodeid); 3554 error = -EINVAL; 3555 ret_nodeid = -1; 3556 goto out; 3557 } 3558 3559 error = dlm_dir_lookup(ls, from_nodeid, ms->m_extra, len, &ret_nodeid); 3560 3561 /* Optimization: we're master so treat lookup as a request */ 3562 if (!error && ret_nodeid == our_nodeid) { 3563 receive_request(ls, ms); 3564 return; 3565 } 3566 out: 3567 send_lookup_reply(ls, ms, ret_nodeid, error); 3568 } 3569 3570 static void receive_remove(struct dlm_ls *ls, struct dlm_message *ms) 3571 { 3572 int len, dir_nodeid, from_nodeid; 3573 3574 from_nodeid = ms->m_header.h_nodeid; 3575 3576 len = receive_extralen(ms); 3577 3578 dir_nodeid = dlm_hash2nodeid(ls, ms->m_hash); 3579 if (dir_nodeid != dlm_our_nodeid()) { 3580 log_error(ls, "remove dir entry dir_nodeid %d from %d", 3581 dir_nodeid, from_nodeid); 3582 return; 3583 } 3584 3585 dlm_dir_remove_entry(ls, from_nodeid, ms->m_extra, len); 3586 } 3587 3588 static void receive_purge(struct dlm_ls *ls, struct dlm_message *ms) 3589 { 3590 do_purge(ls, ms->m_nodeid, ms->m_pid); 3591 } 3592 3593 static void receive_request_reply(struct dlm_ls *ls, struct dlm_message *ms) 3594 { 3595 struct dlm_lkb *lkb; 3596 struct dlm_rsb *r; 3597 int error, mstype, result; 3598 3599 error = find_lkb(ls, ms->m_remid, &lkb); 3600 if (error) { 3601 log_debug(ls, "receive_request_reply from %d no lkb %x", 3602 ms->m_header.h_nodeid, ms->m_remid); 3603 return; 3604 } 3605 3606 r = lkb->lkb_resource; 3607 hold_rsb(r); 3608 lock_rsb(r); 3609 3610 error = validate_message(lkb, ms); 3611 if (error) 3612 goto out; 3613 3614 mstype = lkb->lkb_wait_type; 3615 error = remove_from_waiters(lkb, DLM_MSG_REQUEST_REPLY); 3616 if (error) 3617 goto out; 3618 3619 /* Optimization: the dir node was also the master, so it took our 3620 lookup as a request and sent request reply instead of lookup reply */ 3621 if (mstype == DLM_MSG_LOOKUP) { 3622 r->res_nodeid = ms->m_header.h_nodeid; 3623 lkb->lkb_nodeid = r->res_nodeid; 3624 } 3625 3626 /* this is the value returned from do_request() on the master */ 3627 result = ms->m_result; 3628 3629 switch (result) { 3630 case -EAGAIN: 3631 /* request would block (be queued) on remote master */ 3632 queue_cast(r, lkb, -EAGAIN); 3633 confirm_master(r, -EAGAIN); 3634 unhold_lkb(lkb); /* undoes create_lkb() */ 3635 break; 3636 3637 case -EINPROGRESS: 3638 case 0: 3639 /* request was queued or granted on remote master */ 3640 receive_flags_reply(lkb, ms); 3641 lkb->lkb_remid = ms->m_lkid; 3642 if (is_altmode(lkb)) 3643 munge_altmode(lkb, ms); 3644 if (result) { 3645 add_lkb(r, lkb, DLM_LKSTS_WAITING); 3646 add_timeout(lkb); 3647 } else { 3648 grant_lock_pc(r, lkb, ms); 3649 queue_cast(r, lkb, 0); 3650 } 3651 confirm_master(r, result); 3652 break; 3653 3654 case -EBADR: 3655 case -ENOTBLK: 3656 /* find_rsb failed to find rsb or rsb wasn't master */ 3657 log_debug(ls, "receive_request_reply %x %x master diff %d %d", 3658 lkb->lkb_id, lkb->lkb_flags, r->res_nodeid, result); 3659 r->res_nodeid = -1; 3660 lkb->lkb_nodeid = -1; 3661 3662 if (is_overlap(lkb)) { 3663 /* we'll ignore error in cancel/unlock reply */ 3664 queue_cast_overlap(r, lkb); 3665 confirm_master(r, result); 3666 unhold_lkb(lkb); /* undoes create_lkb() */ 3667 } else 3668 _request_lock(r, lkb); 3669 break; 3670 3671 default: 3672 log_error(ls, "receive_request_reply %x error %d", 3673 lkb->lkb_id, result); 3674 } 3675 3676 if (is_overlap_unlock(lkb) && (result == 0 || result == -EINPROGRESS)) { 3677 log_debug(ls, "receive_request_reply %x result %d unlock", 3678 lkb->lkb_id, result); 3679 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK; 3680 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL; 3681 send_unlock(r, lkb); 3682 } else if (is_overlap_cancel(lkb) && (result == -EINPROGRESS)) { 3683 log_debug(ls, "receive_request_reply %x cancel", lkb->lkb_id); 3684 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK; 3685 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL; 3686 send_cancel(r, lkb); 3687 } else { 3688 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL; 3689 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK; 3690 } 3691 out: 3692 unlock_rsb(r); 3693 put_rsb(r); 3694 dlm_put_lkb(lkb); 3695 } 3696 3697 static void __receive_convert_reply(struct dlm_rsb *r, struct dlm_lkb *lkb, 3698 struct dlm_message *ms) 3699 { 3700 /* this is the value returned from do_convert() on the master */ 3701 switch (ms->m_result) { 3702 case -EAGAIN: 3703 /* convert would block (be queued) on remote master */ 3704 queue_cast(r, lkb, -EAGAIN); 3705 break; 3706 3707 case -EDEADLK: 3708 receive_flags_reply(lkb, ms); 3709 revert_lock_pc(r, lkb); 3710 queue_cast(r, lkb, -EDEADLK); 3711 break; 3712 3713 case -EINPROGRESS: 3714 /* convert was queued on remote master */ 3715 receive_flags_reply(lkb, ms); 3716 if (is_demoted(lkb)) 3717 munge_demoted(lkb); 3718 del_lkb(r, lkb); 3719 add_lkb(r, lkb, DLM_LKSTS_CONVERT); 3720 add_timeout(lkb); 3721 break; 3722 3723 case 0: 3724 /* convert was granted on remote master */ 3725 receive_flags_reply(lkb, ms); 3726 if (is_demoted(lkb)) 3727 munge_demoted(lkb); 3728 grant_lock_pc(r, lkb, ms); 3729 queue_cast(r, lkb, 0); 3730 break; 3731 3732 default: 3733 log_error(r->res_ls, "receive_convert_reply %x error %d", 3734 lkb->lkb_id, ms->m_result); 3735 } 3736 } 3737 3738 static void _receive_convert_reply(struct dlm_lkb *lkb, struct dlm_message *ms) 3739 { 3740 struct dlm_rsb *r = lkb->lkb_resource; 3741 int error; 3742 3743 hold_rsb(r); 3744 lock_rsb(r); 3745 3746 error = validate_message(lkb, ms); 3747 if (error) 3748 goto out; 3749 3750 /* stub reply can happen with waiters_mutex held */ 3751 error = remove_from_waiters_ms(lkb, ms); 3752 if (error) 3753 goto out; 3754 3755 __receive_convert_reply(r, lkb, ms); 3756 out: 3757 unlock_rsb(r); 3758 put_rsb(r); 3759 } 3760 3761 static void receive_convert_reply(struct dlm_ls *ls, struct dlm_message *ms) 3762 { 3763 struct dlm_lkb *lkb; 3764 int error; 3765 3766 error = find_lkb(ls, ms->m_remid, &lkb); 3767 if (error) { 3768 log_debug(ls, "receive_convert_reply from %d no lkb %x", 3769 ms->m_header.h_nodeid, ms->m_remid); 3770 return; 3771 } 3772 3773 _receive_convert_reply(lkb, ms); 3774 dlm_put_lkb(lkb); 3775 } 3776 3777 static void _receive_unlock_reply(struct dlm_lkb *lkb, struct dlm_message *ms) 3778 { 3779 struct dlm_rsb *r = lkb->lkb_resource; 3780 int error; 3781 3782 hold_rsb(r); 3783 lock_rsb(r); 3784 3785 error = validate_message(lkb, ms); 3786 if (error) 3787 goto out; 3788 3789 /* stub reply can happen with waiters_mutex held */ 3790 error = remove_from_waiters_ms(lkb, ms); 3791 if (error) 3792 goto out; 3793 3794 /* this is the value returned from do_unlock() on the master */ 3795 3796 switch (ms->m_result) { 3797 case -DLM_EUNLOCK: 3798 receive_flags_reply(lkb, ms); 3799 remove_lock_pc(r, lkb); 3800 queue_cast(r, lkb, -DLM_EUNLOCK); 3801 break; 3802 case -ENOENT: 3803 break; 3804 default: 3805 log_error(r->res_ls, "receive_unlock_reply %x error %d", 3806 lkb->lkb_id, ms->m_result); 3807 } 3808 out: 3809 unlock_rsb(r); 3810 put_rsb(r); 3811 } 3812 3813 static void receive_unlock_reply(struct dlm_ls *ls, struct dlm_message *ms) 3814 { 3815 struct dlm_lkb *lkb; 3816 int error; 3817 3818 error = find_lkb(ls, ms->m_remid, &lkb); 3819 if (error) { 3820 log_debug(ls, "receive_unlock_reply from %d no lkb %x", 3821 ms->m_header.h_nodeid, ms->m_remid); 3822 return; 3823 } 3824 3825 _receive_unlock_reply(lkb, ms); 3826 dlm_put_lkb(lkb); 3827 } 3828 3829 static void _receive_cancel_reply(struct dlm_lkb *lkb, struct dlm_message *ms) 3830 { 3831 struct dlm_rsb *r = lkb->lkb_resource; 3832 int error; 3833 3834 hold_rsb(r); 3835 lock_rsb(r); 3836 3837 error = validate_message(lkb, ms); 3838 if (error) 3839 goto out; 3840 3841 /* stub reply can happen with waiters_mutex held */ 3842 error = remove_from_waiters_ms(lkb, ms); 3843 if (error) 3844 goto out; 3845 3846 /* this is the value returned from do_cancel() on the master */ 3847 3848 switch (ms->m_result) { 3849 case -DLM_ECANCEL: 3850 receive_flags_reply(lkb, ms); 3851 revert_lock_pc(r, lkb); 3852 queue_cast(r, lkb, -DLM_ECANCEL); 3853 break; 3854 case 0: 3855 break; 3856 default: 3857 log_error(r->res_ls, "receive_cancel_reply %x error %d", 3858 lkb->lkb_id, ms->m_result); 3859 } 3860 out: 3861 unlock_rsb(r); 3862 put_rsb(r); 3863 } 3864 3865 static void receive_cancel_reply(struct dlm_ls *ls, struct dlm_message *ms) 3866 { 3867 struct dlm_lkb *lkb; 3868 int error; 3869 3870 error = find_lkb(ls, ms->m_remid, &lkb); 3871 if (error) { 3872 log_debug(ls, "receive_cancel_reply from %d no lkb %x", 3873 ms->m_header.h_nodeid, ms->m_remid); 3874 return; 3875 } 3876 3877 _receive_cancel_reply(lkb, ms); 3878 dlm_put_lkb(lkb); 3879 } 3880 3881 static void receive_lookup_reply(struct dlm_ls *ls, struct dlm_message *ms) 3882 { 3883 struct dlm_lkb *lkb; 3884 struct dlm_rsb *r; 3885 int error, ret_nodeid; 3886 3887 error = find_lkb(ls, ms->m_lkid, &lkb); 3888 if (error) { 3889 log_error(ls, "receive_lookup_reply no lkb"); 3890 return; 3891 } 3892 3893 /* ms->m_result is the value returned by dlm_dir_lookup on dir node 3894 FIXME: will a non-zero error ever be returned? */ 3895 3896 r = lkb->lkb_resource; 3897 hold_rsb(r); 3898 lock_rsb(r); 3899 3900 error = remove_from_waiters(lkb, DLM_MSG_LOOKUP_REPLY); 3901 if (error) 3902 goto out; 3903 3904 ret_nodeid = ms->m_nodeid; 3905 if (ret_nodeid == dlm_our_nodeid()) { 3906 r->res_nodeid = 0; 3907 ret_nodeid = 0; 3908 r->res_first_lkid = 0; 3909 } else { 3910 /* set_master() will copy res_nodeid to lkb_nodeid */ 3911 r->res_nodeid = ret_nodeid; 3912 } 3913 3914 if (is_overlap(lkb)) { 3915 log_debug(ls, "receive_lookup_reply %x unlock %x", 3916 lkb->lkb_id, lkb->lkb_flags); 3917 queue_cast_overlap(r, lkb); 3918 unhold_lkb(lkb); /* undoes create_lkb() */ 3919 goto out_list; 3920 } 3921 3922 _request_lock(r, lkb); 3923 3924 out_list: 3925 if (!ret_nodeid) 3926 process_lookup_list(r); 3927 out: 3928 unlock_rsb(r); 3929 put_rsb(r); 3930 dlm_put_lkb(lkb); 3931 } 3932 3933 static void _receive_message(struct dlm_ls *ls, struct dlm_message *ms) 3934 { 3935 if (!dlm_is_member(ls, ms->m_header.h_nodeid)) { 3936 log_debug(ls, "ignore non-member message %d from %d %x %x %d", 3937 ms->m_type, ms->m_header.h_nodeid, ms->m_lkid, 3938 ms->m_remid, ms->m_result); 3939 return; 3940 } 3941 3942 switch (ms->m_type) { 3943 3944 /* messages sent to a master node */ 3945 3946 case DLM_MSG_REQUEST: 3947 receive_request(ls, ms); 3948 break; 3949 3950 case DLM_MSG_CONVERT: 3951 receive_convert(ls, ms); 3952 break; 3953 3954 case DLM_MSG_UNLOCK: 3955 receive_unlock(ls, ms); 3956 break; 3957 3958 case DLM_MSG_CANCEL: 3959 receive_cancel(ls, ms); 3960 break; 3961 3962 /* messages sent from a master node (replies to above) */ 3963 3964 case DLM_MSG_REQUEST_REPLY: 3965 receive_request_reply(ls, ms); 3966 break; 3967 3968 case DLM_MSG_CONVERT_REPLY: 3969 receive_convert_reply(ls, ms); 3970 break; 3971 3972 case DLM_MSG_UNLOCK_REPLY: 3973 receive_unlock_reply(ls, ms); 3974 break; 3975 3976 case DLM_MSG_CANCEL_REPLY: 3977 receive_cancel_reply(ls, ms); 3978 break; 3979 3980 /* messages sent from a master node (only two types of async msg) */ 3981 3982 case DLM_MSG_GRANT: 3983 receive_grant(ls, ms); 3984 break; 3985 3986 case DLM_MSG_BAST: 3987 receive_bast(ls, ms); 3988 break; 3989 3990 /* messages sent to a dir node */ 3991 3992 case DLM_MSG_LOOKUP: 3993 receive_lookup(ls, ms); 3994 break; 3995 3996 case DLM_MSG_REMOVE: 3997 receive_remove(ls, ms); 3998 break; 3999 4000 /* messages sent from a dir node (remove has no reply) */ 4001 4002 case DLM_MSG_LOOKUP_REPLY: 4003 receive_lookup_reply(ls, ms); 4004 break; 4005 4006 /* other messages */ 4007 4008 case DLM_MSG_PURGE: 4009 receive_purge(ls, ms); 4010 break; 4011 4012 default: 4013 log_error(ls, "unknown message type %d", ms->m_type); 4014 } 4015 } 4016 4017 /* If the lockspace is in recovery mode (locking stopped), then normal 4018 messages are saved on the requestqueue for processing after recovery is 4019 done. When not in recovery mode, we wait for dlm_recoverd to drain saved 4020 messages off the requestqueue before we process new ones. This occurs right 4021 after recovery completes when we transition from saving all messages on 4022 requestqueue, to processing all the saved messages, to processing new 4023 messages as they arrive. */ 4024 4025 static void dlm_receive_message(struct dlm_ls *ls, struct dlm_message *ms, 4026 int nodeid) 4027 { 4028 if (dlm_locking_stopped(ls)) { 4029 dlm_add_requestqueue(ls, nodeid, ms); 4030 } else { 4031 dlm_wait_requestqueue(ls); 4032 _receive_message(ls, ms); 4033 } 4034 } 4035 4036 /* This is called by dlm_recoverd to process messages that were saved on 4037 the requestqueue. */ 4038 4039 void dlm_receive_message_saved(struct dlm_ls *ls, struct dlm_message *ms) 4040 { 4041 _receive_message(ls, ms); 4042 } 4043 4044 /* This is called by the midcomms layer when something is received for 4045 the lockspace. It could be either a MSG (normal message sent as part of 4046 standard locking activity) or an RCOM (recovery message sent as part of 4047 lockspace recovery). */ 4048 4049 void dlm_receive_buffer(union dlm_packet *p, int nodeid) 4050 { 4051 struct dlm_header *hd = &p->header; 4052 struct dlm_ls *ls; 4053 int type = 0; 4054 4055 switch (hd->h_cmd) { 4056 case DLM_MSG: 4057 dlm_message_in(&p->message); 4058 type = p->message.m_type; 4059 break; 4060 case DLM_RCOM: 4061 dlm_rcom_in(&p->rcom); 4062 type = p->rcom.rc_type; 4063 break; 4064 default: 4065 log_print("invalid h_cmd %d from %u", hd->h_cmd, nodeid); 4066 return; 4067 } 4068 4069 if (hd->h_nodeid != nodeid) { 4070 log_print("invalid h_nodeid %d from %d lockspace %x", 4071 hd->h_nodeid, nodeid, hd->h_lockspace); 4072 return; 4073 } 4074 4075 ls = dlm_find_lockspace_global(hd->h_lockspace); 4076 if (!ls) { 4077 if (dlm_config.ci_log_debug) 4078 log_print("invalid lockspace %x from %d cmd %d type %d", 4079 hd->h_lockspace, nodeid, hd->h_cmd, type); 4080 4081 if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS) 4082 dlm_send_ls_not_ready(nodeid, &p->rcom); 4083 return; 4084 } 4085 4086 /* this rwsem allows dlm_ls_stop() to wait for all dlm_recv threads to 4087 be inactive (in this ls) before transitioning to recovery mode */ 4088 4089 down_read(&ls->ls_recv_active); 4090 if (hd->h_cmd == DLM_MSG) 4091 dlm_receive_message(ls, &p->message, nodeid); 4092 else 4093 dlm_receive_rcom(ls, &p->rcom, nodeid); 4094 up_read(&ls->ls_recv_active); 4095 4096 dlm_put_lockspace(ls); 4097 } 4098 4099 static void recover_convert_waiter(struct dlm_ls *ls, struct dlm_lkb *lkb, 4100 struct dlm_message *ms_stub) 4101 { 4102 if (middle_conversion(lkb)) { 4103 hold_lkb(lkb); 4104 memset(ms_stub, 0, sizeof(struct dlm_message)); 4105 ms_stub->m_flags = DLM_IFL_STUB_MS; 4106 ms_stub->m_type = DLM_MSG_CONVERT_REPLY; 4107 ms_stub->m_result = -EINPROGRESS; 4108 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid; 4109 _receive_convert_reply(lkb, ms_stub); 4110 4111 /* Same special case as in receive_rcom_lock_args() */ 4112 lkb->lkb_grmode = DLM_LOCK_IV; 4113 rsb_set_flag(lkb->lkb_resource, RSB_RECOVER_CONVERT); 4114 unhold_lkb(lkb); 4115 4116 } else if (lkb->lkb_rqmode >= lkb->lkb_grmode) { 4117 lkb->lkb_flags |= DLM_IFL_RESEND; 4118 } 4119 4120 /* lkb->lkb_rqmode < lkb->lkb_grmode shouldn't happen since down 4121 conversions are async; there's no reply from the remote master */ 4122 } 4123 4124 /* A waiting lkb needs recovery if the master node has failed, or 4125 the master node is changing (only when no directory is used) */ 4126 4127 static int waiter_needs_recovery(struct dlm_ls *ls, struct dlm_lkb *lkb) 4128 { 4129 if (dlm_is_removed(ls, lkb->lkb_nodeid)) 4130 return 1; 4131 4132 if (!dlm_no_directory(ls)) 4133 return 0; 4134 4135 if (dlm_dir_nodeid(lkb->lkb_resource) != lkb->lkb_nodeid) 4136 return 1; 4137 4138 return 0; 4139 } 4140 4141 /* Recovery for locks that are waiting for replies from nodes that are now 4142 gone. We can just complete unlocks and cancels by faking a reply from the 4143 dead node. Requests and up-conversions we flag to be resent after 4144 recovery. Down-conversions can just be completed with a fake reply like 4145 unlocks. Conversions between PR and CW need special attention. */ 4146 4147 void dlm_recover_waiters_pre(struct dlm_ls *ls) 4148 { 4149 struct dlm_lkb *lkb, *safe; 4150 struct dlm_message *ms_stub; 4151 int wait_type, stub_unlock_result, stub_cancel_result; 4152 4153 ms_stub = kmalloc(sizeof(struct dlm_message), GFP_KERNEL); 4154 if (!ms_stub) { 4155 log_error(ls, "dlm_recover_waiters_pre no mem"); 4156 return; 4157 } 4158 4159 mutex_lock(&ls->ls_waiters_mutex); 4160 4161 list_for_each_entry_safe(lkb, safe, &ls->ls_waiters, lkb_wait_reply) { 4162 4163 /* exclude debug messages about unlocks because there can be so 4164 many and they aren't very interesting */ 4165 4166 if (lkb->lkb_wait_type != DLM_MSG_UNLOCK) { 4167 log_debug(ls, "recover_waiter %x nodeid %d " 4168 "msg %d to %d", lkb->lkb_id, lkb->lkb_nodeid, 4169 lkb->lkb_wait_type, lkb->lkb_wait_nodeid); 4170 } 4171 4172 /* all outstanding lookups, regardless of destination will be 4173 resent after recovery is done */ 4174 4175 if (lkb->lkb_wait_type == DLM_MSG_LOOKUP) { 4176 lkb->lkb_flags |= DLM_IFL_RESEND; 4177 continue; 4178 } 4179 4180 if (!waiter_needs_recovery(ls, lkb)) 4181 continue; 4182 4183 wait_type = lkb->lkb_wait_type; 4184 stub_unlock_result = -DLM_EUNLOCK; 4185 stub_cancel_result = -DLM_ECANCEL; 4186 4187 /* Main reply may have been received leaving a zero wait_type, 4188 but a reply for the overlapping op may not have been 4189 received. In that case we need to fake the appropriate 4190 reply for the overlap op. */ 4191 4192 if (!wait_type) { 4193 if (is_overlap_cancel(lkb)) { 4194 wait_type = DLM_MSG_CANCEL; 4195 if (lkb->lkb_grmode == DLM_LOCK_IV) 4196 stub_cancel_result = 0; 4197 } 4198 if (is_overlap_unlock(lkb)) { 4199 wait_type = DLM_MSG_UNLOCK; 4200 if (lkb->lkb_grmode == DLM_LOCK_IV) 4201 stub_unlock_result = -ENOENT; 4202 } 4203 4204 log_debug(ls, "rwpre overlap %x %x %d %d %d", 4205 lkb->lkb_id, lkb->lkb_flags, wait_type, 4206 stub_cancel_result, stub_unlock_result); 4207 } 4208 4209 switch (wait_type) { 4210 4211 case DLM_MSG_REQUEST: 4212 lkb->lkb_flags |= DLM_IFL_RESEND; 4213 break; 4214 4215 case DLM_MSG_CONVERT: 4216 recover_convert_waiter(ls, lkb, ms_stub); 4217 break; 4218 4219 case DLM_MSG_UNLOCK: 4220 hold_lkb(lkb); 4221 memset(ms_stub, 0, sizeof(struct dlm_message)); 4222 ms_stub->m_flags = DLM_IFL_STUB_MS; 4223 ms_stub->m_type = DLM_MSG_UNLOCK_REPLY; 4224 ms_stub->m_result = stub_unlock_result; 4225 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid; 4226 _receive_unlock_reply(lkb, ms_stub); 4227 dlm_put_lkb(lkb); 4228 break; 4229 4230 case DLM_MSG_CANCEL: 4231 hold_lkb(lkb); 4232 memset(ms_stub, 0, sizeof(struct dlm_message)); 4233 ms_stub->m_flags = DLM_IFL_STUB_MS; 4234 ms_stub->m_type = DLM_MSG_CANCEL_REPLY; 4235 ms_stub->m_result = stub_cancel_result; 4236 ms_stub->m_header.h_nodeid = lkb->lkb_nodeid; 4237 _receive_cancel_reply(lkb, ms_stub); 4238 dlm_put_lkb(lkb); 4239 break; 4240 4241 default: 4242 log_error(ls, "invalid lkb wait_type %d %d", 4243 lkb->lkb_wait_type, wait_type); 4244 } 4245 schedule(); 4246 } 4247 mutex_unlock(&ls->ls_waiters_mutex); 4248 kfree(ms_stub); 4249 } 4250 4251 static struct dlm_lkb *find_resend_waiter(struct dlm_ls *ls) 4252 { 4253 struct dlm_lkb *lkb; 4254 int found = 0; 4255 4256 mutex_lock(&ls->ls_waiters_mutex); 4257 list_for_each_entry(lkb, &ls->ls_waiters, lkb_wait_reply) { 4258 if (lkb->lkb_flags & DLM_IFL_RESEND) { 4259 hold_lkb(lkb); 4260 found = 1; 4261 break; 4262 } 4263 } 4264 mutex_unlock(&ls->ls_waiters_mutex); 4265 4266 if (!found) 4267 lkb = NULL; 4268 return lkb; 4269 } 4270 4271 /* Deal with lookups and lkb's marked RESEND from _pre. We may now be the 4272 master or dir-node for r. Processing the lkb may result in it being placed 4273 back on waiters. */ 4274 4275 /* We do this after normal locking has been enabled and any saved messages 4276 (in requestqueue) have been processed. We should be confident that at 4277 this point we won't get or process a reply to any of these waiting 4278 operations. But, new ops may be coming in on the rsbs/locks here from 4279 userspace or remotely. */ 4280 4281 /* there may have been an overlap unlock/cancel prior to recovery or after 4282 recovery. if before, the lkb may still have a pos wait_count; if after, the 4283 overlap flag would just have been set and nothing new sent. we can be 4284 confident here than any replies to either the initial op or overlap ops 4285 prior to recovery have been received. */ 4286 4287 int dlm_recover_waiters_post(struct dlm_ls *ls) 4288 { 4289 struct dlm_lkb *lkb; 4290 struct dlm_rsb *r; 4291 int error = 0, mstype, err, oc, ou; 4292 4293 while (1) { 4294 if (dlm_locking_stopped(ls)) { 4295 log_debug(ls, "recover_waiters_post aborted"); 4296 error = -EINTR; 4297 break; 4298 } 4299 4300 lkb = find_resend_waiter(ls); 4301 if (!lkb) 4302 break; 4303 4304 r = lkb->lkb_resource; 4305 hold_rsb(r); 4306 lock_rsb(r); 4307 4308 mstype = lkb->lkb_wait_type; 4309 oc = is_overlap_cancel(lkb); 4310 ou = is_overlap_unlock(lkb); 4311 err = 0; 4312 4313 log_debug(ls, "recover_waiter %x nodeid %d msg %d r_nodeid %d", 4314 lkb->lkb_id, lkb->lkb_nodeid, mstype, r->res_nodeid); 4315 4316 /* At this point we assume that we won't get a reply to any 4317 previous op or overlap op on this lock. First, do a big 4318 remove_from_waiters() for all previous ops. */ 4319 4320 lkb->lkb_flags &= ~DLM_IFL_RESEND; 4321 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_UNLOCK; 4322 lkb->lkb_flags &= ~DLM_IFL_OVERLAP_CANCEL; 4323 lkb->lkb_wait_type = 0; 4324 lkb->lkb_wait_count = 0; 4325 mutex_lock(&ls->ls_waiters_mutex); 4326 list_del_init(&lkb->lkb_wait_reply); 4327 mutex_unlock(&ls->ls_waiters_mutex); 4328 unhold_lkb(lkb); /* for waiters list */ 4329 4330 if (oc || ou) { 4331 /* do an unlock or cancel instead of resending */ 4332 switch (mstype) { 4333 case DLM_MSG_LOOKUP: 4334 case DLM_MSG_REQUEST: 4335 queue_cast(r, lkb, ou ? -DLM_EUNLOCK : 4336 -DLM_ECANCEL); 4337 unhold_lkb(lkb); /* undoes create_lkb() */ 4338 break; 4339 case DLM_MSG_CONVERT: 4340 if (oc) { 4341 queue_cast(r, lkb, -DLM_ECANCEL); 4342 } else { 4343 lkb->lkb_exflags |= DLM_LKF_FORCEUNLOCK; 4344 _unlock_lock(r, lkb); 4345 } 4346 break; 4347 default: 4348 err = 1; 4349 } 4350 } else { 4351 switch (mstype) { 4352 case DLM_MSG_LOOKUP: 4353 case DLM_MSG_REQUEST: 4354 _request_lock(r, lkb); 4355 if (is_master(r)) 4356 confirm_master(r, 0); 4357 break; 4358 case DLM_MSG_CONVERT: 4359 _convert_lock(r, lkb); 4360 break; 4361 default: 4362 err = 1; 4363 } 4364 } 4365 4366 if (err) 4367 log_error(ls, "recover_waiters_post %x %d %x %d %d", 4368 lkb->lkb_id, mstype, lkb->lkb_flags, oc, ou); 4369 unlock_rsb(r); 4370 put_rsb(r); 4371 dlm_put_lkb(lkb); 4372 } 4373 4374 return error; 4375 } 4376 4377 static void purge_queue(struct dlm_rsb *r, struct list_head *queue, 4378 int (*test)(struct dlm_ls *ls, struct dlm_lkb *lkb)) 4379 { 4380 struct dlm_ls *ls = r->res_ls; 4381 struct dlm_lkb *lkb, *safe; 4382 4383 list_for_each_entry_safe(lkb, safe, queue, lkb_statequeue) { 4384 if (test(ls, lkb)) { 4385 rsb_set_flag(r, RSB_LOCKS_PURGED); 4386 del_lkb(r, lkb); 4387 /* this put should free the lkb */ 4388 if (!dlm_put_lkb(lkb)) 4389 log_error(ls, "purged lkb not released"); 4390 } 4391 } 4392 } 4393 4394 static int purge_dead_test(struct dlm_ls *ls, struct dlm_lkb *lkb) 4395 { 4396 return (is_master_copy(lkb) && dlm_is_removed(ls, lkb->lkb_nodeid)); 4397 } 4398 4399 static int purge_mstcpy_test(struct dlm_ls *ls, struct dlm_lkb *lkb) 4400 { 4401 return is_master_copy(lkb); 4402 } 4403 4404 static void purge_dead_locks(struct dlm_rsb *r) 4405 { 4406 purge_queue(r, &r->res_grantqueue, &purge_dead_test); 4407 purge_queue(r, &r->res_convertqueue, &purge_dead_test); 4408 purge_queue(r, &r->res_waitqueue, &purge_dead_test); 4409 } 4410 4411 void dlm_purge_mstcpy_locks(struct dlm_rsb *r) 4412 { 4413 purge_queue(r, &r->res_grantqueue, &purge_mstcpy_test); 4414 purge_queue(r, &r->res_convertqueue, &purge_mstcpy_test); 4415 purge_queue(r, &r->res_waitqueue, &purge_mstcpy_test); 4416 } 4417 4418 /* Get rid of locks held by nodes that are gone. */ 4419 4420 int dlm_purge_locks(struct dlm_ls *ls) 4421 { 4422 struct dlm_rsb *r; 4423 4424 log_debug(ls, "dlm_purge_locks"); 4425 4426 down_write(&ls->ls_root_sem); 4427 list_for_each_entry(r, &ls->ls_root_list, res_root_list) { 4428 hold_rsb(r); 4429 lock_rsb(r); 4430 if (is_master(r)) 4431 purge_dead_locks(r); 4432 unlock_rsb(r); 4433 unhold_rsb(r); 4434 4435 schedule(); 4436 } 4437 up_write(&ls->ls_root_sem); 4438 4439 return 0; 4440 } 4441 4442 static struct dlm_rsb *find_purged_rsb(struct dlm_ls *ls, int bucket) 4443 { 4444 struct dlm_rsb *r, *r_ret = NULL; 4445 4446 spin_lock(&ls->ls_rsbtbl[bucket].lock); 4447 list_for_each_entry(r, &ls->ls_rsbtbl[bucket].list, res_hashchain) { 4448 if (!rsb_flag(r, RSB_LOCKS_PURGED)) 4449 continue; 4450 hold_rsb(r); 4451 rsb_clear_flag(r, RSB_LOCKS_PURGED); 4452 r_ret = r; 4453 break; 4454 } 4455 spin_unlock(&ls->ls_rsbtbl[bucket].lock); 4456 return r_ret; 4457 } 4458 4459 void dlm_grant_after_purge(struct dlm_ls *ls) 4460 { 4461 struct dlm_rsb *r; 4462 int bucket = 0; 4463 4464 while (1) { 4465 r = find_purged_rsb(ls, bucket); 4466 if (!r) { 4467 if (bucket == ls->ls_rsbtbl_size - 1) 4468 break; 4469 bucket++; 4470 continue; 4471 } 4472 lock_rsb(r); 4473 if (is_master(r)) { 4474 grant_pending_locks(r); 4475 confirm_master(r, 0); 4476 } 4477 unlock_rsb(r); 4478 put_rsb(r); 4479 schedule(); 4480 } 4481 } 4482 4483 static struct dlm_lkb *search_remid_list(struct list_head *head, int nodeid, 4484 uint32_t remid) 4485 { 4486 struct dlm_lkb *lkb; 4487 4488 list_for_each_entry(lkb, head, lkb_statequeue) { 4489 if (lkb->lkb_nodeid == nodeid && lkb->lkb_remid == remid) 4490 return lkb; 4491 } 4492 return NULL; 4493 } 4494 4495 static struct dlm_lkb *search_remid(struct dlm_rsb *r, int nodeid, 4496 uint32_t remid) 4497 { 4498 struct dlm_lkb *lkb; 4499 4500 lkb = search_remid_list(&r->res_grantqueue, nodeid, remid); 4501 if (lkb) 4502 return lkb; 4503 lkb = search_remid_list(&r->res_convertqueue, nodeid, remid); 4504 if (lkb) 4505 return lkb; 4506 lkb = search_remid_list(&r->res_waitqueue, nodeid, remid); 4507 if (lkb) 4508 return lkb; 4509 return NULL; 4510 } 4511 4512 /* needs at least dlm_rcom + rcom_lock */ 4513 static int receive_rcom_lock_args(struct dlm_ls *ls, struct dlm_lkb *lkb, 4514 struct dlm_rsb *r, struct dlm_rcom *rc) 4515 { 4516 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf; 4517 4518 lkb->lkb_nodeid = rc->rc_header.h_nodeid; 4519 lkb->lkb_ownpid = le32_to_cpu(rl->rl_ownpid); 4520 lkb->lkb_remid = le32_to_cpu(rl->rl_lkid); 4521 lkb->lkb_exflags = le32_to_cpu(rl->rl_exflags); 4522 lkb->lkb_flags = le32_to_cpu(rl->rl_flags) & 0x0000FFFF; 4523 lkb->lkb_flags |= DLM_IFL_MSTCPY; 4524 lkb->lkb_lvbseq = le32_to_cpu(rl->rl_lvbseq); 4525 lkb->lkb_rqmode = rl->rl_rqmode; 4526 lkb->lkb_grmode = rl->rl_grmode; 4527 /* don't set lkb_status because add_lkb wants to itself */ 4528 4529 lkb->lkb_bastfn = (rl->rl_asts & DLM_CB_BAST) ? &fake_bastfn : NULL; 4530 lkb->lkb_astfn = (rl->rl_asts & DLM_CB_CAST) ? &fake_astfn : NULL; 4531 4532 if (lkb->lkb_exflags & DLM_LKF_VALBLK) { 4533 int lvblen = rc->rc_header.h_length - sizeof(struct dlm_rcom) - 4534 sizeof(struct rcom_lock); 4535 if (lvblen > ls->ls_lvblen) 4536 return -EINVAL; 4537 lkb->lkb_lvbptr = dlm_allocate_lvb(ls); 4538 if (!lkb->lkb_lvbptr) 4539 return -ENOMEM; 4540 memcpy(lkb->lkb_lvbptr, rl->rl_lvb, lvblen); 4541 } 4542 4543 /* Conversions between PR and CW (middle modes) need special handling. 4544 The real granted mode of these converting locks cannot be determined 4545 until all locks have been rebuilt on the rsb (recover_conversion) */ 4546 4547 if (rl->rl_wait_type == cpu_to_le16(DLM_MSG_CONVERT) && 4548 middle_conversion(lkb)) { 4549 rl->rl_status = DLM_LKSTS_CONVERT; 4550 lkb->lkb_grmode = DLM_LOCK_IV; 4551 rsb_set_flag(r, RSB_RECOVER_CONVERT); 4552 } 4553 4554 return 0; 4555 } 4556 4557 /* This lkb may have been recovered in a previous aborted recovery so we need 4558 to check if the rsb already has an lkb with the given remote nodeid/lkid. 4559 If so we just send back a standard reply. If not, we create a new lkb with 4560 the given values and send back our lkid. We send back our lkid by sending 4561 back the rcom_lock struct we got but with the remid field filled in. */ 4562 4563 /* needs at least dlm_rcom + rcom_lock */ 4564 int dlm_recover_master_copy(struct dlm_ls *ls, struct dlm_rcom *rc) 4565 { 4566 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf; 4567 struct dlm_rsb *r; 4568 struct dlm_lkb *lkb; 4569 int error; 4570 4571 if (rl->rl_parent_lkid) { 4572 error = -EOPNOTSUPP; 4573 goto out; 4574 } 4575 4576 error = find_rsb(ls, rl->rl_name, le16_to_cpu(rl->rl_namelen), 4577 R_MASTER, &r); 4578 if (error) 4579 goto out; 4580 4581 lock_rsb(r); 4582 4583 lkb = search_remid(r, rc->rc_header.h_nodeid, le32_to_cpu(rl->rl_lkid)); 4584 if (lkb) { 4585 error = -EEXIST; 4586 goto out_remid; 4587 } 4588 4589 error = create_lkb(ls, &lkb); 4590 if (error) 4591 goto out_unlock; 4592 4593 error = receive_rcom_lock_args(ls, lkb, r, rc); 4594 if (error) { 4595 __put_lkb(ls, lkb); 4596 goto out_unlock; 4597 } 4598 4599 attach_lkb(r, lkb); 4600 add_lkb(r, lkb, rl->rl_status); 4601 error = 0; 4602 4603 out_remid: 4604 /* this is the new value returned to the lock holder for 4605 saving in its process-copy lkb */ 4606 rl->rl_remid = cpu_to_le32(lkb->lkb_id); 4607 4608 out_unlock: 4609 unlock_rsb(r); 4610 put_rsb(r); 4611 out: 4612 if (error) 4613 log_debug(ls, "recover_master_copy %d %x", error, 4614 le32_to_cpu(rl->rl_lkid)); 4615 rl->rl_result = cpu_to_le32(error); 4616 return error; 4617 } 4618 4619 /* needs at least dlm_rcom + rcom_lock */ 4620 int dlm_recover_process_copy(struct dlm_ls *ls, struct dlm_rcom *rc) 4621 { 4622 struct rcom_lock *rl = (struct rcom_lock *) rc->rc_buf; 4623 struct dlm_rsb *r; 4624 struct dlm_lkb *lkb; 4625 int error; 4626 4627 error = find_lkb(ls, le32_to_cpu(rl->rl_lkid), &lkb); 4628 if (error) { 4629 log_error(ls, "recover_process_copy no lkid %x", 4630 le32_to_cpu(rl->rl_lkid)); 4631 return error; 4632 } 4633 4634 DLM_ASSERT(is_process_copy(lkb), dlm_print_lkb(lkb);); 4635 4636 error = le32_to_cpu(rl->rl_result); 4637 4638 r = lkb->lkb_resource; 4639 hold_rsb(r); 4640 lock_rsb(r); 4641 4642 switch (error) { 4643 case -EBADR: 4644 /* There's a chance the new master received our lock before 4645 dlm_recover_master_reply(), this wouldn't happen if we did 4646 a barrier between recover_masters and recover_locks. */ 4647 log_debug(ls, "master copy not ready %x r %lx %s", lkb->lkb_id, 4648 (unsigned long)r, r->res_name); 4649 dlm_send_rcom_lock(r, lkb); 4650 goto out; 4651 case -EEXIST: 4652 log_debug(ls, "master copy exists %x", lkb->lkb_id); 4653 /* fall through */ 4654 case 0: 4655 lkb->lkb_remid = le32_to_cpu(rl->rl_remid); 4656 break; 4657 default: 4658 log_error(ls, "dlm_recover_process_copy unknown error %d %x", 4659 error, lkb->lkb_id); 4660 } 4661 4662 /* an ack for dlm_recover_locks() which waits for replies from 4663 all the locks it sends to new masters */ 4664 dlm_recovered_lock(r); 4665 out: 4666 unlock_rsb(r); 4667 put_rsb(r); 4668 dlm_put_lkb(lkb); 4669 4670 return 0; 4671 } 4672 4673 int dlm_user_request(struct dlm_ls *ls, struct dlm_user_args *ua, 4674 int mode, uint32_t flags, void *name, unsigned int namelen, 4675 unsigned long timeout_cs) 4676 { 4677 struct dlm_lkb *lkb; 4678 struct dlm_args args; 4679 int error; 4680 4681 dlm_lock_recovery(ls); 4682 4683 error = create_lkb(ls, &lkb); 4684 if (error) { 4685 kfree(ua); 4686 goto out; 4687 } 4688 4689 if (flags & DLM_LKF_VALBLK) { 4690 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS); 4691 if (!ua->lksb.sb_lvbptr) { 4692 kfree(ua); 4693 __put_lkb(ls, lkb); 4694 error = -ENOMEM; 4695 goto out; 4696 } 4697 } 4698 4699 /* After ua is attached to lkb it will be freed by dlm_free_lkb(). 4700 When DLM_IFL_USER is set, the dlm knows that this is a userspace 4701 lock and that lkb_astparam is the dlm_user_args structure. */ 4702 4703 error = set_lock_args(mode, &ua->lksb, flags, namelen, timeout_cs, 4704 fake_astfn, ua, fake_bastfn, &args); 4705 lkb->lkb_flags |= DLM_IFL_USER; 4706 4707 if (error) { 4708 __put_lkb(ls, lkb); 4709 goto out; 4710 } 4711 4712 error = request_lock(ls, lkb, name, namelen, &args); 4713 4714 switch (error) { 4715 case 0: 4716 break; 4717 case -EINPROGRESS: 4718 error = 0; 4719 break; 4720 case -EAGAIN: 4721 error = 0; 4722 /* fall through */ 4723 default: 4724 __put_lkb(ls, lkb); 4725 goto out; 4726 } 4727 4728 /* add this new lkb to the per-process list of locks */ 4729 spin_lock(&ua->proc->locks_spin); 4730 hold_lkb(lkb); 4731 list_add_tail(&lkb->lkb_ownqueue, &ua->proc->locks); 4732 spin_unlock(&ua->proc->locks_spin); 4733 out: 4734 dlm_unlock_recovery(ls); 4735 return error; 4736 } 4737 4738 int dlm_user_convert(struct dlm_ls *ls, struct dlm_user_args *ua_tmp, 4739 int mode, uint32_t flags, uint32_t lkid, char *lvb_in, 4740 unsigned long timeout_cs) 4741 { 4742 struct dlm_lkb *lkb; 4743 struct dlm_args args; 4744 struct dlm_user_args *ua; 4745 int error; 4746 4747 dlm_lock_recovery(ls); 4748 4749 error = find_lkb(ls, lkid, &lkb); 4750 if (error) 4751 goto out; 4752 4753 /* user can change the params on its lock when it converts it, or 4754 add an lvb that didn't exist before */ 4755 4756 ua = lkb->lkb_ua; 4757 4758 if (flags & DLM_LKF_VALBLK && !ua->lksb.sb_lvbptr) { 4759 ua->lksb.sb_lvbptr = kzalloc(DLM_USER_LVB_LEN, GFP_NOFS); 4760 if (!ua->lksb.sb_lvbptr) { 4761 error = -ENOMEM; 4762 goto out_put; 4763 } 4764 } 4765 if (lvb_in && ua->lksb.sb_lvbptr) 4766 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN); 4767 4768 ua->xid = ua_tmp->xid; 4769 ua->castparam = ua_tmp->castparam; 4770 ua->castaddr = ua_tmp->castaddr; 4771 ua->bastparam = ua_tmp->bastparam; 4772 ua->bastaddr = ua_tmp->bastaddr; 4773 ua->user_lksb = ua_tmp->user_lksb; 4774 4775 error = set_lock_args(mode, &ua->lksb, flags, 0, timeout_cs, 4776 fake_astfn, ua, fake_bastfn, &args); 4777 if (error) 4778 goto out_put; 4779 4780 error = convert_lock(ls, lkb, &args); 4781 4782 if (error == -EINPROGRESS || error == -EAGAIN || error == -EDEADLK) 4783 error = 0; 4784 out_put: 4785 dlm_put_lkb(lkb); 4786 out: 4787 dlm_unlock_recovery(ls); 4788 kfree(ua_tmp); 4789 return error; 4790 } 4791 4792 int dlm_user_unlock(struct dlm_ls *ls, struct dlm_user_args *ua_tmp, 4793 uint32_t flags, uint32_t lkid, char *lvb_in) 4794 { 4795 struct dlm_lkb *lkb; 4796 struct dlm_args args; 4797 struct dlm_user_args *ua; 4798 int error; 4799 4800 dlm_lock_recovery(ls); 4801 4802 error = find_lkb(ls, lkid, &lkb); 4803 if (error) 4804 goto out; 4805 4806 ua = lkb->lkb_ua; 4807 4808 if (lvb_in && ua->lksb.sb_lvbptr) 4809 memcpy(ua->lksb.sb_lvbptr, lvb_in, DLM_USER_LVB_LEN); 4810 if (ua_tmp->castparam) 4811 ua->castparam = ua_tmp->castparam; 4812 ua->user_lksb = ua_tmp->user_lksb; 4813 4814 error = set_unlock_args(flags, ua, &args); 4815 if (error) 4816 goto out_put; 4817 4818 error = unlock_lock(ls, lkb, &args); 4819 4820 if (error == -DLM_EUNLOCK) 4821 error = 0; 4822 /* from validate_unlock_args() */ 4823 if (error == -EBUSY && (flags & DLM_LKF_FORCEUNLOCK)) 4824 error = 0; 4825 if (error) 4826 goto out_put; 4827 4828 spin_lock(&ua->proc->locks_spin); 4829 /* dlm_user_add_cb() may have already taken lkb off the proc list */ 4830 if (!list_empty(&lkb->lkb_ownqueue)) 4831 list_move(&lkb->lkb_ownqueue, &ua->proc->unlocking); 4832 spin_unlock(&ua->proc->locks_spin); 4833 out_put: 4834 dlm_put_lkb(lkb); 4835 out: 4836 dlm_unlock_recovery(ls); 4837 kfree(ua_tmp); 4838 return error; 4839 } 4840 4841 int dlm_user_cancel(struct dlm_ls *ls, struct dlm_user_args *ua_tmp, 4842 uint32_t flags, uint32_t lkid) 4843 { 4844 struct dlm_lkb *lkb; 4845 struct dlm_args args; 4846 struct dlm_user_args *ua; 4847 int error; 4848 4849 dlm_lock_recovery(ls); 4850 4851 error = find_lkb(ls, lkid, &lkb); 4852 if (error) 4853 goto out; 4854 4855 ua = lkb->lkb_ua; 4856 if (ua_tmp->castparam) 4857 ua->castparam = ua_tmp->castparam; 4858 ua->user_lksb = ua_tmp->user_lksb; 4859 4860 error = set_unlock_args(flags, ua, &args); 4861 if (error) 4862 goto out_put; 4863 4864 error = cancel_lock(ls, lkb, &args); 4865 4866 if (error == -DLM_ECANCEL) 4867 error = 0; 4868 /* from validate_unlock_args() */ 4869 if (error == -EBUSY) 4870 error = 0; 4871 out_put: 4872 dlm_put_lkb(lkb); 4873 out: 4874 dlm_unlock_recovery(ls); 4875 kfree(ua_tmp); 4876 return error; 4877 } 4878 4879 int dlm_user_deadlock(struct dlm_ls *ls, uint32_t flags, uint32_t lkid) 4880 { 4881 struct dlm_lkb *lkb; 4882 struct dlm_args args; 4883 struct dlm_user_args *ua; 4884 struct dlm_rsb *r; 4885 int error; 4886 4887 dlm_lock_recovery(ls); 4888 4889 error = find_lkb(ls, lkid, &lkb); 4890 if (error) 4891 goto out; 4892 4893 ua = lkb->lkb_ua; 4894 4895 error = set_unlock_args(flags, ua, &args); 4896 if (error) 4897 goto out_put; 4898 4899 /* same as cancel_lock(), but set DEADLOCK_CANCEL after lock_rsb */ 4900 4901 r = lkb->lkb_resource; 4902 hold_rsb(r); 4903 lock_rsb(r); 4904 4905 error = validate_unlock_args(lkb, &args); 4906 if (error) 4907 goto out_r; 4908 lkb->lkb_flags |= DLM_IFL_DEADLOCK_CANCEL; 4909 4910 error = _cancel_lock(r, lkb); 4911 out_r: 4912 unlock_rsb(r); 4913 put_rsb(r); 4914 4915 if (error == -DLM_ECANCEL) 4916 error = 0; 4917 /* from validate_unlock_args() */ 4918 if (error == -EBUSY) 4919 error = 0; 4920 out_put: 4921 dlm_put_lkb(lkb); 4922 out: 4923 dlm_unlock_recovery(ls); 4924 return error; 4925 } 4926 4927 /* lkb's that are removed from the waiters list by revert are just left on the 4928 orphans list with the granted orphan locks, to be freed by purge */ 4929 4930 static int orphan_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb) 4931 { 4932 struct dlm_args args; 4933 int error; 4934 4935 hold_lkb(lkb); 4936 mutex_lock(&ls->ls_orphans_mutex); 4937 list_add_tail(&lkb->lkb_ownqueue, &ls->ls_orphans); 4938 mutex_unlock(&ls->ls_orphans_mutex); 4939 4940 set_unlock_args(0, lkb->lkb_ua, &args); 4941 4942 error = cancel_lock(ls, lkb, &args); 4943 if (error == -DLM_ECANCEL) 4944 error = 0; 4945 return error; 4946 } 4947 4948 /* The force flag allows the unlock to go ahead even if the lkb isn't granted. 4949 Regardless of what rsb queue the lock is on, it's removed and freed. */ 4950 4951 static int unlock_proc_lock(struct dlm_ls *ls, struct dlm_lkb *lkb) 4952 { 4953 struct dlm_args args; 4954 int error; 4955 4956 set_unlock_args(DLM_LKF_FORCEUNLOCK, lkb->lkb_ua, &args); 4957 4958 error = unlock_lock(ls, lkb, &args); 4959 if (error == -DLM_EUNLOCK) 4960 error = 0; 4961 return error; 4962 } 4963 4964 /* We have to release clear_proc_locks mutex before calling unlock_proc_lock() 4965 (which does lock_rsb) due to deadlock with receiving a message that does 4966 lock_rsb followed by dlm_user_add_cb() */ 4967 4968 static struct dlm_lkb *del_proc_lock(struct dlm_ls *ls, 4969 struct dlm_user_proc *proc) 4970 { 4971 struct dlm_lkb *lkb = NULL; 4972 4973 mutex_lock(&ls->ls_clear_proc_locks); 4974 if (list_empty(&proc->locks)) 4975 goto out; 4976 4977 lkb = list_entry(proc->locks.next, struct dlm_lkb, lkb_ownqueue); 4978 list_del_init(&lkb->lkb_ownqueue); 4979 4980 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT) 4981 lkb->lkb_flags |= DLM_IFL_ORPHAN; 4982 else 4983 lkb->lkb_flags |= DLM_IFL_DEAD; 4984 out: 4985 mutex_unlock(&ls->ls_clear_proc_locks); 4986 return lkb; 4987 } 4988 4989 /* The ls_clear_proc_locks mutex protects against dlm_user_add_cb() which 4990 1) references lkb->ua which we free here and 2) adds lkbs to proc->asts, 4991 which we clear here. */ 4992 4993 /* proc CLOSING flag is set so no more device_reads should look at proc->asts 4994 list, and no more device_writes should add lkb's to proc->locks list; so we 4995 shouldn't need to take asts_spin or locks_spin here. this assumes that 4996 device reads/writes/closes are serialized -- FIXME: we may need to serialize 4997 them ourself. */ 4998 4999 void dlm_clear_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc) 5000 { 5001 struct dlm_lkb *lkb, *safe; 5002 5003 dlm_lock_recovery(ls); 5004 5005 while (1) { 5006 lkb = del_proc_lock(ls, proc); 5007 if (!lkb) 5008 break; 5009 del_timeout(lkb); 5010 if (lkb->lkb_exflags & DLM_LKF_PERSISTENT) 5011 orphan_proc_lock(ls, lkb); 5012 else 5013 unlock_proc_lock(ls, lkb); 5014 5015 /* this removes the reference for the proc->locks list 5016 added by dlm_user_request, it may result in the lkb 5017 being freed */ 5018 5019 dlm_put_lkb(lkb); 5020 } 5021 5022 mutex_lock(&ls->ls_clear_proc_locks); 5023 5024 /* in-progress unlocks */ 5025 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) { 5026 list_del_init(&lkb->lkb_ownqueue); 5027 lkb->lkb_flags |= DLM_IFL_DEAD; 5028 dlm_put_lkb(lkb); 5029 } 5030 5031 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) { 5032 memset(&lkb->lkb_callbacks, 0, 5033 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE); 5034 list_del_init(&lkb->lkb_cb_list); 5035 dlm_put_lkb(lkb); 5036 } 5037 5038 mutex_unlock(&ls->ls_clear_proc_locks); 5039 dlm_unlock_recovery(ls); 5040 } 5041 5042 static void purge_proc_locks(struct dlm_ls *ls, struct dlm_user_proc *proc) 5043 { 5044 struct dlm_lkb *lkb, *safe; 5045 5046 while (1) { 5047 lkb = NULL; 5048 spin_lock(&proc->locks_spin); 5049 if (!list_empty(&proc->locks)) { 5050 lkb = list_entry(proc->locks.next, struct dlm_lkb, 5051 lkb_ownqueue); 5052 list_del_init(&lkb->lkb_ownqueue); 5053 } 5054 spin_unlock(&proc->locks_spin); 5055 5056 if (!lkb) 5057 break; 5058 5059 lkb->lkb_flags |= DLM_IFL_DEAD; 5060 unlock_proc_lock(ls, lkb); 5061 dlm_put_lkb(lkb); /* ref from proc->locks list */ 5062 } 5063 5064 spin_lock(&proc->locks_spin); 5065 list_for_each_entry_safe(lkb, safe, &proc->unlocking, lkb_ownqueue) { 5066 list_del_init(&lkb->lkb_ownqueue); 5067 lkb->lkb_flags |= DLM_IFL_DEAD; 5068 dlm_put_lkb(lkb); 5069 } 5070 spin_unlock(&proc->locks_spin); 5071 5072 spin_lock(&proc->asts_spin); 5073 list_for_each_entry_safe(lkb, safe, &proc->asts, lkb_cb_list) { 5074 memset(&lkb->lkb_callbacks, 0, 5075 sizeof(struct dlm_callback) * DLM_CALLBACKS_SIZE); 5076 list_del_init(&lkb->lkb_cb_list); 5077 dlm_put_lkb(lkb); 5078 } 5079 spin_unlock(&proc->asts_spin); 5080 } 5081 5082 /* pid of 0 means purge all orphans */ 5083 5084 static void do_purge(struct dlm_ls *ls, int nodeid, int pid) 5085 { 5086 struct dlm_lkb *lkb, *safe; 5087 5088 mutex_lock(&ls->ls_orphans_mutex); 5089 list_for_each_entry_safe(lkb, safe, &ls->ls_orphans, lkb_ownqueue) { 5090 if (pid && lkb->lkb_ownpid != pid) 5091 continue; 5092 unlock_proc_lock(ls, lkb); 5093 list_del_init(&lkb->lkb_ownqueue); 5094 dlm_put_lkb(lkb); 5095 } 5096 mutex_unlock(&ls->ls_orphans_mutex); 5097 } 5098 5099 static int send_purge(struct dlm_ls *ls, int nodeid, int pid) 5100 { 5101 struct dlm_message *ms; 5102 struct dlm_mhandle *mh; 5103 int error; 5104 5105 error = _create_message(ls, sizeof(struct dlm_message), nodeid, 5106 DLM_MSG_PURGE, &ms, &mh); 5107 if (error) 5108 return error; 5109 ms->m_nodeid = nodeid; 5110 ms->m_pid = pid; 5111 5112 return send_message(mh, ms); 5113 } 5114 5115 int dlm_user_purge(struct dlm_ls *ls, struct dlm_user_proc *proc, 5116 int nodeid, int pid) 5117 { 5118 int error = 0; 5119 5120 if (nodeid != dlm_our_nodeid()) { 5121 error = send_purge(ls, nodeid, pid); 5122 } else { 5123 dlm_lock_recovery(ls); 5124 if (pid == current->pid) 5125 purge_proc_locks(ls, proc); 5126 else 5127 do_purge(ls, nodeid, pid); 5128 dlm_unlock_recovery(ls); 5129 } 5130 return error; 5131 } 5132 5133