1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
5 ** Copyright (C) 2004-2005 Red Hat, Inc. All rights reserved.
7 ** This copyrighted material is made available to anyone wishing to use,
8 ** modify, copy, or redistribute it subject to the terms and conditions
9 ** of the GNU General Public License v.2.
11 *******************************************************************************
12 ******************************************************************************/
14 #include "dlm_internal.h"
15 #include "lockspace.h"
28 * Recovery waiting routines: these functions wait for a particular reply from
29 * a remote node, or for the remote node to report a certain status. They need
30 * to abort if the lockspace is stopped indicating a node has failed (perhaps
31 * the one being waited for).
35 * Wait until given function returns non-zero or lockspace is stopped
36 * (LS_RECOVERY_STOP set due to failure of a node in ls_nodes). When another
37 * function thinks it could have completed the waited-on task, they should wake
38 * up ls_wait_general to get an immediate response rather than waiting for the
39 * timer to detect the result. A timer wakes us up periodically while waiting
40 * to see if we should abort due to a node failure. This should only be called
41 * by the dlm_recoverd thread.
44 static void dlm_wait_timer_fn(unsigned long data
)
46 struct dlm_ls
*ls
= (struct dlm_ls
*) data
;
47 mod_timer(&ls
->ls_timer
, jiffies
+ (dlm_config
.ci_recover_timer
* HZ
));
48 wake_up(&ls
->ls_wait_general
);
51 int dlm_wait_function(struct dlm_ls
*ls
, int (*testfn
) (struct dlm_ls
*ls
))
55 init_timer(&ls
->ls_timer
);
56 ls
->ls_timer
.function
= dlm_wait_timer_fn
;
57 ls
->ls_timer
.data
= (long) ls
;
58 ls
->ls_timer
.expires
= jiffies
+ (dlm_config
.ci_recover_timer
* HZ
);
59 add_timer(&ls
->ls_timer
);
61 wait_event(ls
->ls_wait_general
, testfn(ls
) || dlm_recovery_stopped(ls
));
62 del_timer_sync(&ls
->ls_timer
);
64 if (dlm_recovery_stopped(ls
)) {
65 log_debug(ls
, "dlm_wait_function aborted");
72 * An efficient way for all nodes to wait for all others to have a certain
73 * status. The node with the lowest nodeid polls all the others for their
74 * status (wait_status_all) and all the others poll the node with the low id
75 * for its accumulated result (wait_status_low). When all nodes have set
76 * status flag X, then status flag X_ALL will be set on the low nodeid.
79 uint32_t dlm_recover_status(struct dlm_ls
*ls
)
82 spin_lock(&ls
->ls_recover_lock
);
83 status
= ls
->ls_recover_status
;
84 spin_unlock(&ls
->ls_recover_lock
);
88 static void _set_recover_status(struct dlm_ls
*ls
, uint32_t status
)
90 ls
->ls_recover_status
|= status
;
93 void dlm_set_recover_status(struct dlm_ls
*ls
, uint32_t status
)
95 spin_lock(&ls
->ls_recover_lock
);
96 _set_recover_status(ls
, status
);
97 spin_unlock(&ls
->ls_recover_lock
);
100 static int wait_status_all(struct dlm_ls
*ls
, uint32_t wait_status
,
103 struct dlm_rcom
*rc
= ls
->ls_recover_buf
;
104 struct dlm_member
*memb
;
105 int error
= 0, delay
;
107 list_for_each_entry(memb
, &ls
->ls_nodes
, list
) {
110 if (dlm_recovery_stopped(ls
)) {
115 error
= dlm_rcom_status(ls
, memb
->nodeid
, 0);
120 dlm_slot_save(ls
, rc
, memb
);
122 if (rc
->rc_result
& wait_status
)
133 static int wait_status_low(struct dlm_ls
*ls
, uint32_t wait_status
,
134 uint32_t status_flags
)
136 struct dlm_rcom
*rc
= ls
->ls_recover_buf
;
137 int error
= 0, delay
= 0, nodeid
= ls
->ls_low_nodeid
;
140 if (dlm_recovery_stopped(ls
)) {
145 error
= dlm_rcom_status(ls
, nodeid
, status_flags
);
149 if (rc
->rc_result
& wait_status
)
159 static int wait_status(struct dlm_ls
*ls
, uint32_t status
)
161 uint32_t status_all
= status
<< 1;
164 if (ls
->ls_low_nodeid
== dlm_our_nodeid()) {
165 error
= wait_status_all(ls
, status
, 0);
167 dlm_set_recover_status(ls
, status_all
);
169 error
= wait_status_low(ls
, status_all
, 0);
174 int dlm_recover_members_wait(struct dlm_ls
*ls
)
176 struct dlm_member
*memb
;
177 struct dlm_slot
*slots
;
178 int num_slots
, slots_size
;
182 list_for_each_entry(memb
, &ls
->ls_nodes
, list
) {
184 memb
->generation
= 0;
187 if (ls
->ls_low_nodeid
== dlm_our_nodeid()) {
188 error
= wait_status_all(ls
, DLM_RS_NODES
, 1);
192 /* slots array is sparse, slots_size may be > num_slots */
194 rv
= dlm_slots_assign(ls
, &num_slots
, &slots_size
, &slots
, &gen
);
196 spin_lock(&ls
->ls_recover_lock
);
197 _set_recover_status(ls
, DLM_RS_NODES_ALL
);
198 ls
->ls_num_slots
= num_slots
;
199 ls
->ls_slots_size
= slots_size
;
200 ls
->ls_slots
= slots
;
201 ls
->ls_generation
= gen
;
202 spin_unlock(&ls
->ls_recover_lock
);
204 dlm_set_recover_status(ls
, DLM_RS_NODES_ALL
);
207 error
= wait_status_low(ls
, DLM_RS_NODES_ALL
, DLM_RSF_NEED_SLOTS
);
211 dlm_slots_copy_in(ls
);
217 int dlm_recover_directory_wait(struct dlm_ls
*ls
)
219 return wait_status(ls
, DLM_RS_DIR
);
222 int dlm_recover_locks_wait(struct dlm_ls
*ls
)
224 return wait_status(ls
, DLM_RS_LOCKS
);
227 int dlm_recover_done_wait(struct dlm_ls
*ls
)
229 return wait_status(ls
, DLM_RS_DONE
);
233 * The recover_list contains all the rsb's for which we've requested the new
234 * master nodeid. As replies are returned from the resource directories the
235 * rsb's are removed from the list. When the list is empty we're done.
237 * The recover_list is later similarly used for all rsb's for which we've sent
238 * new lkb's and need to receive new corresponding lkid's.
240 * We use the address of the rsb struct as a simple local identifier for the
241 * rsb so we can match an rcom reply with the rsb it was sent for.
244 static int recover_list_empty(struct dlm_ls
*ls
)
248 spin_lock(&ls
->ls_recover_list_lock
);
249 empty
= list_empty(&ls
->ls_recover_list
);
250 spin_unlock(&ls
->ls_recover_list_lock
);
255 static void recover_list_add(struct dlm_rsb
*r
)
257 struct dlm_ls
*ls
= r
->res_ls
;
259 spin_lock(&ls
->ls_recover_list_lock
);
260 if (list_empty(&r
->res_recover_list
)) {
261 list_add_tail(&r
->res_recover_list
, &ls
->ls_recover_list
);
262 ls
->ls_recover_list_count
++;
265 spin_unlock(&ls
->ls_recover_list_lock
);
268 static void recover_list_del(struct dlm_rsb
*r
)
270 struct dlm_ls
*ls
= r
->res_ls
;
272 spin_lock(&ls
->ls_recover_list_lock
);
273 list_del_init(&r
->res_recover_list
);
274 ls
->ls_recover_list_count
--;
275 spin_unlock(&ls
->ls_recover_list_lock
);
280 static struct dlm_rsb
*recover_list_find(struct dlm_ls
*ls
, uint64_t id
)
282 struct dlm_rsb
*r
= NULL
;
284 spin_lock(&ls
->ls_recover_list_lock
);
286 list_for_each_entry(r
, &ls
->ls_recover_list
, res_recover_list
) {
287 if (id
== (unsigned long) r
)
292 spin_unlock(&ls
->ls_recover_list_lock
);
296 static void recover_list_clear(struct dlm_ls
*ls
)
298 struct dlm_rsb
*r
, *s
;
300 spin_lock(&ls
->ls_recover_list_lock
);
301 list_for_each_entry_safe(r
, s
, &ls
->ls_recover_list
, res_recover_list
) {
302 list_del_init(&r
->res_recover_list
);
303 r
->res_recover_locks_count
= 0;
305 ls
->ls_recover_list_count
--;
308 if (ls
->ls_recover_list_count
!= 0) {
309 log_error(ls
, "warning: recover_list_count %d",
310 ls
->ls_recover_list_count
);
311 ls
->ls_recover_list_count
= 0;
313 spin_unlock(&ls
->ls_recover_list_lock
);
317 /* Master recovery: find new master node for rsb's that were
318 mastered on nodes that have been removed.
322 dlm_send_rcom_lookup -> receive_rcom_lookup
324 receive_rcom_lookup_reply <-
325 dlm_recover_master_reply
332 * Set the lock master for all LKBs in a lock queue
333 * If we are the new master of the rsb, we may have received new
334 * MSTCPY locks from other nodes already which we need to ignore
335 * when setting the new nodeid.
338 static void set_lock_master(struct list_head
*queue
, int nodeid
)
342 list_for_each_entry(lkb
, queue
, lkb_statequeue
) {
343 if (!(lkb
->lkb_flags
& DLM_IFL_MSTCPY
)) {
344 lkb
->lkb_nodeid
= nodeid
;
350 static void set_master_lkbs(struct dlm_rsb
*r
)
352 set_lock_master(&r
->res_grantqueue
, r
->res_nodeid
);
353 set_lock_master(&r
->res_convertqueue
, r
->res_nodeid
);
354 set_lock_master(&r
->res_waitqueue
, r
->res_nodeid
);
358 * Propagate the new master nodeid to locks
359 * The NEW_MASTER flag tells dlm_recover_locks() which rsb's to consider.
360 * The NEW_MASTER2 flag tells recover_lvb() and recover_grant() which
364 static void set_new_master(struct dlm_rsb
*r
, int nodeid
)
366 r
->res_nodeid
= nodeid
;
368 rsb_set_flag(r
, RSB_NEW_MASTER
);
369 rsb_set_flag(r
, RSB_NEW_MASTER2
);
373 * We do async lookups on rsb's that need new masters. The rsb's
374 * waiting for a lookup reply are kept on the recover_list.
377 static int recover_master(struct dlm_rsb
*r
)
379 struct dlm_ls
*ls
= r
->res_ls
;
380 int error
, ret_nodeid
;
381 int our_nodeid
= dlm_our_nodeid();
382 int dir_nodeid
= dlm_dir_nodeid(r
);
384 if (dir_nodeid
== our_nodeid
) {
385 error
= dlm_dir_lookup(ls
, our_nodeid
, r
->res_name
,
386 r
->res_length
, &ret_nodeid
);
388 log_error(ls
, "recover dir lookup error %d", error
);
390 if (ret_nodeid
== our_nodeid
)
393 set_new_master(r
, ret_nodeid
);
397 error
= dlm_send_rcom_lookup(r
, dir_nodeid
);
404 * All MSTCPY locks are purged and rebuilt, even if the master stayed the same.
405 * This is necessary because recovery can be started, aborted and restarted,
406 * causing the master nodeid to briefly change during the aborted recovery, and
407 * change back to the original value in the second recovery. The MSTCPY locks
408 * may or may not have been purged during the aborted recovery. Another node
409 * with an outstanding request in waiters list and a request reply saved in the
410 * requestqueue, cannot know whether it should ignore the reply and resend the
411 * request, or accept the reply and complete the request. It must do the
412 * former if the remote node purged MSTCPY locks, and it must do the later if
413 * the remote node did not. This is solved by always purging MSTCPY locks, in
414 * which case, the request reply would always be ignored and the request
418 static int recover_master_static(struct dlm_rsb
*r
)
420 int dir_nodeid
= dlm_dir_nodeid(r
);
421 int new_master
= dir_nodeid
;
423 if (dir_nodeid
== dlm_our_nodeid())
427 dlm_purge_mstcpy_locks(r
);
428 set_new_master(r
, new_master
);
434 * Go through local root resources and for each rsb which has a master which
435 * has departed, get the new master nodeid from the directory. The dir will
436 * assign mastery to the first node to look up the new master. That means
437 * we'll discover in this lookup if we're the new master of any rsb's.
439 * We fire off all the dir lookup requests individually and asynchronously to
440 * the correct dir node.
443 int dlm_recover_masters(struct dlm_ls
*ls
)
446 int error
= 0, count
= 0;
448 log_debug(ls
, "dlm_recover_masters");
450 down_read(&ls
->ls_root_sem
);
451 list_for_each_entry(r
, &ls
->ls_root_list
, res_root_list
) {
452 if (dlm_recovery_stopped(ls
)) {
453 up_read(&ls
->ls_root_sem
);
458 if (dlm_no_directory(ls
))
459 count
+= recover_master_static(r
);
460 else if (!is_master(r
) &&
461 (dlm_is_removed(ls
, r
->res_nodeid
) ||
462 rsb_flag(r
, RSB_NEW_MASTER
))) {
469 up_read(&ls
->ls_root_sem
);
471 log_debug(ls
, "dlm_recover_masters %d resources", count
);
473 error
= dlm_wait_function(ls
, &recover_list_empty
);
476 recover_list_clear(ls
);
480 int dlm_recover_master_reply(struct dlm_ls
*ls
, struct dlm_rcom
*rc
)
485 r
= recover_list_find(ls
, rc
->rc_id
);
487 log_error(ls
, "dlm_recover_master_reply no id %llx",
488 (unsigned long long)rc
->rc_id
);
492 nodeid
= rc
->rc_result
;
493 if (nodeid
== dlm_our_nodeid())
497 set_new_master(r
, nodeid
);
501 if (recover_list_empty(ls
))
502 wake_up(&ls
->ls_wait_general
);
508 /* Lock recovery: rebuild the process-copy locks we hold on a
509 remastered rsb on the new rsb master.
514 dlm_send_rcom_lock -> receive_rcom_lock
515 dlm_recover_master_copy
516 receive_rcom_lock_reply <-
517 dlm_recover_process_copy
522 * keep a count of the number of lkb's we send to the new master; when we get
523 * an equal number of replies then recovery for the rsb is done
526 static int recover_locks_queue(struct dlm_rsb
*r
, struct list_head
*head
)
531 list_for_each_entry(lkb
, head
, lkb_statequeue
) {
532 error
= dlm_send_rcom_lock(r
, lkb
);
535 r
->res_recover_locks_count
++;
541 static int recover_locks(struct dlm_rsb
*r
)
547 DLM_ASSERT(!r
->res_recover_locks_count
, dlm_dump_rsb(r
););
549 error
= recover_locks_queue(r
, &r
->res_grantqueue
);
552 error
= recover_locks_queue(r
, &r
->res_convertqueue
);
555 error
= recover_locks_queue(r
, &r
->res_waitqueue
);
559 if (r
->res_recover_locks_count
)
562 rsb_clear_flag(r
, RSB_NEW_MASTER
);
568 int dlm_recover_locks(struct dlm_ls
*ls
)
571 int error
, count
= 0;
573 down_read(&ls
->ls_root_sem
);
574 list_for_each_entry(r
, &ls
->ls_root_list
, res_root_list
) {
576 rsb_clear_flag(r
, RSB_NEW_MASTER
);
580 if (!rsb_flag(r
, RSB_NEW_MASTER
))
583 if (dlm_recovery_stopped(ls
)) {
585 up_read(&ls
->ls_root_sem
);
589 error
= recover_locks(r
);
591 up_read(&ls
->ls_root_sem
);
595 count
+= r
->res_recover_locks_count
;
597 up_read(&ls
->ls_root_sem
);
599 log_debug(ls
, "dlm_recover_locks %d out", count
);
601 error
= dlm_wait_function(ls
, &recover_list_empty
);
604 recover_list_clear(ls
);
608 void dlm_recovered_lock(struct dlm_rsb
*r
)
610 DLM_ASSERT(rsb_flag(r
, RSB_NEW_MASTER
), dlm_dump_rsb(r
););
612 r
->res_recover_locks_count
--;
613 if (!r
->res_recover_locks_count
) {
614 rsb_clear_flag(r
, RSB_NEW_MASTER
);
618 if (recover_list_empty(r
->res_ls
))
619 wake_up(&r
->res_ls
->ls_wait_general
);
623 * The lvb needs to be recovered on all master rsb's. This includes setting
624 * the VALNOTVALID flag if necessary, and determining the correct lvb contents
625 * based on the lvb's of the locks held on the rsb.
627 * RSB_VALNOTVALID is set if there are only NL/CR locks on the rsb. If it
628 * was already set prior to recovery, it's not cleared, regardless of locks.
630 * The LVB contents are only considered for changing when this is a new master
631 * of the rsb (NEW_MASTER2). Then, the rsb's lvb is taken from any lkb with
632 * mode > CR. If no lkb's exist with mode above CR, the lvb contents are taken
633 * from the lkb with the largest lvb sequence number.
636 static void recover_lvb(struct dlm_rsb
*r
)
638 struct dlm_lkb
*lkb
, *high_lkb
= NULL
;
639 uint32_t high_seq
= 0;
640 int lock_lvb_exists
= 0;
641 int big_lock_exists
= 0;
642 int lvblen
= r
->res_ls
->ls_lvblen
;
644 list_for_each_entry(lkb
, &r
->res_grantqueue
, lkb_statequeue
) {
645 if (!(lkb
->lkb_exflags
& DLM_LKF_VALBLK
))
650 if (lkb
->lkb_grmode
> DLM_LOCK_CR
) {
655 if (((int)lkb
->lkb_lvbseq
- (int)high_seq
) >= 0) {
657 high_seq
= lkb
->lkb_lvbseq
;
661 list_for_each_entry(lkb
, &r
->res_convertqueue
, lkb_statequeue
) {
662 if (!(lkb
->lkb_exflags
& DLM_LKF_VALBLK
))
667 if (lkb
->lkb_grmode
> DLM_LOCK_CR
) {
672 if (((int)lkb
->lkb_lvbseq
- (int)high_seq
) >= 0) {
674 high_seq
= lkb
->lkb_lvbseq
;
679 if (!lock_lvb_exists
)
682 if (!big_lock_exists
)
683 rsb_set_flag(r
, RSB_VALNOTVALID
);
685 /* don't mess with the lvb unless we're the new master */
686 if (!rsb_flag(r
, RSB_NEW_MASTER2
))
689 if (!r
->res_lvbptr
) {
690 r
->res_lvbptr
= dlm_allocate_lvb(r
->res_ls
);
695 if (big_lock_exists
) {
696 r
->res_lvbseq
= lkb
->lkb_lvbseq
;
697 memcpy(r
->res_lvbptr
, lkb
->lkb_lvbptr
, lvblen
);
698 } else if (high_lkb
) {
699 r
->res_lvbseq
= high_lkb
->lkb_lvbseq
;
700 memcpy(r
->res_lvbptr
, high_lkb
->lkb_lvbptr
, lvblen
);
703 memset(r
->res_lvbptr
, 0, lvblen
);
709 /* All master rsb's flagged RECOVER_CONVERT need to be looked at. The locks
710 converting PR->CW or CW->PR need to have their lkb_grmode set. */
712 static void recover_conversion(struct dlm_rsb
*r
)
717 list_for_each_entry(lkb
, &r
->res_grantqueue
, lkb_statequeue
) {
718 if (lkb
->lkb_grmode
== DLM_LOCK_PR
||
719 lkb
->lkb_grmode
== DLM_LOCK_CW
) {
720 grmode
= lkb
->lkb_grmode
;
725 list_for_each_entry(lkb
, &r
->res_convertqueue
, lkb_statequeue
) {
726 if (lkb
->lkb_grmode
!= DLM_LOCK_IV
)
729 lkb
->lkb_grmode
= lkb
->lkb_rqmode
;
731 lkb
->lkb_grmode
= grmode
;
735 /* We've become the new master for this rsb and waiting/converting locks may
736 need to be granted in dlm_recover_grant() due to locks that may have
737 existed from a removed node. */
739 static void recover_grant(struct dlm_rsb
*r
)
741 if (!list_empty(&r
->res_waitqueue
) || !list_empty(&r
->res_convertqueue
))
742 rsb_set_flag(r
, RSB_RECOVER_GRANT
);
745 void dlm_recover_rsbs(struct dlm_ls
*ls
)
748 unsigned int count
= 0;
750 down_read(&ls
->ls_root_sem
);
751 list_for_each_entry(r
, &ls
->ls_root_list
, res_root_list
) {
754 if (rsb_flag(r
, RSB_RECOVER_CONVERT
))
755 recover_conversion(r
);
756 if (rsb_flag(r
, RSB_NEW_MASTER2
))
761 rsb_clear_flag(r
, RSB_RECOVER_CONVERT
);
762 rsb_clear_flag(r
, RSB_NEW_MASTER2
);
765 up_read(&ls
->ls_root_sem
);
768 log_debug(ls
, "dlm_recover_rsbs %d done", count
);
771 /* Create a single list of all root rsb's to be used during recovery */
773 int dlm_create_root_list(struct dlm_ls
*ls
)
779 down_write(&ls
->ls_root_sem
);
780 if (!list_empty(&ls
->ls_root_list
)) {
781 log_error(ls
, "root list not empty");
786 for (i
= 0; i
< ls
->ls_rsbtbl_size
; i
++) {
787 spin_lock(&ls
->ls_rsbtbl
[i
].lock
);
788 for (n
= rb_first(&ls
->ls_rsbtbl
[i
].keep
); n
; n
= rb_next(n
)) {
789 r
= rb_entry(n
, struct dlm_rsb
, res_hashnode
);
790 list_add(&r
->res_root_list
, &ls
->ls_root_list
);
794 /* If we're using a directory, add tossed rsbs to the root
795 list; they'll have entries created in the new directory,
796 but no other recovery steps should do anything with them. */
798 if (dlm_no_directory(ls
)) {
799 spin_unlock(&ls
->ls_rsbtbl
[i
].lock
);
803 for (n
= rb_first(&ls
->ls_rsbtbl
[i
].toss
); n
; n
= rb_next(n
)) {
804 r
= rb_entry(n
, struct dlm_rsb
, res_hashnode
);
805 list_add(&r
->res_root_list
, &ls
->ls_root_list
);
808 spin_unlock(&ls
->ls_rsbtbl
[i
].lock
);
811 up_write(&ls
->ls_root_sem
);
815 void dlm_release_root_list(struct dlm_ls
*ls
)
817 struct dlm_rsb
*r
, *safe
;
819 down_write(&ls
->ls_root_sem
);
820 list_for_each_entry_safe(r
, safe
, &ls
->ls_root_list
, res_root_list
) {
821 list_del_init(&r
->res_root_list
);
824 up_write(&ls
->ls_root_sem
);
827 /* If not using a directory, clear the entire toss list, there's no benefit to
828 caching the master value since it's fixed. If we are using a dir, keep the
829 rsb's we're the master of. Recovery will add them to the root list and from
830 there they'll be entered in the rebuilt directory. */
832 void dlm_clear_toss_list(struct dlm_ls
*ls
)
834 struct rb_node
*n
, *next
;
838 for (i
= 0; i
< ls
->ls_rsbtbl_size
; i
++) {
839 spin_lock(&ls
->ls_rsbtbl
[i
].lock
);
840 for (n
= rb_first(&ls
->ls_rsbtbl
[i
].toss
); n
; n
= next
) {
842 rsb
= rb_entry(n
, struct dlm_rsb
, res_hashnode
);
843 if (dlm_no_directory(ls
) || !is_master(rsb
)) {
844 rb_erase(n
, &ls
->ls_rsbtbl
[i
].toss
);
848 spin_unlock(&ls
->ls_rsbtbl
[i
].lock
);