SAFE public projects git trees. - safe/jmp/linux-2.6/blob - fs/dlm/member.c

   1 /******************************************************************************
   2 *******************************************************************************
   3 **
   4 **  Copyright (C) 2005 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", error);
 180         return error;
 181 }
 182
 183 int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
 184 {
 185         struct dlm_member *memb, *safe;
 186         int i, error, found, pos = 0, neg = 0, low = -1;
 187
 188         /* move departed members from ls_nodes to ls_nodes_gone */
 189
 190         list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) {
 191                 found = 0;
 192                 for (i = 0; i < rv->node_count; i++) {
 193                         if (memb->nodeid == rv->nodeids[i]) {
 194                                 found = 1;
 195                                 break;
 196                         }
 197                 }
 198
 199                 if (!found) {
 200                         neg++;
 201                         dlm_remove_member(ls, memb);
 202                         log_debug(ls, "remove member %d", memb->nodeid);
 203                 }
 204         }
 205
 206         /* add new members to ls_nodes */
 207
 208         for (i = 0; i < rv->node_count; i++) {
 209                 if (dlm_is_member(ls, rv->nodeids[i]))
 210                         continue;
 211                 dlm_add_member(ls, rv->nodeids[i]);
 212                 pos++;
 213                 log_debug(ls, "add member %d", rv->nodeids[i]);
 214         }
 215
 216         list_for_each_entry(memb, &ls->ls_nodes, list) {
 217                 if (low == -1 || memb->nodeid < low)
 218                         low = memb->nodeid;
 219         }
 220         ls->ls_low_nodeid = low;
 221
 222         make_member_array(ls);
 223         dlm_set_recover_status(ls, DLM_RS_NODES);
 224         *neg_out = neg;
 225
 226         error = ping_members(ls);
 227         if (error)
 228                 goto out;
 229
 230         error = dlm_recover_members_wait(ls);
 231  out:
 232         log_debug(ls, "total members %d error %d", ls->ls_num_nodes, error);
 233         return error;
 234 }
 235
 236 /*
 237  * Following called from lockspace.c
 238  */
 239
 240 int dlm_ls_stop(struct dlm_ls *ls)
 241 {
 242         int new;
 243
 244         /*
 245          * A stop cancels any recovery that's in progress (see RECOVERY_STOP,
 246          * dlm_recovery_stopped()) and prevents any new locks from being
 247          * processed (see RUNNING, dlm_locking_stopped()).
 248          */
 249
 250         spin_lock(&ls->ls_recover_lock);
 251         set_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
 252         new = test_and_clear_bit(LSFL_RUNNING, &ls->ls_flags);
 253         ls->ls_recover_seq++;
 254         spin_unlock(&ls->ls_recover_lock);
 255
 256         /*
 257          * This in_recovery lock does two things:
 258          *
 259          * 1) Keeps this function from returning until all threads are out
 260          *    of locking routines and locking is truely stopped.
 261          * 2) Keeps any new requests from being processed until it's unlocked
 262          *    when recovery is complete.
 263          */
 264
 265         if (new)
 266                 down_write(&ls->ls_in_recovery);
 267
 268         /*
 269          * The recoverd suspend/resume makes sure that dlm_recoverd (if
 270          * running) has noticed the clearing of RUNNING above and quit
 271          * processing the previous recovery.  This will be true for all nodes
 272          * before any nodes start the new recovery.
 273          */
 274
 275         dlm_recoverd_suspend(ls);
 276         ls->ls_recover_status = 0;
 277         dlm_recoverd_resume(ls);
 278         return 0;
 279 }
 280
 281 int dlm_ls_start(struct dlm_ls *ls)
 282 {
 283         struct dlm_recover *rv = NULL, *rv_old;
 284         int *ids = NULL;
 285         int error, count;
 286
 287         rv = kzalloc(sizeof(struct dlm_recover), GFP_KERNEL);
 288         if (!rv)
 289                 return -ENOMEM;
 290
 291         error = count = dlm_nodeid_list(ls->ls_name, &ids);
 292         if (error <= 0)
 293                 goto fail;
 294
 295         spin_lock(&ls->ls_recover_lock);
 296
 297         /* the lockspace needs to be stopped before it can be started */
 298
 299         if (!dlm_locking_stopped(ls)) {
 300                 spin_unlock(&ls->ls_recover_lock);
 301                 log_error(ls, "start ignored: lockspace running");
 302                 error = -EINVAL;
 303                 goto fail;
 304         }
 305
 306         rv->nodeids = ids;
 307         rv->node_count = count;
 308         rv->seq = ++ls->ls_recover_seq;
 309         rv_old = ls->ls_recover_args;
 310         ls->ls_recover_args = rv;
 311         spin_unlock(&ls->ls_recover_lock);
 312
 313         if (rv_old) {
 314                 kfree(rv_old->nodeids);
 315                 kfree(rv_old);
 316         }
 317
 318         dlm_recoverd_kick(ls);
 319         return 0;
 320
 321  fail:
 322         kfree(rv);
 323         kfree(ids);
 324         return error;
 325 }
 326