xref: /openbmc/linux/fs/ocfs2/dlm/dlmmaster.c (revision 52fb57e7)
1 /* -*- mode: c; c-basic-offset: 8; -*-
2  * vim: noexpandtab sw=8 ts=8 sts=0:
3  *
4  * dlmmod.c
5  *
6  * standalone DLM module
7  *
8  * Copyright (C) 2004 Oracle.  All rights reserved.
9  *
10  * This program is free software; you can redistribute it and/or
11  * modify it under the terms of the GNU General Public
12  * License as published by the Free Software Foundation; either
13  * version 2 of the License, or (at your option) any later version.
14  *
15  * This program is distributed in the hope that it will be useful,
16  * but WITHOUT ANY WARRANTY; without even the implied warranty of
17  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
18  * General Public License for more details.
19  *
20  * You should have received a copy of the GNU General Public
21  * License along with this program; if not, write to the
22  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
23  * Boston, MA 021110-1307, USA.
24  *
25  */
26 
27 
28 #include <linux/module.h>
29 #include <linux/fs.h>
30 #include <linux/types.h>
31 #include <linux/slab.h>
32 #include <linux/highmem.h>
33 #include <linux/init.h>
34 #include <linux/sysctl.h>
35 #include <linux/random.h>
36 #include <linux/blkdev.h>
37 #include <linux/socket.h>
38 #include <linux/inet.h>
39 #include <linux/spinlock.h>
40 #include <linux/delay.h>
41 
42 
43 #include "cluster/heartbeat.h"
44 #include "cluster/nodemanager.h"
45 #include "cluster/tcp.h"
46 
47 #include "dlmapi.h"
48 #include "dlmcommon.h"
49 #include "dlmdomain.h"
50 #include "dlmdebug.h"
51 
52 #define MLOG_MASK_PREFIX (ML_DLM|ML_DLM_MASTER)
53 #include "cluster/masklog.h"
54 
55 static void dlm_mle_node_down(struct dlm_ctxt *dlm,
56 			      struct dlm_master_list_entry *mle,
57 			      struct o2nm_node *node,
58 			      int idx);
59 static void dlm_mle_node_up(struct dlm_ctxt *dlm,
60 			    struct dlm_master_list_entry *mle,
61 			    struct o2nm_node *node,
62 			    int idx);
63 
64 static void dlm_assert_master_worker(struct dlm_work_item *item, void *data);
65 static int dlm_do_assert_master(struct dlm_ctxt *dlm,
66 				struct dlm_lock_resource *res,
67 				void *nodemap, u32 flags);
68 static void dlm_deref_lockres_worker(struct dlm_work_item *item, void *data);
69 
70 static inline int dlm_mle_equal(struct dlm_ctxt *dlm,
71 				struct dlm_master_list_entry *mle,
72 				const char *name,
73 				unsigned int namelen)
74 {
75 	if (dlm != mle->dlm)
76 		return 0;
77 
78 	if (namelen != mle->mnamelen ||
79 	    memcmp(name, mle->mname, namelen) != 0)
80 		return 0;
81 
82 	return 1;
83 }
84 
85 static struct kmem_cache *dlm_lockres_cache;
86 static struct kmem_cache *dlm_lockname_cache;
87 static struct kmem_cache *dlm_mle_cache;
88 
89 static void dlm_mle_release(struct kref *kref);
90 static void dlm_init_mle(struct dlm_master_list_entry *mle,
91 			enum dlm_mle_type type,
92 			struct dlm_ctxt *dlm,
93 			struct dlm_lock_resource *res,
94 			const char *name,
95 			unsigned int namelen);
96 static void dlm_put_mle(struct dlm_master_list_entry *mle);
97 static void __dlm_put_mle(struct dlm_master_list_entry *mle);
98 static int dlm_find_mle(struct dlm_ctxt *dlm,
99 			struct dlm_master_list_entry **mle,
100 			char *name, unsigned int namelen);
101 
102 static int dlm_do_master_request(struct dlm_lock_resource *res,
103 				 struct dlm_master_list_entry *mle, int to);
104 
105 
106 static int dlm_wait_for_lock_mastery(struct dlm_ctxt *dlm,
107 				     struct dlm_lock_resource *res,
108 				     struct dlm_master_list_entry *mle,
109 				     int *blocked);
110 static int dlm_restart_lock_mastery(struct dlm_ctxt *dlm,
111 				    struct dlm_lock_resource *res,
112 				    struct dlm_master_list_entry *mle,
113 				    int blocked);
114 static int dlm_add_migration_mle(struct dlm_ctxt *dlm,
115 				 struct dlm_lock_resource *res,
116 				 struct dlm_master_list_entry *mle,
117 				 struct dlm_master_list_entry **oldmle,
118 				 const char *name, unsigned int namelen,
119 				 u8 new_master, u8 master);
120 
121 static u8 dlm_pick_migration_target(struct dlm_ctxt *dlm,
122 				    struct dlm_lock_resource *res);
123 static void dlm_remove_nonlocal_locks(struct dlm_ctxt *dlm,
124 				      struct dlm_lock_resource *res);
125 static int dlm_mark_lockres_migrating(struct dlm_ctxt *dlm,
126 				       struct dlm_lock_resource *res,
127 				       u8 target);
128 static int dlm_pre_master_reco_lockres(struct dlm_ctxt *dlm,
129 				       struct dlm_lock_resource *res);
130 
131 
132 int dlm_is_host_down(int errno)
133 {
134 	switch (errno) {
135 		case -EBADF:
136 		case -ECONNREFUSED:
137 		case -ENOTCONN:
138 		case -ECONNRESET:
139 		case -EPIPE:
140 		case -EHOSTDOWN:
141 		case -EHOSTUNREACH:
142 		case -ETIMEDOUT:
143 		case -ECONNABORTED:
144 		case -ENETDOWN:
145 		case -ENETUNREACH:
146 		case -ENETRESET:
147 		case -ESHUTDOWN:
148 		case -ENOPROTOOPT:
149 		case -EINVAL:   /* if returned from our tcp code,
150 				   this means there is no socket */
151 			return 1;
152 	}
153 	return 0;
154 }
155 
156 
157 /*
158  * MASTER LIST FUNCTIONS
159  */
160 
161 
162 /*
163  * regarding master list entries and heartbeat callbacks:
164  *
165  * in order to avoid sleeping and allocation that occurs in
166  * heartbeat, master list entries are simply attached to the
167  * dlm's established heartbeat callbacks.  the mle is attached
168  * when it is created, and since the dlm->spinlock is held at
169  * that time, any heartbeat event will be properly discovered
170  * by the mle.  the mle needs to be detached from the
171  * dlm->mle_hb_events list as soon as heartbeat events are no
172  * longer useful to the mle, and before the mle is freed.
173  *
174  * as a general rule, heartbeat events are no longer needed by
175  * the mle once an "answer" regarding the lock master has been
176  * received.
177  */
178 static inline void __dlm_mle_attach_hb_events(struct dlm_ctxt *dlm,
179 					      struct dlm_master_list_entry *mle)
180 {
181 	assert_spin_locked(&dlm->spinlock);
182 
183 	list_add_tail(&mle->hb_events, &dlm->mle_hb_events);
184 }
185 
186 
187 static inline void __dlm_mle_detach_hb_events(struct dlm_ctxt *dlm,
188 					      struct dlm_master_list_entry *mle)
189 {
190 	if (!list_empty(&mle->hb_events))
191 		list_del_init(&mle->hb_events);
192 }
193 
194 
195 static inline void dlm_mle_detach_hb_events(struct dlm_ctxt *dlm,
196 					    struct dlm_master_list_entry *mle)
197 {
198 	spin_lock(&dlm->spinlock);
199 	__dlm_mle_detach_hb_events(dlm, mle);
200 	spin_unlock(&dlm->spinlock);
201 }
202 
203 static void dlm_get_mle_inuse(struct dlm_master_list_entry *mle)
204 {
205 	struct dlm_ctxt *dlm;
206 	dlm = mle->dlm;
207 
208 	assert_spin_locked(&dlm->spinlock);
209 	assert_spin_locked(&dlm->master_lock);
210 	mle->inuse++;
211 	kref_get(&mle->mle_refs);
212 }
213 
214 static void dlm_put_mle_inuse(struct dlm_master_list_entry *mle)
215 {
216 	struct dlm_ctxt *dlm;
217 	dlm = mle->dlm;
218 
219 	spin_lock(&dlm->spinlock);
220 	spin_lock(&dlm->master_lock);
221 	mle->inuse--;
222 	__dlm_put_mle(mle);
223 	spin_unlock(&dlm->master_lock);
224 	spin_unlock(&dlm->spinlock);
225 
226 }
227 
228 /* remove from list and free */
229 static void __dlm_put_mle(struct dlm_master_list_entry *mle)
230 {
231 	struct dlm_ctxt *dlm;
232 	dlm = mle->dlm;
233 
234 	assert_spin_locked(&dlm->spinlock);
235 	assert_spin_locked(&dlm->master_lock);
236 	if (!atomic_read(&mle->mle_refs.refcount)) {
237 		/* this may or may not crash, but who cares.
238 		 * it's a BUG. */
239 		mlog(ML_ERROR, "bad mle: %p\n", mle);
240 		dlm_print_one_mle(mle);
241 		BUG();
242 	} else
243 		kref_put(&mle->mle_refs, dlm_mle_release);
244 }
245 
246 
247 /* must not have any spinlocks coming in */
248 static void dlm_put_mle(struct dlm_master_list_entry *mle)
249 {
250 	struct dlm_ctxt *dlm;
251 	dlm = mle->dlm;
252 
253 	spin_lock(&dlm->spinlock);
254 	spin_lock(&dlm->master_lock);
255 	__dlm_put_mle(mle);
256 	spin_unlock(&dlm->master_lock);
257 	spin_unlock(&dlm->spinlock);
258 }
259 
260 static inline void dlm_get_mle(struct dlm_master_list_entry *mle)
261 {
262 	kref_get(&mle->mle_refs);
263 }
264 
265 static void dlm_init_mle(struct dlm_master_list_entry *mle,
266 			enum dlm_mle_type type,
267 			struct dlm_ctxt *dlm,
268 			struct dlm_lock_resource *res,
269 			const char *name,
270 			unsigned int namelen)
271 {
272 	assert_spin_locked(&dlm->spinlock);
273 
274 	mle->dlm = dlm;
275 	mle->type = type;
276 	INIT_HLIST_NODE(&mle->master_hash_node);
277 	INIT_LIST_HEAD(&mle->hb_events);
278 	memset(mle->maybe_map, 0, sizeof(mle->maybe_map));
279 	spin_lock_init(&mle->spinlock);
280 	init_waitqueue_head(&mle->wq);
281 	atomic_set(&mle->woken, 0);
282 	kref_init(&mle->mle_refs);
283 	memset(mle->response_map, 0, sizeof(mle->response_map));
284 	mle->master = O2NM_MAX_NODES;
285 	mle->new_master = O2NM_MAX_NODES;
286 	mle->inuse = 0;
287 
288 	BUG_ON(mle->type != DLM_MLE_BLOCK &&
289 	       mle->type != DLM_MLE_MASTER &&
290 	       mle->type != DLM_MLE_MIGRATION);
291 
292 	if (mle->type == DLM_MLE_MASTER) {
293 		BUG_ON(!res);
294 		mle->mleres = res;
295 		memcpy(mle->mname, res->lockname.name, res->lockname.len);
296 		mle->mnamelen = res->lockname.len;
297 		mle->mnamehash = res->lockname.hash;
298 	} else {
299 		BUG_ON(!name);
300 		mle->mleres = NULL;
301 		memcpy(mle->mname, name, namelen);
302 		mle->mnamelen = namelen;
303 		mle->mnamehash = dlm_lockid_hash(name, namelen);
304 	}
305 
306 	atomic_inc(&dlm->mle_tot_count[mle->type]);
307 	atomic_inc(&dlm->mle_cur_count[mle->type]);
308 
309 	/* copy off the node_map and register hb callbacks on our copy */
310 	memcpy(mle->node_map, dlm->domain_map, sizeof(mle->node_map));
311 	memcpy(mle->vote_map, dlm->domain_map, sizeof(mle->vote_map));
312 	clear_bit(dlm->node_num, mle->vote_map);
313 	clear_bit(dlm->node_num, mle->node_map);
314 
315 	/* attach the mle to the domain node up/down events */
316 	__dlm_mle_attach_hb_events(dlm, mle);
317 }
318 
319 void __dlm_unlink_mle(struct dlm_ctxt *dlm, struct dlm_master_list_entry *mle)
320 {
321 	assert_spin_locked(&dlm->spinlock);
322 	assert_spin_locked(&dlm->master_lock);
323 
324 	if (!hlist_unhashed(&mle->master_hash_node))
325 		hlist_del_init(&mle->master_hash_node);
326 }
327 
328 void __dlm_insert_mle(struct dlm_ctxt *dlm, struct dlm_master_list_entry *mle)
329 {
330 	struct hlist_head *bucket;
331 
332 	assert_spin_locked(&dlm->master_lock);
333 
334 	bucket = dlm_master_hash(dlm, mle->mnamehash);
335 	hlist_add_head(&mle->master_hash_node, bucket);
336 }
337 
338 /* returns 1 if found, 0 if not */
339 static int dlm_find_mle(struct dlm_ctxt *dlm,
340 			struct dlm_master_list_entry **mle,
341 			char *name, unsigned int namelen)
342 {
343 	struct dlm_master_list_entry *tmpmle;
344 	struct hlist_head *bucket;
345 	unsigned int hash;
346 
347 	assert_spin_locked(&dlm->master_lock);
348 
349 	hash = dlm_lockid_hash(name, namelen);
350 	bucket = dlm_master_hash(dlm, hash);
351 	hlist_for_each_entry(tmpmle, bucket, master_hash_node) {
352 		if (!dlm_mle_equal(dlm, tmpmle, name, namelen))
353 			continue;
354 		dlm_get_mle(tmpmle);
355 		*mle = tmpmle;
356 		return 1;
357 	}
358 	return 0;
359 }
360 
361 void dlm_hb_event_notify_attached(struct dlm_ctxt *dlm, int idx, int node_up)
362 {
363 	struct dlm_master_list_entry *mle;
364 
365 	assert_spin_locked(&dlm->spinlock);
366 
367 	list_for_each_entry(mle, &dlm->mle_hb_events, hb_events) {
368 		if (node_up)
369 			dlm_mle_node_up(dlm, mle, NULL, idx);
370 		else
371 			dlm_mle_node_down(dlm, mle, NULL, idx);
372 	}
373 }
374 
375 static void dlm_mle_node_down(struct dlm_ctxt *dlm,
376 			      struct dlm_master_list_entry *mle,
377 			      struct o2nm_node *node, int idx)
378 {
379 	spin_lock(&mle->spinlock);
380 
381 	if (!test_bit(idx, mle->node_map))
382 		mlog(0, "node %u already removed from nodemap!\n", idx);
383 	else
384 		clear_bit(idx, mle->node_map);
385 
386 	spin_unlock(&mle->spinlock);
387 }
388 
389 static void dlm_mle_node_up(struct dlm_ctxt *dlm,
390 			    struct dlm_master_list_entry *mle,
391 			    struct o2nm_node *node, int idx)
392 {
393 	spin_lock(&mle->spinlock);
394 
395 	if (test_bit(idx, mle->node_map))
396 		mlog(0, "node %u already in node map!\n", idx);
397 	else
398 		set_bit(idx, mle->node_map);
399 
400 	spin_unlock(&mle->spinlock);
401 }
402 
403 
404 int dlm_init_mle_cache(void)
405 {
406 	dlm_mle_cache = kmem_cache_create("o2dlm_mle",
407 					  sizeof(struct dlm_master_list_entry),
408 					  0, SLAB_HWCACHE_ALIGN,
409 					  NULL);
410 	if (dlm_mle_cache == NULL)
411 		return -ENOMEM;
412 	return 0;
413 }
414 
415 void dlm_destroy_mle_cache(void)
416 {
417 	if (dlm_mle_cache)
418 		kmem_cache_destroy(dlm_mle_cache);
419 }
420 
421 static void dlm_mle_release(struct kref *kref)
422 {
423 	struct dlm_master_list_entry *mle;
424 	struct dlm_ctxt *dlm;
425 
426 	mle = container_of(kref, struct dlm_master_list_entry, mle_refs);
427 	dlm = mle->dlm;
428 
429 	assert_spin_locked(&dlm->spinlock);
430 	assert_spin_locked(&dlm->master_lock);
431 
432 	mlog(0, "Releasing mle for %.*s, type %d\n", mle->mnamelen, mle->mname,
433 	     mle->type);
434 
435 	/* remove from list if not already */
436 	__dlm_unlink_mle(dlm, mle);
437 
438 	/* detach the mle from the domain node up/down events */
439 	__dlm_mle_detach_hb_events(dlm, mle);
440 
441 	atomic_dec(&dlm->mle_cur_count[mle->type]);
442 
443 	/* NOTE: kfree under spinlock here.
444 	 * if this is bad, we can move this to a freelist. */
445 	kmem_cache_free(dlm_mle_cache, mle);
446 }
447 
448 
449 /*
450  * LOCK RESOURCE FUNCTIONS
451  */
452 
453 int dlm_init_master_caches(void)
454 {
455 	dlm_lockres_cache = kmem_cache_create("o2dlm_lockres",
456 					      sizeof(struct dlm_lock_resource),
457 					      0, SLAB_HWCACHE_ALIGN, NULL);
458 	if (!dlm_lockres_cache)
459 		goto bail;
460 
461 	dlm_lockname_cache = kmem_cache_create("o2dlm_lockname",
462 					       DLM_LOCKID_NAME_MAX, 0,
463 					       SLAB_HWCACHE_ALIGN, NULL);
464 	if (!dlm_lockname_cache)
465 		goto bail;
466 
467 	return 0;
468 bail:
469 	dlm_destroy_master_caches();
470 	return -ENOMEM;
471 }
472 
473 void dlm_destroy_master_caches(void)
474 {
475 	if (dlm_lockname_cache) {
476 		kmem_cache_destroy(dlm_lockname_cache);
477 		dlm_lockname_cache = NULL;
478 	}
479 
480 	if (dlm_lockres_cache) {
481 		kmem_cache_destroy(dlm_lockres_cache);
482 		dlm_lockres_cache = NULL;
483 	}
484 }
485 
486 static void dlm_lockres_release(struct kref *kref)
487 {
488 	struct dlm_lock_resource *res;
489 	struct dlm_ctxt *dlm;
490 
491 	res = container_of(kref, struct dlm_lock_resource, refs);
492 	dlm = res->dlm;
493 
494 	/* This should not happen -- all lockres' have a name
495 	 * associated with them at init time. */
496 	BUG_ON(!res->lockname.name);
497 
498 	mlog(0, "destroying lockres %.*s\n", res->lockname.len,
499 	     res->lockname.name);
500 
501 	spin_lock(&dlm->track_lock);
502 	if (!list_empty(&res->tracking))
503 		list_del_init(&res->tracking);
504 	else {
505 		mlog(ML_ERROR, "Resource %.*s not on the Tracking list\n",
506 		     res->lockname.len, res->lockname.name);
507 		dlm_print_one_lock_resource(res);
508 	}
509 	spin_unlock(&dlm->track_lock);
510 
511 	atomic_dec(&dlm->res_cur_count);
512 
513 	if (!hlist_unhashed(&res->hash_node) ||
514 	    !list_empty(&res->granted) ||
515 	    !list_empty(&res->converting) ||
516 	    !list_empty(&res->blocked) ||
517 	    !list_empty(&res->dirty) ||
518 	    !list_empty(&res->recovering) ||
519 	    !list_empty(&res->purge)) {
520 		mlog(ML_ERROR,
521 		     "Going to BUG for resource %.*s."
522 		     "  We're on a list! [%c%c%c%c%c%c%c]\n",
523 		     res->lockname.len, res->lockname.name,
524 		     !hlist_unhashed(&res->hash_node) ? 'H' : ' ',
525 		     !list_empty(&res->granted) ? 'G' : ' ',
526 		     !list_empty(&res->converting) ? 'C' : ' ',
527 		     !list_empty(&res->blocked) ? 'B' : ' ',
528 		     !list_empty(&res->dirty) ? 'D' : ' ',
529 		     !list_empty(&res->recovering) ? 'R' : ' ',
530 		     !list_empty(&res->purge) ? 'P' : ' ');
531 
532 		dlm_print_one_lock_resource(res);
533 	}
534 
535 	/* By the time we're ready to blow this guy away, we shouldn't
536 	 * be on any lists. */
537 	BUG_ON(!hlist_unhashed(&res->hash_node));
538 	BUG_ON(!list_empty(&res->granted));
539 	BUG_ON(!list_empty(&res->converting));
540 	BUG_ON(!list_empty(&res->blocked));
541 	BUG_ON(!list_empty(&res->dirty));
542 	BUG_ON(!list_empty(&res->recovering));
543 	BUG_ON(!list_empty(&res->purge));
544 
545 	kmem_cache_free(dlm_lockname_cache, (void *)res->lockname.name);
546 
547 	kmem_cache_free(dlm_lockres_cache, res);
548 }
549 
550 void dlm_lockres_put(struct dlm_lock_resource *res)
551 {
552 	kref_put(&res->refs, dlm_lockres_release);
553 }
554 
555 static void dlm_init_lockres(struct dlm_ctxt *dlm,
556 			     struct dlm_lock_resource *res,
557 			     const char *name, unsigned int namelen)
558 {
559 	char *qname;
560 
561 	/* If we memset here, we lose our reference to the kmalloc'd
562 	 * res->lockname.name, so be sure to init every field
563 	 * correctly! */
564 
565 	qname = (char *) res->lockname.name;
566 	memcpy(qname, name, namelen);
567 
568 	res->lockname.len = namelen;
569 	res->lockname.hash = dlm_lockid_hash(name, namelen);
570 
571 	init_waitqueue_head(&res->wq);
572 	spin_lock_init(&res->spinlock);
573 	INIT_HLIST_NODE(&res->hash_node);
574 	INIT_LIST_HEAD(&res->granted);
575 	INIT_LIST_HEAD(&res->converting);
576 	INIT_LIST_HEAD(&res->blocked);
577 	INIT_LIST_HEAD(&res->dirty);
578 	INIT_LIST_HEAD(&res->recovering);
579 	INIT_LIST_HEAD(&res->purge);
580 	INIT_LIST_HEAD(&res->tracking);
581 	atomic_set(&res->asts_reserved, 0);
582 	res->migration_pending = 0;
583 	res->inflight_locks = 0;
584 	res->inflight_assert_workers = 0;
585 
586 	res->dlm = dlm;
587 
588 	kref_init(&res->refs);
589 
590 	atomic_inc(&dlm->res_tot_count);
591 	atomic_inc(&dlm->res_cur_count);
592 
593 	/* just for consistency */
594 	spin_lock(&res->spinlock);
595 	dlm_set_lockres_owner(dlm, res, DLM_LOCK_RES_OWNER_UNKNOWN);
596 	spin_unlock(&res->spinlock);
597 
598 	res->state = DLM_LOCK_RES_IN_PROGRESS;
599 
600 	res->last_used = 0;
601 
602 	spin_lock(&dlm->spinlock);
603 	list_add_tail(&res->tracking, &dlm->tracking_list);
604 	spin_unlock(&dlm->spinlock);
605 
606 	memset(res->lvb, 0, DLM_LVB_LEN);
607 	memset(res->refmap, 0, sizeof(res->refmap));
608 }
609 
610 struct dlm_lock_resource *dlm_new_lockres(struct dlm_ctxt *dlm,
611 				   const char *name,
612 				   unsigned int namelen)
613 {
614 	struct dlm_lock_resource *res = NULL;
615 
616 	res = kmem_cache_zalloc(dlm_lockres_cache, GFP_NOFS);
617 	if (!res)
618 		goto error;
619 
620 	res->lockname.name = kmem_cache_zalloc(dlm_lockname_cache, GFP_NOFS);
621 	if (!res->lockname.name)
622 		goto error;
623 
624 	dlm_init_lockres(dlm, res, name, namelen);
625 	return res;
626 
627 error:
628 	if (res)
629 		kmem_cache_free(dlm_lockres_cache, res);
630 	return NULL;
631 }
632 
633 void dlm_lockres_set_refmap_bit(struct dlm_ctxt *dlm,
634 				struct dlm_lock_resource *res, int bit)
635 {
636 	assert_spin_locked(&res->spinlock);
637 
638 	mlog(0, "res %.*s, set node %u, %ps()\n", res->lockname.len,
639 	     res->lockname.name, bit, __builtin_return_address(0));
640 
641 	set_bit(bit, res->refmap);
642 }
643 
644 void dlm_lockres_clear_refmap_bit(struct dlm_ctxt *dlm,
645 				  struct dlm_lock_resource *res, int bit)
646 {
647 	assert_spin_locked(&res->spinlock);
648 
649 	mlog(0, "res %.*s, clr node %u, %ps()\n", res->lockname.len,
650 	     res->lockname.name, bit, __builtin_return_address(0));
651 
652 	clear_bit(bit, res->refmap);
653 }
654 
655 static void __dlm_lockres_grab_inflight_ref(struct dlm_ctxt *dlm,
656 				   struct dlm_lock_resource *res)
657 {
658 	res->inflight_locks++;
659 
660 	mlog(0, "%s: res %.*s, inflight++: now %u, %ps()\n", dlm->name,
661 	     res->lockname.len, res->lockname.name, res->inflight_locks,
662 	     __builtin_return_address(0));
663 }
664 
665 void dlm_lockres_grab_inflight_ref(struct dlm_ctxt *dlm,
666 				   struct dlm_lock_resource *res)
667 {
668 	assert_spin_locked(&res->spinlock);
669 	__dlm_lockres_grab_inflight_ref(dlm, res);
670 }
671 
672 void dlm_lockres_drop_inflight_ref(struct dlm_ctxt *dlm,
673 				   struct dlm_lock_resource *res)
674 {
675 	assert_spin_locked(&res->spinlock);
676 
677 	BUG_ON(res->inflight_locks == 0);
678 
679 	res->inflight_locks--;
680 
681 	mlog(0, "%s: res %.*s, inflight--: now %u, %ps()\n", dlm->name,
682 	     res->lockname.len, res->lockname.name, res->inflight_locks,
683 	     __builtin_return_address(0));
684 
685 	wake_up(&res->wq);
686 }
687 
688 void __dlm_lockres_grab_inflight_worker(struct dlm_ctxt *dlm,
689 		struct dlm_lock_resource *res)
690 {
691 	assert_spin_locked(&res->spinlock);
692 	res->inflight_assert_workers++;
693 	mlog(0, "%s:%.*s: inflight assert worker++: now %u\n",
694 			dlm->name, res->lockname.len, res->lockname.name,
695 			res->inflight_assert_workers);
696 }
697 
698 static void __dlm_lockres_drop_inflight_worker(struct dlm_ctxt *dlm,
699 		struct dlm_lock_resource *res)
700 {
701 	assert_spin_locked(&res->spinlock);
702 	BUG_ON(res->inflight_assert_workers == 0);
703 	res->inflight_assert_workers--;
704 	mlog(0, "%s:%.*s: inflight assert worker--: now %u\n",
705 			dlm->name, res->lockname.len, res->lockname.name,
706 			res->inflight_assert_workers);
707 }
708 
709 static void dlm_lockres_drop_inflight_worker(struct dlm_ctxt *dlm,
710 		struct dlm_lock_resource *res)
711 {
712 	spin_lock(&res->spinlock);
713 	__dlm_lockres_drop_inflight_worker(dlm, res);
714 	spin_unlock(&res->spinlock);
715 }
716 
717 /*
718  * lookup a lock resource by name.
719  * may already exist in the hashtable.
720  * lockid is null terminated
721  *
722  * if not, allocate enough for the lockres and for
723  * the temporary structure used in doing the mastering.
724  *
725  * also, do a lookup in the dlm->master_list to see
726  * if another node has begun mastering the same lock.
727  * if so, there should be a block entry in there
728  * for this name, and we should *not* attempt to master
729  * the lock here.   need to wait around for that node
730  * to assert_master (or die).
731  *
732  */
733 struct dlm_lock_resource * dlm_get_lock_resource(struct dlm_ctxt *dlm,
734 					  const char *lockid,
735 					  int namelen,
736 					  int flags)
737 {
738 	struct dlm_lock_resource *tmpres=NULL, *res=NULL;
739 	struct dlm_master_list_entry *mle = NULL;
740 	struct dlm_master_list_entry *alloc_mle = NULL;
741 	int blocked = 0;
742 	int ret, nodenum;
743 	struct dlm_node_iter iter;
744 	unsigned int hash;
745 	int tries = 0;
746 	int bit, wait_on_recovery = 0;
747 
748 	BUG_ON(!lockid);
749 
750 	hash = dlm_lockid_hash(lockid, namelen);
751 
752 	mlog(0, "get lockres %s (len %d)\n", lockid, namelen);
753 
754 lookup:
755 	spin_lock(&dlm->spinlock);
756 	tmpres = __dlm_lookup_lockres_full(dlm, lockid, namelen, hash);
757 	if (tmpres) {
758 		spin_unlock(&dlm->spinlock);
759 		spin_lock(&tmpres->spinlock);
760 
761 		/*
762 		 * Right after dlm spinlock was released, dlm_thread could have
763 		 * purged the lockres. Check if lockres got unhashed. If so
764 		 * start over.
765 		 */
766 		if (hlist_unhashed(&tmpres->hash_node)) {
767 			spin_unlock(&tmpres->spinlock);
768 			dlm_lockres_put(tmpres);
769 			tmpres = NULL;
770 			goto lookup;
771 		}
772 
773 		/* Wait on the thread that is mastering the resource */
774 		if (tmpres->owner == DLM_LOCK_RES_OWNER_UNKNOWN) {
775 			__dlm_wait_on_lockres(tmpres);
776 			BUG_ON(tmpres->owner == DLM_LOCK_RES_OWNER_UNKNOWN);
777 			spin_unlock(&tmpres->spinlock);
778 			dlm_lockres_put(tmpres);
779 			tmpres = NULL;
780 			goto lookup;
781 		}
782 
783 		/* Wait on the resource purge to complete before continuing */
784 		if (tmpres->state & DLM_LOCK_RES_DROPPING_REF) {
785 			BUG_ON(tmpres->owner == dlm->node_num);
786 			__dlm_wait_on_lockres_flags(tmpres,
787 						    DLM_LOCK_RES_DROPPING_REF);
788 			spin_unlock(&tmpres->spinlock);
789 			dlm_lockres_put(tmpres);
790 			tmpres = NULL;
791 			goto lookup;
792 		}
793 
794 		/* Grab inflight ref to pin the resource */
795 		dlm_lockres_grab_inflight_ref(dlm, tmpres);
796 
797 		spin_unlock(&tmpres->spinlock);
798 		if (res)
799 			dlm_lockres_put(res);
800 		res = tmpres;
801 		goto leave;
802 	}
803 
804 	if (!res) {
805 		spin_unlock(&dlm->spinlock);
806 		mlog(0, "allocating a new resource\n");
807 		/* nothing found and we need to allocate one. */
808 		alloc_mle = kmem_cache_alloc(dlm_mle_cache, GFP_NOFS);
809 		if (!alloc_mle)
810 			goto leave;
811 		res = dlm_new_lockres(dlm, lockid, namelen);
812 		if (!res)
813 			goto leave;
814 		goto lookup;
815 	}
816 
817 	mlog(0, "no lockres found, allocated our own: %p\n", res);
818 
819 	if (flags & LKM_LOCAL) {
820 		/* caller knows it's safe to assume it's not mastered elsewhere
821 		 * DONE!  return right away */
822 		spin_lock(&res->spinlock);
823 		dlm_change_lockres_owner(dlm, res, dlm->node_num);
824 		__dlm_insert_lockres(dlm, res);
825 		dlm_lockres_grab_inflight_ref(dlm, res);
826 		spin_unlock(&res->spinlock);
827 		spin_unlock(&dlm->spinlock);
828 		/* lockres still marked IN_PROGRESS */
829 		goto wake_waiters;
830 	}
831 
832 	/* check master list to see if another node has started mastering it */
833 	spin_lock(&dlm->master_lock);
834 
835 	/* if we found a block, wait for lock to be mastered by another node */
836 	blocked = dlm_find_mle(dlm, &mle, (char *)lockid, namelen);
837 	if (blocked) {
838 		int mig;
839 		if (mle->type == DLM_MLE_MASTER) {
840 			mlog(ML_ERROR, "master entry for nonexistent lock!\n");
841 			BUG();
842 		}
843 		mig = (mle->type == DLM_MLE_MIGRATION);
844 		/* if there is a migration in progress, let the migration
845 		 * finish before continuing.  we can wait for the absence
846 		 * of the MIGRATION mle: either the migrate finished or
847 		 * one of the nodes died and the mle was cleaned up.
848 		 * if there is a BLOCK here, but it already has a master
849 		 * set, we are too late.  the master does not have a ref
850 		 * for us in the refmap.  detach the mle and drop it.
851 		 * either way, go back to the top and start over. */
852 		if (mig || mle->master != O2NM_MAX_NODES) {
853 			BUG_ON(mig && mle->master == dlm->node_num);
854 			/* we arrived too late.  the master does not
855 			 * have a ref for us. retry. */
856 			mlog(0, "%s:%.*s: late on %s\n",
857 			     dlm->name, namelen, lockid,
858 			     mig ?  "MIGRATION" : "BLOCK");
859 			spin_unlock(&dlm->master_lock);
860 			spin_unlock(&dlm->spinlock);
861 
862 			/* master is known, detach */
863 			if (!mig)
864 				dlm_mle_detach_hb_events(dlm, mle);
865 			dlm_put_mle(mle);
866 			mle = NULL;
867 			/* this is lame, but we can't wait on either
868 			 * the mle or lockres waitqueue here */
869 			if (mig)
870 				msleep(100);
871 			goto lookup;
872 		}
873 	} else {
874 		/* go ahead and try to master lock on this node */
875 		mle = alloc_mle;
876 		/* make sure this does not get freed below */
877 		alloc_mle = NULL;
878 		dlm_init_mle(mle, DLM_MLE_MASTER, dlm, res, NULL, 0);
879 		set_bit(dlm->node_num, mle->maybe_map);
880 		__dlm_insert_mle(dlm, mle);
881 
882 		/* still holding the dlm spinlock, check the recovery map
883 		 * to see if there are any nodes that still need to be
884 		 * considered.  these will not appear in the mle nodemap
885 		 * but they might own this lockres.  wait on them. */
886 		bit = find_next_bit(dlm->recovery_map, O2NM_MAX_NODES, 0);
887 		if (bit < O2NM_MAX_NODES) {
888 			mlog(0, "%s: res %.*s, At least one node (%d) "
889 			     "to recover before lock mastery can begin\n",
890 			     dlm->name, namelen, (char *)lockid, bit);
891 			wait_on_recovery = 1;
892 		}
893 	}
894 
895 	/* at this point there is either a DLM_MLE_BLOCK or a
896 	 * DLM_MLE_MASTER on the master list, so it's safe to add the
897 	 * lockres to the hashtable.  anyone who finds the lock will
898 	 * still have to wait on the IN_PROGRESS. */
899 
900 	/* finally add the lockres to its hash bucket */
901 	__dlm_insert_lockres(dlm, res);
902 
903 	/* since this lockres is new it doesn't not require the spinlock */
904 	__dlm_lockres_grab_inflight_ref(dlm, res);
905 
906 	/* get an extra ref on the mle in case this is a BLOCK
907 	 * if so, the creator of the BLOCK may try to put the last
908 	 * ref at this time in the assert master handler, so we
909 	 * need an extra one to keep from a bad ptr deref. */
910 	dlm_get_mle_inuse(mle);
911 	spin_unlock(&dlm->master_lock);
912 	spin_unlock(&dlm->spinlock);
913 
914 redo_request:
915 	while (wait_on_recovery) {
916 		/* any cluster changes that occurred after dropping the
917 		 * dlm spinlock would be detectable be a change on the mle,
918 		 * so we only need to clear out the recovery map once. */
919 		if (dlm_is_recovery_lock(lockid, namelen)) {
920 			mlog(0, "%s: Recovery map is not empty, but must "
921 			     "master $RECOVERY lock now\n", dlm->name);
922 			if (!dlm_pre_master_reco_lockres(dlm, res))
923 				wait_on_recovery = 0;
924 			else {
925 				mlog(0, "%s: waiting 500ms for heartbeat state "
926 				    "change\n", dlm->name);
927 				msleep(500);
928 			}
929 			continue;
930 		}
931 
932 		dlm_kick_recovery_thread(dlm);
933 		msleep(1000);
934 		dlm_wait_for_recovery(dlm);
935 
936 		spin_lock(&dlm->spinlock);
937 		bit = find_next_bit(dlm->recovery_map, O2NM_MAX_NODES, 0);
938 		if (bit < O2NM_MAX_NODES) {
939 			mlog(0, "%s: res %.*s, At least one node (%d) "
940 			     "to recover before lock mastery can begin\n",
941 			     dlm->name, namelen, (char *)lockid, bit);
942 			wait_on_recovery = 1;
943 		} else
944 			wait_on_recovery = 0;
945 		spin_unlock(&dlm->spinlock);
946 
947 		if (wait_on_recovery)
948 			dlm_wait_for_node_recovery(dlm, bit, 10000);
949 	}
950 
951 	/* must wait for lock to be mastered elsewhere */
952 	if (blocked)
953 		goto wait;
954 
955 	ret = -EINVAL;
956 	dlm_node_iter_init(mle->vote_map, &iter);
957 	while ((nodenum = dlm_node_iter_next(&iter)) >= 0) {
958 		ret = dlm_do_master_request(res, mle, nodenum);
959 		if (ret < 0)
960 			mlog_errno(ret);
961 		if (mle->master != O2NM_MAX_NODES) {
962 			/* found a master ! */
963 			if (mle->master <= nodenum)
964 				break;
965 			/* if our master request has not reached the master
966 			 * yet, keep going until it does.  this is how the
967 			 * master will know that asserts are needed back to
968 			 * the lower nodes. */
969 			mlog(0, "%s: res %.*s, Requests only up to %u but "
970 			     "master is %u, keep going\n", dlm->name, namelen,
971 			     lockid, nodenum, mle->master);
972 		}
973 	}
974 
975 wait:
976 	/* keep going until the response map includes all nodes */
977 	ret = dlm_wait_for_lock_mastery(dlm, res, mle, &blocked);
978 	if (ret < 0) {
979 		wait_on_recovery = 1;
980 		mlog(0, "%s: res %.*s, Node map changed, redo the master "
981 		     "request now, blocked=%d\n", dlm->name, res->lockname.len,
982 		     res->lockname.name, blocked);
983 		if (++tries > 20) {
984 			mlog(ML_ERROR, "%s: res %.*s, Spinning on "
985 			     "dlm_wait_for_lock_mastery, blocked = %d\n",
986 			     dlm->name, res->lockname.len,
987 			     res->lockname.name, blocked);
988 			dlm_print_one_lock_resource(res);
989 			dlm_print_one_mle(mle);
990 			tries = 0;
991 		}
992 		goto redo_request;
993 	}
994 
995 	mlog(0, "%s: res %.*s, Mastered by %u\n", dlm->name, res->lockname.len,
996 	     res->lockname.name, res->owner);
997 	/* make sure we never continue without this */
998 	BUG_ON(res->owner == O2NM_MAX_NODES);
999 
1000 	/* master is known, detach if not already detached */
1001 	dlm_mle_detach_hb_events(dlm, mle);
1002 	dlm_put_mle(mle);
1003 	/* put the extra ref */
1004 	dlm_put_mle_inuse(mle);
1005 
1006 wake_waiters:
1007 	spin_lock(&res->spinlock);
1008 	res->state &= ~DLM_LOCK_RES_IN_PROGRESS;
1009 	spin_unlock(&res->spinlock);
1010 	wake_up(&res->wq);
1011 
1012 leave:
1013 	/* need to free the unused mle */
1014 	if (alloc_mle)
1015 		kmem_cache_free(dlm_mle_cache, alloc_mle);
1016 
1017 	return res;
1018 }
1019 
1020 
1021 #define DLM_MASTERY_TIMEOUT_MS   5000
1022 
1023 static int dlm_wait_for_lock_mastery(struct dlm_ctxt *dlm,
1024 				     struct dlm_lock_resource *res,
1025 				     struct dlm_master_list_entry *mle,
1026 				     int *blocked)
1027 {
1028 	u8 m;
1029 	int ret, bit;
1030 	int map_changed, voting_done;
1031 	int assert, sleep;
1032 
1033 recheck:
1034 	ret = 0;
1035 	assert = 0;
1036 
1037 	/* check if another node has already become the owner */
1038 	spin_lock(&res->spinlock);
1039 	if (res->owner != DLM_LOCK_RES_OWNER_UNKNOWN) {
1040 		mlog(0, "%s:%.*s: owner is suddenly %u\n", dlm->name,
1041 		     res->lockname.len, res->lockname.name, res->owner);
1042 		spin_unlock(&res->spinlock);
1043 		/* this will cause the master to re-assert across
1044 		 * the whole cluster, freeing up mles */
1045 		if (res->owner != dlm->node_num) {
1046 			ret = dlm_do_master_request(res, mle, res->owner);
1047 			if (ret < 0) {
1048 				/* give recovery a chance to run */
1049 				mlog(ML_ERROR, "link to %u went down?: %d\n", res->owner, ret);
1050 				msleep(500);
1051 				goto recheck;
1052 			}
1053 		}
1054 		ret = 0;
1055 		goto leave;
1056 	}
1057 	spin_unlock(&res->spinlock);
1058 
1059 	spin_lock(&mle->spinlock);
1060 	m = mle->master;
1061 	map_changed = (memcmp(mle->vote_map, mle->node_map,
1062 			      sizeof(mle->vote_map)) != 0);
1063 	voting_done = (memcmp(mle->vote_map, mle->response_map,
1064 			     sizeof(mle->vote_map)) == 0);
1065 
1066 	/* restart if we hit any errors */
1067 	if (map_changed) {
1068 		int b;
1069 		mlog(0, "%s: %.*s: node map changed, restarting\n",
1070 		     dlm->name, res->lockname.len, res->lockname.name);
1071 		ret = dlm_restart_lock_mastery(dlm, res, mle, *blocked);
1072 		b = (mle->type == DLM_MLE_BLOCK);
1073 		if ((*blocked && !b) || (!*blocked && b)) {
1074 			mlog(0, "%s:%.*s: status change: old=%d new=%d\n",
1075 			     dlm->name, res->lockname.len, res->lockname.name,
1076 			     *blocked, b);
1077 			*blocked = b;
1078 		}
1079 		spin_unlock(&mle->spinlock);
1080 		if (ret < 0) {
1081 			mlog_errno(ret);
1082 			goto leave;
1083 		}
1084 		mlog(0, "%s:%.*s: restart lock mastery succeeded, "
1085 		     "rechecking now\n", dlm->name, res->lockname.len,
1086 		     res->lockname.name);
1087 		goto recheck;
1088 	} else {
1089 		if (!voting_done) {
1090 			mlog(0, "map not changed and voting not done "
1091 			     "for %s:%.*s\n", dlm->name, res->lockname.len,
1092 			     res->lockname.name);
1093 		}
1094 	}
1095 
1096 	if (m != O2NM_MAX_NODES) {
1097 		/* another node has done an assert!
1098 		 * all done! */
1099 		sleep = 0;
1100 	} else {
1101 		sleep = 1;
1102 		/* have all nodes responded? */
1103 		if (voting_done && !*blocked) {
1104 			bit = find_next_bit(mle->maybe_map, O2NM_MAX_NODES, 0);
1105 			if (dlm->node_num <= bit) {
1106 				/* my node number is lowest.
1107 			 	 * now tell other nodes that I am
1108 				 * mastering this. */
1109 				mle->master = dlm->node_num;
1110 				/* ref was grabbed in get_lock_resource
1111 				 * will be dropped in dlmlock_master */
1112 				assert = 1;
1113 				sleep = 0;
1114 			}
1115 			/* if voting is done, but we have not received
1116 			 * an assert master yet, we must sleep */
1117 		}
1118 	}
1119 
1120 	spin_unlock(&mle->spinlock);
1121 
1122 	/* sleep if we haven't finished voting yet */
1123 	if (sleep) {
1124 		unsigned long timeo = msecs_to_jiffies(DLM_MASTERY_TIMEOUT_MS);
1125 
1126 		/*
1127 		if (atomic_read(&mle->mle_refs.refcount) < 2)
1128 			mlog(ML_ERROR, "mle (%p) refs=%d, name=%.*s\n", mle,
1129 			atomic_read(&mle->mle_refs.refcount),
1130 			res->lockname.len, res->lockname.name);
1131 		*/
1132 		atomic_set(&mle->woken, 0);
1133 		(void)wait_event_timeout(mle->wq,
1134 					 (atomic_read(&mle->woken) == 1),
1135 					 timeo);
1136 		if (res->owner == O2NM_MAX_NODES) {
1137 			mlog(0, "%s:%.*s: waiting again\n", dlm->name,
1138 			     res->lockname.len, res->lockname.name);
1139 			goto recheck;
1140 		}
1141 		mlog(0, "done waiting, master is %u\n", res->owner);
1142 		ret = 0;
1143 		goto leave;
1144 	}
1145 
1146 	ret = 0;   /* done */
1147 	if (assert) {
1148 		m = dlm->node_num;
1149 		mlog(0, "about to master %.*s here, this=%u\n",
1150 		     res->lockname.len, res->lockname.name, m);
1151 		ret = dlm_do_assert_master(dlm, res, mle->vote_map, 0);
1152 		if (ret) {
1153 			/* This is a failure in the network path,
1154 			 * not in the response to the assert_master
1155 			 * (any nonzero response is a BUG on this node).
1156 			 * Most likely a socket just got disconnected
1157 			 * due to node death. */
1158 			mlog_errno(ret);
1159 		}
1160 		/* no longer need to restart lock mastery.
1161 		 * all living nodes have been contacted. */
1162 		ret = 0;
1163 	}
1164 
1165 	/* set the lockres owner */
1166 	spin_lock(&res->spinlock);
1167 	/* mastery reference obtained either during
1168 	 * assert_master_handler or in get_lock_resource */
1169 	dlm_change_lockres_owner(dlm, res, m);
1170 	spin_unlock(&res->spinlock);
1171 
1172 leave:
1173 	return ret;
1174 }
1175 
1176 struct dlm_bitmap_diff_iter
1177 {
1178 	int curnode;
1179 	unsigned long *orig_bm;
1180 	unsigned long *cur_bm;
1181 	unsigned long diff_bm[BITS_TO_LONGS(O2NM_MAX_NODES)];
1182 };
1183 
1184 enum dlm_node_state_change
1185 {
1186 	NODE_DOWN = -1,
1187 	NODE_NO_CHANGE = 0,
1188 	NODE_UP
1189 };
1190 
1191 static void dlm_bitmap_diff_iter_init(struct dlm_bitmap_diff_iter *iter,
1192 				      unsigned long *orig_bm,
1193 				      unsigned long *cur_bm)
1194 {
1195 	unsigned long p1, p2;
1196 	int i;
1197 
1198 	iter->curnode = -1;
1199 	iter->orig_bm = orig_bm;
1200 	iter->cur_bm = cur_bm;
1201 
1202 	for (i = 0; i < BITS_TO_LONGS(O2NM_MAX_NODES); i++) {
1203        		p1 = *(iter->orig_bm + i);
1204 	       	p2 = *(iter->cur_bm + i);
1205 		iter->diff_bm[i] = (p1 & ~p2) | (p2 & ~p1);
1206 	}
1207 }
1208 
1209 static int dlm_bitmap_diff_iter_next(struct dlm_bitmap_diff_iter *iter,
1210 				     enum dlm_node_state_change *state)
1211 {
1212 	int bit;
1213 
1214 	if (iter->curnode >= O2NM_MAX_NODES)
1215 		return -ENOENT;
1216 
1217 	bit = find_next_bit(iter->diff_bm, O2NM_MAX_NODES,
1218 			    iter->curnode+1);
1219 	if (bit >= O2NM_MAX_NODES) {
1220 		iter->curnode = O2NM_MAX_NODES;
1221 		return -ENOENT;
1222 	}
1223 
1224 	/* if it was there in the original then this node died */
1225 	if (test_bit(bit, iter->orig_bm))
1226 		*state = NODE_DOWN;
1227 	else
1228 		*state = NODE_UP;
1229 
1230 	iter->curnode = bit;
1231 	return bit;
1232 }
1233 
1234 
1235 static int dlm_restart_lock_mastery(struct dlm_ctxt *dlm,
1236 				    struct dlm_lock_resource *res,
1237 				    struct dlm_master_list_entry *mle,
1238 				    int blocked)
1239 {
1240 	struct dlm_bitmap_diff_iter bdi;
1241 	enum dlm_node_state_change sc;
1242 	int node;
1243 	int ret = 0;
1244 
1245 	mlog(0, "something happened such that the "
1246 	     "master process may need to be restarted!\n");
1247 
1248 	assert_spin_locked(&mle->spinlock);
1249 
1250 	dlm_bitmap_diff_iter_init(&bdi, mle->vote_map, mle->node_map);
1251 	node = dlm_bitmap_diff_iter_next(&bdi, &sc);
1252 	while (node >= 0) {
1253 		if (sc == NODE_UP) {
1254 			/* a node came up.  clear any old vote from
1255 			 * the response map and set it in the vote map
1256 			 * then restart the mastery. */
1257 			mlog(ML_NOTICE, "node %d up while restarting\n", node);
1258 
1259 			/* redo the master request, but only for the new node */
1260 			mlog(0, "sending request to new node\n");
1261 			clear_bit(node, mle->response_map);
1262 			set_bit(node, mle->vote_map);
1263 		} else {
1264 			mlog(ML_ERROR, "node down! %d\n", node);
1265 			if (blocked) {
1266 				int lowest = find_next_bit(mle->maybe_map,
1267 						       O2NM_MAX_NODES, 0);
1268 
1269 				/* act like it was never there */
1270 				clear_bit(node, mle->maybe_map);
1271 
1272 			       	if (node == lowest) {
1273 					mlog(0, "expected master %u died"
1274 					    " while this node was blocked "
1275 					    "waiting on it!\n", node);
1276 					lowest = find_next_bit(mle->maybe_map,
1277 						       	O2NM_MAX_NODES,
1278 						       	lowest+1);
1279 					if (lowest < O2NM_MAX_NODES) {
1280 						mlog(0, "%s:%.*s:still "
1281 						     "blocked. waiting on %u "
1282 						     "now\n", dlm->name,
1283 						     res->lockname.len,
1284 						     res->lockname.name,
1285 						     lowest);
1286 					} else {
1287 						/* mle is an MLE_BLOCK, but
1288 						 * there is now nothing left to
1289 						 * block on.  we need to return
1290 						 * all the way back out and try
1291 						 * again with an MLE_MASTER.
1292 						 * dlm_do_local_recovery_cleanup
1293 						 * has already run, so the mle
1294 						 * refcount is ok */
1295 						mlog(0, "%s:%.*s: no "
1296 						     "longer blocking. try to "
1297 						     "master this here\n",
1298 						     dlm->name,
1299 						     res->lockname.len,
1300 						     res->lockname.name);
1301 						mle->type = DLM_MLE_MASTER;
1302 						mle->mleres = res;
1303 					}
1304 				}
1305 			}
1306 
1307 			/* now blank out everything, as if we had never
1308 			 * contacted anyone */
1309 			memset(mle->maybe_map, 0, sizeof(mle->maybe_map));
1310 			memset(mle->response_map, 0, sizeof(mle->response_map));
1311 			/* reset the vote_map to the current node_map */
1312 			memcpy(mle->vote_map, mle->node_map,
1313 			       sizeof(mle->node_map));
1314 			/* put myself into the maybe map */
1315 			if (mle->type != DLM_MLE_BLOCK)
1316 				set_bit(dlm->node_num, mle->maybe_map);
1317 		}
1318 		ret = -EAGAIN;
1319 		node = dlm_bitmap_diff_iter_next(&bdi, &sc);
1320 	}
1321 	return ret;
1322 }
1323 
1324 
1325 /*
1326  * DLM_MASTER_REQUEST_MSG
1327  *
1328  * returns: 0 on success,
1329  *          -errno on a network error
1330  *
1331  * on error, the caller should assume the target node is "dead"
1332  *
1333  */
1334 
1335 static int dlm_do_master_request(struct dlm_lock_resource *res,
1336 				 struct dlm_master_list_entry *mle, int to)
1337 {
1338 	struct dlm_ctxt *dlm = mle->dlm;
1339 	struct dlm_master_request request;
1340 	int ret, response=0, resend;
1341 
1342 	memset(&request, 0, sizeof(request));
1343 	request.node_idx = dlm->node_num;
1344 
1345 	BUG_ON(mle->type == DLM_MLE_MIGRATION);
1346 
1347 	request.namelen = (u8)mle->mnamelen;
1348 	memcpy(request.name, mle->mname, request.namelen);
1349 
1350 again:
1351 	ret = o2net_send_message(DLM_MASTER_REQUEST_MSG, dlm->key, &request,
1352 				 sizeof(request), to, &response);
1353 	if (ret < 0)  {
1354 		if (ret == -ESRCH) {
1355 			/* should never happen */
1356 			mlog(ML_ERROR, "TCP stack not ready!\n");
1357 			BUG();
1358 		} else if (ret == -EINVAL) {
1359 			mlog(ML_ERROR, "bad args passed to o2net!\n");
1360 			BUG();
1361 		} else if (ret == -ENOMEM) {
1362 			mlog(ML_ERROR, "out of memory while trying to send "
1363 			     "network message!  retrying\n");
1364 			/* this is totally crude */
1365 			msleep(50);
1366 			goto again;
1367 		} else if (!dlm_is_host_down(ret)) {
1368 			/* not a network error. bad. */
1369 			mlog_errno(ret);
1370 			mlog(ML_ERROR, "unhandled error!");
1371 			BUG();
1372 		}
1373 		/* all other errors should be network errors,
1374 		 * and likely indicate node death */
1375 		mlog(ML_ERROR, "link to %d went down!\n", to);
1376 		goto out;
1377 	}
1378 
1379 	ret = 0;
1380 	resend = 0;
1381 	spin_lock(&mle->spinlock);
1382 	switch (response) {
1383 		case DLM_MASTER_RESP_YES:
1384 			set_bit(to, mle->response_map);
1385 			mlog(0, "node %u is the master, response=YES\n", to);
1386 			mlog(0, "%s:%.*s: master node %u now knows I have a "
1387 			     "reference\n", dlm->name, res->lockname.len,
1388 			     res->lockname.name, to);
1389 			mle->master = to;
1390 			break;
1391 		case DLM_MASTER_RESP_NO:
1392 			mlog(0, "node %u not master, response=NO\n", to);
1393 			set_bit(to, mle->response_map);
1394 			break;
1395 		case DLM_MASTER_RESP_MAYBE:
1396 			mlog(0, "node %u not master, response=MAYBE\n", to);
1397 			set_bit(to, mle->response_map);
1398 			set_bit(to, mle->maybe_map);
1399 			break;
1400 		case DLM_MASTER_RESP_ERROR:
1401 			mlog(0, "node %u hit an error, resending\n", to);
1402 			resend = 1;
1403 			response = 0;
1404 			break;
1405 		default:
1406 			mlog(ML_ERROR, "bad response! %u\n", response);
1407 			BUG();
1408 	}
1409 	spin_unlock(&mle->spinlock);
1410 	if (resend) {
1411 		/* this is also totally crude */
1412 		msleep(50);
1413 		goto again;
1414 	}
1415 
1416 out:
1417 	return ret;
1418 }
1419 
1420 /*
1421  * locks that can be taken here:
1422  * dlm->spinlock
1423  * res->spinlock
1424  * mle->spinlock
1425  * dlm->master_list
1426  *
1427  * if possible, TRIM THIS DOWN!!!
1428  */
1429 int dlm_master_request_handler(struct o2net_msg *msg, u32 len, void *data,
1430 			       void **ret_data)
1431 {
1432 	u8 response = DLM_MASTER_RESP_MAYBE;
1433 	struct dlm_ctxt *dlm = data;
1434 	struct dlm_lock_resource *res = NULL;
1435 	struct dlm_master_request *request = (struct dlm_master_request *) msg->buf;
1436 	struct dlm_master_list_entry *mle = NULL, *tmpmle = NULL;
1437 	char *name;
1438 	unsigned int namelen, hash;
1439 	int found, ret;
1440 	int set_maybe;
1441 	int dispatch_assert = 0;
1442 
1443 	if (!dlm_grab(dlm))
1444 		return DLM_MASTER_RESP_NO;
1445 
1446 	if (!dlm_domain_fully_joined(dlm)) {
1447 		response = DLM_MASTER_RESP_NO;
1448 		goto send_response;
1449 	}
1450 
1451 	name = request->name;
1452 	namelen = request->namelen;
1453 	hash = dlm_lockid_hash(name, namelen);
1454 
1455 	if (namelen > DLM_LOCKID_NAME_MAX) {
1456 		response = DLM_IVBUFLEN;
1457 		goto send_response;
1458 	}
1459 
1460 way_up_top:
1461 	spin_lock(&dlm->spinlock);
1462 	res = __dlm_lookup_lockres(dlm, name, namelen, hash);
1463 	if (res) {
1464 		spin_unlock(&dlm->spinlock);
1465 
1466 		/* take care of the easy cases up front */
1467 		spin_lock(&res->spinlock);
1468 
1469 		/*
1470 		 * Right after dlm spinlock was released, dlm_thread could have
1471 		 * purged the lockres. Check if lockres got unhashed. If so
1472 		 * start over.
1473 		 */
1474 		if (hlist_unhashed(&res->hash_node)) {
1475 			spin_unlock(&res->spinlock);
1476 			dlm_lockres_put(res);
1477 			goto way_up_top;
1478 		}
1479 
1480 		if (res->state & (DLM_LOCK_RES_RECOVERING|
1481 				  DLM_LOCK_RES_MIGRATING)) {
1482 			spin_unlock(&res->spinlock);
1483 			mlog(0, "returning DLM_MASTER_RESP_ERROR since res is "
1484 			     "being recovered/migrated\n");
1485 			response = DLM_MASTER_RESP_ERROR;
1486 			if (mle)
1487 				kmem_cache_free(dlm_mle_cache, mle);
1488 			goto send_response;
1489 		}
1490 
1491 		if (res->owner == dlm->node_num) {
1492 			dlm_lockres_set_refmap_bit(dlm, res, request->node_idx);
1493 			spin_unlock(&res->spinlock);
1494 			response = DLM_MASTER_RESP_YES;
1495 			if (mle)
1496 				kmem_cache_free(dlm_mle_cache, mle);
1497 
1498 			/* this node is the owner.
1499 			 * there is some extra work that needs to
1500 			 * happen now.  the requesting node has
1501 			 * caused all nodes up to this one to
1502 			 * create mles.  this node now needs to
1503 			 * go back and clean those up. */
1504 			dispatch_assert = 1;
1505 			goto send_response;
1506 		} else if (res->owner != DLM_LOCK_RES_OWNER_UNKNOWN) {
1507 			spin_unlock(&res->spinlock);
1508 			// mlog(0, "node %u is the master\n", res->owner);
1509 			response = DLM_MASTER_RESP_NO;
1510 			if (mle)
1511 				kmem_cache_free(dlm_mle_cache, mle);
1512 			goto send_response;
1513 		}
1514 
1515 		/* ok, there is no owner.  either this node is
1516 		 * being blocked, or it is actively trying to
1517 		 * master this lock. */
1518 		if (!(res->state & DLM_LOCK_RES_IN_PROGRESS)) {
1519 			mlog(ML_ERROR, "lock with no owner should be "
1520 			     "in-progress!\n");
1521 			BUG();
1522 		}
1523 
1524 		// mlog(0, "lockres is in progress...\n");
1525 		spin_lock(&dlm->master_lock);
1526 		found = dlm_find_mle(dlm, &tmpmle, name, namelen);
1527 		if (!found) {
1528 			mlog(ML_ERROR, "no mle found for this lock!\n");
1529 			BUG();
1530 		}
1531 		set_maybe = 1;
1532 		spin_lock(&tmpmle->spinlock);
1533 		if (tmpmle->type == DLM_MLE_BLOCK) {
1534 			// mlog(0, "this node is waiting for "
1535 			// "lockres to be mastered\n");
1536 			response = DLM_MASTER_RESP_NO;
1537 		} else if (tmpmle->type == DLM_MLE_MIGRATION) {
1538 			mlog(0, "node %u is master, but trying to migrate to "
1539 			     "node %u.\n", tmpmle->master, tmpmle->new_master);
1540 			if (tmpmle->master == dlm->node_num) {
1541 				mlog(ML_ERROR, "no owner on lockres, but this "
1542 				     "node is trying to migrate it to %u?!\n",
1543 				     tmpmle->new_master);
1544 				BUG();
1545 			} else {
1546 				/* the real master can respond on its own */
1547 				response = DLM_MASTER_RESP_NO;
1548 			}
1549 		} else if (tmpmle->master != DLM_LOCK_RES_OWNER_UNKNOWN) {
1550 			set_maybe = 0;
1551 			if (tmpmle->master == dlm->node_num) {
1552 				response = DLM_MASTER_RESP_YES;
1553 				/* this node will be the owner.
1554 				 * go back and clean the mles on any
1555 				 * other nodes */
1556 				dispatch_assert = 1;
1557 				dlm_lockres_set_refmap_bit(dlm, res,
1558 							   request->node_idx);
1559 			} else
1560 				response = DLM_MASTER_RESP_NO;
1561 		} else {
1562 			// mlog(0, "this node is attempting to "
1563 			// "master lockres\n");
1564 			response = DLM_MASTER_RESP_MAYBE;
1565 		}
1566 		if (set_maybe)
1567 			set_bit(request->node_idx, tmpmle->maybe_map);
1568 		spin_unlock(&tmpmle->spinlock);
1569 
1570 		spin_unlock(&dlm->master_lock);
1571 		spin_unlock(&res->spinlock);
1572 
1573 		/* keep the mle attached to heartbeat events */
1574 		dlm_put_mle(tmpmle);
1575 		if (mle)
1576 			kmem_cache_free(dlm_mle_cache, mle);
1577 		goto send_response;
1578 	}
1579 
1580 	/*
1581 	 * lockres doesn't exist on this node
1582 	 * if there is an MLE_BLOCK, return NO
1583 	 * if there is an MLE_MASTER, return MAYBE
1584 	 * otherwise, add an MLE_BLOCK, return NO
1585 	 */
1586 	spin_lock(&dlm->master_lock);
1587 	found = dlm_find_mle(dlm, &tmpmle, name, namelen);
1588 	if (!found) {
1589 		/* this lockid has never been seen on this node yet */
1590 		// mlog(0, "no mle found\n");
1591 		if (!mle) {
1592 			spin_unlock(&dlm->master_lock);
1593 			spin_unlock(&dlm->spinlock);
1594 
1595 			mle = kmem_cache_alloc(dlm_mle_cache, GFP_NOFS);
1596 			if (!mle) {
1597 				response = DLM_MASTER_RESP_ERROR;
1598 				mlog_errno(-ENOMEM);
1599 				goto send_response;
1600 			}
1601 			goto way_up_top;
1602 		}
1603 
1604 		// mlog(0, "this is second time thru, already allocated, "
1605 		// "add the block.\n");
1606 		dlm_init_mle(mle, DLM_MLE_BLOCK, dlm, NULL, name, namelen);
1607 		set_bit(request->node_idx, mle->maybe_map);
1608 		__dlm_insert_mle(dlm, mle);
1609 		response = DLM_MASTER_RESP_NO;
1610 	} else {
1611 		// mlog(0, "mle was found\n");
1612 		set_maybe = 1;
1613 		spin_lock(&tmpmle->spinlock);
1614 		if (tmpmle->master == dlm->node_num) {
1615 			mlog(ML_ERROR, "no lockres, but an mle with this node as master!\n");
1616 			BUG();
1617 		}
1618 		if (tmpmle->type == DLM_MLE_BLOCK)
1619 			response = DLM_MASTER_RESP_NO;
1620 		else if (tmpmle->type == DLM_MLE_MIGRATION) {
1621 			mlog(0, "migration mle was found (%u->%u)\n",
1622 			     tmpmle->master, tmpmle->new_master);
1623 			/* real master can respond on its own */
1624 			response = DLM_MASTER_RESP_NO;
1625 		} else
1626 			response = DLM_MASTER_RESP_MAYBE;
1627 		if (set_maybe)
1628 			set_bit(request->node_idx, tmpmle->maybe_map);
1629 		spin_unlock(&tmpmle->spinlock);
1630 	}
1631 	spin_unlock(&dlm->master_lock);
1632 	spin_unlock(&dlm->spinlock);
1633 
1634 	if (found) {
1635 		/* keep the mle attached to heartbeat events */
1636 		dlm_put_mle(tmpmle);
1637 	}
1638 send_response:
1639 	/*
1640 	 * __dlm_lookup_lockres() grabbed a reference to this lockres.
1641 	 * The reference is released by dlm_assert_master_worker() under
1642 	 * the call to dlm_dispatch_assert_master().  If
1643 	 * dlm_assert_master_worker() isn't called, we drop it here.
1644 	 */
1645 	if (dispatch_assert) {
1646 		if (response != DLM_MASTER_RESP_YES)
1647 			mlog(ML_ERROR, "invalid response %d\n", response);
1648 		if (!res) {
1649 			mlog(ML_ERROR, "bad lockres while trying to assert!\n");
1650 			BUG();
1651 		}
1652 		mlog(0, "%u is the owner of %.*s, cleaning everyone else\n",
1653 			     dlm->node_num, res->lockname.len, res->lockname.name);
1654 		spin_lock(&res->spinlock);
1655 		ret = dlm_dispatch_assert_master(dlm, res, 0, request->node_idx,
1656 						 DLM_ASSERT_MASTER_MLE_CLEANUP);
1657 		if (ret < 0) {
1658 			mlog(ML_ERROR, "failed to dispatch assert master work\n");
1659 			response = DLM_MASTER_RESP_ERROR;
1660 			dlm_lockres_put(res);
1661 		} else
1662 			__dlm_lockres_grab_inflight_worker(dlm, res);
1663 		spin_unlock(&res->spinlock);
1664 	} else {
1665 		if (res)
1666 			dlm_lockres_put(res);
1667 	}
1668 
1669 	dlm_put(dlm);
1670 	return response;
1671 }
1672 
1673 /*
1674  * DLM_ASSERT_MASTER_MSG
1675  */
1676 
1677 
1678 /*
1679  * NOTE: this can be used for debugging
1680  * can periodically run all locks owned by this node
1681  * and re-assert across the cluster...
1682  */
1683 static int dlm_do_assert_master(struct dlm_ctxt *dlm,
1684 				struct dlm_lock_resource *res,
1685 				void *nodemap, u32 flags)
1686 {
1687 	struct dlm_assert_master assert;
1688 	int to, tmpret;
1689 	struct dlm_node_iter iter;
1690 	int ret = 0;
1691 	int reassert;
1692 	const char *lockname = res->lockname.name;
1693 	unsigned int namelen = res->lockname.len;
1694 
1695 	BUG_ON(namelen > O2NM_MAX_NAME_LEN);
1696 
1697 	spin_lock(&res->spinlock);
1698 	res->state |= DLM_LOCK_RES_SETREF_INPROG;
1699 	spin_unlock(&res->spinlock);
1700 
1701 again:
1702 	reassert = 0;
1703 
1704 	/* note that if this nodemap is empty, it returns 0 */
1705 	dlm_node_iter_init(nodemap, &iter);
1706 	while ((to = dlm_node_iter_next(&iter)) >= 0) {
1707 		int r = 0;
1708 		struct dlm_master_list_entry *mle = NULL;
1709 
1710 		mlog(0, "sending assert master to %d (%.*s)\n", to,
1711 		     namelen, lockname);
1712 		memset(&assert, 0, sizeof(assert));
1713 		assert.node_idx = dlm->node_num;
1714 		assert.namelen = namelen;
1715 		memcpy(assert.name, lockname, namelen);
1716 		assert.flags = cpu_to_be32(flags);
1717 
1718 		tmpret = o2net_send_message(DLM_ASSERT_MASTER_MSG, dlm->key,
1719 					    &assert, sizeof(assert), to, &r);
1720 		if (tmpret < 0) {
1721 			mlog(ML_ERROR, "Error %d when sending message %u (key "
1722 			     "0x%x) to node %u\n", tmpret,
1723 			     DLM_ASSERT_MASTER_MSG, dlm->key, to);
1724 			if (!dlm_is_host_down(tmpret)) {
1725 				mlog(ML_ERROR, "unhandled error=%d!\n", tmpret);
1726 				BUG();
1727 			}
1728 			/* a node died.  finish out the rest of the nodes. */
1729 			mlog(0, "link to %d went down!\n", to);
1730 			/* any nonzero status return will do */
1731 			ret = tmpret;
1732 			r = 0;
1733 		} else if (r < 0) {
1734 			/* ok, something horribly messed.  kill thyself. */
1735 			mlog(ML_ERROR,"during assert master of %.*s to %u, "
1736 			     "got %d.\n", namelen, lockname, to, r);
1737 			spin_lock(&dlm->spinlock);
1738 			spin_lock(&dlm->master_lock);
1739 			if (dlm_find_mle(dlm, &mle, (char *)lockname,
1740 					 namelen)) {
1741 				dlm_print_one_mle(mle);
1742 				__dlm_put_mle(mle);
1743 			}
1744 			spin_unlock(&dlm->master_lock);
1745 			spin_unlock(&dlm->spinlock);
1746 			BUG();
1747 		}
1748 
1749 		if (r & DLM_ASSERT_RESPONSE_REASSERT &&
1750 		    !(r & DLM_ASSERT_RESPONSE_MASTERY_REF)) {
1751 				mlog(ML_ERROR, "%.*s: very strange, "
1752 				     "master MLE but no lockres on %u\n",
1753 				     namelen, lockname, to);
1754 		}
1755 
1756 		if (r & DLM_ASSERT_RESPONSE_REASSERT) {
1757 			mlog(0, "%.*s: node %u create mles on other "
1758 			     "nodes and requests a re-assert\n",
1759 			     namelen, lockname, to);
1760 			reassert = 1;
1761 		}
1762 		if (r & DLM_ASSERT_RESPONSE_MASTERY_REF) {
1763 			mlog(0, "%.*s: node %u has a reference to this "
1764 			     "lockres, set the bit in the refmap\n",
1765 			     namelen, lockname, to);
1766 			spin_lock(&res->spinlock);
1767 			dlm_lockres_set_refmap_bit(dlm, res, to);
1768 			spin_unlock(&res->spinlock);
1769 		}
1770 	}
1771 
1772 	if (reassert)
1773 		goto again;
1774 
1775 	spin_lock(&res->spinlock);
1776 	res->state &= ~DLM_LOCK_RES_SETREF_INPROG;
1777 	spin_unlock(&res->spinlock);
1778 	wake_up(&res->wq);
1779 
1780 	return ret;
1781 }
1782 
1783 /*
1784  * locks that can be taken here:
1785  * dlm->spinlock
1786  * res->spinlock
1787  * mle->spinlock
1788  * dlm->master_list
1789  *
1790  * if possible, TRIM THIS DOWN!!!
1791  */
1792 int dlm_assert_master_handler(struct o2net_msg *msg, u32 len, void *data,
1793 			      void **ret_data)
1794 {
1795 	struct dlm_ctxt *dlm = data;
1796 	struct dlm_master_list_entry *mle = NULL;
1797 	struct dlm_assert_master *assert = (struct dlm_assert_master *)msg->buf;
1798 	struct dlm_lock_resource *res = NULL;
1799 	char *name;
1800 	unsigned int namelen, hash;
1801 	u32 flags;
1802 	int master_request = 0, have_lockres_ref = 0;
1803 	int ret = 0;
1804 
1805 	if (!dlm_grab(dlm))
1806 		return 0;
1807 
1808 	name = assert->name;
1809 	namelen = assert->namelen;
1810 	hash = dlm_lockid_hash(name, namelen);
1811 	flags = be32_to_cpu(assert->flags);
1812 
1813 	if (namelen > DLM_LOCKID_NAME_MAX) {
1814 		mlog(ML_ERROR, "Invalid name length!");
1815 		goto done;
1816 	}
1817 
1818 	spin_lock(&dlm->spinlock);
1819 
1820 	if (flags)
1821 		mlog(0, "assert_master with flags: %u\n", flags);
1822 
1823 	/* find the MLE */
1824 	spin_lock(&dlm->master_lock);
1825 	if (!dlm_find_mle(dlm, &mle, name, namelen)) {
1826 		/* not an error, could be master just re-asserting */
1827 		mlog(0, "just got an assert_master from %u, but no "
1828 		     "MLE for it! (%.*s)\n", assert->node_idx,
1829 		     namelen, name);
1830 	} else {
1831 		int bit = find_next_bit (mle->maybe_map, O2NM_MAX_NODES, 0);
1832 		if (bit >= O2NM_MAX_NODES) {
1833 			/* not necessarily an error, though less likely.
1834 			 * could be master just re-asserting. */
1835 			mlog(0, "no bits set in the maybe_map, but %u "
1836 			     "is asserting! (%.*s)\n", assert->node_idx,
1837 			     namelen, name);
1838 		} else if (bit != assert->node_idx) {
1839 			if (flags & DLM_ASSERT_MASTER_MLE_CLEANUP) {
1840 				mlog(0, "master %u was found, %u should "
1841 				     "back off\n", assert->node_idx, bit);
1842 			} else {
1843 				/* with the fix for bug 569, a higher node
1844 				 * number winning the mastery will respond
1845 				 * YES to mastery requests, but this node
1846 				 * had no way of knowing.  let it pass. */
1847 				mlog(0, "%u is the lowest node, "
1848 				     "%u is asserting. (%.*s)  %u must "
1849 				     "have begun after %u won.\n", bit,
1850 				     assert->node_idx, namelen, name, bit,
1851 				     assert->node_idx);
1852 			}
1853 		}
1854 		if (mle->type == DLM_MLE_MIGRATION) {
1855 			if (flags & DLM_ASSERT_MASTER_MLE_CLEANUP) {
1856 				mlog(0, "%s:%.*s: got cleanup assert"
1857 				     " from %u for migration\n",
1858 				     dlm->name, namelen, name,
1859 				     assert->node_idx);
1860 			} else if (!(flags & DLM_ASSERT_MASTER_FINISH_MIGRATION)) {
1861 				mlog(0, "%s:%.*s: got unrelated assert"
1862 				     " from %u for migration, ignoring\n",
1863 				     dlm->name, namelen, name,
1864 				     assert->node_idx);
1865 				__dlm_put_mle(mle);
1866 				spin_unlock(&dlm->master_lock);
1867 				spin_unlock(&dlm->spinlock);
1868 				goto done;
1869 			}
1870 		}
1871 	}
1872 	spin_unlock(&dlm->master_lock);
1873 
1874 	/* ok everything checks out with the MLE
1875 	 * now check to see if there is a lockres */
1876 	res = __dlm_lookup_lockres(dlm, name, namelen, hash);
1877 	if (res) {
1878 		spin_lock(&res->spinlock);
1879 		if (res->state & DLM_LOCK_RES_RECOVERING)  {
1880 			mlog(ML_ERROR, "%u asserting but %.*s is "
1881 			     "RECOVERING!\n", assert->node_idx, namelen, name);
1882 			goto kill;
1883 		}
1884 		if (!mle) {
1885 			if (res->owner != DLM_LOCK_RES_OWNER_UNKNOWN &&
1886 			    res->owner != assert->node_idx) {
1887 				mlog(ML_ERROR, "DIE! Mastery assert from %u, "
1888 				     "but current owner is %u! (%.*s)\n",
1889 				     assert->node_idx, res->owner, namelen,
1890 				     name);
1891 				__dlm_print_one_lock_resource(res);
1892 				BUG();
1893 			}
1894 		} else if (mle->type != DLM_MLE_MIGRATION) {
1895 			if (res->owner != DLM_LOCK_RES_OWNER_UNKNOWN) {
1896 				/* owner is just re-asserting */
1897 				if (res->owner == assert->node_idx) {
1898 					mlog(0, "owner %u re-asserting on "
1899 					     "lock %.*s\n", assert->node_idx,
1900 					     namelen, name);
1901 					goto ok;
1902 				}
1903 				mlog(ML_ERROR, "got assert_master from "
1904 				     "node %u, but %u is the owner! "
1905 				     "(%.*s)\n", assert->node_idx,
1906 				     res->owner, namelen, name);
1907 				goto kill;
1908 			}
1909 			if (!(res->state & DLM_LOCK_RES_IN_PROGRESS)) {
1910 				mlog(ML_ERROR, "got assert from %u, but lock "
1911 				     "with no owner should be "
1912 				     "in-progress! (%.*s)\n",
1913 				     assert->node_idx,
1914 				     namelen, name);
1915 				goto kill;
1916 			}
1917 		} else /* mle->type == DLM_MLE_MIGRATION */ {
1918 			/* should only be getting an assert from new master */
1919 			if (assert->node_idx != mle->new_master) {
1920 				mlog(ML_ERROR, "got assert from %u, but "
1921 				     "new master is %u, and old master "
1922 				     "was %u (%.*s)\n",
1923 				     assert->node_idx, mle->new_master,
1924 				     mle->master, namelen, name);
1925 				goto kill;
1926 			}
1927 
1928 		}
1929 ok:
1930 		spin_unlock(&res->spinlock);
1931 	}
1932 
1933 	// mlog(0, "woo!  got an assert_master from node %u!\n",
1934 	// 	     assert->node_idx);
1935 	if (mle) {
1936 		int extra_ref = 0;
1937 		int nn = -1;
1938 		int rr, err = 0;
1939 
1940 		spin_lock(&mle->spinlock);
1941 		if (mle->type == DLM_MLE_BLOCK || mle->type == DLM_MLE_MIGRATION)
1942 			extra_ref = 1;
1943 		else {
1944 			/* MASTER mle: if any bits set in the response map
1945 			 * then the calling node needs to re-assert to clear
1946 			 * up nodes that this node contacted */
1947 			while ((nn = find_next_bit (mle->response_map, O2NM_MAX_NODES,
1948 						    nn+1)) < O2NM_MAX_NODES) {
1949 				if (nn != dlm->node_num && nn != assert->node_idx) {
1950 					master_request = 1;
1951 					break;
1952 				}
1953 			}
1954 		}
1955 		mle->master = assert->node_idx;
1956 		atomic_set(&mle->woken, 1);
1957 		wake_up(&mle->wq);
1958 		spin_unlock(&mle->spinlock);
1959 
1960 		if (res) {
1961 			int wake = 0;
1962 			spin_lock(&res->spinlock);
1963 			if (mle->type == DLM_MLE_MIGRATION) {
1964 				mlog(0, "finishing off migration of lockres %.*s, "
1965 			     		"from %u to %u\n",
1966 			       		res->lockname.len, res->lockname.name,
1967 			       		dlm->node_num, mle->new_master);
1968 				res->state &= ~DLM_LOCK_RES_MIGRATING;
1969 				wake = 1;
1970 				dlm_change_lockres_owner(dlm, res, mle->new_master);
1971 				BUG_ON(res->state & DLM_LOCK_RES_DIRTY);
1972 			} else {
1973 				dlm_change_lockres_owner(dlm, res, mle->master);
1974 			}
1975 			spin_unlock(&res->spinlock);
1976 			have_lockres_ref = 1;
1977 			if (wake)
1978 				wake_up(&res->wq);
1979 		}
1980 
1981 		/* master is known, detach if not already detached.
1982 		 * ensures that only one assert_master call will happen
1983 		 * on this mle. */
1984 		spin_lock(&dlm->master_lock);
1985 
1986 		rr = atomic_read(&mle->mle_refs.refcount);
1987 		if (mle->inuse > 0) {
1988 			if (extra_ref && rr < 3)
1989 				err = 1;
1990 			else if (!extra_ref && rr < 2)
1991 				err = 1;
1992 		} else {
1993 			if (extra_ref && rr < 2)
1994 				err = 1;
1995 			else if (!extra_ref && rr < 1)
1996 				err = 1;
1997 		}
1998 		if (err) {
1999 			mlog(ML_ERROR, "%s:%.*s: got assert master from %u "
2000 			     "that will mess up this node, refs=%d, extra=%d, "
2001 			     "inuse=%d\n", dlm->name, namelen, name,
2002 			     assert->node_idx, rr, extra_ref, mle->inuse);
2003 			dlm_print_one_mle(mle);
2004 		}
2005 		__dlm_unlink_mle(dlm, mle);
2006 		__dlm_mle_detach_hb_events(dlm, mle);
2007 		__dlm_put_mle(mle);
2008 		if (extra_ref) {
2009 			/* the assert master message now balances the extra
2010 		 	 * ref given by the master / migration request message.
2011 		 	 * if this is the last put, it will be removed
2012 		 	 * from the list. */
2013 			__dlm_put_mle(mle);
2014 		}
2015 		spin_unlock(&dlm->master_lock);
2016 	} else if (res) {
2017 		if (res->owner != assert->node_idx) {
2018 			mlog(0, "assert_master from %u, but current "
2019 			     "owner is %u (%.*s), no mle\n", assert->node_idx,
2020 			     res->owner, namelen, name);
2021 		}
2022 	}
2023 	spin_unlock(&dlm->spinlock);
2024 
2025 done:
2026 	ret = 0;
2027 	if (res) {
2028 		spin_lock(&res->spinlock);
2029 		res->state |= DLM_LOCK_RES_SETREF_INPROG;
2030 		spin_unlock(&res->spinlock);
2031 		*ret_data = (void *)res;
2032 	}
2033 	dlm_put(dlm);
2034 	if (master_request) {
2035 		mlog(0, "need to tell master to reassert\n");
2036 		/* positive. negative would shoot down the node. */
2037 		ret |= DLM_ASSERT_RESPONSE_REASSERT;
2038 		if (!have_lockres_ref) {
2039 			mlog(ML_ERROR, "strange, got assert from %u, MASTER "
2040 			     "mle present here for %s:%.*s, but no lockres!\n",
2041 			     assert->node_idx, dlm->name, namelen, name);
2042 		}
2043 	}
2044 	if (have_lockres_ref) {
2045 		/* let the master know we have a reference to the lockres */
2046 		ret |= DLM_ASSERT_RESPONSE_MASTERY_REF;
2047 		mlog(0, "%s:%.*s: got assert from %u, need a ref\n",
2048 		     dlm->name, namelen, name, assert->node_idx);
2049 	}
2050 	return ret;
2051 
2052 kill:
2053 	/* kill the caller! */
2054 	mlog(ML_ERROR, "Bad message received from another node.  Dumping state "
2055 	     "and killing the other node now!  This node is OK and can continue.\n");
2056 	__dlm_print_one_lock_resource(res);
2057 	spin_unlock(&res->spinlock);
2058 	spin_lock(&dlm->master_lock);
2059 	if (mle)
2060 		__dlm_put_mle(mle);
2061 	spin_unlock(&dlm->master_lock);
2062 	spin_unlock(&dlm->spinlock);
2063 	*ret_data = (void *)res;
2064 	dlm_put(dlm);
2065 	return -EINVAL;
2066 }
2067 
2068 void dlm_assert_master_post_handler(int status, void *data, void *ret_data)
2069 {
2070 	struct dlm_lock_resource *res = (struct dlm_lock_resource *)ret_data;
2071 
2072 	if (ret_data) {
2073 		spin_lock(&res->spinlock);
2074 		res->state &= ~DLM_LOCK_RES_SETREF_INPROG;
2075 		spin_unlock(&res->spinlock);
2076 		wake_up(&res->wq);
2077 		dlm_lockres_put(res);
2078 	}
2079 	return;
2080 }
2081 
2082 int dlm_dispatch_assert_master(struct dlm_ctxt *dlm,
2083 			       struct dlm_lock_resource *res,
2084 			       int ignore_higher, u8 request_from, u32 flags)
2085 {
2086 	struct dlm_work_item *item;
2087 	item = kzalloc(sizeof(*item), GFP_ATOMIC);
2088 	if (!item)
2089 		return -ENOMEM;
2090 
2091 
2092 	/* queue up work for dlm_assert_master_worker */
2093 	dlm_grab(dlm);  /* get an extra ref for the work item */
2094 	dlm_init_work_item(dlm, item, dlm_assert_master_worker, NULL);
2095 	item->u.am.lockres = res; /* already have a ref */
2096 	/* can optionally ignore node numbers higher than this node */
2097 	item->u.am.ignore_higher = ignore_higher;
2098 	item->u.am.request_from = request_from;
2099 	item->u.am.flags = flags;
2100 
2101 	if (ignore_higher)
2102 		mlog(0, "IGNORE HIGHER: %.*s\n", res->lockname.len,
2103 		     res->lockname.name);
2104 
2105 	spin_lock(&dlm->work_lock);
2106 	list_add_tail(&item->list, &dlm->work_list);
2107 	spin_unlock(&dlm->work_lock);
2108 
2109 	queue_work(dlm->dlm_worker, &dlm->dispatched_work);
2110 	return 0;
2111 }
2112 
2113 static void dlm_assert_master_worker(struct dlm_work_item *item, void *data)
2114 {
2115 	struct dlm_ctxt *dlm = data;
2116 	int ret = 0;
2117 	struct dlm_lock_resource *res;
2118 	unsigned long nodemap[BITS_TO_LONGS(O2NM_MAX_NODES)];
2119 	int ignore_higher;
2120 	int bit;
2121 	u8 request_from;
2122 	u32 flags;
2123 
2124 	dlm = item->dlm;
2125 	res = item->u.am.lockres;
2126 	ignore_higher = item->u.am.ignore_higher;
2127 	request_from = item->u.am.request_from;
2128 	flags = item->u.am.flags;
2129 
2130 	spin_lock(&dlm->spinlock);
2131 	memcpy(nodemap, dlm->domain_map, sizeof(nodemap));
2132 	spin_unlock(&dlm->spinlock);
2133 
2134 	clear_bit(dlm->node_num, nodemap);
2135 	if (ignore_higher) {
2136 		/* if is this just to clear up mles for nodes below
2137 		 * this node, do not send the message to the original
2138 		 * caller or any node number higher than this */
2139 		clear_bit(request_from, nodemap);
2140 		bit = dlm->node_num;
2141 		while (1) {
2142 			bit = find_next_bit(nodemap, O2NM_MAX_NODES,
2143 					    bit+1);
2144 		       	if (bit >= O2NM_MAX_NODES)
2145 				break;
2146 			clear_bit(bit, nodemap);
2147 		}
2148 	}
2149 
2150 	/*
2151 	 * If we're migrating this lock to someone else, we are no
2152 	 * longer allowed to assert out own mastery.  OTOH, we need to
2153 	 * prevent migration from starting while we're still asserting
2154 	 * our dominance.  The reserved ast delays migration.
2155 	 */
2156 	spin_lock(&res->spinlock);
2157 	if (res->state & DLM_LOCK_RES_MIGRATING) {
2158 		mlog(0, "Someone asked us to assert mastery, but we're "
2159 		     "in the middle of migration.  Skipping assert, "
2160 		     "the new master will handle that.\n");
2161 		spin_unlock(&res->spinlock);
2162 		goto put;
2163 	} else
2164 		__dlm_lockres_reserve_ast(res);
2165 	spin_unlock(&res->spinlock);
2166 
2167 	/* this call now finishes out the nodemap
2168 	 * even if one or more nodes die */
2169 	mlog(0, "worker about to master %.*s here, this=%u\n",
2170 		     res->lockname.len, res->lockname.name, dlm->node_num);
2171 	ret = dlm_do_assert_master(dlm, res, nodemap, flags);
2172 	if (ret < 0) {
2173 		/* no need to restart, we are done */
2174 		if (!dlm_is_host_down(ret))
2175 			mlog_errno(ret);
2176 	}
2177 
2178 	/* Ok, we've asserted ourselves.  Let's let migration start. */
2179 	dlm_lockres_release_ast(dlm, res);
2180 
2181 put:
2182 	dlm_lockres_drop_inflight_worker(dlm, res);
2183 
2184 	dlm_lockres_put(res);
2185 
2186 	mlog(0, "finished with dlm_assert_master_worker\n");
2187 }
2188 
2189 /* SPECIAL CASE for the $RECOVERY lock used by the recovery thread.
2190  * We cannot wait for node recovery to complete to begin mastering this
2191  * lockres because this lockres is used to kick off recovery! ;-)
2192  * So, do a pre-check on all living nodes to see if any of those nodes
2193  * think that $RECOVERY is currently mastered by a dead node.  If so,
2194  * we wait a short time to allow that node to get notified by its own
2195  * heartbeat stack, then check again.  All $RECOVERY lock resources
2196  * mastered by dead nodes are purged when the hearbeat callback is
2197  * fired, so we can know for sure that it is safe to continue once
2198  * the node returns a live node or no node.  */
2199 static int dlm_pre_master_reco_lockres(struct dlm_ctxt *dlm,
2200 				       struct dlm_lock_resource *res)
2201 {
2202 	struct dlm_node_iter iter;
2203 	int nodenum;
2204 	int ret = 0;
2205 	u8 master = DLM_LOCK_RES_OWNER_UNKNOWN;
2206 
2207 	spin_lock(&dlm->spinlock);
2208 	dlm_node_iter_init(dlm->domain_map, &iter);
2209 	spin_unlock(&dlm->spinlock);
2210 
2211 	while ((nodenum = dlm_node_iter_next(&iter)) >= 0) {
2212 		/* do not send to self */
2213 		if (nodenum == dlm->node_num)
2214 			continue;
2215 		ret = dlm_do_master_requery(dlm, res, nodenum, &master);
2216 		if (ret < 0) {
2217 			mlog_errno(ret);
2218 			if (!dlm_is_host_down(ret))
2219 				BUG();
2220 			/* host is down, so answer for that node would be
2221 			 * DLM_LOCK_RES_OWNER_UNKNOWN.  continue. */
2222 			ret = 0;
2223 		}
2224 
2225 		if (master != DLM_LOCK_RES_OWNER_UNKNOWN) {
2226 			/* check to see if this master is in the recovery map */
2227 			spin_lock(&dlm->spinlock);
2228 			if (test_bit(master, dlm->recovery_map)) {
2229 				mlog(ML_NOTICE, "%s: node %u has not seen "
2230 				     "node %u go down yet, and thinks the "
2231 				     "dead node is mastering the recovery "
2232 				     "lock.  must wait.\n", dlm->name,
2233 				     nodenum, master);
2234 				ret = -EAGAIN;
2235 			}
2236 			spin_unlock(&dlm->spinlock);
2237 			mlog(0, "%s: reco lock master is %u\n", dlm->name,
2238 			     master);
2239 			break;
2240 		}
2241 	}
2242 	return ret;
2243 }
2244 
2245 /*
2246  * DLM_DEREF_LOCKRES_MSG
2247  */
2248 
2249 int dlm_drop_lockres_ref(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)
2250 {
2251 	struct dlm_deref_lockres deref;
2252 	int ret = 0, r;
2253 	const char *lockname;
2254 	unsigned int namelen;
2255 
2256 	lockname = res->lockname.name;
2257 	namelen = res->lockname.len;
2258 	BUG_ON(namelen > O2NM_MAX_NAME_LEN);
2259 
2260 	memset(&deref, 0, sizeof(deref));
2261 	deref.node_idx = dlm->node_num;
2262 	deref.namelen = namelen;
2263 	memcpy(deref.name, lockname, namelen);
2264 
2265 	ret = o2net_send_message(DLM_DEREF_LOCKRES_MSG, dlm->key,
2266 				 &deref, sizeof(deref), res->owner, &r);
2267 	if (ret < 0)
2268 		mlog(ML_ERROR, "%s: res %.*s, error %d send DEREF to node %u\n",
2269 		     dlm->name, namelen, lockname, ret, res->owner);
2270 	else if (r < 0) {
2271 		/* BAD.  other node says I did not have a ref. */
2272 		mlog(ML_ERROR, "%s: res %.*s, DEREF to node %u got %d\n",
2273 		     dlm->name, namelen, lockname, res->owner, r);
2274 		dlm_print_one_lock_resource(res);
2275 		BUG();
2276 	}
2277 	return ret;
2278 }
2279 
2280 int dlm_deref_lockres_handler(struct o2net_msg *msg, u32 len, void *data,
2281 			      void **ret_data)
2282 {
2283 	struct dlm_ctxt *dlm = data;
2284 	struct dlm_deref_lockres *deref = (struct dlm_deref_lockres *)msg->buf;
2285 	struct dlm_lock_resource *res = NULL;
2286 	char *name;
2287 	unsigned int namelen;
2288 	int ret = -EINVAL;
2289 	u8 node;
2290 	unsigned int hash;
2291 	struct dlm_work_item *item;
2292 	int cleared = 0;
2293 	int dispatch = 0;
2294 
2295 	if (!dlm_grab(dlm))
2296 		return 0;
2297 
2298 	name = deref->name;
2299 	namelen = deref->namelen;
2300 	node = deref->node_idx;
2301 
2302 	if (namelen > DLM_LOCKID_NAME_MAX) {
2303 		mlog(ML_ERROR, "Invalid name length!");
2304 		goto done;
2305 	}
2306 	if (deref->node_idx >= O2NM_MAX_NODES) {
2307 		mlog(ML_ERROR, "Invalid node number: %u\n", node);
2308 		goto done;
2309 	}
2310 
2311 	hash = dlm_lockid_hash(name, namelen);
2312 
2313 	spin_lock(&dlm->spinlock);
2314 	res = __dlm_lookup_lockres_full(dlm, name, namelen, hash);
2315 	if (!res) {
2316 		spin_unlock(&dlm->spinlock);
2317 		mlog(ML_ERROR, "%s:%.*s: bad lockres name\n",
2318 		     dlm->name, namelen, name);
2319 		goto done;
2320 	}
2321 	spin_unlock(&dlm->spinlock);
2322 
2323 	spin_lock(&res->spinlock);
2324 	if (res->state & DLM_LOCK_RES_SETREF_INPROG)
2325 		dispatch = 1;
2326 	else {
2327 		BUG_ON(res->state & DLM_LOCK_RES_DROPPING_REF);
2328 		if (test_bit(node, res->refmap)) {
2329 			dlm_lockres_clear_refmap_bit(dlm, res, node);
2330 			cleared = 1;
2331 		}
2332 	}
2333 	spin_unlock(&res->spinlock);
2334 
2335 	if (!dispatch) {
2336 		if (cleared)
2337 			dlm_lockres_calc_usage(dlm, res);
2338 		else {
2339 			mlog(ML_ERROR, "%s:%.*s: node %u trying to drop ref "
2340 		     	"but it is already dropped!\n", dlm->name,
2341 		     	res->lockname.len, res->lockname.name, node);
2342 			dlm_print_one_lock_resource(res);
2343 		}
2344 		ret = 0;
2345 		goto done;
2346 	}
2347 
2348 	item = kzalloc(sizeof(*item), GFP_NOFS);
2349 	if (!item) {
2350 		ret = -ENOMEM;
2351 		mlog_errno(ret);
2352 		goto done;
2353 	}
2354 
2355 	dlm_init_work_item(dlm, item, dlm_deref_lockres_worker, NULL);
2356 	item->u.dl.deref_res = res;
2357 	item->u.dl.deref_node = node;
2358 
2359 	spin_lock(&dlm->work_lock);
2360 	list_add_tail(&item->list, &dlm->work_list);
2361 	spin_unlock(&dlm->work_lock);
2362 
2363 	queue_work(dlm->dlm_worker, &dlm->dispatched_work);
2364 	return 0;
2365 
2366 done:
2367 	if (res)
2368 		dlm_lockres_put(res);
2369 	dlm_put(dlm);
2370 
2371 	return ret;
2372 }
2373 
2374 static void dlm_deref_lockres_worker(struct dlm_work_item *item, void *data)
2375 {
2376 	struct dlm_ctxt *dlm;
2377 	struct dlm_lock_resource *res;
2378 	u8 node;
2379 	u8 cleared = 0;
2380 
2381 	dlm = item->dlm;
2382 	res = item->u.dl.deref_res;
2383 	node = item->u.dl.deref_node;
2384 
2385 	spin_lock(&res->spinlock);
2386 	BUG_ON(res->state & DLM_LOCK_RES_DROPPING_REF);
2387 	if (test_bit(node, res->refmap)) {
2388 		__dlm_wait_on_lockres_flags(res, DLM_LOCK_RES_SETREF_INPROG);
2389 		dlm_lockres_clear_refmap_bit(dlm, res, node);
2390 		cleared = 1;
2391 	}
2392 	spin_unlock(&res->spinlock);
2393 
2394 	if (cleared) {
2395 		mlog(0, "%s:%.*s node %u ref dropped in dispatch\n",
2396 		     dlm->name, res->lockname.len, res->lockname.name, node);
2397 		dlm_lockres_calc_usage(dlm, res);
2398 	} else {
2399 		mlog(ML_ERROR, "%s:%.*s: node %u trying to drop ref "
2400 		     "but it is already dropped!\n", dlm->name,
2401 		     res->lockname.len, res->lockname.name, node);
2402 		dlm_print_one_lock_resource(res);
2403 	}
2404 
2405 	dlm_lockres_put(res);
2406 }
2407 
2408 /*
2409  * A migrateable resource is one that is :
2410  * 1. locally mastered, and,
2411  * 2. zero local locks, and,
2412  * 3. one or more non-local locks, or, one or more references
2413  * Returns 1 if yes, 0 if not.
2414  */
2415 static int dlm_is_lockres_migrateable(struct dlm_ctxt *dlm,
2416 				      struct dlm_lock_resource *res)
2417 {
2418 	enum dlm_lockres_list idx;
2419 	int nonlocal = 0, node_ref;
2420 	struct list_head *queue;
2421 	struct dlm_lock *lock;
2422 	u64 cookie;
2423 
2424 	assert_spin_locked(&res->spinlock);
2425 
2426 	/* delay migration when the lockres is in MIGRATING state */
2427 	if (res->state & DLM_LOCK_RES_MIGRATING)
2428 		return 0;
2429 
2430 	/* delay migration when the lockres is in RECOCERING state */
2431 	if (res->state & DLM_LOCK_RES_RECOVERING)
2432 		return 0;
2433 
2434 	if (res->owner != dlm->node_num)
2435 		return 0;
2436 
2437         for (idx = DLM_GRANTED_LIST; idx <= DLM_BLOCKED_LIST; idx++) {
2438 		queue = dlm_list_idx_to_ptr(res, idx);
2439 		list_for_each_entry(lock, queue, list) {
2440 			if (lock->ml.node != dlm->node_num) {
2441 				nonlocal++;
2442 				continue;
2443 			}
2444 			cookie = be64_to_cpu(lock->ml.cookie);
2445 			mlog(0, "%s: Not migrateable res %.*s, lock %u:%llu on "
2446 			     "%s list\n", dlm->name, res->lockname.len,
2447 			     res->lockname.name,
2448 			     dlm_get_lock_cookie_node(cookie),
2449 			     dlm_get_lock_cookie_seq(cookie),
2450 			     dlm_list_in_text(idx));
2451 			return 0;
2452 		}
2453 	}
2454 
2455 	if (!nonlocal) {
2456 		node_ref = find_next_bit(res->refmap, O2NM_MAX_NODES, 0);
2457 		if (node_ref >= O2NM_MAX_NODES)
2458 			return 0;
2459 	}
2460 
2461 	mlog(0, "%s: res %.*s, Migrateable\n", dlm->name, res->lockname.len,
2462 	     res->lockname.name);
2463 
2464 	return 1;
2465 }
2466 
2467 /*
2468  * DLM_MIGRATE_LOCKRES
2469  */
2470 
2471 
2472 static int dlm_migrate_lockres(struct dlm_ctxt *dlm,
2473 			       struct dlm_lock_resource *res, u8 target)
2474 {
2475 	struct dlm_master_list_entry *mle = NULL;
2476 	struct dlm_master_list_entry *oldmle = NULL;
2477  	struct dlm_migratable_lockres *mres = NULL;
2478 	int ret = 0;
2479 	const char *name;
2480 	unsigned int namelen;
2481 	int mle_added = 0;
2482 	int wake = 0;
2483 
2484 	if (!dlm_grab(dlm))
2485 		return -EINVAL;
2486 
2487 	BUG_ON(target == O2NM_MAX_NODES);
2488 
2489 	name = res->lockname.name;
2490 	namelen = res->lockname.len;
2491 
2492 	mlog(0, "%s: Migrating %.*s to node %u\n", dlm->name, namelen, name,
2493 	     target);
2494 
2495 	/* preallocate up front. if this fails, abort */
2496 	ret = -ENOMEM;
2497 	mres = (struct dlm_migratable_lockres *) __get_free_page(GFP_NOFS);
2498 	if (!mres) {
2499 		mlog_errno(ret);
2500 		goto leave;
2501 	}
2502 
2503 	mle = kmem_cache_alloc(dlm_mle_cache, GFP_NOFS);
2504 	if (!mle) {
2505 		mlog_errno(ret);
2506 		goto leave;
2507 	}
2508 	ret = 0;
2509 
2510 	/*
2511 	 * clear any existing master requests and
2512 	 * add the migration mle to the list
2513 	 */
2514 	spin_lock(&dlm->spinlock);
2515 	spin_lock(&dlm->master_lock);
2516 	ret = dlm_add_migration_mle(dlm, res, mle, &oldmle, name,
2517 				    namelen, target, dlm->node_num);
2518 	spin_unlock(&dlm->master_lock);
2519 	spin_unlock(&dlm->spinlock);
2520 
2521 	if (ret == -EEXIST) {
2522 		mlog(0, "another process is already migrating it\n");
2523 		goto fail;
2524 	}
2525 	mle_added = 1;
2526 
2527 	/*
2528 	 * set the MIGRATING flag and flush asts
2529 	 * if we fail after this we need to re-dirty the lockres
2530 	 */
2531 	if (dlm_mark_lockres_migrating(dlm, res, target) < 0) {
2532 		mlog(ML_ERROR, "tried to migrate %.*s to %u, but "
2533 		     "the target went down.\n", res->lockname.len,
2534 		     res->lockname.name, target);
2535 		spin_lock(&res->spinlock);
2536 		res->state &= ~DLM_LOCK_RES_MIGRATING;
2537 		wake = 1;
2538 		spin_unlock(&res->spinlock);
2539 		ret = -EINVAL;
2540 	}
2541 
2542 fail:
2543 	if (oldmle) {
2544 		/* master is known, detach if not already detached */
2545 		dlm_mle_detach_hb_events(dlm, oldmle);
2546 		dlm_put_mle(oldmle);
2547 	}
2548 
2549 	if (ret < 0) {
2550 		if (mle_added) {
2551 			dlm_mle_detach_hb_events(dlm, mle);
2552 			dlm_put_mle(mle);
2553 		} else if (mle) {
2554 			kmem_cache_free(dlm_mle_cache, mle);
2555 			mle = NULL;
2556 		}
2557 		goto leave;
2558 	}
2559 
2560 	/*
2561 	 * at this point, we have a migration target, an mle
2562 	 * in the master list, and the MIGRATING flag set on
2563 	 * the lockres
2564 	 */
2565 
2566 	/* now that remote nodes are spinning on the MIGRATING flag,
2567 	 * ensure that all assert_master work is flushed. */
2568 	flush_workqueue(dlm->dlm_worker);
2569 
2570 	/* get an extra reference on the mle.
2571 	 * otherwise the assert_master from the new
2572 	 * master will destroy this.
2573 	 * also, make sure that all callers of dlm_get_mle
2574 	 * take both dlm->spinlock and dlm->master_lock */
2575 	spin_lock(&dlm->spinlock);
2576 	spin_lock(&dlm->master_lock);
2577 	dlm_get_mle_inuse(mle);
2578 	spin_unlock(&dlm->master_lock);
2579 	spin_unlock(&dlm->spinlock);
2580 
2581 	/* notify new node and send all lock state */
2582 	/* call send_one_lockres with migration flag.
2583 	 * this serves as notice to the target node that a
2584 	 * migration is starting. */
2585 	ret = dlm_send_one_lockres(dlm, res, mres, target,
2586 				   DLM_MRES_MIGRATION);
2587 
2588 	if (ret < 0) {
2589 		mlog(0, "migration to node %u failed with %d\n",
2590 		     target, ret);
2591 		/* migration failed, detach and clean up mle */
2592 		dlm_mle_detach_hb_events(dlm, mle);
2593 		dlm_put_mle(mle);
2594 		dlm_put_mle_inuse(mle);
2595 		spin_lock(&res->spinlock);
2596 		res->state &= ~DLM_LOCK_RES_MIGRATING;
2597 		wake = 1;
2598 		spin_unlock(&res->spinlock);
2599 		if (dlm_is_host_down(ret))
2600 			dlm_wait_for_node_death(dlm, target,
2601 						DLM_NODE_DEATH_WAIT_MAX);
2602 		goto leave;
2603 	}
2604 
2605 	/* at this point, the target sends a message to all nodes,
2606 	 * (using dlm_do_migrate_request).  this node is skipped since
2607 	 * we had to put an mle in the list to begin the process.  this
2608 	 * node now waits for target to do an assert master.  this node
2609 	 * will be the last one notified, ensuring that the migration
2610 	 * is complete everywhere.  if the target dies while this is
2611 	 * going on, some nodes could potentially see the target as the
2612 	 * master, so it is important that my recovery finds the migration
2613 	 * mle and sets the master to UNKNOWN. */
2614 
2615 
2616 	/* wait for new node to assert master */
2617 	while (1) {
2618 		ret = wait_event_interruptible_timeout(mle->wq,
2619 					(atomic_read(&mle->woken) == 1),
2620 					msecs_to_jiffies(5000));
2621 
2622 		if (ret >= 0) {
2623 		       	if (atomic_read(&mle->woken) == 1 ||
2624 			    res->owner == target)
2625 				break;
2626 
2627 			mlog(0, "%s:%.*s: timed out during migration\n",
2628 			     dlm->name, res->lockname.len, res->lockname.name);
2629 			/* avoid hang during shutdown when migrating lockres
2630 			 * to a node which also goes down */
2631 			if (dlm_is_node_dead(dlm, target)) {
2632 				mlog(0, "%s:%.*s: expected migration "
2633 				     "target %u is no longer up, restarting\n",
2634 				     dlm->name, res->lockname.len,
2635 				     res->lockname.name, target);
2636 				ret = -EINVAL;
2637 				/* migration failed, detach and clean up mle */
2638 				dlm_mle_detach_hb_events(dlm, mle);
2639 				dlm_put_mle(mle);
2640 				dlm_put_mle_inuse(mle);
2641 				spin_lock(&res->spinlock);
2642 				res->state &= ~DLM_LOCK_RES_MIGRATING;
2643 				wake = 1;
2644 				spin_unlock(&res->spinlock);
2645 				goto leave;
2646 			}
2647 		} else
2648 			mlog(0, "%s:%.*s: caught signal during migration\n",
2649 			     dlm->name, res->lockname.len, res->lockname.name);
2650 	}
2651 
2652 	/* all done, set the owner, clear the flag */
2653 	spin_lock(&res->spinlock);
2654 	dlm_set_lockres_owner(dlm, res, target);
2655 	res->state &= ~DLM_LOCK_RES_MIGRATING;
2656 	dlm_remove_nonlocal_locks(dlm, res);
2657 	spin_unlock(&res->spinlock);
2658 	wake_up(&res->wq);
2659 
2660 	/* master is known, detach if not already detached */
2661 	dlm_mle_detach_hb_events(dlm, mle);
2662 	dlm_put_mle_inuse(mle);
2663 	ret = 0;
2664 
2665 	dlm_lockres_calc_usage(dlm, res);
2666 
2667 leave:
2668 	/* re-dirty the lockres if we failed */
2669 	if (ret < 0)
2670 		dlm_kick_thread(dlm, res);
2671 
2672 	/* wake up waiters if the MIGRATING flag got set
2673 	 * but migration failed */
2674 	if (wake)
2675 		wake_up(&res->wq);
2676 
2677 	if (mres)
2678 		free_page((unsigned long)mres);
2679 
2680 	dlm_put(dlm);
2681 
2682 	mlog(0, "%s: Migrating %.*s to %u, returns %d\n", dlm->name, namelen,
2683 	     name, target, ret);
2684 	return ret;
2685 }
2686 
2687 #define DLM_MIGRATION_RETRY_MS  100
2688 
2689 /*
2690  * Should be called only after beginning the domain leave process.
2691  * There should not be any remaining locks on nonlocal lock resources,
2692  * and there should be no local locks left on locally mastered resources.
2693  *
2694  * Called with the dlm spinlock held, may drop it to do migration, but
2695  * will re-acquire before exit.
2696  *
2697  * Returns: 1 if dlm->spinlock was dropped/retaken, 0 if never dropped
2698  */
2699 int dlm_empty_lockres(struct dlm_ctxt *dlm, struct dlm_lock_resource *res)
2700 {
2701 	int ret;
2702 	int lock_dropped = 0;
2703 	u8 target = O2NM_MAX_NODES;
2704 
2705 	assert_spin_locked(&dlm->spinlock);
2706 
2707 	spin_lock(&res->spinlock);
2708 	if (dlm_is_lockres_migrateable(dlm, res))
2709 		target = dlm_pick_migration_target(dlm, res);
2710 	spin_unlock(&res->spinlock);
2711 
2712 	if (target == O2NM_MAX_NODES)
2713 		goto leave;
2714 
2715 	/* Wheee! Migrate lockres here! Will sleep so drop spinlock. */
2716 	spin_unlock(&dlm->spinlock);
2717 	lock_dropped = 1;
2718 	ret = dlm_migrate_lockres(dlm, res, target);
2719 	if (ret)
2720 		mlog(0, "%s: res %.*s, Migrate to node %u failed with %d\n",
2721 		     dlm->name, res->lockname.len, res->lockname.name,
2722 		     target, ret);
2723 	spin_lock(&dlm->spinlock);
2724 leave:
2725 	return lock_dropped;
2726 }
2727 
2728 int dlm_lock_basts_flushed(struct dlm_ctxt *dlm, struct dlm_lock *lock)
2729 {
2730 	int ret;
2731 	spin_lock(&dlm->ast_lock);
2732 	spin_lock(&lock->spinlock);
2733 	ret = (list_empty(&lock->bast_list) && !lock->bast_pending);
2734 	spin_unlock(&lock->spinlock);
2735 	spin_unlock(&dlm->ast_lock);
2736 	return ret;
2737 }
2738 
2739 static int dlm_migration_can_proceed(struct dlm_ctxt *dlm,
2740 				     struct dlm_lock_resource *res,
2741 				     u8 mig_target)
2742 {
2743 	int can_proceed;
2744 	spin_lock(&res->spinlock);
2745 	can_proceed = !!(res->state & DLM_LOCK_RES_MIGRATING);
2746 	spin_unlock(&res->spinlock);
2747 
2748 	/* target has died, so make the caller break out of the
2749 	 * wait_event, but caller must recheck the domain_map */
2750 	spin_lock(&dlm->spinlock);
2751 	if (!test_bit(mig_target, dlm->domain_map))
2752 		can_proceed = 1;
2753 	spin_unlock(&dlm->spinlock);
2754 	return can_proceed;
2755 }
2756 
2757 static int dlm_lockres_is_dirty(struct dlm_ctxt *dlm,
2758 				struct dlm_lock_resource *res)
2759 {
2760 	int ret;
2761 	spin_lock(&res->spinlock);
2762 	ret = !!(res->state & DLM_LOCK_RES_DIRTY);
2763 	spin_unlock(&res->spinlock);
2764 	return ret;
2765 }
2766 
2767 
2768 static int dlm_mark_lockres_migrating(struct dlm_ctxt *dlm,
2769 				       struct dlm_lock_resource *res,
2770 				       u8 target)
2771 {
2772 	int ret = 0;
2773 
2774 	mlog(0, "dlm_mark_lockres_migrating: %.*s, from %u to %u\n",
2775 	       res->lockname.len, res->lockname.name, dlm->node_num,
2776 	       target);
2777 	/* need to set MIGRATING flag on lockres.  this is done by
2778 	 * ensuring that all asts have been flushed for this lockres. */
2779 	spin_lock(&res->spinlock);
2780 	BUG_ON(res->migration_pending);
2781 	res->migration_pending = 1;
2782 	/* strategy is to reserve an extra ast then release
2783 	 * it below, letting the release do all of the work */
2784 	__dlm_lockres_reserve_ast(res);
2785 	spin_unlock(&res->spinlock);
2786 
2787 	/* now flush all the pending asts */
2788 	dlm_kick_thread(dlm, res);
2789 	/* before waiting on DIRTY, block processes which may
2790 	 * try to dirty the lockres before MIGRATING is set */
2791 	spin_lock(&res->spinlock);
2792 	BUG_ON(res->state & DLM_LOCK_RES_BLOCK_DIRTY);
2793 	res->state |= DLM_LOCK_RES_BLOCK_DIRTY;
2794 	spin_unlock(&res->spinlock);
2795 	/* now wait on any pending asts and the DIRTY state */
2796 	wait_event(dlm->ast_wq, !dlm_lockres_is_dirty(dlm, res));
2797 	dlm_lockres_release_ast(dlm, res);
2798 
2799 	mlog(0, "about to wait on migration_wq, dirty=%s\n",
2800 	       res->state & DLM_LOCK_RES_DIRTY ? "yes" : "no");
2801 	/* if the extra ref we just put was the final one, this
2802 	 * will pass thru immediately.  otherwise, we need to wait
2803 	 * for the last ast to finish. */
2804 again:
2805 	ret = wait_event_interruptible_timeout(dlm->migration_wq,
2806 		   dlm_migration_can_proceed(dlm, res, target),
2807 		   msecs_to_jiffies(1000));
2808 	if (ret < 0) {
2809 		mlog(0, "woken again: migrating? %s, dead? %s\n",
2810 		       res->state & DLM_LOCK_RES_MIGRATING ? "yes":"no",
2811 		       test_bit(target, dlm->domain_map) ? "no":"yes");
2812 	} else {
2813 		mlog(0, "all is well: migrating? %s, dead? %s\n",
2814 		       res->state & DLM_LOCK_RES_MIGRATING ? "yes":"no",
2815 		       test_bit(target, dlm->domain_map) ? "no":"yes");
2816 	}
2817 	if (!dlm_migration_can_proceed(dlm, res, target)) {
2818 		mlog(0, "trying again...\n");
2819 		goto again;
2820 	}
2821 
2822 	ret = 0;
2823 	/* did the target go down or die? */
2824 	spin_lock(&dlm->spinlock);
2825 	if (!test_bit(target, dlm->domain_map)) {
2826 		mlog(ML_ERROR, "aha. migration target %u just went down\n",
2827 		     target);
2828 		ret = -EHOSTDOWN;
2829 	}
2830 	spin_unlock(&dlm->spinlock);
2831 
2832 	/*
2833 	 * if target is down, we need to clear DLM_LOCK_RES_BLOCK_DIRTY for
2834 	 * another try; otherwise, we are sure the MIGRATING state is there,
2835 	 * drop the unneded state which blocked threads trying to DIRTY
2836 	 */
2837 	spin_lock(&res->spinlock);
2838 	BUG_ON(!(res->state & DLM_LOCK_RES_BLOCK_DIRTY));
2839 	res->state &= ~DLM_LOCK_RES_BLOCK_DIRTY;
2840 	if (!ret)
2841 		BUG_ON(!(res->state & DLM_LOCK_RES_MIGRATING));
2842 	spin_unlock(&res->spinlock);
2843 
2844 	/*
2845 	 * at this point:
2846 	 *
2847 	 *   o the DLM_LOCK_RES_MIGRATING flag is set if target not down
2848 	 *   o there are no pending asts on this lockres
2849 	 *   o all processes trying to reserve an ast on this
2850 	 *     lockres must wait for the MIGRATING flag to clear
2851 	 */
2852 	return ret;
2853 }
2854 
2855 /* last step in the migration process.
2856  * original master calls this to free all of the dlm_lock
2857  * structures that used to be for other nodes. */
2858 static void dlm_remove_nonlocal_locks(struct dlm_ctxt *dlm,
2859 				      struct dlm_lock_resource *res)
2860 {
2861 	struct list_head *queue = &res->granted;
2862 	int i, bit;
2863 	struct dlm_lock *lock, *next;
2864 
2865 	assert_spin_locked(&res->spinlock);
2866 
2867 	BUG_ON(res->owner == dlm->node_num);
2868 
2869 	for (i=0; i<3; i++) {
2870 		list_for_each_entry_safe(lock, next, queue, list) {
2871 			if (lock->ml.node != dlm->node_num) {
2872 				mlog(0, "putting lock for node %u\n",
2873 				     lock->ml.node);
2874 				/* be extra careful */
2875 				BUG_ON(!list_empty(&lock->ast_list));
2876 				BUG_ON(!list_empty(&lock->bast_list));
2877 				BUG_ON(lock->ast_pending);
2878 				BUG_ON(lock->bast_pending);
2879 				dlm_lockres_clear_refmap_bit(dlm, res,
2880 							     lock->ml.node);
2881 				list_del_init(&lock->list);
2882 				dlm_lock_put(lock);
2883 				/* In a normal unlock, we would have added a
2884 				 * DLM_UNLOCK_FREE_LOCK action. Force it. */
2885 				dlm_lock_put(lock);
2886 			}
2887 		}
2888 		queue++;
2889 	}
2890 	bit = 0;
2891 	while (1) {
2892 		bit = find_next_bit(res->refmap, O2NM_MAX_NODES, bit);
2893 		if (bit >= O2NM_MAX_NODES)
2894 			break;
2895 		/* do not clear the local node reference, if there is a
2896 		 * process holding this, let it drop the ref itself */
2897 		if (bit != dlm->node_num) {
2898 			mlog(0, "%s:%.*s: node %u had a ref to this "
2899 			     "migrating lockres, clearing\n", dlm->name,
2900 			     res->lockname.len, res->lockname.name, bit);
2901 			dlm_lockres_clear_refmap_bit(dlm, res, bit);
2902 		}
2903 		bit++;
2904 	}
2905 }
2906 
2907 /*
2908  * Pick a node to migrate the lock resource to. This function selects a
2909  * potential target based first on the locks and then on refmap. It skips
2910  * nodes that are in the process of exiting the domain.
2911  */
2912 static u8 dlm_pick_migration_target(struct dlm_ctxt *dlm,
2913 				    struct dlm_lock_resource *res)
2914 {
2915 	enum dlm_lockres_list idx;
2916 	struct list_head *queue = &res->granted;
2917 	struct dlm_lock *lock;
2918 	int noderef;
2919 	u8 nodenum = O2NM_MAX_NODES;
2920 
2921 	assert_spin_locked(&dlm->spinlock);
2922 	assert_spin_locked(&res->spinlock);
2923 
2924 	/* Go through all the locks */
2925 	for (idx = DLM_GRANTED_LIST; idx <= DLM_BLOCKED_LIST; idx++) {
2926 		queue = dlm_list_idx_to_ptr(res, idx);
2927 		list_for_each_entry(lock, queue, list) {
2928 			if (lock->ml.node == dlm->node_num)
2929 				continue;
2930 			if (test_bit(lock->ml.node, dlm->exit_domain_map))
2931 				continue;
2932 			nodenum = lock->ml.node;
2933 			goto bail;
2934 		}
2935 	}
2936 
2937 	/* Go thru the refmap */
2938 	noderef = -1;
2939 	while (1) {
2940 		noderef = find_next_bit(res->refmap, O2NM_MAX_NODES,
2941 					noderef + 1);
2942 		if (noderef >= O2NM_MAX_NODES)
2943 			break;
2944 		if (noderef == dlm->node_num)
2945 			continue;
2946 		if (test_bit(noderef, dlm->exit_domain_map))
2947 			continue;
2948 		nodenum = noderef;
2949 		goto bail;
2950 	}
2951 
2952 bail:
2953 	return nodenum;
2954 }
2955 
2956 /* this is called by the new master once all lockres
2957  * data has been received */
2958 static int dlm_do_migrate_request(struct dlm_ctxt *dlm,
2959 				  struct dlm_lock_resource *res,
2960 				  u8 master, u8 new_master,
2961 				  struct dlm_node_iter *iter)
2962 {
2963 	struct dlm_migrate_request migrate;
2964 	int ret, skip, status = 0;
2965 	int nodenum;
2966 
2967 	memset(&migrate, 0, sizeof(migrate));
2968 	migrate.namelen = res->lockname.len;
2969 	memcpy(migrate.name, res->lockname.name, migrate.namelen);
2970 	migrate.new_master = new_master;
2971 	migrate.master = master;
2972 
2973 	ret = 0;
2974 
2975 	/* send message to all nodes, except the master and myself */
2976 	while ((nodenum = dlm_node_iter_next(iter)) >= 0) {
2977 		if (nodenum == master ||
2978 		    nodenum == new_master)
2979 			continue;
2980 
2981 		/* We could race exit domain. If exited, skip. */
2982 		spin_lock(&dlm->spinlock);
2983 		skip = (!test_bit(nodenum, dlm->domain_map));
2984 		spin_unlock(&dlm->spinlock);
2985 		if (skip) {
2986 			clear_bit(nodenum, iter->node_map);
2987 			continue;
2988 		}
2989 
2990 		ret = o2net_send_message(DLM_MIGRATE_REQUEST_MSG, dlm->key,
2991 					 &migrate, sizeof(migrate), nodenum,
2992 					 &status);
2993 		if (ret < 0) {
2994 			mlog(ML_ERROR, "%s: res %.*s, Error %d send "
2995 			     "MIGRATE_REQUEST to node %u\n", dlm->name,
2996 			     migrate.namelen, migrate.name, ret, nodenum);
2997 			if (!dlm_is_host_down(ret)) {
2998 				mlog(ML_ERROR, "unhandled error=%d!\n", ret);
2999 				BUG();
3000 			}
3001 			clear_bit(nodenum, iter->node_map);
3002 			ret = 0;
3003 		} else if (status < 0) {
3004 			mlog(0, "migrate request (node %u) returned %d!\n",
3005 			     nodenum, status);
3006 			ret = status;
3007 		} else if (status == DLM_MIGRATE_RESPONSE_MASTERY_REF) {
3008 			/* during the migration request we short-circuited
3009 			 * the mastery of the lockres.  make sure we have
3010 			 * a mastery ref for nodenum */
3011 			mlog(0, "%s:%.*s: need ref for node %u\n",
3012 			     dlm->name, res->lockname.len, res->lockname.name,
3013 			     nodenum);
3014 			spin_lock(&res->spinlock);
3015 			dlm_lockres_set_refmap_bit(dlm, res, nodenum);
3016 			spin_unlock(&res->spinlock);
3017 		}
3018 	}
3019 
3020 	if (ret < 0)
3021 		mlog_errno(ret);
3022 
3023 	mlog(0, "returning ret=%d\n", ret);
3024 	return ret;
3025 }
3026 
3027 
3028 /* if there is an existing mle for this lockres, we now know who the master is.
3029  * (the one who sent us *this* message) we can clear it up right away.
3030  * since the process that put the mle on the list still has a reference to it,
3031  * we can unhash it now, set the master and wake the process.  as a result,
3032  * we will have no mle in the list to start with.  now we can add an mle for
3033  * the migration and this should be the only one found for those scanning the
3034  * list.  */
3035 int dlm_migrate_request_handler(struct o2net_msg *msg, u32 len, void *data,
3036 				void **ret_data)
3037 {
3038 	struct dlm_ctxt *dlm = data;
3039 	struct dlm_lock_resource *res = NULL;
3040 	struct dlm_migrate_request *migrate = (struct dlm_migrate_request *) msg->buf;
3041 	struct dlm_master_list_entry *mle = NULL, *oldmle = NULL;
3042 	const char *name;
3043 	unsigned int namelen, hash;
3044 	int ret = 0;
3045 
3046 	if (!dlm_grab(dlm))
3047 		return -EINVAL;
3048 
3049 	name = migrate->name;
3050 	namelen = migrate->namelen;
3051 	hash = dlm_lockid_hash(name, namelen);
3052 
3053 	/* preallocate.. if this fails, abort */
3054 	mle = kmem_cache_alloc(dlm_mle_cache, GFP_NOFS);
3055 
3056 	if (!mle) {
3057 		ret = -ENOMEM;
3058 		goto leave;
3059 	}
3060 
3061 	/* check for pre-existing lock */
3062 	spin_lock(&dlm->spinlock);
3063 	res = __dlm_lookup_lockres(dlm, name, namelen, hash);
3064 	if (res) {
3065 		spin_lock(&res->spinlock);
3066 		if (res->state & DLM_LOCK_RES_RECOVERING) {
3067 			/* if all is working ok, this can only mean that we got
3068 		 	* a migrate request from a node that we now see as
3069 		 	* dead.  what can we do here?  drop it to the floor? */
3070 			spin_unlock(&res->spinlock);
3071 			mlog(ML_ERROR, "Got a migrate request, but the "
3072 			     "lockres is marked as recovering!");
3073 			kmem_cache_free(dlm_mle_cache, mle);
3074 			ret = -EINVAL; /* need a better solution */
3075 			goto unlock;
3076 		}
3077 		res->state |= DLM_LOCK_RES_MIGRATING;
3078 		spin_unlock(&res->spinlock);
3079 	}
3080 
3081 	spin_lock(&dlm->master_lock);
3082 	/* ignore status.  only nonzero status would BUG. */
3083 	ret = dlm_add_migration_mle(dlm, res, mle, &oldmle,
3084 				    name, namelen,
3085 				    migrate->new_master,
3086 				    migrate->master);
3087 
3088 	spin_unlock(&dlm->master_lock);
3089 unlock:
3090 	spin_unlock(&dlm->spinlock);
3091 
3092 	if (oldmle) {
3093 		/* master is known, detach if not already detached */
3094 		dlm_mle_detach_hb_events(dlm, oldmle);
3095 		dlm_put_mle(oldmle);
3096 	}
3097 
3098 	if (res)
3099 		dlm_lockres_put(res);
3100 leave:
3101 	dlm_put(dlm);
3102 	return ret;
3103 }
3104 
3105 /* must be holding dlm->spinlock and dlm->master_lock
3106  * when adding a migration mle, we can clear any other mles
3107  * in the master list because we know with certainty that
3108  * the master is "master".  so we remove any old mle from
3109  * the list after setting it's master field, and then add
3110  * the new migration mle.  this way we can hold with the rule
3111  * of having only one mle for a given lock name at all times. */
3112 static int dlm_add_migration_mle(struct dlm_ctxt *dlm,
3113 				 struct dlm_lock_resource *res,
3114 				 struct dlm_master_list_entry *mle,
3115 				 struct dlm_master_list_entry **oldmle,
3116 				 const char *name, unsigned int namelen,
3117 				 u8 new_master, u8 master)
3118 {
3119 	int found;
3120 	int ret = 0;
3121 
3122 	*oldmle = NULL;
3123 
3124 	assert_spin_locked(&dlm->spinlock);
3125 	assert_spin_locked(&dlm->master_lock);
3126 
3127 	/* caller is responsible for any ref taken here on oldmle */
3128 	found = dlm_find_mle(dlm, oldmle, (char *)name, namelen);
3129 	if (found) {
3130 		struct dlm_master_list_entry *tmp = *oldmle;
3131 		spin_lock(&tmp->spinlock);
3132 		if (tmp->type == DLM_MLE_MIGRATION) {
3133 			if (master == dlm->node_num) {
3134 				/* ah another process raced me to it */
3135 				mlog(0, "tried to migrate %.*s, but some "
3136 				     "process beat me to it\n",
3137 				     namelen, name);
3138 				ret = -EEXIST;
3139 			} else {
3140 				/* bad.  2 NODES are trying to migrate! */
3141 				mlog(ML_ERROR, "migration error  mle: "
3142 				     "master=%u new_master=%u // request: "
3143 				     "master=%u new_master=%u // "
3144 				     "lockres=%.*s\n",
3145 				     tmp->master, tmp->new_master,
3146 				     master, new_master,
3147 				     namelen, name);
3148 				BUG();
3149 			}
3150 		} else {
3151 			/* this is essentially what assert_master does */
3152 			tmp->master = master;
3153 			atomic_set(&tmp->woken, 1);
3154 			wake_up(&tmp->wq);
3155 			/* remove it so that only one mle will be found */
3156 			__dlm_unlink_mle(dlm, tmp);
3157 			__dlm_mle_detach_hb_events(dlm, tmp);
3158 			if (tmp->type == DLM_MLE_MASTER) {
3159 				ret = DLM_MIGRATE_RESPONSE_MASTERY_REF;
3160 				mlog(0, "%s:%.*s: master=%u, newmaster=%u, "
3161 						"telling master to get ref "
3162 						"for cleared out mle during "
3163 						"migration\n", dlm->name,
3164 						namelen, name, master,
3165 						new_master);
3166 			}
3167 		}
3168 		spin_unlock(&tmp->spinlock);
3169 	}
3170 
3171 	/* now add a migration mle to the tail of the list */
3172 	dlm_init_mle(mle, DLM_MLE_MIGRATION, dlm, res, name, namelen);
3173 	mle->new_master = new_master;
3174 	/* the new master will be sending an assert master for this.
3175 	 * at that point we will get the refmap reference */
3176 	mle->master = master;
3177 	/* do this for consistency with other mle types */
3178 	set_bit(new_master, mle->maybe_map);
3179 	__dlm_insert_mle(dlm, mle);
3180 
3181 	return ret;
3182 }
3183 
3184 /*
3185  * Sets the owner of the lockres, associated to the mle, to UNKNOWN
3186  */
3187 static struct dlm_lock_resource *dlm_reset_mleres_owner(struct dlm_ctxt *dlm,
3188 					struct dlm_master_list_entry *mle)
3189 {
3190 	struct dlm_lock_resource *res;
3191 
3192 	/* Find the lockres associated to the mle and set its owner to UNK */
3193 	res = __dlm_lookup_lockres(dlm, mle->mname, mle->mnamelen,
3194 				   mle->mnamehash);
3195 	if (res) {
3196 		spin_unlock(&dlm->master_lock);
3197 
3198 		/* move lockres onto recovery list */
3199 		spin_lock(&res->spinlock);
3200 		dlm_set_lockres_owner(dlm, res, DLM_LOCK_RES_OWNER_UNKNOWN);
3201 		dlm_move_lockres_to_recovery_list(dlm, res);
3202 		spin_unlock(&res->spinlock);
3203 		dlm_lockres_put(res);
3204 
3205 		/* about to get rid of mle, detach from heartbeat */
3206 		__dlm_mle_detach_hb_events(dlm, mle);
3207 
3208 		/* dump the mle */
3209 		spin_lock(&dlm->master_lock);
3210 		__dlm_put_mle(mle);
3211 		spin_unlock(&dlm->master_lock);
3212 	}
3213 
3214 	return res;
3215 }
3216 
3217 static void dlm_clean_migration_mle(struct dlm_ctxt *dlm,
3218 				    struct dlm_master_list_entry *mle)
3219 {
3220 	__dlm_mle_detach_hb_events(dlm, mle);
3221 
3222 	spin_lock(&mle->spinlock);
3223 	__dlm_unlink_mle(dlm, mle);
3224 	atomic_set(&mle->woken, 1);
3225 	spin_unlock(&mle->spinlock);
3226 
3227 	wake_up(&mle->wq);
3228 }
3229 
3230 static void dlm_clean_block_mle(struct dlm_ctxt *dlm,
3231 				struct dlm_master_list_entry *mle, u8 dead_node)
3232 {
3233 	int bit;
3234 
3235 	BUG_ON(mle->type != DLM_MLE_BLOCK);
3236 
3237 	spin_lock(&mle->spinlock);
3238 	bit = find_next_bit(mle->maybe_map, O2NM_MAX_NODES, 0);
3239 	if (bit != dead_node) {
3240 		mlog(0, "mle found, but dead node %u would not have been "
3241 		     "master\n", dead_node);
3242 		spin_unlock(&mle->spinlock);
3243 	} else {
3244 		/* Must drop the refcount by one since the assert_master will
3245 		 * never arrive. This may result in the mle being unlinked and
3246 		 * freed, but there may still be a process waiting in the
3247 		 * dlmlock path which is fine. */
3248 		mlog(0, "node %u was expected master\n", dead_node);
3249 		atomic_set(&mle->woken, 1);
3250 		spin_unlock(&mle->spinlock);
3251 		wake_up(&mle->wq);
3252 
3253 		/* Do not need events any longer, so detach from heartbeat */
3254 		__dlm_mle_detach_hb_events(dlm, mle);
3255 		__dlm_put_mle(mle);
3256 	}
3257 }
3258 
3259 void dlm_clean_master_list(struct dlm_ctxt *dlm, u8 dead_node)
3260 {
3261 	struct dlm_master_list_entry *mle;
3262 	struct dlm_lock_resource *res;
3263 	struct hlist_head *bucket;
3264 	struct hlist_node *tmp;
3265 	unsigned int i;
3266 
3267 	mlog(0, "dlm=%s, dead node=%u\n", dlm->name, dead_node);
3268 top:
3269 	assert_spin_locked(&dlm->spinlock);
3270 
3271 	/* clean the master list */
3272 	spin_lock(&dlm->master_lock);
3273 	for (i = 0; i < DLM_HASH_BUCKETS; i++) {
3274 		bucket = dlm_master_hash(dlm, i);
3275 		hlist_for_each_entry_safe(mle, tmp, bucket, master_hash_node) {
3276 			BUG_ON(mle->type != DLM_MLE_BLOCK &&
3277 			       mle->type != DLM_MLE_MASTER &&
3278 			       mle->type != DLM_MLE_MIGRATION);
3279 
3280 			/* MASTER mles are initiated locally. The waiting
3281 			 * process will notice the node map change shortly.
3282 			 * Let that happen as normal. */
3283 			if (mle->type == DLM_MLE_MASTER)
3284 				continue;
3285 
3286 			/* BLOCK mles are initiated by other nodes. Need to
3287 			 * clean up if the dead node would have been the
3288 			 * master. */
3289 			if (mle->type == DLM_MLE_BLOCK) {
3290 				dlm_clean_block_mle(dlm, mle, dead_node);
3291 				continue;
3292 			}
3293 
3294 			/* Everything else is a MIGRATION mle */
3295 
3296 			/* The rule for MIGRATION mles is that the master
3297 			 * becomes UNKNOWN if *either* the original or the new
3298 			 * master dies. All UNKNOWN lockres' are sent to
3299 			 * whichever node becomes the recovery master. The new
3300 			 * master is responsible for determining if there is
3301 			 * still a master for this lockres, or if he needs to
3302 			 * take over mastery. Either way, this node should
3303 			 * expect another message to resolve this. */
3304 
3305 			if (mle->master != dead_node &&
3306 			    mle->new_master != dead_node)
3307 				continue;
3308 
3309 			/* If we have reached this point, this mle needs to be
3310 			 * removed from the list and freed. */
3311 			dlm_clean_migration_mle(dlm, mle);
3312 
3313 			mlog(0, "%s: node %u died during migration from "
3314 			     "%u to %u!\n", dlm->name, dead_node, mle->master,
3315 			     mle->new_master);
3316 
3317 			/* If we find a lockres associated with the mle, we've
3318 			 * hit this rare case that messes up our lock ordering.
3319 			 * If so, we need to drop the master lock so that we can
3320 			 * take the lockres lock, meaning that we will have to
3321 			 * restart from the head of list. */
3322 			res = dlm_reset_mleres_owner(dlm, mle);
3323 			if (res)
3324 				/* restart */
3325 				goto top;
3326 
3327 			/* This may be the last reference */
3328 			__dlm_put_mle(mle);
3329 		}
3330 	}
3331 	spin_unlock(&dlm->master_lock);
3332 }
3333 
3334 int dlm_finish_migration(struct dlm_ctxt *dlm, struct dlm_lock_resource *res,
3335 			 u8 old_master)
3336 {
3337 	struct dlm_node_iter iter;
3338 	int ret = 0;
3339 
3340 	spin_lock(&dlm->spinlock);
3341 	dlm_node_iter_init(dlm->domain_map, &iter);
3342 	clear_bit(old_master, iter.node_map);
3343 	clear_bit(dlm->node_num, iter.node_map);
3344 	spin_unlock(&dlm->spinlock);
3345 
3346 	/* ownership of the lockres is changing.  account for the
3347 	 * mastery reference here since old_master will briefly have
3348 	 * a reference after the migration completes */
3349 	spin_lock(&res->spinlock);
3350 	dlm_lockres_set_refmap_bit(dlm, res, old_master);
3351 	spin_unlock(&res->spinlock);
3352 
3353 	mlog(0, "now time to do a migrate request to other nodes\n");
3354 	ret = dlm_do_migrate_request(dlm, res, old_master,
3355 				     dlm->node_num, &iter);
3356 	if (ret < 0) {
3357 		mlog_errno(ret);
3358 		goto leave;
3359 	}
3360 
3361 	mlog(0, "doing assert master of %.*s to all except the original node\n",
3362 	     res->lockname.len, res->lockname.name);
3363 	/* this call now finishes out the nodemap
3364 	 * even if one or more nodes die */
3365 	ret = dlm_do_assert_master(dlm, res, iter.node_map,
3366 				   DLM_ASSERT_MASTER_FINISH_MIGRATION);
3367 	if (ret < 0) {
3368 		/* no longer need to retry.  all living nodes contacted. */
3369 		mlog_errno(ret);
3370 		ret = 0;
3371 	}
3372 
3373 	memset(iter.node_map, 0, sizeof(iter.node_map));
3374 	set_bit(old_master, iter.node_map);
3375 	mlog(0, "doing assert master of %.*s back to %u\n",
3376 	     res->lockname.len, res->lockname.name, old_master);
3377 	ret = dlm_do_assert_master(dlm, res, iter.node_map,
3378 				   DLM_ASSERT_MASTER_FINISH_MIGRATION);
3379 	if (ret < 0) {
3380 		mlog(0, "assert master to original master failed "
3381 		     "with %d.\n", ret);
3382 		/* the only nonzero status here would be because of
3383 		 * a dead original node.  we're done. */
3384 		ret = 0;
3385 	}
3386 
3387 	/* all done, set the owner, clear the flag */
3388 	spin_lock(&res->spinlock);
3389 	dlm_set_lockres_owner(dlm, res, dlm->node_num);
3390 	res->state &= ~DLM_LOCK_RES_MIGRATING;
3391 	spin_unlock(&res->spinlock);
3392 	/* re-dirty it on the new master */
3393 	dlm_kick_thread(dlm, res);
3394 	wake_up(&res->wq);
3395 leave:
3396 	return ret;
3397 }
3398 
3399 /*
3400  * LOCKRES AST REFCOUNT
3401  * this is integral to migration
3402  */
3403 
3404 /* for future intent to call an ast, reserve one ahead of time.
3405  * this should be called only after waiting on the lockres
3406  * with dlm_wait_on_lockres, and while still holding the
3407  * spinlock after the call. */
3408 void __dlm_lockres_reserve_ast(struct dlm_lock_resource *res)
3409 {
3410 	assert_spin_locked(&res->spinlock);
3411 	if (res->state & DLM_LOCK_RES_MIGRATING) {
3412 		__dlm_print_one_lock_resource(res);
3413 	}
3414 	BUG_ON(res->state & DLM_LOCK_RES_MIGRATING);
3415 
3416 	atomic_inc(&res->asts_reserved);
3417 }
3418 
3419 /*
3420  * used to drop the reserved ast, either because it went unused,
3421  * or because the ast/bast was actually called.
3422  *
3423  * also, if there is a pending migration on this lockres,
3424  * and this was the last pending ast on the lockres,
3425  * atomically set the MIGRATING flag before we drop the lock.
3426  * this is how we ensure that migration can proceed with no
3427  * asts in progress.  note that it is ok if the state of the
3428  * queues is such that a lock should be granted in the future
3429  * or that a bast should be fired, because the new master will
3430  * shuffle the lists on this lockres as soon as it is migrated.
3431  */
3432 void dlm_lockres_release_ast(struct dlm_ctxt *dlm,
3433 			     struct dlm_lock_resource *res)
3434 {
3435 	if (!atomic_dec_and_lock(&res->asts_reserved, &res->spinlock))
3436 		return;
3437 
3438 	if (!res->migration_pending) {
3439 		spin_unlock(&res->spinlock);
3440 		return;
3441 	}
3442 
3443 	BUG_ON(res->state & DLM_LOCK_RES_MIGRATING);
3444 	res->migration_pending = 0;
3445 	res->state |= DLM_LOCK_RES_MIGRATING;
3446 	spin_unlock(&res->spinlock);
3447 	wake_up(&res->wq);
3448 	wake_up(&dlm->migration_wq);
3449 }
3450 
3451 void dlm_force_free_mles(struct dlm_ctxt *dlm)
3452 {
3453 	int i;
3454 	struct hlist_head *bucket;
3455 	struct dlm_master_list_entry *mle;
3456 	struct hlist_node *tmp;
3457 
3458 	/*
3459 	 * We notified all other nodes that we are exiting the domain and
3460 	 * marked the dlm state to DLM_CTXT_LEAVING. If any mles are still
3461 	 * around we force free them and wake any processes that are waiting
3462 	 * on the mles
3463 	 */
3464 	spin_lock(&dlm->spinlock);
3465 	spin_lock(&dlm->master_lock);
3466 
3467 	BUG_ON(dlm->dlm_state != DLM_CTXT_LEAVING);
3468 	BUG_ON((find_next_bit(dlm->domain_map, O2NM_MAX_NODES, 0) < O2NM_MAX_NODES));
3469 
3470 	for (i = 0; i < DLM_HASH_BUCKETS; i++) {
3471 		bucket = dlm_master_hash(dlm, i);
3472 		hlist_for_each_entry_safe(mle, tmp, bucket, master_hash_node) {
3473 			if (mle->type != DLM_MLE_BLOCK) {
3474 				mlog(ML_ERROR, "bad mle: %p\n", mle);
3475 				dlm_print_one_mle(mle);
3476 			}
3477 			atomic_set(&mle->woken, 1);
3478 			wake_up(&mle->wq);
3479 
3480 			__dlm_unlink_mle(dlm, mle);
3481 			__dlm_mle_detach_hb_events(dlm, mle);
3482 			__dlm_put_mle(mle);
3483 		}
3484 	}
3485 	spin_unlock(&dlm->master_lock);
3486 	spin_unlock(&dlm->spinlock);
3487 }
3488