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