1 /****************************************************************************** 2 ******************************************************************************* 3 ** 4 ** Copyright (C) 2005-2011 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 #include "dlm_internal.h" 14 #include "lockspace.h" 15 #include "member.h" 16 #include "recoverd.h" 17 #include "recover.h" 18 #include "rcom.h" 19 #include "config.h" 20 #include "lowcomms.h" 21 22 int dlm_slots_version(struct dlm_header *h) 23 { 24 if ((h->h_version & 0x0000FFFF) < DLM_HEADER_SLOTS) 25 return 0; 26 return 1; 27 } 28 29 void dlm_slot_save(struct dlm_ls *ls, struct dlm_rcom *rc, 30 struct dlm_member *memb) 31 { 32 struct rcom_config *rf = (struct rcom_config *)rc->rc_buf; 33 34 if (!dlm_slots_version(&rc->rc_header)) 35 return; 36 37 memb->slot = le16_to_cpu(rf->rf_our_slot); 38 memb->generation = le32_to_cpu(rf->rf_generation); 39 } 40 41 void dlm_slots_copy_out(struct dlm_ls *ls, struct dlm_rcom *rc) 42 { 43 struct dlm_slot *slot; 44 struct rcom_slot *ro; 45 int i; 46 47 ro = (struct rcom_slot *)(rc->rc_buf + sizeof(struct rcom_config)); 48 49 /* ls_slots array is sparse, but not rcom_slots */ 50 51 for (i = 0; i < ls->ls_slots_size; i++) { 52 slot = &ls->ls_slots[i]; 53 if (!slot->nodeid) 54 continue; 55 ro->ro_nodeid = cpu_to_le32(slot->nodeid); 56 ro->ro_slot = cpu_to_le16(slot->slot); 57 ro++; 58 } 59 } 60 61 #define SLOT_DEBUG_LINE 128 62 63 static void log_debug_slots(struct dlm_ls *ls, uint32_t gen, int num_slots, 64 struct rcom_slot *ro0, struct dlm_slot *array, 65 int array_size) 66 { 67 char line[SLOT_DEBUG_LINE]; 68 int len = SLOT_DEBUG_LINE - 1; 69 int pos = 0; 70 int ret, i; 71 72 if (!dlm_config.ci_log_debug) 73 return; 74 75 memset(line, 0, sizeof(line)); 76 77 if (array) { 78 for (i = 0; i < array_size; i++) { 79 if (!array[i].nodeid) 80 continue; 81 82 ret = snprintf(line + pos, len - pos, " %d:%d", 83 array[i].slot, array[i].nodeid); 84 if (ret >= len - pos) 85 break; 86 pos += ret; 87 } 88 } else if (ro0) { 89 for (i = 0; i < num_slots; i++) { 90 ret = snprintf(line + pos, len - pos, " %d:%d", 91 ro0[i].ro_slot, ro0[i].ro_nodeid); 92 if (ret >= len - pos) 93 break; 94 pos += ret; 95 } 96 } 97 98 log_debug(ls, "generation %u slots %d%s", gen, num_slots, line); 99 } 100 101 int dlm_slots_copy_in(struct dlm_ls *ls) 102 { 103 struct dlm_member *memb; 104 struct dlm_rcom *rc = ls->ls_recover_buf; 105 struct rcom_config *rf = (struct rcom_config *)rc->rc_buf; 106 struct rcom_slot *ro0, *ro; 107 int our_nodeid = dlm_our_nodeid(); 108 int i, num_slots; 109 uint32_t gen; 110 111 if (!dlm_slots_version(&rc->rc_header)) 112 return -1; 113 114 gen = le32_to_cpu(rf->rf_generation); 115 if (gen <= ls->ls_generation) { 116 log_error(ls, "dlm_slots_copy_in gen %u old %u", 117 gen, ls->ls_generation); 118 } 119 ls->ls_generation = gen; 120 121 num_slots = le16_to_cpu(rf->rf_num_slots); 122 if (!num_slots) 123 return -1; 124 125 ro0 = (struct rcom_slot *)(rc->rc_buf + sizeof(struct rcom_config)); 126 127 for (i = 0, ro = ro0; i < num_slots; i++, ro++) { 128 ro->ro_nodeid = le32_to_cpu(ro->ro_nodeid); 129 ro->ro_slot = le16_to_cpu(ro->ro_slot); 130 } 131 132 log_debug_slots(ls, gen, num_slots, ro0, NULL, 0); 133 134 list_for_each_entry(memb, &ls->ls_nodes, list) { 135 for (i = 0, ro = ro0; i < num_slots; i++, ro++) { 136 if (ro->ro_nodeid != memb->nodeid) 137 continue; 138 memb->slot = ro->ro_slot; 139 memb->slot_prev = memb->slot; 140 break; 141 } 142 143 if (memb->nodeid == our_nodeid) { 144 if (ls->ls_slot && ls->ls_slot != memb->slot) { 145 log_error(ls, "dlm_slots_copy_in our slot " 146 "changed %d %d", ls->ls_slot, 147 memb->slot); 148 return -1; 149 } 150 151 if (!ls->ls_slot) 152 ls->ls_slot = memb->slot; 153 } 154 155 if (!memb->slot) { 156 log_error(ls, "dlm_slots_copy_in nodeid %d no slot", 157 memb->nodeid); 158 return -1; 159 } 160 } 161 162 return 0; 163 } 164 165 /* for any nodes that do not support slots, we will not have set memb->slot 166 in wait_status_all(), so memb->slot will remain -1, and we will not 167 assign slots or set ls_num_slots here */ 168 169 int dlm_slots_assign(struct dlm_ls *ls, int *num_slots, int *slots_size, 170 struct dlm_slot **slots_out, uint32_t *gen_out) 171 { 172 struct dlm_member *memb; 173 struct dlm_slot *array; 174 int our_nodeid = dlm_our_nodeid(); 175 int array_size, max_slots, i; 176 int need = 0; 177 int max = 0; 178 int num = 0; 179 uint32_t gen = 0; 180 181 /* our own memb struct will have slot -1 gen 0 */ 182 183 list_for_each_entry(memb, &ls->ls_nodes, list) { 184 if (memb->nodeid == our_nodeid) { 185 memb->slot = ls->ls_slot; 186 memb->generation = ls->ls_generation; 187 break; 188 } 189 } 190 191 list_for_each_entry(memb, &ls->ls_nodes, list) { 192 if (memb->generation > gen) 193 gen = memb->generation; 194 195 /* node doesn't support slots */ 196 197 if (memb->slot == -1) 198 return -1; 199 200 /* node needs a slot assigned */ 201 202 if (!memb->slot) 203 need++; 204 205 /* node has a slot assigned */ 206 207 num++; 208 209 if (!max || max < memb->slot) 210 max = memb->slot; 211 212 /* sanity check, once slot is assigned it shouldn't change */ 213 214 if (memb->slot_prev && memb->slot && memb->slot_prev != memb->slot) { 215 log_error(ls, "nodeid %d slot changed %d %d", 216 memb->nodeid, memb->slot_prev, memb->slot); 217 return -1; 218 } 219 memb->slot_prev = memb->slot; 220 } 221 222 array_size = max + need; 223 224 array = kzalloc(array_size * sizeof(struct dlm_slot), GFP_NOFS); 225 if (!array) 226 return -ENOMEM; 227 228 num = 0; 229 230 /* fill in slots (offsets) that are used */ 231 232 list_for_each_entry(memb, &ls->ls_nodes, list) { 233 if (!memb->slot) 234 continue; 235 236 if (memb->slot > array_size) { 237 log_error(ls, "invalid slot number %d", memb->slot); 238 kfree(array); 239 return -1; 240 } 241 242 array[memb->slot - 1].nodeid = memb->nodeid; 243 array[memb->slot - 1].slot = memb->slot; 244 num++; 245 } 246 247 /* assign new slots from unused offsets */ 248 249 list_for_each_entry(memb, &ls->ls_nodes, list) { 250 if (memb->slot) 251 continue; 252 253 for (i = 0; i < array_size; i++) { 254 if (array[i].nodeid) 255 continue; 256 257 memb->slot = i + 1; 258 memb->slot_prev = memb->slot; 259 array[i].nodeid = memb->nodeid; 260 array[i].slot = memb->slot; 261 num++; 262 263 if (!ls->ls_slot && memb->nodeid == our_nodeid) 264 ls->ls_slot = memb->slot; 265 break; 266 } 267 268 if (!memb->slot) { 269 log_error(ls, "no free slot found"); 270 kfree(array); 271 return -1; 272 } 273 } 274 275 gen++; 276 277 log_debug_slots(ls, gen, num, NULL, array, array_size); 278 279 max_slots = (dlm_config.ci_buffer_size - sizeof(struct dlm_rcom) - 280 sizeof(struct rcom_config)) / sizeof(struct rcom_slot); 281 282 if (num > max_slots) { 283 log_error(ls, "num_slots %d exceeds max_slots %d", 284 num, max_slots); 285 kfree(array); 286 return -1; 287 } 288 289 *gen_out = gen; 290 *slots_out = array; 291 *slots_size = array_size; 292 *num_slots = num; 293 return 0; 294 } 295 296 static void add_ordered_member(struct dlm_ls *ls, struct dlm_member *new) 297 { 298 struct dlm_member *memb = NULL; 299 struct list_head *tmp; 300 struct list_head *newlist = &new->list; 301 struct list_head *head = &ls->ls_nodes; 302 303 list_for_each(tmp, head) { 304 memb = list_entry(tmp, struct dlm_member, list); 305 if (new->nodeid < memb->nodeid) 306 break; 307 } 308 309 if (!memb) 310 list_add_tail(newlist, head); 311 else { 312 /* FIXME: can use list macro here */ 313 newlist->prev = tmp->prev; 314 newlist->next = tmp; 315 tmp->prev->next = newlist; 316 tmp->prev = newlist; 317 } 318 } 319 320 static int dlm_add_member(struct dlm_ls *ls, struct dlm_config_node *node) 321 { 322 struct dlm_member *memb; 323 int error; 324 325 memb = kzalloc(sizeof(struct dlm_member), GFP_NOFS); 326 if (!memb) 327 return -ENOMEM; 328 329 error = dlm_lowcomms_connect_node(node->nodeid); 330 if (error < 0) { 331 kfree(memb); 332 return error; 333 } 334 335 memb->nodeid = node->nodeid; 336 memb->weight = node->weight; 337 memb->comm_seq = node->comm_seq; 338 add_ordered_member(ls, memb); 339 ls->ls_num_nodes++; 340 return 0; 341 } 342 343 static struct dlm_member *find_memb(struct list_head *head, int nodeid) 344 { 345 struct dlm_member *memb; 346 347 list_for_each_entry(memb, head, list) { 348 if (memb->nodeid == nodeid) 349 return memb; 350 } 351 return NULL; 352 } 353 354 int dlm_is_member(struct dlm_ls *ls, int nodeid) 355 { 356 if (find_memb(&ls->ls_nodes, nodeid)) 357 return 1; 358 return 0; 359 } 360 361 int dlm_is_removed(struct dlm_ls *ls, int nodeid) 362 { 363 if (find_memb(&ls->ls_nodes_gone, nodeid)) 364 return 1; 365 return 0; 366 } 367 368 static void clear_memb_list(struct list_head *head) 369 { 370 struct dlm_member *memb; 371 372 while (!list_empty(head)) { 373 memb = list_entry(head->next, struct dlm_member, list); 374 list_del(&memb->list); 375 kfree(memb); 376 } 377 } 378 379 void dlm_clear_members(struct dlm_ls *ls) 380 { 381 clear_memb_list(&ls->ls_nodes); 382 ls->ls_num_nodes = 0; 383 } 384 385 void dlm_clear_members_gone(struct dlm_ls *ls) 386 { 387 clear_memb_list(&ls->ls_nodes_gone); 388 } 389 390 static void make_member_array(struct dlm_ls *ls) 391 { 392 struct dlm_member *memb; 393 int i, w, x = 0, total = 0, all_zero = 0, *array; 394 395 kfree(ls->ls_node_array); 396 ls->ls_node_array = NULL; 397 398 list_for_each_entry(memb, &ls->ls_nodes, list) { 399 if (memb->weight) 400 total += memb->weight; 401 } 402 403 /* all nodes revert to weight of 1 if all have weight 0 */ 404 405 if (!total) { 406 total = ls->ls_num_nodes; 407 all_zero = 1; 408 } 409 410 ls->ls_total_weight = total; 411 412 array = kmalloc(sizeof(int) * total, GFP_NOFS); 413 if (!array) 414 return; 415 416 list_for_each_entry(memb, &ls->ls_nodes, list) { 417 if (!all_zero && !memb->weight) 418 continue; 419 420 if (all_zero) 421 w = 1; 422 else 423 w = memb->weight; 424 425 DLM_ASSERT(x < total, printk("total %d x %d\n", total, x);); 426 427 for (i = 0; i < w; i++) 428 array[x++] = memb->nodeid; 429 } 430 431 ls->ls_node_array = array; 432 } 433 434 /* send a status request to all members just to establish comms connections */ 435 436 static int ping_members(struct dlm_ls *ls) 437 { 438 struct dlm_member *memb; 439 int error = 0; 440 441 list_for_each_entry(memb, &ls->ls_nodes, list) { 442 error = dlm_recovery_stopped(ls); 443 if (error) 444 break; 445 error = dlm_rcom_status(ls, memb->nodeid, 0); 446 if (error) 447 break; 448 } 449 if (error) 450 log_debug(ls, "ping_members aborted %d last nodeid %d", 451 error, ls->ls_recover_nodeid); 452 return error; 453 } 454 455 static void dlm_lsop_recover_prep(struct dlm_ls *ls) 456 { 457 if (!ls->ls_ops || !ls->ls_ops->recover_prep) 458 return; 459 ls->ls_ops->recover_prep(ls->ls_ops_arg); 460 } 461 462 static void dlm_lsop_recover_slot(struct dlm_ls *ls, struct dlm_member *memb) 463 { 464 struct dlm_slot slot; 465 uint32_t seq; 466 int error; 467 468 if (!ls->ls_ops || !ls->ls_ops->recover_slot) 469 return; 470 471 /* if there is no comms connection with this node 472 or the present comms connection is newer 473 than the one when this member was added, then 474 we consider the node to have failed (versus 475 being removed due to dlm_release_lockspace) */ 476 477 error = dlm_comm_seq(memb->nodeid, &seq); 478 479 if (!error && seq == memb->comm_seq) 480 return; 481 482 slot.nodeid = memb->nodeid; 483 slot.slot = memb->slot; 484 485 ls->ls_ops->recover_slot(ls->ls_ops_arg, &slot); 486 } 487 488 void dlm_lsop_recover_done(struct dlm_ls *ls) 489 { 490 struct dlm_member *memb; 491 struct dlm_slot *slots; 492 int i, num; 493 494 if (!ls->ls_ops || !ls->ls_ops->recover_done) 495 return; 496 497 num = ls->ls_num_nodes; 498 499 slots = kzalloc(num * sizeof(struct dlm_slot), GFP_KERNEL); 500 if (!slots) 501 return; 502 503 i = 0; 504 list_for_each_entry(memb, &ls->ls_nodes, list) { 505 if (i == num) { 506 log_error(ls, "dlm_lsop_recover_done bad num %d", num); 507 goto out; 508 } 509 slots[i].nodeid = memb->nodeid; 510 slots[i].slot = memb->slot; 511 i++; 512 } 513 514 ls->ls_ops->recover_done(ls->ls_ops_arg, slots, num, 515 ls->ls_slot, ls->ls_generation); 516 out: 517 kfree(slots); 518 } 519 520 static struct dlm_config_node *find_config_node(struct dlm_recover *rv, 521 int nodeid) 522 { 523 int i; 524 525 for (i = 0; i < rv->nodes_count; i++) { 526 if (rv->nodes[i].nodeid == nodeid) 527 return &rv->nodes[i]; 528 } 529 return NULL; 530 } 531 532 int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out) 533 { 534 struct dlm_member *memb, *safe; 535 struct dlm_config_node *node; 536 int i, error, neg = 0, low = -1; 537 538 /* previously removed members that we've not finished removing need to 539 count as a negative change so the "neg" recovery steps will happen */ 540 541 list_for_each_entry(memb, &ls->ls_nodes_gone, list) { 542 log_debug(ls, "prev removed member %d", memb->nodeid); 543 neg++; 544 } 545 546 /* move departed members from ls_nodes to ls_nodes_gone */ 547 548 list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) { 549 node = find_config_node(rv, memb->nodeid); 550 if (node && !node->new) 551 continue; 552 553 if (!node) { 554 log_debug(ls, "remove member %d", memb->nodeid); 555 } else { 556 /* removed and re-added */ 557 log_debug(ls, "remove member %d comm_seq %u %u", 558 memb->nodeid, memb->comm_seq, node->comm_seq); 559 } 560 561 neg++; 562 list_move(&memb->list, &ls->ls_nodes_gone); 563 ls->ls_num_nodes--; 564 dlm_lsop_recover_slot(ls, memb); 565 } 566 567 /* add new members to ls_nodes */ 568 569 for (i = 0; i < rv->nodes_count; i++) { 570 node = &rv->nodes[i]; 571 if (dlm_is_member(ls, node->nodeid)) 572 continue; 573 dlm_add_member(ls, node); 574 log_debug(ls, "add member %d", node->nodeid); 575 } 576 577 list_for_each_entry(memb, &ls->ls_nodes, list) { 578 if (low == -1 || memb->nodeid < low) 579 low = memb->nodeid; 580 } 581 ls->ls_low_nodeid = low; 582 583 make_member_array(ls); 584 *neg_out = neg; 585 586 error = ping_members(ls); 587 if (!error || error == -EPROTO) { 588 /* new_lockspace() may be waiting to know if the config 589 is good or bad */ 590 ls->ls_members_result = error; 591 complete(&ls->ls_members_done); 592 } 593 594 log_debug(ls, "dlm_recover_members %d nodes", ls->ls_num_nodes); 595 return error; 596 } 597 598 /* Userspace guarantees that dlm_ls_stop() has completed on all nodes before 599 dlm_ls_start() is called on any of them to start the new recovery. */ 600 601 int dlm_ls_stop(struct dlm_ls *ls) 602 { 603 int new; 604 605 /* 606 * Prevent dlm_recv from being in the middle of something when we do 607 * the stop. This includes ensuring dlm_recv isn't processing a 608 * recovery message (rcom), while dlm_recoverd is aborting and 609 * resetting things from an in-progress recovery. i.e. we want 610 * dlm_recoverd to abort its recovery without worrying about dlm_recv 611 * processing an rcom at the same time. Stopping dlm_recv also makes 612 * it easy for dlm_receive_message() to check locking stopped and add a 613 * message to the requestqueue without races. 614 */ 615 616 down_write(&ls->ls_recv_active); 617 618 /* 619 * Abort any recovery that's in progress (see RECOVER_STOP, 620 * dlm_recovery_stopped()) and tell any other threads running in the 621 * dlm to quit any processing (see RUNNING, dlm_locking_stopped()). 622 */ 623 624 spin_lock(&ls->ls_recover_lock); 625 set_bit(LSFL_RECOVER_STOP, &ls->ls_flags); 626 new = test_and_clear_bit(LSFL_RUNNING, &ls->ls_flags); 627 ls->ls_recover_seq++; 628 spin_unlock(&ls->ls_recover_lock); 629 630 /* 631 * Let dlm_recv run again, now any normal messages will be saved on the 632 * requestqueue for later. 633 */ 634 635 up_write(&ls->ls_recv_active); 636 637 /* 638 * This in_recovery lock does two things: 639 * 1) Keeps this function from returning until all threads are out 640 * of locking routines and locking is truly stopped. 641 * 2) Keeps any new requests from being processed until it's unlocked 642 * when recovery is complete. 643 */ 644 645 if (new) { 646 set_bit(LSFL_RECOVER_DOWN, &ls->ls_flags); 647 wake_up_process(ls->ls_recoverd_task); 648 wait_event(ls->ls_recover_lock_wait, 649 test_bit(LSFL_RECOVER_LOCK, &ls->ls_flags)); 650 } 651 652 /* 653 * The recoverd suspend/resume makes sure that dlm_recoverd (if 654 * running) has noticed RECOVER_STOP above and quit processing the 655 * previous recovery. 656 */ 657 658 dlm_recoverd_suspend(ls); 659 660 spin_lock(&ls->ls_recover_lock); 661 kfree(ls->ls_slots); 662 ls->ls_slots = NULL; 663 ls->ls_num_slots = 0; 664 ls->ls_slots_size = 0; 665 ls->ls_recover_status = 0; 666 spin_unlock(&ls->ls_recover_lock); 667 668 dlm_recoverd_resume(ls); 669 670 if (!ls->ls_recover_begin) 671 ls->ls_recover_begin = jiffies; 672 673 dlm_lsop_recover_prep(ls); 674 return 0; 675 } 676 677 int dlm_ls_start(struct dlm_ls *ls) 678 { 679 struct dlm_recover *rv = NULL, *rv_old; 680 struct dlm_config_node *nodes; 681 int error, count; 682 683 rv = kzalloc(sizeof(struct dlm_recover), GFP_NOFS); 684 if (!rv) 685 return -ENOMEM; 686 687 error = dlm_config_nodes(ls->ls_name, &nodes, &count); 688 if (error < 0) 689 goto fail; 690 691 spin_lock(&ls->ls_recover_lock); 692 693 /* the lockspace needs to be stopped before it can be started */ 694 695 if (!dlm_locking_stopped(ls)) { 696 spin_unlock(&ls->ls_recover_lock); 697 log_error(ls, "start ignored: lockspace running"); 698 error = -EINVAL; 699 goto fail; 700 } 701 702 rv->nodes = nodes; 703 rv->nodes_count = count; 704 rv->seq = ++ls->ls_recover_seq; 705 rv_old = ls->ls_recover_args; 706 ls->ls_recover_args = rv; 707 spin_unlock(&ls->ls_recover_lock); 708 709 if (rv_old) { 710 log_error(ls, "unused recovery %llx %d", 711 (unsigned long long)rv_old->seq, rv_old->nodes_count); 712 kfree(rv_old->nodes); 713 kfree(rv_old); 714 } 715 716 set_bit(LSFL_RECOVER_WORK, &ls->ls_flags); 717 wake_up_process(ls->ls_recoverd_task); 718 return 0; 719 720 fail: 721 kfree(rv); 722 kfree(nodes); 723 return error; 724 } 725 726