dlmdomain.c 48.6 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34
/* -*- mode: c; c-basic-offset: 8; -*-
 * vim: noexpandtab sw=8 ts=8 sts=0:
 *
 * dlmdomain.c
 *
 * defines domain join / leave apis
 *
 * Copyright (C) 2004 Oracle.  All rights reserved.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public
 * License along with this program; if not, write to the
 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 * Boston, MA 021110-1307, USA.
 *
 */

#include <linux/module.h>
#include <linux/types.h>
#include <linux/slab.h>
#include <linux/highmem.h>
#include <linux/init.h>
#include <linux/spinlock.h>
#include <linux/delay.h>
#include <linux/err.h>
S
Sunil Mushran 已提交
35
#include <linux/debugfs.h>
36 37 38 39 40 41 42 43

#include "cluster/heartbeat.h"
#include "cluster/nodemanager.h"
#include "cluster/tcp.h"

#include "dlmapi.h"
#include "dlmcommon.h"
#include "dlmdomain.h"
S
Sunil Mushran 已提交
44
#include "dlmdebug.h"
45 46 47 48 49 50

#include "dlmver.h"

#define MLOG_MASK_PREFIX (ML_DLM|ML_DLM_DOMAIN)
#include "cluster/masklog.h"

51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80
/*
 * ocfs2 node maps are array of long int, which limits to send them freely
 * across the wire due to endianness issues. To workaround this, we convert
 * long ints to byte arrays. Following 3 routines are helper functions to
 * set/test/copy bits within those array of bytes
 */
static inline void byte_set_bit(u8 nr, u8 map[])
{
	map[nr >> 3] |= (1UL << (nr & 7));
}

static inline int byte_test_bit(u8 nr, u8 map[])
{
	return ((1UL << (nr & 7)) & (map[nr >> 3])) != 0;
}

static inline void byte_copymap(u8 dmap[], unsigned long smap[],
			unsigned int sz)
{
	unsigned int nn;

	if (!sz)
		return;

	memset(dmap, 0, ((sz + 7) >> 3));
	for (nn = 0 ; nn < sz; nn++)
		if (test_bit(nn, smap))
			byte_set_bit(nn, dmap);
}

81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98
static void dlm_free_pagevec(void **vec, int pages)
{
	while (pages--)
		free_page((unsigned long)vec[pages]);
	kfree(vec);
}

static void **dlm_alloc_pagevec(int pages)
{
	void **vec = kmalloc(pages * sizeof(void *), GFP_KERNEL);
	int i;

	if (!vec)
		return NULL;

	for (i = 0; i < pages; i++)
		if (!(vec[i] = (void *)__get_free_page(GFP_KERNEL)))
			goto out_free;
99

100
	mlog(0, "Allocated DLM hash pagevec; %d pages (%lu expected), %lu buckets per page\n",
101 102
	     pages, (unsigned long)DLM_HASH_PAGES,
	     (unsigned long)DLM_BUCKETS_PER_PAGE);
103 104 105 106 107 108
	return vec;
out_free:
	dlm_free_pagevec(vec, i);
	return NULL;
}

109 110 111 112 113 114 115 116 117 118 119 120 121
/*
 *
 * spinlock lock ordering: if multiple locks are needed, obey this ordering:
 *    dlm_domain_lock
 *    struct dlm_ctxt->spinlock
 *    struct dlm_lock_resource->spinlock
 *    struct dlm_ctxt->master_lock
 *    struct dlm_ctxt->ast_lock
 *    dlm_master_list_entry->spinlock
 *    dlm_lock->spinlock
 *
 */

I
Ingo Molnar 已提交
122
DEFINE_SPINLOCK(dlm_domain_lock);
123 124 125
LIST_HEAD(dlm_domains);
static DECLARE_WAIT_QUEUE_HEAD(dlm_domain_events);

126 127 128 129 130 131 132 133 134 135 136
/*
 * The supported protocol version for DLM communication.  Running domains
 * will have a negotiated version with the same major number and a minor
 * number equal or smaller.  The dlm_ctxt->dlm_locking_proto field should
 * be used to determine what a running domain is actually using.
 */
static const struct dlm_protocol_version dlm_protocol = {
	.pv_major = 1,
	.pv_minor = 0,
};

137 138
#define DLM_DOMAIN_BACKOFF_MS 200

139 140 141 142 143 144 145 146
static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data,
				  void **ret_data);
static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data,
				     void **ret_data);
static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data,
				   void **ret_data);
static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data,
				   void **ret_data);
147 148
static int dlm_protocol_compare(struct dlm_protocol_version *existing,
				struct dlm_protocol_version *request);
149 150 151 152 153

static void dlm_unregister_domain_handlers(struct dlm_ctxt *dlm);

void __dlm_unhash_lockres(struct dlm_lock_resource *lockres)
{
154 155 156 157
	if (!hlist_unhashed(&lockres->hash_node)) {
		hlist_del_init(&lockres->hash_node);
		dlm_lockres_put(lockres);
	}
158 159 160 161 162
}

void __dlm_insert_lockres(struct dlm_ctxt *dlm,
		       struct dlm_lock_resource *res)
{
163
	struct hlist_head *bucket;
164 165 166 167 168
	struct qstr *q;

	assert_spin_locked(&dlm->spinlock);

	q = &res->lockname;
169
	bucket = dlm_lockres_hash(dlm, q->hash);
170 171 172 173

	/* get a reference for our hashtable */
	dlm_lockres_get(res);

174
	hlist_add_head(&res->hash_node, bucket);
175 176
}

177 178 179 180
struct dlm_lock_resource * __dlm_lookup_lockres_full(struct dlm_ctxt *dlm,
						     const char *name,
						     unsigned int len,
						     unsigned int hash)
181
{
182
	struct hlist_head *bucket;
183
	struct hlist_node *list;
184 185 186 187 188

	mlog_entry("%.*s\n", len, name);

	assert_spin_locked(&dlm->spinlock);

189 190
	bucket = dlm_lockres_hash(dlm, hash);

191 192 193 194 195 196 197 198 199 200 201
	hlist_for_each(list, bucket) {
		struct dlm_lock_resource *res = hlist_entry(list,
			struct dlm_lock_resource, hash_node);
		if (res->lockname.name[0] != name[0])
			continue;
		if (unlikely(res->lockname.len != len))
			continue;
		if (memcmp(res->lockname.name + 1, name + 1, len - 1))
			continue;
		dlm_lockres_get(res);
		return res;
202
	}
203
	return NULL;
204 205
}

206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236
/* intended to be called by functions which do not care about lock
 * resources which are being purged (most net _handler functions).
 * this will return NULL for any lock resource which is found but
 * currently in the process of dropping its mastery reference.
 * use __dlm_lookup_lockres_full when you need the lock resource
 * regardless (e.g. dlm_get_lock_resource) */
struct dlm_lock_resource * __dlm_lookup_lockres(struct dlm_ctxt *dlm,
						const char *name,
						unsigned int len,
						unsigned int hash)
{
	struct dlm_lock_resource *res = NULL;

	mlog_entry("%.*s\n", len, name);

	assert_spin_locked(&dlm->spinlock);

	res = __dlm_lookup_lockres_full(dlm, name, len, hash);
	if (res) {
		spin_lock(&res->spinlock);
		if (res->state & DLM_LOCK_RES_DROPPING_REF) {
			spin_unlock(&res->spinlock);
			dlm_lockres_put(res);
			return NULL;
		}
		spin_unlock(&res->spinlock);
	}

	return res;
}

237 238 239 240 241
struct dlm_lock_resource * dlm_lookup_lockres(struct dlm_ctxt *dlm,
				    const char *name,
				    unsigned int len)
{
	struct dlm_lock_resource *res;
242
	unsigned int hash = dlm_lockid_hash(name, len);
243 244

	spin_lock(&dlm->spinlock);
245
	res = __dlm_lookup_lockres(dlm, name, len, hash);
246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300
	spin_unlock(&dlm->spinlock);
	return res;
}

static struct dlm_ctxt * __dlm_lookup_domain_full(const char *domain, int len)
{
	struct dlm_ctxt *tmp = NULL;
	struct list_head *iter;

	assert_spin_locked(&dlm_domain_lock);

	/* tmp->name here is always NULL terminated,
	 * but domain may not be! */
	list_for_each(iter, &dlm_domains) {
		tmp = list_entry (iter, struct dlm_ctxt, list);
		if (strlen(tmp->name) == len &&
		    memcmp(tmp->name, domain, len)==0)
			break;
		tmp = NULL;
	}

	return tmp;
}

/* For null terminated domain strings ONLY */
static struct dlm_ctxt * __dlm_lookup_domain(const char *domain)
{
	assert_spin_locked(&dlm_domain_lock);

	return __dlm_lookup_domain_full(domain, strlen(domain));
}


/* returns true on one of two conditions:
 * 1) the domain does not exist
 * 2) the domain exists and it's state is "joined" */
static int dlm_wait_on_domain_helper(const char *domain)
{
	int ret = 0;
	struct dlm_ctxt *tmp = NULL;

	spin_lock(&dlm_domain_lock);

	tmp = __dlm_lookup_domain(domain);
	if (!tmp)
		ret = 1;
	else if (tmp->dlm_state == DLM_CTXT_JOINED)
		ret = 1;

	spin_unlock(&dlm_domain_lock);
	return ret;
}

static void dlm_free_ctxt_mem(struct dlm_ctxt *dlm)
{
S
Sunil Mushran 已提交
301 302
	dlm_destroy_debugfs_subroot(dlm);

303
	if (dlm->lockres_hash)
304
		dlm_free_pagevec((void **)dlm->lockres_hash, DLM_HASH_PAGES);
305

306 307 308
	if (dlm->master_hash)
		dlm_free_pagevec((void **)dlm->master_hash, DLM_HASH_PAGES);

309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390
	if (dlm->name)
		kfree(dlm->name);

	kfree(dlm);
}

/* A little strange - this function will be called while holding
 * dlm_domain_lock and is expected to be holding it on the way out. We
 * will however drop and reacquire it multiple times */
static void dlm_ctxt_release(struct kref *kref)
{
	struct dlm_ctxt *dlm;

	dlm = container_of(kref, struct dlm_ctxt, dlm_refs);

	BUG_ON(dlm->num_joins);
	BUG_ON(dlm->dlm_state == DLM_CTXT_JOINED);

	/* we may still be in the list if we hit an error during join. */
	list_del_init(&dlm->list);

	spin_unlock(&dlm_domain_lock);

	mlog(0, "freeing memory from domain %s\n", dlm->name);

	wake_up(&dlm_domain_events);

	dlm_free_ctxt_mem(dlm);

	spin_lock(&dlm_domain_lock);
}

void dlm_put(struct dlm_ctxt *dlm)
{
	spin_lock(&dlm_domain_lock);
	kref_put(&dlm->dlm_refs, dlm_ctxt_release);
	spin_unlock(&dlm_domain_lock);
}

static void __dlm_get(struct dlm_ctxt *dlm)
{
	kref_get(&dlm->dlm_refs);
}

/* given a questionable reference to a dlm object, gets a reference if
 * it can find it in the list, otherwise returns NULL in which case
 * you shouldn't trust your pointer. */
struct dlm_ctxt *dlm_grab(struct dlm_ctxt *dlm)
{
	struct list_head *iter;
	struct dlm_ctxt *target = NULL;

	spin_lock(&dlm_domain_lock);

	list_for_each(iter, &dlm_domains) {
		target = list_entry (iter, struct dlm_ctxt, list);

		if (target == dlm) {
			__dlm_get(target);
			break;
		}

		target = NULL;
	}

	spin_unlock(&dlm_domain_lock);

	return target;
}

int dlm_domain_fully_joined(struct dlm_ctxt *dlm)
{
	int ret;

	spin_lock(&dlm_domain_lock);
	ret = (dlm->dlm_state == DLM_CTXT_JOINED) ||
		(dlm->dlm_state == DLM_CTXT_IN_SHUTDOWN);
	spin_unlock(&dlm_domain_lock);

	return ret;
}

391 392 393 394 395 396 397 398 399
static void dlm_destroy_dlm_worker(struct dlm_ctxt *dlm)
{
	if (dlm->dlm_worker) {
		flush_workqueue(dlm->dlm_worker);
		destroy_workqueue(dlm->dlm_worker);
		dlm->dlm_worker = NULL;
	}
}

400 401 402
static void dlm_complete_dlm_shutdown(struct dlm_ctxt *dlm)
{
	dlm_unregister_domain_handlers(dlm);
403
	dlm_debug_shutdown(dlm);
404 405
	dlm_complete_thread(dlm);
	dlm_complete_recovery_thread(dlm);
406
	dlm_destroy_dlm_worker(dlm);
407 408 409 410 411 412 413 414 415 416 417 418

	/* We've left the domain. Now we can take ourselves out of the
	 * list and allow the kref stuff to help us free the
	 * memory. */
	spin_lock(&dlm_domain_lock);
	list_del_init(&dlm->list);
	spin_unlock(&dlm_domain_lock);

	/* Wake up anyone waiting for us to remove this domain */
	wake_up(&dlm_domain_events);
}

419
static int dlm_migrate_all_locks(struct dlm_ctxt *dlm)
420
{
421
	int i, num, n, ret = 0;
422
	struct dlm_lock_resource *res;
423 424 425
	struct hlist_node *iter;
	struct hlist_head *bucket;
	int dropped;
426 427

	mlog(0, "Migrating locks from domain %s\n", dlm->name);
428 429

	num = 0;
430
	spin_lock(&dlm->spinlock);
431
	for (i = 0; i < DLM_HASH_BUCKETS; i++) {
432 433 434 435 436 437 438 439
redo_bucket:
		n = 0;
		bucket = dlm_lockres_hash(dlm, i);
		iter = bucket->first;
		while (iter) {
			n++;
			res = hlist_entry(iter, struct dlm_lock_resource,
					  hash_node);
440
			dlm_lockres_get(res);
441 442 443 444 445 446 447 448 449
			/* migrate, if necessary.  this will drop the dlm
			 * spinlock and retake it if it does migration. */
			dropped = dlm_empty_lockres(dlm, res);

			spin_lock(&res->spinlock);
			__dlm_lockres_calc_usage(dlm, res);
			iter = res->hash_node.next;
			spin_unlock(&res->spinlock);

450
			dlm_lockres_put(res);
451 452 453

			if (dropped)
				goto redo_bucket;
454
		}
455
		cond_resched_lock(&dlm->spinlock);
456 457 458
		num += n;
		mlog(0, "%s: touched %d lockreses in bucket %d "
		     "(tot=%d)\n", dlm->name, n, i, num);
459 460
	}
	spin_unlock(&dlm->spinlock);
461 462 463 464 465 466 467 468 469
	wake_up(&dlm->dlm_thread_wq);

	/* let the dlm thread take care of purging, keep scanning until
	 * nothing remains in the hash */
	if (num) {
		mlog(0, "%s: %d lock resources in hash last pass\n",
		     dlm->name, num);
		ret = -EAGAIN;
	}
470
	mlog(0, "DONE Migrating locks from domain %s\n", dlm->name);
471
	return ret;
472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513
}

static int dlm_no_joining_node(struct dlm_ctxt *dlm)
{
	int ret;

	spin_lock(&dlm->spinlock);
	ret = dlm->joining_node == DLM_LOCK_RES_OWNER_UNKNOWN;
	spin_unlock(&dlm->spinlock);

	return ret;
}

static void dlm_mark_domain_leaving(struct dlm_ctxt *dlm)
{
	/* Yikes, a double spinlock! I need domain_lock for the dlm
	 * state and the dlm spinlock for join state... Sorry! */
again:
	spin_lock(&dlm_domain_lock);
	spin_lock(&dlm->spinlock);

	if (dlm->joining_node != DLM_LOCK_RES_OWNER_UNKNOWN) {
		mlog(0, "Node %d is joining, we wait on it.\n",
			  dlm->joining_node);
		spin_unlock(&dlm->spinlock);
		spin_unlock(&dlm_domain_lock);

		wait_event(dlm->dlm_join_events, dlm_no_joining_node(dlm));
		goto again;
	}

	dlm->dlm_state = DLM_CTXT_LEAVING;
	spin_unlock(&dlm->spinlock);
	spin_unlock(&dlm_domain_lock);
}

static void __dlm_print_nodes(struct dlm_ctxt *dlm)
{
	int node = -1;

	assert_spin_locked(&dlm->spinlock);

514
	printk(KERN_NOTICE "o2dlm: Nodes in domain %s: ", dlm->name);
515 516 517

	while ((node = find_next_bit(dlm->domain_map, O2NM_MAX_NODES,
				     node + 1)) < O2NM_MAX_NODES) {
S
Sunil Mushran 已提交
518
		printk("%d ", node);
519
	}
S
Sunil Mushran 已提交
520
	printk("\n");
521 522
}

523 524
static int dlm_exit_domain_handler(struct o2net_msg *msg, u32 len, void *data,
				   void **ret_data)
525 526 527 528 529 530 531 532 533 534 535 536
{
	struct dlm_ctxt *dlm = data;
	unsigned int node;
	struct dlm_exit_domain *exit_msg = (struct dlm_exit_domain *) msg->buf;

	mlog_entry("%p %u %p", msg, len, data);

	if (!dlm_grab(dlm))
		return 0;

	node = exit_msg->node_idx;

537
	printk(KERN_NOTICE "o2dlm: Node %u leaves domain %s\n", node, dlm->name);
538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567

	spin_lock(&dlm->spinlock);
	clear_bit(node, dlm->domain_map);
	__dlm_print_nodes(dlm);

	/* notify anything attached to the heartbeat events */
	dlm_hb_event_notify_attached(dlm, node, 0);

	spin_unlock(&dlm->spinlock);

	dlm_put(dlm);

	return 0;
}

static int dlm_send_one_domain_exit(struct dlm_ctxt *dlm,
				    unsigned int node)
{
	int status;
	struct dlm_exit_domain leave_msg;

	mlog(0, "Asking node %u if we can leave the domain %s me = %u\n",
		  node, dlm->name, dlm->node_num);

	memset(&leave_msg, 0, sizeof(leave_msg));
	leave_msg.node_idx = dlm->node_num;

	status = o2net_send_message(DLM_EXIT_DOMAIN_MSG, dlm->key,
				    &leave_msg, sizeof(leave_msg), node,
				    NULL);
568 569 570
	if (status < 0)
		mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to "
		     "node %u\n", status, DLM_EXIT_DOMAIN_MSG, dlm->key, node);
571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654
	mlog(0, "status return %d from o2net_send_message\n", status);

	return status;
}


static void dlm_leave_domain(struct dlm_ctxt *dlm)
{
	int node, clear_node, status;

	/* At this point we've migrated away all our locks and won't
	 * accept mastership of new ones. The dlm is responsible for
	 * almost nothing now. We make sure not to confuse any joining
	 * nodes and then commence shutdown procedure. */

	spin_lock(&dlm->spinlock);
	/* Clear ourselves from the domain map */
	clear_bit(dlm->node_num, dlm->domain_map);
	while ((node = find_next_bit(dlm->domain_map, O2NM_MAX_NODES,
				     0)) < O2NM_MAX_NODES) {
		/* Drop the dlm spinlock. This is safe wrt the domain_map.
		 * -nodes cannot be added now as the
		 *   query_join_handlers knows to respond with OK_NO_MAP
		 * -we catch the right network errors if a node is
		 *   removed from the map while we're sending him the
		 *   exit message. */
		spin_unlock(&dlm->spinlock);

		clear_node = 1;

		status = dlm_send_one_domain_exit(dlm, node);
		if (status < 0 &&
		    status != -ENOPROTOOPT &&
		    status != -ENOTCONN) {
			mlog(ML_NOTICE, "Error %d sending domain exit message "
			     "to node %d\n", status, node);

			/* Not sure what to do here but lets sleep for
			 * a bit in case this was a transient
			 * error... */
			msleep(DLM_DOMAIN_BACKOFF_MS);
			clear_node = 0;
		}

		spin_lock(&dlm->spinlock);
		/* If we're not clearing the node bit then we intend
		 * to loop back around to try again. */
		if (clear_node)
			clear_bit(node, dlm->domain_map);
	}
	spin_unlock(&dlm->spinlock);
}

int dlm_joined(struct dlm_ctxt *dlm)
{
	int ret = 0;

	spin_lock(&dlm_domain_lock);

	if (dlm->dlm_state == DLM_CTXT_JOINED)
		ret = 1;

	spin_unlock(&dlm_domain_lock);

	return ret;
}

int dlm_shutting_down(struct dlm_ctxt *dlm)
{
	int ret = 0;

	spin_lock(&dlm_domain_lock);

	if (dlm->dlm_state == DLM_CTXT_IN_SHUTDOWN)
		ret = 1;

	spin_unlock(&dlm_domain_lock);

	return ret;
}

void dlm_unregister_domain(struct dlm_ctxt *dlm)
{
	int leave = 0;
655
	struct dlm_lock_resource *res;
656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679

	spin_lock(&dlm_domain_lock);
	BUG_ON(dlm->dlm_state != DLM_CTXT_JOINED);
	BUG_ON(!dlm->num_joins);

	dlm->num_joins--;
	if (!dlm->num_joins) {
		/* We mark it "in shutdown" now so new register
		 * requests wait until we've completely left the
		 * domain. Don't use DLM_CTXT_LEAVING yet as we still
		 * want new domain joins to communicate with us at
		 * least until we've completed migration of our
		 * resources. */
		dlm->dlm_state = DLM_CTXT_IN_SHUTDOWN;
		leave = 1;
	}
	spin_unlock(&dlm_domain_lock);

	if (leave) {
		mlog(0, "shutting down domain %s\n", dlm->name);

		/* We changed dlm state, notify the thread */
		dlm_kick_thread(dlm, NULL);

680
		while (dlm_migrate_all_locks(dlm)) {
681 682
			/* Give dlm_thread time to purge the lockres' */
			msleep(500);
683 684
			mlog(0, "%s: more migration to do\n", dlm->name);
		}
685 686 687 688 689 690 691 692 693

		/* This list should be empty. If not, print remaining lockres */
		if (!list_empty(&dlm->tracking_list)) {
			mlog(ML_ERROR, "Following lockres' are still on the "
			     "tracking list:\n");
			list_for_each_entry(res, &dlm->tracking_list, tracking)
				dlm_print_one_lock_resource(res);
		}

694 695 696 697 698 699 700 701
		dlm_mark_domain_leaving(dlm);
		dlm_leave_domain(dlm);
		dlm_complete_dlm_shutdown(dlm);
	}
	dlm_put(dlm);
}
EXPORT_SYMBOL_GPL(dlm_unregister_domain);

702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733
static int dlm_query_join_proto_check(char *proto_type, int node,
				      struct dlm_protocol_version *ours,
				      struct dlm_protocol_version *request)
{
	int rc;
	struct dlm_protocol_version proto = *request;

	if (!dlm_protocol_compare(ours, &proto)) {
		mlog(0,
		     "node %u wanted to join with %s locking protocol "
		     "%u.%u, we respond with %u.%u\n",
		     node, proto_type,
		     request->pv_major,
		     request->pv_minor,
		     proto.pv_major, proto.pv_minor);
		request->pv_minor = proto.pv_minor;
		rc = 0;
	} else {
		mlog(ML_NOTICE,
		     "Node %u wanted to join with %s locking "
		     "protocol %u.%u, but we have %u.%u, disallowing\n",
		     node, proto_type,
		     request->pv_major,
		     request->pv_minor,
		     ours->pv_major,
		     ours->pv_minor);
		rc = 1;
	}

	return rc;
}

734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764
/*
 * struct dlm_query_join_packet is made up of four one-byte fields.  They
 * are effectively in big-endian order already.  However, little-endian
 * machines swap them before putting the packet on the wire (because
 * query_join's response is a status, and that status is treated as a u32
 * on the wire).  Thus, a big-endian and little-endian machines will treat
 * this structure differently.
 *
 * The solution is to have little-endian machines swap the structure when
 * converting from the structure to the u32 representation.  This will
 * result in the structure having the correct format on the wire no matter
 * the host endian format.
 */
static void dlm_query_join_packet_to_wire(struct dlm_query_join_packet *packet,
					  u32 *wire)
{
	union dlm_query_join_response response;

	response.packet = *packet;
	*wire = cpu_to_be32(response.intval);
}

static void dlm_query_join_wire_to_packet(u32 wire,
					  struct dlm_query_join_packet *packet)
{
	union dlm_query_join_response response;

	response.intval = cpu_to_be32(wire);
	*packet = response.packet;
}

765 766
static int dlm_query_join_handler(struct o2net_msg *msg, u32 len, void *data,
				  void **ret_data)
767 768
{
	struct dlm_query_join_request *query;
769 770
	struct dlm_query_join_packet packet = {
		.code = JOIN_DISALLOW,
771
	};
772
	struct dlm_ctxt *dlm = NULL;
773
	u32 response;
774
	u8 nodenum;
775 776 777 778 779 780 781 782 783 784 785 786 787 788 789

	query = (struct dlm_query_join_request *) msg->buf;

	mlog(0, "node %u wants to join domain %s\n", query->node_idx,
		  query->domain);

	/*
	 * If heartbeat doesn't consider the node live, tell it
	 * to back off and try again.  This gives heartbeat a chance
	 * to catch up.
	 */
	if (!o2hb_check_node_heartbeating(query->node_idx)) {
		mlog(0, "node %u is not in our live map yet\n",
		     query->node_idx);

790
		packet.code = JOIN_DISALLOW;
791 792 793
		goto respond;
	}

794
	packet.code = JOIN_OK_NO_MAP;
795 796 797

	spin_lock(&dlm_domain_lock);
	dlm = __dlm_lookup_domain_full(query->domain, query->name_len);
798 799 800 801 802 803 804 805 806 807 808 809
	if (!dlm)
		goto unlock_respond;

	/*
	 * There is a small window where the joining node may not see the
	 * node(s) that just left but still part of the cluster. DISALLOW
	 * join request if joining node has different node map.
	 */
	nodenum=0;
	while (nodenum < O2NM_MAX_NODES) {
		if (test_bit(nodenum, dlm->domain_map)) {
			if (!byte_test_bit(nodenum, query->node_map)) {
810 811 812
				mlog(0, "disallow join as node %u does not "
				     "have node %u in its nodemap\n",
				     query->node_idx, nodenum);
813
				packet.code = JOIN_DISALLOW;
814 815 816 817 818 819
				goto unlock_respond;
			}
		}
		nodenum++;
	}

820
	/* Once the dlm ctxt is marked as leaving then we don't want
821
	 * to be put in someone's domain map.
K
Kurt Hackel 已提交
822 823
	 * Also, explicitly disallow joining at certain troublesome
	 * times (ie. during recovery). */
824
	if (dlm && dlm->dlm_state != DLM_CTXT_LEAVING) {
K
Kurt Hackel 已提交
825
		int bit = query->node_idx;
826 827 828 829 830 831 832
		spin_lock(&dlm->spinlock);

		if (dlm->dlm_state == DLM_CTXT_NEW &&
		    dlm->joining_node == DLM_LOCK_RES_OWNER_UNKNOWN) {
			/*If this is a brand new context and we
			 * haven't started our join process yet, then
			 * the other node won the race. */
833
			packet.code = JOIN_OK_NO_MAP;
834 835
		} else if (dlm->joining_node != DLM_LOCK_RES_OWNER_UNKNOWN) {
			/* Disallow parallel joins. */
836
			packet.code = JOIN_DISALLOW;
K
Kurt Hackel 已提交
837
		} else if (dlm->reco.state & DLM_RECO_STATE_ACTIVE) {
838
			mlog(0, "node %u trying to join, but recovery "
K
Kurt Hackel 已提交
839
			     "is ongoing.\n", bit);
840
			packet.code = JOIN_DISALLOW;
K
Kurt Hackel 已提交
841
		} else if (test_bit(bit, dlm->recovery_map)) {
842
			mlog(0, "node %u trying to join, but it "
K
Kurt Hackel 已提交
843
			     "still needs recovery.\n", bit);
844
			packet.code = JOIN_DISALLOW;
K
Kurt Hackel 已提交
845
		} else if (test_bit(bit, dlm->domain_map)) {
846
			mlog(0, "node %u trying to join, but it "
K
Kurt Hackel 已提交
847 848
			     "is still in the domain! needs recovery?\n",
			     bit);
849
			packet.code = JOIN_DISALLOW;
850 851 852 853 854
		} else {
			/* Alright we're fully a part of this domain
			 * so we keep some state as to who's joining
			 * and indicate to him that needs to be fixed
			 * up. */
855 856 857 858 859

			/* Make sure we speak compatible locking protocols.  */
			if (dlm_query_join_proto_check("DLM", bit,
						       &dlm->dlm_locking_proto,
						       &query->dlm_proto)) {
860
				packet.code = JOIN_PROTOCOL_MISMATCH;
861 862 863
			} else if (dlm_query_join_proto_check("fs", bit,
							      &dlm->fs_locking_proto,
							      &query->fs_proto)) {
864
				packet.code = JOIN_PROTOCOL_MISMATCH;
865
			} else {
866 867 868
				packet.dlm_minor = query->dlm_proto.pv_minor;
				packet.fs_minor = query->fs_proto.pv_minor;
				packet.code = JOIN_OK;
869 870
				__dlm_set_joining_node(dlm, query->node_idx);
			}
871 872 873 874
		}

		spin_unlock(&dlm->spinlock);
	}
875
unlock_respond:
876 877 878
	spin_unlock(&dlm_domain_lock);

respond:
879
	mlog(0, "We respond with %u\n", packet.code);
880

881 882
	dlm_query_join_packet_to_wire(&packet, &response);
	return response;
883 884
}

885 886
static int dlm_assert_joined_handler(struct o2net_msg *msg, u32 len, void *data,
				     void **ret_data)
887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908
{
	struct dlm_assert_joined *assert;
	struct dlm_ctxt *dlm = NULL;

	assert = (struct dlm_assert_joined *) msg->buf;

	mlog(0, "node %u asserts join on domain %s\n", assert->node_idx,
		  assert->domain);

	spin_lock(&dlm_domain_lock);
	dlm = __dlm_lookup_domain_full(assert->domain, assert->name_len);
	/* XXX should we consider no dlm ctxt an error? */
	if (dlm) {
		spin_lock(&dlm->spinlock);

		/* Alright, this node has officially joined our
		 * domain. Set him in the map and clean up our
		 * leftover join state. */
		BUG_ON(dlm->joining_node != assert->node_idx);
		set_bit(assert->node_idx, dlm->domain_map);
		__dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN);

909
		printk(KERN_NOTICE "o2dlm: Node %u joins domain %s\n",
S
Sunil Mushran 已提交
910
		       assert->node_idx, dlm->name);
911 912 913 914 915 916 917 918 919 920 921 922
		__dlm_print_nodes(dlm);

		/* notify anything attached to the heartbeat events */
		dlm_hb_event_notify_attached(dlm, assert->node_idx, 1);

		spin_unlock(&dlm->spinlock);
	}
	spin_unlock(&dlm_domain_lock);

	return 0;
}

923 924
static int dlm_cancel_join_handler(struct o2net_msg *msg, u32 len, void *data,
				   void **ret_data)
925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966
{
	struct dlm_cancel_join *cancel;
	struct dlm_ctxt *dlm = NULL;

	cancel = (struct dlm_cancel_join *) msg->buf;

	mlog(0, "node %u cancels join on domain %s\n", cancel->node_idx,
		  cancel->domain);

	spin_lock(&dlm_domain_lock);
	dlm = __dlm_lookup_domain_full(cancel->domain, cancel->name_len);

	if (dlm) {
		spin_lock(&dlm->spinlock);

		/* Yikes, this guy wants to cancel his join. No
		 * problem, we simply cleanup our join state. */
		BUG_ON(dlm->joining_node != cancel->node_idx);
		__dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN);

		spin_unlock(&dlm->spinlock);
	}
	spin_unlock(&dlm_domain_lock);

	return 0;
}

static int dlm_send_one_join_cancel(struct dlm_ctxt *dlm,
				    unsigned int node)
{
	int status;
	struct dlm_cancel_join cancel_msg;

	memset(&cancel_msg, 0, sizeof(cancel_msg));
	cancel_msg.node_idx = dlm->node_num;
	cancel_msg.name_len = strlen(dlm->name);
	memcpy(cancel_msg.domain, dlm->name, cancel_msg.name_len);

	status = o2net_send_message(DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY,
				    &cancel_msg, sizeof(cancel_msg), node,
				    NULL);
	if (status < 0) {
967 968 969
		mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to "
		     "node %u\n", status, DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY,
		     node);
970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988
		goto bail;
	}

bail:
	return status;
}

/* map_size should be in bytes. */
static int dlm_send_join_cancels(struct dlm_ctxt *dlm,
				 unsigned long *node_map,
				 unsigned int map_size)
{
	int status, tmpstat;
	unsigned int node;

	if (map_size != (BITS_TO_LONGS(O2NM_MAX_NODES) *
			 sizeof(unsigned long))) {
		mlog(ML_ERROR,
		     "map_size %u != BITS_TO_LONGS(O2NM_MAX_NODES) %u\n",
989
		     map_size, (unsigned)BITS_TO_LONGS(O2NM_MAX_NODES));
990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015
		return -EINVAL;
	}

	status = 0;
	node = -1;
	while ((node = find_next_bit(node_map, O2NM_MAX_NODES,
				     node + 1)) < O2NM_MAX_NODES) {
		if (node == dlm->node_num)
			continue;

		tmpstat = dlm_send_one_join_cancel(dlm, node);
		if (tmpstat) {
			mlog(ML_ERROR, "Error return %d cancelling join on "
			     "node %d\n", tmpstat, node);
			if (!status)
				status = tmpstat;
		}
	}

	if (status)
		mlog_errno(status);
	return status;
}

static int dlm_request_join(struct dlm_ctxt *dlm,
			    int node,
1016
			    enum dlm_query_join_response_code *response)
1017
{
1018
	int status;
1019
	struct dlm_query_join_request join_msg;
1020 1021
	struct dlm_query_join_packet packet;
	u32 join_resp;
1022 1023 1024 1025 1026 1027 1028

	mlog(0, "querying node %d\n", node);

	memset(&join_msg, 0, sizeof(join_msg));
	join_msg.node_idx = dlm->node_num;
	join_msg.name_len = strlen(dlm->name);
	memcpy(join_msg.domain, dlm->name, join_msg.name_len);
1029 1030
	join_msg.dlm_proto = dlm->dlm_locking_proto;
	join_msg.fs_proto = dlm->fs_locking_proto;
1031

1032 1033 1034
	/* copy live node map to join message */
	byte_copymap(join_msg.node_map, dlm->live_nodes_map, O2NM_MAX_NODES);

1035
	status = o2net_send_message(DLM_QUERY_JOIN_MSG, DLM_MOD_KEY, &join_msg,
1036
				    sizeof(join_msg), node, &join_resp);
1037
	if (status < 0 && status != -ENOPROTOOPT) {
1038 1039 1040
		mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to "
		     "node %u\n", status, DLM_QUERY_JOIN_MSG, DLM_MOD_KEY,
		     node);
1041 1042
		goto bail;
	}
1043
	dlm_query_join_wire_to_packet(join_resp, &packet);
1044 1045 1046 1047 1048 1049 1050 1051

	/* -ENOPROTOOPT from the net code means the other side isn't
	    listening for our message type -- that's fine, it means
	    his dlm isn't up, so we can consider him a 'yes' but not
	    joined into the domain.  */
	if (status == -ENOPROTOOPT) {
		status = 0;
		*response = JOIN_OK_NO_MAP;
1052 1053 1054 1055
	} else if (packet.code == JOIN_DISALLOW ||
		   packet.code == JOIN_OK_NO_MAP) {
		*response = packet.code;
	} else if (packet.code == JOIN_PROTOCOL_MISMATCH) {
1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066
		mlog(ML_NOTICE,
		     "This node requested DLM locking protocol %u.%u and "
		     "filesystem locking protocol %u.%u.  At least one of "
		     "the protocol versions on node %d is not compatible, "
		     "disconnecting\n",
		     dlm->dlm_locking_proto.pv_major,
		     dlm->dlm_locking_proto.pv_minor,
		     dlm->fs_locking_proto.pv_major,
		     dlm->fs_locking_proto.pv_minor,
		     node);
		status = -EPROTO;
1067 1068 1069
		*response = packet.code;
	} else if (packet.code == JOIN_OK) {
		*response = packet.code;
1070
		/* Use the same locking protocol as the remote node */
1071 1072
		dlm->dlm_locking_proto.pv_minor = packet.dlm_minor;
		dlm->fs_locking_proto.pv_minor = packet.fs_minor;
1073 1074 1075 1076 1077 1078 1079 1080
		mlog(0,
		     "Node %d responds JOIN_OK with DLM locking protocol "
		     "%u.%u and fs locking protocol %u.%u\n",
		     node,
		     dlm->dlm_locking_proto.pv_major,
		     dlm->dlm_locking_proto.pv_minor,
		     dlm->fs_locking_proto.pv_major,
		     dlm->fs_locking_proto.pv_minor);
1081 1082
	} else {
		status = -EINVAL;
1083
		mlog(ML_ERROR, "invalid response %d from node %u\n",
1084
		     packet.code, node);
1085 1086 1087
	}

	mlog(0, "status %d, node %d response is %d\n", status, node,
1088
	     *response);
1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110

bail:
	return status;
}

static int dlm_send_one_join_assert(struct dlm_ctxt *dlm,
				    unsigned int node)
{
	int status;
	struct dlm_assert_joined assert_msg;

	mlog(0, "Sending join assert to node %u\n", node);

	memset(&assert_msg, 0, sizeof(assert_msg));
	assert_msg.node_idx = dlm->node_num;
	assert_msg.name_len = strlen(dlm->name);
	memcpy(assert_msg.domain, dlm->name, assert_msg.name_len);

	status = o2net_send_message(DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY,
				    &assert_msg, sizeof(assert_msg), node,
				    NULL);
	if (status < 0)
1111 1112 1113
		mlog(ML_ERROR, "Error %d when sending message %u (key 0x%x) to "
		     "node %u\n", status, DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY,
		     node);
1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158

	return status;
}

static void dlm_send_join_asserts(struct dlm_ctxt *dlm,
				  unsigned long *node_map)
{
	int status, node, live;

	status = 0;
	node = -1;
	while ((node = find_next_bit(node_map, O2NM_MAX_NODES,
				     node + 1)) < O2NM_MAX_NODES) {
		if (node == dlm->node_num)
			continue;

		do {
			/* It is very important that this message be
			 * received so we spin until either the node
			 * has died or it gets the message. */
			status = dlm_send_one_join_assert(dlm, node);

			spin_lock(&dlm->spinlock);
			live = test_bit(node, dlm->live_nodes_map);
			spin_unlock(&dlm->spinlock);

			if (status) {
				mlog(ML_ERROR, "Error return %d asserting "
				     "join on node %d\n", status, node);

				/* give us some time between errors... */
				if (live)
					msleep(DLM_DOMAIN_BACKOFF_MS);
			}
		} while (status && live);
	}
}

struct domain_join_ctxt {
	unsigned long live_map[BITS_TO_LONGS(O2NM_MAX_NODES)];
	unsigned long yes_resp_map[BITS_TO_LONGS(O2NM_MAX_NODES)];
};

static int dlm_should_restart_join(struct dlm_ctxt *dlm,
				   struct domain_join_ctxt *ctxt,
1159
				   enum dlm_query_join_response_code response)
1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184
{
	int ret;

	if (response == JOIN_DISALLOW) {
		mlog(0, "Latest response of disallow -- should restart\n");
		return 1;
	}

	spin_lock(&dlm->spinlock);
	/* For now, we restart the process if the node maps have
	 * changed at all */
	ret = memcmp(ctxt->live_map, dlm->live_nodes_map,
		     sizeof(dlm->live_nodes_map));
	spin_unlock(&dlm->spinlock);

	if (ret)
		mlog(0, "Node maps changed -- should restart\n");

	return ret;
}

static int dlm_try_to_join_domain(struct dlm_ctxt *dlm)
{
	int status = 0, tmpstat, node;
	struct domain_join_ctxt *ctxt;
1185
	enum dlm_query_join_response_code response = JOIN_DISALLOW;
1186 1187 1188

	mlog_entry("%p", dlm);

1189
	ctxt = kzalloc(sizeof(*ctxt), GFP_KERNEL);
1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278
	if (!ctxt) {
		status = -ENOMEM;
		mlog_errno(status);
		goto bail;
	}

	/* group sem locking should work for us here -- we're already
	 * registered for heartbeat events so filling this should be
	 * atomic wrt getting those handlers called. */
	o2hb_fill_node_map(dlm->live_nodes_map, sizeof(dlm->live_nodes_map));

	spin_lock(&dlm->spinlock);
	memcpy(ctxt->live_map, dlm->live_nodes_map, sizeof(ctxt->live_map));

	__dlm_set_joining_node(dlm, dlm->node_num);

	spin_unlock(&dlm->spinlock);

	node = -1;
	while ((node = find_next_bit(ctxt->live_map, O2NM_MAX_NODES,
				     node + 1)) < O2NM_MAX_NODES) {
		if (node == dlm->node_num)
			continue;

		status = dlm_request_join(dlm, node, &response);
		if (status < 0) {
			mlog_errno(status);
			goto bail;
		}

		/* Ok, either we got a response or the node doesn't have a
		 * dlm up. */
		if (response == JOIN_OK)
			set_bit(node, ctxt->yes_resp_map);

		if (dlm_should_restart_join(dlm, ctxt, response)) {
			status = -EAGAIN;
			goto bail;
		}
	}

	mlog(0, "Yay, done querying nodes!\n");

	/* Yay, everyone agree's we can join the domain. My domain is
	 * comprised of all nodes who were put in the
	 * yes_resp_map. Copy that into our domain map and send a join
	 * assert message to clean up everyone elses state. */
	spin_lock(&dlm->spinlock);
	memcpy(dlm->domain_map, ctxt->yes_resp_map,
	       sizeof(ctxt->yes_resp_map));
	set_bit(dlm->node_num, dlm->domain_map);
	spin_unlock(&dlm->spinlock);

	dlm_send_join_asserts(dlm, ctxt->yes_resp_map);

	/* Joined state *must* be set before the joining node
	 * information, otherwise the query_join handler may read no
	 * current joiner but a state of NEW and tell joining nodes
	 * we're not in the domain. */
	spin_lock(&dlm_domain_lock);
	dlm->dlm_state = DLM_CTXT_JOINED;
	dlm->num_joins++;
	spin_unlock(&dlm_domain_lock);

bail:
	spin_lock(&dlm->spinlock);
	__dlm_set_joining_node(dlm, DLM_LOCK_RES_OWNER_UNKNOWN);
	if (!status)
		__dlm_print_nodes(dlm);
	spin_unlock(&dlm->spinlock);

	if (ctxt) {
		/* Do we need to send a cancel message to any nodes? */
		if (status < 0) {
			tmpstat = dlm_send_join_cancels(dlm,
							ctxt->yes_resp_map,
							sizeof(ctxt->yes_resp_map));
			if (tmpstat < 0)
				mlog_errno(tmpstat);
		}
		kfree(ctxt);
	}

	mlog(0, "returning %d\n", status);
	return status;
}

static void dlm_unregister_domain_handlers(struct dlm_ctxt *dlm)
{
1279 1280
	o2hb_unregister_callback(NULL, &dlm->dlm_hb_up);
	o2hb_unregister_callback(NULL, &dlm->dlm_hb_down);
1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291
	o2net_unregister_handler_list(&dlm->dlm_domain_handlers);
}

static int dlm_register_domain_handlers(struct dlm_ctxt *dlm)
{
	int status;

	mlog(0, "registering handlers.\n");

	o2hb_setup_callback(&dlm->dlm_hb_down, O2HB_NODE_DOWN_CB,
			    dlm_hb_node_down_cb, dlm, DLM_HB_NODE_DOWN_PRI);
1292
	status = o2hb_register_callback(NULL, &dlm->dlm_hb_down);
1293 1294 1295 1296 1297
	if (status)
		goto bail;

	o2hb_setup_callback(&dlm->dlm_hb_up, O2HB_NODE_UP_CB,
			    dlm_hb_node_up_cb, dlm, DLM_HB_NODE_UP_PRI);
1298
	status = o2hb_register_callback(NULL, &dlm->dlm_hb_up);
1299 1300 1301 1302 1303 1304
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_MASTER_REQUEST_MSG, dlm->key,
					sizeof(struct dlm_master_request),
					dlm_master_request_handler,
1305
					dlm, NULL, &dlm->dlm_domain_handlers);
1306 1307 1308 1309 1310 1311
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_ASSERT_MASTER_MSG, dlm->key,
					sizeof(struct dlm_assert_master),
					dlm_assert_master_handler,
1312 1313
					dlm, dlm_assert_master_post_handler,
					&dlm->dlm_domain_handlers);
1314 1315 1316 1317 1318 1319
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_CREATE_LOCK_MSG, dlm->key,
					sizeof(struct dlm_create_lock),
					dlm_create_lock_handler,
1320
					dlm, NULL, &dlm->dlm_domain_handlers);
1321 1322 1323 1324 1325 1326
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_CONVERT_LOCK_MSG, dlm->key,
					DLM_CONVERT_LOCK_MAX_LEN,
					dlm_convert_lock_handler,
1327
					dlm, NULL, &dlm->dlm_domain_handlers);
1328 1329 1330 1331 1332 1333
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_UNLOCK_LOCK_MSG, dlm->key,
					DLM_UNLOCK_LOCK_MAX_LEN,
					dlm_unlock_lock_handler,
1334
					dlm, NULL, &dlm->dlm_domain_handlers);
1335 1336 1337 1338 1339 1340
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_PROXY_AST_MSG, dlm->key,
					DLM_PROXY_AST_MAX_LEN,
					dlm_proxy_ast_handler,
1341
					dlm, NULL, &dlm->dlm_domain_handlers);
1342 1343 1344 1345 1346 1347
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_EXIT_DOMAIN_MSG, dlm->key,
					sizeof(struct dlm_exit_domain),
					dlm_exit_domain_handler,
1348
					dlm, NULL, &dlm->dlm_domain_handlers);
1349 1350 1351
	if (status)
		goto bail;

1352 1353 1354
	status = o2net_register_handler(DLM_DEREF_LOCKRES_MSG, dlm->key,
					sizeof(struct dlm_deref_lockres),
					dlm_deref_lockres_handler,
1355
					dlm, NULL, &dlm->dlm_domain_handlers);
1356 1357 1358
	if (status)
		goto bail;

1359 1360 1361
	status = o2net_register_handler(DLM_MIGRATE_REQUEST_MSG, dlm->key,
					sizeof(struct dlm_migrate_request),
					dlm_migrate_request_handler,
1362
					dlm, NULL, &dlm->dlm_domain_handlers);
1363 1364 1365 1366 1367 1368
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_MIG_LOCKRES_MSG, dlm->key,
					DLM_MIG_LOCKRES_MAX_LEN,
					dlm_mig_lockres_handler,
1369
					dlm, NULL, &dlm->dlm_domain_handlers);
1370 1371 1372 1373 1374 1375
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_MASTER_REQUERY_MSG, dlm->key,
					sizeof(struct dlm_master_requery),
					dlm_master_requery_handler,
1376
					dlm, NULL, &dlm->dlm_domain_handlers);
1377 1378 1379 1380 1381 1382
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_LOCK_REQUEST_MSG, dlm->key,
					sizeof(struct dlm_lock_request),
					dlm_request_all_locks_handler,
1383
					dlm, NULL, &dlm->dlm_domain_handlers);
1384 1385 1386 1387 1388 1389
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_RECO_DATA_DONE_MSG, dlm->key,
					sizeof(struct dlm_reco_data_done),
					dlm_reco_data_done_handler,
1390
					dlm, NULL, &dlm->dlm_domain_handlers);
1391 1392 1393 1394 1395 1396
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_BEGIN_RECO_MSG, dlm->key,
					sizeof(struct dlm_begin_reco),
					dlm_begin_reco_handler,
1397
					dlm, NULL, &dlm->dlm_domain_handlers);
1398 1399 1400 1401 1402 1403
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_FINALIZE_RECO_MSG, dlm->key,
					sizeof(struct dlm_finalize_reco),
					dlm_finalize_reco_handler,
1404
					dlm, NULL, &dlm->dlm_domain_handlers);
1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417
	if (status)
		goto bail;

bail:
	if (status)
		dlm_unregister_domain_handlers(dlm);

	return status;
}

static int dlm_join_domain(struct dlm_ctxt *dlm)
{
	int status;
1418 1419
	unsigned int backoff;
	unsigned int total_backoff = 0;
1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430

	BUG_ON(!dlm);

	mlog(0, "Join domain %s\n", dlm->name);

	status = dlm_register_domain_handlers(dlm);
	if (status) {
		mlog_errno(status);
		goto bail;
	}

1431 1432 1433 1434 1435 1436
	status = dlm_debug_init(dlm);
	if (status < 0) {
		mlog_errno(status);
		goto bail;
	}

1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448
	status = dlm_launch_thread(dlm);
	if (status < 0) {
		mlog_errno(status);
		goto bail;
	}

	status = dlm_launch_recovery_thread(dlm);
	if (status < 0) {
		mlog_errno(status);
		goto bail;
	}

1449 1450 1451 1452 1453 1454 1455
	dlm->dlm_worker = create_singlethread_workqueue("dlm_wq");
	if (!dlm->dlm_worker) {
		status = -ENOMEM;
		mlog_errno(status);
		goto bail;
	}

1456 1457 1458 1459 1460 1461
	do {
		status = dlm_try_to_join_domain(dlm);

		/* If we're racing another node to the join, then we
		 * need to back off temporarily and let them
		 * complete. */
1462
#define	DLM_JOIN_TIMEOUT_MSECS	90000
1463 1464 1465 1466 1467 1468
		if (status == -EAGAIN) {
			if (signal_pending(current)) {
				status = -ERESTARTSYS;
				goto bail;
			}

1469 1470 1471 1472 1473 1474 1475 1476 1477
			if (total_backoff >
			    msecs_to_jiffies(DLM_JOIN_TIMEOUT_MSECS)) {
				status = -ERESTARTSYS;
				mlog(ML_NOTICE, "Timed out joining dlm domain "
				     "%s after %u msecs\n", dlm->name,
				     jiffies_to_msecs(total_backoff));
				goto bail;
			}

1478 1479 1480 1481 1482 1483 1484 1485 1486
			/*
			 * <chip> After you!
			 * <dale> No, after you!
			 * <chip> I insist!
			 * <dale> But you first!
			 * ...
			 */
			backoff = (unsigned int)(jiffies & 0x3);
			backoff *= DLM_DOMAIN_BACKOFF_MS;
1487
			total_backoff += backoff;
1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503
			mlog(0, "backoff %d\n", backoff);
			msleep(backoff);
		}
	} while (status == -EAGAIN);

	if (status < 0) {
		mlog_errno(status);
		goto bail;
	}

	status = 0;
bail:
	wake_up(&dlm_domain_events);

	if (status) {
		dlm_unregister_domain_handlers(dlm);
1504
		dlm_debug_shutdown(dlm);
1505 1506
		dlm_complete_thread(dlm);
		dlm_complete_recovery_thread(dlm);
1507
		dlm_destroy_dlm_worker(dlm);
1508 1509 1510 1511 1512 1513 1514 1515 1516
	}

	return status;
}

static struct dlm_ctxt *dlm_alloc_ctxt(const char *domain,
				u32 key)
{
	int i;
S
Sunil Mushran 已提交
1517
	int ret;
1518 1519
	struct dlm_ctxt *dlm = NULL;

1520
	dlm = kzalloc(sizeof(*dlm), GFP_KERNEL);
1521 1522 1523 1524 1525
	if (!dlm) {
		mlog_errno(-ENOMEM);
		goto leave;
	}

J
Julia Lawall 已提交
1526
	dlm->name = kstrdup(domain, GFP_KERNEL);
1527 1528 1529 1530 1531 1532 1533
	if (dlm->name == NULL) {
		mlog_errno(-ENOMEM);
		kfree(dlm);
		dlm = NULL;
		goto leave;
	}

1534
	dlm->lockres_hash = (struct hlist_head **)dlm_alloc_pagevec(DLM_HASH_PAGES);
1535
	if (!dlm->lockres_hash) {
1536 1537 1538 1539 1540 1541 1542
		mlog_errno(-ENOMEM);
		kfree(dlm->name);
		kfree(dlm);
		dlm = NULL;
		goto leave;
	}

1543 1544
	for (i = 0; i < DLM_HASH_BUCKETS; i++)
		INIT_HLIST_HEAD(dlm_lockres_hash(dlm, i));
1545

1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559
	dlm->master_hash = (struct hlist_head **)
				dlm_alloc_pagevec(DLM_HASH_PAGES);
	if (!dlm->master_hash) {
		mlog_errno(-ENOMEM);
		dlm_free_pagevec((void **)dlm->lockres_hash, DLM_HASH_PAGES);
		kfree(dlm->name);
		kfree(dlm);
		dlm = NULL;
		goto leave;
	}

	for (i = 0; i < DLM_HASH_BUCKETS; i++)
		INIT_HLIST_HEAD(dlm_master_hash(dlm, i));

1560 1561 1562
	dlm->key = key;
	dlm->node_num = o2nm_this_node();

S
Sunil Mushran 已提交
1563 1564
	ret = dlm_create_debugfs_subroot(dlm);
	if (ret < 0) {
1565
		dlm_free_pagevec((void **)dlm->master_hash, DLM_HASH_PAGES);
S
Sunil Mushran 已提交
1566 1567 1568 1569 1570 1571 1572
		dlm_free_pagevec((void **)dlm->lockres_hash, DLM_HASH_PAGES);
		kfree(dlm->name);
		kfree(dlm);
		dlm = NULL;
		goto leave;
	}

1573 1574 1575
	spin_lock_init(&dlm->spinlock);
	spin_lock_init(&dlm->master_lock);
	spin_lock_init(&dlm->ast_lock);
1576
	spin_lock_init(&dlm->track_lock);
1577 1578 1579 1580 1581 1582 1583
	INIT_LIST_HEAD(&dlm->list);
	INIT_LIST_HEAD(&dlm->dirty_list);
	INIT_LIST_HEAD(&dlm->reco.resources);
	INIT_LIST_HEAD(&dlm->reco.received);
	INIT_LIST_HEAD(&dlm->reco.node_data);
	INIT_LIST_HEAD(&dlm->purge_list);
	INIT_LIST_HEAD(&dlm->dlm_domain_handlers);
1584
	INIT_LIST_HEAD(&dlm->tracking_list);
1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598
	dlm->reco.state = 0;

	INIT_LIST_HEAD(&dlm->pending_asts);
	INIT_LIST_HEAD(&dlm->pending_basts);

	mlog(0, "dlm->recovery_map=%p, &(dlm->recovery_map[0])=%p\n",
		  dlm->recovery_map, &(dlm->recovery_map[0]));

	memset(dlm->recovery_map, 0, sizeof(dlm->recovery_map));
	memset(dlm->live_nodes_map, 0, sizeof(dlm->live_nodes_map));
	memset(dlm->domain_map, 0, sizeof(dlm->domain_map));

	dlm->dlm_thread_task = NULL;
	dlm->dlm_reco_thread_task = NULL;
1599
	dlm->dlm_worker = NULL;
1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612
	init_waitqueue_head(&dlm->dlm_thread_wq);
	init_waitqueue_head(&dlm->dlm_reco_thread_wq);
	init_waitqueue_head(&dlm->reco.event);
	init_waitqueue_head(&dlm->ast_wq);
	init_waitqueue_head(&dlm->migration_wq);
	INIT_LIST_HEAD(&dlm->mle_hb_events);

	dlm->joining_node = DLM_LOCK_RES_OWNER_UNKNOWN;
	init_waitqueue_head(&dlm->dlm_join_events);

	dlm->reco.new_master = O2NM_INVALID_NODE_NUM;
	dlm->reco.dead_node = O2NM_INVALID_NODE_NUM;

1613 1614
	atomic_set(&dlm->res_tot_count, 0);
	atomic_set(&dlm->res_cur_count, 0);
S
Sunil Mushran 已提交
1615 1616 1617 1618 1619
	for (i = 0; i < DLM_MLE_NUM_TYPES; ++i) {
		atomic_set(&dlm->mle_tot_count[i], 0);
		atomic_set(&dlm->mle_cur_count[i], 0);
	}

1620 1621
	spin_lock_init(&dlm->work_lock);
	INIT_LIST_HEAD(&dlm->work_list);
D
David Howells 已提交
1622
	INIT_WORK(&dlm->dispatched_work, dlm_dispatch_work);
1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636

	kref_init(&dlm->dlm_refs);
	dlm->dlm_state = DLM_CTXT_NEW;

	INIT_LIST_HEAD(&dlm->dlm_eviction_callbacks);

	mlog(0, "context init: refcount %u\n",
		  atomic_read(&dlm->dlm_refs.refcount));

leave:
	return dlm;
}

/*
1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664
 * Compare a requested locking protocol version against the current one.
 *
 * If the major numbers are different, they are incompatible.
 * If the current minor is greater than the request, they are incompatible.
 * If the current minor is less than or equal to the request, they are
 * compatible, and the requester should run at the current minor version.
 */
static int dlm_protocol_compare(struct dlm_protocol_version *existing,
				struct dlm_protocol_version *request)
{
	if (existing->pv_major != request->pv_major)
		return 1;

	if (existing->pv_minor > request->pv_minor)
		return 1;

	if (existing->pv_minor < request->pv_minor)
		request->pv_minor = existing->pv_minor;

	return 0;
}

/*
 * dlm_register_domain: one-time setup per "domain".
 *
 * The filesystem passes in the requested locking version via proto.
 * If registration was successful, proto will contain the negotiated
 * locking protocol.
1665 1666
 */
struct dlm_ctxt * dlm_register_domain(const char *domain,
1667 1668
			       u32 key,
			       struct dlm_protocol_version *fs_proto)
1669 1670 1671 1672 1673
{
	int ret;
	struct dlm_ctxt *dlm = NULL;
	struct dlm_ctxt *new_ctxt = NULL;

1674
	if (strlen(domain) >= O2NM_MAX_NAME_LEN) {
1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710
		ret = -ENAMETOOLONG;
		mlog(ML_ERROR, "domain name length too long\n");
		goto leave;
	}

	if (!o2hb_check_local_node_heartbeating()) {
		mlog(ML_ERROR, "the local node has not been configured, or is "
		     "not heartbeating\n");
		ret = -EPROTO;
		goto leave;
	}

	mlog(0, "register called for domain \"%s\"\n", domain);

retry:
	dlm = NULL;
	if (signal_pending(current)) {
		ret = -ERESTARTSYS;
		mlog_errno(ret);
		goto leave;
	}

	spin_lock(&dlm_domain_lock);

	dlm = __dlm_lookup_domain(domain);
	if (dlm) {
		if (dlm->dlm_state != DLM_CTXT_JOINED) {
			spin_unlock(&dlm_domain_lock);

			mlog(0, "This ctxt is not joined yet!\n");
			wait_event_interruptible(dlm_domain_events,
						 dlm_wait_on_domain_helper(
							 domain));
			goto retry;
		}

1711
		if (dlm_protocol_compare(&dlm->fs_locking_proto, fs_proto)) {
1712
			spin_unlock(&dlm_domain_lock);
1713 1714 1715 1716 1717 1718 1719 1720
			mlog(ML_ERROR,
			     "Requested locking protocol version is not "
			     "compatible with already registered domain "
			     "\"%s\"\n", domain);
			ret = -EPROTO;
			goto leave;
		}

1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750
		__dlm_get(dlm);
		dlm->num_joins++;

		spin_unlock(&dlm_domain_lock);

		ret = 0;
		goto leave;
	}

	/* doesn't exist */
	if (!new_ctxt) {
		spin_unlock(&dlm_domain_lock);

		new_ctxt = dlm_alloc_ctxt(domain, key);
		if (new_ctxt)
			goto retry;

		ret = -ENOMEM;
		mlog_errno(ret);
		goto leave;
	}

	/* a little variable switch-a-roo here... */
	dlm = new_ctxt;
	new_ctxt = NULL;

	/* add the new domain */
	list_add_tail(&dlm->list, &dlm_domains);
	spin_unlock(&dlm_domain_lock);

1751 1752 1753 1754 1755 1756 1757
	/*
	 * Pass the locking protocol version into the join.  If the join
	 * succeeds, it will have the negotiated protocol set.
	 */
	dlm->dlm_locking_proto = dlm_protocol;
	dlm->fs_locking_proto = *fs_proto;

1758 1759 1760 1761 1762 1763 1764
	ret = dlm_join_domain(dlm);
	if (ret) {
		mlog_errno(ret);
		dlm_put(dlm);
		goto leave;
	}

1765 1766 1767
	/* Tell the caller what locking protocol we negotiated */
	*fs_proto = dlm->fs_locking_proto;

1768 1769 1770 1771 1772 1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793
	ret = 0;
leave:
	if (new_ctxt)
		dlm_free_ctxt_mem(new_ctxt);

	if (ret < 0)
		dlm = ERR_PTR(ret);

	return dlm;
}
EXPORT_SYMBOL_GPL(dlm_register_domain);

static LIST_HEAD(dlm_join_handlers);

static void dlm_unregister_net_handlers(void)
{
	o2net_unregister_handler_list(&dlm_join_handlers);
}

static int dlm_register_net_handlers(void)
{
	int status = 0;

	status = o2net_register_handler(DLM_QUERY_JOIN_MSG, DLM_MOD_KEY,
					sizeof(struct dlm_query_join_request),
					dlm_query_join_handler,
1794
					NULL, NULL, &dlm_join_handlers);
1795 1796 1797 1798 1799 1800
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_ASSERT_JOINED_MSG, DLM_MOD_KEY,
					sizeof(struct dlm_assert_joined),
					dlm_assert_joined_handler,
1801
					NULL, NULL, &dlm_join_handlers);
1802 1803 1804 1805 1806 1807
	if (status)
		goto bail;

	status = o2net_register_handler(DLM_CANCEL_JOIN_MSG, DLM_MOD_KEY,
					sizeof(struct dlm_cancel_join),
					dlm_cancel_join_handler,
1808
					NULL, NULL, &dlm_join_handlers);
1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878

bail:
	if (status < 0)
		dlm_unregister_net_handlers();

	return status;
}

/* Domain eviction callback handling.
 *
 * The file system requires notification of node death *before* the
 * dlm completes it's recovery work, otherwise it may be able to
 * acquire locks on resources requiring recovery. Since the dlm can
 * evict a node from it's domain *before* heartbeat fires, a similar
 * mechanism is required. */

/* Eviction is not expected to happen often, so a per-domain lock is
 * not necessary. Eviction callbacks are allowed to sleep for short
 * periods of time. */
static DECLARE_RWSEM(dlm_callback_sem);

void dlm_fire_domain_eviction_callbacks(struct dlm_ctxt *dlm,
					int node_num)
{
	struct list_head *iter;
	struct dlm_eviction_cb *cb;

	down_read(&dlm_callback_sem);
	list_for_each(iter, &dlm->dlm_eviction_callbacks) {
		cb = list_entry(iter, struct dlm_eviction_cb, ec_item);

		cb->ec_func(node_num, cb->ec_data);
	}
	up_read(&dlm_callback_sem);
}

void dlm_setup_eviction_cb(struct dlm_eviction_cb *cb,
			   dlm_eviction_func *f,
			   void *data)
{
	INIT_LIST_HEAD(&cb->ec_item);
	cb->ec_func = f;
	cb->ec_data = data;
}
EXPORT_SYMBOL_GPL(dlm_setup_eviction_cb);

void dlm_register_eviction_cb(struct dlm_ctxt *dlm,
			      struct dlm_eviction_cb *cb)
{
	down_write(&dlm_callback_sem);
	list_add_tail(&cb->ec_item, &dlm->dlm_eviction_callbacks);
	up_write(&dlm_callback_sem);
}
EXPORT_SYMBOL_GPL(dlm_register_eviction_cb);

void dlm_unregister_eviction_cb(struct dlm_eviction_cb *cb)
{
	down_write(&dlm_callback_sem);
	list_del_init(&cb->ec_item);
	up_write(&dlm_callback_sem);
}
EXPORT_SYMBOL_GPL(dlm_unregister_eviction_cb);

static int __init dlm_init(void)
{
	int status;

	dlm_print_version();

	status = dlm_init_mle_cache();
1879 1880
	if (status) {
		mlog(ML_ERROR, "Could not create o2dlm_mle slabcache\n");
1881 1882 1883 1884 1885 1886 1887 1888 1889 1890 1891 1892 1893 1894
		goto error;
	}

	status = dlm_init_master_caches();
	if (status) {
		mlog(ML_ERROR, "Could not create o2dlm_lockres and "
		     "o2dlm_lockname slabcaches\n");
		goto error;
	}

	status = dlm_init_lock_cache();
	if (status) {
		mlog(ML_ERROR, "Count not create o2dlm_lock slabcache\n");
		goto error;
1895
	}
1896 1897 1898

	status = dlm_register_net_handlers();
	if (status) {
1899 1900
		mlog(ML_ERROR, "Unable to register network handlers\n");
		goto error;
1901 1902
	}

S
Sunil Mushran 已提交
1903 1904 1905 1906
	status = dlm_create_debugfs_root();
	if (status)
		goto error;

1907
	return 0;
1908
error:
S
Sunil Mushran 已提交
1909
	dlm_unregister_net_handlers();
1910 1911 1912 1913
	dlm_destroy_lock_cache();
	dlm_destroy_master_caches();
	dlm_destroy_mle_cache();
	return -1;
1914 1915 1916 1917
}

static void __exit dlm_exit (void)
{
S
Sunil Mushran 已提交
1918
	dlm_destroy_debugfs_root();
1919
	dlm_unregister_net_handlers();
1920 1921
	dlm_destroy_lock_cache();
	dlm_destroy_master_caches();
1922 1923 1924 1925 1926 1927 1928 1929
	dlm_destroy_mle_cache();
}

MODULE_AUTHOR("Oracle");
MODULE_LICENSE("GPL");

module_init(dlm_init);
module_exit(dlm_exit);