/*
* Propogate the new master nodeid to locks
* The NEW_MASTER flag tells dlm_recover_locks() which rsb's to consider.
- * The NEW_MASTER2 flag tells recover_lvb() which rsb's to consider.
+ * The NEW_MASTER2 flag tells recover_lvb() and set_locks_purged() which
+ * rsb's to consider.
*/
static void set_new_master(struct dlm_rsb *r, int nodeid)
r = recover_list_find(ls, rc->rc_id);
if (!r) {
- log_error(ls, "dlm_recover_master_reply no id %"PRIx64"",
- rc->rc_id);
+ log_error(ls, "dlm_recover_master_reply no id %llx",
+ (unsigned long long)rc->rc_id);
goto out;
}
return error;
}
-static int all_queues_empty(struct dlm_rsb *r)
-{
- if (!list_empty(&r->res_grantqueue) ||
- !list_empty(&r->res_convertqueue) ||
- !list_empty(&r->res_waitqueue))
- return FALSE;
- return TRUE;
-}
-
static int recover_locks(struct dlm_rsb *r)
{
int error = 0;
lock_rsb(r);
- if (all_queues_empty(r))
- goto out;
- DLM_ASSERT(!r->res_recover_locks_count, dlm_print_rsb(r););
+ DLM_ASSERT(!r->res_recover_locks_count, dlm_dump_rsb(r););
error = recover_locks_queue(r, &r->res_grantqueue);
if (error)
void dlm_recovered_lock(struct dlm_rsb *r)
{
- DLM_ASSERT(rsb_flag(r, RSB_NEW_MASTER), dlm_print_rsb(r););
+ DLM_ASSERT(rsb_flag(r, RSB_NEW_MASTER), dlm_dump_rsb(r););
r->res_recover_locks_count--;
if (!r->res_recover_locks_count) {
{
struct dlm_lkb *lkb, *high_lkb = NULL;
uint32_t high_seq = 0;
- int lock_lvb_exists = FALSE;
- int big_lock_exists = FALSE;
+ int lock_lvb_exists = 0;
+ int big_lock_exists = 0;
int lvblen = r->res_ls->ls_lvblen;
list_for_each_entry(lkb, &r->res_grantqueue, lkb_statequeue) {
if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
continue;
- lock_lvb_exists = TRUE;
+ lock_lvb_exists = 1;
if (lkb->lkb_grmode > DLM_LOCK_CR) {
- big_lock_exists = TRUE;
+ big_lock_exists = 1;
goto setflag;
}
if (!(lkb->lkb_exflags & DLM_LKF_VALBLK))
continue;
- lock_lvb_exists = TRUE;
+ lock_lvb_exists = 1;
if (lkb->lkb_grmode > DLM_LOCK_CR) {
- big_lock_exists = TRUE;
+ big_lock_exists = 1;
goto setflag;
}
}
}
+/* We've become the new master for this rsb and waiting/converting locks may
+ need to be granted in dlm_grant_after_purge() due to locks that may have
+ existed from a removed node. */
+
+static void set_locks_purged(struct dlm_rsb *r)
+{
+ if (!list_empty(&r->res_waitqueue) || !list_empty(&r->res_convertqueue))
+ rsb_set_flag(r, RSB_LOCKS_PURGED);
+}
+
void dlm_recover_rsbs(struct dlm_ls *ls)
{
struct dlm_rsb *r;
if (is_master(r)) {
if (rsb_flag(r, RSB_RECOVER_CONVERT))
recover_conversion(r);
+ if (rsb_flag(r, RSB_NEW_MASTER2))
+ set_locks_purged(r);
recover_lvb(r);
count++;
}
rsb_clear_flag(r, RSB_RECOVER_CONVERT);
+ rsb_clear_flag(r, RSB_NEW_MASTER2);
unlock_rsb(r);
}
up_read(&ls->ls_root_sem);