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