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_joined(struct dlm_ctxt *dlm) 678{ 679 int ret = 0; 680 681 spin_lock(&dlm_domain_lock); 682 683 if (dlm->dlm_state == DLM_CTXT_JOINED) 684 ret = 1; 685 686 spin_unlock(&dlm_domain_lock); 687 688 return ret; 689} 690 691int dlm_shutting_down(struct dlm_ctxt *dlm) 692{ 693 int ret = 0; 694 695 spin_lock(&dlm_domain_lock); 696 697 if (dlm->dlm_state == DLM_CTXT_IN_SHUTDOWN) 698 ret = 1; 699 700 spin_unlock(&dlm_domain_lock); 701 702 return ret; 703} 704 705void dlm_unregister_domain(struct dlm_ctxt *dlm) 706{ 707 int leave = 0; 708 struct dlm_lock_resource *res; 709 710 spin_lock(&dlm_domain_lock); 711 BUG_ON(dlm->dlm_state != DLM_CTXT_JOINED); 712 BUG_ON(!dlm->num_joins); 713 714 dlm->num_joins--; 715 if (!dlm->num_joins) { 716 /* We mark it "in shutdown" now so new register 717 * requests wait until we've completely left the 718 * domain. Don't use DLM_CTXT_LEAVING yet as we still 719 * want new domain joins to communicate with us at 720 * least until we've completed migration of our 721 * resources. */ 722 dlm->dlm_state = DLM_CTXT_IN_SHUTDOWN; 723 leave = 1; 724 } 725 spin_unlock(&dlm_domain_lock); 726 727 if (leave) { 728 mlog(0, "shutting down domain %s\n", dlm->name); 729 dlm_begin_exit_domain(dlm); 730 731 /* We changed dlm state, notify the thread */ 732 dlm_kick_thread(dlm, NULL); 733 734 while (dlm_migrate_all_locks(dlm)) { 735 /* Give dlm_thread time to purge the lockres' */ 736 msleep(500); 737 mlog(0, "%s: more migration to do\n", dlm->name); 738 } 739 740 /* This list should be empty. If not, print remaining lockres */ 741 if (!list_empty(&dlm->tracking_list)) { 742 mlog(ML_ERROR, "Following lockres' are still on the " 743 "tracking list:\n"); 744 list_for_each_entry(res, &dlm->tracking_list, tracking) 745 dlm_print_one_lock_resource(res); 746 } 747 748 dlm_mark_domain_leaving(dlm); 749 dlm_leave_domain(dlm); 750 printk(KERN_NOTICE "o2dlm: Leaving domain %s\n", dlm->name); 751 dlm_force_free_mles(dlm); 752 dlm_complete_dlm_shutdown(dlm); 753 } 754 dlm_put(dlm); 755} 756EXPORT_SYMBOL_GPL(dlm_unregister_domain); 757 758static int dlm_query_join_proto_check(char *proto_type, int node, 759 struct dlm_protocol_version *ours, 760 struct dlm_protocol_version *request) 761{ 762 int rc; 763 struct dlm_protocol_version proto = *request; 764 765 if (!dlm_protocol_compare(ours, &proto)) { 766 mlog(0, 767 "node %u wanted to join with %s locking protocol " 768 "%u.%u, we respond with %u.%u\n", 769 node, proto_type, 770 request->pv_major, 771 request->pv_minor, 772 proto.pv_major, proto.pv_minor); 773 request->pv_minor = proto.pv_minor; 774 rc = 0; 775 } else { 776 mlog(ML_NOTICE, 777 "Node %u wanted to join with %s locking " 778 "protocol %u.%u, but we have %u.%u, disallowing\n", 779 node, proto_type, 780 request->pv_major, 781 request->pv_minor, 782 ours->pv_major, 783 ours->pv_minor); 784 rc = 1; 785 } 786 787 return rc; 788} 789 790/* 791 * struct dlm_query_join_packet is made up of four one-byte fields. They 792 * are effectively in big-endian order already. However, little-endian 793 * machines swap them before putting the packet on the wire (because 794 * query_join's response is a status, and that status is treated as a u32 795 * on the wire). Thus, a big-endian and little-endian machines will treat 796 * this structure differently. 797 * 798 * The solution is to have little-endian machines swap the structure when 799 * converting from the structure to the u32 representation. This will 800 * result in the structure having the correct format on the wire no matter 801 * the host endian format. 802 */ 803static void dlm_query_join_packet_to_wire(struct dlm_query_join_packet *packet, 804 u32 *wire) 805{ 806 union dlm_query_join_response response; 807 808 response.packet = *packet; 809 *wire = be32_to_cpu(response.intval); 810} 811 812static void dlm_query_join_wire_to_packet(u32 wire, 813 struct dlm_query_join_packet *packet) 814{ 815 union dlm_query_join_response response; 816 817 response.intval = cpu_to_be32(wire); 818 *packet = response.packet; 819} 820 821static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data, 822 void **ret_data) 823{ 824 struct dlm_query_join_request *query; 825 struct dlm_query_join_packet packet = { 826 .code = JOIN_DISALLOW, 827 }; 828 struct dlm_ctxt *dlm = NULL; 829 u32 response; 830 u8 nodenum; 831 832 query = (struct dlm_query_join_request *) msg->buf; 833 834 mlog(0, "node %u wants to join domain %s\n", query->node_idx, 835 query->domain); 836 837 /* 838 * If heartbeat doesn't consider the node live, tell it 839 * to back off and try again. This gives heartbeat a chance 840 * to catch up. 841 */ 842 if (!o2hb_check_node_heartbeating_no_sem(query->node_idx)) { 843 mlog(0, "node %u is not in our live map yet\n", 844 query->node_idx); 845 846 packet.code = JOIN_DISALLOW; 847 goto respond; 848 } 849 850 packet.code = JOIN_OK_NO_MAP; 851 852 spin_lock(&dlm_domain_lock); 853 dlm = __dlm_lookup_domain_full(query->domain, query->name_len); 854 if (!dlm) 855 goto unlock_respond; 856 857 /* 858 * There is a small window where the joining node may not see the 859 * node(s) that just left but still part of the cluster. DISALLOW 860 * join request if joining node has different node map. 861 */ 862 nodenum=0; 863 while (nodenum < O2NM_MAX_NODES) { 864 if (test_bit(nodenum, dlm->domain_map)) { 865 if (!byte_test_bit(nodenum, query->node_map)) { 866 mlog(0, "disallow join as node %u does not " 867 "have node %u in its nodemap\n", 868 query->node_idx, nodenum); 869 packet.code = JOIN_DISALLOW; 870 goto unlock_respond; 871 } 872 } 873 nodenum++; 874 } 875 876 /* Once the dlm ctxt is marked as leaving then we don't want 877 * to be put in someone's domain map. 878 * Also, explicitly disallow joining at certain troublesome 879 * times (ie. during recovery). */ 880 if (dlm && dlm->dlm_state != DLM_CTXT_LEAVING) { 881 int bit = query->node_idx; 882 spin_lock(&dlm->spinlock); 883 884 if (dlm->dlm_state == DLM_CTXT_NEW && 885 dlm->joining_node == DLM_LOCK_RES_OWNER_UNKNOWN) { 886 /*If this is a brand new context and we 887 * haven't started our join process yet, then 888 * the other node won the race. */ 889 packet.code = JOIN_OK_NO_MAP; 890 } else if (dlm->joining_node != DLM_LOCK_RES_OWNER_UNKNOWN) { 891 /* Disallow parallel joins. */ 892 packet.code = JOIN_DISALLOW; 893 } else if (dlm->reco.state & DLM_RECO_STATE_ACTIVE) { 894 mlog(0, "node %u trying to join, but recovery " 895 "is ongoing.\n", bit); 896 packet.code = JOIN_DISALLOW; 897 } else if (test_bit(bit, dlm->recovery_map)) { 898 mlog(0, "node %u trying to join, but it " 899 "still needs recovery.\n", bit); 900 packet.code = JOIN_DISALLOW; 901 } else if (test_bit(bit, dlm->domain_map)) { 902 mlog(0, "node %u trying to join, but it " 903 "is still in the domain! needs recovery?\n", 904 bit); 905 packet.code = JOIN_DISALLOW; 906 } else { 907 /* Alright we're fully a part of this domain 908 * so we keep some state as to who's joining 909 * and indicate to him that needs to be fixed 910 * up. */ 911 912 /* Make sure we speak compatible locking protocols. */ 913 if (dlm_query_join_proto_check("DLM", bit, 914 &dlm->dlm_locking_proto, 915 &query->dlm_proto)) { 916 packet.code = JOIN_PROTOCOL_MISMATCH; 917 } else if (dlm_query_join_proto_check("fs", bit, 918 &dlm->fs_locking_proto, 919 &query->fs_proto)) { 920 packet.code = JOIN_PROTOCOL_MISMATCH; 921 } else { 922 packet.dlm_minor = query->dlm_proto.pv_minor; 923 packet.fs_minor = query->fs_proto.pv_minor; 924 packet.code = JOIN_OK; 925 __dlm_set_joining_node(dlm, query->node_idx); 926 } 927 } 928 929 spin_unlock(&dlm->spinlock); 930 } 931unlock_respond: 932 spin_unlock(&dlm_domain_lock); 933 934respond: 935 mlog(0, "We respond with %u\n", packet.code); 936 937 dlm_query_join_packet_to_wire(&packet, &response); 938 return response; 939} 940 941static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data, 942 void **ret_data) 943{ 944 struct dlm_assert_joined *assert; 945 struct dlm_ctxt *dlm = NULL; 946 947 assert = (struct dlm_assert_joined *) msg->buf; 948 949 mlog(0, "node %u asserts join on domain %s\n", assert->node_idx, 950 assert->domain); 951 952 spin_lock(&dlm_domain_lock); 953 dlm = __dlm_lookup_domain_full(assert->domain, assert->name_len); 954 /* XXX should we consider no dlm ctxt an error? */ 955 if (dlm) { 956 spin_lock(&dlm->spinlock); 957 958 /* Alright, this node has officially joined our 959 * domain. Set him in the map and clean up our 960 * leftover join state. */ 961 BUG_ON(dlm->joining_node != assert->node_idx); 962 963 if (dlm->reco.state & DLM_RECO_STATE_ACTIVE) { 964 mlog(0, "dlm recovery is ongoing, disallow join\n"); 965 spin_unlock(&dlm->spinlock); 966 spin_unlock(&dlm_domain_lock); 967 return -EAGAIN; 968 } 969 970 set_bit(assert->node_idx, dlm->domain_map); 971 clear_bit(assert->node_idx, dlm->exit_domain_map); 972 __dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN); 973 974 printk(KERN_NOTICE "o2dlm: Node %u joins domain %s ", 975 assert->node_idx, dlm->name); 976 __dlm_print_nodes(dlm); 977 978 /* notify anything attached to the heartbeat events */ 979 dlm_hb_event_notify_attached(dlm, assert->node_idx, 1); 980 981 spin_unlock(&dlm->spinlock); 982 } 983 spin_unlock(&dlm_domain_lock); 984 985 return 0; 986} 987 988static int dlm_match_regions(struct dlm_ctxt *dlm, 989 struct dlm_query_region *qr, 990 char *local, int locallen) 991{ 992 char *remote = qr->qr_regions; 993 char *l, *r; 994 int localnr, i, j, foundit; 995 int status = 0; 996 997 if (!o2hb_global_heartbeat_active()) { 998 if (qr->qr_numregions) { 999 mlog(ML_ERROR, "Domain %s: Joining node %d has global " 1000 "heartbeat enabled but local node %d does not\n", 1001 qr->qr_domain, qr->qr_node, dlm->node_num); 1002 status = -EINVAL; 1003 } 1004 goto bail; 1005 } 1006 1007 if (o2hb_global_heartbeat_active() && !qr->qr_numregions) { 1008 mlog(ML_ERROR, "Domain %s: Local node %d has global " 1009 "heartbeat enabled but joining node %d does not\n", 1010 qr->qr_domain, dlm->node_num, qr->qr_node); 1011 status = -EINVAL; 1012 goto bail; 1013 } 1014 1015 r = remote; 1016 for (i = 0; i < qr->qr_numregions; ++i) { 1017 mlog(0, "Region %.*s\n", O2HB_MAX_REGION_NAME_LEN, r); 1018 r += O2HB_MAX_REGION_NAME_LEN; 1019 } 1020 1021 localnr = min(O2NM_MAX_REGIONS, locallen/O2HB_MAX_REGION_NAME_LEN); 1022 localnr = o2hb_get_all_regions(local, (u8)localnr); 1023 1024 /* compare local regions with remote */ 1025 l = local; 1026 for (i = 0; i < localnr; ++i) { 1027 foundit = 0; 1028 r = remote; 1029 for (j = 0; j <= qr->qr_numregions; ++j) { 1030 if (!memcmp(l, r, O2HB_MAX_REGION_NAME_LEN)) { 1031 foundit = 1; 1032 break; 1033 } 1034 r += O2HB_MAX_REGION_NAME_LEN; 1035 } 1036 if (!foundit) { 1037 status = -EINVAL; 1038 mlog(ML_ERROR, "Domain %s: Region '%.*s' registered " 1039 "in local node %d but not in joining node %d\n", 1040 qr->qr_domain, O2HB_MAX_REGION_NAME_LEN, l, 1041 dlm->node_num, qr->qr_node); 1042 goto bail; 1043 } 1044 l += O2HB_MAX_REGION_NAME_LEN; 1045 } 1046 1047 /* compare remote with local regions */ 1048 r = remote; 1049 for (i = 0; i < qr->qr_numregions; ++i) { 1050 foundit = 0; 1051 l = local; 1052 for (j = 0; j < localnr; ++j) { 1053 if (!memcmp(r, l, O2HB_MAX_REGION_NAME_LEN)) { 1054 foundit = 1; 1055 break; 1056 } 1057 l += O2HB_MAX_REGION_NAME_LEN; 1058 } 1059 if (!foundit) { 1060 status = -EINVAL; 1061 mlog(ML_ERROR, "Domain %s: Region '%.*s' registered " 1062 "in joining node %d but not in local node %d\n", 1063 qr->qr_domain, O2HB_MAX_REGION_NAME_LEN, r, 1064 qr->qr_node, dlm->node_num); 1065 goto bail; 1066 } 1067 r += O2HB_MAX_REGION_NAME_LEN; 1068 } 1069 1070bail: 1071 return status; 1072} 1073 1074static int dlm_send_regions(struct dlm_ctxt *dlm, unsigned long *node_map) 1075{ 1076 struct dlm_query_region *qr = NULL; 1077 int status, ret = 0, i; 1078 char *p; 1079 1080 if (find_next_bit(node_map, O2NM_MAX_NODES, 0) >= O2NM_MAX_NODES) 1081 goto bail; 1082 1083 qr = kzalloc(sizeof(struct dlm_query_region), GFP_KERNEL); 1084 if (!qr) { 1085 ret = -ENOMEM; 1086 mlog_errno(ret); 1087 goto bail; 1088 } 1089 1090 qr->qr_node = dlm->node_num; 1091 qr->qr_namelen = strlen(dlm->name); 1092 memcpy(qr->qr_domain, dlm->name, qr->qr_namelen); 1093 /* if local hb, the numregions will be zero */ 1094 if (o2hb_global_heartbeat_active()) 1095 qr->qr_numregions = o2hb_get_all_regions(qr->qr_regions, 1096 O2NM_MAX_REGIONS); 1097 1098 p = qr->qr_regions; 1099 for (i = 0; i < qr->qr_numregions; ++i, p += O2HB_MAX_REGION_NAME_LEN) 1100 mlog(0, "Region %.*s\n", O2HB_MAX_REGION_NAME_LEN, p); 1101 1102 i = -1; 1103 while ((i = find_next_bit(node_map, O2NM_MAX_NODES, 1104 i + 1)) < O2NM_MAX_NODES) { 1105 if (i == dlm->node_num) 1106 continue; 1107 1108 mlog(0, "Sending regions to node %d\n", i); 1109 1110 ret = o2net_send_message(DLM_QUERY_REGION, DLM_MOD_KEY, qr, 1111 sizeof(struct dlm_query_region), 1112 i, &status); 1113 if (ret >= 0) 1114 ret = status; 1115 if (ret) { 1116 mlog(ML_ERROR, "Region mismatch %d, node %d\n", 1117 ret, i); 1118 break; 1119 } 1120 } 1121 1122bail: 1123 kfree(qr); 1124 return ret; 1125} 1126 1127static int dlm_query_region_handler(struct o2net_msg *msg, u32 len, 1128 void *data, void **ret_data) 1129{ 1130 struct dlm_query_region *qr; 1131 struct dlm_ctxt *dlm = NULL; 1132 char *local = NULL; 1133 int status = 0; 1134 1135 qr = (struct dlm_query_region *) msg->buf; 1136 1137 mlog(0, "Node %u queries hb regions on domain %s\n", qr->qr_node, 1138 qr->qr_domain); 1139 1140 /* buffer used in dlm_mast_regions() */ 1141 local = kmalloc(sizeof(qr->qr_regions), GFP_KERNEL); 1142 if (!local) 1143 return -ENOMEM; 1144 1145 status = -EINVAL; 1146 1147 spin_lock(&dlm_domain_lock); 1148 dlm = __dlm_lookup_domain_full(qr->qr_domain, qr->qr_namelen); 1149 if (!dlm) { 1150 mlog(ML_ERROR, "Node %d queried hb regions on domain %s " 1151 "before join domain\n", qr->qr_node, qr->qr_domain); 1152 goto out_domain_lock; 1153 } 1154 1155 spin_lock(&dlm->spinlock); 1156 if (dlm->joining_node != qr->qr_node) { 1157 mlog(ML_ERROR, "Node %d queried hb regions on domain %s " 1158 "but joining node is %d\n", qr->qr_node, qr->qr_domain, 1159 dlm->joining_node); 1160 goto out_dlm_lock; 1161 } 1162 1163 /* Support for global heartbeat was added in 1.1 */ 1164 if (dlm->dlm_locking_proto.pv_major == 1 && 1165 dlm->dlm_locking_proto.pv_minor == 0) { 1166 mlog(ML_ERROR, "Node %d queried hb regions on domain %s " 1167 "but active dlm protocol is %d.%d\n", qr->qr_node, 1168 qr->qr_domain, dlm->dlm_locking_proto.pv_major, 1169 dlm->dlm_locking_proto.pv_minor); 1170 goto out_dlm_lock; 1171 } 1172 1173 status = dlm_match_regions(dlm, qr, local, sizeof(qr->qr_regions)); 1174 1175out_dlm_lock: 1176 spin_unlock(&dlm->spinlock); 1177 1178out_domain_lock: 1179 spin_unlock(&dlm_domain_lock); 1180 1181 kfree(local); 1182 1183 return status; 1184} 1185 1186static int dlm_match_nodes(struct dlm_ctxt *dlm, struct dlm_query_nodeinfo *qn) 1187{ 1188 struct o2nm_node *local; 1189 struct dlm_node_info *remote; 1190 int i, j; 1191 int status = 0; 1192 1193 for (j = 0; j < qn->qn_numnodes; ++j) 1194 mlog(0, "Node %3d, %pI4:%u\n", qn->qn_nodes[j].ni_nodenum, 1195 &(qn->qn_nodes[j].ni_ipv4_address), 1196 ntohs(qn->qn_nodes[j].ni_ipv4_port)); 1197 1198 for (i = 0; i < O2NM_MAX_NODES && !status; ++i) { 1199 local = o2nm_get_node_by_num(i); 1200 remote = NULL; 1201 for (j = 0; j < qn->qn_numnodes; ++j) { 1202 if (qn->qn_nodes[j].ni_nodenum == i) { 1203 remote = &(qn->qn_nodes[j]); 1204 break; 1205 } 1206 } 1207 1208 if (!local && !remote) 1209 continue; 1210 1211 if ((local && !remote) || (!local && remote)) 1212 status = -EINVAL; 1213 1214 if (!status && 1215 ((remote->ni_nodenum != local->nd_num) || 1216 (remote->ni_ipv4_port != local->nd_ipv4_port) || 1217 (remote->ni_ipv4_address != local->nd_ipv4_address))) 1218 status = -EINVAL; 1219 1220 if (status) { 1221 if (remote && !local) 1222 mlog(ML_ERROR, "Domain %s: Node %d (%pI4:%u) " 1223 "registered in joining node %d but not in " 1224 "local node %d\n", qn->qn_domain, 1225 remote->ni_nodenum, 1226 &(remote->ni_ipv4_address), 1227 ntohs(remote->ni_ipv4_port), 1228 qn->qn_nodenum, dlm->node_num); 1229 if (local && !remote) 1230 mlog(ML_ERROR, "Domain %s: Node %d (%pI4:%u) " 1231 "registered in local node %d but not in " 1232 "joining node %d\n", qn->qn_domain, 1233 local->nd_num, &(local->nd_ipv4_address), 1234 ntohs(local->nd_ipv4_port), 1235 dlm->node_num, qn->qn_nodenum); 1236 BUG_ON((!local && !remote)); 1237 } 1238 1239 if (local) 1240 o2nm_node_put(local); 1241 } 1242 1243 return status; 1244} 1245 1246static int dlm_send_nodeinfo(struct dlm_ctxt *dlm, unsigned long *node_map) 1247{ 1248 struct dlm_query_nodeinfo *qn = NULL; 1249 struct o2nm_node *node; 1250 int ret = 0, status, count, i; 1251 1252 if (find_next_bit(node_map, O2NM_MAX_NODES, 0) >= O2NM_MAX_NODES) 1253 goto bail; 1254 1255 qn = kzalloc(sizeof(struct dlm_query_nodeinfo), GFP_KERNEL); 1256 if (!qn) { 1257 ret = -ENOMEM; 1258 mlog_errno(ret); 1259 goto bail; 1260 } 1261 1262 for (i = 0, count = 0; i < O2NM_MAX_NODES; ++i) { 1263 node = o2nm_get_node_by_num(i); 1264 if (!node) 1265 continue; 1266 qn->qn_nodes[count].ni_nodenum = node->nd_num; 1267 qn->qn_nodes[count].ni_ipv4_port = node->nd_ipv4_port; 1268 qn->qn_nodes[count].ni_ipv4_address = node->nd_ipv4_address; 1269 mlog(0, "Node %3d, %pI4:%u\n", node->nd_num, 1270 &(node->nd_ipv4_address), ntohs(node->nd_ipv4_port)); 1271 ++count; 1272 o2nm_node_put(node); 1273 } 1274 1275 qn->qn_nodenum = dlm->node_num; 1276 qn->qn_numnodes = count; 1277 qn->qn_namelen = strlen(dlm->name); 1278 memcpy(qn->qn_domain, dlm->name, qn->qn_namelen); 1279 1280 i = -1; 1281 while ((i = find_next_bit(node_map, O2NM_MAX_NODES, 1282 i + 1)) < O2NM_MAX_NODES) { 1283 if (i == dlm->node_num) 1284 continue; 1285 1286 mlog(0, "Sending nodeinfo to node %d\n", i); 1287 1288 ret = o2net_send_message(DLM_QUERY_NODEINFO, DLM_MOD_KEY, 1289 qn, sizeof(struct dlm_query_nodeinfo), 1290 i, &status); 1291 if (ret >= 0) 1292 ret = status; 1293 if (ret) { 1294 mlog(ML_ERROR, "node mismatch %d, node %d\n", ret, i); 1295 break; 1296 } 1297 } 1298 1299bail: 1300 kfree(qn); 1301 return ret; 1302} 1303 1304static int dlm_query_nodeinfo_handler(struct o2net_msg *msg, u32 len, 1305 void *data, void **ret_data) 1306{ 1307 struct dlm_query_nodeinfo *qn; 1308 struct dlm_ctxt *dlm = NULL; 1309 int locked = 0, status = -EINVAL; 1310 1311 qn = (struct dlm_query_nodeinfo *) msg->buf; 1312 1313 mlog(0, "Node %u queries nodes on domain %s\n", qn->qn_nodenum, 1314 qn->qn_domain); 1315 1316 spin_lock(&dlm_domain_lock); 1317 dlm = __dlm_lookup_domain_full(qn->qn_domain, qn->qn_namelen); 1318 if (!dlm) { 1319 mlog(ML_ERROR, "Node %d queried nodes on domain %s before " 1320 "join domain\n", qn->qn_nodenum, qn->qn_domain); 1321 goto bail; 1322 } 1323 1324 spin_lock(&dlm->spinlock); 1325 locked = 1; 1326 if (dlm->joining_node != qn->qn_nodenum) { 1327 mlog(ML_ERROR, "Node %d queried nodes on domain %s but " 1328 "joining node is %d\n", qn->qn_nodenum, qn->qn_domain, 1329 dlm->joining_node); 1330 goto bail; 1331 } 1332 1333 /* Support for node query was added in 1.1 */ 1334 if (dlm->dlm_locking_proto.pv_major == 1 && 1335 dlm->dlm_locking_proto.pv_minor == 0) { 1336 mlog(ML_ERROR, "Node %d queried nodes on domain %s " 1337 "but active dlm protocol is %d.%d\n", qn->qn_nodenum, 1338 qn->qn_domain, dlm->dlm_locking_proto.pv_major, 1339 dlm->dlm_locking_proto.pv_minor); 1340 goto bail; 1341 } 1342 1343 status = dlm_match_nodes(dlm, qn); 1344 1345bail: 1346 if (locked) 1347 spin_unlock(&dlm->spinlock); 1348 spin_unlock(&dlm_domain_lock); 1349 1350 return status; 1351} 1352 1353static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data, 1354 void **ret_data) 1355{ 1356 struct dlm_cancel_join *cancel; 1357 struct dlm_ctxt *dlm = NULL; 1358 1359 cancel = (struct dlm_cancel_join *) msg->buf; 1360 1361 mlog(0, "node %u cancels join on domain %s\n", cancel->node_idx, 1362 cancel->domain); 1363 1364 spin_lock(&dlm_domain_lock); 1365 dlm = __dlm_lookup_domain_full(cancel->domain, cancel->name_len); 1366 1367 if (dlm) { 1368 spin_lock(&dlm->spinlock); 1369 1370 /* Yikes, this guy wants to cancel his join. No 1371 * problem, we simply cleanup our join state. */ 1372 BUG_ON(dlm->joining_node != cancel->node_idx); 1373 __dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN); 1374 1375 spin_unlock(&dlm->spinlock); 1376 } 1377 spin_unlock(&dlm_domain_lock); 1378 1379 return 0; 1380} 1381 1382static int dlm_send_one_join_cancel(struct dlm_ctxt *dlm, 1383 unsigned int node) 1384{ 1385 int status; 1386 struct dlm_cancel_join cancel_msg; 1387 1388 memset(&cancel_msg, 0, sizeof(cancel_msg)); 1389 cancel_msg.node_idx = dlm->node_num; 1390 cancel_msg.name_len = strlen(dlm->name); 1391 memcpy(cancel_msg.domain, dlm->name, cancel_msg.name_len); 1392 1393 status = o2net_send_message(DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY, 1394 &cancel_msg, sizeof(cancel_msg), node, 1395 NULL); 1396 if (status < 0) { 1397 mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to " 1398 "node %u\n", status, DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY, 1399 node); 1400 goto bail; 1401 } 1402 1403bail: 1404 return status; 1405} 1406 1407/* map_size should be in bytes. */ 1408static int dlm_send_join_cancels(struct dlm_ctxt *dlm, 1409 unsigned long *node_map, 1410 unsigned int map_size) 1411{ 1412 int status, tmpstat; 1413 unsigned int node; 1414 1415 if (map_size != (BITS_TO_LONGS(O2NM_MAX_NODES) * 1416 sizeof(unsigned long))) { 1417 mlog(ML_ERROR, 1418 "map_size %u != BITS_TO_LONGS(O2NM_MAX_NODES) %u\n", 1419 map_size, (unsigned)BITS_TO_LONGS(O2NM_MAX_NODES)); 1420 return -EINVAL; 1421 } 1422 1423 status = 0; 1424 node = -1; 1425 while ((node = find_next_bit(node_map, O2NM_MAX_NODES, 1426 node + 1)) < O2NM_MAX_NODES) { 1427 if (node == dlm->node_num) 1428 continue; 1429 1430 tmpstat = dlm_send_one_join_cancel(dlm, node); 1431 if (tmpstat) { 1432 mlog(ML_ERROR, "Error return %d cancelling join on " 1433 "node %d\n", tmpstat, node); 1434 if (!status) 1435 status = tmpstat; 1436 } 1437 } 1438 1439 if (status) 1440 mlog_errno(status); 1441 return status; 1442} 1443 1444static int dlm_request_join(struct dlm_ctxt *dlm, 1445 int node, 1446 enum dlm_query_join_response_code *response) 1447{ 1448 int status; 1449 struct dlm_query_join_request join_msg; 1450 struct dlm_query_join_packet packet; 1451 u32 join_resp; 1452 1453 mlog(0, "querying node %d\n", node); 1454 1455 memset(&join_msg, 0, sizeof(join_msg)); 1456 join_msg.node_idx = dlm->node_num; 1457 join_msg.name_len = strlen(dlm->name); 1458 memcpy(join_msg.domain, dlm->name, join_msg.name_len); 1459 join_msg.dlm_proto = dlm->dlm_locking_proto; 1460 join_msg.fs_proto = dlm->fs_locking_proto; 1461 1462 /* copy live node map to join message */ 1463 byte_copymap(join_msg.node_map, dlm->live_nodes_map, O2NM_MAX_NODES); 1464 1465 status = o2net_send_message(DLM_QUERY_JOIN_MSG, DLM_MOD_KEY, &join_msg, 1466 sizeof(join_msg), node, &join_resp); 1467 if (status < 0 && status != -ENOPROTOOPT) { 1468 mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to " 1469 "node %u\n", status, DLM_QUERY_JOIN_MSG, DLM_MOD_KEY, 1470 node); 1471 goto bail; 1472 } 1473 dlm_query_join_wire_to_packet(join_resp, &packet); 1474 1475 /* -ENOPROTOOPT from the net code means the other side isn't 1476 listening for our message type -- that's fine, it means 1477 his dlm isn't up, so we can consider him a 'yes' but not 1478 joined into the domain. */ 1479 if (status == -ENOPROTOOPT) { 1480 status = 0; 1481 *response = JOIN_OK_NO_MAP; 1482 } else if (packet.code == JOIN_DISALLOW || 1483 packet.code == JOIN_OK_NO_MAP) { 1484 *response = packet.code; 1485 } else if (packet.code == JOIN_PROTOCOL_MISMATCH) { 1486 mlog(ML_NOTICE, 1487 "This node requested DLM locking protocol %u.%u and " 1488 "filesystem locking protocol %u.%u. At least one of " 1489 "the protocol versions on node %d is not compatible, " 1490 "disconnecting\n", 1491 dlm->dlm_locking_proto.pv_major, 1492 dlm->dlm_locking_proto.pv_minor, 1493 dlm->fs_locking_proto.pv_major, 1494 dlm->fs_locking_proto.pv_minor, 1495 node); 1496 status = -EPROTO; 1497 *response = packet.code; 1498 } else if (packet.code == JOIN_OK) { 1499 *response = packet.code; 1500 /* Use the same locking protocol as the remote node */ 1501 dlm->dlm_locking_proto.pv_minor = packet.dlm_minor; 1502 dlm->fs_locking_proto.pv_minor = packet.fs_minor; 1503 mlog(0, 1504 "Node %d responds JOIN_OK with DLM locking protocol " 1505 "%u.%u and fs locking protocol %u.%u\n", 1506 node, 1507 dlm->dlm_locking_proto.pv_major, 1508 dlm->dlm_locking_proto.pv_minor, 1509 dlm->fs_locking_proto.pv_major, 1510 dlm->fs_locking_proto.pv_minor); 1511 } else { 1512 status = -EINVAL; 1513 mlog(ML_ERROR, "invalid response %d from node %u\n", 1514 packet.code, node); 1515 } 1516 1517 mlog(0, "status %d, node %d response is %d\n", status, node, 1518 *response); 1519 1520bail: 1521 return status; 1522} 1523 1524static int dlm_send_one_join_assert(struct dlm_ctxt *dlm, 1525 unsigned int node) 1526{ 1527 int status; 1528 int ret; 1529 struct dlm_assert_joined assert_msg; 1530 1531 mlog(0, "Sending join assert to node %u\n", node); 1532 1533 memset(&assert_msg, 0, sizeof(assert_msg)); 1534 assert_msg.node_idx = dlm->node_num; 1535 assert_msg.name_len = strlen(dlm->name); 1536 memcpy(assert_msg.domain, dlm->name, assert_msg.name_len); 1537 1538 status = o2net_send_message(DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY, 1539 &assert_msg, sizeof(assert_msg), node, 1540 &ret); 1541 if (status < 0) 1542 mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to " 1543 "node %u\n", status, DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY, 1544 node); 1545 else 1546 status = ret; 1547 1548 return status; 1549} 1550 1551static void dlm_send_join_asserts(struct dlm_ctxt *dlm, 1552 unsigned long *node_map) 1553{ 1554 int status, node, live; 1555 1556 status = 0; 1557 node = -1; 1558 while ((node = find_next_bit(node_map, O2NM_MAX_NODES, 1559 node + 1)) < O2NM_MAX_NODES) { 1560 if (node == dlm->node_num) 1561 continue; 1562 1563 do { 1564 /* It is very important that this message be 1565 * received so we spin until either the node 1566 * has died or it gets the message. */ 1567 status = dlm_send_one_join_assert(dlm, node); 1568 1569 spin_lock(&dlm->spinlock); 1570 live = test_bit(node, dlm->live_nodes_map); 1571 spin_unlock(&dlm->spinlock); 1572 1573 if (status) { 1574 mlog(ML_ERROR, "Error return %d asserting " 1575 "join on node %d\n", status, node); 1576 1577 /* give us some time between errors... */ 1578 if (live) 1579 msleep(DLM_DOMAIN_BACKOFF_MS); 1580 } 1581 } while (status && live); 1582 } 1583} 1584 1585struct domain_join_ctxt { 1586 unsigned long live_map[BITS_TO_LONGS(O2NM_MAX_NODES)]; 1587 unsigned long yes_resp_map[BITS_TO_LONGS(O2NM_MAX_NODES)]; 1588}; 1589 1590static int dlm_should_restart_join(struct dlm_ctxt *dlm, 1591 struct domain_join_ctxt *ctxt, 1592 enum dlm_query_join_response_code response) 1593{ 1594 int ret; 1595 1596 if (response == JOIN_DISALLOW) { 1597 mlog(0, "Latest response of disallow -- should restart\n"); 1598 return 1; 1599 } 1600 1601 spin_lock(&dlm->spinlock); 1602 /* For now, we restart the process if the node maps have 1603 * changed at all */ 1604 ret = memcmp(ctxt->live_map, dlm->live_nodes_map, 1605 sizeof(dlm->live_nodes_map)); 1606 spin_unlock(&dlm->spinlock); 1607 1608 if (ret) 1609 mlog(0, "Node maps changed -- should restart\n"); 1610 1611 return ret; 1612} 1613 1614static int dlm_try_to_join_domain(struct dlm_ctxt *dlm) 1615{ 1616 int status = 0, tmpstat, node; 1617 struct domain_join_ctxt *ctxt; 1618 enum dlm_query_join_response_code response = JOIN_DISALLOW; 1619 1620 mlog(0, "%p", dlm); 1621 1622 ctxt = kzalloc(sizeof(*ctxt), GFP_KERNEL); 1623 if (!ctxt) { 1624 status = -ENOMEM; 1625 mlog_errno(status); 1626 goto bail; 1627 } 1628 1629 /* group sem locking should work for us here -- we're already 1630 * registered for heartbeat events so filling this should be 1631 * atomic wrt getting those handlers called. */ 1632 o2hb_fill_node_map(dlm->live_nodes_map, sizeof(dlm->live_nodes_map)); 1633 1634 spin_lock(&dlm->spinlock); 1635 memcpy(ctxt->live_map, dlm->live_nodes_map, sizeof(ctxt->live_map)); 1636 1637 __dlm_set_joining_node(dlm, dlm->node_num); 1638 1639 spin_unlock(&dlm->spinlock); 1640 1641 node = -1; 1642 while ((node = find_next_bit(ctxt->live_map, O2NM_MAX_NODES, 1643 node + 1)) < O2NM_MAX_NODES) { 1644 if (node == dlm->node_num) 1645 continue; 1646 1647 status = dlm_request_join(dlm, node, &response); 1648 if (status < 0) { 1649 mlog_errno(status); 1650 goto bail; 1651 } 1652 1653 /* Ok, either we got a response or the node doesn't have a 1654 * dlm up. */ 1655 if (response == JOIN_OK) 1656 set_bit(node, ctxt->yes_resp_map); 1657 1658 if (dlm_should_restart_join(dlm, ctxt, response)) { 1659 status = -EAGAIN; 1660 goto bail; 1661 } 1662 } 1663 1664 mlog(0, "Yay, done querying nodes!\n"); 1665 1666 /* Yay, everyone agree's we can join the domain. My domain is 1667 * comprised of all nodes who were put in the 1668 * yes_resp_map. Copy that into our domain map and send a join 1669 * assert message to clean up everyone elses state. */ 1670 spin_lock(&dlm->spinlock); 1671 memcpy(dlm->domain_map, ctxt->yes_resp_map, 1672 sizeof(ctxt->yes_resp_map)); 1673 set_bit(dlm->node_num, dlm->domain_map); 1674 spin_unlock(&dlm->spinlock); 1675 1676 /* Support for global heartbeat and node info was added in 1.1 */ 1677 if (dlm->dlm_locking_proto.pv_major > 1 || 1678 dlm->dlm_locking_proto.pv_minor > 0) { 1679 status = dlm_send_nodeinfo(dlm, ctxt->yes_resp_map); 1680 if (status) { 1681 mlog_errno(status); 1682 goto bail; 1683 } 1684 status = dlm_send_regions(dlm, ctxt->yes_resp_map); 1685 if (status) { 1686 mlog_errno(status); 1687 goto bail; 1688 } 1689 } 1690 1691 dlm_send_join_asserts(dlm, ctxt->yes_resp_map); 1692 1693 /* Joined state *must* be set before the joining node 1694 * information, otherwise the query_join handler may read no 1695 * current joiner but a state of NEW and tell joining nodes 1696 * we're not in the domain. */ 1697 spin_lock(&dlm_domain_lock); 1698 dlm->dlm_state = DLM_CTXT_JOINED; 1699 dlm->num_joins++; 1700 spin_unlock(&dlm_domain_lock); 1701 1702bail: 1703 spin_lock(&dlm->spinlock); 1704 __dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN); 1705 if (!status) { 1706 printk(KERN_NOTICE "o2dlm: Joining domain %s ", dlm->name); 1707 __dlm_print_nodes(dlm); 1708 } 1709 spin_unlock(&dlm->spinlock); 1710 1711 if (ctxt) { 1712 /* Do we need to send a cancel message to any nodes? */ 1713 if (status < 0) { 1714 tmpstat = dlm_send_join_cancels(dlm, 1715 ctxt->yes_resp_map, 1716 sizeof(ctxt->yes_resp_map)); 1717 if (tmpstat < 0) 1718 mlog_errno(tmpstat); 1719 } 1720 kfree(ctxt); 1721 } 1722 1723 mlog(0, "returning %d\n", status); 1724 return status; 1725} 1726 1727static void dlm_unregister_domain_handlers(struct dlm_ctxt *dlm) 1728{ 1729 o2hb_unregister_callback(dlm->name, &dlm->dlm_hb_up); 1730 o2hb_unregister_callback(dlm->name, &dlm->dlm_hb_down); 1731 o2net_unregister_handler_list(&dlm->dlm_domain_handlers); 1732} 1733 1734static int dlm_register_domain_handlers(struct dlm_ctxt *dlm) 1735{ 1736 int status; 1737 1738 mlog(0, "registering handlers.\n"); 1739 1740 o2hb_setup_callback(&dlm->dlm_hb_down, O2HB_NODE_DOWN_CB, 1741 dlm_hb_node_down_cb, dlm, DLM_HB_NODE_DOWN_PRI); 1742 status = o2hb_register_callback(dlm->name, &dlm->dlm_hb_down); 1743 if (status) 1744 goto bail; 1745 1746 o2hb_setup_callback(&dlm->dlm_hb_up, O2HB_NODE_UP_CB, 1747 dlm_hb_node_up_cb, dlm, DLM_HB_NODE_UP_PRI); 1748 status = o2hb_register_callback(dlm->name, &dlm->dlm_hb_up); 1749 if (status) 1750 goto bail; 1751 1752 status = o2net_register_handler(DLM_MASTER_REQUEST_MSG, dlm->key, 1753 sizeof(struct dlm_master_request), 1754 dlm_master_request_handler, 1755 dlm, NULL, &dlm->dlm_domain_handlers); 1756 if (status) 1757 goto bail; 1758 1759 status = o2net_register_handler(DLM_ASSERT_MASTER_MSG, dlm->key, 1760 sizeof(struct dlm_assert_master), 1761 dlm_assert_master_handler, 1762 dlm, dlm_assert_master_post_handler, 1763 &dlm->dlm_domain_handlers); 1764 if (status) 1765 goto bail; 1766 1767 status = o2net_register_handler(DLM_CREATE_LOCK_MSG, dlm->key, 1768 sizeof(struct dlm_create_lock), 1769 dlm_create_lock_handler, 1770 dlm, NULL, &dlm->dlm_domain_handlers); 1771 if (status) 1772 goto bail; 1773 1774 status = o2net_register_handler(DLM_CONVERT_LOCK_MSG, dlm->key, 1775 DLM_CONVERT_LOCK_MAX_LEN, 1776 dlm_convert_lock_handler, 1777 dlm, NULL, &dlm->dlm_domain_handlers); 1778 if (status) 1779 goto bail; 1780 1781 status = o2net_register_handler(DLM_UNLOCK_LOCK_MSG, dlm->key, 1782 DLM_UNLOCK_LOCK_MAX_LEN, 1783 dlm_unlock_lock_handler, 1784 dlm, NULL, &dlm->dlm_domain_handlers); 1785 if (status) 1786 goto bail; 1787 1788 status = o2net_register_handler(DLM_PROXY_AST_MSG, dlm->key, 1789 DLM_PROXY_AST_MAX_LEN, 1790 dlm_proxy_ast_handler, 1791 dlm, NULL, &dlm->dlm_domain_handlers); 1792 if (status) 1793 goto bail; 1794 1795 status = o2net_register_handler(DLM_EXIT_DOMAIN_MSG, dlm->key, 1796 sizeof(struct dlm_exit_domain), 1797 dlm_exit_domain_handler, 1798 dlm, NULL, &dlm->dlm_domain_handlers); 1799 if (status) 1800 goto bail; 1801 1802 status = o2net_register_handler(DLM_DEREF_LOCKRES_MSG, dlm->key, 1803 sizeof(struct dlm_deref_lockres), 1804 dlm_deref_lockres_handler, 1805 dlm, NULL, &dlm->dlm_domain_handlers); 1806 if (status) 1807 goto bail; 1808 1809 status = o2net_register_handler(DLM_MIGRATE_REQUEST_MSG, dlm->key, 1810 sizeof(struct dlm_migrate_request), 1811 dlm_migrate_request_handler, 1812 dlm, NULL, &dlm->dlm_domain_handlers); 1813 if (status) 1814 goto bail; 1815 1816 status = o2net_register_handler(DLM_MIG_LOCKRES_MSG, dlm->key, 1817 DLM_MIG_LOCKRES_MAX_LEN, 1818 dlm_mig_lockres_handler, 1819 dlm, NULL, &dlm->dlm_domain_handlers); 1820 if (status) 1821 goto bail; 1822 1823 status = o2net_register_handler(DLM_MASTER_REQUERY_MSG, dlm->key, 1824 sizeof(struct dlm_master_requery), 1825 dlm_master_requery_handler, 1826 dlm, NULL, &dlm->dlm_domain_handlers); 1827 if (status) 1828 goto bail; 1829 1830 status = o2net_register_handler(DLM_LOCK_REQUEST_MSG, dlm->key, 1831 sizeof(struct dlm_lock_request), 1832 dlm_request_all_locks_handler, 1833 dlm, NULL, &dlm->dlm_domain_handlers); 1834 if (status) 1835 goto bail; 1836 1837 status = o2net_register_handler(DLM_RECO_DATA_DONE_MSG, dlm->key, 1838 sizeof(struct dlm_reco_data_done), 1839 dlm_reco_data_done_handler, 1840 dlm, NULL, &dlm->dlm_domain_handlers); 1841 if (status) 1842 goto bail; 1843 1844 status = o2net_register_handler(DLM_BEGIN_RECO_MSG, dlm->key, 1845 sizeof(struct dlm_begin_reco), 1846 dlm_begin_reco_handler, 1847 dlm, NULL, &dlm->dlm_domain_handlers); 1848 if (status) 1849 goto bail; 1850 1851 status = o2net_register_handler(DLM_FINALIZE_RECO_MSG, dlm->key, 1852 sizeof(struct dlm_finalize_reco), 1853 dlm_finalize_reco_handler, 1854 dlm, NULL, &dlm->dlm_domain_handlers); 1855 if (status) 1856 goto bail; 1857 1858 status = o2net_register_handler(DLM_BEGIN_EXIT_DOMAIN_MSG, dlm->key, 1859 sizeof(struct dlm_exit_domain), 1860 dlm_begin_exit_domain_handler, 1861 dlm, NULL, &dlm->dlm_domain_handlers); 1862 if (status) 1863 goto bail; 1864 1865bail: 1866 if (status) 1867 dlm_unregister_domain_handlers(dlm); 1868 1869 return status; 1870} 1871 1872static int dlm_join_domain(struct dlm_ctxt *dlm) 1873{ 1874 int status; 1875 unsigned int backoff; 1876 unsigned int total_backoff = 0; 1877 1878 BUG_ON(!dlm); 1879 1880 mlog(0, "Join domain %s\n", dlm->name); 1881 1882 status = dlm_register_domain_handlers(dlm); 1883 if (status) { 1884 mlog_errno(status); 1885 goto bail; 1886 } 1887 1888 status = dlm_launch_thread(dlm); 1889 if (status < 0) { 1890 mlog_errno(status); 1891 goto bail; 1892 } 1893 1894 status = dlm_launch_recovery_thread(dlm); 1895 if (status < 0) { 1896 mlog_errno(status); 1897 goto bail; 1898 } 1899 1900 status = dlm_debug_init(dlm); 1901 if (status < 0) { 1902 mlog_errno(status); 1903 goto bail; 1904 } 1905 1906 dlm->dlm_worker = create_singlethread_workqueue("dlm_wq"); 1907 if (!dlm->dlm_worker) { 1908 status = -ENOMEM; 1909 mlog_errno(status); 1910 goto bail; 1911 } 1912 1913 do { 1914 status = dlm_try_to_join_domain(dlm); 1915 1916 /* If we're racing another node to the join, then we 1917 * need to back off temporarily and let them 1918 * complete. */ 1919#define DLM_JOIN_TIMEOUT_MSECS 90000 1920 if (status == -EAGAIN) { 1921 if (signal_pending(current)) { 1922 status = -ERESTARTSYS; 1923 goto bail; 1924 } 1925 1926 if (total_backoff > DLM_JOIN_TIMEOUT_MSECS) { 1927 status = -ERESTARTSYS; 1928 mlog(ML_NOTICE, "Timed out joining dlm domain " 1929 "%s after %u msecs\n", dlm->name, 1930 total_backoff); 1931 goto bail; 1932 } 1933 1934 /* 1935 * <chip> After you! 1936 * <dale> No, after you! 1937 * <chip> I insist! 1938 * <dale> But you first! 1939 * ... 1940 */ 1941 backoff = (unsigned int)(jiffies & 0x3); 1942 backoff *= DLM_DOMAIN_BACKOFF_MS; 1943 total_backoff += backoff; 1944 mlog(0, "backoff %d\n", backoff); 1945 msleep(backoff); 1946 } 1947 } while (status == -EAGAIN); 1948 1949 if (status < 0) { 1950 mlog_errno(status); 1951 goto bail; 1952 } 1953 1954 status = 0; 1955bail: 1956 wake_up(&dlm_domain_events); 1957 1958 if (status) { 1959 dlm_unregister_domain_handlers(dlm); 1960 dlm_debug_shutdown(dlm); 1961 dlm_complete_thread(dlm); 1962 dlm_complete_recovery_thread(dlm); 1963 dlm_destroy_dlm_worker(dlm); 1964 } 1965 1966 return status; 1967} 1968 1969static struct dlm_ctxt *dlm_alloc_ctxt(const char *domain, 1970 u32 key) 1971{ 1972 int i; 1973 int ret; 1974 struct dlm_ctxt *dlm = NULL; 1975 1976 dlm = kzalloc(sizeof(*dlm), GFP_KERNEL); 1977 if (!dlm) { 1978 ret = -ENOMEM; 1979 mlog_errno(ret); 1980 goto leave; 1981 } 1982 1983 dlm->name = kstrdup(domain, GFP_KERNEL); 1984 if (dlm->name == NULL) { 1985 ret = -ENOMEM; 1986 mlog_errno(ret); 1987 goto leave; 1988 } 1989 1990 dlm->lockres_hash = (struct hlist_head **)dlm_alloc_pagevec(DLM_HASH_PAGES); 1991 if (!dlm->lockres_hash) { 1992 ret = -ENOMEM; 1993 mlog_errno(ret); 1994 goto leave; 1995 } 1996 1997 for (i = 0; i < DLM_HASH_BUCKETS; i++) 1998 INIT_HLIST_HEAD(dlm_lockres_hash(dlm, i)); 1999 2000 dlm->master_hash = (struct hlist_head **) 2001 dlm_alloc_pagevec(DLM_HASH_PAGES); 2002 if (!dlm->master_hash) { 2003 ret = -ENOMEM; 2004 mlog_errno(ret); 2005 goto leave; 2006 } 2007 2008 for (i = 0; i < DLM_HASH_BUCKETS; i++) 2009 INIT_HLIST_HEAD(dlm_master_hash(dlm, i)); 2010 2011 dlm->key = key; 2012 dlm->node_num = o2nm_this_node(); 2013 2014 ret = dlm_create_debugfs_subroot(dlm); 2015 if (ret < 0) 2016 goto leave; 2017 2018 spin_lock_init(&dlm->spinlock); 2019 spin_lock_init(&dlm->master_lock); 2020 spin_lock_init(&dlm->ast_lock); 2021 spin_lock_init(&dlm->track_lock); 2022 INIT_LIST_HEAD(&dlm->list); 2023 INIT_LIST_HEAD(&dlm->dirty_list); 2024 INIT_LIST_HEAD(&dlm->reco.resources); 2025 INIT_LIST_HEAD(&dlm->reco.node_data); 2026 INIT_LIST_HEAD(&dlm->purge_list); 2027 INIT_LIST_HEAD(&dlm->dlm_domain_handlers); 2028 INIT_LIST_HEAD(&dlm->tracking_list); 2029 dlm->reco.state = 0; 2030 2031 INIT_LIST_HEAD(&dlm->pending_asts); 2032 INIT_LIST_HEAD(&dlm->pending_basts); 2033 2034 mlog(0, "dlm->recovery_map=%p, &(dlm->recovery_map[0])=%p\n", 2035 dlm->recovery_map, &(dlm->recovery_map[0])); 2036 2037 memset(dlm->recovery_map, 0, sizeof(dlm->recovery_map)); 2038 memset(dlm->live_nodes_map, 0, sizeof(dlm->live_nodes_map)); 2039 memset(dlm->domain_map, 0, sizeof(dlm->domain_map)); 2040 2041 dlm->dlm_thread_task = NULL; 2042 dlm->dlm_reco_thread_task = NULL; 2043 dlm->dlm_worker = NULL; 2044 init_waitqueue_head(&dlm->dlm_thread_wq); 2045 init_waitqueue_head(&dlm->dlm_reco_thread_wq); 2046 init_waitqueue_head(&dlm->reco.event); 2047 init_waitqueue_head(&dlm->ast_wq); 2048 init_waitqueue_head(&dlm->migration_wq); 2049 INIT_LIST_HEAD(&dlm->mle_hb_events); 2050 2051 dlm->joining_node = DLM_LOCK_RES_OWNER_UNKNOWN; 2052 init_waitqueue_head(&dlm->dlm_join_events); 2053 2054 dlm->reco.new_master = O2NM_INVALID_NODE_NUM; 2055 dlm->reco.dead_node = O2NM_INVALID_NODE_NUM; 2056 2057 atomic_set(&dlm->res_tot_count, 0); 2058 atomic_set(&dlm->res_cur_count, 0); 2059 for (i = 0; i < DLM_MLE_NUM_TYPES; ++i) { 2060 atomic_set(&dlm->mle_tot_count[i], 0); 2061 atomic_set(&dlm->mle_cur_count[i], 0); 2062 } 2063 2064 spin_lock_init(&dlm->work_lock); 2065 INIT_LIST_HEAD(&dlm->work_list); 2066 INIT_WORK(&dlm->dispatched_work, dlm_dispatch_work); 2067 2068 kref_init(&dlm->dlm_refs); 2069 dlm->dlm_state = DLM_CTXT_NEW; 2070 2071 INIT_LIST_HEAD(&dlm->dlm_eviction_callbacks); 2072 2073 mlog(0, "context init: refcount %u\n", 2074 atomic_read(&dlm->dlm_refs.refcount)); 2075 2076leave: 2077 if (ret < 0 && dlm) { 2078 if (dlm->master_hash) 2079 dlm_free_pagevec((void **)dlm->master_hash, 2080 DLM_HASH_PAGES); 2081 2082 if (dlm->lockres_hash) 2083 dlm_free_pagevec((void **)dlm->lockres_hash, 2084 DLM_HASH_PAGES); 2085 2086 kfree(dlm->name); 2087 kfree(dlm); 2088 dlm = NULL; 2089 } 2090 return dlm; 2091} 2092 2093/* 2094 * Compare a requested locking protocol version against the current one. 2095 * 2096 * If the major numbers are different, they are incompatible. 2097 * If the current minor is greater than the request, they are incompatible. 2098 * If the current minor is less than or equal to the request, they are 2099 * compatible, and the requester should run at the current minor version. 2100 */ 2101static int dlm_protocol_compare(struct dlm_protocol_version *existing, 2102 struct dlm_protocol_version *request) 2103{ 2104 if (existing->pv_major != request->pv_major) 2105 return 1; 2106 2107 if (existing->pv_minor > request->pv_minor) 2108 return 1; 2109 2110 if (existing->pv_minor < request->pv_minor) 2111 request->pv_minor = existing->pv_minor; 2112 2113 return 0; 2114} 2115 2116/* 2117 * dlm_register_domain: one-time setup per "domain". 2118 * 2119 * The filesystem passes in the requested locking version via proto. 2120 * If registration was successful, proto will contain the negotiated 2121 * locking protocol. 2122 */ 2123struct dlm_ctxt * dlm_register_domain(const char *domain, 2124 u32 key, 2125 struct dlm_protocol_version *fs_proto) 2126{ 2127 int ret; 2128 struct dlm_ctxt *dlm = NULL; 2129 struct dlm_ctxt *new_ctxt = NULL; 2130 2131 if (strlen(domain) >= O2NM_MAX_NAME_LEN) { 2132 ret = -ENAMETOOLONG; 2133 mlog(ML_ERROR, "domain name length too long\n"); 2134 goto leave; 2135 } 2136 2137 mlog(0, "register called for domain \"%s\"\n", domain); 2138 2139retry: 2140 dlm = NULL; 2141 if (signal_pending(current)) { 2142 ret = -ERESTARTSYS; 2143 mlog_errno(ret); 2144 goto leave; 2145 } 2146 2147 spin_lock(&dlm_domain_lock); 2148 2149 dlm = __dlm_lookup_domain(domain); 2150 if (dlm) { 2151 if (dlm->dlm_state != DLM_CTXT_JOINED) { 2152 spin_unlock(&dlm_domain_lock); 2153 2154 mlog(0, "This ctxt is not joined yet!\n"); 2155 wait_event_interruptible(dlm_domain_events, 2156 dlm_wait_on_domain_helper( 2157 domain)); 2158 goto retry; 2159 } 2160 2161 if (dlm_protocol_compare(&dlm->fs_locking_proto, fs_proto)) { 2162 spin_unlock(&dlm_domain_lock); 2163 mlog(ML_ERROR, 2164 "Requested locking protocol version is not " 2165 "compatible with already registered domain " 2166 "\"%s\"\n", domain); 2167 ret = -EPROTO; 2168 goto leave; 2169 } 2170 2171 __dlm_get(dlm); 2172 dlm->num_joins++; 2173 2174 spin_unlock(&dlm_domain_lock); 2175 2176 ret = 0; 2177 goto leave; 2178 } 2179 2180 /* doesn't exist */ 2181 if (!new_ctxt) { 2182 spin_unlock(&dlm_domain_lock); 2183 2184 new_ctxt = dlm_alloc_ctxt(domain, key); 2185 if (new_ctxt) 2186 goto retry; 2187 2188 ret = -ENOMEM; 2189 mlog_errno(ret); 2190 goto leave; 2191 } 2192 2193 /* a little variable switch-a-roo here... */ 2194 dlm = new_ctxt; 2195 new_ctxt = NULL; 2196 2197 /* add the new domain */ 2198 list_add_tail(&dlm->list, &dlm_domains); 2199 spin_unlock(&dlm_domain_lock); 2200 2201 /* 2202 * Pass the locking protocol version into the join. If the join 2203 * succeeds, it will have the negotiated protocol set. 2204 */ 2205 dlm->dlm_locking_proto = dlm_protocol; 2206 dlm->fs_locking_proto = *fs_proto; 2207 2208 ret = dlm_join_domain(dlm); 2209 if (ret) { 2210 mlog_errno(ret); 2211 dlm_put(dlm); 2212 goto leave; 2213 } 2214 2215 /* Tell the caller what locking protocol we negotiated */ 2216 *fs_proto = dlm->fs_locking_proto; 2217 2218 ret = 0; 2219leave: 2220 if (new_ctxt) 2221 dlm_free_ctxt_mem(new_ctxt); 2222 2223 if (ret < 0) 2224 dlm = ERR_PTR(ret); 2225 2226 return dlm; 2227} 2228EXPORT_SYMBOL_GPL(dlm_register_domain); 2229 2230static LIST_HEAD(dlm_join_handlers); 2231 2232static void dlm_unregister_net_handlers(void) 2233{ 2234 o2net_unregister_handler_list(&dlm_join_handlers); 2235} 2236 2237static int dlm_register_net_handlers(void) 2238{ 2239 int status = 0; 2240 2241 status = o2net_register_handler(DLM_QUERY_JOIN_MSG, DLM_MOD_KEY, 2242 sizeof(struct dlm_query_join_request), 2243 dlm_query_join_handler, 2244 NULL, NULL, &dlm_join_handlers); 2245 if (status) 2246 goto bail; 2247 2248 status = o2net_register_handler(DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY, 2249 sizeof(struct dlm_assert_joined), 2250 dlm_assert_joined_handler, 2251 NULL, NULL, &dlm_join_handlers); 2252 if (status) 2253 goto bail; 2254 2255 status = o2net_register_handler(DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY, 2256 sizeof(struct dlm_cancel_join), 2257 dlm_cancel_join_handler, 2258 NULL, NULL, &dlm_join_handlers); 2259 if (status) 2260 goto bail; 2261 2262 status = o2net_register_handler(DLM_QUERY_REGION, DLM_MOD_KEY, 2263 sizeof(struct dlm_query_region), 2264 dlm_query_region_handler, 2265 NULL, NULL, &dlm_join_handlers); 2266 2267 if (status) 2268 goto bail; 2269 2270 status = o2net_register_handler(DLM_QUERY_NODEINFO, DLM_MOD_KEY, 2271 sizeof(struct dlm_query_nodeinfo), 2272 dlm_query_nodeinfo_handler, 2273 NULL, NULL, &dlm_join_handlers); 2274bail: 2275 if (status < 0) 2276 dlm_unregister_net_handlers(); 2277 2278 return status; 2279} 2280 2281/* Domain eviction callback handling. 2282 * 2283 * The file system requires notification of node death *before* the 2284 * dlm completes it's recovery work, otherwise it may be able to 2285 * acquire locks on resources requiring recovery. Since the dlm can 2286 * evict a node from it's domain *before* heartbeat fires, a similar 2287 * mechanism is required. */ 2288 2289/* Eviction is not expected to happen often, so a per-domain lock is 2290 * not necessary. Eviction callbacks are allowed to sleep for short 2291 * periods of time. */ 2292static DECLARE_RWSEM(dlm_callback_sem); 2293 2294void dlm_fire_domain_eviction_callbacks(struct dlm_ctxt *dlm, 2295 int node_num) 2296{ 2297 struct dlm_eviction_cb *cb; 2298 2299 down_read(&dlm_callback_sem); 2300 list_for_each_entry(cb, &dlm->dlm_eviction_callbacks, ec_item) { 2301 cb->ec_func(node_num, cb->ec_data); 2302 } 2303 up_read(&dlm_callback_sem); 2304} 2305 2306void dlm_setup_eviction_cb(struct dlm_eviction_cb *cb, 2307 dlm_eviction_func *f, 2308 void *data) 2309{ 2310 INIT_LIST_HEAD(&cb->ec_item); 2311 cb->ec_func = f; 2312 cb->ec_data = data; 2313} 2314EXPORT_SYMBOL_GPL(dlm_setup_eviction_cb); 2315 2316void dlm_register_eviction_cb(struct dlm_ctxt *dlm, 2317 struct dlm_eviction_cb *cb) 2318{ 2319 down_write(&dlm_callback_sem); 2320 list_add_tail(&cb->ec_item, &dlm->dlm_eviction_callbacks); 2321 up_write(&dlm_callback_sem); 2322} 2323EXPORT_SYMBOL_GPL(dlm_register_eviction_cb); 2324 2325void dlm_unregister_eviction_cb(struct dlm_eviction_cb *cb) 2326{ 2327 down_write(&dlm_callback_sem); 2328 list_del_init(&cb->ec_item); 2329 up_write(&dlm_callback_sem); 2330} 2331EXPORT_SYMBOL_GPL(dlm_unregister_eviction_cb); 2332 2333static int __init dlm_init(void) 2334{ 2335 int status; 2336 2337 status = dlm_init_mle_cache(); 2338 if (status) { 2339 mlog(ML_ERROR, "Could not create o2dlm_mle slabcache\n"); 2340 goto error; 2341 } 2342 2343 status = dlm_init_master_caches(); 2344 if (status) { 2345 mlog(ML_ERROR, "Could not create o2dlm_lockres and " 2346 "o2dlm_lockname slabcaches\n"); 2347 goto error; 2348 } 2349 2350 status = dlm_init_lock_cache(); 2351 if (status) { 2352 mlog(ML_ERROR, "Count not create o2dlm_lock slabcache\n"); 2353 goto error; 2354 } 2355 2356 status = dlm_register_net_handlers(); 2357 if (status) { 2358 mlog(ML_ERROR, "Unable to register network handlers\n"); 2359 goto error; 2360 } 2361 2362 status = dlm_create_debugfs_root(); 2363 if (status) 2364 goto error; 2365 2366 return 0; 2367error: 2368 dlm_unregister_net_handlers(); 2369 dlm_destroy_lock_cache(); 2370 dlm_destroy_master_caches(); 2371 dlm_destroy_mle_cache(); 2372 return -1; 2373} 2374 2375static void __exit dlm_exit (void) 2376{ 2377 dlm_destroy_debugfs_root(); 2378 dlm_unregister_net_handlers(); 2379 dlm_destroy_lock_cache(); 2380 dlm_destroy_master_caches(); 2381 dlm_destroy_mle_cache(); 2382} 2383 2384MODULE_AUTHOR("Oracle"); 2385MODULE_LICENSE("GPL"); 2386MODULE_DESCRIPTION("OCFS2 Distributed Lock Management"); 2387 2388module_init(dlm_init); 2389module_exit(dlm_exit); 2390