1 // SPDX-License-Identifier: GPL-2.0-only
5 **  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
6 **  Copyright (C) 2004-2005 Red Hat, Inc.  All rights reserved.
25  * We use the upper 16 bits of the hash value to select the directory node.
26  * Low bits are used for distribution of rsb's among hash buckets on each node.
28  * To give the exact range wanted (0 to num_nodes-1), we apply a modulus of
33 int dlm_hash2nodeid(struct dlm_ls *ls, uint32_t hash)  in dlm_hash2nodeid()  argument
37 	if (ls->ls_num_nodes == 1)  in dlm_hash2nodeid()
40 		node = (hash >> 16) % ls->ls_total_weight;  in dlm_hash2nodeid()
41 		return ls->ls_node_array[node];  in dlm_hash2nodeid()
47 	return r->res_dir_nodeid;  in dlm_dir_nodeid()
50 void dlm_recover_dir_nodeid(struct dlm_ls *ls, const struct list_head *root_list)  in dlm_recover_dir_nodeid()  argument
55 		r->res_dir_nodeid = dlm_hash2nodeid(ls, r->res_hash);  in dlm_recover_dir_nodeid()
59 int dlm_recover_directory(struct dlm_ls *ls, uint64_t seq)  in dlm_recover_directory()  argument
63 	int error = -ENOMEM, last_len, nodeid, result;  in dlm_recover_directory()
67 	log_rinfo(ls, "dlm_recover_directory");  in dlm_recover_directory()
69 	if (dlm_no_directory(ls))  in dlm_recover_directory()
76 	list_for_each_entry(memb, &ls->ls_nodes, list) {  in dlm_recover_directory()
77 		if (memb->nodeid == dlm_our_nodeid())  in dlm_recover_directory()
85 			if (dlm_recovery_stopped(ls)) {  in dlm_recover_directory()
86 				error = -EINTR;  in dlm_recover_directory()
90 			error = dlm_rcom_names(ls, memb->nodeid,  in dlm_recover_directory()
101 			b = ls->ls_recover_buf->rc_buf;  in dlm_recover_directory()
102 			left = le16_to_cpu(ls->ls_recover_buf->rc_header.h_length);  in dlm_recover_directory()
103 			left -= sizeof(struct dlm_rcom);  in dlm_recover_directory()
108 				error = -EINVAL;  in dlm_recover_directory()
115 				left -= sizeof(__be16);  in dlm_recover_directory()
132 				error = dlm_master_lookup(ls, memb->nodeid,  in dlm_recover_directory()
137 					log_error(ls, "recover_dir lookup %d",  in dlm_recover_directory()
144 				 * memb->nodeid which says it is the master.  in dlm_recover_directory()
148 				    nodeid != memb->nodeid) {  in dlm_recover_directory()
150 					log_error(ls, "recover_dir lookup %d "  in dlm_recover_directory()
152 						  result, nodeid, memb->nodeid,  in dlm_recover_directory()
160 				 * master nodeid matches memb->nodeid. */  in dlm_recover_directory()
163 				    nodeid == memb->nodeid) {  in dlm_recover_directory()
168 				 * added with memb->nodeid as the master. */  in dlm_recover_directory()
177 				left -= namelen;  in dlm_recover_directory()
187 	dlm_set_recover_status(ls, DLM_RS_DIR);  in dlm_recover_directory()
189 	log_rinfo(ls, "dlm_recover_directory %u in %u new",  in dlm_recover_directory()
197 static struct dlm_rsb *find_rsb_root(struct dlm_ls *ls, const char *name,  in find_rsb_root()  argument
203 	read_lock_bh(&ls->ls_rsbtbl_lock);  in find_rsb_root()
204 	rv = dlm_search_rsb_tree(&ls->ls_rsbtbl, name, len, &r);  in find_rsb_root()
205 	read_unlock_bh(&ls->ls_rsbtbl_lock);  in find_rsb_root()
209 	list_for_each_entry(r, &ls->ls_masters_list, res_masters_list) {  in find_rsb_root()
210 		if (len == r->res_length && !memcmp(name, r->res_name, len)) {  in find_rsb_root()
211 			log_debug(ls, "find_rsb_root revert to root_list %s",  in find_rsb_root()
212 				  r->res_name);  in find_rsb_root()
236 static void drop_dir_ctx(struct dlm_ls *ls, int nodeid)  in drop_dir_ctx()  argument
240 	write_lock_bh(&ls->ls_dir_dump_lock);  in drop_dir_ctx()
241 	list_for_each_entry_safe(dd, safe, &ls->ls_dir_dump_list, list) {  in drop_dir_ctx()
242 		if (dd->nodeid_init == nodeid) {  in drop_dir_ctx()
243 			log_error(ls, "drop dump seq %llu",  in drop_dir_ctx()
244 				 (unsigned long long)dd->seq_init);  in drop_dir_ctx()
245 			list_del(&dd->list);  in drop_dir_ctx()
249 	write_unlock_bh(&ls->ls_dir_dump_lock);  in drop_dir_ctx()
252 static struct dlm_dir_dump *lookup_dir_dump(struct dlm_ls *ls, int nodeid)  in lookup_dir_dump()  argument
256 	read_lock_bh(&ls->ls_dir_dump_lock);  in lookup_dir_dump()
257 	list_for_each_entry(iter, &ls->ls_dir_dump_list, list) {  in lookup_dir_dump()
258 		if (iter->nodeid_init == nodeid) {  in lookup_dir_dump()
263 	read_unlock_bh(&ls->ls_dir_dump_lock);  in lookup_dir_dump()
268 static struct dlm_dir_dump *init_dir_dump(struct dlm_ls *ls, int nodeid)  in init_dir_dump()  argument
272 	dd = lookup_dir_dump(ls, nodeid);  in init_dir_dump()
274 		log_error(ls, "found ongoing dir dump for node %d, will drop it",  in init_dir_dump()
276 		drop_dir_ctx(ls, nodeid);  in init_dir_dump()
283 	dd->seq_init = ls->ls_recover_seq;  in init_dir_dump()
284 	dd->nodeid_init = nodeid;  in init_dir_dump()
286 	write_lock_bh(&ls->ls_dir_dump_lock);  in init_dir_dump()
287 	list_add(&dd->list, &ls->ls_dir_dump_list);  in init_dir_dump()
288 	write_unlock_bh(&ls->ls_dir_dump_lock);  in init_dir_dump()
297 void dlm_copy_master_names(struct dlm_ls *ls, const char *inbuf, int inlen,  in dlm_copy_master_names()  argument
306 	read_lock_bh(&ls->ls_masters_lock);  in dlm_copy_master_names()
309 		dd = lookup_dir_dump(ls, nodeid);  in dlm_copy_master_names()
311 			log_error(ls, "failed to lookup dir dump context nodeid: %d",  in dlm_copy_master_names()
317 		r = find_rsb_root(ls, inbuf, inlen);  in dlm_copy_master_names()
319 			log_error(ls, "copy_master_names from %d start %d %.*s",  in dlm_copy_master_names()
323 		list = r->res_masters_list.next;  in dlm_copy_master_names()
326 		if (dd->last != &r->res_masters_list ||  in dlm_copy_master_names()
327 		    dd->seq_init != ls->ls_recover_seq) {  in dlm_copy_master_names()
328 			log_error(ls, "failed dir dump sanity check seq_init: %llu seq: %llu",  in dlm_copy_master_names()
329 				  (unsigned long long)dd->seq_init,  in dlm_copy_master_names()
330 				  (unsigned long long)ls->ls_recover_seq);  in dlm_copy_master_names()
334 		dd = init_dir_dump(ls, nodeid);  in dlm_copy_master_names()
336 			log_error(ls, "failed to allocate dir dump context");  in dlm_copy_master_names()
341 		list = ls->ls_masters_list.next;  in dlm_copy_master_names()
342 		dd->last = list;  in dlm_copy_master_names()
345 	for (offset = 0; list != &ls->ls_masters_list; list = list->next) {  in dlm_copy_master_names()
355 		 * name (r->res_length) +  in dlm_copy_master_names()
356 		 * end-of-block record 0x0000 (uint16_t)  in dlm_copy_master_names()
359 		if (offset + sizeof(uint16_t)*2 + r->res_length > outlen) {  in dlm_copy_master_names()
360 			/* Write end-of-block record */  in dlm_copy_master_names()
364 			dd->sent_msg++;  in dlm_copy_master_names()
368 		be_namelen = cpu_to_be16(r->res_length);  in dlm_copy_master_names()
371 		memcpy(outbuf + offset, r->res_name, r->res_length);  in dlm_copy_master_names()
372 		offset += r->res_length;  in dlm_copy_master_names()
373 		dd->sent_res++;  in dlm_copy_master_names()
374 		dd->last = list;  in dlm_copy_master_names()
382 	if ((list == &ls->ls_masters_list) &&  in dlm_copy_master_names()
388 		dd->sent_msg++;  in dlm_copy_master_names()
389 		log_rinfo(ls, "dlm_recover_directory nodeid %d sent %u res out %u messages",  in dlm_copy_master_names()
390 			  nodeid, dd->sent_res, dd->sent_msg);  in dlm_copy_master_names()
392 		write_lock_bh(&ls->ls_dir_dump_lock);  in dlm_copy_master_names()
393 		list_del_init(&dd->list);  in dlm_copy_master_names()
394 		write_unlock_bh(&ls->ls_dir_dump_lock);  in dlm_copy_master_names()
398 	read_unlock_bh(&ls->ls_masters_lock);  in dlm_copy_master_names()