[DLM] abort recovery more quickly
[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", 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