1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) 2005 Red Hat, Inc. All rights reserved.
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.
10 *******************************************************************************
11 ******************************************************************************/
13 #include "dlm_internal.h"
14 #include "lockspace.h"
22 * Following called by dlm_recoverd thread
25 static void add_ordered_member(struct dlm_ls *ls, struct dlm_member *new)
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;
32 list_for_each(tmp, head) {
33 memb = list_entry(tmp, struct dlm_member, list);
34 if (new->nodeid < memb->nodeid)
39 list_add_tail(newlist, head);
41 /* FIXME: can use list macro here */
42 newlist->prev = tmp->prev;
44 tmp->prev->next = newlist;
49 static int dlm_add_member(struct dlm_ls *ls, int nodeid)
51 struct dlm_member *memb;
54 memb = kzalloc(sizeof(struct dlm_member), GFP_KERNEL);
58 w = dlm_node_weight(ls->ls_name, nodeid);
62 memb->nodeid = nodeid;
64 add_ordered_member(ls, memb);
69 static void dlm_remove_member(struct dlm_ls *ls, struct dlm_member *memb)
71 list_move(&memb->list, &ls->ls_nodes_gone);
75 static int dlm_is_member(struct dlm_ls *ls, int nodeid)
77 struct dlm_member *memb;
79 list_for_each_entry(memb, &ls->ls_nodes, list) {
80 if (memb->nodeid == nodeid)
86 int dlm_is_removed(struct dlm_ls *ls, int nodeid)
88 struct dlm_member *memb;
90 list_for_each_entry(memb, &ls->ls_nodes_gone, list) {
91 if (memb->nodeid == nodeid)
97 static void clear_memb_list(struct list_head *head)
99 struct dlm_member *memb;
101 while (!list_empty(head)) {
102 memb = list_entry(head->next, struct dlm_member, list);
103 list_del(&memb->list);
108 void dlm_clear_members(struct dlm_ls *ls)
110 clear_memb_list(&ls->ls_nodes);
111 ls->ls_num_nodes = 0;
114 void dlm_clear_members_gone(struct dlm_ls *ls)
116 clear_memb_list(&ls->ls_nodes_gone);
119 static void make_member_array(struct dlm_ls *ls)
121 struct dlm_member *memb;
122 int i, w, x = 0, total = 0, all_zero = 0, *array;
124 kfree(ls->ls_node_array);
125 ls->ls_node_array = NULL;
127 list_for_each_entry(memb, &ls->ls_nodes, list) {
129 total += memb->weight;
132 /* all nodes revert to weight of 1 if all have weight 0 */
135 total = ls->ls_num_nodes;
139 ls->ls_total_weight = total;
141 array = kmalloc(sizeof(int) * total, GFP_KERNEL);
145 list_for_each_entry(memb, &ls->ls_nodes, list) {
146 if (!all_zero && !memb->weight)
154 DLM_ASSERT(x < total, printk("total %d x %d\n", total, x););
156 for (i = 0; i < w; i++)
157 array[x++] = memb->nodeid;
160 ls->ls_node_array = array;
163 /* send a status request to all members just to establish comms connections */
165 static void ping_members(struct dlm_ls *ls)
167 struct dlm_member *memb;
168 list_for_each_entry(memb, &ls->ls_nodes, list)
169 dlm_rcom_status(ls, memb->nodeid);
172 int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out)
174 struct dlm_member *memb, *safe;
175 int i, error, found, pos = 0, neg = 0, low = -1;
177 /* move departed members from ls_nodes to ls_nodes_gone */
179 list_for_each_entry_safe(memb, safe, &ls->ls_nodes, list) {
181 for (i = 0; i < rv->node_count; i++) {
182 if (memb->nodeid == rv->nodeids[i]) {
190 dlm_remove_member(ls, memb);
191 log_debug(ls, "remove member %d", memb->nodeid);
195 /* add new members to ls_nodes */
197 for (i = 0; i < rv->node_count; i++) {
198 if (dlm_is_member(ls, rv->nodeids[i]))
200 dlm_add_member(ls, rv->nodeids[i]);
202 log_debug(ls, "add member %d", rv->nodeids[i]);
205 list_for_each_entry(memb, &ls->ls_nodes, list) {
206 if (low == -1 || memb->nodeid < low)
209 ls->ls_low_nodeid = low;
211 make_member_array(ls);
212 dlm_set_recover_status(ls, DLM_RS_NODES);
217 error = dlm_recover_members_wait(ls);
218 log_debug(ls, "total members %d", ls->ls_num_nodes);
223 * Following called from lockspace.c
226 int dlm_ls_stop(struct dlm_ls *ls)
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()).
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);
243 * This in_recovery lock does two things:
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.
252 down_write(&ls->ls_in_recovery);
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.
261 dlm_recoverd_suspend(ls);
262 ls->ls_recover_status = 0;
263 dlm_recoverd_resume(ls);
267 int dlm_ls_start(struct dlm_ls *ls)
269 struct dlm_recover *rv = NULL, *rv_old;
273 rv = kzalloc(sizeof(struct dlm_recover), GFP_KERNEL);
277 error = count = dlm_nodeid_list(ls->ls_name, &ids);
281 spin_lock(&ls->ls_recover_lock);
283 /* the lockspace needs to be stopped before it can be started */
285 if (!dlm_locking_stopped(ls)) {
286 spin_unlock(&ls->ls_recover_lock);
287 log_error(ls, "start ignored: lockspace running");
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);
300 kfree(rv_old->nodeids);
304 dlm_recoverd_kick(ls);