[DLM] fix aborted recovery during node removal
[linux-3.10.git] / 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         /* 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)
237                 goto out;
238
239         error = dlm_recover_members_wait(ls);
240  out:
241         log_debug(ls, "total members %d error %d", ls->ls_num_nodes, error);
242         return error;
243 }
244
245 /*
246  * Following called from lockspace.c
247  */
248
249 int dlm_ls_stop(struct dlm_ls *ls)
250 {
251         int new;
252
253         /*
254          * A stop cancels any recovery that's in progress (see RECOVERY_STOP,
255          * dlm_recovery_stopped()) and prevents any new locks from being
256          * processed (see RUNNING, dlm_locking_stopped()).
257          */
258
259         spin_lock(&ls->ls_recover_lock);
260         set_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
261         new = test_and_clear_bit(LSFL_RUNNING, &ls->ls_flags);
262         ls->ls_recover_seq++;
263         spin_unlock(&ls->ls_recover_lock);
264
265         /*
266          * This in_recovery lock does two things:
267          *
268          * 1) Keeps this function from returning until all threads are out
269          *    of locking routines and locking is truely stopped.
270          * 2) Keeps any new requests from being processed until it's unlocked
271          *    when recovery is complete.
272          */
273
274         if (new)
275                 down_write(&ls->ls_in_recovery);
276
277         /*
278          * The recoverd suspend/resume makes sure that dlm_recoverd (if
279          * running) has noticed the clearing of RUNNING above and quit
280          * processing the previous recovery.  This will be true for all nodes
281          * before any nodes start the new recovery.
282          */
283
284         dlm_recoverd_suspend(ls);
285         ls->ls_recover_status = 0;
286         dlm_recoverd_resume(ls);
287         return 0;
288 }
289
290 int dlm_ls_start(struct dlm_ls *ls)
291 {
292         struct dlm_recover *rv = NULL, *rv_old;
293         int *ids = NULL;
294         int error, count;
295
296         rv = kzalloc(sizeof(struct dlm_recover), GFP_KERNEL);
297         if (!rv)
298                 return -ENOMEM;
299
300         error = count = dlm_nodeid_list(ls->ls_name, &ids);
301         if (error <= 0)
302                 goto fail;
303
304         spin_lock(&ls->ls_recover_lock);
305
306         /* the lockspace needs to be stopped before it can be started */
307
308         if (!dlm_locking_stopped(ls)) {
309                 spin_unlock(&ls->ls_recover_lock);
310                 log_error(ls, "start ignored: lockspace running");
311                 error = -EINVAL;
312                 goto fail;
313         }
314
315         rv->nodeids = ids;
316         rv->node_count = count;
317         rv->seq = ++ls->ls_recover_seq;
318         rv_old = ls->ls_recover_args;
319         ls->ls_recover_args = rv;
320         spin_unlock(&ls->ls_recover_lock);
321
322         if (rv_old) {
323                 kfree(rv_old->nodeids);
324                 kfree(rv_old);
325         }
326
327         dlm_recoverd_kick(ls);
328         return 0;
329
330  fail:
331         kfree(rv);
332         kfree(ids);
333         return error;
334 }
335