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