dlm: remove shared message stub for recovery
[linux-3.10.git] / fs / dlm / member.c
index ebb33c3..b12532e 100644 (file)
@@ -1,7 +1,7 @@
 /******************************************************************************
 *******************************************************************************
 **
-**  Copyright (C) 2005 Red Hat, Inc.  All rights reserved.
+**  Copyright (C) 2005-2009 Red Hat, Inc.  All rights reserved.
 **
 **  This copyrighted material is made available to anyone wishing to use,
 **  modify, copy, or redistribute it subject to the terms and conditions
 #include "recover.h"
 #include "rcom.h"
 #include "config.h"
-
-/*
- * Following called by dlm_recoverd thread
- */
+#include "lowcomms.h"
 
 static void add_ordered_member(struct dlm_ls *ls, struct dlm_member *new)
 {
@@ -49,15 +46,23 @@ static void add_ordered_member(struct dlm_ls *ls, struct dlm_member *new)
 static int dlm_add_member(struct dlm_ls *ls, int nodeid)
 {
        struct dlm_member *memb;
-       int w;
+       int w, error;
 
-       memb = kzalloc(sizeof(struct dlm_member), GFP_KERNEL);
+       memb = kzalloc(sizeof(struct dlm_member), GFP_NOFS);
        if (!memb)
                return -ENOMEM;
 
        w = dlm_node_weight(ls->ls_name, nodeid);
-       if (w < 0)
+       if (w < 0) {
+               kfree(memb);
                return w;
+       }
+
+       error = dlm_lowcomms_connect_node(nodeid);
+       if (error < 0) {
+               kfree(memb);
+               return error;
+       }
 
        memb->nodeid = nodeid;
        memb->weight = w;
@@ -72,7 +77,7 @@ static void dlm_remove_member(struct dlm_ls *ls, struct dlm_member *memb)
        ls->ls_num_nodes--;
 }
 
-static int dlm_is_member(struct dlm_ls *ls, int nodeid)
+int dlm_is_member(struct dlm_ls *ls, int nodeid)
 {
        struct dlm_member *memb;
 
@@ -138,7 +143,7 @@ static void make_member_array(struct dlm_ls *ls)
 
        ls->ls_total_weight = total;
 
-       array = kmalloc(sizeof(int) * total, GFP_KERNEL);
+       array = kmalloc(sizeof(int) * total, GFP_NOFS);
        if (!array)
                return;
 
@@ -176,7 +181,8 @@ static int ping_members(struct dlm_ls *ls)
                        break;
        }
        if (error)
-               log_debug(ls, "ping_members aborted %d", error);
+               log_debug(ls, "ping_members aborted %d last nodeid %d",
+                         error, ls->ls_recover_nodeid);
        return error;
 }
 
@@ -185,6 +191,14 @@ int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
        struct dlm_member *memb, *safe;
        int i, error, found, pos = 0, neg = 0, low = -1;
 
+       /* previously removed members that we've not finished removing need to
+          count as a negative change so the "neg" recovery steps will happen */
+
+       list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
+               log_debug(ls, "prev removed member %d", memb->nodeid);
+               neg++;
+       }
+
        /* move departed members from ls_nodes to ls_nodes_gone */
 
        list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) {
@@ -203,6 +217,23 @@ int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
                }
        }
 
+       /* Add an entry to ls_nodes_gone for members that were removed and
+          then added again, so that previous state for these nodes will be
+          cleared during recovery. */
+
+       for (i = 0; i < rv->new_count; i++) {
+               if (!dlm_is_member(ls, rv->new[i]))
+                       continue;
+               log_debug(ls, "new nodeid %d is a re-added member", rv->new[i]);
+
+               memb = kzalloc(sizeof(struct dlm_member), GFP_NOFS);
+               if (!memb)
+                       return -ENOMEM;
+               memb->nodeid = rv->new[i];
+               list_add_tail(&memb->list, &ls->ls_nodes_gone);
+               neg++;
+       }
+
        /* add new members to ls_nodes */
 
        for (i = 0; i < rv->node_count; i++) {
@@ -224,6 +255,12 @@ int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
        *neg_out = neg;
 
        error = ping_members(ls);
+       if (!error || error == -EPROTO) {
+               /* new_lockspace() may be waiting to know if the config
+                  is good or bad */
+               ls->ls_members_result = error;
+               complete(&ls->ls_members_done);
+       }
        if (error)
                goto out;
 
@@ -233,18 +270,30 @@ int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
        return error;
 }
 
-/*
- * Following called from lockspace.c
- */
+/* Userspace guarantees that dlm_ls_stop() has completed on all nodes before
+   dlm_ls_start() is called on any of them to start the new recovery. */
 
 int dlm_ls_stop(struct dlm_ls *ls)
 {
        int new;
 
        /*
-        * A stop cancels any recovery that's in progress (see RECOVERY_STOP,
-        * dlm_recovery_stopped()) and prevents any new locks from being
-        * processed (see RUNNING, dlm_locking_stopped()).
+        * Prevent dlm_recv from being in the middle of something when we do
+        * the stop.  This includes ensuring dlm_recv isn't processing a
+        * recovery message (rcom), while dlm_recoverd is aborting and
+        * resetting things from an in-progress recovery.  i.e. we want
+        * dlm_recoverd to abort its recovery without worrying about dlm_recv
+        * processing an rcom at the same time.  Stopping dlm_recv also makes
+        * it easy for dlm_receive_message() to check locking stopped and add a
+        * message to the requestqueue without races.
+        */
+
+       down_write(&ls->ls_recv_active);
+
+       /*
+        * Abort any recovery that's in progress (see RECOVERY_STOP,
+        * dlm_recovery_stopped()) and tell any other threads running in the
+        * dlm to quit any processing (see RUNNING, dlm_locking_stopped()).
         */
 
        spin_lock(&ls->ls_recover_lock);
@@ -254,10 +303,16 @@ int dlm_ls_stop(struct dlm_ls *ls)
        spin_unlock(&ls->ls_recover_lock);
 
        /*
+        * Let dlm_recv run again, now any normal messages will be saved on the
+        * requestqueue for later.
+        */
+
+       up_write(&ls->ls_recv_active);
+
+       /*
         * This in_recovery lock does two things:
-        *
         * 1) Keeps this function from returning until all threads are out
-        *    of locking routines and locking is truely stopped.
+        *    of locking routines and locking is truly stopped.
         * 2) Keeps any new requests from being processed until it's unlocked
         *    when recovery is complete.
         */
@@ -267,29 +322,32 @@ int dlm_ls_stop(struct dlm_ls *ls)
 
        /*
         * The recoverd suspend/resume makes sure that dlm_recoverd (if
-        * running) has noticed the clearing of RUNNING above and quit
-        * processing the previous recovery.  This will be true for all nodes
-        * before any nodes start the new recovery.
+        * running) has noticed RECOVERY_STOP above and quit processing the
+        * previous recovery.
         */
 
        dlm_recoverd_suspend(ls);
        ls->ls_recover_status = 0;
        dlm_recoverd_resume(ls);
+
+       if (!ls->ls_recover_begin)
+               ls->ls_recover_begin = jiffies;
        return 0;
 }
 
 int dlm_ls_start(struct dlm_ls *ls)
 {
        struct dlm_recover *rv = NULL, *rv_old;
-       int *ids = NULL;
-       int error, count;
+       int *ids = NULL, *new = NULL;
+       int error, ids_count = 0, new_count = 0;
 
-       rv = kzalloc(sizeof(struct dlm_recover), GFP_KERNEL);
+       rv = kzalloc(sizeof(struct dlm_recover), GFP_NOFS);
        if (!rv)
                return -ENOMEM;
 
-       error = count = dlm_nodeid_list(ls->ls_name, &ids);
-       if (error <= 0)
+       error = dlm_nodeid_list(ls->ls_name, &ids, &ids_count,
+                               &new, &new_count);
+       if (error < 0)
                goto fail;
 
        spin_lock(&ls->ls_recover_lock);
@@ -304,14 +362,19 @@ int dlm_ls_start(struct dlm_ls *ls)
        }
 
        rv->nodeids = ids;
-       rv->node_count = count;
+       rv->node_count = ids_count;
+       rv->new = new;
+       rv->new_count = new_count;
        rv->seq = ++ls->ls_recover_seq;
        rv_old = ls->ls_recover_args;
        ls->ls_recover_args = rv;
        spin_unlock(&ls->ls_recover_lock);
 
        if (rv_old) {
+               log_error(ls, "unused recovery %llx %d",
+                         (unsigned long long)rv_old->seq, rv_old->node_count);
                kfree(rv_old->nodeids);
+               kfree(rv_old->new);
                kfree(rv_old);
        }
 
@@ -321,6 +384,7 @@ int dlm_ls_start(struct dlm_ls *ls)
  fail:
        kfree(rv);
        kfree(ids);
+       kfree(new);
        return error;
 }