fs/dlm/recover.c

42 int dlm_wait_function(struct dlm_ls *ls, int (*testfn) (struct dlm_ls *ls))
48 		rv = wait_event_timeout(ls->ls_wait_general,
49 					testfn(ls) || dlm_recovery_stopped(ls),
53 		if (test_bit(LSFL_RCOM_WAIT, &ls->ls_flags)) {
54 			log_debug(ls, "dlm_wait_function timed out");
59 	if (dlm_recovery_stopped(ls)) {
60 		log_debug(ls, "dlm_wait_function aborted");
74 uint32_t dlm_recover_status(struct dlm_ls *ls)
77 	spin_lock_bh(&ls->ls_recover_lock);
78 	status = ls->ls_recover_status;
79 	spin_unlock_bh(&ls->ls_recover_lock);
83 static void _set_recover_status(struct dlm_ls *ls, uint32_t status)
85 	ls->ls_recover_status |= status;
88 void dlm_set_recover_status(struct dlm_ls *ls, uint32_t status)
90 	spin_lock_bh(&ls->ls_recover_lock);
91 	_set_recover_status(ls, status);
92 	spin_unlock_bh(&ls->ls_recover_lock);
95 static int wait_status_all(struct dlm_ls *ls, uint32_t wait_status,
98 	struct dlm_rcom *rc = ls->ls_recover_buf;
102 	list_for_each_entry(memb, &ls->ls_nodes, list) {
105 			if (dlm_recovery_stopped(ls)) {
110 			error = dlm_rcom_status(ls, memb->nodeid, 0, seq);
115 				dlm_slot_save(ls, rc, memb);
128 static int wait_status_low(struct dlm_ls *ls, uint32_t wait_status,
131 	struct dlm_rcom *rc = ls->ls_recover_buf;
132 	int error = 0, delay = 0, nodeid = ls->ls_low_nodeid;
135 		if (dlm_recovery_stopped(ls)) {
140 		error = dlm_rcom_status(ls, nodeid, status_flags, seq);
154 static int wait_status(struct dlm_ls *ls, uint32_t status, uint64_t seq)
159 	if (ls->ls_low_nodeid == dlm_our_nodeid()) {
160 		error = wait_status_all(ls, status, 0, seq);
162 			dlm_set_recover_status(ls, status_all);
164 		error = wait_status_low(ls, status_all, 0, seq);
169 int dlm_recover_members_wait(struct dlm_ls *ls, uint64_t seq)
177 	list_for_each_entry(memb, &ls->ls_nodes, list) {
182 	if (ls->ls_low_nodeid == dlm_our_nodeid()) {
183 		error = wait_status_all(ls, DLM_RS_NODES, 1, seq);
189 		rv = dlm_slots_assign(ls, &num_slots, &slots_size, &slots, &gen);
191 			spin_lock_bh(&ls->ls_recover_lock);
192 			_set_recover_status(ls, DLM_RS_NODES_ALL);
193 			ls->ls_num_slots = num_slots;
194 			ls->ls_slots_size = slots_size;
195 			ls->ls_slots = slots;
196 			ls->ls_generation = gen;
197 			spin_unlock_bh(&ls->ls_recover_lock);
199 			dlm_set_recover_status(ls, DLM_RS_NODES_ALL);
202 		error = wait_status_low(ls, DLM_RS_NODES_ALL,
207 		dlm_slots_copy_in(ls);
213 int dlm_recover_directory_wait(struct dlm_ls *ls, uint64_t seq)
215 	return wait_status(ls, DLM_RS_DIR, seq);
218 int dlm_recover_locks_wait(struct dlm_ls *ls, uint64_t seq)
220 	return wait_status(ls, DLM_RS_LOCKS, seq);
223 int dlm_recover_done_wait(struct dlm_ls *ls, uint64_t seq)
225 	return wait_status(ls, DLM_RS_DONE, seq);
240 static int recover_list_empty(struct dlm_ls *ls)
244 	spin_lock_bh(&ls->ls_recover_list_lock);
245 	empty = list_empty(&ls->ls_recover_list);
246 	spin_unlock_bh(&ls->ls_recover_list_lock);
253 	struct dlm_ls *ls = r->res_ls;
255 	spin_lock_bh(&ls->ls_recover_list_lock);
257 		list_add_tail(&r->res_recover_list, &ls->ls_recover_list);
258 		ls->ls_recover_list_count++;
261 	spin_unlock_bh(&ls->ls_recover_list_lock);
266 	struct dlm_ls *ls = r->res_ls;
268 	spin_lock_bh(&ls->ls_recover_list_lock);
270 	ls->ls_recover_list_count--;
271 	spin_unlock_bh(&ls->ls_recover_list_lock);
276 static void recover_list_clear(struct dlm_ls *ls)
280 	spin_lock_bh(&ls->ls_recover_list_lock);
281 	list_for_each_entry_safe(r, s, &ls->ls_recover_list, res_recover_list) {
285 		ls->ls_recover_list_count--;
288 	if (ls->ls_recover_list_count != 0) {
289 		log_error(ls, "warning: recover_list_count %d",
290 			  ls->ls_recover_list_count);
291 		ls->ls_recover_list_count = 0;
293 	spin_unlock_bh(&ls->ls_recover_list_lock);
296 static int recover_idr_empty(struct dlm_ls *ls)
300 	spin_lock_bh(&ls->ls_recover_idr_lock);
301 	if (ls->ls_recover_list_count)
303 	spin_unlock_bh(&ls->ls_recover_idr_lock);
310 	struct dlm_ls *ls = r->res_ls;
313 	spin_lock_bh(&ls->ls_recover_idr_lock);
318 	rv = idr_alloc(&ls->ls_recover_idr, r, 1, 0, GFP_NOWAIT);
323 	ls->ls_recover_list_count++;
327 	spin_unlock_bh(&ls->ls_recover_idr_lock);
333 	struct dlm_ls *ls = r->res_ls;
335 	spin_lock_bh(&ls->ls_recover_idr_lock);
336 	idr_remove(&ls->ls_recover_idr, r->res_id);
338 	ls->ls_recover_list_count--;
339 	spin_unlock_bh(&ls->ls_recover_idr_lock);
344 static struct dlm_rsb *recover_idr_find(struct dlm_ls *ls, uint64_t id)
348 	spin_lock_bh(&ls->ls_recover_idr_lock);
349 	r = idr_find(&ls->ls_recover_idr, (int)id);
350 	spin_unlock_bh(&ls->ls_recover_idr_lock);
354 static void recover_idr_clear(struct dlm_ls *ls)
359 	spin_lock_bh(&ls->ls_recover_idr_lock);
361 	idr_for_each_entry(&ls->ls_recover_idr, r, id) {
362 		idr_remove(&ls->ls_recover_idr, id);
365 		ls->ls_recover_list_count--;
370 	if (ls->ls_recover_list_count != 0) {
371 		log_error(ls, "warning: recover_list_count %d",
372 			  ls->ls_recover_list_count);
373 		ls->ls_recover_list_count = 0;
375 	spin_unlock_bh(&ls->ls_recover_idr_lock);
445 	struct dlm_ls *ls = r->res_ls;
453 	is_removed = dlm_is_removed(ls, r->res_nodeid);
522 int dlm_recover_masters(struct dlm_ls *ls, uint64_t seq,
528 	int nodir = dlm_no_directory(ls);
531 	log_rinfo(ls, "dlm_recover_masters");
534 		if (dlm_recovery_stopped(ls)) {
552 	log_rinfo(ls, "dlm_recover_masters %u of %u", count, total);
554 	error = dlm_wait_function(ls, &recover_idr_empty);
557 		recover_idr_clear(ls);
561 int dlm_recover_master_reply(struct dlm_ls *ls, const struct dlm_rcom *rc)
566 	r = recover_idr_find(ls, le64_to_cpu(rc->rc_id));
568 		log_error(ls, "dlm_recover_master_reply no id %llx",
587 	if (recover_idr_empty(ls))
588 		wake_up(&ls->ls_wait_general);
655 int dlm_recover_locks(struct dlm_ls *ls, uint64_t seq,
670 		if (dlm_recovery_stopped(ls)) {
682 	log_rinfo(ls, "dlm_recover_locks %d out", count);
684 	error = dlm_wait_function(ls, &recover_list_empty);
687 		recover_list_clear(ls);
812 	struct dlm_ls *ls = r->res_ls;
828 			log_debug(ls, "recover_conversion %x set gr to rq %d",
832 			log_debug(ls, "recover_conversion %x set gr %d",
849 void dlm_recover_rsbs(struct dlm_ls *ls, const struct list_head *root_list)
877 		log_rinfo(ls, "dlm_recover_rsbs %d done", count);
882 void dlm_clear_toss(struct dlm_ls *ls)
887 	write_lock_bh(&ls->ls_rsbtbl_lock);
888 	list_for_each_entry_safe(r, safe, &ls->ls_toss, res_rsbs_list) {
890 		rhashtable_remove_fast(&ls->ls_rsbtbl, &r->res_node,
900 	write_unlock_bh(&ls->ls_rsbtbl_lock);
903 		log_rinfo(ls, "dlm_clear_toss %u done", count);