git.oblomov.eu Git - linux-2.6/blob - fs/dlm/member.c

   1 /******************************************************************************
   2 *******************************************************************************
   3 **
   4 **  Copyright (C) 2005-2007 Red Hat, Inc.  All rights reserved.
   5 **
   6 **  This copyrighted material is made available to anyone wishing to use,
   7 **  modify, copy, or redistribute it subject to the terms and conditions
   8 **  of the GNU General Public License v.2.
   9 **
  10 *******************************************************************************
  11 ******************************************************************************/
  12
  13 #include "dlm_internal.h"
  14 #include "lockspace.h"
  15 #include "member.h"
  16 #include "recoverd.h"
  17 #include "recover.h"
  18 #include "rcom.h"
  19 #include "config.h"
  20
  21 /*
  22  * Following called by dlm_recoverd thread
  23  */
  24
  25 static void add_ordered_member(struct dlm_ls *ls, struct dlm_member *new)
  26 {
  27         struct dlm_member *memb = NULL;
  28         struct list_head *tmp;
  29         struct list_head *newlist = &new->list;
  30         struct list_head *head = &ls->ls_nodes;
  31
  32         list_for_each(tmp, head) {
  33                 memb = list_entry(tmp, struct dlm_member, list);
  34                 if (new->nodeid < memb->nodeid)
  35                         break;
  36         }
  37
  38         if (!memb)
  39                 list_add_tail(newlist, head);
  40         else {
  41                 /* FIXME: can use list macro here */
  42                 newlist->prev = tmp->prev;
  43                 newlist->next = tmp;
  44                 tmp->prev->next = newlist;
  45                 tmp->prev = newlist;
  46         }
  47 }
  48
  49 static int dlm_add_member(struct dlm_ls *ls, int nodeid)
  50 {
  51         struct dlm_member *memb;
  52         int w;
  53
  54         memb = kzalloc(sizeof(struct dlm_member), GFP_KERNEL);
  55         if (!memb)
  56                 return -ENOMEM;
  57
  58         w = dlm_node_weight(ls->ls_name, nodeid);
  59         if (w < 0)
  60                 return w;
  61
  62         memb->nodeid = nodeid;
  63         memb->weight = w;
  64         add_ordered_member(ls, memb);
  65         ls->ls_num_nodes++;
  66         return 0;
  67 }
  68
  69 static void dlm_remove_member(struct dlm_ls *ls, struct dlm_member *memb)
  70 {
  71         list_move(&memb->list, &ls->ls_nodes_gone);
  72         ls->ls_num_nodes--;
  73 }
  74
  75 static int dlm_is_member(struct dlm_ls *ls, int nodeid)
  76 {
  77         struct dlm_member *memb;
  78
  79         list_for_each_entry(memb, &ls->ls_nodes, list) {
  80                 if (memb->nodeid == nodeid)
  81                         return 1;
  82         }
  83         return 0;
  84 }
  85
  86 int dlm_is_removed(struct dlm_ls *ls, int nodeid)
  87 {
  88         struct dlm_member *memb;
  89
  90         list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
  91                 if (memb->nodeid == nodeid)
  92                         return 1;
  93         }
  94         return 0;
  95 }
  96
  97 static void clear_memb_list(struct list_head *head)
  98 {
  99         struct dlm_member *memb;
 100
 101         while (!list_empty(head)) {
 102                 memb = list_entry(head->next, struct dlm_member, list);
 103                 list_del(&memb->list);
 104                 kfree(memb);
 105         }
 106 }
 107
 108 void dlm_clear_members(struct dlm_ls *ls)
 109 {
 110         clear_memb_list(&ls->ls_nodes);
 111         ls->ls_num_nodes = 0;
 112 }
 113
 114 void dlm_clear_members_gone(struct dlm_ls *ls)
 115 {
 116         clear_memb_list(&ls->ls_nodes_gone);
 117 }
 118
 119 static void make_member_array(struct dlm_ls *ls)
 120 {
 121         struct dlm_member *memb;
 122         int i, w, x = 0, total = 0, all_zero = 0, *array;
 123
 124         kfree(ls->ls_node_array);
 125         ls->ls_node_array = NULL;
 126
 127         list_for_each_entry(memb, &ls->ls_nodes, list) {
 128                 if (memb->weight)
 129                         total += memb->weight;
 130         }
 131
 132         /* all nodes revert to weight of 1 if all have weight 0 */
 133
 134         if (!total) {
 135                 total = ls->ls_num_nodes;
 136                 all_zero = 1;
 137         }
 138
 139         ls->ls_total_weight = total;
 140
 141         array = kmalloc(sizeof(int) * total, GFP_KERNEL);
 142         if (!array)
 143                 return;
 144
 145         list_for_each_entry(memb, &ls->ls_nodes, list) {
 146                 if (!all_zero && !memb->weight)
 147                         continue;
 148
 149                 if (all_zero)
 150                         w = 1;
 151                 else
 152                         w = memb->weight;
 153
 154                 DLM_ASSERT(x < total, printk("total %d x %d\n", total, x););
 155
 156                 for (i = 0; i < w; i++)
 157                         array[x++] = memb->nodeid;
 158         }
 159
 160         ls->ls_node_array = array;
 161 }
 162
 163 /* send a status request to all members just to establish comms connections */
 164
 165 static int ping_members(struct dlm_ls *ls)
 166 {
 167         struct dlm_member *memb;
 168         int error = 0;
 169
 170         list_for_each_entry(memb, &ls->ls_nodes, list) {
 171                 error = dlm_recovery_stopped(ls);
 172                 if (error)
 173                         break;
 174                 error = dlm_rcom_status(ls, memb->nodeid);
 175                 if (error)
 176                         break;
 177         }
 178         if (error)
 179                 log_debug(ls, "ping_members aborted %d last nodeid %d",
 180                           error, ls->ls_recover_nodeid);
 181         return error;
 182 }
 183
 184 int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
 185 {
 186         struct dlm_member *memb, *safe;
 187         int i, error, found, pos = 0, neg = 0, low = -1;
 188
 189         /* previously removed members that we've not finished removing need to
 190            count as a negative change so the "neg" recovery steps will happen */
 191
 192         list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
 193                 log_debug(ls, "prev removed member %d", memb->nodeid);
 194                 neg++;
 195         }
 196
 197         /* move departed members from ls_nodes to ls_nodes_gone */
 198
 199         list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) {
 200                 found = 0;
 201                 for (i = 0; i < rv->node_count; i++) {
 202                         if (memb->nodeid == rv->nodeids[i]) {
 203                                 found = 1;
 204                                 break;
 205                         }
 206                 }
 207
 208                 if (!found) {
 209                         neg++;
 210                         dlm_remove_member(ls, memb);
 211                         log_debug(ls, "remove member %d", memb->nodeid);
 212                 }
 213         }
 214
 215         /* add new members to ls_nodes */
 216
 217         for (i = 0; i < rv->node_count; i++) {
 218                 if (dlm_is_member(ls, rv->nodeids[i]))
 219                         continue;
 220                 dlm_add_member(ls, rv->nodeids[i]);
 221                 pos++;
 222                 log_debug(ls, "add member %d", rv->nodeids[i]);
 223         }
 224
 225         list_for_each_entry(memb, &ls->ls_nodes, list) {
 226                 if (low == -1 || memb->nodeid < low)
 227                         low = memb->nodeid;
 228         }
 229         ls->ls_low_nodeid = low;
 230
 231         make_member_array(ls);
 232         dlm_set_recover_status(ls, DLM_RS_NODES);
 233         *neg_out = neg;
 234
 235         error = ping_members(ls);
 236         if (!error || error == -EPROTO) {
 237                 /* new_lockspace() may be waiting to know if the config
 238                    is good or bad */
 239                 ls->ls_members_result = error;
 240                 complete(&ls->ls_members_done);
 241         }
 242         if (error)
 243                 goto out;
 244
 245         error = dlm_recover_members_wait(ls);
 246  out:
 247         log_debug(ls, "total members %d error %d", ls->ls_num_nodes, error);
 248         return error;
 249 }
 250
 251 /*
 252  * Following called from lockspace.c
 253  */
 254
 255 int dlm_ls_stop(struct dlm_ls *ls)
 256 {
 257         int new;
 258
 259         /*
 260          * A stop cancels any recovery that's in progress (see RECOVERY_STOP,
 261          * dlm_recovery_stopped()) and prevents any new locks from being
 262          * processed (see RUNNING, dlm_locking_stopped()).
 263          */
 264
 265         spin_lock(&ls->ls_recover_lock);
 266         set_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
 267         new = test_and_clear_bit(LSFL_RUNNING, &ls->ls_flags);
 268         ls->ls_recover_seq++;
 269         spin_unlock(&ls->ls_recover_lock);
 270
 271         /*
 272          * This in_recovery lock does two things:
 273          *
 274          * 1) Keeps this function from returning until all threads are out
 275          *    of locking routines and locking is truely stopped.
 276          * 2) Keeps any new requests from being processed until it's unlocked
 277          *    when recovery is complete.
 278          */
 279
 280         if (new)
 281                 down_write(&ls->ls_in_recovery);
 282
 283         /*
 284          * The recoverd suspend/resume makes sure that dlm_recoverd (if
 285          * running) has noticed the clearing of RUNNING above and quit
 286          * processing the previous recovery.  This will be true for all nodes
 287          * before any nodes start the new recovery.
 288          */
 289
 290         dlm_recoverd_suspend(ls);
 291         ls->ls_recover_status = 0;
 292         dlm_recoverd_resume(ls);
 293
 294         if (!ls->ls_recover_begin)
 295                 ls->ls_recover_begin = jiffies;
 296         return 0;
 297 }
 298
 299 int dlm_ls_start(struct dlm_ls *ls)
 300 {
 301         struct dlm_recover *rv = NULL, *rv_old;
 302         int *ids = NULL;
 303         int error, count;
 304
 305         rv = kzalloc(sizeof(struct dlm_recover), GFP_KERNEL);
 306         if (!rv)
 307                 return -ENOMEM;
 308
 309         error = count = dlm_nodeid_list(ls->ls_name, &ids);
 310         if (error <= 0)
 311                 goto fail;
 312
 313         spin_lock(&ls->ls_recover_lock);
 314
 315         /* the lockspace needs to be stopped before it can be started */
 316
 317         if (!dlm_locking_stopped(ls)) {
 318                 spin_unlock(&ls->ls_recover_lock);
 319                 log_error(ls, "start ignored: lockspace running");
 320                 error = -EINVAL;
 321                 goto fail;
 322         }
 323
 324         rv->nodeids = ids;
 325         rv->node_count = count;
 326         rv->seq = ++ls->ls_recover_seq;
 327         rv_old = ls->ls_recover_args;
 328         ls->ls_recover_args = rv;
 329         spin_unlock(&ls->ls_recover_lock);
 330
 331         if (rv_old) {
 332                 kfree(rv_old->nodeids);
 333                 kfree(rv_old);
 334         }
 335
 336         dlm_recoverd_kick(ls);
 337         return 0;
 338
 339  fail:
 340         kfree(rv);
 341         kfree(ids);
 342         return error;
 343 }
 344