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