1 /* -*- mode: c; c-basic-offset: 8; -*- 2 * vim: noexpandtab sw=8 ts=8 sts=0: 3 * 4 * dlmdomain.c 5 * 6 * defines domain join / leave apis 7 * 8 * Copyright (C) 2004 Oracle. All rights reserved. 9 * 10 * This program is free software; you can redistribute it and/or 11 * modify it under the terms of the GNU General Public 12 * License as published by the Free Software Foundation; either 13 * version 2 of the License, or (at your option) any later version. 14 * 15 * This program is distributed in the hope that it will be useful, 16 * but WITHOUT ANY WARRANTY; without even the implied warranty of 17 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 18 * General Public License for more details. 19 * 20 * You should have received a copy of the GNU General Public 21 * License along with this program; if not, write to the 22 * Free Software Foundation, Inc., 59 Temple Place - Suite 330, 23 * Boston, MA 021110-1307, USA. 24 * 25 */ 26 27 #include <linux/module.h> 28 #include <linux/types.h> 29 #include <linux/slab.h> 30 #include <linux/highmem.h> 31 #include <linux/init.h> 32 #include <linux/spinlock.h> 33 #include <linux/delay.h> 34 #include <linux/err.h> 35 #include <linux/debugfs.h> 36 37 #include "cluster/heartbeat.h" 38 #include "cluster/nodemanager.h" 39 #include "cluster/tcp.h" 40 41 #include "dlmapi.h" 42 #include "dlmcommon.h" 43 #include "dlmdomain.h" 44 #include "dlmdebug.h" 45 46 #include "dlmver.h" 47 48 #define MLOG_MASK_PREFIX (ML_DLM|ML_DLM_DOMAIN) 49 #include "cluster/masklog.h" 50 51 /* 52 * ocfs2 node maps are array of long int, which limits to send them freely 53 * across the wire due to endianness issues. To workaround this, we convert 54 * long ints to byte arrays. Following 3 routines are helper functions to 55 * set/test/copy bits within those array of bytes 56 */ 57 static inline void byte_set_bit(u8 nr, u8 map[]) 58 { 59 map[nr >> 3] |= (1UL << (nr & 7)); 60 } 61 62 static inline int byte_test_bit(u8 nr, u8 map[]) 63 { 64 return ((1UL << (nr & 7)) & (map[nr >> 3])) != 0; 65 } 66 67 static inline void byte_copymap(u8 dmap[], unsigned long smap[], 68 unsigned int sz) 69 { 70 unsigned int nn; 71 72 if (!sz) 73 return; 74 75 memset(dmap, 0, ((sz + 7) >> 3)); 76 for (nn = 0 ; nn < sz; nn++) 77 if (test_bit(nn, smap)) 78 byte_set_bit(nn, dmap); 79 } 80 81 static void dlm_free_pagevec(void **vec, int pages) 82 { 83 while (pages--) 84 free_page((unsigned long)vec[pages]); 85 kfree(vec); 86 } 87 88 static void **dlm_alloc_pagevec(int pages) 89 { 90 void **vec = kmalloc(pages * sizeof(void *), GFP_KERNEL); 91 int i; 92 93 if (!vec) 94 return NULL; 95 96 for (i = 0; i < pages; i++) 97 if (!(vec[i] = (void *)__get_free_page(GFP_KERNEL))) 98 goto out_free; 99 100 mlog(0, "Allocated DLM hash pagevec; %d pages (%lu expected), %lu buckets per page\n", 101 pages, (unsigned long)DLM_HASH_PAGES, 102 (unsigned long)DLM_BUCKETS_PER_PAGE); 103 return vec; 104 out_free: 105 dlm_free_pagevec(vec, i); 106 return NULL; 107 } 108 109 /* 110 * 111 * spinlock lock ordering: if multiple locks are needed, obey this ordering: 112 * dlm_domain_lock 113 * struct dlm_ctxt->spinlock 114 * struct dlm_lock_resource->spinlock 115 * struct dlm_ctxt->master_lock 116 * struct dlm_ctxt->ast_lock 117 * dlm_master_list_entry->spinlock 118 * dlm_lock->spinlock 119 * 120 */ 121 122 DEFINE_SPINLOCK(dlm_domain_lock); 123 LIST_HEAD(dlm_domains); 124 static DECLARE_WAIT_QUEUE_HEAD(dlm_domain_events); 125 126 /* 127 * The supported protocol version for DLM communication. Running domains 128 * will have a negotiated version with the same major number and a minor 129 * number equal or smaller. The dlm_ctxt->dlm_locking_proto field should 130 * be used to determine what a running domain is actually using. 131 * 132 * New in version 1.1: 133 * - Message DLM_QUERY_REGION added to support global heartbeat 134 * - Message DLM_QUERY_NODEINFO added to allow online node removes 135 * New in version 1.2: 136 * - Message DLM_BEGIN_EXIT_DOMAIN_MSG added to mark start of exit domain 137 */ 138 static const struct dlm_protocol_version dlm_protocol = { 139 .pv_major = 1, 140 .pv_minor = 2, 141 }; 142 143 #define DLM_DOMAIN_BACKOFF_MS 200 144 145 static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data, 146 void **ret_data); 147 static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data, 148 void **ret_data); 149 static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data, 150 void **ret_data); 151 static int dlm_query_region_handler(struct o2net_msg *msg, u32 len, 152 void *data, void **ret_data); 153 static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data, 154 void **ret_data); 155 static int dlm_protocol_compare(struct dlm_protocol_version *existing, 156 struct dlm_protocol_version *request); 157 158 static void dlm_unregister_domain_handlers(struct dlm_ctxt *dlm); 159 160 void __dlm_unhash_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res) 161 { 162 if (hlist_unhashed(&res->hash_node)) 163 return; 164 165 mlog(0, "%s: Unhash res %.*s\n", dlm->name, res->lockname.len, 166 res->lockname.name); 167 hlist_del_init(&res->hash_node); 168 dlm_lockres_put(res); 169 } 170 171 void __dlm_insert_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res) 172 { 173 struct hlist_head *bucket; 174 struct qstr *q; 175 176 assert_spin_locked(&dlm->spinlock); 177 178 q = &res->lockname; 179 bucket = dlm_lockres_hash(dlm, q->hash); 180 181 /* get a reference for our hashtable */ 182 dlm_lockres_get(res); 183 184 hlist_add_head(&res->hash_node, bucket); 185 186 mlog(0, "%s: Hash res %.*s\n", dlm->name, res->lockname.len, 187 res->lockname.name); 188 } 189 190 struct dlm_lock_resource * __dlm_lookup_lockres_full(struct dlm_ctxt *dlm, 191 const char *name, 192 unsigned int len, 193 unsigned int hash) 194 { 195 struct hlist_head *bucket; 196 struct hlist_node *list; 197 198 mlog(0, "%.*s\n", len, name); 199 200 assert_spin_locked(&dlm->spinlock); 201 202 bucket = dlm_lockres_hash(dlm, hash); 203 204 hlist_for_each(list, bucket) { 205 struct dlm_lock_resource *res = hlist_entry(list, 206 struct dlm_lock_resource, hash_node); 207 if (res->lockname.name[0] != name[0]) 208 continue; 209 if (unlikely(res->lockname.len != len)) 210 continue; 211 if (memcmp(res->lockname.name + 1, name + 1, len - 1)) 212 continue; 213 dlm_lockres_get(res); 214 return res; 215 } 216 return NULL; 217 } 218 219 /* intended to be called by functions which do not care about lock 220 * resources which are being purged (most net _handler functions). 221 * this will return NULL for any lock resource which is found but 222 * currently in the process of dropping its mastery reference. 223 * use __dlm_lookup_lockres_full when you need the lock resource 224 * regardless (e.g. dlm_get_lock_resource) */ 225 struct dlm_lock_resource * __dlm_lookup_lockres(struct dlm_ctxt *dlm, 226 const char *name, 227 unsigned int len, 228 unsigned int hash) 229 { 230 struct dlm_lock_resource *res = NULL; 231 232 mlog(0, "%.*s\n", len, name); 233 234 assert_spin_locked(&dlm->spinlock); 235 236 res = __dlm_lookup_lockres_full(dlm, name, len, hash); 237 if (res) { 238 spin_lock(&res->spinlock); 239 if (res->state & DLM_LOCK_RES_DROPPING_REF) { 240 spin_unlock(&res->spinlock); 241 dlm_lockres_put(res); 242 return NULL; 243 } 244 spin_unlock(&res->spinlock); 245 } 246 247 return res; 248 } 249 250 struct dlm_lock_resource * dlm_lookup_lockres(struct dlm_ctxt *dlm, 251 const char *name, 252 unsigned int len) 253 { 254 struct dlm_lock_resource *res; 255 unsigned int hash = dlm_lockid_hash(name, len); 256 257 spin_lock(&dlm->spinlock); 258 res = __dlm_lookup_lockres(dlm, name, len, hash); 259 spin_unlock(&dlm->spinlock); 260 return res; 261 } 262 263 static struct dlm_ctxt * __dlm_lookup_domain_full(const char *domain, int len) 264 { 265 struct dlm_ctxt *tmp = NULL; 266 struct list_head *iter; 267 268 assert_spin_locked(&dlm_domain_lock); 269 270 /* tmp->name here is always NULL terminated, 271 * but domain may not be! */ 272 list_for_each(iter, &dlm_domains) { 273 tmp = list_entry (iter, struct dlm_ctxt, list); 274 if (strlen(tmp->name) == len && 275 memcmp(tmp->name, domain, len)==0) 276 break; 277 tmp = NULL; 278 } 279 280 return tmp; 281 } 282 283 /* For null terminated domain strings ONLY */ 284 static struct dlm_ctxt * __dlm_lookup_domain(const char *domain) 285 { 286 assert_spin_locked(&dlm_domain_lock); 287 288 return __dlm_lookup_domain_full(domain, strlen(domain)); 289 } 290 291 292 /* returns true on one of two conditions: 293 * 1) the domain does not exist 294 * 2) the domain exists and it's state is "joined" */ 295 static int dlm_wait_on_domain_helper(const char *domain) 296 { 297 int ret = 0; 298 struct dlm_ctxt *tmp = NULL; 299 300 spin_lock(&dlm_domain_lock); 301 302 tmp = __dlm_lookup_domain(domain); 303 if (!tmp) 304 ret = 1; 305 else if (tmp->dlm_state == DLM_CTXT_JOINED) 306 ret = 1; 307 308 spin_unlock(&dlm_domain_lock); 309 return ret; 310 } 311 312 static void dlm_free_ctxt_mem(struct dlm_ctxt *dlm) 313 { 314 dlm_destroy_debugfs_subroot(dlm); 315 316 if (dlm->lockres_hash) 317 dlm_free_pagevec((void **)dlm->lockres_hash, DLM_HASH_PAGES); 318 319 if (dlm->master_hash) 320 dlm_free_pagevec((void **)dlm->master_hash, DLM_HASH_PAGES); 321 322 if (dlm->name) 323 kfree(dlm->name); 324 325 kfree(dlm); 326 } 327 328 /* A little strange - this function will be called while holding 329 * dlm_domain_lock and is expected to be holding it on the way out. We 330 * will however drop and reacquire it multiple times */ 331 static void dlm_ctxt_release(struct kref *kref) 332 { 333 struct dlm_ctxt *dlm; 334 335 dlm = container_of(kref, struct dlm_ctxt, dlm_refs); 336 337 BUG_ON(dlm->num_joins); 338 BUG_ON(dlm->dlm_state == DLM_CTXT_JOINED); 339 340 /* we may still be in the list if we hit an error during join. */ 341 list_del_init(&dlm->list); 342 343 spin_unlock(&dlm_domain_lock); 344 345 mlog(0, "freeing memory from domain %s\n", dlm->name); 346 347 wake_up(&dlm_domain_events); 348 349 dlm_free_ctxt_mem(dlm); 350 351 spin_lock(&dlm_domain_lock); 352 } 353 354 void dlm_put(struct dlm_ctxt *dlm) 355 { 356 spin_lock(&dlm_domain_lock); 357 kref_put(&dlm->dlm_refs, dlm_ctxt_release); 358 spin_unlock(&dlm_domain_lock); 359 } 360 361 static void __dlm_get(struct dlm_ctxt *dlm) 362 { 363 kref_get(&dlm->dlm_refs); 364 } 365 366 /* given a questionable reference to a dlm object, gets a reference if 367 * it can find it in the list, otherwise returns NULL in which case 368 * you shouldn't trust your pointer. */ 369 struct dlm_ctxt *dlm_grab(struct dlm_ctxt *dlm) 370 { 371 struct list_head *iter; 372 struct dlm_ctxt *target = NULL; 373 374 spin_lock(&dlm_domain_lock); 375 376 list_for_each(iter, &dlm_domains) { 377 target = list_entry (iter, struct dlm_ctxt, list); 378 379 if (target == dlm) { 380 __dlm_get(target); 381 break; 382 } 383 384 target = NULL; 385 } 386 387 spin_unlock(&dlm_domain_lock); 388 389 return target; 390 } 391 392 int dlm_domain_fully_joined(struct dlm_ctxt *dlm) 393 { 394 int ret; 395 396 spin_lock(&dlm_domain_lock); 397 ret = (dlm->dlm_state == DLM_CTXT_JOINED) || 398 (dlm->dlm_state == DLM_CTXT_IN_SHUTDOWN); 399 spin_unlock(&dlm_domain_lock); 400 401 return ret; 402 } 403 404 static void dlm_destroy_dlm_worker(struct dlm_ctxt *dlm) 405 { 406 if (dlm->dlm_worker) { 407 flush_workqueue(dlm->dlm_worker); 408 destroy_workqueue(dlm->dlm_worker); 409 dlm->dlm_worker = NULL; 410 } 411 } 412 413 static void dlm_complete_dlm_shutdown(struct dlm_ctxt *dlm) 414 { 415 dlm_unregister_domain_handlers(dlm); 416 dlm_debug_shutdown(dlm); 417 dlm_complete_thread(dlm); 418 dlm_complete_recovery_thread(dlm); 419 dlm_destroy_dlm_worker(dlm); 420 421 /* We've left the domain. Now we can take ourselves out of the 422 * list and allow the kref stuff to help us free the 423 * memory. */ 424 spin_lock(&dlm_domain_lock); 425 list_del_init(&dlm->list); 426 spin_unlock(&dlm_domain_lock); 427 428 /* Wake up anyone waiting for us to remove this domain */ 429 wake_up(&dlm_domain_events); 430 } 431 432 static int dlm_migrate_all_locks(struct dlm_ctxt *dlm) 433 { 434 int i, num, n, ret = 0; 435 struct dlm_lock_resource *res; 436 struct hlist_node *iter; 437 struct hlist_head *bucket; 438 int dropped; 439 440 mlog(0, "Migrating locks from domain %s\n", dlm->name); 441 442 num = 0; 443 spin_lock(&dlm->spinlock); 444 for (i = 0; i < DLM_HASH_BUCKETS; i++) { 445 redo_bucket: 446 n = 0; 447 bucket = dlm_lockres_hash(dlm, i); 448 iter = bucket->first; 449 while (iter) { 450 n++; 451 res = hlist_entry(iter, struct dlm_lock_resource, 452 hash_node); 453 dlm_lockres_get(res); 454 /* migrate, if necessary. this will drop the dlm 455 * spinlock and retake it if it does migration. */ 456 dropped = dlm_empty_lockres(dlm, res); 457 458 spin_lock(&res->spinlock); 459 if (dropped) 460 __dlm_lockres_calc_usage(dlm, res); 461 else 462 iter = res->hash_node.next; 463 spin_unlock(&res->spinlock); 464 465 dlm_lockres_put(res); 466 467 if (dropped) { 468 cond_resched_lock(&dlm->spinlock); 469 goto redo_bucket; 470 } 471 } 472 cond_resched_lock(&dlm->spinlock); 473 num += n; 474 } 475 spin_unlock(&dlm->spinlock); 476 wake_up(&dlm->dlm_thread_wq); 477 478 /* let the dlm thread take care of purging, keep scanning until 479 * nothing remains in the hash */ 480 if (num) { 481 mlog(0, "%s: %d lock resources in hash last pass\n", 482 dlm->name, num); 483 ret = -EAGAIN; 484 } 485 mlog(0, "DONE Migrating locks from domain %s\n", dlm->name); 486 return ret; 487 } 488 489 static int dlm_no_joining_node(struct dlm_ctxt *dlm) 490 { 491 int ret; 492 493 spin_lock(&dlm->spinlock); 494 ret = dlm->joining_node == DLM_LOCK_RES_OWNER_UNKNOWN; 495 spin_unlock(&dlm->spinlock); 496 497 return ret; 498 } 499 500 static int dlm_begin_exit_domain_handler(struct o2net_msg *msg, u32 len, 501 void *data, void **ret_data) 502 { 503 struct dlm_ctxt *dlm = data; 504 unsigned int node; 505 struct dlm_exit_domain *exit_msg = (struct dlm_exit_domain *) msg->buf; 506 507 if (!dlm_grab(dlm)) 508 return 0; 509 510 node = exit_msg->node_idx; 511 mlog(0, "%s: Node %u sent a begin exit domain message\n", dlm->name, node); 512 513 spin_lock(&dlm->spinlock); 514 set_bit(node, dlm->exit_domain_map); 515 spin_unlock(&dlm->spinlock); 516 517 dlm_put(dlm); 518 519 return 0; 520 } 521 522 static void dlm_mark_domain_leaving(struct dlm_ctxt *dlm) 523 { 524 /* Yikes, a double spinlock! I need domain_lock for the dlm 525 * state and the dlm spinlock for join state... Sorry! */ 526 again: 527 spin_lock(&dlm_domain_lock); 528 spin_lock(&dlm->spinlock); 529 530 if (dlm->joining_node != DLM_LOCK_RES_OWNER_UNKNOWN) { 531 mlog(0, "Node %d is joining, we wait on it.\n", 532 dlm->joining_node); 533 spin_unlock(&dlm->spinlock); 534 spin_unlock(&dlm_domain_lock); 535 536 wait_event(dlm->dlm_join_events, dlm_no_joining_node(dlm)); 537 goto again; 538 } 539 540 dlm->dlm_state = DLM_CTXT_LEAVING; 541 spin_unlock(&dlm->spinlock); 542 spin_unlock(&dlm_domain_lock); 543 } 544 545 static void __dlm_print_nodes(struct dlm_ctxt *dlm) 546 { 547 int node = -1, num = 0; 548 549 assert_spin_locked(&dlm->spinlock); 550 551 printk("( "); 552 while ((node = find_next_bit(dlm->domain_map, O2NM_MAX_NODES, 553 node + 1)) < O2NM_MAX_NODES) { 554 printk("%d ", node); 555 ++num; 556 } 557 printk(") %u nodes\n", num); 558 } 559 560 static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data, 561 void **ret_data) 562 { 563 struct dlm_ctxt *dlm = data; 564 unsigned int node; 565 struct dlm_exit_domain *exit_msg = (struct dlm_exit_domain *) msg->buf; 566 567 mlog(0, "%p %u %p", msg, len, data); 568 569 if (!dlm_grab(dlm)) 570 return 0; 571 572 node = exit_msg->node_idx; 573 574 spin_lock(&dlm->spinlock); 575 clear_bit(node, dlm->domain_map); 576 clear_bit(node, dlm->exit_domain_map); 577 printk(KERN_NOTICE "o2dlm: Node %u leaves domain %s ", node, dlm->name); 578 __dlm_print_nodes(dlm); 579 580 /* notify anything attached to the heartbeat events */ 581 dlm_hb_event_notify_attached(dlm, node, 0); 582 583 spin_unlock(&dlm->spinlock); 584 585 dlm_put(dlm); 586 587 return 0; 588 } 589 590 static int dlm_send_one_domain_exit(struct dlm_ctxt *dlm, u32 msg_type, 591 unsigned int node) 592 { 593 int status; 594 struct dlm_exit_domain leave_msg; 595 596 mlog(0, "%s: Sending domain exit message %u to node %u\n", dlm->name, 597 msg_type, node); 598 599 memset(&leave_msg, 0, sizeof(leave_msg)); 600 leave_msg.node_idx = dlm->node_num; 601 602 status = o2net_send_message(msg_type, dlm->key, &leave_msg, 603 sizeof(leave_msg), node, NULL); 604 if (status < 0) 605 mlog(ML_ERROR, "Error %d sending domain exit message %u " 606 "to node %u on domain %s\n", status, msg_type, node, 607 dlm->name); 608 609 return status; 610 } 611 612 static void dlm_begin_exit_domain(struct dlm_ctxt *dlm) 613 { 614 int node = -1; 615 616 /* Support for begin exit domain was added in 1.2 */ 617 if (dlm->dlm_locking_proto.pv_major == 1 && 618 dlm->dlm_locking_proto.pv_minor < 2) 619 return; 620 621 /* 622 * Unlike DLM_EXIT_DOMAIN_MSG, DLM_BEGIN_EXIT_DOMAIN_MSG is purely 623 * informational. Meaning if a node does not receive the message, 624 * so be it. 625 */ 626 spin_lock(&dlm->spinlock); 627 while (1) { 628 node = find_next_bit(dlm->domain_map, O2NM_MAX_NODES, node + 1); 629 if (node >= O2NM_MAX_NODES) 630 break; 631 if (node == dlm->node_num) 632 continue; 633 634 spin_unlock(&dlm->spinlock); 635 dlm_send_one_domain_exit(dlm, DLM_BEGIN_EXIT_DOMAIN_MSG, node); 636 spin_lock(&dlm->spinlock); 637 } 638 spin_unlock(&dlm->spinlock); 639 } 640 641 static void dlm_leave_domain(struct dlm_ctxt *dlm) 642 { 643 int node, clear_node, status; 644 645 /* At this point we've migrated away all our locks and won't 646 * accept mastership of new ones. The dlm is responsible for 647 * almost nothing now. We make sure not to confuse any joining 648 * nodes and then commence shutdown procedure. */ 649 650 spin_lock(&dlm->spinlock); 651 /* Clear ourselves from the domain map */ 652 clear_bit(dlm->node_num, dlm->domain_map); 653 while ((node = find_next_bit(dlm->domain_map, O2NM_MAX_NODES, 654 0)) < O2NM_MAX_NODES) { 655 /* Drop the dlm spinlock. This is safe wrt the domain_map. 656 * -nodes cannot be added now as the 657 * query_join_handlers knows to respond with OK_NO_MAP 658 * -we catch the right network errors if a node is 659 * removed from the map while we're sending him the 660 * exit message. */ 661 spin_unlock(&dlm->spinlock); 662 663 clear_node = 1; 664 665 status = dlm_send_one_domain_exit(dlm, DLM_EXIT_DOMAIN_MSG, 666 node); 667 if (status < 0 && 668 status != -ENOPROTOOPT && 669 status != -ENOTCONN) { 670 mlog(ML_NOTICE, "Error %d sending domain exit message " 671 "to node %d\n", status, node); 672 673 /* Not sure what to do here but lets sleep for 674 * a bit in case this was a transient 675 * error... */ 676 msleep(DLM_DOMAIN_BACKOFF_MS); 677 clear_node = 0; 678 } 679 680 spin_lock(&dlm->spinlock); 681 /* If we're not clearing the node bit then we intend 682 * to loop back around to try again. */ 683 if (clear_node) 684 clear_bit(node, dlm->domain_map); 685 } 686 spin_unlock(&dlm->spinlock); 687 } 688 689 int dlm_joined(struct dlm_ctxt *dlm) 690 { 691 int ret = 0; 692 693 spin_lock(&dlm_domain_lock); 694 695 if (dlm->dlm_state == DLM_CTXT_JOINED) 696 ret = 1; 697 698 spin_unlock(&dlm_domain_lock); 699 700 return ret; 701 } 702 703 int dlm_shutting_down(struct dlm_ctxt *dlm) 704 { 705 int ret = 0; 706 707 spin_lock(&dlm_domain_lock); 708 709 if (dlm->dlm_state == DLM_CTXT_IN_SHUTDOWN) 710 ret = 1; 711 712 spin_unlock(&dlm_domain_lock); 713 714 return ret; 715 } 716 717 void dlm_unregister_domain(struct dlm_ctxt *dlm) 718 { 719 int leave = 0; 720 struct dlm_lock_resource *res; 721 722 spin_lock(&dlm_domain_lock); 723 BUG_ON(dlm->dlm_state != DLM_CTXT_JOINED); 724 BUG_ON(!dlm->num_joins); 725 726 dlm->num_joins--; 727 if (!dlm->num_joins) { 728 /* We mark it "in shutdown" now so new register 729 * requests wait until we've completely left the 730 * domain. Don't use DLM_CTXT_LEAVING yet as we still 731 * want new domain joins to communicate with us at 732 * least until we've completed migration of our 733 * resources. */ 734 dlm->dlm_state = DLM_CTXT_IN_SHUTDOWN; 735 leave = 1; 736 } 737 spin_unlock(&dlm_domain_lock); 738 739 if (leave) { 740 mlog(0, "shutting down domain %s\n", dlm->name); 741 dlm_begin_exit_domain(dlm); 742 743 /* We changed dlm state, notify the thread */ 744 dlm_kick_thread(dlm, NULL); 745 746 while (dlm_migrate_all_locks(dlm)) { 747 /* Give dlm_thread time to purge the lockres' */ 748 msleep(500); 749 mlog(0, "%s: more migration to do\n", dlm->name); 750 } 751 752 /* This list should be empty. If not, print remaining lockres */ 753 if (!list_empty(&dlm->tracking_list)) { 754 mlog(ML_ERROR, "Following lockres' are still on the " 755 "tracking list:\n"); 756 list_for_each_entry(res, &dlm->tracking_list, tracking) 757 dlm_print_one_lock_resource(res); 758 } 759 760 dlm_mark_domain_leaving(dlm); 761 dlm_leave_domain(dlm); 762 printk(KERN_NOTICE "o2dlm: Leaving domain %s\n", dlm->name); 763 dlm_force_free_mles(dlm); 764 dlm_complete_dlm_shutdown(dlm); 765 } 766 dlm_put(dlm); 767 } 768 EXPORT_SYMBOL_GPL(dlm_unregister_domain); 769 770 static int dlm_query_join_proto_check(char *proto_type, int node, 771 struct dlm_protocol_version *ours, 772 struct dlm_protocol_version *request) 773 { 774 int rc; 775 struct dlm_protocol_version proto = *request; 776 777 if (!dlm_protocol_compare(ours, &proto)) { 778 mlog(0, 779 "node %u wanted to join with %s locking protocol " 780 "%u.%u, we respond with %u.%u\n", 781 node, proto_type, 782 request->pv_major, 783 request->pv_minor, 784 proto.pv_major, proto.pv_minor); 785 request->pv_minor = proto.pv_minor; 786 rc = 0; 787 } else { 788 mlog(ML_NOTICE, 789 "Node %u wanted to join with %s locking " 790 "protocol %u.%u, but we have %u.%u, disallowing\n", 791 node, proto_type, 792 request->pv_major, 793 request->pv_minor, 794 ours->pv_major, 795 ours->pv_minor); 796 rc = 1; 797 } 798 799 return rc; 800 } 801 802 /* 803 * struct dlm_query_join_packet is made up of four one-byte fields. They 804 * are effectively in big-endian order already. However, little-endian 805 * machines swap them before putting the packet on the wire (because 806 * query_join's response is a status, and that status is treated as a u32 807 * on the wire). Thus, a big-endian and little-endian machines will treat 808 * this structure differently. 809 * 810 * The solution is to have little-endian machines swap the structure when 811 * converting from the structure to the u32 representation. This will 812 * result in the structure having the correct format on the wire no matter 813 * the host endian format. 814 */ 815 static void dlm_query_join_packet_to_wire(struct dlm_query_join_packet *packet, 816 u32 *wire) 817 { 818 union dlm_query_join_response response; 819 820 response.packet = *packet; 821 *wire = be32_to_cpu(response.intval); 822 } 823 824 static void dlm_query_join_wire_to_packet(u32 wire, 825 struct dlm_query_join_packet *packet) 826 { 827 union dlm_query_join_response response; 828 829 response.intval = cpu_to_be32(wire); 830 *packet = response.packet; 831 } 832 833 static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data, 834 void **ret_data) 835 { 836 struct dlm_query_join_request *query; 837 struct dlm_query_join_packet packet = { 838 .code = JOIN_DISALLOW, 839 }; 840 struct dlm_ctxt *dlm = NULL; 841 u32 response; 842 u8 nodenum; 843 844 query = (struct dlm_query_join_request *) msg->buf; 845 846 mlog(0, "node %u wants to join domain %s\n", query->node_idx, 847 query->domain); 848 849 /* 850 * If heartbeat doesn't consider the node live, tell it 851 * to back off and try again. This gives heartbeat a chance 852 * to catch up. 853 */ 854 if (!o2hb_check_node_heartbeating(query->node_idx)) { 855 mlog(0, "node %u is not in our live map yet\n", 856 query->node_idx); 857 858 packet.code = JOIN_DISALLOW; 859 goto respond; 860 } 861 862 packet.code = JOIN_OK_NO_MAP; 863 864 spin_lock(&dlm_domain_lock); 865 dlm = __dlm_lookup_domain_full(query->domain, query->name_len); 866 if (!dlm) 867 goto unlock_respond; 868 869 /* 870 * There is a small window where the joining node may not see the 871 * node(s) that just left but still part of the cluster. DISALLOW 872 * join request if joining node has different node map. 873 */ 874 nodenum=0; 875 while (nodenum < O2NM_MAX_NODES) { 876 if (test_bit(nodenum, dlm->domain_map)) { 877 if (!byte_test_bit(nodenum, query->node_map)) { 878 mlog(0, "disallow join as node %u does not " 879 "have node %u in its nodemap\n", 880 query->node_idx, nodenum); 881 packet.code = JOIN_DISALLOW; 882 goto unlock_respond; 883 } 884 } 885 nodenum++; 886 } 887 888 /* Once the dlm ctxt is marked as leaving then we don't want 889 * to be put in someone's domain map. 890 * Also, explicitly disallow joining at certain troublesome 891 * times (ie. during recovery). */ 892 if (dlm && dlm->dlm_state != DLM_CTXT_LEAVING) { 893 int bit = query->node_idx; 894 spin_lock(&dlm->spinlock); 895 896 if (dlm->dlm_state == DLM_CTXT_NEW && 897 dlm->joining_node == DLM_LOCK_RES_OWNER_UNKNOWN) { 898 /*If this is a brand new context and we 899 * haven't started our join process yet, then 900 * the other node won the race. */ 901 packet.code = JOIN_OK_NO_MAP; 902 } else if (dlm->joining_node != DLM_LOCK_RES_OWNER_UNKNOWN) { 903 /* Disallow parallel joins. */ 904 packet.code = JOIN_DISALLOW; 905 } else if (dlm->reco.state & DLM_RECO_STATE_ACTIVE) { 906 mlog(0, "node %u trying to join, but recovery " 907 "is ongoing.\n", bit); 908 packet.code = JOIN_DISALLOW; 909 } else if (test_bit(bit, dlm->recovery_map)) { 910 mlog(0, "node %u trying to join, but it " 911 "still needs recovery.\n", bit); 912 packet.code = JOIN_DISALLOW; 913 } else if (test_bit(bit, dlm->domain_map)) { 914 mlog(0, "node %u trying to join, but it " 915 "is still in the domain! needs recovery?\n", 916 bit); 917 packet.code = JOIN_DISALLOW; 918 } else { 919 /* Alright we're fully a part of this domain 920 * so we keep some state as to who's joining 921 * and indicate to him that needs to be fixed 922 * up. */ 923 924 /* Make sure we speak compatible locking protocols. */ 925 if (dlm_query_join_proto_check("DLM", bit, 926 &dlm->dlm_locking_proto, 927 &query->dlm_proto)) { 928 packet.code = JOIN_PROTOCOL_MISMATCH; 929 } else if (dlm_query_join_proto_check("fs", bit, 930 &dlm->fs_locking_proto, 931 &query->fs_proto)) { 932 packet.code = JOIN_PROTOCOL_MISMATCH; 933 } else { 934 packet.dlm_minor = query->dlm_proto.pv_minor; 935 packet.fs_minor = query->fs_proto.pv_minor; 936 packet.code = JOIN_OK; 937 __dlm_set_joining_node(dlm, query->node_idx); 938 } 939 } 940 941 spin_unlock(&dlm->spinlock); 942 } 943 unlock_respond: 944 spin_unlock(&dlm_domain_lock); 945 946 respond: 947 mlog(0, "We respond with %u\n", packet.code); 948 949 dlm_query_join_packet_to_wire(&packet, &response); 950 return response; 951 } 952 953 static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data, 954 void **ret_data) 955 { 956 struct dlm_assert_joined *assert; 957 struct dlm_ctxt *dlm = NULL; 958 959 assert = (struct dlm_assert_joined *) msg->buf; 960 961 mlog(0, "node %u asserts join on domain %s\n", assert->node_idx, 962 assert->domain); 963 964 spin_lock(&dlm_domain_lock); 965 dlm = __dlm_lookup_domain_full(assert->domain, assert->name_len); 966 /* XXX should we consider no dlm ctxt an error? */ 967 if (dlm) { 968 spin_lock(&dlm->spinlock); 969 970 /* Alright, this node has officially joined our 971 * domain. Set him in the map and clean up our 972 * leftover join state. */ 973 BUG_ON(dlm->joining_node != assert->node_idx); 974 set_bit(assert->node_idx, dlm->domain_map); 975 clear_bit(assert->node_idx, dlm->exit_domain_map); 976 __dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN); 977 978 printk(KERN_NOTICE "o2dlm: Node %u joins domain %s ", 979 assert->node_idx, dlm->name); 980 __dlm_print_nodes(dlm); 981 982 /* notify anything attached to the heartbeat events */ 983 dlm_hb_event_notify_attached(dlm, assert->node_idx, 1); 984 985 spin_unlock(&dlm->spinlock); 986 } 987 spin_unlock(&dlm_domain_lock); 988 989 return 0; 990 } 991 992 static int dlm_match_regions(struct dlm_ctxt *dlm, 993 struct dlm_query_region *qr, 994 char *local, int locallen) 995 { 996 char *remote = qr->qr_regions; 997 char *l, *r; 998 int localnr, i, j, foundit; 999 int status = 0; 1000 1001 if (!o2hb_global_heartbeat_active()) { 1002 if (qr->qr_numregions) { 1003 mlog(ML_ERROR, "Domain %s: Joining node %d has global " 1004 "heartbeat enabled but local node %d does not\n", 1005 qr->qr_domain, qr->qr_node, dlm->node_num); 1006 status = -EINVAL; 1007 } 1008 goto bail; 1009 } 1010 1011 if (o2hb_global_heartbeat_active() && !qr->qr_numregions) { 1012 mlog(ML_ERROR, "Domain %s: Local node %d has global " 1013 "heartbeat enabled but joining node %d does not\n", 1014 qr->qr_domain, dlm->node_num, qr->qr_node); 1015 status = -EINVAL; 1016 goto bail; 1017 } 1018 1019 r = remote; 1020 for (i = 0; i < qr->qr_numregions; ++i) { 1021 mlog(0, "Region %.*s\n", O2HB_MAX_REGION_NAME_LEN, r); 1022 r += O2HB_MAX_REGION_NAME_LEN; 1023 } 1024 1025 localnr = min(O2NM_MAX_REGIONS, locallen/O2HB_MAX_REGION_NAME_LEN); 1026 localnr = o2hb_get_all_regions(local, (u8)localnr); 1027 1028 /* compare local regions with remote */ 1029 l = local; 1030 for (i = 0; i < localnr; ++i) { 1031 foundit = 0; 1032 r = remote; 1033 for (j = 0; j <= qr->qr_numregions; ++j) { 1034 if (!memcmp(l, r, O2HB_MAX_REGION_NAME_LEN)) { 1035 foundit = 1; 1036 break; 1037 } 1038 r += O2HB_MAX_REGION_NAME_LEN; 1039 } 1040 if (!foundit) { 1041 status = -EINVAL; 1042 mlog(ML_ERROR, "Domain %s: Region '%.*s' registered " 1043 "in local node %d but not in joining node %d\n", 1044 qr->qr_domain, O2HB_MAX_REGION_NAME_LEN, l, 1045 dlm->node_num, qr->qr_node); 1046 goto bail; 1047 } 1048 l += O2HB_MAX_REGION_NAME_LEN; 1049 } 1050 1051 /* compare remote with local regions */ 1052 r = remote; 1053 for (i = 0; i < qr->qr_numregions; ++i) { 1054 foundit = 0; 1055 l = local; 1056 for (j = 0; j < localnr; ++j) { 1057 if (!memcmp(r, l, O2HB_MAX_REGION_NAME_LEN)) { 1058 foundit = 1; 1059 break; 1060 } 1061 l += O2HB_MAX_REGION_NAME_LEN; 1062 } 1063 if (!foundit) { 1064 status = -EINVAL; 1065 mlog(ML_ERROR, "Domain %s: Region '%.*s' registered " 1066 "in joining node %d but not in local node %d\n", 1067 qr->qr_domain, O2HB_MAX_REGION_NAME_LEN, r, 1068 qr->qr_node, dlm->node_num); 1069 goto bail; 1070 } 1071 r += O2HB_MAX_REGION_NAME_LEN; 1072 } 1073 1074 bail: 1075 return status; 1076 } 1077 1078 static int dlm_send_regions(struct dlm_ctxt *dlm, unsigned long *node_map) 1079 { 1080 struct dlm_query_region *qr = NULL; 1081 int status, ret = 0, i; 1082 char *p; 1083 1084 if (find_next_bit(node_map, O2NM_MAX_NODES, 0) >= O2NM_MAX_NODES) 1085 goto bail; 1086 1087 qr = kzalloc(sizeof(struct dlm_query_region), GFP_KERNEL); 1088 if (!qr) { 1089 ret = -ENOMEM; 1090 mlog_errno(ret); 1091 goto bail; 1092 } 1093 1094 qr->qr_node = dlm->node_num; 1095 qr->qr_namelen = strlen(dlm->name); 1096 memcpy(qr->qr_domain, dlm->name, qr->qr_namelen); 1097 /* if local hb, the numregions will be zero */ 1098 if (o2hb_global_heartbeat_active()) 1099 qr->qr_numregions = o2hb_get_all_regions(qr->qr_regions, 1100 O2NM_MAX_REGIONS); 1101 1102 p = qr->qr_regions; 1103 for (i = 0; i < qr->qr_numregions; ++i, p += O2HB_MAX_REGION_NAME_LEN) 1104 mlog(0, "Region %.*s\n", O2HB_MAX_REGION_NAME_LEN, p); 1105 1106 i = -1; 1107 while ((i = find_next_bit(node_map, O2NM_MAX_NODES, 1108 i + 1)) < O2NM_MAX_NODES) { 1109 if (i == dlm->node_num) 1110 continue; 1111 1112 mlog(0, "Sending regions to node %d\n", i); 1113 1114 ret = o2net_send_message(DLM_QUERY_REGION, DLM_MOD_KEY, qr, 1115 sizeof(struct dlm_query_region), 1116 i, &status); 1117 if (ret >= 0) 1118 ret = status; 1119 if (ret) { 1120 mlog(ML_ERROR, "Region mismatch %d, node %d\n", 1121 ret, i); 1122 break; 1123 } 1124 } 1125 1126 bail: 1127 kfree(qr); 1128 return ret; 1129 } 1130 1131 static int dlm_query_region_handler(struct o2net_msg *msg, u32 len, 1132 void *data, void **ret_data) 1133 { 1134 struct dlm_query_region *qr; 1135 struct dlm_ctxt *dlm = NULL; 1136 char *local = NULL; 1137 int status = 0; 1138 int locked = 0; 1139 1140 qr = (struct dlm_query_region *) msg->buf; 1141 1142 mlog(0, "Node %u queries hb regions on domain %s\n", qr->qr_node, 1143 qr->qr_domain); 1144 1145 /* buffer used in dlm_mast_regions() */ 1146 local = kmalloc(sizeof(qr->qr_regions), GFP_KERNEL); 1147 if (!local) { 1148 status = -ENOMEM; 1149 goto bail; 1150 } 1151 1152 status = -EINVAL; 1153 1154 spin_lock(&dlm_domain_lock); 1155 dlm = __dlm_lookup_domain_full(qr->qr_domain, qr->qr_namelen); 1156 if (!dlm) { 1157 mlog(ML_ERROR, "Node %d queried hb regions on domain %s " 1158 "before join domain\n", qr->qr_node, qr->qr_domain); 1159 goto bail; 1160 } 1161 1162 spin_lock(&dlm->spinlock); 1163 locked = 1; 1164 if (dlm->joining_node != qr->qr_node) { 1165 mlog(ML_ERROR, "Node %d queried hb regions on domain %s " 1166 "but joining node is %d\n", qr->qr_node, qr->qr_domain, 1167 dlm->joining_node); 1168 goto bail; 1169 } 1170 1171 /* Support for global heartbeat was added in 1.1 */ 1172 if (dlm->dlm_locking_proto.pv_major == 1 && 1173 dlm->dlm_locking_proto.pv_minor == 0) { 1174 mlog(ML_ERROR, "Node %d queried hb regions on domain %s " 1175 "but active dlm protocol is %d.%d\n", qr->qr_node, 1176 qr->qr_domain, dlm->dlm_locking_proto.pv_major, 1177 dlm->dlm_locking_proto.pv_minor); 1178 goto bail; 1179 } 1180 1181 status = dlm_match_regions(dlm, qr, local, sizeof(qr->qr_regions)); 1182 1183 bail: 1184 if (locked) 1185 spin_unlock(&dlm->spinlock); 1186 spin_unlock(&dlm_domain_lock); 1187 1188 kfree(local); 1189 1190 return status; 1191 } 1192 1193 static int dlm_match_nodes(struct dlm_ctxt *dlm, struct dlm_query_nodeinfo *qn) 1194 { 1195 struct o2nm_node *local; 1196 struct dlm_node_info *remote; 1197 int i, j; 1198 int status = 0; 1199 1200 for (j = 0; j < qn->qn_numnodes; ++j) 1201 mlog(0, "Node %3d, %pI4:%u\n", qn->qn_nodes[j].ni_nodenum, 1202 &(qn->qn_nodes[j].ni_ipv4_address), 1203 ntohs(qn->qn_nodes[j].ni_ipv4_port)); 1204 1205 for (i = 0; i < O2NM_MAX_NODES && !status; ++i) { 1206 local = o2nm_get_node_by_num(i); 1207 remote = NULL; 1208 for (j = 0; j < qn->qn_numnodes; ++j) { 1209 if (qn->qn_nodes[j].ni_nodenum == i) { 1210 remote = &(qn->qn_nodes[j]); 1211 break; 1212 } 1213 } 1214 1215 if (!local && !remote) 1216 continue; 1217 1218 if ((local && !remote) || (!local && remote)) 1219 status = -EINVAL; 1220 1221 if (!status && 1222 ((remote->ni_nodenum != local->nd_num) || 1223 (remote->ni_ipv4_port != local->nd_ipv4_port) || 1224 (remote->ni_ipv4_address != local->nd_ipv4_address))) 1225 status = -EINVAL; 1226 1227 if (status) { 1228 if (remote && !local) 1229 mlog(ML_ERROR, "Domain %s: Node %d (%pI4:%u) " 1230 "registered in joining node %d but not in " 1231 "local node %d\n", qn->qn_domain, 1232 remote->ni_nodenum, 1233 &(remote->ni_ipv4_address), 1234 ntohs(remote->ni_ipv4_port), 1235 qn->qn_nodenum, dlm->node_num); 1236 if (local && !remote) 1237 mlog(ML_ERROR, "Domain %s: Node %d (%pI4:%u) " 1238 "registered in local node %d but not in " 1239 "joining node %d\n", qn->qn_domain, 1240 local->nd_num, &(local->nd_ipv4_address), 1241 ntohs(local->nd_ipv4_port), 1242 dlm->node_num, qn->qn_nodenum); 1243 BUG_ON((!local && !remote)); 1244 } 1245 1246 if (local) 1247 o2nm_node_put(local); 1248 } 1249 1250 return status; 1251 } 1252 1253 static int dlm_send_nodeinfo(struct dlm_ctxt *dlm, unsigned long *node_map) 1254 { 1255 struct dlm_query_nodeinfo *qn = NULL; 1256 struct o2nm_node *node; 1257 int ret = 0, status, count, i; 1258 1259 if (find_next_bit(node_map, O2NM_MAX_NODES, 0) >= O2NM_MAX_NODES) 1260 goto bail; 1261 1262 qn = kzalloc(sizeof(struct dlm_query_nodeinfo), GFP_KERNEL); 1263 if (!qn) { 1264 ret = -ENOMEM; 1265 mlog_errno(ret); 1266 goto bail; 1267 } 1268 1269 for (i = 0, count = 0; i < O2NM_MAX_NODES; ++i) { 1270 node = o2nm_get_node_by_num(i); 1271 if (!node) 1272 continue; 1273 qn->qn_nodes[count].ni_nodenum = node->nd_num; 1274 qn->qn_nodes[count].ni_ipv4_port = node->nd_ipv4_port; 1275 qn->qn_nodes[count].ni_ipv4_address = node->nd_ipv4_address; 1276 mlog(0, "Node %3d, %pI4:%u\n", node->nd_num, 1277 &(node->nd_ipv4_address), ntohs(node->nd_ipv4_port)); 1278 ++count; 1279 o2nm_node_put(node); 1280 } 1281 1282 qn->qn_nodenum = dlm->node_num; 1283 qn->qn_numnodes = count; 1284 qn->qn_namelen = strlen(dlm->name); 1285 memcpy(qn->qn_domain, dlm->name, qn->qn_namelen); 1286 1287 i = -1; 1288 while ((i = find_next_bit(node_map, O2NM_MAX_NODES, 1289 i + 1)) < O2NM_MAX_NODES) { 1290 if (i == dlm->node_num) 1291 continue; 1292 1293 mlog(0, "Sending nodeinfo to node %d\n", i); 1294 1295 ret = o2net_send_message(DLM_QUERY_NODEINFO, DLM_MOD_KEY, 1296 qn, sizeof(struct dlm_query_nodeinfo), 1297 i, &status); 1298 if (ret >= 0) 1299 ret = status; 1300 if (ret) { 1301 mlog(ML_ERROR, "node mismatch %d, node %d\n", ret, i); 1302 break; 1303 } 1304 } 1305 1306 bail: 1307 kfree(qn); 1308 return ret; 1309 } 1310 1311 static int dlm_query_nodeinfo_handler(struct o2net_msg *msg, u32 len, 1312 void *data, void **ret_data) 1313 { 1314 struct dlm_query_nodeinfo *qn; 1315 struct dlm_ctxt *dlm = NULL; 1316 int locked = 0, status = -EINVAL; 1317 1318 qn = (struct dlm_query_nodeinfo *) msg->buf; 1319 1320 mlog(0, "Node %u queries nodes on domain %s\n", qn->qn_nodenum, 1321 qn->qn_domain); 1322 1323 spin_lock(&dlm_domain_lock); 1324 dlm = __dlm_lookup_domain_full(qn->qn_domain, qn->qn_namelen); 1325 if (!dlm) { 1326 mlog(ML_ERROR, "Node %d queried nodes on domain %s before " 1327 "join domain\n", qn->qn_nodenum, qn->qn_domain); 1328 goto bail; 1329 } 1330 1331 spin_lock(&dlm->spinlock); 1332 locked = 1; 1333 if (dlm->joining_node != qn->qn_nodenum) { 1334 mlog(ML_ERROR, "Node %d queried nodes on domain %s but " 1335 "joining node is %d\n", qn->qn_nodenum, qn->qn_domain, 1336 dlm->joining_node); 1337 goto bail; 1338 } 1339 1340 /* Support for node query was added in 1.1 */ 1341 if (dlm->dlm_locking_proto.pv_major == 1 && 1342 dlm->dlm_locking_proto.pv_minor == 0) { 1343 mlog(ML_ERROR, "Node %d queried nodes on domain %s " 1344 "but active dlm protocol is %d.%d\n", qn->qn_nodenum, 1345 qn->qn_domain, dlm->dlm_locking_proto.pv_major, 1346 dlm->dlm_locking_proto.pv_minor); 1347 goto bail; 1348 } 1349 1350 status = dlm_match_nodes(dlm, qn); 1351 1352 bail: 1353 if (locked) 1354 spin_unlock(&dlm->spinlock); 1355 spin_unlock(&dlm_domain_lock); 1356 1357 return status; 1358 } 1359 1360 static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data, 1361 void **ret_data) 1362 { 1363 struct dlm_cancel_join *cancel; 1364 struct dlm_ctxt *dlm = NULL; 1365 1366 cancel = (struct dlm_cancel_join *) msg->buf; 1367 1368 mlog(0, "node %u cancels join on domain %s\n", cancel->node_idx, 1369 cancel->domain); 1370 1371 spin_lock(&dlm_domain_lock); 1372 dlm = __dlm_lookup_domain_full(cancel->domain, cancel->name_len); 1373 1374 if (dlm) { 1375 spin_lock(&dlm->spinlock); 1376 1377 /* Yikes, this guy wants to cancel his join. No 1378 * problem, we simply cleanup our join state. */ 1379 BUG_ON(dlm->joining_node != cancel->node_idx); 1380 __dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN); 1381 1382 spin_unlock(&dlm->spinlock); 1383 } 1384 spin_unlock(&dlm_domain_lock); 1385 1386 return 0; 1387 } 1388 1389 static int dlm_send_one_join_cancel(struct dlm_ctxt *dlm, 1390 unsigned int node) 1391 { 1392 int status; 1393 struct dlm_cancel_join cancel_msg; 1394 1395 memset(&cancel_msg, 0, sizeof(cancel_msg)); 1396 cancel_msg.node_idx = dlm->node_num; 1397 cancel_msg.name_len = strlen(dlm->name); 1398 memcpy(cancel_msg.domain, dlm->name, cancel_msg.name_len); 1399 1400 status = o2net_send_message(DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY, 1401 &cancel_msg, sizeof(cancel_msg), node, 1402 NULL); 1403 if (status < 0) { 1404 mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to " 1405 "node %u\n", status, DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY, 1406 node); 1407 goto bail; 1408 } 1409 1410 bail: 1411 return status; 1412 } 1413 1414 /* map_size should be in bytes. */ 1415 static int dlm_send_join_cancels(struct dlm_ctxt *dlm, 1416 unsigned long *node_map, 1417 unsigned int map_size) 1418 { 1419 int status, tmpstat; 1420 unsigned int node; 1421 1422 if (map_size != (BITS_TO_LONGS(O2NM_MAX_NODES) * 1423 sizeof(unsigned long))) { 1424 mlog(ML_ERROR, 1425 "map_size %u != BITS_TO_LONGS(O2NM_MAX_NODES) %u\n", 1426 map_size, (unsigned)BITS_TO_LONGS(O2NM_MAX_NODES)); 1427 return -EINVAL; 1428 } 1429 1430 status = 0; 1431 node = -1; 1432 while ((node = find_next_bit(node_map, O2NM_MAX_NODES, 1433 node + 1)) < O2NM_MAX_NODES) { 1434 if (node == dlm->node_num) 1435 continue; 1436 1437 tmpstat = dlm_send_one_join_cancel(dlm, node); 1438 if (tmpstat) { 1439 mlog(ML_ERROR, "Error return %d cancelling join on " 1440 "node %d\n", tmpstat, node); 1441 if (!status) 1442 status = tmpstat; 1443 } 1444 } 1445 1446 if (status) 1447 mlog_errno(status); 1448 return status; 1449 } 1450 1451 static int dlm_request_join(struct dlm_ctxt *dlm, 1452 int node, 1453 enum dlm_query_join_response_code *response) 1454 { 1455 int status; 1456 struct dlm_query_join_request join_msg; 1457 struct dlm_query_join_packet packet; 1458 u32 join_resp; 1459 1460 mlog(0, "querying node %d\n", node); 1461 1462 memset(&join_msg, 0, sizeof(join_msg)); 1463 join_msg.node_idx = dlm->node_num; 1464 join_msg.name_len = strlen(dlm->name); 1465 memcpy(join_msg.domain, dlm->name, join_msg.name_len); 1466 join_msg.dlm_proto = dlm->dlm_locking_proto; 1467 join_msg.fs_proto = dlm->fs_locking_proto; 1468 1469 /* copy live node map to join message */ 1470 byte_copymap(join_msg.node_map, dlm->live_nodes_map, O2NM_MAX_NODES); 1471 1472 status = o2net_send_message(DLM_QUERY_JOIN_MSG, DLM_MOD_KEY, &join_msg, 1473 sizeof(join_msg), node, &join_resp); 1474 if (status < 0 && status != -ENOPROTOOPT) { 1475 mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to " 1476 "node %u\n", status, DLM_QUERY_JOIN_MSG, DLM_MOD_KEY, 1477 node); 1478 goto bail; 1479 } 1480 dlm_query_join_wire_to_packet(join_resp, &packet); 1481 1482 /* -ENOPROTOOPT from the net code means the other side isn't 1483 listening for our message type -- that's fine, it means 1484 his dlm isn't up, so we can consider him a 'yes' but not 1485 joined into the domain. */ 1486 if (status == -ENOPROTOOPT) { 1487 status = 0; 1488 *response = JOIN_OK_NO_MAP; 1489 } else if (packet.code == JOIN_DISALLOW || 1490 packet.code == JOIN_OK_NO_MAP) { 1491 *response = packet.code; 1492 } else if (packet.code == JOIN_PROTOCOL_MISMATCH) { 1493 mlog(ML_NOTICE, 1494 "This node requested DLM locking protocol %u.%u and " 1495 "filesystem locking protocol %u.%u. At least one of " 1496 "the protocol versions on node %d is not compatible, " 1497 "disconnecting\n", 1498 dlm->dlm_locking_proto.pv_major, 1499 dlm->dlm_locking_proto.pv_minor, 1500 dlm->fs_locking_proto.pv_major, 1501 dlm->fs_locking_proto.pv_minor, 1502 node); 1503 status = -EPROTO; 1504 *response = packet.code; 1505 } else if (packet.code == JOIN_OK) { 1506 *response = packet.code; 1507 /* Use the same locking protocol as the remote node */ 1508 dlm->dlm_locking_proto.pv_minor = packet.dlm_minor; 1509 dlm->fs_locking_proto.pv_minor = packet.fs_minor; 1510 mlog(0, 1511 "Node %d responds JOIN_OK with DLM locking protocol " 1512 "%u.%u and fs locking protocol %u.%u\n", 1513 node, 1514 dlm->dlm_locking_proto.pv_major, 1515 dlm->dlm_locking_proto.pv_minor, 1516 dlm->fs_locking_proto.pv_major, 1517 dlm->fs_locking_proto.pv_minor); 1518 } else { 1519 status = -EINVAL; 1520 mlog(ML_ERROR, "invalid response %d from node %u\n", 1521 packet.code, node); 1522 } 1523 1524 mlog(0, "status %d, node %d response is %d\n", status, node, 1525 *response); 1526 1527 bail: 1528 return status; 1529 } 1530 1531 static int dlm_send_one_join_assert(struct dlm_ctxt *dlm, 1532 unsigned int node) 1533 { 1534 int status; 1535 struct dlm_assert_joined assert_msg; 1536 1537 mlog(0, "Sending join assert to node %u\n", node); 1538 1539 memset(&assert_msg, 0, sizeof(assert_msg)); 1540 assert_msg.node_idx = dlm->node_num; 1541 assert_msg.name_len = strlen(dlm->name); 1542 memcpy(assert_msg.domain, dlm->name, assert_msg.name_len); 1543 1544 status = o2net_send_message(DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY, 1545 &assert_msg, sizeof(assert_msg), node, 1546 NULL); 1547 if (status < 0) 1548 mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to " 1549 "node %u\n", status, DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY, 1550 node); 1551 1552 return status; 1553 } 1554 1555 static void dlm_send_join_asserts(struct dlm_ctxt *dlm, 1556 unsigned long *node_map) 1557 { 1558 int status, node, live; 1559 1560 status = 0; 1561 node = -1; 1562 while ((node = find_next_bit(node_map, O2NM_MAX_NODES, 1563 node + 1)) < O2NM_MAX_NODES) { 1564 if (node == dlm->node_num) 1565 continue; 1566 1567 do { 1568 /* It is very important that this message be 1569 * received so we spin until either the node 1570 * has died or it gets the message. */ 1571 status = dlm_send_one_join_assert(dlm, node); 1572 1573 spin_lock(&dlm->spinlock); 1574 live = test_bit(node, dlm->live_nodes_map); 1575 spin_unlock(&dlm->spinlock); 1576 1577 if (status) { 1578 mlog(ML_ERROR, "Error return %d asserting " 1579 "join on node %d\n", status, node); 1580 1581 /* give us some time between errors... */ 1582 if (live) 1583 msleep(DLM_DOMAIN_BACKOFF_MS); 1584 } 1585 } while (status && live); 1586 } 1587 } 1588 1589 struct domain_join_ctxt { 1590 unsigned long live_map[BITS_TO_LONGS(O2NM_MAX_NODES)]; 1591 unsigned long yes_resp_map[BITS_TO_LONGS(O2NM_MAX_NODES)]; 1592 }; 1593 1594 static int dlm_should_restart_join(struct dlm_ctxt *dlm, 1595 struct domain_join_ctxt *ctxt, 1596 enum dlm_query_join_response_code response) 1597 { 1598 int ret; 1599 1600 if (response == JOIN_DISALLOW) { 1601 mlog(0, "Latest response of disallow -- should restart\n"); 1602 return 1; 1603 } 1604 1605 spin_lock(&dlm->spinlock); 1606 /* For now, we restart the process if the node maps have 1607 * changed at all */ 1608 ret = memcmp(ctxt->live_map, dlm->live_nodes_map, 1609 sizeof(dlm->live_nodes_map)); 1610 spin_unlock(&dlm->spinlock); 1611 1612 if (ret) 1613 mlog(0, "Node maps changed -- should restart\n"); 1614 1615 return ret; 1616 } 1617 1618 static int dlm_try_to_join_domain(struct dlm_ctxt *dlm) 1619 { 1620 int status = 0, tmpstat, node; 1621 struct domain_join_ctxt *ctxt; 1622 enum dlm_query_join_response_code response = JOIN_DISALLOW; 1623 1624 mlog(0, "%p", dlm); 1625 1626 ctxt = kzalloc(sizeof(*ctxt), GFP_KERNEL); 1627 if (!ctxt) { 1628 status = -ENOMEM; 1629 mlog_errno(status); 1630 goto bail; 1631 } 1632 1633 /* group sem locking should work for us here -- we're already 1634 * registered for heartbeat events so filling this should be 1635 * atomic wrt getting those handlers called. */ 1636 o2hb_fill_node_map(dlm->live_nodes_map, sizeof(dlm->live_nodes_map)); 1637 1638 spin_lock(&dlm->spinlock); 1639 memcpy(ctxt->live_map, dlm->live_nodes_map, sizeof(ctxt->live_map)); 1640 1641 __dlm_set_joining_node(dlm, dlm->node_num); 1642 1643 spin_unlock(&dlm->spinlock); 1644 1645 node = -1; 1646 while ((node = find_next_bit(ctxt->live_map, O2NM_MAX_NODES, 1647 node + 1)) < O2NM_MAX_NODES) { 1648 if (node == dlm->node_num) 1649 continue; 1650 1651 status = dlm_request_join(dlm, node, &response); 1652 if (status < 0) { 1653 mlog_errno(status); 1654 goto bail; 1655 } 1656 1657 /* Ok, either we got a response or the node doesn't have a 1658 * dlm up. */ 1659 if (response == JOIN_OK) 1660 set_bit(node, ctxt->yes_resp_map); 1661 1662 if (dlm_should_restart_join(dlm, ctxt, response)) { 1663 status = -EAGAIN; 1664 goto bail; 1665 } 1666 } 1667 1668 mlog(0, "Yay, done querying nodes!\n"); 1669 1670 /* Yay, everyone agree's we can join the domain. My domain is 1671 * comprised of all nodes who were put in the 1672 * yes_resp_map. Copy that into our domain map and send a join 1673 * assert message to clean up everyone elses state. */ 1674 spin_lock(&dlm->spinlock); 1675 memcpy(dlm->domain_map, ctxt->yes_resp_map, 1676 sizeof(ctxt->yes_resp_map)); 1677 set_bit(dlm->node_num, dlm->domain_map); 1678 spin_unlock(&dlm->spinlock); 1679 1680 /* Support for global heartbeat and node info was added in 1.1 */ 1681 if (dlm->dlm_locking_proto.pv_major > 1 || 1682 dlm->dlm_locking_proto.pv_minor > 0) { 1683 status = dlm_send_nodeinfo(dlm, ctxt->yes_resp_map); 1684 if (status) { 1685 mlog_errno(status); 1686 goto bail; 1687 } 1688 status = dlm_send_regions(dlm, ctxt->yes_resp_map); 1689 if (status) { 1690 mlog_errno(status); 1691 goto bail; 1692 } 1693 } 1694 1695 dlm_send_join_asserts(dlm, ctxt->yes_resp_map); 1696 1697 /* Joined state *must* be set before the joining node 1698 * information, otherwise the query_join handler may read no 1699 * current joiner but a state of NEW and tell joining nodes 1700 * we're not in the domain. */ 1701 spin_lock(&dlm_domain_lock); 1702 dlm->dlm_state = DLM_CTXT_JOINED; 1703 dlm->num_joins++; 1704 spin_unlock(&dlm_domain_lock); 1705 1706 bail: 1707 spin_lock(&dlm->spinlock); 1708 __dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN); 1709 if (!status) { 1710 printk(KERN_NOTICE "o2dlm: Joining domain %s ", dlm->name); 1711 __dlm_print_nodes(dlm); 1712 } 1713 spin_unlock(&dlm->spinlock); 1714 1715 if (ctxt) { 1716 /* Do we need to send a cancel message to any nodes? */ 1717 if (status < 0) { 1718 tmpstat = dlm_send_join_cancels(dlm, 1719 ctxt->yes_resp_map, 1720 sizeof(ctxt->yes_resp_map)); 1721 if (tmpstat < 0) 1722 mlog_errno(tmpstat); 1723 } 1724 kfree(ctxt); 1725 } 1726 1727 mlog(0, "returning %d\n", status); 1728 return status; 1729 } 1730 1731 static void dlm_unregister_domain_handlers(struct dlm_ctxt *dlm) 1732 { 1733 o2hb_unregister_callback(dlm->name, &dlm->dlm_hb_up); 1734 o2hb_unregister_callback(dlm->name, &dlm->dlm_hb_down); 1735 o2net_unregister_handler_list(&dlm->dlm_domain_handlers); 1736 } 1737 1738 static int dlm_register_domain_handlers(struct dlm_ctxt *dlm) 1739 { 1740 int status; 1741 1742 mlog(0, "registering handlers.\n"); 1743 1744 o2hb_setup_callback(&dlm->dlm_hb_down, O2HB_NODE_DOWN_CB, 1745 dlm_hb_node_down_cb, dlm, DLM_HB_NODE_DOWN_PRI); 1746 status = o2hb_register_callback(dlm->name, &dlm->dlm_hb_down); 1747 if (status) 1748 goto bail; 1749 1750 o2hb_setup_callback(&dlm->dlm_hb_up, O2HB_NODE_UP_CB, 1751 dlm_hb_node_up_cb, dlm, DLM_HB_NODE_UP_PRI); 1752 status = o2hb_register_callback(dlm->name, &dlm->dlm_hb_up); 1753 if (status) 1754 goto bail; 1755 1756 status = o2net_register_handler(DLM_MASTER_REQUEST_MSG, dlm->key, 1757 sizeof(struct dlm_master_request), 1758 dlm_master_request_handler, 1759 dlm, NULL, &dlm->dlm_domain_handlers); 1760 if (status) 1761 goto bail; 1762 1763 status = o2net_register_handler(DLM_ASSERT_MASTER_MSG, dlm->key, 1764 sizeof(struct dlm_assert_master), 1765 dlm_assert_master_handler, 1766 dlm, dlm_assert_master_post_handler, 1767 &dlm->dlm_domain_handlers); 1768 if (status) 1769 goto bail; 1770 1771 status = o2net_register_handler(DLM_CREATE_LOCK_MSG, dlm->key, 1772 sizeof(struct dlm_create_lock), 1773 dlm_create_lock_handler, 1774 dlm, NULL, &dlm->dlm_domain_handlers); 1775 if (status) 1776 goto bail; 1777 1778 status = o2net_register_handler(DLM_CONVERT_LOCK_MSG, dlm->key, 1779 DLM_CONVERT_LOCK_MAX_LEN, 1780 dlm_convert_lock_handler, 1781 dlm, NULL, &dlm->dlm_domain_handlers); 1782 if (status) 1783 goto bail; 1784 1785 status = o2net_register_handler(DLM_UNLOCK_LOCK_MSG, dlm->key, 1786 DLM_UNLOCK_LOCK_MAX_LEN, 1787 dlm_unlock_lock_handler, 1788 dlm, NULL, &dlm->dlm_domain_handlers); 1789 if (status) 1790 goto bail; 1791 1792 status = o2net_register_handler(DLM_PROXY_AST_MSG, dlm->key, 1793 DLM_PROXY_AST_MAX_LEN, 1794 dlm_proxy_ast_handler, 1795 dlm, NULL, &dlm->dlm_domain_handlers); 1796 if (status) 1797 goto bail; 1798 1799 status = o2net_register_handler(DLM_EXIT_DOMAIN_MSG, dlm->key, 1800 sizeof(struct dlm_exit_domain), 1801 dlm_exit_domain_handler, 1802 dlm, NULL, &dlm->dlm_domain_handlers); 1803 if (status) 1804 goto bail; 1805 1806 status = o2net_register_handler(DLM_DEREF_LOCKRES_MSG, dlm->key, 1807 sizeof(struct dlm_deref_lockres), 1808 dlm_deref_lockres_handler, 1809 dlm, NULL, &dlm->dlm_domain_handlers); 1810 if (status) 1811 goto bail; 1812 1813 status = o2net_register_handler(DLM_MIGRATE_REQUEST_MSG, dlm->key, 1814 sizeof(struct dlm_migrate_request), 1815 dlm_migrate_request_handler, 1816 dlm, NULL, &dlm->dlm_domain_handlers); 1817 if (status) 1818 goto bail; 1819 1820 status = o2net_register_handler(DLM_MIG_LOCKRES_MSG, dlm->key, 1821 DLM_MIG_LOCKRES_MAX_LEN, 1822 dlm_mig_lockres_handler, 1823 dlm, NULL, &dlm->dlm_domain_handlers); 1824 if (status) 1825 goto bail; 1826 1827 status = o2net_register_handler(DLM_MASTER_REQUERY_MSG, dlm->key, 1828 sizeof(struct dlm_master_requery), 1829 dlm_master_requery_handler, 1830 dlm, NULL, &dlm->dlm_domain_handlers); 1831 if (status) 1832 goto bail; 1833 1834 status = o2net_register_handler(DLM_LOCK_REQUEST_MSG, dlm->key, 1835 sizeof(struct dlm_lock_request), 1836 dlm_request_all_locks_handler, 1837 dlm, NULL, &dlm->dlm_domain_handlers); 1838 if (status) 1839 goto bail; 1840 1841 status = o2net_register_handler(DLM_RECO_DATA_DONE_MSG, dlm->key, 1842 sizeof(struct dlm_reco_data_done), 1843 dlm_reco_data_done_handler, 1844 dlm, NULL, &dlm->dlm_domain_handlers); 1845 if (status) 1846 goto bail; 1847 1848 status = o2net_register_handler(DLM_BEGIN_RECO_MSG, dlm->key, 1849 sizeof(struct dlm_begin_reco), 1850 dlm_begin_reco_handler, 1851 dlm, NULL, &dlm->dlm_domain_handlers); 1852 if (status) 1853 goto bail; 1854 1855 status = o2net_register_handler(DLM_FINALIZE_RECO_MSG, dlm->key, 1856 sizeof(struct dlm_finalize_reco), 1857 dlm_finalize_reco_handler, 1858 dlm, NULL, &dlm->dlm_domain_handlers); 1859 if (status) 1860 goto bail; 1861 1862 status = o2net_register_handler(DLM_BEGIN_EXIT_DOMAIN_MSG, dlm->key, 1863 sizeof(struct dlm_exit_domain), 1864 dlm_begin_exit_domain_handler, 1865 dlm, NULL, &dlm->dlm_domain_handlers); 1866 if (status) 1867 goto bail; 1868 1869 bail: 1870 if (status) 1871 dlm_unregister_domain_handlers(dlm); 1872 1873 return status; 1874 } 1875 1876 static int dlm_join_domain(struct dlm_ctxt *dlm) 1877 { 1878 int status; 1879 unsigned int backoff; 1880 unsigned int total_backoff = 0; 1881 1882 BUG_ON(!dlm); 1883 1884 mlog(0, "Join domain %s\n", dlm->name); 1885 1886 status = dlm_register_domain_handlers(dlm); 1887 if (status) { 1888 mlog_errno(status); 1889 goto bail; 1890 } 1891 1892 status = dlm_debug_init(dlm); 1893 if (status < 0) { 1894 mlog_errno(status); 1895 goto bail; 1896 } 1897 1898 status = dlm_launch_thread(dlm); 1899 if (status < 0) { 1900 mlog_errno(status); 1901 goto bail; 1902 } 1903 1904 status = dlm_launch_recovery_thread(dlm); 1905 if (status < 0) { 1906 mlog_errno(status); 1907 goto bail; 1908 } 1909 1910 dlm->dlm_worker = create_singlethread_workqueue("dlm_wq"); 1911 if (!dlm->dlm_worker) { 1912 status = -ENOMEM; 1913 mlog_errno(status); 1914 goto bail; 1915 } 1916 1917 do { 1918 status = dlm_try_to_join_domain(dlm); 1919 1920 /* If we're racing another node to the join, then we 1921 * need to back off temporarily and let them 1922 * complete. */ 1923 #define DLM_JOIN_TIMEOUT_MSECS 90000 1924 if (status == -EAGAIN) { 1925 if (signal_pending(current)) { 1926 status = -ERESTARTSYS; 1927 goto bail; 1928 } 1929 1930 if (total_backoff > 1931 msecs_to_jiffies(DLM_JOIN_TIMEOUT_MSECS)) { 1932 status = -ERESTARTSYS; 1933 mlog(ML_NOTICE, "Timed out joining dlm domain " 1934 "%s after %u msecs\n", dlm->name, 1935 jiffies_to_msecs(total_backoff)); 1936 goto bail; 1937 } 1938 1939 /* 1940 * <chip> After you! 1941 * <dale> No, after you! 1942 * <chip> I insist! 1943 * <dale> But you first! 1944 * ... 1945 */ 1946 backoff = (unsigned int)(jiffies & 0x3); 1947 backoff *= DLM_DOMAIN_BACKOFF_MS; 1948 total_backoff += backoff; 1949 mlog(0, "backoff %d\n", backoff); 1950 msleep(backoff); 1951 } 1952 } while (status == -EAGAIN); 1953 1954 if (status < 0) { 1955 mlog_errno(status); 1956 goto bail; 1957 } 1958 1959 status = 0; 1960 bail: 1961 wake_up(&dlm_domain_events); 1962 1963 if (status) { 1964 dlm_unregister_domain_handlers(dlm); 1965 dlm_debug_shutdown(dlm); 1966 dlm_complete_thread(dlm); 1967 dlm_complete_recovery_thread(dlm); 1968 dlm_destroy_dlm_worker(dlm); 1969 } 1970 1971 return status; 1972 } 1973 1974 static struct dlm_ctxt *dlm_alloc_ctxt(const char *domain, 1975 u32 key) 1976 { 1977 int i; 1978 int ret; 1979 struct dlm_ctxt *dlm = NULL; 1980 1981 dlm = kzalloc(sizeof(*dlm), GFP_KERNEL); 1982 if (!dlm) { 1983 mlog_errno(-ENOMEM); 1984 goto leave; 1985 } 1986 1987 dlm->name = kstrdup(domain, GFP_KERNEL); 1988 if (dlm->name == NULL) { 1989 mlog_errno(-ENOMEM); 1990 kfree(dlm); 1991 dlm = NULL; 1992 goto leave; 1993 } 1994 1995 dlm->lockres_hash = (struct hlist_head **)dlm_alloc_pagevec(DLM_HASH_PAGES); 1996 if (!dlm->lockres_hash) { 1997 mlog_errno(-ENOMEM); 1998 kfree(dlm->name); 1999 kfree(dlm); 2000 dlm = NULL; 2001 goto leave; 2002 } 2003 2004 for (i = 0; i < DLM_HASH_BUCKETS; i++) 2005 INIT_HLIST_HEAD(dlm_lockres_hash(dlm, i)); 2006 2007 dlm->master_hash = (struct hlist_head **) 2008 dlm_alloc_pagevec(DLM_HASH_PAGES); 2009 if (!dlm->master_hash) { 2010 mlog_errno(-ENOMEM); 2011 dlm_free_pagevec((void **)dlm->lockres_hash, DLM_HASH_PAGES); 2012 kfree(dlm->name); 2013 kfree(dlm); 2014 dlm = NULL; 2015 goto leave; 2016 } 2017 2018 for (i = 0; i < DLM_HASH_BUCKETS; i++) 2019 INIT_HLIST_HEAD(dlm_master_hash(dlm, i)); 2020 2021 dlm->key = key; 2022 dlm->node_num = o2nm_this_node(); 2023 2024 ret = dlm_create_debugfs_subroot(dlm); 2025 if (ret < 0) { 2026 dlm_free_pagevec((void **)dlm->master_hash, DLM_HASH_PAGES); 2027 dlm_free_pagevec((void **)dlm->lockres_hash, DLM_HASH_PAGES); 2028 kfree(dlm->name); 2029 kfree(dlm); 2030 dlm = NULL; 2031 goto leave; 2032 } 2033 2034 spin_lock_init(&dlm->spinlock); 2035 spin_lock_init(&dlm->master_lock); 2036 spin_lock_init(&dlm->ast_lock); 2037 spin_lock_init(&dlm->track_lock); 2038 INIT_LIST_HEAD(&dlm->list); 2039 INIT_LIST_HEAD(&dlm->dirty_list); 2040 INIT_LIST_HEAD(&dlm->reco.resources); 2041 INIT_LIST_HEAD(&dlm->reco.received); 2042 INIT_LIST_HEAD(&dlm->reco.node_data); 2043 INIT_LIST_HEAD(&dlm->purge_list); 2044 INIT_LIST_HEAD(&dlm->dlm_domain_handlers); 2045 INIT_LIST_HEAD(&dlm->tracking_list); 2046 dlm->reco.state = 0; 2047 2048 INIT_LIST_HEAD(&dlm->pending_asts); 2049 INIT_LIST_HEAD(&dlm->pending_basts); 2050 2051 mlog(0, "dlm->recovery_map=%p, &(dlm->recovery_map[0])=%p\n", 2052 dlm->recovery_map, &(dlm->recovery_map[0])); 2053 2054 memset(dlm->recovery_map, 0, sizeof(dlm->recovery_map)); 2055 memset(dlm->live_nodes_map, 0, sizeof(dlm->live_nodes_map)); 2056 memset(dlm->domain_map, 0, sizeof(dlm->domain_map)); 2057 2058 dlm->dlm_thread_task = NULL; 2059 dlm->dlm_reco_thread_task = NULL; 2060 dlm->dlm_worker = NULL; 2061 init_waitqueue_head(&dlm->dlm_thread_wq); 2062 init_waitqueue_head(&dlm->dlm_reco_thread_wq); 2063 init_waitqueue_head(&dlm->reco.event); 2064 init_waitqueue_head(&dlm->ast_wq); 2065 init_waitqueue_head(&dlm->migration_wq); 2066 INIT_LIST_HEAD(&dlm->mle_hb_events); 2067 2068 dlm->joining_node = DLM_LOCK_RES_OWNER_UNKNOWN; 2069 init_waitqueue_head(&dlm->dlm_join_events); 2070 2071 dlm->reco.new_master = O2NM_INVALID_NODE_NUM; 2072 dlm->reco.dead_node = O2NM_INVALID_NODE_NUM; 2073 2074 atomic_set(&dlm->res_tot_count, 0); 2075 atomic_set(&dlm->res_cur_count, 0); 2076 for (i = 0; i < DLM_MLE_NUM_TYPES; ++i) { 2077 atomic_set(&dlm->mle_tot_count[i], 0); 2078 atomic_set(&dlm->mle_cur_count[i], 0); 2079 } 2080 2081 spin_lock_init(&dlm->work_lock); 2082 INIT_LIST_HEAD(&dlm->work_list); 2083 INIT_WORK(&dlm->dispatched_work, dlm_dispatch_work); 2084 2085 kref_init(&dlm->dlm_refs); 2086 dlm->dlm_state = DLM_CTXT_NEW; 2087 2088 INIT_LIST_HEAD(&dlm->dlm_eviction_callbacks); 2089 2090 mlog(0, "context init: refcount %u\n", 2091 atomic_read(&dlm->dlm_refs.refcount)); 2092 2093 leave: 2094 return dlm; 2095 } 2096 2097 /* 2098 * Compare a requested locking protocol version against the current one. 2099 * 2100 * If the major numbers are different, they are incompatible. 2101 * If the current minor is greater than the request, they are incompatible. 2102 * If the current minor is less than or equal to the request, they are 2103 * compatible, and the requester should run at the current minor version. 2104 */ 2105 static int dlm_protocol_compare(struct dlm_protocol_version *existing, 2106 struct dlm_protocol_version *request) 2107 { 2108 if (existing->pv_major != request->pv_major) 2109 return 1; 2110 2111 if (existing->pv_minor > request->pv_minor) 2112 return 1; 2113 2114 if (existing->pv_minor < request->pv_minor) 2115 request->pv_minor = existing->pv_minor; 2116 2117 return 0; 2118 } 2119 2120 /* 2121 * dlm_register_domain: one-time setup per "domain". 2122 * 2123 * The filesystem passes in the requested locking version via proto. 2124 * If registration was successful, proto will contain the negotiated 2125 * locking protocol. 2126 */ 2127 struct dlm_ctxt * dlm_register_domain(const char *domain, 2128 u32 key, 2129 struct dlm_protocol_version *fs_proto) 2130 { 2131 int ret; 2132 struct dlm_ctxt *dlm = NULL; 2133 struct dlm_ctxt *new_ctxt = NULL; 2134 2135 if (strlen(domain) >= O2NM_MAX_NAME_LEN) { 2136 ret = -ENAMETOOLONG; 2137 mlog(ML_ERROR, "domain name length too long\n"); 2138 goto leave; 2139 } 2140 2141 mlog(0, "register called for domain \"%s\"\n", domain); 2142 2143 retry: 2144 dlm = NULL; 2145 if (signal_pending(current)) { 2146 ret = -ERESTARTSYS; 2147 mlog_errno(ret); 2148 goto leave; 2149 } 2150 2151 spin_lock(&dlm_domain_lock); 2152 2153 dlm = __dlm_lookup_domain(domain); 2154 if (dlm) { 2155 if (dlm->dlm_state != DLM_CTXT_JOINED) { 2156 spin_unlock(&dlm_domain_lock); 2157 2158 mlog(0, "This ctxt is not joined yet!\n"); 2159 wait_event_interruptible(dlm_domain_events, 2160 dlm_wait_on_domain_helper( 2161 domain)); 2162 goto retry; 2163 } 2164 2165 if (dlm_protocol_compare(&dlm->fs_locking_proto, fs_proto)) { 2166 spin_unlock(&dlm_domain_lock); 2167 mlog(ML_ERROR, 2168 "Requested locking protocol version is not " 2169 "compatible with already registered domain " 2170 "\"%s\"\n", domain); 2171 ret = -EPROTO; 2172 goto leave; 2173 } 2174 2175 __dlm_get(dlm); 2176 dlm->num_joins++; 2177 2178 spin_unlock(&dlm_domain_lock); 2179 2180 ret = 0; 2181 goto leave; 2182 } 2183 2184 /* doesn't exist */ 2185 if (!new_ctxt) { 2186 spin_unlock(&dlm_domain_lock); 2187 2188 new_ctxt = dlm_alloc_ctxt(domain, key); 2189 if (new_ctxt) 2190 goto retry; 2191 2192 ret = -ENOMEM; 2193 mlog_errno(ret); 2194 goto leave; 2195 } 2196 2197 /* a little variable switch-a-roo here... */ 2198 dlm = new_ctxt; 2199 new_ctxt = NULL; 2200 2201 /* add the new domain */ 2202 list_add_tail(&dlm->list, &dlm_domains); 2203 spin_unlock(&dlm_domain_lock); 2204 2205 /* 2206 * Pass the locking protocol version into the join. If the join 2207 * succeeds, it will have the negotiated protocol set. 2208 */ 2209 dlm->dlm_locking_proto = dlm_protocol; 2210 dlm->fs_locking_proto = *fs_proto; 2211 2212 ret = dlm_join_domain(dlm); 2213 if (ret) { 2214 mlog_errno(ret); 2215 dlm_put(dlm); 2216 goto leave; 2217 } 2218 2219 /* Tell the caller what locking protocol we negotiated */ 2220 *fs_proto = dlm->fs_locking_proto; 2221 2222 ret = 0; 2223 leave: 2224 if (new_ctxt) 2225 dlm_free_ctxt_mem(new_ctxt); 2226 2227 if (ret < 0) 2228 dlm = ERR_PTR(ret); 2229 2230 return dlm; 2231 } 2232 EXPORT_SYMBOL_GPL(dlm_register_domain); 2233 2234 static LIST_HEAD(dlm_join_handlers); 2235 2236 static void dlm_unregister_net_handlers(void) 2237 { 2238 o2net_unregister_handler_list(&dlm_join_handlers); 2239 } 2240 2241 static int dlm_register_net_handlers(void) 2242 { 2243 int status = 0; 2244 2245 status = o2net_register_handler(DLM_QUERY_JOIN_MSG, DLM_MOD_KEY, 2246 sizeof(struct dlm_query_join_request), 2247 dlm_query_join_handler, 2248 NULL, NULL, &dlm_join_handlers); 2249 if (status) 2250 goto bail; 2251 2252 status = o2net_register_handler(DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY, 2253 sizeof(struct dlm_assert_joined), 2254 dlm_assert_joined_handler, 2255 NULL, NULL, &dlm_join_handlers); 2256 if (status) 2257 goto bail; 2258 2259 status = o2net_register_handler(DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY, 2260 sizeof(struct dlm_cancel_join), 2261 dlm_cancel_join_handler, 2262 NULL, NULL, &dlm_join_handlers); 2263 if (status) 2264 goto bail; 2265 2266 status = o2net_register_handler(DLM_QUERY_REGION, DLM_MOD_KEY, 2267 sizeof(struct dlm_query_region), 2268 dlm_query_region_handler, 2269 NULL, NULL, &dlm_join_handlers); 2270 2271 if (status) 2272 goto bail; 2273 2274 status = o2net_register_handler(DLM_QUERY_NODEINFO, DLM_MOD_KEY, 2275 sizeof(struct dlm_query_nodeinfo), 2276 dlm_query_nodeinfo_handler, 2277 NULL, NULL, &dlm_join_handlers); 2278 bail: 2279 if (status < 0) 2280 dlm_unregister_net_handlers(); 2281 2282 return status; 2283 } 2284 2285 /* Domain eviction callback handling. 2286 * 2287 * The file system requires notification of node death *before* the 2288 * dlm completes it's recovery work, otherwise it may be able to 2289 * acquire locks on resources requiring recovery. Since the dlm can 2290 * evict a node from it's domain *before* heartbeat fires, a similar 2291 * mechanism is required. */ 2292 2293 /* Eviction is not expected to happen often, so a per-domain lock is 2294 * not necessary. Eviction callbacks are allowed to sleep for short 2295 * periods of time. */ 2296 static DECLARE_RWSEM(dlm_callback_sem); 2297 2298 void dlm_fire_domain_eviction_callbacks(struct dlm_ctxt *dlm, 2299 int node_num) 2300 { 2301 struct list_head *iter; 2302 struct dlm_eviction_cb *cb; 2303 2304 down_read(&dlm_callback_sem); 2305 list_for_each(iter, &dlm->dlm_eviction_callbacks) { 2306 cb = list_entry(iter, struct dlm_eviction_cb, ec_item); 2307 2308 cb->ec_func(node_num, cb->ec_data); 2309 } 2310 up_read(&dlm_callback_sem); 2311 } 2312 2313 void dlm_setup_eviction_cb(struct dlm_eviction_cb *cb, 2314 dlm_eviction_func *f, 2315 void *data) 2316 { 2317 INIT_LIST_HEAD(&cb->ec_item); 2318 cb->ec_func = f; 2319 cb->ec_data = data; 2320 } 2321 EXPORT_SYMBOL_GPL(dlm_setup_eviction_cb); 2322 2323 void dlm_register_eviction_cb(struct dlm_ctxt *dlm, 2324 struct dlm_eviction_cb *cb) 2325 { 2326 down_write(&dlm_callback_sem); 2327 list_add_tail(&cb->ec_item, &dlm->dlm_eviction_callbacks); 2328 up_write(&dlm_callback_sem); 2329 } 2330 EXPORT_SYMBOL_GPL(dlm_register_eviction_cb); 2331 2332 void dlm_unregister_eviction_cb(struct dlm_eviction_cb *cb) 2333 { 2334 down_write(&dlm_callback_sem); 2335 list_del_init(&cb->ec_item); 2336 up_write(&dlm_callback_sem); 2337 } 2338 EXPORT_SYMBOL_GPL(dlm_unregister_eviction_cb); 2339 2340 static int __init dlm_init(void) 2341 { 2342 int status; 2343 2344 dlm_print_version(); 2345 2346 status = dlm_init_mle_cache(); 2347 if (status) { 2348 mlog(ML_ERROR, "Could not create o2dlm_mle slabcache\n"); 2349 goto error; 2350 } 2351 2352 status = dlm_init_master_caches(); 2353 if (status) { 2354 mlog(ML_ERROR, "Could not create o2dlm_lockres and " 2355 "o2dlm_lockname slabcaches\n"); 2356 goto error; 2357 } 2358 2359 status = dlm_init_lock_cache(); 2360 if (status) { 2361 mlog(ML_ERROR, "Count not create o2dlm_lock slabcache\n"); 2362 goto error; 2363 } 2364 2365 status = dlm_register_net_handlers(); 2366 if (status) { 2367 mlog(ML_ERROR, "Unable to register network handlers\n"); 2368 goto error; 2369 } 2370 2371 status = dlm_create_debugfs_root(); 2372 if (status) 2373 goto error; 2374 2375 return 0; 2376 error: 2377 dlm_unregister_net_handlers(); 2378 dlm_destroy_lock_cache(); 2379 dlm_destroy_master_caches(); 2380 dlm_destroy_mle_cache(); 2381 return -1; 2382 } 2383 2384 static void __exit dlm_exit (void) 2385 { 2386 dlm_destroy_debugfs_root(); 2387 dlm_unregister_net_handlers(); 2388 dlm_destroy_lock_cache(); 2389 dlm_destroy_master_caches(); 2390 dlm_destroy_mle_cache(); 2391 } 2392 2393 MODULE_AUTHOR("Oracle"); 2394 MODULE_LICENSE("GPL"); 2395 2396 module_init(dlm_init); 2397 module_exit(dlm_exit); 2398