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