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
,
242 static int dlm_pre_master_reco_lockres(struct dlm_ctxt
*dlm
,
243 struct dlm_lock_resource
*res
);
246 int dlm_is_host_down(int errno
)
263 case -EINVAL
: /* if returned from our tcp code,
264 this means there is no socket */
272 * MASTER LIST FUNCTIONS
277 * regarding master list entries and heartbeat callbacks:
279 * in order to avoid sleeping and allocation that occurs in
280 * heartbeat, master list entries are simply attached to the
281 * dlm's established heartbeat callbacks. the mle is attached
282 * when it is created, and since the dlm->spinlock is held at
283 * that time, any heartbeat event will be properly discovered
284 * by the mle. the mle needs to be detached from the
285 * dlm->mle_hb_events list as soon as heartbeat events are no
286 * longer useful to the mle, and before the mle is freed.
288 * as a general rule, heartbeat events are no longer needed by
289 * the mle once an "answer" regarding the lock master has been
292 static inline void __dlm_mle_attach_hb_events(struct dlm_ctxt
*dlm
,
293 struct dlm_master_list_entry
*mle
)
295 assert_spin_locked(&dlm
->spinlock
);
297 list_add_tail(&mle
->hb_events
, &dlm
->mle_hb_events
);
301 static inline void __dlm_mle_detach_hb_events(struct dlm_ctxt
*dlm
,
302 struct dlm_master_list_entry
*mle
)
304 if (!list_empty(&mle
->hb_events
))
305 list_del_init(&mle
->hb_events
);
309 static inline void dlm_mle_detach_hb_events(struct dlm_ctxt
*dlm
,
310 struct dlm_master_list_entry
*mle
)
312 spin_lock(&dlm
->spinlock
);
313 __dlm_mle_detach_hb_events(dlm
, mle
);
314 spin_unlock(&dlm
->spinlock
);
317 /* remove from list and free */
318 static void __dlm_put_mle(struct dlm_master_list_entry
*mle
)
320 struct dlm_ctxt
*dlm
;
323 assert_spin_locked(&dlm
->spinlock
);
324 assert_spin_locked(&dlm
->master_lock
);
325 BUG_ON(!atomic_read(&mle
->mle_refs
.refcount
));
327 kref_put(&mle
->mle_refs
, dlm_mle_release
);
331 /* must not have any spinlocks coming in */
332 static void dlm_put_mle(struct dlm_master_list_entry
*mle
)
334 struct dlm_ctxt
*dlm
;
337 spin_lock(&dlm
->spinlock
);
338 spin_lock(&dlm
->master_lock
);
340 spin_unlock(&dlm
->master_lock
);
341 spin_unlock(&dlm
->spinlock
);
344 static inline void dlm_get_mle(struct dlm_master_list_entry
*mle
)
346 kref_get(&mle
->mle_refs
);
349 static void dlm_init_mle(struct dlm_master_list_entry
*mle
,
350 enum dlm_mle_type type
,
351 struct dlm_ctxt
*dlm
,
352 struct dlm_lock_resource
*res
,
354 unsigned int namelen
)
356 assert_spin_locked(&dlm
->spinlock
);
360 INIT_LIST_HEAD(&mle
->list
);
361 INIT_LIST_HEAD(&mle
->hb_events
);
362 memset(mle
->maybe_map
, 0, sizeof(mle
->maybe_map
));
363 spin_lock_init(&mle
->spinlock
);
364 init_waitqueue_head(&mle
->wq
);
365 atomic_set(&mle
->woken
, 0);
366 kref_init(&mle
->mle_refs
);
367 memset(mle
->response_map
, 0, sizeof(mle
->response_map
));
368 mle
->master
= O2NM_MAX_NODES
;
369 mle
->new_master
= O2NM_MAX_NODES
;
371 if (mle
->type
== DLM_MLE_MASTER
) {
374 } else if (mle
->type
== DLM_MLE_BLOCK
) {
376 memcpy(mle
->u
.name
.name
, name
, namelen
);
377 mle
->u
.name
.len
= namelen
;
378 } else /* DLM_MLE_MIGRATION */ {
380 memcpy(mle
->u
.name
.name
, name
, namelen
);
381 mle
->u
.name
.len
= namelen
;
384 /* copy off the node_map and register hb callbacks on our copy */
385 memcpy(mle
->node_map
, dlm
->domain_map
, sizeof(mle
->node_map
));
386 memcpy(mle
->vote_map
, dlm
->domain_map
, sizeof(mle
->vote_map
));
387 clear_bit(dlm
->node_num
, mle
->vote_map
);
388 clear_bit(dlm
->node_num
, mle
->node_map
);
390 /* attach the mle to the domain node up/down events */
391 __dlm_mle_attach_hb_events(dlm
, mle
);
395 /* returns 1 if found, 0 if not */
396 static int dlm_find_mle(struct dlm_ctxt
*dlm
,
397 struct dlm_master_list_entry
**mle
,
398 char *name
, unsigned int namelen
)
400 struct dlm_master_list_entry
*tmpmle
;
401 struct list_head
*iter
;
403 assert_spin_locked(&dlm
->master_lock
);
405 list_for_each(iter
, &dlm
->master_list
) {
406 tmpmle
= list_entry(iter
, struct dlm_master_list_entry
, list
);
407 if (!dlm_mle_equal(dlm
, tmpmle
, name
, namelen
))
416 void dlm_hb_event_notify_attached(struct dlm_ctxt
*dlm
, int idx
, int node_up
)
418 struct dlm_master_list_entry
*mle
;
419 struct list_head
*iter
;
421 assert_spin_locked(&dlm
->spinlock
);
423 list_for_each(iter
, &dlm
->mle_hb_events
) {
424 mle
= list_entry(iter
, struct dlm_master_list_entry
,
427 dlm_mle_node_up(dlm
, mle
, NULL
, idx
);
429 dlm_mle_node_down(dlm
, mle
, NULL
, idx
);
433 static void dlm_mle_node_down(struct dlm_ctxt
*dlm
,
434 struct dlm_master_list_entry
*mle
,
435 struct o2nm_node
*node
, int idx
)
437 spin_lock(&mle
->spinlock
);
439 if (!test_bit(idx
, mle
->node_map
))
440 mlog(0, "node %u already removed from nodemap!\n", idx
);
442 clear_bit(idx
, mle
->node_map
);
444 spin_unlock(&mle
->spinlock
);
447 static void dlm_mle_node_up(struct dlm_ctxt
*dlm
,
448 struct dlm_master_list_entry
*mle
,
449 struct o2nm_node
*node
, int idx
)
451 spin_lock(&mle
->spinlock
);
453 if (test_bit(idx
, mle
->node_map
))
454 mlog(0, "node %u already in node map!\n", idx
);
456 set_bit(idx
, mle
->node_map
);
458 spin_unlock(&mle
->spinlock
);
462 int dlm_init_mle_cache(void)
464 dlm_mle_cache
= kmem_cache_create("dlm_mle_cache",
465 sizeof(struct dlm_master_list_entry
),
466 0, SLAB_HWCACHE_ALIGN
,
468 if (dlm_mle_cache
== NULL
)
473 void dlm_destroy_mle_cache(void)
476 kmem_cache_destroy(dlm_mle_cache
);
479 static void dlm_mle_release(struct kref
*kref
)
481 struct dlm_master_list_entry
*mle
;
482 struct dlm_ctxt
*dlm
;
486 mle
= container_of(kref
, struct dlm_master_list_entry
, mle_refs
);
489 if (mle
->type
!= DLM_MLE_MASTER
) {
490 mlog(0, "calling mle_release for %.*s, type %d\n",
491 mle
->u
.name
.len
, mle
->u
.name
.name
, mle
->type
);
493 mlog(0, "calling mle_release for %.*s, type %d\n",
494 mle
->u
.res
->lockname
.len
,
495 mle
->u
.res
->lockname
.name
, mle
->type
);
497 assert_spin_locked(&dlm
->spinlock
);
498 assert_spin_locked(&dlm
->master_lock
);
500 /* remove from list if not already */
501 if (!list_empty(&mle
->list
))
502 list_del_init(&mle
->list
);
504 /* detach the mle from the domain node up/down events */
505 __dlm_mle_detach_hb_events(dlm
, mle
);
507 /* NOTE: kfree under spinlock here.
508 * if this is bad, we can move this to a freelist. */
509 kmem_cache_free(dlm_mle_cache
, mle
);
514 * LOCK RESOURCE FUNCTIONS
517 static void dlm_set_lockres_owner(struct dlm_ctxt
*dlm
,
518 struct dlm_lock_resource
*res
,
521 assert_spin_locked(&res
->spinlock
);
523 mlog_entry("%.*s, %u\n", res
->lockname
.len
, res
->lockname
.name
, owner
);
525 if (owner
== dlm
->node_num
)
526 atomic_inc(&dlm
->local_resources
);
527 else if (owner
== DLM_LOCK_RES_OWNER_UNKNOWN
)
528 atomic_inc(&dlm
->unknown_resources
);
530 atomic_inc(&dlm
->remote_resources
);
535 void dlm_change_lockres_owner(struct dlm_ctxt
*dlm
,
536 struct dlm_lock_resource
*res
, u8 owner
)
538 assert_spin_locked(&res
->spinlock
);
540 if (owner
== res
->owner
)
543 if (res
->owner
== dlm
->node_num
)
544 atomic_dec(&dlm
->local_resources
);
545 else if (res
->owner
== DLM_LOCK_RES_OWNER_UNKNOWN
)
546 atomic_dec(&dlm
->unknown_resources
);
548 atomic_dec(&dlm
->remote_resources
);
550 dlm_set_lockres_owner(dlm
, res
, owner
);
554 static void dlm_lockres_release(struct kref
*kref
)
556 struct dlm_lock_resource
*res
;
558 res
= container_of(kref
, struct dlm_lock_resource
, refs
);
560 /* This should not happen -- all lockres' have a name
561 * associated with them at init time. */
562 BUG_ON(!res
->lockname
.name
);
564 mlog(0, "destroying lockres %.*s\n", res
->lockname
.len
,
567 /* By the time we're ready to blow this guy away, we shouldn't
568 * be on any lists. */
569 BUG_ON(!hlist_unhashed(&res
->hash_node
));
570 BUG_ON(!list_empty(&res
->granted
));
571 BUG_ON(!list_empty(&res
->converting
));
572 BUG_ON(!list_empty(&res
->blocked
));
573 BUG_ON(!list_empty(&res
->dirty
));
574 BUG_ON(!list_empty(&res
->recovering
));
575 BUG_ON(!list_empty(&res
->purge
));
577 kfree(res
->lockname
.name
);
582 void dlm_lockres_put(struct dlm_lock_resource
*res
)
584 kref_put(&res
->refs
, dlm_lockres_release
);
587 static void dlm_init_lockres(struct dlm_ctxt
*dlm
,
588 struct dlm_lock_resource
*res
,
589 const char *name
, unsigned int namelen
)
593 /* If we memset here, we lose our reference to the kmalloc'd
594 * res->lockname.name, so be sure to init every field
597 qname
= (char *) res
->lockname
.name
;
598 memcpy(qname
, name
, namelen
);
600 res
->lockname
.len
= namelen
;
601 res
->lockname
.hash
= dlm_lockid_hash(name
, namelen
);
603 init_waitqueue_head(&res
->wq
);
604 spin_lock_init(&res
->spinlock
);
605 INIT_HLIST_NODE(&res
->hash_node
);
606 INIT_LIST_HEAD(&res
->granted
);
607 INIT_LIST_HEAD(&res
->converting
);
608 INIT_LIST_HEAD(&res
->blocked
);
609 INIT_LIST_HEAD(&res
->dirty
);
610 INIT_LIST_HEAD(&res
->recovering
);
611 INIT_LIST_HEAD(&res
->purge
);
612 atomic_set(&res
->asts_reserved
, 0);
613 res
->migration_pending
= 0;
615 kref_init(&res
->refs
);
617 /* just for consistency */
618 spin_lock(&res
->spinlock
);
619 dlm_set_lockres_owner(dlm
, res
, DLM_LOCK_RES_OWNER_UNKNOWN
);
620 spin_unlock(&res
->spinlock
);
622 res
->state
= DLM_LOCK_RES_IN_PROGRESS
;
626 memset(res
->lvb
, 0, DLM_LVB_LEN
);
629 struct dlm_lock_resource
*dlm_new_lockres(struct dlm_ctxt
*dlm
,
631 unsigned int namelen
)
633 struct dlm_lock_resource
*res
;
635 res
= kmalloc(sizeof(struct dlm_lock_resource
), GFP_KERNEL
);
639 res
->lockname
.name
= kmalloc(namelen
, GFP_KERNEL
);
640 if (!res
->lockname
.name
) {
645 dlm_init_lockres(dlm
, res
, name
, namelen
);
650 * lookup a lock resource by name.
651 * may already exist in the hashtable.
652 * lockid is null terminated
654 * if not, allocate enough for the lockres and for
655 * the temporary structure used in doing the mastering.
657 * also, do a lookup in the dlm->master_list to see
658 * if another node has begun mastering the same lock.
659 * if so, there should be a block entry in there
660 * for this name, and we should *not* attempt to master
661 * the lock here. need to wait around for that node
662 * to assert_master (or die).
665 struct dlm_lock_resource
* dlm_get_lock_resource(struct dlm_ctxt
*dlm
,
669 struct dlm_lock_resource
*tmpres
=NULL
, *res
=NULL
;
670 struct dlm_master_list_entry
*mle
= NULL
;
671 struct dlm_master_list_entry
*alloc_mle
= NULL
;
674 struct dlm_node_iter iter
;
675 unsigned int namelen
, hash
;
677 int bit
, wait_on_recovery
= 0;
681 namelen
= strlen(lockid
);
682 hash
= dlm_lockid_hash(lockid
, namelen
);
684 mlog(0, "get lockres %s (len %d)\n", lockid
, namelen
);
687 spin_lock(&dlm
->spinlock
);
688 tmpres
= __dlm_lookup_lockres(dlm
, lockid
, namelen
, hash
);
690 spin_unlock(&dlm
->spinlock
);
691 mlog(0, "found in hash!\n");
693 dlm_lockres_put(res
);
699 spin_unlock(&dlm
->spinlock
);
700 mlog(0, "allocating a new resource\n");
701 /* nothing found and we need to allocate one. */
702 alloc_mle
= (struct dlm_master_list_entry
*)
703 kmem_cache_alloc(dlm_mle_cache
, GFP_KERNEL
);
706 res
= dlm_new_lockres(dlm
, lockid
, namelen
);
712 mlog(0, "no lockres found, allocated our own: %p\n", res
);
714 if (flags
& LKM_LOCAL
) {
715 /* caller knows it's safe to assume it's not mastered elsewhere
716 * DONE! return right away */
717 spin_lock(&res
->spinlock
);
718 dlm_change_lockres_owner(dlm
, res
, dlm
->node_num
);
719 __dlm_insert_lockres(dlm
, res
);
720 spin_unlock(&res
->spinlock
);
721 spin_unlock(&dlm
->spinlock
);
722 /* lockres still marked IN_PROGRESS */
726 /* check master list to see if another node has started mastering it */
727 spin_lock(&dlm
->master_lock
);
729 /* if we found a block, wait for lock to be mastered by another node */
730 blocked
= dlm_find_mle(dlm
, &mle
, (char *)lockid
, namelen
);
732 if (mle
->type
== DLM_MLE_MASTER
) {
733 mlog(ML_ERROR
, "master entry for nonexistent lock!\n");
735 } else if (mle
->type
== DLM_MLE_MIGRATION
) {
736 /* migration is in progress! */
737 /* the good news is that we now know the
738 * "current" master (mle->master). */
740 spin_unlock(&dlm
->master_lock
);
741 assert_spin_locked(&dlm
->spinlock
);
743 /* set the lockres owner and hash it */
744 spin_lock(&res
->spinlock
);
745 dlm_set_lockres_owner(dlm
, res
, mle
->master
);
746 __dlm_insert_lockres(dlm
, res
);
747 spin_unlock(&res
->spinlock
);
748 spin_unlock(&dlm
->spinlock
);
750 /* master is known, detach */
751 dlm_mle_detach_hb_events(dlm
, mle
);
757 /* go ahead and try to master lock on this node */
759 /* make sure this does not get freed below */
761 dlm_init_mle(mle
, DLM_MLE_MASTER
, dlm
, res
, NULL
, 0);
762 set_bit(dlm
->node_num
, mle
->maybe_map
);
763 list_add(&mle
->list
, &dlm
->master_list
);
765 /* still holding the dlm spinlock, check the recovery map
766 * to see if there are any nodes that still need to be
767 * considered. these will not appear in the mle nodemap
768 * but they might own this lockres. wait on them. */
769 bit
= find_next_bit(dlm
->recovery_map
, O2NM_MAX_NODES
, 0);
770 if (bit
< O2NM_MAX_NODES
) {
771 mlog(ML_NOTICE
, "%s:%.*s: at least one node (%d) to"
772 "recover before lock mastery can begin\n",
773 dlm
->name
, namelen
, (char *)lockid
, bit
);
774 wait_on_recovery
= 1;
778 /* at this point there is either a DLM_MLE_BLOCK or a
779 * DLM_MLE_MASTER on the master list, so it's safe to add the
780 * lockres to the hashtable. anyone who finds the lock will
781 * still have to wait on the IN_PROGRESS. */
783 /* finally add the lockres to its hash bucket */
784 __dlm_insert_lockres(dlm
, res
);
785 /* get an extra ref on the mle in case this is a BLOCK
786 * if so, the creator of the BLOCK may try to put the last
787 * ref at this time in the assert master handler, so we
788 * need an extra one to keep from a bad ptr deref. */
790 spin_unlock(&dlm
->master_lock
);
791 spin_unlock(&dlm
->spinlock
);
793 while (wait_on_recovery
) {
794 /* any cluster changes that occurred after dropping the
795 * dlm spinlock would be detectable be a change on the mle,
796 * so we only need to clear out the recovery map once. */
797 if (dlm_is_recovery_lock(lockid
, namelen
)) {
798 mlog(ML_NOTICE
, "%s: recovery map is not empty, but "
799 "must master $RECOVERY lock now\n", dlm
->name
);
800 if (!dlm_pre_master_reco_lockres(dlm
, res
))
801 wait_on_recovery
= 0;
803 mlog(0, "%s: waiting 500ms for heartbeat state "
804 "change\n", dlm
->name
);
810 dlm_kick_recovery_thread(dlm
);
812 dlm_wait_for_recovery(dlm
);
814 spin_lock(&dlm
->spinlock
);
815 bit
= find_next_bit(dlm
->recovery_map
, O2NM_MAX_NODES
, 0);
816 if (bit
< O2NM_MAX_NODES
) {
817 mlog(ML_NOTICE
, "%s:%.*s: at least one node (%d) to"
818 "recover before lock mastery can begin\n",
819 dlm
->name
, namelen
, (char *)lockid
, bit
);
820 wait_on_recovery
= 1;
822 wait_on_recovery
= 0;
823 spin_unlock(&dlm
->spinlock
);
826 /* must wait for lock to be mastered elsewhere */
832 dlm_node_iter_init(mle
->vote_map
, &iter
);
833 while ((nodenum
= dlm_node_iter_next(&iter
)) >= 0) {
834 ret
= dlm_do_master_request(mle
, nodenum
);
837 if (mle
->master
!= O2NM_MAX_NODES
) {
838 /* found a master ! */
839 if (mle
->master
<= nodenum
)
841 /* if our master request has not reached the master
842 * yet, keep going until it does. this is how the
843 * master will know that asserts are needed back to
844 * the lower nodes. */
845 mlog(0, "%s:%.*s: requests only up to %u but master "
846 "is %u, keep going\n", dlm
->name
, namelen
,
847 lockid
, nodenum
, mle
->master
);
852 /* keep going until the response map includes all nodes */
853 ret
= dlm_wait_for_lock_mastery(dlm
, res
, mle
, &blocked
);
855 mlog(0, "%s:%.*s: node map changed, redo the "
856 "master request now, blocked=%d\n",
857 dlm
->name
, res
->lockname
.len
,
858 res
->lockname
.name
, blocked
);
860 mlog(ML_ERROR
, "%s:%.*s: spinning on "
861 "dlm_wait_for_lock_mastery, blocked=%d\n",
862 dlm
->name
, res
->lockname
.len
,
863 res
->lockname
.name
, blocked
);
864 dlm_print_one_lock_resource(res
);
865 /* dlm_print_one_mle(mle); */
871 mlog(0, "lockres mastered by %u\n", res
->owner
);
872 /* make sure we never continue without this */
873 BUG_ON(res
->owner
== O2NM_MAX_NODES
);
875 /* master is known, detach if not already detached */
876 dlm_mle_detach_hb_events(dlm
, mle
);
878 /* put the extra ref */
882 spin_lock(&res
->spinlock
);
883 res
->state
&= ~DLM_LOCK_RES_IN_PROGRESS
;
884 spin_unlock(&res
->spinlock
);
888 /* need to free the unused mle */
890 kmem_cache_free(dlm_mle_cache
, alloc_mle
);
896 #define DLM_MASTERY_TIMEOUT_MS 5000
898 static int dlm_wait_for_lock_mastery(struct dlm_ctxt
*dlm
,
899 struct dlm_lock_resource
*res
,
900 struct dlm_master_list_entry
*mle
,
905 int map_changed
, voting_done
;
912 /* check if another node has already become the owner */
913 spin_lock(&res
->spinlock
);
914 if (res
->owner
!= DLM_LOCK_RES_OWNER_UNKNOWN
) {
915 mlog(0, "%s:%.*s: owner is suddenly %u\n", dlm
->name
,
916 res
->lockname
.len
, res
->lockname
.name
, res
->owner
);
917 spin_unlock(&res
->spinlock
);
918 /* this will cause the master to re-assert across
919 * the whole cluster, freeing up mles */
920 ret
= dlm_do_master_request(mle
, res
->owner
);
922 /* give recovery a chance to run */
923 mlog(ML_ERROR
, "link to %u went down?: %d\n", res
->owner
, ret
);
930 spin_unlock(&res
->spinlock
);
932 spin_lock(&mle
->spinlock
);
934 map_changed
= (memcmp(mle
->vote_map
, mle
->node_map
,
935 sizeof(mle
->vote_map
)) != 0);
936 voting_done
= (memcmp(mle
->vote_map
, mle
->response_map
,
937 sizeof(mle
->vote_map
)) == 0);
939 /* restart if we hit any errors */
942 mlog(0, "%s: %.*s: node map changed, restarting\n",
943 dlm
->name
, res
->lockname
.len
, res
->lockname
.name
);
944 ret
= dlm_restart_lock_mastery(dlm
, res
, mle
, *blocked
);
945 b
= (mle
->type
== DLM_MLE_BLOCK
);
946 if ((*blocked
&& !b
) || (!*blocked
&& b
)) {
947 mlog(0, "%s:%.*s: status change: old=%d new=%d\n",
948 dlm
->name
, res
->lockname
.len
, res
->lockname
.name
,
952 spin_unlock(&mle
->spinlock
);
957 mlog(0, "%s:%.*s: restart lock mastery succeeded, "
958 "rechecking now\n", dlm
->name
, res
->lockname
.len
,
963 if (m
!= O2NM_MAX_NODES
) {
964 /* another node has done an assert!
969 /* have all nodes responded? */
970 if (voting_done
&& !*blocked
) {
971 bit
= find_next_bit(mle
->maybe_map
, O2NM_MAX_NODES
, 0);
972 if (dlm
->node_num
<= bit
) {
973 /* my node number is lowest.
974 * now tell other nodes that I am
976 mle
->master
= dlm
->node_num
;
980 /* if voting is done, but we have not received
981 * an assert master yet, we must sleep */
985 spin_unlock(&mle
->spinlock
);
987 /* sleep if we haven't finished voting yet */
989 unsigned long timeo
= msecs_to_jiffies(DLM_MASTERY_TIMEOUT_MS
);
992 if (atomic_read(&mle->mle_refs.refcount) < 2)
993 mlog(ML_ERROR, "mle (%p) refs=%d, name=%.*s\n", mle,
994 atomic_read(&mle->mle_refs.refcount),
995 res->lockname.len, res->lockname.name);
997 atomic_set(&mle
->woken
, 0);
998 (void)wait_event_timeout(mle
->wq
,
999 (atomic_read(&mle
->woken
) == 1),
1001 if (res
->owner
== O2NM_MAX_NODES
) {
1002 mlog(0, "waiting again\n");
1005 mlog(0, "done waiting, master is %u\n", res
->owner
);
1013 mlog(0, "about to master %.*s here, this=%u\n",
1014 res
->lockname
.len
, res
->lockname
.name
, m
);
1015 ret
= dlm_do_assert_master(dlm
, res
->lockname
.name
,
1016 res
->lockname
.len
, mle
->vote_map
, 0);
1018 /* This is a failure in the network path,
1019 * not in the response to the assert_master
1020 * (any nonzero response is a BUG on this node).
1021 * Most likely a socket just got disconnected
1022 * due to node death. */
1025 /* no longer need to restart lock mastery.
1026 * all living nodes have been contacted. */
1030 /* set the lockres owner */
1031 spin_lock(&res
->spinlock
);
1032 dlm_change_lockres_owner(dlm
, res
, m
);
1033 spin_unlock(&res
->spinlock
);
1039 struct dlm_bitmap_diff_iter
1042 unsigned long *orig_bm
;
1043 unsigned long *cur_bm
;
1044 unsigned long diff_bm
[BITS_TO_LONGS(O2NM_MAX_NODES
)];
1047 enum dlm_node_state_change
1054 static void dlm_bitmap_diff_iter_init(struct dlm_bitmap_diff_iter
*iter
,
1055 unsigned long *orig_bm
,
1056 unsigned long *cur_bm
)
1058 unsigned long p1
, p2
;
1062 iter
->orig_bm
= orig_bm
;
1063 iter
->cur_bm
= cur_bm
;
1065 for (i
= 0; i
< BITS_TO_LONGS(O2NM_MAX_NODES
); i
++) {
1066 p1
= *(iter
->orig_bm
+ i
);
1067 p2
= *(iter
->cur_bm
+ i
);
1068 iter
->diff_bm
[i
] = (p1
& ~p2
) | (p2
& ~p1
);
1072 static int dlm_bitmap_diff_iter_next(struct dlm_bitmap_diff_iter
*iter
,
1073 enum dlm_node_state_change
*state
)
1077 if (iter
->curnode
>= O2NM_MAX_NODES
)
1080 bit
= find_next_bit(iter
->diff_bm
, O2NM_MAX_NODES
,
1082 if (bit
>= O2NM_MAX_NODES
) {
1083 iter
->curnode
= O2NM_MAX_NODES
;
1087 /* if it was there in the original then this node died */
1088 if (test_bit(bit
, iter
->orig_bm
))
1093 iter
->curnode
= bit
;
1098 static int dlm_restart_lock_mastery(struct dlm_ctxt
*dlm
,
1099 struct dlm_lock_resource
*res
,
1100 struct dlm_master_list_entry
*mle
,
1103 struct dlm_bitmap_diff_iter bdi
;
1104 enum dlm_node_state_change sc
;
1108 mlog(0, "something happened such that the "
1109 "master process may need to be restarted!\n");
1111 assert_spin_locked(&mle
->spinlock
);
1113 dlm_bitmap_diff_iter_init(&bdi
, mle
->vote_map
, mle
->node_map
);
1114 node
= dlm_bitmap_diff_iter_next(&bdi
, &sc
);
1116 if (sc
== NODE_UP
) {
1117 /* a node came up. clear any old vote from
1118 * the response map and set it in the vote map
1119 * then restart the mastery. */
1120 mlog(ML_NOTICE
, "node %d up while restarting\n", node
);
1122 /* redo the master request, but only for the new node */
1123 mlog(0, "sending request to new node\n");
1124 clear_bit(node
, mle
->response_map
);
1125 set_bit(node
, mle
->vote_map
);
1127 mlog(ML_ERROR
, "node down! %d\n", node
);
1129 /* if the node wasn't involved in mastery skip it,
1130 * but clear it out from the maps so that it will
1131 * not affect mastery of this lockres */
1132 clear_bit(node
, mle
->response_map
);
1133 clear_bit(node
, mle
->vote_map
);
1134 if (!test_bit(node
, mle
->maybe_map
))
1137 /* if we're already blocked on lock mastery, and the
1138 * dead node wasn't the expected master, or there is
1139 * another node in the maybe_map, keep waiting */
1141 int lowest
= find_next_bit(mle
->maybe_map
,
1144 /* act like it was never there */
1145 clear_bit(node
, mle
->maybe_map
);
1150 mlog(ML_ERROR
, "expected master %u died while "
1151 "this node was blocked waiting on it!\n",
1153 lowest
= find_next_bit(mle
->maybe_map
,
1156 if (lowest
< O2NM_MAX_NODES
) {
1157 mlog(0, "still blocked. waiting "
1158 "on %u now\n", lowest
);
1162 /* mle is an MLE_BLOCK, but there is now
1163 * nothing left to block on. we need to return
1164 * all the way back out and try again with
1165 * an MLE_MASTER. dlm_do_local_recovery_cleanup
1166 * has already run, so the mle refcount is ok */
1167 mlog(0, "no longer blocking. we can "
1168 "try to master this here\n");
1169 mle
->type
= DLM_MLE_MASTER
;
1170 memset(mle
->maybe_map
, 0,
1171 sizeof(mle
->maybe_map
));
1172 memset(mle
->response_map
, 0,
1173 sizeof(mle
->maybe_map
));
1174 memcpy(mle
->vote_map
, mle
->node_map
,
1175 sizeof(mle
->node_map
));
1177 set_bit(dlm
->node_num
, mle
->maybe_map
);
1183 clear_bit(node
, mle
->maybe_map
);
1184 if (node
> dlm
->node_num
)
1187 mlog(0, "dead node in map!\n");
1188 /* yuck. go back and re-contact all nodes
1189 * in the vote_map, removing this node. */
1190 memset(mle
->response_map
, 0,
1191 sizeof(mle
->response_map
));
1195 node
= dlm_bitmap_diff_iter_next(&bdi
, &sc
);
1202 * DLM_MASTER_REQUEST_MSG
1204 * returns: 0 on success,
1205 * -errno on a network error
1207 * on error, the caller should assume the target node is "dead"
1211 static int dlm_do_master_request(struct dlm_master_list_entry
*mle
, int to
)
1213 struct dlm_ctxt
*dlm
= mle
->dlm
;
1214 struct dlm_master_request request
;
1215 int ret
, response
=0, resend
;
1217 memset(&request
, 0, sizeof(request
));
1218 request
.node_idx
= dlm
->node_num
;
1220 BUG_ON(mle
->type
== DLM_MLE_MIGRATION
);
1222 if (mle
->type
!= DLM_MLE_MASTER
) {
1223 request
.namelen
= mle
->u
.name
.len
;
1224 memcpy(request
.name
, mle
->u
.name
.name
, request
.namelen
);
1226 request
.namelen
= mle
->u
.res
->lockname
.len
;
1227 memcpy(request
.name
, mle
->u
.res
->lockname
.name
,
1232 ret
= o2net_send_message(DLM_MASTER_REQUEST_MSG
, dlm
->key
, &request
,
1233 sizeof(request
), to
, &response
);
1235 if (ret
== -ESRCH
) {
1236 /* should never happen */
1237 mlog(ML_ERROR
, "TCP stack not ready!\n");
1239 } else if (ret
== -EINVAL
) {
1240 mlog(ML_ERROR
, "bad args passed to o2net!\n");
1242 } else if (ret
== -ENOMEM
) {
1243 mlog(ML_ERROR
, "out of memory while trying to send "
1244 "network message! retrying\n");
1245 /* this is totally crude */
1248 } else if (!dlm_is_host_down(ret
)) {
1249 /* not a network error. bad. */
1251 mlog(ML_ERROR
, "unhandled error!");
1254 /* all other errors should be network errors,
1255 * and likely indicate node death */
1256 mlog(ML_ERROR
, "link to %d went down!\n", to
);
1262 spin_lock(&mle
->spinlock
);
1264 case DLM_MASTER_RESP_YES
:
1265 set_bit(to
, mle
->response_map
);
1266 mlog(0, "node %u is the master, response=YES\n", to
);
1269 case DLM_MASTER_RESP_NO
:
1270 mlog(0, "node %u not master, response=NO\n", to
);
1271 set_bit(to
, mle
->response_map
);
1273 case DLM_MASTER_RESP_MAYBE
:
1274 mlog(0, "node %u not master, response=MAYBE\n", to
);
1275 set_bit(to
, mle
->response_map
);
1276 set_bit(to
, mle
->maybe_map
);
1278 case DLM_MASTER_RESP_ERROR
:
1279 mlog(0, "node %u hit an error, resending\n", to
);
1284 mlog(ML_ERROR
, "bad response! %u\n", response
);
1287 spin_unlock(&mle
->spinlock
);
1289 /* this is also totally crude */
1299 * locks that can be taken here:
1305 * if possible, TRIM THIS DOWN!!!
1307 int dlm_master_request_handler(struct o2net_msg
*msg
, u32 len
, void *data
)
1309 u8 response
= DLM_MASTER_RESP_MAYBE
;
1310 struct dlm_ctxt
*dlm
= data
;
1311 struct dlm_lock_resource
*res
= NULL
;
1312 struct dlm_master_request
*request
= (struct dlm_master_request
*) msg
->buf
;
1313 struct dlm_master_list_entry
*mle
= NULL
, *tmpmle
= NULL
;
1315 unsigned int namelen
, hash
;
1318 int dispatch_assert
= 0;
1321 return DLM_MASTER_RESP_NO
;
1323 if (!dlm_domain_fully_joined(dlm
)) {
1324 response
= DLM_MASTER_RESP_NO
;
1328 name
= request
->name
;
1329 namelen
= request
->namelen
;
1330 hash
= dlm_lockid_hash(name
, namelen
);
1332 if (namelen
> DLM_LOCKID_NAME_MAX
) {
1333 response
= DLM_IVBUFLEN
;
1338 spin_lock(&dlm
->spinlock
);
1339 res
= __dlm_lookup_lockres(dlm
, name
, namelen
, hash
);
1341 spin_unlock(&dlm
->spinlock
);
1343 /* take care of the easy cases up front */
1344 spin_lock(&res
->spinlock
);
1345 if (res
->state
& DLM_LOCK_RES_RECOVERING
) {
1346 spin_unlock(&res
->spinlock
);
1347 mlog(0, "returning DLM_MASTER_RESP_ERROR since res is "
1348 "being recovered\n");
1349 response
= DLM_MASTER_RESP_ERROR
;
1351 kmem_cache_free(dlm_mle_cache
, mle
);
1355 if (res
->owner
== dlm
->node_num
) {
1356 spin_unlock(&res
->spinlock
);
1357 // mlog(0, "this node is the master\n");
1358 response
= DLM_MASTER_RESP_YES
;
1360 kmem_cache_free(dlm_mle_cache
, mle
);
1362 /* this node is the owner.
1363 * there is some extra work that needs to
1364 * happen now. the requesting node has
1365 * caused all nodes up to this one to
1366 * create mles. this node now needs to
1367 * go back and clean those up. */
1368 dispatch_assert
= 1;
1370 } else if (res
->owner
!= DLM_LOCK_RES_OWNER_UNKNOWN
) {
1371 spin_unlock(&res
->spinlock
);
1372 // mlog(0, "node %u is the master\n", res->owner);
1373 response
= DLM_MASTER_RESP_NO
;
1375 kmem_cache_free(dlm_mle_cache
, mle
);
1379 /* ok, there is no owner. either this node is
1380 * being blocked, or it is actively trying to
1381 * master this lock. */
1382 if (!(res
->state
& DLM_LOCK_RES_IN_PROGRESS
)) {
1383 mlog(ML_ERROR
, "lock with no owner should be "
1388 // mlog(0, "lockres is in progress...\n");
1389 spin_lock(&dlm
->master_lock
);
1390 found
= dlm_find_mle(dlm
, &tmpmle
, name
, namelen
);
1392 mlog(ML_ERROR
, "no mle found for this lock!\n");
1396 spin_lock(&tmpmle
->spinlock
);
1397 if (tmpmle
->type
== DLM_MLE_BLOCK
) {
1398 // mlog(0, "this node is waiting for "
1399 // "lockres to be mastered\n");
1400 response
= DLM_MASTER_RESP_NO
;
1401 } else if (tmpmle
->type
== DLM_MLE_MIGRATION
) {
1402 mlog(0, "node %u is master, but trying to migrate to "
1403 "node %u.\n", tmpmle
->master
, tmpmle
->new_master
);
1404 if (tmpmle
->master
== dlm
->node_num
) {
1405 response
= DLM_MASTER_RESP_YES
;
1406 mlog(ML_ERROR
, "no owner on lockres, but this "
1407 "node is trying to migrate it to %u?!\n",
1408 tmpmle
->new_master
);
1411 /* the real master can respond on its own */
1412 response
= DLM_MASTER_RESP_NO
;
1414 } else if (tmpmle
->master
!= DLM_LOCK_RES_OWNER_UNKNOWN
) {
1416 if (tmpmle
->master
== dlm
->node_num
) {
1417 response
= DLM_MASTER_RESP_YES
;
1418 /* this node will be the owner.
1419 * go back and clean the mles on any
1421 dispatch_assert
= 1;
1423 response
= DLM_MASTER_RESP_NO
;
1425 // mlog(0, "this node is attempting to "
1426 // "master lockres\n");
1427 response
= DLM_MASTER_RESP_MAYBE
;
1430 set_bit(request
->node_idx
, tmpmle
->maybe_map
);
1431 spin_unlock(&tmpmle
->spinlock
);
1433 spin_unlock(&dlm
->master_lock
);
1434 spin_unlock(&res
->spinlock
);
1436 /* keep the mle attached to heartbeat events */
1437 dlm_put_mle(tmpmle
);
1439 kmem_cache_free(dlm_mle_cache
, mle
);
1444 * lockres doesn't exist on this node
1445 * if there is an MLE_BLOCK, return NO
1446 * if there is an MLE_MASTER, return MAYBE
1447 * otherwise, add an MLE_BLOCK, return NO
1449 spin_lock(&dlm
->master_lock
);
1450 found
= dlm_find_mle(dlm
, &tmpmle
, name
, namelen
);
1452 /* this lockid has never been seen on this node yet */
1453 // mlog(0, "no mle found\n");
1455 spin_unlock(&dlm
->master_lock
);
1456 spin_unlock(&dlm
->spinlock
);
1458 mle
= (struct dlm_master_list_entry
*)
1459 kmem_cache_alloc(dlm_mle_cache
, GFP_KERNEL
);
1461 response
= DLM_MASTER_RESP_ERROR
;
1462 mlog_errno(-ENOMEM
);
1465 spin_lock(&dlm
->spinlock
);
1466 dlm_init_mle(mle
, DLM_MLE_BLOCK
, dlm
, NULL
,
1468 spin_unlock(&dlm
->spinlock
);
1472 // mlog(0, "this is second time thru, already allocated, "
1473 // "add the block.\n");
1474 set_bit(request
->node_idx
, mle
->maybe_map
);
1475 list_add(&mle
->list
, &dlm
->master_list
);
1476 response
= DLM_MASTER_RESP_NO
;
1478 // mlog(0, "mle was found\n");
1480 spin_lock(&tmpmle
->spinlock
);
1481 if (tmpmle
->master
== dlm
->node_num
) {
1482 mlog(ML_ERROR
, "no lockres, but an mle with this node as master!\n");
1485 if (tmpmle
->type
== DLM_MLE_BLOCK
)
1486 response
= DLM_MASTER_RESP_NO
;
1487 else if (tmpmle
->type
== DLM_MLE_MIGRATION
) {
1488 mlog(0, "migration mle was found (%u->%u)\n",
1489 tmpmle
->master
, tmpmle
->new_master
);
1490 /* real master can respond on its own */
1491 response
= DLM_MASTER_RESP_NO
;
1493 response
= DLM_MASTER_RESP_MAYBE
;
1495 set_bit(request
->node_idx
, tmpmle
->maybe_map
);
1496 spin_unlock(&tmpmle
->spinlock
);
1498 spin_unlock(&dlm
->master_lock
);
1499 spin_unlock(&dlm
->spinlock
);
1502 /* keep the mle attached to heartbeat events */
1503 dlm_put_mle(tmpmle
);
1507 if (dispatch_assert
) {
1508 if (response
!= DLM_MASTER_RESP_YES
)
1509 mlog(ML_ERROR
, "invalid response %d\n", response
);
1511 mlog(ML_ERROR
, "bad lockres while trying to assert!\n");
1514 mlog(0, "%u is the owner of %.*s, cleaning everyone else\n",
1515 dlm
->node_num
, res
->lockname
.len
, res
->lockname
.name
);
1516 ret
= dlm_dispatch_assert_master(dlm
, res
, 0, request
->node_idx
,
1517 DLM_ASSERT_MASTER_MLE_CLEANUP
);
1519 mlog(ML_ERROR
, "failed to dispatch assert master work\n");
1520 response
= DLM_MASTER_RESP_ERROR
;
1529 * DLM_ASSERT_MASTER_MSG
1534 * NOTE: this can be used for debugging
1535 * can periodically run all locks owned by this node
1536 * and re-assert across the cluster...
1538 static int dlm_do_assert_master(struct dlm_ctxt
*dlm
, const char *lockname
,
1539 unsigned int namelen
, void *nodemap
,
1542 struct dlm_assert_master
assert;
1544 struct dlm_node_iter iter
;
1548 BUG_ON(namelen
> O2NM_MAX_NAME_LEN
);
1552 /* note that if this nodemap is empty, it returns 0 */
1553 dlm_node_iter_init(nodemap
, &iter
);
1554 while ((to
= dlm_node_iter_next(&iter
)) >= 0) {
1556 mlog(0, "sending assert master to %d (%.*s)\n", to
,
1558 memset(&assert, 0, sizeof(assert));
1559 assert.node_idx
= dlm
->node_num
;
1560 assert.namelen
= namelen
;
1561 memcpy(assert.name
, lockname
, namelen
);
1562 assert.flags
= cpu_to_be32(flags
);
1564 tmpret
= o2net_send_message(DLM_ASSERT_MASTER_MSG
, dlm
->key
,
1565 &assert, sizeof(assert), to
, &r
);
1567 mlog(ML_ERROR
, "assert_master returned %d!\n", tmpret
);
1568 if (!dlm_is_host_down(tmpret
)) {
1569 mlog(ML_ERROR
, "unhandled error!\n");
1572 /* a node died. finish out the rest of the nodes. */
1573 mlog(ML_ERROR
, "link to %d went down!\n", to
);
1574 /* any nonzero status return will do */
1577 /* ok, something horribly messed. kill thyself. */
1578 mlog(ML_ERROR
,"during assert master of %.*s to %u, "
1579 "got %d.\n", namelen
, lockname
, to
, r
);
1580 dlm_dump_lock_resources(dlm
);
1582 } else if (r
== EAGAIN
) {
1583 mlog(0, "%.*s: node %u create mles on other "
1584 "nodes and requests a re-assert\n",
1585 namelen
, lockname
, to
);
1597 * locks that can be taken here:
1603 * if possible, TRIM THIS DOWN!!!
1605 int dlm_assert_master_handler(struct o2net_msg
*msg
, u32 len
, void *data
)
1607 struct dlm_ctxt
*dlm
= data
;
1608 struct dlm_master_list_entry
*mle
= NULL
;
1609 struct dlm_assert_master
*assert = (struct dlm_assert_master
*)msg
->buf
;
1610 struct dlm_lock_resource
*res
= NULL
;
1612 unsigned int namelen
, hash
;
1614 int master_request
= 0;
1620 name
= assert->name
;
1621 namelen
= assert->namelen
;
1622 hash
= dlm_lockid_hash(name
, namelen
);
1623 flags
= be32_to_cpu(assert->flags
);
1625 if (namelen
> DLM_LOCKID_NAME_MAX
) {
1626 mlog(ML_ERROR
, "Invalid name length!");
1630 spin_lock(&dlm
->spinlock
);
1633 mlog(0, "assert_master with flags: %u\n", flags
);
1636 spin_lock(&dlm
->master_lock
);
1637 if (!dlm_find_mle(dlm
, &mle
, name
, namelen
)) {
1638 /* not an error, could be master just re-asserting */
1639 mlog(0, "just got an assert_master from %u, but no "
1640 "MLE for it! (%.*s)\n", assert->node_idx
,
1643 int bit
= find_next_bit (mle
->maybe_map
, O2NM_MAX_NODES
, 0);
1644 if (bit
>= O2NM_MAX_NODES
) {
1645 /* not necessarily an error, though less likely.
1646 * could be master just re-asserting. */
1647 mlog(ML_ERROR
, "no bits set in the maybe_map, but %u "
1648 "is asserting! (%.*s)\n", assert->node_idx
,
1650 } else if (bit
!= assert->node_idx
) {
1651 if (flags
& DLM_ASSERT_MASTER_MLE_CLEANUP
) {
1652 mlog(0, "master %u was found, %u should "
1653 "back off\n", assert->node_idx
, bit
);
1655 /* with the fix for bug 569, a higher node
1656 * number winning the mastery will respond
1657 * YES to mastery requests, but this node
1658 * had no way of knowing. let it pass. */
1659 mlog(ML_ERROR
, "%u is the lowest node, "
1660 "%u is asserting. (%.*s) %u must "
1661 "have begun after %u won.\n", bit
,
1662 assert->node_idx
, namelen
, name
, bit
,
1667 spin_unlock(&dlm
->master_lock
);
1669 /* ok everything checks out with the MLE
1670 * now check to see if there is a lockres */
1671 res
= __dlm_lookup_lockres(dlm
, name
, namelen
, hash
);
1673 spin_lock(&res
->spinlock
);
1674 if (res
->state
& DLM_LOCK_RES_RECOVERING
) {
1675 mlog(ML_ERROR
, "%u asserting but %.*s is "
1676 "RECOVERING!\n", assert->node_idx
, namelen
, name
);
1680 if (res
->owner
!= assert->node_idx
) {
1681 mlog(ML_ERROR
, "assert_master from "
1682 "%u, but current owner is "
1684 assert->node_idx
, res
->owner
,
1688 } else if (mle
->type
!= DLM_MLE_MIGRATION
) {
1689 if (res
->owner
!= DLM_LOCK_RES_OWNER_UNKNOWN
) {
1690 /* owner is just re-asserting */
1691 if (res
->owner
== assert->node_idx
) {
1692 mlog(0, "owner %u re-asserting on "
1693 "lock %.*s\n", assert->node_idx
,
1697 mlog(ML_ERROR
, "got assert_master from "
1698 "node %u, but %u is the owner! "
1699 "(%.*s)\n", assert->node_idx
,
1700 res
->owner
, namelen
, name
);
1703 if (!(res
->state
& DLM_LOCK_RES_IN_PROGRESS
)) {
1704 mlog(ML_ERROR
, "got assert from %u, but lock "
1705 "with no owner should be "
1706 "in-progress! (%.*s)\n",
1711 } else /* mle->type == DLM_MLE_MIGRATION */ {
1712 /* should only be getting an assert from new master */
1713 if (assert->node_idx
!= mle
->new_master
) {
1714 mlog(ML_ERROR
, "got assert from %u, but "
1715 "new master is %u, and old master "
1717 assert->node_idx
, mle
->new_master
,
1718 mle
->master
, namelen
, name
);
1724 spin_unlock(&res
->spinlock
);
1726 spin_unlock(&dlm
->spinlock
);
1728 // mlog(0, "woo! got an assert_master from node %u!\n",
1729 // assert->node_idx);
1734 spin_lock(&mle
->spinlock
);
1735 if (mle
->type
== DLM_MLE_BLOCK
|| mle
->type
== DLM_MLE_MIGRATION
)
1738 /* MASTER mle: if any bits set in the response map
1739 * then the calling node needs to re-assert to clear
1740 * up nodes that this node contacted */
1741 while ((nn
= find_next_bit (mle
->response_map
, O2NM_MAX_NODES
,
1742 nn
+1)) < O2NM_MAX_NODES
) {
1743 if (nn
!= dlm
->node_num
&& nn
!= assert->node_idx
)
1747 mle
->master
= assert->node_idx
;
1748 atomic_set(&mle
->woken
, 1);
1750 spin_unlock(&mle
->spinlock
);
1752 if (mle
->type
== DLM_MLE_MIGRATION
&& res
) {
1753 mlog(0, "finishing off migration of lockres %.*s, "
1755 res
->lockname
.len
, res
->lockname
.name
,
1756 dlm
->node_num
, mle
->new_master
);
1757 spin_lock(&res
->spinlock
);
1758 res
->state
&= ~DLM_LOCK_RES_MIGRATING
;
1759 dlm_change_lockres_owner(dlm
, res
, mle
->new_master
);
1760 BUG_ON(res
->state
& DLM_LOCK_RES_DIRTY
);
1761 spin_unlock(&res
->spinlock
);
1763 /* master is known, detach if not already detached */
1764 dlm_mle_detach_hb_events(dlm
, mle
);
1768 /* the assert master message now balances the extra
1769 * ref given by the master / migration request message.
1770 * if this is the last put, it will be removed
1779 dlm_lockres_put(res
);
1781 if (master_request
) {
1782 mlog(0, "need to tell master to reassert\n");
1783 ret
= EAGAIN
; // positive. negative would shoot down the node.
1788 /* kill the caller! */
1789 spin_unlock(&res
->spinlock
);
1790 spin_unlock(&dlm
->spinlock
);
1791 dlm_lockres_put(res
);
1792 mlog(ML_ERROR
, "Bad message received from another node. Dumping state "
1793 "and killing the other node now! This node is OK and can continue.\n");
1794 dlm_dump_lock_resources(dlm
);
1799 int dlm_dispatch_assert_master(struct dlm_ctxt
*dlm
,
1800 struct dlm_lock_resource
*res
,
1801 int ignore_higher
, u8 request_from
, u32 flags
)
1803 struct dlm_work_item
*item
;
1804 item
= kcalloc(1, sizeof(*item
), GFP_KERNEL
);
1809 /* queue up work for dlm_assert_master_worker */
1810 dlm_grab(dlm
); /* get an extra ref for the work item */
1811 dlm_init_work_item(dlm
, item
, dlm_assert_master_worker
, NULL
);
1812 item
->u
.am
.lockres
= res
; /* already have a ref */
1813 /* can optionally ignore node numbers higher than this node */
1814 item
->u
.am
.ignore_higher
= ignore_higher
;
1815 item
->u
.am
.request_from
= request_from
;
1816 item
->u
.am
.flags
= flags
;
1819 mlog(0, "IGNORE HIGHER: %.*s\n", res
->lockname
.len
,
1820 res
->lockname
.name
);
1822 spin_lock(&dlm
->work_lock
);
1823 list_add_tail(&item
->list
, &dlm
->work_list
);
1824 spin_unlock(&dlm
->work_lock
);
1826 schedule_work(&dlm
->dispatched_work
);
1830 static void dlm_assert_master_worker(struct dlm_work_item
*item
, void *data
)
1832 struct dlm_ctxt
*dlm
= data
;
1834 struct dlm_lock_resource
*res
;
1835 unsigned long nodemap
[BITS_TO_LONGS(O2NM_MAX_NODES
)];
1842 res
= item
->u
.am
.lockres
;
1843 ignore_higher
= item
->u
.am
.ignore_higher
;
1844 request_from
= item
->u
.am
.request_from
;
1845 flags
= item
->u
.am
.flags
;
1847 spin_lock(&dlm
->spinlock
);
1848 memcpy(nodemap
, dlm
->domain_map
, sizeof(nodemap
));
1849 spin_unlock(&dlm
->spinlock
);
1851 clear_bit(dlm
->node_num
, nodemap
);
1852 if (ignore_higher
) {
1853 /* if is this just to clear up mles for nodes below
1854 * this node, do not send the message to the original
1855 * caller or any node number higher than this */
1856 clear_bit(request_from
, nodemap
);
1857 bit
= dlm
->node_num
;
1859 bit
= find_next_bit(nodemap
, O2NM_MAX_NODES
,
1861 if (bit
>= O2NM_MAX_NODES
)
1863 clear_bit(bit
, nodemap
);
1867 /* this call now finishes out the nodemap
1868 * even if one or more nodes die */
1869 mlog(0, "worker about to master %.*s here, this=%u\n",
1870 res
->lockname
.len
, res
->lockname
.name
, dlm
->node_num
);
1871 ret
= dlm_do_assert_master(dlm
, res
->lockname
.name
,
1875 /* no need to restart, we are done */
1879 dlm_lockres_put(res
);
1881 mlog(0, "finished with dlm_assert_master_worker\n");
1884 /* SPECIAL CASE for the $RECOVERY lock used by the recovery thread.
1885 * We cannot wait for node recovery to complete to begin mastering this
1886 * lockres because this lockres is used to kick off recovery! ;-)
1887 * So, do a pre-check on all living nodes to see if any of those nodes
1888 * think that $RECOVERY is currently mastered by a dead node. If so,
1889 * we wait a short time to allow that node to get notified by its own
1890 * heartbeat stack, then check again. All $RECOVERY lock resources
1891 * mastered by dead nodes are purged when the hearbeat callback is
1892 * fired, so we can know for sure that it is safe to continue once
1893 * the node returns a live node or no node. */
1894 static int dlm_pre_master_reco_lockres(struct dlm_ctxt
*dlm
,
1895 struct dlm_lock_resource
*res
)
1897 struct dlm_node_iter iter
;
1900 u8 master
= DLM_LOCK_RES_OWNER_UNKNOWN
;
1902 spin_lock(&dlm
->spinlock
);
1903 dlm_node_iter_init(dlm
->domain_map
, &iter
);
1904 spin_unlock(&dlm
->spinlock
);
1906 while ((nodenum
= dlm_node_iter_next(&iter
)) >= 0) {
1907 /* do not send to self */
1908 if (nodenum
== dlm
->node_num
)
1910 ret
= dlm_do_master_requery(dlm
, res
, nodenum
, &master
);
1913 if (!dlm_is_host_down(ret
))
1915 /* host is down, so answer for that node would be
1916 * DLM_LOCK_RES_OWNER_UNKNOWN. continue. */
1919 if (master
!= DLM_LOCK_RES_OWNER_UNKNOWN
) {
1920 /* check to see if this master is in the recovery map */
1921 spin_lock(&dlm
->spinlock
);
1922 if (test_bit(master
, dlm
->recovery_map
)) {
1923 mlog(ML_NOTICE
, "%s: node %u has not seen "
1924 "node %u go down yet, and thinks the "
1925 "dead node is mastering the recovery "
1926 "lock. must wait.\n", dlm
->name
,
1930 spin_unlock(&dlm
->spinlock
);
1931 mlog(0, "%s: reco lock master is %u\n", dlm
->name
,
1941 * DLM_MIGRATE_LOCKRES
1945 int dlm_migrate_lockres(struct dlm_ctxt
*dlm
, struct dlm_lock_resource
*res
,
1948 struct dlm_master_list_entry
*mle
= NULL
;
1949 struct dlm_master_list_entry
*oldmle
= NULL
;
1950 struct dlm_migratable_lockres
*mres
= NULL
;
1953 unsigned int namelen
;
1955 struct list_head
*queue
, *iter
;
1957 struct dlm_lock
*lock
;
1963 name
= res
->lockname
.name
;
1964 namelen
= res
->lockname
.len
;
1966 mlog(0, "migrating %.*s to %u\n", namelen
, name
, target
);
1969 * ensure this lockres is a proper candidate for migration
1971 spin_lock(&res
->spinlock
);
1972 if (res
->owner
== DLM_LOCK_RES_OWNER_UNKNOWN
) {
1973 mlog(0, "cannot migrate lockres with unknown owner!\n");
1974 spin_unlock(&res
->spinlock
);
1977 if (res
->owner
!= dlm
->node_num
) {
1978 mlog(0, "cannot migrate lockres this node doesn't own!\n");
1979 spin_unlock(&res
->spinlock
);
1982 mlog(0, "checking queues...\n");
1983 queue
= &res
->granted
;
1984 for (i
=0; i
<3; i
++) {
1985 list_for_each(iter
, queue
) {
1986 lock
= list_entry (iter
, struct dlm_lock
, list
);
1988 if (lock
->ml
.node
== dlm
->node_num
) {
1989 mlog(0, "found a lock owned by this node "
1990 "still on the %s queue! will not "
1991 "migrate this lockres\n",
1993 (i
==1 ? "converting" : "blocked"));
1994 spin_unlock(&res
->spinlock
);
2001 mlog(0, "all locks on this lockres are nonlocal. continuing\n");
2002 spin_unlock(&res
->spinlock
);
2006 mlog(0, "no locks were found on this lockres! done!\n");
2012 * preallocate up front
2013 * if this fails, abort
2017 mres
= (struct dlm_migratable_lockres
*) __get_free_page(GFP_KERNEL
);
2023 mle
= (struct dlm_master_list_entry
*) kmem_cache_alloc(dlm_mle_cache
,
2032 * find a node to migrate the lockres to
2035 mlog(0, "picking a migration node\n");
2036 spin_lock(&dlm
->spinlock
);
2037 /* pick a new node */
2038 if (!test_bit(target
, dlm
->domain_map
) ||
2039 target
>= O2NM_MAX_NODES
) {
2040 target
= dlm_pick_migration_target(dlm
, res
);
2042 mlog(0, "node %u chosen for migration\n", target
);
2044 if (target
>= O2NM_MAX_NODES
||
2045 !test_bit(target
, dlm
->domain_map
)) {
2046 /* target chosen is not alive */
2051 spin_unlock(&dlm
->spinlock
);
2055 mlog(0, "continuing with target = %u\n", target
);
2058 * clear any existing master requests and
2059 * add the migration mle to the list
2061 spin_lock(&dlm
->master_lock
);
2062 ret
= dlm_add_migration_mle(dlm
, res
, mle
, &oldmle
, name
,
2063 namelen
, target
, dlm
->node_num
);
2064 spin_unlock(&dlm
->master_lock
);
2065 spin_unlock(&dlm
->spinlock
);
2067 if (ret
== -EEXIST
) {
2068 mlog(0, "another process is already migrating it\n");
2074 * set the MIGRATING flag and flush asts
2075 * if we fail after this we need to re-dirty the lockres
2077 if (dlm_mark_lockres_migrating(dlm
, res
, target
) < 0) {
2078 mlog(ML_ERROR
, "tried to migrate %.*s to %u, but "
2079 "the target went down.\n", res
->lockname
.len
,
2080 res
->lockname
.name
, target
);
2081 spin_lock(&res
->spinlock
);
2082 res
->state
&= ~DLM_LOCK_RES_MIGRATING
;
2083 spin_unlock(&res
->spinlock
);
2089 /* master is known, detach if not already detached */
2090 dlm_mle_detach_hb_events(dlm
, oldmle
);
2091 dlm_put_mle(oldmle
);
2096 dlm_mle_detach_hb_events(dlm
, mle
);
2099 kmem_cache_free(dlm_mle_cache
, mle
);
2105 * at this point, we have a migration target, an mle
2106 * in the master list, and the MIGRATING flag set on
2111 /* get an extra reference on the mle.
2112 * otherwise the assert_master from the new
2113 * master will destroy this.
2114 * also, make sure that all callers of dlm_get_mle
2115 * take both dlm->spinlock and dlm->master_lock */
2116 spin_lock(&dlm
->spinlock
);
2117 spin_lock(&dlm
->master_lock
);
2119 spin_unlock(&dlm
->master_lock
);
2120 spin_unlock(&dlm
->spinlock
);
2122 /* notify new node and send all lock state */
2123 /* call send_one_lockres with migration flag.
2124 * this serves as notice to the target node that a
2125 * migration is starting. */
2126 ret
= dlm_send_one_lockres(dlm
, res
, mres
, target
,
2127 DLM_MRES_MIGRATION
);
2130 mlog(0, "migration to node %u failed with %d\n",
2132 /* migration failed, detach and clean up mle */
2133 dlm_mle_detach_hb_events(dlm
, mle
);
2139 /* at this point, the target sends a message to all nodes,
2140 * (using dlm_do_migrate_request). this node is skipped since
2141 * we had to put an mle in the list to begin the process. this
2142 * node now waits for target to do an assert master. this node
2143 * will be the last one notified, ensuring that the migration
2144 * is complete everywhere. if the target dies while this is
2145 * going on, some nodes could potentially see the target as the
2146 * master, so it is important that my recovery finds the migration
2147 * mle and sets the master to UNKNONWN. */
2150 /* wait for new node to assert master */
2152 ret
= wait_event_interruptible_timeout(mle
->wq
,
2153 (atomic_read(&mle
->woken
) == 1),
2154 msecs_to_jiffies(5000));
2157 if (atomic_read(&mle
->woken
) == 1 ||
2158 res
->owner
== target
)
2161 mlog(0, "timed out during migration\n");
2162 /* avoid hang during shutdown when migrating lockres
2163 * to a node which also goes down */
2164 if (dlm_is_node_dead(dlm
, target
)) {
2165 mlog(0, "%s:%.*s: expected migration target %u "
2166 "is no longer up. restarting.\n",
2167 dlm
->name
, res
->lockname
.len
,
2168 res
->lockname
.name
, target
);
2172 if (ret
== -ERESTARTSYS
) {
2173 /* migration failed, detach and clean up mle */
2174 dlm_mle_detach_hb_events(dlm
, mle
);
2179 /* TODO: if node died: stop, clean up, return error */
2182 /* all done, set the owner, clear the flag */
2183 spin_lock(&res
->spinlock
);
2184 dlm_set_lockres_owner(dlm
, res
, target
);
2185 res
->state
&= ~DLM_LOCK_RES_MIGRATING
;
2186 dlm_remove_nonlocal_locks(dlm
, res
);
2187 spin_unlock(&res
->spinlock
);
2190 /* master is known, detach if not already detached */
2191 dlm_mle_detach_hb_events(dlm
, mle
);
2195 dlm_lockres_calc_usage(dlm
, res
);
2198 /* re-dirty the lockres if we failed */
2200 dlm_kick_thread(dlm
, res
);
2204 free_page((unsigned long)mres
);
2208 mlog(0, "returning %d\n", ret
);
2211 EXPORT_SYMBOL_GPL(dlm_migrate_lockres
);
2213 int dlm_lock_basts_flushed(struct dlm_ctxt
*dlm
, struct dlm_lock
*lock
)
2216 spin_lock(&dlm
->ast_lock
);
2217 spin_lock(&lock
->spinlock
);
2218 ret
= (list_empty(&lock
->bast_list
) && !lock
->bast_pending
);
2219 spin_unlock(&lock
->spinlock
);
2220 spin_unlock(&dlm
->ast_lock
);
2224 static int dlm_migration_can_proceed(struct dlm_ctxt
*dlm
,
2225 struct dlm_lock_resource
*res
,
2229 spin_lock(&res
->spinlock
);
2230 can_proceed
= !!(res
->state
& DLM_LOCK_RES_MIGRATING
);
2231 spin_unlock(&res
->spinlock
);
2233 /* target has died, so make the caller break out of the
2234 * wait_event, but caller must recheck the domain_map */
2235 spin_lock(&dlm
->spinlock
);
2236 if (!test_bit(mig_target
, dlm
->domain_map
))
2238 spin_unlock(&dlm
->spinlock
);
2242 int dlm_lockres_is_dirty(struct dlm_ctxt
*dlm
, struct dlm_lock_resource
*res
)
2245 spin_lock(&res
->spinlock
);
2246 ret
= !!(res
->state
& DLM_LOCK_RES_DIRTY
);
2247 spin_unlock(&res
->spinlock
);
2252 static int dlm_mark_lockres_migrating(struct dlm_ctxt
*dlm
,
2253 struct dlm_lock_resource
*res
,
2258 mlog(0, "dlm_mark_lockres_migrating: %.*s, from %u to %u\n",
2259 res
->lockname
.len
, res
->lockname
.name
, dlm
->node_num
,
2261 /* need to set MIGRATING flag on lockres. this is done by
2262 * ensuring that all asts have been flushed for this lockres. */
2263 spin_lock(&res
->spinlock
);
2264 BUG_ON(res
->migration_pending
);
2265 res
->migration_pending
= 1;
2266 /* strategy is to reserve an extra ast then release
2267 * it below, letting the release do all of the work */
2268 __dlm_lockres_reserve_ast(res
);
2269 spin_unlock(&res
->spinlock
);
2271 /* now flush all the pending asts.. hang out for a bit */
2272 dlm_kick_thread(dlm
, res
);
2273 wait_event(dlm
->ast_wq
, !dlm_lockres_is_dirty(dlm
, res
));
2274 dlm_lockres_release_ast(dlm
, res
);
2276 mlog(0, "about to wait on migration_wq, dirty=%s\n",
2277 res
->state
& DLM_LOCK_RES_DIRTY
? "yes" : "no");
2278 /* if the extra ref we just put was the final one, this
2279 * will pass thru immediately. otherwise, we need to wait
2280 * for the last ast to finish. */
2282 ret
= wait_event_interruptible_timeout(dlm
->migration_wq
,
2283 dlm_migration_can_proceed(dlm
, res
, target
),
2284 msecs_to_jiffies(1000));
2286 mlog(0, "woken again: migrating? %s, dead? %s\n",
2287 res
->state
& DLM_LOCK_RES_MIGRATING
? "yes":"no",
2288 test_bit(target
, dlm
->domain_map
) ? "no":"yes");
2290 mlog(0, "all is well: migrating? %s, dead? %s\n",
2291 res
->state
& DLM_LOCK_RES_MIGRATING
? "yes":"no",
2292 test_bit(target
, dlm
->domain_map
) ? "no":"yes");
2294 if (!dlm_migration_can_proceed(dlm
, res
, target
)) {
2295 mlog(0, "trying again...\n");
2299 /* did the target go down or die? */
2300 spin_lock(&dlm
->spinlock
);
2301 if (!test_bit(target
, dlm
->domain_map
)) {
2302 mlog(ML_ERROR
, "aha. migration target %u just went down\n",
2306 spin_unlock(&dlm
->spinlock
);
2311 * o the DLM_LOCK_RES_MIGRATING flag is set
2312 * o there are no pending asts on this lockres
2313 * o all processes trying to reserve an ast on this
2314 * lockres must wait for the MIGRATING flag to clear
2319 /* last step in the migration process.
2320 * original master calls this to free all of the dlm_lock
2321 * structures that used to be for other nodes. */
2322 static void dlm_remove_nonlocal_locks(struct dlm_ctxt
*dlm
,
2323 struct dlm_lock_resource
*res
)
2325 struct list_head
*iter
, *iter2
;
2326 struct list_head
*queue
= &res
->granted
;
2328 struct dlm_lock
*lock
;
2330 assert_spin_locked(&res
->spinlock
);
2332 BUG_ON(res
->owner
== dlm
->node_num
);
2334 for (i
=0; i
<3; i
++) {
2335 list_for_each_safe(iter
, iter2
, queue
) {
2336 lock
= list_entry (iter
, struct dlm_lock
, list
);
2337 if (lock
->ml
.node
!= dlm
->node_num
) {
2338 mlog(0, "putting lock for node %u\n",
2340 /* be extra careful */
2341 BUG_ON(!list_empty(&lock
->ast_list
));
2342 BUG_ON(!list_empty(&lock
->bast_list
));
2343 BUG_ON(lock
->ast_pending
);
2344 BUG_ON(lock
->bast_pending
);
2345 list_del_init(&lock
->list
);
2353 /* for now this is not too intelligent. we will
2354 * need stats to make this do the right thing.
2355 * this just finds the first lock on one of the
2356 * queues and uses that node as the target. */
2357 static u8
dlm_pick_migration_target(struct dlm_ctxt
*dlm
,
2358 struct dlm_lock_resource
*res
)
2361 struct list_head
*queue
= &res
->granted
;
2362 struct list_head
*iter
;
2363 struct dlm_lock
*lock
;
2366 assert_spin_locked(&dlm
->spinlock
);
2368 spin_lock(&res
->spinlock
);
2369 for (i
=0; i
<3; i
++) {
2370 list_for_each(iter
, queue
) {
2371 /* up to the caller to make sure this node
2373 lock
= list_entry (iter
, struct dlm_lock
, list
);
2374 if (lock
->ml
.node
!= dlm
->node_num
) {
2375 spin_unlock(&res
->spinlock
);
2376 return lock
->ml
.node
;
2381 spin_unlock(&res
->spinlock
);
2382 mlog(0, "have not found a suitable target yet! checking domain map\n");
2384 /* ok now we're getting desperate. pick anyone alive. */
2387 nodenum
= find_next_bit(dlm
->domain_map
,
2388 O2NM_MAX_NODES
, nodenum
+1);
2389 mlog(0, "found %d in domain map\n", nodenum
);
2390 if (nodenum
>= O2NM_MAX_NODES
)
2392 if (nodenum
!= dlm
->node_num
) {
2393 mlog(0, "picking %d\n", nodenum
);
2398 mlog(0, "giving up. no master to migrate to\n");
2399 return DLM_LOCK_RES_OWNER_UNKNOWN
;
2404 /* this is called by the new master once all lockres
2405 * data has been received */
2406 static int dlm_do_migrate_request(struct dlm_ctxt
*dlm
,
2407 struct dlm_lock_resource
*res
,
2408 u8 master
, u8 new_master
,
2409 struct dlm_node_iter
*iter
)
2411 struct dlm_migrate_request migrate
;
2412 int ret
, status
= 0;
2415 memset(&migrate
, 0, sizeof(migrate
));
2416 migrate
.namelen
= res
->lockname
.len
;
2417 memcpy(migrate
.name
, res
->lockname
.name
, migrate
.namelen
);
2418 migrate
.new_master
= new_master
;
2419 migrate
.master
= master
;
2423 /* send message to all nodes, except the master and myself */
2424 while ((nodenum
= dlm_node_iter_next(iter
)) >= 0) {
2425 if (nodenum
== master
||
2426 nodenum
== new_master
)
2429 ret
= o2net_send_message(DLM_MIGRATE_REQUEST_MSG
, dlm
->key
,
2430 &migrate
, sizeof(migrate
), nodenum
,
2434 else if (status
< 0) {
2435 mlog(0, "migrate request (node %u) returned %d!\n",
2444 mlog(0, "returning ret=%d\n", ret
);
2449 /* if there is an existing mle for this lockres, we now know who the master is.
2450 * (the one who sent us *this* message) we can clear it up right away.
2451 * since the process that put the mle on the list still has a reference to it,
2452 * we can unhash it now, set the master and wake the process. as a result,
2453 * we will have no mle in the list to start with. now we can add an mle for
2454 * the migration and this should be the only one found for those scanning the
2456 int dlm_migrate_request_handler(struct o2net_msg
*msg
, u32 len
, void *data
)
2458 struct dlm_ctxt
*dlm
= data
;
2459 struct dlm_lock_resource
*res
= NULL
;
2460 struct dlm_migrate_request
*migrate
= (struct dlm_migrate_request
*) msg
->buf
;
2461 struct dlm_master_list_entry
*mle
= NULL
, *oldmle
= NULL
;
2463 unsigned int namelen
, hash
;
2469 name
= migrate
->name
;
2470 namelen
= migrate
->namelen
;
2471 hash
= dlm_lockid_hash(name
, namelen
);
2473 /* preallocate.. if this fails, abort */
2474 mle
= (struct dlm_master_list_entry
*) kmem_cache_alloc(dlm_mle_cache
,
2482 /* check for pre-existing lock */
2483 spin_lock(&dlm
->spinlock
);
2484 res
= __dlm_lookup_lockres(dlm
, name
, namelen
, hash
);
2485 spin_lock(&dlm
->master_lock
);
2488 spin_lock(&res
->spinlock
);
2489 if (res
->state
& DLM_LOCK_RES_RECOVERING
) {
2490 /* if all is working ok, this can only mean that we got
2491 * a migrate request from a node that we now see as
2492 * dead. what can we do here? drop it to the floor? */
2493 spin_unlock(&res
->spinlock
);
2494 mlog(ML_ERROR
, "Got a migrate request, but the "
2495 "lockres is marked as recovering!");
2496 kmem_cache_free(dlm_mle_cache
, mle
);
2497 ret
= -EINVAL
; /* need a better solution */
2500 res
->state
|= DLM_LOCK_RES_MIGRATING
;
2501 spin_unlock(&res
->spinlock
);
2504 /* ignore status. only nonzero status would BUG. */
2505 ret
= dlm_add_migration_mle(dlm
, res
, mle
, &oldmle
,
2507 migrate
->new_master
,
2511 spin_unlock(&dlm
->master_lock
);
2512 spin_unlock(&dlm
->spinlock
);
2515 /* master is known, detach if not already detached */
2516 dlm_mle_detach_hb_events(dlm
, oldmle
);
2517 dlm_put_mle(oldmle
);
2521 dlm_lockres_put(res
);
2527 /* must be holding dlm->spinlock and dlm->master_lock
2528 * when adding a migration mle, we can clear any other mles
2529 * in the master list because we know with certainty that
2530 * the master is "master". so we remove any old mle from
2531 * the list after setting it's master field, and then add
2532 * the new migration mle. this way we can hold with the rule
2533 * of having only one mle for a given lock name at all times. */
2534 static int dlm_add_migration_mle(struct dlm_ctxt
*dlm
,
2535 struct dlm_lock_resource
*res
,
2536 struct dlm_master_list_entry
*mle
,
2537 struct dlm_master_list_entry
**oldmle
,
2538 const char *name
, unsigned int namelen
,
2539 u8 new_master
, u8 master
)
2548 assert_spin_locked(&dlm
->spinlock
);
2549 assert_spin_locked(&dlm
->master_lock
);
2551 /* caller is responsible for any ref taken here on oldmle */
2552 found
= dlm_find_mle(dlm
, oldmle
, (char *)name
, namelen
);
2554 struct dlm_master_list_entry
*tmp
= *oldmle
;
2555 spin_lock(&tmp
->spinlock
);
2556 if (tmp
->type
== DLM_MLE_MIGRATION
) {
2557 if (master
== dlm
->node_num
) {
2558 /* ah another process raced me to it */
2559 mlog(0, "tried to migrate %.*s, but some "
2560 "process beat me to it\n",
2564 /* bad. 2 NODES are trying to migrate! */
2565 mlog(ML_ERROR
, "migration error mle: "
2566 "master=%u new_master=%u // request: "
2567 "master=%u new_master=%u // "
2569 tmp
->master
, tmp
->new_master
,
2575 /* this is essentially what assert_master does */
2576 tmp
->master
= master
;
2577 atomic_set(&tmp
->woken
, 1);
2579 /* remove it from the list so that only one
2580 * mle will be found */
2581 list_del_init(&tmp
->list
);
2583 spin_unlock(&tmp
->spinlock
);
2586 /* now add a migration mle to the tail of the list */
2587 dlm_init_mle(mle
, DLM_MLE_MIGRATION
, dlm
, res
, name
, namelen
);
2588 mle
->new_master
= new_master
;
2589 mle
->master
= master
;
2590 /* do this for consistency with other mle types */
2591 set_bit(new_master
, mle
->maybe_map
);
2592 list_add(&mle
->list
, &dlm
->master_list
);
2598 void dlm_clean_master_list(struct dlm_ctxt
*dlm
, u8 dead_node
)
2600 struct list_head
*iter
, *iter2
;
2601 struct dlm_master_list_entry
*mle
;
2602 struct dlm_lock_resource
*res
;
2605 mlog_entry("dlm=%s, dead node=%u\n", dlm
->name
, dead_node
);
2607 assert_spin_locked(&dlm
->spinlock
);
2609 /* clean the master list */
2610 spin_lock(&dlm
->master_lock
);
2611 list_for_each_safe(iter
, iter2
, &dlm
->master_list
) {
2612 mle
= list_entry(iter
, struct dlm_master_list_entry
, list
);
2614 BUG_ON(mle
->type
!= DLM_MLE_BLOCK
&&
2615 mle
->type
!= DLM_MLE_MASTER
&&
2616 mle
->type
!= DLM_MLE_MIGRATION
);
2618 /* MASTER mles are initiated locally. the waiting
2619 * process will notice the node map change
2620 * shortly. let that happen as normal. */
2621 if (mle
->type
== DLM_MLE_MASTER
)
2625 /* BLOCK mles are initiated by other nodes.
2626 * need to clean up if the dead node would have
2627 * been the master. */
2628 if (mle
->type
== DLM_MLE_BLOCK
) {
2631 spin_lock(&mle
->spinlock
);
2632 bit
= find_next_bit(mle
->maybe_map
, O2NM_MAX_NODES
, 0);
2633 if (bit
!= dead_node
) {
2634 mlog(0, "mle found, but dead node %u would "
2635 "not have been master\n", dead_node
);
2636 spin_unlock(&mle
->spinlock
);
2638 /* must drop the refcount by one since the
2639 * assert_master will never arrive. this
2640 * may result in the mle being unlinked and
2641 * freed, but there may still be a process
2642 * waiting in the dlmlock path which is fine. */
2643 mlog(ML_ERROR
, "node %u was expected master\n",
2645 atomic_set(&mle
->woken
, 1);
2646 spin_unlock(&mle
->spinlock
);
2648 /* do not need events any longer, so detach
2650 __dlm_mle_detach_hb_events(dlm
, mle
);
2656 /* everything else is a MIGRATION mle */
2658 /* the rule for MIGRATION mles is that the master
2659 * becomes UNKNOWN if *either* the original or
2660 * the new master dies. all UNKNOWN lockreses
2661 * are sent to whichever node becomes the recovery
2662 * master. the new master is responsible for
2663 * determining if there is still a master for
2664 * this lockres, or if he needs to take over
2665 * mastery. either way, this node should expect
2666 * another message to resolve this. */
2667 if (mle
->master
!= dead_node
&&
2668 mle
->new_master
!= dead_node
)
2671 /* if we have reached this point, this mle needs to
2672 * be removed from the list and freed. */
2674 /* remove from the list early. NOTE: unlinking
2675 * list_head while in list_for_each_safe */
2676 spin_lock(&mle
->spinlock
);
2677 list_del_init(&mle
->list
);
2678 atomic_set(&mle
->woken
, 1);
2679 spin_unlock(&mle
->spinlock
);
2682 mlog(0, "node %u died during migration from "
2683 "%u to %u!\n", dead_node
,
2684 mle
->master
, mle
->new_master
);
2685 /* if there is a lockres associated with this
2686 * mle, find it and set its owner to UNKNOWN */
2687 hash
= dlm_lockid_hash(mle
->u
.name
.name
, mle
->u
.name
.len
);
2688 res
= __dlm_lookup_lockres(dlm
, mle
->u
.name
.name
,
2689 mle
->u
.name
.len
, hash
);
2691 /* unfortunately if we hit this rare case, our
2692 * lock ordering is messed. we need to drop
2693 * the master lock so that we can take the
2694 * lockres lock, meaning that we will have to
2695 * restart from the head of list. */
2696 spin_unlock(&dlm
->master_lock
);
2698 /* move lockres onto recovery list */
2699 spin_lock(&res
->spinlock
);
2700 dlm_set_lockres_owner(dlm
, res
,
2701 DLM_LOCK_RES_OWNER_UNKNOWN
);
2702 dlm_move_lockres_to_recovery_list(dlm
, res
);
2703 spin_unlock(&res
->spinlock
);
2704 dlm_lockres_put(res
);
2706 /* about to get rid of mle, detach from heartbeat */
2707 __dlm_mle_detach_hb_events(dlm
, mle
);
2710 spin_lock(&dlm
->master_lock
);
2712 spin_unlock(&dlm
->master_lock
);
2718 /* this may be the last reference */
2721 spin_unlock(&dlm
->master_lock
);
2725 int dlm_finish_migration(struct dlm_ctxt
*dlm
, struct dlm_lock_resource
*res
,
2728 struct dlm_node_iter iter
;
2731 spin_lock(&dlm
->spinlock
);
2732 dlm_node_iter_init(dlm
->domain_map
, &iter
);
2733 clear_bit(old_master
, iter
.node_map
);
2734 clear_bit(dlm
->node_num
, iter
.node_map
);
2735 spin_unlock(&dlm
->spinlock
);
2737 mlog(0, "now time to do a migrate request to other nodes\n");
2738 ret
= dlm_do_migrate_request(dlm
, res
, old_master
,
2739 dlm
->node_num
, &iter
);
2745 mlog(0, "doing assert master of %.*s to all except the original node\n",
2746 res
->lockname
.len
, res
->lockname
.name
);
2747 /* this call now finishes out the nodemap
2748 * even if one or more nodes die */
2749 ret
= dlm_do_assert_master(dlm
, res
->lockname
.name
,
2750 res
->lockname
.len
, iter
.node_map
,
2751 DLM_ASSERT_MASTER_FINISH_MIGRATION
);
2753 /* no longer need to retry. all living nodes contacted. */
2758 memset(iter
.node_map
, 0, sizeof(iter
.node_map
));
2759 set_bit(old_master
, iter
.node_map
);
2760 mlog(0, "doing assert master of %.*s back to %u\n",
2761 res
->lockname
.len
, res
->lockname
.name
, old_master
);
2762 ret
= dlm_do_assert_master(dlm
, res
->lockname
.name
,
2763 res
->lockname
.len
, iter
.node_map
,
2764 DLM_ASSERT_MASTER_FINISH_MIGRATION
);
2766 mlog(0, "assert master to original master failed "
2768 /* the only nonzero status here would be because of
2769 * a dead original node. we're done. */
2773 /* all done, set the owner, clear the flag */
2774 spin_lock(&res
->spinlock
);
2775 dlm_set_lockres_owner(dlm
, res
, dlm
->node_num
);
2776 res
->state
&= ~DLM_LOCK_RES_MIGRATING
;
2777 spin_unlock(&res
->spinlock
);
2778 /* re-dirty it on the new master */
2779 dlm_kick_thread(dlm
, res
);
2786 * LOCKRES AST REFCOUNT
2787 * this is integral to migration
2790 /* for future intent to call an ast, reserve one ahead of time.
2791 * this should be called only after waiting on the lockres
2792 * with dlm_wait_on_lockres, and while still holding the
2793 * spinlock after the call. */
2794 void __dlm_lockres_reserve_ast(struct dlm_lock_resource
*res
)
2796 assert_spin_locked(&res
->spinlock
);
2797 if (res
->state
& DLM_LOCK_RES_MIGRATING
) {
2798 __dlm_print_one_lock_resource(res
);
2800 BUG_ON(res
->state
& DLM_LOCK_RES_MIGRATING
);
2802 atomic_inc(&res
->asts_reserved
);
2806 * used to drop the reserved ast, either because it went unused,
2807 * or because the ast/bast was actually called.
2809 * also, if there is a pending migration on this lockres,
2810 * and this was the last pending ast on the lockres,
2811 * atomically set the MIGRATING flag before we drop the lock.
2812 * this is how we ensure that migration can proceed with no
2813 * asts in progress. note that it is ok if the state of the
2814 * queues is such that a lock should be granted in the future
2815 * or that a bast should be fired, because the new master will
2816 * shuffle the lists on this lockres as soon as it is migrated.
2818 void dlm_lockres_release_ast(struct dlm_ctxt
*dlm
,
2819 struct dlm_lock_resource
*res
)
2821 if (!atomic_dec_and_lock(&res
->asts_reserved
, &res
->spinlock
))
2824 if (!res
->migration_pending
) {
2825 spin_unlock(&res
->spinlock
);
2829 BUG_ON(res
->state
& DLM_LOCK_RES_MIGRATING
);
2830 res
->migration_pending
= 0;
2831 res
->state
|= DLM_LOCK_RES_MIGRATING
;
2832 spin_unlock(&res
->spinlock
);
2834 wake_up(&dlm
->migration_wq
);