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