Merge branch 'master' of git://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux-2.6
[linux-2.6] / fs / dlm / member.c
1 /******************************************************************************
2 *******************************************************************************
3 **
4 **  Copyright (C) 2005-2007 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                 kfree(memb);
61                 return w;
62         }
63
64         memb->nodeid = nodeid;
65         memb->weight = w;
66         add_ordered_member(ls, memb);
67         ls->ls_num_nodes++;
68         return 0;
69 }
70
71 static void dlm_remove_member(struct dlm_ls *ls, struct dlm_member *memb)
72 {
73         list_move(&memb->list, &ls->ls_nodes_gone);
74         ls->ls_num_nodes--;
75 }
76
77 static int dlm_is_member(struct dlm_ls *ls, int nodeid)
78 {
79         struct dlm_member *memb;
80
81         list_for_each_entry(memb, &ls->ls_nodes, list) {
82                 if (memb->nodeid == nodeid)
83                         return 1;
84         }
85         return 0;
86 }
87
88 int dlm_is_removed(struct dlm_ls *ls, int nodeid)
89 {
90         struct dlm_member *memb;
91
92         list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
93                 if (memb->nodeid == nodeid)
94                         return 1;
95         }
96         return 0;
97 }
98
99 static void clear_memb_list(struct list_head *head)
100 {
101         struct dlm_member *memb;
102
103         while (!list_empty(head)) {
104                 memb = list_entry(head->next, struct dlm_member, list);
105                 list_del(&memb->list);
106                 kfree(memb);
107         }
108 }
109
110 void dlm_clear_members(struct dlm_ls *ls)
111 {
112         clear_memb_list(&ls->ls_nodes);
113         ls->ls_num_nodes = 0;
114 }
115
116 void dlm_clear_members_gone(struct dlm_ls *ls)
117 {
118         clear_memb_list(&ls->ls_nodes_gone);
119 }
120
121 static void make_member_array(struct dlm_ls *ls)
122 {
123         struct dlm_member *memb;
124         int i, w, x = 0, total = 0, all_zero = 0, *array;
125
126         kfree(ls->ls_node_array);
127         ls->ls_node_array = NULL;
128
129         list_for_each_entry(memb, &ls->ls_nodes, list) {
130                 if (memb->weight)
131                         total += memb->weight;
132         }
133
134         /* all nodes revert to weight of 1 if all have weight 0 */
135
136         if (!total) {
137                 total = ls->ls_num_nodes;
138                 all_zero = 1;
139         }
140
141         ls->ls_total_weight = total;
142
143         array = kmalloc(sizeof(int) * total, GFP_KERNEL);
144         if (!array)
145                 return;
146
147         list_for_each_entry(memb, &ls->ls_nodes, list) {
148                 if (!all_zero && !memb->weight)
149                         continue;
150
151                 if (all_zero)
152                         w = 1;
153                 else
154                         w = memb->weight;
155
156                 DLM_ASSERT(x < total, printk("total %d x %d\n", total, x););
157
158                 for (i = 0; i < w; i++)
159                         array[x++] = memb->nodeid;
160         }
161
162         ls->ls_node_array = array;
163 }
164
165 /* send a status request to all members just to establish comms connections */
166
167 static int ping_members(struct dlm_ls *ls)
168 {
169         struct dlm_member *memb;
170         int error = 0;
171
172         list_for_each_entry(memb, &ls->ls_nodes, list) {
173                 error = dlm_recovery_stopped(ls);
174                 if (error)
175                         break;
176                 error = dlm_rcom_status(ls, memb->nodeid);
177                 if (error)
178                         break;
179         }
180         if (error)
181                 log_debug(ls, "ping_members aborted %d last nodeid %d",
182                           error, ls->ls_recover_nodeid);
183         return error;
184 }
185
186 int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
187 {
188         struct dlm_member *memb, *safe;
189         int i, error, found, pos = 0, neg = 0, low = -1;
190
191         /* previously removed members that we've not finished removing need to
192            count as a negative change so the "neg" recovery steps will happen */
193
194         list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
195                 log_debug(ls, "prev removed member %d", memb->nodeid);
196                 neg++;
197         }
198
199         /* move departed members from ls_nodes to ls_nodes_gone */
200
201         list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) {
202                 found = 0;
203                 for (i = 0; i < rv->node_count; i++) {
204                         if (memb->nodeid == rv->nodeids[i]) {
205                                 found = 1;
206                                 break;
207                         }
208                 }
209
210                 if (!found) {
211                         neg++;
212                         dlm_remove_member(ls, memb);
213                         log_debug(ls, "remove member %d", memb->nodeid);
214                 }
215         }
216
217         /* add new members to ls_nodes */
218
219         for (i = 0; i < rv->node_count; i++) {
220                 if (dlm_is_member(ls, rv->nodeids[i]))
221                         continue;
222                 dlm_add_member(ls, rv->nodeids[i]);
223                 pos++;
224                 log_debug(ls, "add member %d", rv->nodeids[i]);
225         }
226
227         list_for_each_entry(memb, &ls->ls_nodes, list) {
228                 if (low == -1 || memb->nodeid < low)
229                         low = memb->nodeid;
230         }
231         ls->ls_low_nodeid = low;
232
233         make_member_array(ls);
234         dlm_set_recover_status(ls, DLM_RS_NODES);
235         *neg_out = neg;
236
237         error = ping_members(ls);
238         if (!error || error == -EPROTO) {
239                 /* new_lockspace() may be waiting to know if the config
240                    is good or bad */
241                 ls->ls_members_result = error;
242                 complete(&ls->ls_members_done);
243         }
244         if (error)
245                 goto out;
246
247         error = dlm_recover_members_wait(ls);
248  out:
249         log_debug(ls, "total members %d error %d", ls->ls_num_nodes, error);
250         return error;
251 }
252
253 /*
254  * Following called from lockspace.c
255  */
256
257 int dlm_ls_stop(struct dlm_ls *ls)
258 {
259         int new;
260
261         /*
262          * A stop cancels any recovery that's in progress (see RECOVERY_STOP,
263          * dlm_recovery_stopped()) and prevents any new locks from being
264          * processed (see RUNNING, dlm_locking_stopped()).
265          */
266
267         spin_lock(&ls->ls_recover_lock);
268         set_bit(LSFL_RECOVERY_STOP, &ls->ls_flags);
269         new = test_and_clear_bit(LSFL_RUNNING, &ls->ls_flags);
270         ls->ls_recover_seq++;
271         spin_unlock(&ls->ls_recover_lock);
272
273         /*
274          * This in_recovery lock does two things:
275          *
276          * 1) Keeps this function from returning until all threads are out
277          *    of locking routines and locking is truely stopped.
278          * 2) Keeps any new requests from being processed until it's unlocked
279          *    when recovery is complete.
280          */
281
282         if (new)
283                 down_write(&ls->ls_in_recovery);
284
285         /*
286          * The recoverd suspend/resume makes sure that dlm_recoverd (if
287          * running) has noticed the clearing of RUNNING above and quit
288          * processing the previous recovery.  This will be true for all nodes
289          * before any nodes start the new recovery.
290          */
291
292         dlm_recoverd_suspend(ls);
293         ls->ls_recover_status = 0;
294         dlm_recoverd_resume(ls);
295
296         if (!ls->ls_recover_begin)
297                 ls->ls_recover_begin = jiffies;
298         return 0;
299 }
300
301 int dlm_ls_start(struct dlm_ls *ls)
302 {
303         struct dlm_recover *rv = NULL, *rv_old;
304         int *ids = NULL;
305         int error, count;
306
307         rv = kzalloc(sizeof(struct dlm_recover), GFP_KERNEL);
308         if (!rv)
309                 return -ENOMEM;
310
311         error = count = dlm_nodeid_list(ls->ls_name, &ids);
312         if (error <= 0)
313                 goto fail;
314
315         spin_lock(&ls->ls_recover_lock);
316
317         /* the lockspace needs to be stopped before it can be started */
318
319         if (!dlm_locking_stopped(ls)) {
320                 spin_unlock(&ls->ls_recover_lock);
321                 log_error(ls, "start ignored: lockspace running");
322                 error = -EINVAL;
323                 goto fail;
324         }
325
326         rv->nodeids = ids;
327         rv->node_count = count;
328         rv->seq = ++ls->ls_recover_seq;
329         rv_old = ls->ls_recover_args;
330         ls->ls_recover_args = rv;
331         spin_unlock(&ls->ls_recover_lock);
332
333         if (rv_old) {
334                 kfree(rv_old->nodeids);
335                 kfree(rv_old);
336         }
337
338         dlm_recoverd_kick(ls);
339         return 0;
340
341  fail:
342         kfree(rv);
343         kfree(ids);
344         return error;
345 }
346