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