1 /* -*- mode: c; c-basic-offset: 8; -*-
2 * vim: noexpandtab sw=8 ts=8 sts=0:
6 * standalone DLM module
8 * Copyright (C) 2004 Oracle. All rights reserved.
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.
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.
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.
28 #include <linux/module.h>
30 #include <linux/types.h>
31 #include <linux/slab.h>
32 #include <linux/highmem.h>
33 #include <linux/utsname.h>
34 #include <linux/init.h>
35 #include <linux/sysctl.h>
36 #include <linux/random.h>
37 #include <linux/blkdev.h>
38 #include <linux/socket.h>
39 #include <linux/inet.h>
40 #include <linux/spinlock.h>
41 #include <linux/delay.h>
44 #include "cluster/heartbeat.h"
45 #include "cluster/nodemanager.h"
46 #include "cluster/tcp.h"
49 #include "dlmcommon.h"
51 #include "dlmdomain.h"
53 #define MLOG_MASK_PREFIX (ML_DLM|ML_DLM_MASTER)
54 #include "cluster/masklog.h"
65 u8 name
[DLM_LOCKID_NAME_MAX
];
68 struct dlm_master_list_entry
70 struct list_head list
;
71 struct list_head hb_events
;
77 unsigned long maybe_map
[BITS_TO_LONGS(O2NM_MAX_NODES
)];
78 unsigned long vote_map
[BITS_TO_LONGS(O2NM_MAX_NODES
)];
79 unsigned long response_map
[BITS_TO_LONGS(O2NM_MAX_NODES
)];
80 unsigned long node_map
[BITS_TO_LONGS(O2NM_MAX_NODES
)];
83 enum dlm_mle_type type
;
84 struct o2hb_callback_func mle_hb_up
;
85 struct o2hb_callback_func mle_hb_down
;
87 struct dlm_lock_resource
*res
;
88 struct dlm_lock_name name
;
92 static void dlm_mle_node_down(struct dlm_ctxt
*dlm
,
93 struct dlm_master_list_entry
*mle
,
94 struct o2nm_node
*node
,
96 static void dlm_mle_node_up(struct dlm_ctxt
*dlm
,
97 struct dlm_master_list_entry
*mle
,
98 struct o2nm_node
*node
,
101 static void dlm_assert_master_worker(struct dlm_work_item
*item
, void *data
);
102 static int dlm_do_assert_master(struct dlm_ctxt
*dlm
, const char *lockname
,
103 unsigned int namelen
, void *nodemap
,
106 static inline int dlm_mle_equal(struct dlm_ctxt
*dlm
,
107 struct dlm_master_list_entry
*mle
,
109 unsigned int namelen
)
111 struct dlm_lock_resource
*res
;
116 if (mle
->type
== DLM_MLE_BLOCK
||
117 mle
->type
== DLM_MLE_MIGRATION
) {
118 if (namelen
!= mle
->u
.name
.len
||
119 memcmp(name
, mle
->u
.name
.name
, namelen
)!=0)
123 if (namelen
!= res
->lockname
.len
||
124 memcmp(res
->lockname
.name
, name
, namelen
) != 0)
131 /* Code here is included but defined out as it aids debugging */
133 void dlm_print_one_mle(struct dlm_master_list_entry
*mle
)
139 unsigned int namelen
;
144 if (mle
->type
== DLM_MLE_BLOCK
)
146 else if (mle
->type
== DLM_MLE_MASTER
)
150 refs
= atomic_read(&k
->refcount
);
151 master
= mle
->master
;
152 attached
= (list_empty(&mle
->hb_events
) ? 'N' : 'Y');
154 if (mle
->type
!= DLM_MLE_MASTER
) {
155 namelen
= mle
->u
.name
.len
;
156 name
= mle
->u
.name
.name
;
158 namelen
= mle
->u
.res
->lockname
.len
;
159 name
= mle
->u
.res
->lockname
.name
;
162 mlog(ML_NOTICE
, " #%3d: %3s %3d %3u %3u %c (%d)%.*s\n",
163 i
, type
, refs
, master
, mle
->new_master
, attached
,
164 namelen
, namelen
, name
);
167 static void dlm_dump_mles(struct dlm_ctxt
*dlm
)
169 struct dlm_master_list_entry
*mle
;
170 struct list_head
*iter
;
172 mlog(ML_NOTICE
, "dumping all mles for domain %s:\n", dlm
->name
);
173 mlog(ML_NOTICE
, " ####: type refs owner new events? lockname nodemap votemap respmap maybemap\n");
174 spin_lock(&dlm
->master_lock
);
175 list_for_each(iter
, &dlm
->master_list
) {
176 mle
= list_entry(iter
, struct dlm_master_list_entry
, list
);
177 dlm_print_one_mle(mle
);
179 spin_unlock(&dlm
->master_lock
);
182 int dlm_dump_all_mles(const char __user
*data
, unsigned int len
)
184 struct list_head
*iter
;
185 struct dlm_ctxt
*dlm
;
187 spin_lock(&dlm_domain_lock
);
188 list_for_each(iter
, &dlm_domains
) {
189 dlm
= list_entry (iter
, struct dlm_ctxt
, list
);
190 mlog(ML_NOTICE
, "found dlm: %p, name=%s\n", dlm
, dlm
->name
);
193 spin_unlock(&dlm_domain_lock
);
196 EXPORT_SYMBOL_GPL(dlm_dump_all_mles
);
201 static kmem_cache_t
*dlm_mle_cache
= NULL
;
204 static void dlm_mle_release(struct kref
*kref
);
205 static void dlm_init_mle(struct dlm_master_list_entry
*mle
,
206 enum dlm_mle_type type
,
207 struct dlm_ctxt
*dlm
,
208 struct dlm_lock_resource
*res
,
210 unsigned int namelen
);
211 static void dlm_put_mle(struct dlm_master_list_entry
*mle
);
212 static void __dlm_put_mle(struct dlm_master_list_entry
*mle
);
213 static int dlm_find_mle(struct dlm_ctxt
*dlm
,
214 struct dlm_master_list_entry
**mle
,
215 char *name
, unsigned int namelen
);
217 static int dlm_do_master_request(struct dlm_master_list_entry
*mle
, int to
);
220 static int dlm_wait_for_lock_mastery(struct dlm_ctxt
*dlm
,
221 struct dlm_lock_resource
*res
,
222 struct dlm_master_list_entry
*mle
,
224 static int dlm_restart_lock_mastery(struct dlm_ctxt
*dlm
,
225 struct dlm_lock_resource
*res
,
226 struct dlm_master_list_entry
*mle
,
228 static int dlm_add_migration_mle(struct dlm_ctxt
*dlm
,
229 struct dlm_lock_resource
*res
,
230 struct dlm_master_list_entry
*mle
,
231 struct dlm_master_list_entry
**oldmle
,
232 const char *name
, unsigned int namelen
,
233 u8 new_master
, u8 master
);
235 static u8
dlm_pick_migration_target(struct dlm_ctxt
*dlm
,
236 struct dlm_lock_resource
*res
);
237 static void dlm_remove_nonlocal_locks(struct dlm_ctxt
*dlm
,
238 struct dlm_lock_resource
*res
);
239 static int dlm_mark_lockres_migrating(struct dlm_ctxt
*dlm
,
240 struct dlm_lock_resource
*res
,
244 int dlm_is_host_down(int errno
)
261 case -EINVAL
: /* if returned from our tcp code,
262 this means there is no socket */
270 * MASTER LIST FUNCTIONS
275 * regarding master list entries and heartbeat callbacks:
277 * in order to avoid sleeping and allocation that occurs in
278 * heartbeat, master list entries are simply attached to the
279 * dlm's established heartbeat callbacks. the mle is attached
280 * when it is created, and since the dlm->spinlock is held at
281 * that time, any heartbeat event will be properly discovered
282 * by the mle. the mle needs to be detached from the
283 * dlm->mle_hb_events list as soon as heartbeat events are no
284 * longer useful to the mle, and before the mle is freed.
286 * as a general rule, heartbeat events are no longer needed by
287 * the mle once an "answer" regarding the lock master has been
290 static inline void __dlm_mle_attach_hb_events(struct dlm_ctxt
*dlm
,
291 struct dlm_master_list_entry
*mle
)
293 assert_spin_locked(&dlm
->spinlock
);
295 list_add_tail(&mle
->hb_events
, &dlm
->mle_hb_events
);
299 static inline void __dlm_mle_detach_hb_events(struct dlm_ctxt
*dlm
,
300 struct dlm_master_list_entry
*mle
)
302 if (!list_empty(&mle
->hb_events
))
303 list_del_init(&mle
->hb_events
);
307 static inline void dlm_mle_detach_hb_events(struct dlm_ctxt
*dlm
,
308 struct dlm_master_list_entry
*mle
)
310 spin_lock(&dlm
->spinlock
);
311 __dlm_mle_detach_hb_events(dlm
, mle
);
312 spin_unlock(&dlm
->spinlock
);
315 /* remove from list and free */
316 static void __dlm_put_mle(struct dlm_master_list_entry
*mle
)
318 struct dlm_ctxt
*dlm
;
321 assert_spin_locked(&dlm
->spinlock
);
322 assert_spin_locked(&dlm
->master_lock
);
323 BUG_ON(!atomic_read(&mle
->mle_refs
.refcount
));
325 kref_put(&mle
->mle_refs
, dlm_mle_release
);
329 /* must not have any spinlocks coming in */
330 static void dlm_put_mle(struct dlm_master_list_entry
*mle
)
332 struct dlm_ctxt
*dlm
;
335 spin_lock(&dlm
->spinlock
);
336 spin_lock(&dlm
->master_lock
);
338 spin_unlock(&dlm
->master_lock
);
339 spin_unlock(&dlm
->spinlock
);
342 static inline void dlm_get_mle(struct dlm_master_list_entry
*mle
)
344 kref_get(&mle
->mle_refs
);
347 static void dlm_init_mle(struct dlm_master_list_entry
*mle
,
348 enum dlm_mle_type type
,
349 struct dlm_ctxt
*dlm
,
350 struct dlm_lock_resource
*res
,
352 unsigned int namelen
)
354 assert_spin_locked(&dlm
->spinlock
);
358 INIT_LIST_HEAD(&mle
->list
);
359 INIT_LIST_HEAD(&mle
->hb_events
);
360 memset(mle
->maybe_map
, 0, sizeof(mle
->maybe_map
));
361 spin_lock_init(&mle
->spinlock
);
362 init_waitqueue_head(&mle
->wq
);
363 atomic_set(&mle
->woken
, 0);
364 kref_init(&mle
->mle_refs
);
365 memset(mle
->response_map
, 0, sizeof(mle
->response_map
));
366 mle
->master
= O2NM_MAX_NODES
;
367 mle
->new_master
= O2NM_MAX_NODES
;
369 if (mle
->type
== DLM_MLE_MASTER
) {
372 } else if (mle
->type
== DLM_MLE_BLOCK
) {
374 memcpy(mle
->u
.name
.name
, name
, namelen
);
375 mle
->u
.name
.len
= namelen
;
376 } else /* DLM_MLE_MIGRATION */ {
378 memcpy(mle
->u
.name
.name
, name
, namelen
);
379 mle
->u
.name
.len
= namelen
;
382 /* copy off the node_map and register hb callbacks on our copy */
383 memcpy(mle
->node_map
, dlm
->domain_map
, sizeof(mle
->node_map
));
384 memcpy(mle
->vote_map
, dlm
->domain_map
, sizeof(mle
->vote_map
));
385 clear_bit(dlm
->node_num
, mle
->vote_map
);
386 clear_bit(dlm
->node_num
, mle
->node_map
);
388 /* attach the mle to the domain node up/down events */
389 __dlm_mle_attach_hb_events(dlm
, mle
);
393 /* returns 1 if found, 0 if not */
394 static int dlm_find_mle(struct dlm_ctxt
*dlm
,
395 struct dlm_master_list_entry
**mle
,
396 char *name
, unsigned int namelen
)
398 struct dlm_master_list_entry
*tmpmle
;
399 struct list_head
*iter
;
401 assert_spin_locked(&dlm
->master_lock
);
403 list_for_each(iter
, &dlm
->master_list
) {
404 tmpmle
= list_entry(iter
, struct dlm_master_list_entry
, list
);
405 if (!dlm_mle_equal(dlm
, tmpmle
, name
, namelen
))
414 void dlm_hb_event_notify_attached(struct dlm_ctxt
*dlm
, int idx
, int node_up
)
416 struct dlm_master_list_entry
*mle
;
417 struct list_head
*iter
;
419 assert_spin_locked(&dlm
->spinlock
);
421 list_for_each(iter
, &dlm
->mle_hb_events
) {
422 mle
= list_entry(iter
, struct dlm_master_list_entry
,
425 dlm_mle_node_up(dlm
, mle
, NULL
, idx
);
427 dlm_mle_node_down(dlm
, mle
, NULL
, idx
);
431 static void dlm_mle_node_down(struct dlm_ctxt
*dlm
,
432 struct dlm_master_list_entry
*mle
,
433 struct o2nm_node
*node
, int idx
)
435 spin_lock(&mle
->spinlock
);
437 if (!test_bit(idx
, mle
->node_map
))
438 mlog(0, "node %u already removed from nodemap!\n", idx
);
440 clear_bit(idx
, mle
->node_map
);
442 spin_unlock(&mle
->spinlock
);
445 static void dlm_mle_node_up(struct dlm_ctxt
*dlm
,
446 struct dlm_master_list_entry
*mle
,
447 struct o2nm_node
*node
, int idx
)
449 spin_lock(&mle
->spinlock
);
451 if (test_bit(idx
, mle
->node_map
))
452 mlog(0, "node %u already in node map!\n", idx
);
454 set_bit(idx
, mle
->node_map
);
456 spin_unlock(&mle
->spinlock
);
460 int dlm_init_mle_cache(void)
462 dlm_mle_cache
= kmem_cache_create("dlm_mle_cache",
463 sizeof(struct dlm_master_list_entry
),
464 0, SLAB_HWCACHE_ALIGN
,
466 if (dlm_mle_cache
== NULL
)
471 void dlm_destroy_mle_cache(void)
474 kmem_cache_destroy(dlm_mle_cache
);
477 static void dlm_mle_release(struct kref
*kref
)
479 struct dlm_master_list_entry
*mle
;
480 struct dlm_ctxt
*dlm
;
484 mle
= container_of(kref
, struct dlm_master_list_entry
, mle_refs
);
487 if (mle
->type
!= DLM_MLE_MASTER
) {
488 mlog(0, "calling mle_release for %.*s, type %d\n",
489 mle
->u
.name
.len
, mle
->u
.name
.name
, mle
->type
);
491 mlog(0, "calling mle_release for %.*s, type %d\n",
492 mle
->u
.res
->lockname
.len
,
493 mle
->u
.res
->lockname
.name
, mle
->type
);
495 assert_spin_locked(&dlm
->spinlock
);
496 assert_spin_locked(&dlm
->master_lock
);
498 /* remove from list if not already */
499 if (!list_empty(&mle
->list
))
500 list_del_init(&mle
->list
);
502 /* detach the mle from the domain node up/down events */
503 __dlm_mle_detach_hb_events(dlm
, mle
);
505 /* NOTE: kfree under spinlock here.
506 * if this is bad, we can move this to a freelist. */
507 kmem_cache_free(dlm_mle_cache
, mle
);
512 * LOCK RESOURCE FUNCTIONS
515 static void dlm_set_lockres_owner(struct dlm_ctxt
*dlm
,
516 struct dlm_lock_resource
*res
,
519 assert_spin_locked(&res
->spinlock
);
521 mlog_entry("%.*s, %u\n", res
->lockname
.len
, res
->lockname
.name
, owner
);
523 if (owner
== dlm
->node_num
)
524 atomic_inc(&dlm
->local_resources
);
525 else if (owner
== DLM_LOCK_RES_OWNER_UNKNOWN
)
526 atomic_inc(&dlm
->unknown_resources
);
528 atomic_inc(&dlm
->remote_resources
);
533 void dlm_change_lockres_owner(struct dlm_ctxt
*dlm
,
534 struct dlm_lock_resource
*res
, u8 owner
)
536 assert_spin_locked(&res
->spinlock
);
538 if (owner
== res
->owner
)
541 if (res
->owner
== dlm
->node_num
)
542 atomic_dec(&dlm
->local_resources
);
543 else if (res
->owner
== DLM_LOCK_RES_OWNER_UNKNOWN
)
544 atomic_dec(&dlm
->unknown_resources
);
546 atomic_dec(&dlm
->remote_resources
);
548 dlm_set_lockres_owner(dlm
, res
, owner
);
552 static void dlm_lockres_release(struct kref
*kref
)
554 struct dlm_lock_resource
*res
;
556 res
= container_of(kref
, struct dlm_lock_resource
, refs
);
558 /* This should not happen -- all lockres' have a name
559 * associated with them at init time. */
560 BUG_ON(!res
->lockname
.name
);
562 mlog(0, "destroying lockres %.*s\n", res
->lockname
.len
,
565 /* By the time we're ready to blow this guy away, we shouldn't
566 * be on any lists. */
567 BUG_ON(!hlist_unhashed(&res
->hash_node
));
568 BUG_ON(!list_empty(&res
->granted
));
569 BUG_ON(!list_empty(&res
->converting
));
570 BUG_ON(!list_empty(&res
->blocked
));
571 BUG_ON(!list_empty(&res
->dirty
));
572 BUG_ON(!list_empty(&res
->recovering
));
573 BUG_ON(!list_empty(&res
->purge
));
575 kfree(res
->lockname
.name
);
580 void dlm_lockres_get(struct dlm_lock_resource
*res
)
582 kref_get(&res
->refs
);
585 void dlm_lockres_put(struct dlm_lock_resource
*res
)
587 kref_put(&res
->refs
, dlm_lockres_release
);
590 static void dlm_init_lockres(struct dlm_ctxt
*dlm
,
591 struct dlm_lock_resource
*res
,
592 const char *name
, unsigned int namelen
)
596 /* If we memset here, we lose our reference to the kmalloc'd
597 * res->lockname.name, so be sure to init every field
600 qname
= (char *) res
->lockname
.name
;
601 memcpy(qname
, name
, namelen
);
603 res
->lockname
.len
= namelen
;
604 res
->lockname
.hash
= full_name_hash(name
, namelen
);
606 init_waitqueue_head(&res
->wq
);
607 spin_lock_init(&res
->spinlock
);
608 INIT_HLIST_NODE(&res
->hash_node
);
609 INIT_LIST_HEAD(&res
->granted
);
610 INIT_LIST_HEAD(&res
->converting
);
611 INIT_LIST_HEAD(&res
->blocked
);
612 INIT_LIST_HEAD(&res
->dirty
);
613 INIT_LIST_HEAD(&res
->recovering
);
614 INIT_LIST_HEAD(&res
->purge
);
615 atomic_set(&res
->asts_reserved
, 0);
616 res
->migration_pending
= 0;
618 kref_init(&res
->refs
);
620 /* just for consistency */
621 spin_lock(&res
->spinlock
);
622 dlm_set_lockres_owner(dlm
, res
, DLM_LOCK_RES_OWNER_UNKNOWN
);
623 spin_unlock(&res
->spinlock
);
625 res
->state
= DLM_LOCK_RES_IN_PROGRESS
;
629 memset(res
->lvb
, 0, DLM_LVB_LEN
);
632 struct dlm_lock_resource
*dlm_new_lockres(struct dlm_ctxt
*dlm
,
634 unsigned int namelen
)
636 struct dlm_lock_resource
*res
;
638 res
= kmalloc(sizeof(struct dlm_lock_resource
), GFP_KERNEL
);
642 res
->lockname
.name
= kmalloc(namelen
, GFP_KERNEL
);
643 if (!res
->lockname
.name
) {
648 dlm_init_lockres(dlm
, res
, name
, namelen
);
653 * lookup a lock resource by name.
654 * may already exist in the hashtable.
655 * lockid is null terminated
657 * if not, allocate enough for the lockres and for
658 * the temporary structure used in doing the mastering.
660 * also, do a lookup in the dlm->master_list to see
661 * if another node has begun mastering the same lock.
662 * if so, there should be a block entry in there
663 * for this name, and we should *not* attempt to master
664 * the lock here. need to wait around for that node
665 * to assert_master (or die).
668 struct dlm_lock_resource
* dlm_get_lock_resource(struct dlm_ctxt
*dlm
,
672 struct dlm_lock_resource
*tmpres
=NULL
, *res
=NULL
;
673 struct dlm_master_list_entry
*mle
= NULL
;
674 struct dlm_master_list_entry
*alloc_mle
= NULL
;
677 struct dlm_node_iter iter
;
678 unsigned int namelen
;
683 namelen
= strlen(lockid
);
685 mlog(0, "get lockres %s (len %d)\n", lockid
, namelen
);
688 spin_lock(&dlm
->spinlock
);
689 tmpres
= __dlm_lookup_lockres(dlm
, lockid
, namelen
);
691 spin_unlock(&dlm
->spinlock
);
692 mlog(0, "found in hash!\n");
694 dlm_lockres_put(res
);
700 spin_unlock(&dlm
->spinlock
);
701 mlog(0, "allocating a new resource\n");
702 /* nothing found and we need to allocate one. */
703 alloc_mle
= (struct dlm_master_list_entry
*)
704 kmem_cache_alloc(dlm_mle_cache
, GFP_KERNEL
);
707 res
= dlm_new_lockres(dlm
, lockid
, namelen
);
713 mlog(0, "no lockres found, allocated our own: %p\n", res
);
715 if (flags
& LKM_LOCAL
) {
716 /* caller knows it's safe to assume it's not mastered elsewhere
717 * DONE! return right away */
718 spin_lock(&res
->spinlock
);
719 dlm_change_lockres_owner(dlm
, res
, dlm
->node_num
);
720 __dlm_insert_lockres(dlm
, res
);
721 spin_unlock(&res
->spinlock
);
722 spin_unlock(&dlm
->spinlock
);
723 /* lockres still marked IN_PROGRESS */
727 /* check master list to see if another node has started mastering it */
728 spin_lock(&dlm
->master_lock
);
730 /* if we found a block, wait for lock to be mastered by another node */
731 blocked
= dlm_find_mle(dlm
, &mle
, (char *)lockid
, namelen
);
733 if (mle
->type
== DLM_MLE_MASTER
) {
734 mlog(ML_ERROR
, "master entry for nonexistent lock!\n");
736 } else if (mle
->type
== DLM_MLE_MIGRATION
) {
737 /* migration is in progress! */
738 /* the good news is that we now know the
739 * "current" master (mle->master). */
741 spin_unlock(&dlm
->master_lock
);
742 assert_spin_locked(&dlm
->spinlock
);
744 /* set the lockres owner and hash it */
745 spin_lock(&res
->spinlock
);
746 dlm_set_lockres_owner(dlm
, res
, mle
->master
);
747 __dlm_insert_lockres(dlm
, res
);
748 spin_unlock(&res
->spinlock
);
749 spin_unlock(&dlm
->spinlock
);
751 /* master is known, detach */
752 dlm_mle_detach_hb_events(dlm
, mle
);
758 /* go ahead and try to master lock on this node */
760 /* make sure this does not get freed below */
762 dlm_init_mle(mle
, DLM_MLE_MASTER
, dlm
, res
, NULL
, 0);
763 set_bit(dlm
->node_num
, mle
->maybe_map
);
764 list_add(&mle
->list
, &dlm
->master_list
);
767 /* at this point there is either a DLM_MLE_BLOCK or a
768 * DLM_MLE_MASTER on the master list, so it's safe to add the
769 * lockres to the hashtable. anyone who finds the lock will
770 * still have to wait on the IN_PROGRESS. */
772 /* finally add the lockres to its hash bucket */
773 __dlm_insert_lockres(dlm
, res
);
774 /* get an extra ref on the mle in case this is a BLOCK
775 * if so, the creator of the BLOCK may try to put the last
776 * ref at this time in the assert master handler, so we
777 * need an extra one to keep from a bad ptr deref. */
779 spin_unlock(&dlm
->master_lock
);
780 spin_unlock(&dlm
->spinlock
);
782 /* must wait for lock to be mastered elsewhere */
788 dlm_node_iter_init(mle
->vote_map
, &iter
);
789 while ((nodenum
= dlm_node_iter_next(&iter
)) >= 0) {
790 ret
= dlm_do_master_request(mle
, nodenum
);
793 if (mle
->master
!= O2NM_MAX_NODES
) {
794 /* found a master ! */
800 /* keep going until the response map includes all nodes */
801 ret
= dlm_wait_for_lock_mastery(dlm
, res
, mle
, &blocked
);
803 mlog(0, "%s:%.*s: node map changed, redo the "
804 "master request now, blocked=%d\n",
805 dlm
->name
, res
->lockname
.len
,
806 res
->lockname
.name
, blocked
);
808 mlog(ML_ERROR
, "%s:%.*s: spinning on "
809 "dlm_wait_for_lock_mastery, blocked=%d\n",
810 dlm
->name
, res
->lockname
.len
,
811 res
->lockname
.name
, blocked
);
812 dlm_print_one_lock_resource(res
);
813 /* dlm_print_one_mle(mle); */
819 mlog(0, "lockres mastered by %u\n", res
->owner
);
820 /* make sure we never continue without this */
821 BUG_ON(res
->owner
== O2NM_MAX_NODES
);
823 /* master is known, detach if not already detached */
824 dlm_mle_detach_hb_events(dlm
, mle
);
826 /* put the extra ref */
830 spin_lock(&res
->spinlock
);
831 res
->state
&= ~DLM_LOCK_RES_IN_PROGRESS
;
832 spin_unlock(&res
->spinlock
);
836 /* need to free the unused mle */
838 kmem_cache_free(dlm_mle_cache
, alloc_mle
);
844 #define DLM_MASTERY_TIMEOUT_MS 5000
846 static int dlm_wait_for_lock_mastery(struct dlm_ctxt
*dlm
,
847 struct dlm_lock_resource
*res
,
848 struct dlm_master_list_entry
*mle
,
853 int map_changed
, voting_done
;
860 /* check if another node has already become the owner */
861 spin_lock(&res
->spinlock
);
862 if (res
->owner
!= DLM_LOCK_RES_OWNER_UNKNOWN
) {
863 spin_unlock(&res
->spinlock
);
866 spin_unlock(&res
->spinlock
);
868 spin_lock(&mle
->spinlock
);
870 map_changed
= (memcmp(mle
->vote_map
, mle
->node_map
,
871 sizeof(mle
->vote_map
)) != 0);
872 voting_done
= (memcmp(mle
->vote_map
, mle
->response_map
,
873 sizeof(mle
->vote_map
)) == 0);
875 /* restart if we hit any errors */
878 mlog(0, "%s: %.*s: node map changed, restarting\n",
879 dlm
->name
, res
->lockname
.len
, res
->lockname
.name
);
880 ret
= dlm_restart_lock_mastery(dlm
, res
, mle
, *blocked
);
881 b
= (mle
->type
== DLM_MLE_BLOCK
);
882 if ((*blocked
&& !b
) || (!*blocked
&& b
)) {
883 mlog(0, "%s:%.*s: status change: old=%d new=%d\n",
884 dlm
->name
, res
->lockname
.len
, res
->lockname
.name
,
888 spin_unlock(&mle
->spinlock
);
893 mlog(0, "%s:%.*s: restart lock mastery succeeded, "
894 "rechecking now\n", dlm
->name
, res
->lockname
.len
,
899 if (m
!= O2NM_MAX_NODES
) {
900 /* another node has done an assert!
905 /* have all nodes responded? */
906 if (voting_done
&& !*blocked
) {
907 bit
= find_next_bit(mle
->maybe_map
, O2NM_MAX_NODES
, 0);
908 if (dlm
->node_num
<= bit
) {
909 /* my node number is lowest.
910 * now tell other nodes that I am
912 mle
->master
= dlm
->node_num
;
916 /* if voting is done, but we have not received
917 * an assert master yet, we must sleep */
921 spin_unlock(&mle
->spinlock
);
923 /* sleep if we haven't finished voting yet */
925 unsigned long timeo
= msecs_to_jiffies(DLM_MASTERY_TIMEOUT_MS
);
928 if (atomic_read(&mle->mle_refs.refcount) < 2)
929 mlog(ML_ERROR, "mle (%p) refs=%d, name=%.*s\n", mle,
930 atomic_read(&mle->mle_refs.refcount),
931 res->lockname.len, res->lockname.name);
933 atomic_set(&mle
->woken
, 0);
934 (void)wait_event_timeout(mle
->wq
,
935 (atomic_read(&mle
->woken
) == 1),
937 if (res
->owner
== O2NM_MAX_NODES
) {
938 mlog(0, "waiting again\n");
941 mlog(0, "done waiting, master is %u\n", res
->owner
);
949 mlog(0, "about to master %.*s here, this=%u\n",
950 res
->lockname
.len
, res
->lockname
.name
, m
);
951 ret
= dlm_do_assert_master(dlm
, res
->lockname
.name
,
952 res
->lockname
.len
, mle
->vote_map
, 0);
954 /* This is a failure in the network path,
955 * not in the response to the assert_master
956 * (any nonzero response is a BUG on this node).
957 * Most likely a socket just got disconnected
958 * due to node death. */
961 /* no longer need to restart lock mastery.
962 * all living nodes have been contacted. */
966 /* set the lockres owner */
967 spin_lock(&res
->spinlock
);
968 dlm_change_lockres_owner(dlm
, res
, m
);
969 spin_unlock(&res
->spinlock
);
975 struct dlm_bitmap_diff_iter
978 unsigned long *orig_bm
;
979 unsigned long *cur_bm
;
980 unsigned long diff_bm
[BITS_TO_LONGS(O2NM_MAX_NODES
)];
983 enum dlm_node_state_change
990 static void dlm_bitmap_diff_iter_init(struct dlm_bitmap_diff_iter
*iter
,
991 unsigned long *orig_bm
,
992 unsigned long *cur_bm
)
994 unsigned long p1
, p2
;
998 iter
->orig_bm
= orig_bm
;
999 iter
->cur_bm
= cur_bm
;
1001 for (i
= 0; i
< BITS_TO_LONGS(O2NM_MAX_NODES
); i
++) {
1002 p1
= *(iter
->orig_bm
+ i
);
1003 p2
= *(iter
->cur_bm
+ i
);
1004 iter
->diff_bm
[i
] = (p1
& ~p2
) | (p2
& ~p1
);
1008 static int dlm_bitmap_diff_iter_next(struct dlm_bitmap_diff_iter
*iter
,
1009 enum dlm_node_state_change
*state
)
1013 if (iter
->curnode
>= O2NM_MAX_NODES
)
1016 bit
= find_next_bit(iter
->diff_bm
, O2NM_MAX_NODES
,
1018 if (bit
>= O2NM_MAX_NODES
) {
1019 iter
->curnode
= O2NM_MAX_NODES
;
1023 /* if it was there in the original then this node died */
1024 if (test_bit(bit
, iter
->orig_bm
))
1029 iter
->curnode
= bit
;
1034 static int dlm_restart_lock_mastery(struct dlm_ctxt
*dlm
,
1035 struct dlm_lock_resource
*res
,
1036 struct dlm_master_list_entry
*mle
,
1039 struct dlm_bitmap_diff_iter bdi
;
1040 enum dlm_node_state_change sc
;
1044 mlog(0, "something happened such that the "
1045 "master process may need to be restarted!\n");
1047 assert_spin_locked(&mle
->spinlock
);
1049 dlm_bitmap_diff_iter_init(&bdi
, mle
->vote_map
, mle
->node_map
);
1050 node
= dlm_bitmap_diff_iter_next(&bdi
, &sc
);
1052 if (sc
== NODE_UP
) {
1053 /* a node came up. clear any old vote from
1054 * the response map and set it in the vote map
1055 * then restart the mastery. */
1056 mlog(ML_NOTICE
, "node %d up while restarting\n", node
);
1058 /* redo the master request, but only for the new node */
1059 mlog(0, "sending request to new node\n");
1060 clear_bit(node
, mle
->response_map
);
1061 set_bit(node
, mle
->vote_map
);
1063 mlog(ML_ERROR
, "node down! %d\n", node
);
1065 /* if the node wasn't involved in mastery skip it,
1066 * but clear it out from the maps so that it will
1067 * not affect mastery of this lockres */
1068 clear_bit(node
, mle
->response_map
);
1069 clear_bit(node
, mle
->vote_map
);
1070 if (!test_bit(node
, mle
->maybe_map
))
1073 /* if we're already blocked on lock mastery, and the
1074 * dead node wasn't the expected master, or there is
1075 * another node in the maybe_map, keep waiting */
1077 int lowest
= find_next_bit(mle
->maybe_map
,
1080 /* act like it was never there */
1081 clear_bit(node
, mle
->maybe_map
);
1086 mlog(ML_ERROR
, "expected master %u died while "
1087 "this node was blocked waiting on it!\n",
1089 lowest
= find_next_bit(mle
->maybe_map
,
1092 if (lowest
< O2NM_MAX_NODES
) {
1093 mlog(0, "still blocked. waiting "
1094 "on %u now\n", lowest
);
1098 /* mle is an MLE_BLOCK, but there is now
1099 * nothing left to block on. we need to return
1100 * all the way back out and try again with
1101 * an MLE_MASTER. dlm_do_local_recovery_cleanup
1102 * has already run, so the mle refcount is ok */
1103 mlog(0, "no longer blocking. we can "
1104 "try to master this here\n");
1105 mle
->type
= DLM_MLE_MASTER
;
1106 memset(mle
->maybe_map
, 0,
1107 sizeof(mle
->maybe_map
));
1108 memset(mle
->response_map
, 0,
1109 sizeof(mle
->maybe_map
));
1110 memcpy(mle
->vote_map
, mle
->node_map
,
1111 sizeof(mle
->node_map
));
1113 set_bit(dlm
->node_num
, mle
->maybe_map
);
1119 clear_bit(node
, mle
->maybe_map
);
1120 if (node
> dlm
->node_num
)
1123 mlog(0, "dead node in map!\n");
1124 /* yuck. go back and re-contact all nodes
1125 * in the vote_map, removing this node. */
1126 memset(mle
->response_map
, 0,
1127 sizeof(mle
->response_map
));
1131 node
= dlm_bitmap_diff_iter_next(&bdi
, &sc
);
1138 * DLM_MASTER_REQUEST_MSG
1140 * returns: 0 on success,
1141 * -errno on a network error
1143 * on error, the caller should assume the target node is "dead"
1147 static int dlm_do_master_request(struct dlm_master_list_entry
*mle
, int to
)
1149 struct dlm_ctxt
*dlm
= mle
->dlm
;
1150 struct dlm_master_request request
;
1151 int ret
, response
=0, resend
;
1153 memset(&request
, 0, sizeof(request
));
1154 request
.node_idx
= dlm
->node_num
;
1156 BUG_ON(mle
->type
== DLM_MLE_MIGRATION
);
1158 if (mle
->type
!= DLM_MLE_MASTER
) {
1159 request
.namelen
= mle
->u
.name
.len
;
1160 memcpy(request
.name
, mle
->u
.name
.name
, request
.namelen
);
1162 request
.namelen
= mle
->u
.res
->lockname
.len
;
1163 memcpy(request
.name
, mle
->u
.res
->lockname
.name
,
1168 ret
= o2net_send_message(DLM_MASTER_REQUEST_MSG
, dlm
->key
, &request
,
1169 sizeof(request
), to
, &response
);
1171 if (ret
== -ESRCH
) {
1172 /* should never happen */
1173 mlog(ML_ERROR
, "TCP stack not ready!\n");
1175 } else if (ret
== -EINVAL
) {
1176 mlog(ML_ERROR
, "bad args passed to o2net!\n");
1178 } else if (ret
== -ENOMEM
) {
1179 mlog(ML_ERROR
, "out of memory while trying to send "
1180 "network message! retrying\n");
1181 /* this is totally crude */
1184 } else if (!dlm_is_host_down(ret
)) {
1185 /* not a network error. bad. */
1187 mlog(ML_ERROR
, "unhandled error!");
1190 /* all other errors should be network errors,
1191 * and likely indicate node death */
1192 mlog(ML_ERROR
, "link to %d went down!\n", to
);
1198 spin_lock(&mle
->spinlock
);
1200 case DLM_MASTER_RESP_YES
:
1201 set_bit(to
, mle
->response_map
);
1202 mlog(0, "node %u is the master, response=YES\n", to
);
1205 case DLM_MASTER_RESP_NO
:
1206 mlog(0, "node %u not master, response=NO\n", to
);
1207 set_bit(to
, mle
->response_map
);
1209 case DLM_MASTER_RESP_MAYBE
:
1210 mlog(0, "node %u not master, response=MAYBE\n", to
);
1211 set_bit(to
, mle
->response_map
);
1212 set_bit(to
, mle
->maybe_map
);
1214 case DLM_MASTER_RESP_ERROR
:
1215 mlog(0, "node %u hit an error, resending\n", to
);
1220 mlog(ML_ERROR
, "bad response! %u\n", response
);
1223 spin_unlock(&mle
->spinlock
);
1225 /* this is also totally crude */
1235 * locks that can be taken here:
1241 * if possible, TRIM THIS DOWN!!!
1243 int dlm_master_request_handler(struct o2net_msg
*msg
, u32 len
, void *data
)
1245 u8 response
= DLM_MASTER_RESP_MAYBE
;
1246 struct dlm_ctxt
*dlm
= data
;
1247 struct dlm_lock_resource
*res
;
1248 struct dlm_master_request
*request
= (struct dlm_master_request
*) msg
->buf
;
1249 struct dlm_master_list_entry
*mle
= NULL
, *tmpmle
= NULL
;
1251 unsigned int namelen
;
1256 return DLM_MASTER_RESP_NO
;
1258 if (!dlm_domain_fully_joined(dlm
)) {
1259 response
= DLM_MASTER_RESP_NO
;
1263 name
= request
->name
;
1264 namelen
= request
->namelen
;
1266 if (namelen
> DLM_LOCKID_NAME_MAX
) {
1267 response
= DLM_IVBUFLEN
;
1272 spin_lock(&dlm
->spinlock
);
1273 res
= __dlm_lookup_lockres(dlm
, name
, namelen
);
1275 spin_unlock(&dlm
->spinlock
);
1277 /* take care of the easy cases up front */
1278 spin_lock(&res
->spinlock
);
1279 if (res
->state
& DLM_LOCK_RES_RECOVERING
) {
1280 spin_unlock(&res
->spinlock
);
1281 mlog(0, "returning DLM_MASTER_RESP_ERROR since res is "
1282 "being recovered\n");
1283 response
= DLM_MASTER_RESP_ERROR
;
1285 kmem_cache_free(dlm_mle_cache
, mle
);
1289 if (res
->owner
== dlm
->node_num
) {
1290 u32 flags
= DLM_ASSERT_MASTER_MLE_CLEANUP
;
1291 spin_unlock(&res
->spinlock
);
1292 // mlog(0, "this node is the master\n");
1293 response
= DLM_MASTER_RESP_YES
;
1295 kmem_cache_free(dlm_mle_cache
, mle
);
1297 /* this node is the owner.
1298 * there is some extra work that needs to
1299 * happen now. the requesting node has
1300 * caused all nodes up to this one to
1301 * create mles. this node now needs to
1302 * go back and clean those up. */
1303 mlog(0, "%u is the owner of %.*s, cleaning everyone else\n",
1304 dlm
->node_num
, res
->lockname
.len
, res
->lockname
.name
);
1305 ret
= dlm_dispatch_assert_master(dlm
, res
, 1,
1309 mlog(ML_ERROR
, "failed to dispatch assert "
1311 response
= DLM_MASTER_RESP_ERROR
;
1314 } else if (res
->owner
!= DLM_LOCK_RES_OWNER_UNKNOWN
) {
1315 spin_unlock(&res
->spinlock
);
1316 // mlog(0, "node %u is the master\n", res->owner);
1317 response
= DLM_MASTER_RESP_NO
;
1319 kmem_cache_free(dlm_mle_cache
, mle
);
1323 /* ok, there is no owner. either this node is
1324 * being blocked, or it is actively trying to
1325 * master this lock. */
1326 if (!(res
->state
& DLM_LOCK_RES_IN_PROGRESS
)) {
1327 mlog(ML_ERROR
, "lock with no owner should be "
1332 // mlog(0, "lockres is in progress...\n");
1333 spin_lock(&dlm
->master_lock
);
1334 found
= dlm_find_mle(dlm
, &tmpmle
, name
, namelen
);
1336 mlog(ML_ERROR
, "no mle found for this lock!\n");
1340 spin_lock(&tmpmle
->spinlock
);
1341 if (tmpmle
->type
== DLM_MLE_BLOCK
) {
1342 // mlog(0, "this node is waiting for "
1343 // "lockres to be mastered\n");
1344 response
= DLM_MASTER_RESP_NO
;
1345 } else if (tmpmle
->type
== DLM_MLE_MIGRATION
) {
1346 mlog(0, "node %u is master, but trying to migrate to "
1347 "node %u.\n", tmpmle
->master
, tmpmle
->new_master
);
1348 if (tmpmle
->master
== dlm
->node_num
) {
1349 response
= DLM_MASTER_RESP_YES
;
1350 mlog(ML_ERROR
, "no owner on lockres, but this "
1351 "node is trying to migrate it to %u?!\n",
1352 tmpmle
->new_master
);
1355 /* the real master can respond on its own */
1356 response
= DLM_MASTER_RESP_NO
;
1358 } else if (tmpmle
->master
!= DLM_LOCK_RES_OWNER_UNKNOWN
) {
1360 if (tmpmle
->master
== dlm
->node_num
)
1361 response
= DLM_MASTER_RESP_YES
;
1363 response
= DLM_MASTER_RESP_NO
;
1365 // mlog(0, "this node is attempting to "
1366 // "master lockres\n");
1367 response
= DLM_MASTER_RESP_MAYBE
;
1370 set_bit(request
->node_idx
, tmpmle
->maybe_map
);
1371 spin_unlock(&tmpmle
->spinlock
);
1373 spin_unlock(&dlm
->master_lock
);
1374 spin_unlock(&res
->spinlock
);
1376 /* keep the mle attached to heartbeat events */
1377 dlm_put_mle(tmpmle
);
1379 kmem_cache_free(dlm_mle_cache
, mle
);
1384 * lockres doesn't exist on this node
1385 * if there is an MLE_BLOCK, return NO
1386 * if there is an MLE_MASTER, return MAYBE
1387 * otherwise, add an MLE_BLOCK, return NO
1389 spin_lock(&dlm
->master_lock
);
1390 found
= dlm_find_mle(dlm
, &tmpmle
, name
, namelen
);
1392 /* this lockid has never been seen on this node yet */
1393 // mlog(0, "no mle found\n");
1395 spin_unlock(&dlm
->master_lock
);
1396 spin_unlock(&dlm
->spinlock
);
1398 mle
= (struct dlm_master_list_entry
*)
1399 kmem_cache_alloc(dlm_mle_cache
, GFP_KERNEL
);
1401 // bad bad bad... this sucks.
1402 response
= DLM_MASTER_RESP_ERROR
;
1405 spin_lock(&dlm
->spinlock
);
1406 dlm_init_mle(mle
, DLM_MLE_BLOCK
, dlm
, NULL
,
1408 spin_unlock(&dlm
->spinlock
);
1412 // mlog(0, "this is second time thru, already allocated, "
1413 // "add the block.\n");
1414 set_bit(request
->node_idx
, mle
->maybe_map
);
1415 list_add(&mle
->list
, &dlm
->master_list
);
1416 response
= DLM_MASTER_RESP_NO
;
1418 // mlog(0, "mle was found\n");
1420 spin_lock(&tmpmle
->spinlock
);
1421 if (tmpmle
->type
== DLM_MLE_BLOCK
)
1422 response
= DLM_MASTER_RESP_NO
;
1423 else if (tmpmle
->type
== DLM_MLE_MIGRATION
) {
1424 mlog(0, "migration mle was found (%u->%u)\n",
1425 tmpmle
->master
, tmpmle
->new_master
);
1426 if (tmpmle
->master
== dlm
->node_num
) {
1427 mlog(ML_ERROR
, "no lockres, but migration mle "
1428 "says that this node is master!\n");
1431 /* real master can respond on its own */
1432 response
= DLM_MASTER_RESP_NO
;
1434 if (tmpmle
->master
== dlm
->node_num
) {
1435 response
= DLM_MASTER_RESP_YES
;
1438 response
= DLM_MASTER_RESP_MAYBE
;
1441 set_bit(request
->node_idx
, tmpmle
->maybe_map
);
1442 spin_unlock(&tmpmle
->spinlock
);
1444 spin_unlock(&dlm
->master_lock
);
1445 spin_unlock(&dlm
->spinlock
);
1448 /* keep the mle attached to heartbeat events */
1449 dlm_put_mle(tmpmle
);
1457 * DLM_ASSERT_MASTER_MSG
1462 * NOTE: this can be used for debugging
1463 * can periodically run all locks owned by this node
1464 * and re-assert across the cluster...
1466 static int dlm_do_assert_master(struct dlm_ctxt
*dlm
, const char *lockname
,
1467 unsigned int namelen
, void *nodemap
,
1470 struct dlm_assert_master
assert;
1472 struct dlm_node_iter iter
;
1475 BUG_ON(namelen
> O2NM_MAX_NAME_LEN
);
1477 /* note that if this nodemap is empty, it returns 0 */
1478 dlm_node_iter_init(nodemap
, &iter
);
1479 while ((to
= dlm_node_iter_next(&iter
)) >= 0) {
1481 mlog(0, "sending assert master to %d (%.*s)\n", to
,
1483 memset(&assert, 0, sizeof(assert));
1484 assert.node_idx
= dlm
->node_num
;
1485 assert.namelen
= namelen
;
1486 memcpy(assert.name
, lockname
, namelen
);
1487 assert.flags
= cpu_to_be32(flags
);
1489 tmpret
= o2net_send_message(DLM_ASSERT_MASTER_MSG
, dlm
->key
,
1490 &assert, sizeof(assert), to
, &r
);
1492 mlog(ML_ERROR
, "assert_master returned %d!\n", tmpret
);
1493 if (!dlm_is_host_down(tmpret
)) {
1494 mlog(ML_ERROR
, "unhandled error!\n");
1497 /* a node died. finish out the rest of the nodes. */
1498 mlog(ML_ERROR
, "link to %d went down!\n", to
);
1499 /* any nonzero status return will do */
1502 /* ok, something horribly messed. kill thyself. */
1503 mlog(ML_ERROR
,"during assert master of %.*s to %u, "
1504 "got %d.\n", namelen
, lockname
, to
, r
);
1505 dlm_dump_lock_resources(dlm
);
1514 * locks that can be taken here:
1520 * if possible, TRIM THIS DOWN!!!
1522 int dlm_assert_master_handler(struct o2net_msg
*msg
, u32 len
, void *data
)
1524 struct dlm_ctxt
*dlm
= data
;
1525 struct dlm_master_list_entry
*mle
= NULL
;
1526 struct dlm_assert_master
*assert = (struct dlm_assert_master
*)msg
->buf
;
1527 struct dlm_lock_resource
*res
= NULL
;
1529 unsigned int namelen
;
1535 name
= assert->name
;
1536 namelen
= assert->namelen
;
1537 flags
= be32_to_cpu(assert->flags
);
1539 if (namelen
> DLM_LOCKID_NAME_MAX
) {
1540 mlog(ML_ERROR
, "Invalid name length!");
1544 spin_lock(&dlm
->spinlock
);
1547 mlog(0, "assert_master with flags: %u\n", flags
);
1550 spin_lock(&dlm
->master_lock
);
1551 if (!dlm_find_mle(dlm
, &mle
, name
, namelen
)) {
1552 /* not an error, could be master just re-asserting */
1553 mlog(0, "just got an assert_master from %u, but no "
1554 "MLE for it! (%.*s)\n", assert->node_idx
,
1557 int bit
= find_next_bit (mle
->maybe_map
, O2NM_MAX_NODES
, 0);
1558 if (bit
>= O2NM_MAX_NODES
) {
1559 /* not necessarily an error, though less likely.
1560 * could be master just re-asserting. */
1561 mlog(ML_ERROR
, "no bits set in the maybe_map, but %u "
1562 "is asserting! (%.*s)\n", assert->node_idx
,
1564 } else if (bit
!= assert->node_idx
) {
1565 if (flags
& DLM_ASSERT_MASTER_MLE_CLEANUP
) {
1566 mlog(0, "master %u was found, %u should "
1567 "back off\n", assert->node_idx
, bit
);
1569 /* with the fix for bug 569, a higher node
1570 * number winning the mastery will respond
1571 * YES to mastery requests, but this node
1572 * had no way of knowing. let it pass. */
1573 mlog(ML_ERROR
, "%u is the lowest node, "
1574 "%u is asserting. (%.*s) %u must "
1575 "have begun after %u won.\n", bit
,
1576 assert->node_idx
, namelen
, name
, bit
,
1581 spin_unlock(&dlm
->master_lock
);
1583 /* ok everything checks out with the MLE
1584 * now check to see if there is a lockres */
1585 res
= __dlm_lookup_lockres(dlm
, name
, namelen
);
1587 spin_lock(&res
->spinlock
);
1588 if (res
->state
& DLM_LOCK_RES_RECOVERING
) {
1589 mlog(ML_ERROR
, "%u asserting but %.*s is "
1590 "RECOVERING!\n", assert->node_idx
, namelen
, name
);
1594 if (res
->owner
!= assert->node_idx
) {
1595 mlog(ML_ERROR
, "assert_master from "
1596 "%u, but current owner is "
1598 assert->node_idx
, res
->owner
,
1602 } else if (mle
->type
!= DLM_MLE_MIGRATION
) {
1603 if (res
->owner
!= DLM_LOCK_RES_OWNER_UNKNOWN
) {
1604 /* owner is just re-asserting */
1605 if (res
->owner
== assert->node_idx
) {
1606 mlog(0, "owner %u re-asserting on "
1607 "lock %.*s\n", assert->node_idx
,
1611 mlog(ML_ERROR
, "got assert_master from "
1612 "node %u, but %u is the owner! "
1613 "(%.*s)\n", assert->node_idx
,
1614 res
->owner
, namelen
, name
);
1617 if (!(res
->state
& DLM_LOCK_RES_IN_PROGRESS
)) {
1618 mlog(ML_ERROR
, "got assert from %u, but lock "
1619 "with no owner should be "
1620 "in-progress! (%.*s)\n",
1625 } else /* mle->type == DLM_MLE_MIGRATION */ {
1626 /* should only be getting an assert from new master */
1627 if (assert->node_idx
!= mle
->new_master
) {
1628 mlog(ML_ERROR
, "got assert from %u, but "
1629 "new master is %u, and old master "
1631 assert->node_idx
, mle
->new_master
,
1632 mle
->master
, namelen
, name
);
1638 spin_unlock(&res
->spinlock
);
1640 spin_unlock(&dlm
->spinlock
);
1642 // mlog(0, "woo! got an assert_master from node %u!\n",
1643 // assert->node_idx);
1647 spin_lock(&mle
->spinlock
);
1648 extra_ref
= !!(mle
->type
== DLM_MLE_BLOCK
1649 || mle
->type
== DLM_MLE_MIGRATION
);
1650 mle
->master
= assert->node_idx
;
1651 atomic_set(&mle
->woken
, 1);
1653 spin_unlock(&mle
->spinlock
);
1655 if (mle
->type
== DLM_MLE_MIGRATION
&& res
) {
1656 mlog(0, "finishing off migration of lockres %.*s, "
1658 res
->lockname
.len
, res
->lockname
.name
,
1659 dlm
->node_num
, mle
->new_master
);
1660 spin_lock(&res
->spinlock
);
1661 res
->state
&= ~DLM_LOCK_RES_MIGRATING
;
1662 dlm_change_lockres_owner(dlm
, res
, mle
->new_master
);
1663 BUG_ON(res
->state
& DLM_LOCK_RES_DIRTY
);
1664 spin_unlock(&res
->spinlock
);
1666 /* master is known, detach if not already detached */
1667 dlm_mle_detach_hb_events(dlm
, mle
);
1671 /* the assert master message now balances the extra
1672 * ref given by the master / migration request message.
1673 * if this is the last put, it will be removed
1681 dlm_lockres_put(res
);
1686 /* kill the caller! */
1687 spin_unlock(&res
->spinlock
);
1688 spin_unlock(&dlm
->spinlock
);
1689 dlm_lockres_put(res
);
1690 mlog(ML_ERROR
, "Bad message received from another node. Dumping state "
1691 "and killing the other node now! This node is OK and can continue.\n");
1692 dlm_dump_lock_resources(dlm
);
1697 int dlm_dispatch_assert_master(struct dlm_ctxt
*dlm
,
1698 struct dlm_lock_resource
*res
,
1699 int ignore_higher
, u8 request_from
, u32 flags
)
1701 struct dlm_work_item
*item
;
1702 item
= kcalloc(1, sizeof(*item
), GFP_KERNEL
);
1707 /* queue up work for dlm_assert_master_worker */
1708 dlm_grab(dlm
); /* get an extra ref for the work item */
1709 dlm_init_work_item(dlm
, item
, dlm_assert_master_worker
, NULL
);
1710 item
->u
.am
.lockres
= res
; /* already have a ref */
1711 /* can optionally ignore node numbers higher than this node */
1712 item
->u
.am
.ignore_higher
= ignore_higher
;
1713 item
->u
.am
.request_from
= request_from
;
1714 item
->u
.am
.flags
= flags
;
1716 spin_lock(&dlm
->work_lock
);
1717 list_add_tail(&item
->list
, &dlm
->work_list
);
1718 spin_unlock(&dlm
->work_lock
);
1720 schedule_work(&dlm
->dispatched_work
);
1724 static void dlm_assert_master_worker(struct dlm_work_item
*item
, void *data
)
1726 struct dlm_ctxt
*dlm
= data
;
1728 struct dlm_lock_resource
*res
;
1729 unsigned long nodemap
[BITS_TO_LONGS(O2NM_MAX_NODES
)];
1736 res
= item
->u
.am
.lockres
;
1737 ignore_higher
= item
->u
.am
.ignore_higher
;
1738 request_from
= item
->u
.am
.request_from
;
1739 flags
= item
->u
.am
.flags
;
1741 spin_lock(&dlm
->spinlock
);
1742 memcpy(nodemap
, dlm
->domain_map
, sizeof(nodemap
));
1743 spin_unlock(&dlm
->spinlock
);
1745 clear_bit(dlm
->node_num
, nodemap
);
1746 if (ignore_higher
) {
1747 /* if is this just to clear up mles for nodes below
1748 * this node, do not send the message to the original
1749 * caller or any node number higher than this */
1750 clear_bit(request_from
, nodemap
);
1751 bit
= dlm
->node_num
;
1753 bit
= find_next_bit(nodemap
, O2NM_MAX_NODES
,
1755 if (bit
>= O2NM_MAX_NODES
)
1757 clear_bit(bit
, nodemap
);
1761 /* this call now finishes out the nodemap
1762 * even if one or more nodes die */
1763 mlog(0, "worker about to master %.*s here, this=%u\n",
1764 res
->lockname
.len
, res
->lockname
.name
, dlm
->node_num
);
1765 ret
= dlm_do_assert_master(dlm
, res
->lockname
.name
,
1769 /* no need to restart, we are done */
1773 dlm_lockres_put(res
);
1775 mlog(0, "finished with dlm_assert_master_worker\n");
1780 * DLM_MIGRATE_LOCKRES
1784 int dlm_migrate_lockres(struct dlm_ctxt
*dlm
, struct dlm_lock_resource
*res
,
1787 struct dlm_master_list_entry
*mle
= NULL
;
1788 struct dlm_master_list_entry
*oldmle
= NULL
;
1789 struct dlm_migratable_lockres
*mres
= NULL
;
1792 unsigned int namelen
;
1794 struct list_head
*queue
, *iter
;
1796 struct dlm_lock
*lock
;
1802 name
= res
->lockname
.name
;
1803 namelen
= res
->lockname
.len
;
1805 mlog(0, "migrating %.*s to %u\n", namelen
, name
, target
);
1808 * ensure this lockres is a proper candidate for migration
1810 spin_lock(&res
->spinlock
);
1811 if (res
->owner
== DLM_LOCK_RES_OWNER_UNKNOWN
) {
1812 mlog(0, "cannot migrate lockres with unknown owner!\n");
1813 spin_unlock(&res
->spinlock
);
1816 if (res
->owner
!= dlm
->node_num
) {
1817 mlog(0, "cannot migrate lockres this node doesn't own!\n");
1818 spin_unlock(&res
->spinlock
);
1821 mlog(0, "checking queues...\n");
1822 queue
= &res
->granted
;
1823 for (i
=0; i
<3; i
++) {
1824 list_for_each(iter
, queue
) {
1825 lock
= list_entry (iter
, struct dlm_lock
, list
);
1827 if (lock
->ml
.node
== dlm
->node_num
) {
1828 mlog(0, "found a lock owned by this node "
1829 "still on the %s queue! will not "
1830 "migrate this lockres\n",
1832 (i
==1 ? "converting" : "blocked"));
1833 spin_unlock(&res
->spinlock
);
1840 mlog(0, "all locks on this lockres are nonlocal. continuing\n");
1841 spin_unlock(&res
->spinlock
);
1845 mlog(0, "no locks were found on this lockres! done!\n");
1851 * preallocate up front
1852 * if this fails, abort
1856 mres
= (struct dlm_migratable_lockres
*) __get_free_page(GFP_KERNEL
);
1862 mle
= (struct dlm_master_list_entry
*) kmem_cache_alloc(dlm_mle_cache
,
1871 * find a node to migrate the lockres to
1874 mlog(0, "picking a migration node\n");
1875 spin_lock(&dlm
->spinlock
);
1876 /* pick a new node */
1877 if (!test_bit(target
, dlm
->domain_map
) ||
1878 target
>= O2NM_MAX_NODES
) {
1879 target
= dlm_pick_migration_target(dlm
, res
);
1881 mlog(0, "node %u chosen for migration\n", target
);
1883 if (target
>= O2NM_MAX_NODES
||
1884 !test_bit(target
, dlm
->domain_map
)) {
1885 /* target chosen is not alive */
1890 spin_unlock(&dlm
->spinlock
);
1894 mlog(0, "continuing with target = %u\n", target
);
1897 * clear any existing master requests and
1898 * add the migration mle to the list
1900 spin_lock(&dlm
->master_lock
);
1901 ret
= dlm_add_migration_mle(dlm
, res
, mle
, &oldmle
, name
,
1902 namelen
, target
, dlm
->node_num
);
1903 spin_unlock(&dlm
->master_lock
);
1904 spin_unlock(&dlm
->spinlock
);
1906 if (ret
== -EEXIST
) {
1907 mlog(0, "another process is already migrating it\n");
1913 * set the MIGRATING flag and flush asts
1914 * if we fail after this we need to re-dirty the lockres
1916 if (dlm_mark_lockres_migrating(dlm
, res
, target
) < 0) {
1917 mlog(ML_ERROR
, "tried to migrate %.*s to %u, but "
1918 "the target went down.\n", res
->lockname
.len
,
1919 res
->lockname
.name
, target
);
1920 spin_lock(&res
->spinlock
);
1921 res
->state
&= ~DLM_LOCK_RES_MIGRATING
;
1922 spin_unlock(&res
->spinlock
);
1928 /* master is known, detach if not already detached */
1929 dlm_mle_detach_hb_events(dlm
, oldmle
);
1930 dlm_put_mle(oldmle
);
1935 dlm_mle_detach_hb_events(dlm
, mle
);
1938 kmem_cache_free(dlm_mle_cache
, mle
);
1944 * at this point, we have a migration target, an mle
1945 * in the master list, and the MIGRATING flag set on
1950 /* get an extra reference on the mle.
1951 * otherwise the assert_master from the new
1952 * master will destroy this.
1953 * also, make sure that all callers of dlm_get_mle
1954 * take both dlm->spinlock and dlm->master_lock */
1955 spin_lock(&dlm
->spinlock
);
1956 spin_lock(&dlm
->master_lock
);
1958 spin_unlock(&dlm
->master_lock
);
1959 spin_unlock(&dlm
->spinlock
);
1961 /* notify new node and send all lock state */
1962 /* call send_one_lockres with migration flag.
1963 * this serves as notice to the target node that a
1964 * migration is starting. */
1965 ret
= dlm_send_one_lockres(dlm
, res
, mres
, target
,
1966 DLM_MRES_MIGRATION
);
1969 mlog(0, "migration to node %u failed with %d\n",
1971 /* migration failed, detach and clean up mle */
1972 dlm_mle_detach_hb_events(dlm
, mle
);
1978 /* at this point, the target sends a message to all nodes,
1979 * (using dlm_do_migrate_request). this node is skipped since
1980 * we had to put an mle in the list to begin the process. this
1981 * node now waits for target to do an assert master. this node
1982 * will be the last one notified, ensuring that the migration
1983 * is complete everywhere. if the target dies while this is
1984 * going on, some nodes could potentially see the target as the
1985 * master, so it is important that my recovery finds the migration
1986 * mle and sets the master to UNKNONWN. */
1989 /* wait for new node to assert master */
1991 ret
= wait_event_interruptible_timeout(mle
->wq
,
1992 (atomic_read(&mle
->woken
) == 1),
1993 msecs_to_jiffies(5000));
1996 if (atomic_read(&mle
->woken
) == 1 ||
1997 res
->owner
== target
)
2000 mlog(0, "timed out during migration\n");
2001 /* avoid hang during shutdown when migrating lockres
2002 * to a node which also goes down */
2003 if (dlm_is_node_dead(dlm
, target
)) {
2004 mlog(0, "%s:%.*s: expected migration target %u "
2005 "is no longer up. restarting.\n",
2006 dlm
->name
, res
->lockname
.len
,
2007 res
->lockname
.name
, target
);
2011 if (ret
== -ERESTARTSYS
) {
2012 /* migration failed, detach and clean up mle */
2013 dlm_mle_detach_hb_events(dlm
, mle
);
2018 /* TODO: if node died: stop, clean up, return error */
2021 /* all done, set the owner, clear the flag */
2022 spin_lock(&res
->spinlock
);
2023 dlm_set_lockres_owner(dlm
, res
, target
);
2024 res
->state
&= ~DLM_LOCK_RES_MIGRATING
;
2025 dlm_remove_nonlocal_locks(dlm
, res
);
2026 spin_unlock(&res
->spinlock
);
2029 /* master is known, detach if not already detached */
2030 dlm_mle_detach_hb_events(dlm
, mle
);
2034 dlm_lockres_calc_usage(dlm
, res
);
2037 /* re-dirty the lockres if we failed */
2039 dlm_kick_thread(dlm
, res
);
2043 free_page((unsigned long)mres
);
2047 mlog(0, "returning %d\n", ret
);
2050 EXPORT_SYMBOL_GPL(dlm_migrate_lockres
);
2052 int dlm_lock_basts_flushed(struct dlm_ctxt
*dlm
, struct dlm_lock
*lock
)
2055 spin_lock(&dlm
->ast_lock
);
2056 spin_lock(&lock
->spinlock
);
2057 ret
= (list_empty(&lock
->bast_list
) && !lock
->bast_pending
);
2058 spin_unlock(&lock
->spinlock
);
2059 spin_unlock(&dlm
->ast_lock
);
2063 static int dlm_migration_can_proceed(struct dlm_ctxt
*dlm
,
2064 struct dlm_lock_resource
*res
,
2068 spin_lock(&res
->spinlock
);
2069 can_proceed
= !!(res
->state
& DLM_LOCK_RES_MIGRATING
);
2070 spin_unlock(&res
->spinlock
);
2072 /* target has died, so make the caller break out of the
2073 * wait_event, but caller must recheck the domain_map */
2074 spin_lock(&dlm
->spinlock
);
2075 if (!test_bit(mig_target
, dlm
->domain_map
))
2077 spin_unlock(&dlm
->spinlock
);
2081 int dlm_lockres_is_dirty(struct dlm_ctxt
*dlm
, struct dlm_lock_resource
*res
)
2084 spin_lock(&res
->spinlock
);
2085 ret
= !!(res
->state
& DLM_LOCK_RES_DIRTY
);
2086 spin_unlock(&res
->spinlock
);
2091 static int dlm_mark_lockres_migrating(struct dlm_ctxt
*dlm
,
2092 struct dlm_lock_resource
*res
,
2097 mlog(0, "dlm_mark_lockres_migrating: %.*s, from %u to %u\n",
2098 res
->lockname
.len
, res
->lockname
.name
, dlm
->node_num
,
2100 /* need to set MIGRATING flag on lockres. this is done by
2101 * ensuring that all asts have been flushed for this lockres. */
2102 spin_lock(&res
->spinlock
);
2103 BUG_ON(res
->migration_pending
);
2104 res
->migration_pending
= 1;
2105 /* strategy is to reserve an extra ast then release
2106 * it below, letting the release do all of the work */
2107 __dlm_lockres_reserve_ast(res
);
2108 spin_unlock(&res
->spinlock
);
2110 /* now flush all the pending asts.. hang out for a bit */
2111 dlm_kick_thread(dlm
, res
);
2112 wait_event(dlm
->ast_wq
, !dlm_lockres_is_dirty(dlm
, res
));
2113 dlm_lockres_release_ast(dlm
, res
);
2115 mlog(0, "about to wait on migration_wq, dirty=%s\n",
2116 res
->state
& DLM_LOCK_RES_DIRTY
? "yes" : "no");
2117 /* if the extra ref we just put was the final one, this
2118 * will pass thru immediately. otherwise, we need to wait
2119 * for the last ast to finish. */
2121 ret
= wait_event_interruptible_timeout(dlm
->migration_wq
,
2122 dlm_migration_can_proceed(dlm
, res
, target
),
2123 msecs_to_jiffies(1000));
2125 mlog(0, "woken again: migrating? %s, dead? %s\n",
2126 res
->state
& DLM_LOCK_RES_MIGRATING
? "yes":"no",
2127 test_bit(target
, dlm
->domain_map
) ? "no":"yes");
2129 mlog(0, "all is well: migrating? %s, dead? %s\n",
2130 res
->state
& DLM_LOCK_RES_MIGRATING
? "yes":"no",
2131 test_bit(target
, dlm
->domain_map
) ? "no":"yes");
2133 if (!dlm_migration_can_proceed(dlm
, res
, target
)) {
2134 mlog(0, "trying again...\n");
2138 /* did the target go down or die? */
2139 spin_lock(&dlm
->spinlock
);
2140 if (!test_bit(target
, dlm
->domain_map
)) {
2141 mlog(ML_ERROR
, "aha. migration target %u just went down\n",
2145 spin_unlock(&dlm
->spinlock
);
2150 * o the DLM_LOCK_RES_MIGRATING flag is set
2151 * o there are no pending asts on this lockres
2152 * o all processes trying to reserve an ast on this
2153 * lockres must wait for the MIGRATING flag to clear
2158 /* last step in the migration process.
2159 * original master calls this to free all of the dlm_lock
2160 * structures that used to be for other nodes. */
2161 static void dlm_remove_nonlocal_locks(struct dlm_ctxt
*dlm
,
2162 struct dlm_lock_resource
*res
)
2164 struct list_head
*iter
, *iter2
;
2165 struct list_head
*queue
= &res
->granted
;
2167 struct dlm_lock
*lock
;
2169 assert_spin_locked(&res
->spinlock
);
2171 BUG_ON(res
->owner
== dlm
->node_num
);
2173 for (i
=0; i
<3; i
++) {
2174 list_for_each_safe(iter
, iter2
, queue
) {
2175 lock
= list_entry (iter
, struct dlm_lock
, list
);
2176 if (lock
->ml
.node
!= dlm
->node_num
) {
2177 mlog(0, "putting lock for node %u\n",
2179 /* be extra careful */
2180 BUG_ON(!list_empty(&lock
->ast_list
));
2181 BUG_ON(!list_empty(&lock
->bast_list
));
2182 BUG_ON(lock
->ast_pending
);
2183 BUG_ON(lock
->bast_pending
);
2184 list_del_init(&lock
->list
);
2192 /* for now this is not too intelligent. we will
2193 * need stats to make this do the right thing.
2194 * this just finds the first lock on one of the
2195 * queues and uses that node as the target. */
2196 static u8
dlm_pick_migration_target(struct dlm_ctxt
*dlm
,
2197 struct dlm_lock_resource
*res
)
2200 struct list_head
*queue
= &res
->granted
;
2201 struct list_head
*iter
;
2202 struct dlm_lock
*lock
;
2205 assert_spin_locked(&dlm
->spinlock
);
2207 spin_lock(&res
->spinlock
);
2208 for (i
=0; i
<3; i
++) {
2209 list_for_each(iter
, queue
) {
2210 /* up to the caller to make sure this node
2212 lock
= list_entry (iter
, struct dlm_lock
, list
);
2213 if (lock
->ml
.node
!= dlm
->node_num
) {
2214 spin_unlock(&res
->spinlock
);
2215 return lock
->ml
.node
;
2220 spin_unlock(&res
->spinlock
);
2221 mlog(0, "have not found a suitable target yet! checking domain map\n");
2223 /* ok now we're getting desperate. pick anyone alive. */
2226 nodenum
= find_next_bit(dlm
->domain_map
,
2227 O2NM_MAX_NODES
, nodenum
+1);
2228 mlog(0, "found %d in domain map\n", nodenum
);
2229 if (nodenum
>= O2NM_MAX_NODES
)
2231 if (nodenum
!= dlm
->node_num
) {
2232 mlog(0, "picking %d\n", nodenum
);
2237 mlog(0, "giving up. no master to migrate to\n");
2238 return DLM_LOCK_RES_OWNER_UNKNOWN
;
2243 /* this is called by the new master once all lockres
2244 * data has been received */
2245 static int dlm_do_migrate_request(struct dlm_ctxt
*dlm
,
2246 struct dlm_lock_resource
*res
,
2247 u8 master
, u8 new_master
,
2248 struct dlm_node_iter
*iter
)
2250 struct dlm_migrate_request migrate
;
2251 int ret
, status
= 0;
2254 memset(&migrate
, 0, sizeof(migrate
));
2255 migrate
.namelen
= res
->lockname
.len
;
2256 memcpy(migrate
.name
, res
->lockname
.name
, migrate
.namelen
);
2257 migrate
.new_master
= new_master
;
2258 migrate
.master
= master
;
2262 /* send message to all nodes, except the master and myself */
2263 while ((nodenum
= dlm_node_iter_next(iter
)) >= 0) {
2264 if (nodenum
== master
||
2265 nodenum
== new_master
)
2268 ret
= o2net_send_message(DLM_MIGRATE_REQUEST_MSG
, dlm
->key
,
2269 &migrate
, sizeof(migrate
), nodenum
,
2273 else if (status
< 0) {
2274 mlog(0, "migrate request (node %u) returned %d!\n",
2283 mlog(0, "returning ret=%d\n", ret
);
2288 /* if there is an existing mle for this lockres, we now know who the master is.
2289 * (the one who sent us *this* message) we can clear it up right away.
2290 * since the process that put the mle on the list still has a reference to it,
2291 * we can unhash it now, set the master and wake the process. as a result,
2292 * we will have no mle in the list to start with. now we can add an mle for
2293 * the migration and this should be the only one found for those scanning the
2295 int dlm_migrate_request_handler(struct o2net_msg
*msg
, u32 len
, void *data
)
2297 struct dlm_ctxt
*dlm
= data
;
2298 struct dlm_lock_resource
*res
= NULL
;
2299 struct dlm_migrate_request
*migrate
= (struct dlm_migrate_request
*) msg
->buf
;
2300 struct dlm_master_list_entry
*mle
= NULL
, *oldmle
= NULL
;
2302 unsigned int namelen
;
2308 name
= migrate
->name
;
2309 namelen
= migrate
->namelen
;
2311 /* preallocate.. if this fails, abort */
2312 mle
= (struct dlm_master_list_entry
*) kmem_cache_alloc(dlm_mle_cache
,
2320 /* check for pre-existing lock */
2321 spin_lock(&dlm
->spinlock
);
2322 res
= __dlm_lookup_lockres(dlm
, name
, namelen
);
2323 spin_lock(&dlm
->master_lock
);
2326 spin_lock(&res
->spinlock
);
2327 if (res
->state
& DLM_LOCK_RES_RECOVERING
) {
2328 /* if all is working ok, this can only mean that we got
2329 * a migrate request from a node that we now see as
2330 * dead. what can we do here? drop it to the floor? */
2331 spin_unlock(&res
->spinlock
);
2332 mlog(ML_ERROR
, "Got a migrate request, but the "
2333 "lockres is marked as recovering!");
2334 kmem_cache_free(dlm_mle_cache
, mle
);
2335 ret
= -EINVAL
; /* need a better solution */
2338 res
->state
|= DLM_LOCK_RES_MIGRATING
;
2339 spin_unlock(&res
->spinlock
);
2342 /* ignore status. only nonzero status would BUG. */
2343 ret
= dlm_add_migration_mle(dlm
, res
, mle
, &oldmle
,
2345 migrate
->new_master
,
2349 spin_unlock(&dlm
->master_lock
);
2350 spin_unlock(&dlm
->spinlock
);
2353 /* master is known, detach if not already detached */
2354 dlm_mle_detach_hb_events(dlm
, oldmle
);
2355 dlm_put_mle(oldmle
);
2359 dlm_lockres_put(res
);
2365 /* must be holding dlm->spinlock and dlm->master_lock
2366 * when adding a migration mle, we can clear any other mles
2367 * in the master list because we know with certainty that
2368 * the master is "master". so we remove any old mle from
2369 * the list after setting it's master field, and then add
2370 * the new migration mle. this way we can hold with the rule
2371 * of having only one mle for a given lock name at all times. */
2372 static int dlm_add_migration_mle(struct dlm_ctxt
*dlm
,
2373 struct dlm_lock_resource
*res
,
2374 struct dlm_master_list_entry
*mle
,
2375 struct dlm_master_list_entry
**oldmle
,
2376 const char *name
, unsigned int namelen
,
2377 u8 new_master
, u8 master
)
2386 assert_spin_locked(&dlm
->spinlock
);
2387 assert_spin_locked(&dlm
->master_lock
);
2389 /* caller is responsible for any ref taken here on oldmle */
2390 found
= dlm_find_mle(dlm
, oldmle
, (char *)name
, namelen
);
2392 struct dlm_master_list_entry
*tmp
= *oldmle
;
2393 spin_lock(&tmp
->spinlock
);
2394 if (tmp
->type
== DLM_MLE_MIGRATION
) {
2395 if (master
== dlm
->node_num
) {
2396 /* ah another process raced me to it */
2397 mlog(0, "tried to migrate %.*s, but some "
2398 "process beat me to it\n",
2402 /* bad. 2 NODES are trying to migrate! */
2403 mlog(ML_ERROR
, "migration error mle: "
2404 "master=%u new_master=%u // request: "
2405 "master=%u new_master=%u // "
2407 tmp
->master
, tmp
->new_master
,
2413 /* this is essentially what assert_master does */
2414 tmp
->master
= master
;
2415 atomic_set(&tmp
->woken
, 1);
2417 /* remove it from the list so that only one
2418 * mle will be found */
2419 list_del_init(&tmp
->list
);
2421 spin_unlock(&tmp
->spinlock
);
2424 /* now add a migration mle to the tail of the list */
2425 dlm_init_mle(mle
, DLM_MLE_MIGRATION
, dlm
, res
, name
, namelen
);
2426 mle
->new_master
= new_master
;
2427 mle
->master
= master
;
2428 /* do this for consistency with other mle types */
2429 set_bit(new_master
, mle
->maybe_map
);
2430 list_add(&mle
->list
, &dlm
->master_list
);
2436 void dlm_clean_master_list(struct dlm_ctxt
*dlm
, u8 dead_node
)
2438 struct list_head
*iter
, *iter2
;
2439 struct dlm_master_list_entry
*mle
;
2440 struct dlm_lock_resource
*res
;
2442 mlog_entry("dlm=%s, dead node=%u\n", dlm
->name
, dead_node
);
2444 assert_spin_locked(&dlm
->spinlock
);
2446 /* clean the master list */
2447 spin_lock(&dlm
->master_lock
);
2448 list_for_each_safe(iter
, iter2
, &dlm
->master_list
) {
2449 mle
= list_entry(iter
, struct dlm_master_list_entry
, list
);
2451 BUG_ON(mle
->type
!= DLM_MLE_BLOCK
&&
2452 mle
->type
!= DLM_MLE_MASTER
&&
2453 mle
->type
!= DLM_MLE_MIGRATION
);
2455 /* MASTER mles are initiated locally. the waiting
2456 * process will notice the node map change
2457 * shortly. let that happen as normal. */
2458 if (mle
->type
== DLM_MLE_MASTER
)
2462 /* BLOCK mles are initiated by other nodes.
2463 * need to clean up if the dead node would have
2464 * been the master. */
2465 if (mle
->type
== DLM_MLE_BLOCK
) {
2468 spin_lock(&mle
->spinlock
);
2469 bit
= find_next_bit(mle
->maybe_map
, O2NM_MAX_NODES
, 0);
2470 if (bit
!= dead_node
) {
2471 mlog(0, "mle found, but dead node %u would "
2472 "not have been master\n", dead_node
);
2473 spin_unlock(&mle
->spinlock
);
2475 /* must drop the refcount by one since the
2476 * assert_master will never arrive. this
2477 * may result in the mle being unlinked and
2478 * freed, but there may still be a process
2479 * waiting in the dlmlock path which is fine. */
2480 mlog(ML_ERROR
, "node %u was expected master\n",
2482 atomic_set(&mle
->woken
, 1);
2483 spin_unlock(&mle
->spinlock
);
2485 /* do not need events any longer, so detach
2487 __dlm_mle_detach_hb_events(dlm
, mle
);
2493 /* everything else is a MIGRATION mle */
2495 /* the rule for MIGRATION mles is that the master
2496 * becomes UNKNOWN if *either* the original or
2497 * the new master dies. all UNKNOWN lockreses
2498 * are sent to whichever node becomes the recovery
2499 * master. the new master is responsible for
2500 * determining if there is still a master for
2501 * this lockres, or if he needs to take over
2502 * mastery. either way, this node should expect
2503 * another message to resolve this. */
2504 if (mle
->master
!= dead_node
&&
2505 mle
->new_master
!= dead_node
)
2508 /* if we have reached this point, this mle needs to
2509 * be removed from the list and freed. */
2511 /* remove from the list early. NOTE: unlinking
2512 * list_head while in list_for_each_safe */
2513 spin_lock(&mle
->spinlock
);
2514 list_del_init(&mle
->list
);
2515 atomic_set(&mle
->woken
, 1);
2516 spin_unlock(&mle
->spinlock
);
2519 mlog(0, "node %u died during migration from "
2520 "%u to %u!\n", dead_node
,
2521 mle
->master
, mle
->new_master
);
2522 /* if there is a lockres associated with this
2523 * mle, find it and set its owner to UNKNOWN */
2524 res
= __dlm_lookup_lockres(dlm
, mle
->u
.name
.name
,
2527 /* unfortunately if we hit this rare case, our
2528 * lock ordering is messed. we need to drop
2529 * the master lock so that we can take the
2530 * lockres lock, meaning that we will have to
2531 * restart from the head of list. */
2532 spin_unlock(&dlm
->master_lock
);
2534 /* move lockres onto recovery list */
2535 spin_lock(&res
->spinlock
);
2536 dlm_set_lockres_owner(dlm
, res
,
2537 DLM_LOCK_RES_OWNER_UNKNOWN
);
2538 dlm_move_lockres_to_recovery_list(dlm
, res
);
2539 spin_unlock(&res
->spinlock
);
2540 dlm_lockres_put(res
);
2542 /* about to get rid of mle, detach from heartbeat */
2543 __dlm_mle_detach_hb_events(dlm
, mle
);
2546 spin_lock(&dlm
->master_lock
);
2548 spin_unlock(&dlm
->master_lock
);
2554 /* this may be the last reference */
2557 spin_unlock(&dlm
->master_lock
);
2561 int dlm_finish_migration(struct dlm_ctxt
*dlm
, struct dlm_lock_resource
*res
,
2564 struct dlm_node_iter iter
;
2567 spin_lock(&dlm
->spinlock
);
2568 dlm_node_iter_init(dlm
->domain_map
, &iter
);
2569 clear_bit(old_master
, iter
.node_map
);
2570 clear_bit(dlm
->node_num
, iter
.node_map
);
2571 spin_unlock(&dlm
->spinlock
);
2573 mlog(0, "now time to do a migrate request to other nodes\n");
2574 ret
= dlm_do_migrate_request(dlm
, res
, old_master
,
2575 dlm
->node_num
, &iter
);
2581 mlog(0, "doing assert master of %.*s to all except the original node\n",
2582 res
->lockname
.len
, res
->lockname
.name
);
2583 /* this call now finishes out the nodemap
2584 * even if one or more nodes die */
2585 ret
= dlm_do_assert_master(dlm
, res
->lockname
.name
,
2586 res
->lockname
.len
, iter
.node_map
,
2587 DLM_ASSERT_MASTER_FINISH_MIGRATION
);
2589 /* no longer need to retry. all living nodes contacted. */
2594 memset(iter
.node_map
, 0, sizeof(iter
.node_map
));
2595 set_bit(old_master
, iter
.node_map
);
2596 mlog(0, "doing assert master of %.*s back to %u\n",
2597 res
->lockname
.len
, res
->lockname
.name
, old_master
);
2598 ret
= dlm_do_assert_master(dlm
, res
->lockname
.name
,
2599 res
->lockname
.len
, iter
.node_map
,
2600 DLM_ASSERT_MASTER_FINISH_MIGRATION
);
2602 mlog(0, "assert master to original master failed "
2604 /* the only nonzero status here would be because of
2605 * a dead original node. we're done. */
2609 /* all done, set the owner, clear the flag */
2610 spin_lock(&res
->spinlock
);
2611 dlm_set_lockres_owner(dlm
, res
, dlm
->node_num
);
2612 res
->state
&= ~DLM_LOCK_RES_MIGRATING
;
2613 spin_unlock(&res
->spinlock
);
2614 /* re-dirty it on the new master */
2615 dlm_kick_thread(dlm
, res
);
2622 * LOCKRES AST REFCOUNT
2623 * this is integral to migration
2626 /* for future intent to call an ast, reserve one ahead of time.
2627 * this should be called only after waiting on the lockres
2628 * with dlm_wait_on_lockres, and while still holding the
2629 * spinlock after the call. */
2630 void __dlm_lockres_reserve_ast(struct dlm_lock_resource
*res
)
2632 assert_spin_locked(&res
->spinlock
);
2633 if (res
->state
& DLM_LOCK_RES_MIGRATING
) {
2634 __dlm_print_one_lock_resource(res
);
2636 BUG_ON(res
->state
& DLM_LOCK_RES_MIGRATING
);
2638 atomic_inc(&res
->asts_reserved
);
2642 * used to drop the reserved ast, either because it went unused,
2643 * or because the ast/bast was actually called.
2645 * also, if there is a pending migration on this lockres,
2646 * and this was the last pending ast on the lockres,
2647 * atomically set the MIGRATING flag before we drop the lock.
2648 * this is how we ensure that migration can proceed with no
2649 * asts in progress. note that it is ok if the state of the
2650 * queues is such that a lock should be granted in the future
2651 * or that a bast should be fired, because the new master will
2652 * shuffle the lists on this lockres as soon as it is migrated.
2654 void dlm_lockres_release_ast(struct dlm_ctxt
*dlm
,
2655 struct dlm_lock_resource
*res
)
2657 if (!atomic_dec_and_lock(&res
->asts_reserved
, &res
->spinlock
))
2660 if (!res
->migration_pending
) {
2661 spin_unlock(&res
->spinlock
);
2665 BUG_ON(res
->state
& DLM_LOCK_RES_MIGRATING
);
2666 res
->migration_pending
= 0;
2667 res
->state
|= DLM_LOCK_RES_MIGRATING
;
2668 spin_unlock(&res
->spinlock
);
2670 wake_up(&dlm
->migration_wq
);