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