[linux.git] / fs / dlm / dir.c

/******************************************************************************
*******************************************************************************
**
**  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
**  Copyright (C) 2004-2005 Red Hat, Inc.  All rights reserved.
**
**  This copyrighted material is made available to anyone wishing to use,
**  modify, copy, or redistribute it subject to the terms and conditions
**  of the GNU General Public License v.2.
**
*******************************************************************************
******************************************************************************/

#include "dlm_internal.h"
#include "lockspace.h"
#include "member.h"
#include "lowcomms.h"
#include "rcom.h"
#include "config.h"
#include "memory.h"
#include "recover.h"
#include "util.h"
#include "lock.h"
#include "dir.h"


static void put_free_de(struct dlm_ls *ls, struct dlm_direntry *de)
{
	spin_lock(&ls->ls_recover_list_lock);
	list_add(&de->list, &ls->ls_recover_list);
	spin_unlock(&ls->ls_recover_list_lock);
}

static struct dlm_direntry *get_free_de(struct dlm_ls *ls, int len)
{
	int found = 0;
	struct dlm_direntry *de;

	spin_lock(&ls->ls_recover_list_lock);
	list_for_each_entry(de, &ls->ls_recover_list, list) {
		if (de->length == len) {
			list_del(&de->list);
			de->master_nodeid = 0;
			memset(de->name, 0, len);
			found = 1;
			break;
		}
	}
	spin_unlock(&ls->ls_recover_list_lock);

	if (!found)
		de = kzalloc(sizeof(struct dlm_direntry) + len, GFP_KERNEL);
	return de;
}

void dlm_clear_free_entries(struct dlm_ls *ls)
{
	struct dlm_direntry *de;

	spin_lock(&ls->ls_recover_list_lock);
	while (!list_empty(&ls->ls_recover_list)) {
		de = list_entry(ls->ls_recover_list.next, struct dlm_direntry,
				list);
		list_del(&de->list);
		kfree(de);
	}
	spin_unlock(&ls->ls_recover_list_lock);
}

/*
 * We use the upper 16 bits of the hash value to select the directory node.
 * Low bits are used for distribution of rsb's among hash buckets on each node.
 *
 * To give the exact range wanted (0 to num_nodes-1), we apply a modulus of
 * num_nodes to the hash value.  This value in the desired range is used as an
 * offset into the sorted list of nodeid's to give the particular nodeid.
 */

int dlm_hash2nodeid(struct dlm_ls *ls, uint32_t hash)
{
	struct list_head *tmp;
	struct dlm_member *memb = NULL;
	uint32_t node, n = 0;
	int nodeid;

	if (ls->ls_num_nodes == 1) {
		nodeid = dlm_our_nodeid();
		goto out;
	}

	if (ls->ls_node_array) {
		node = (hash >> 16) % ls->ls_total_weight;
		nodeid = ls->ls_node_array[node];
		goto out;
	}

	/* make_member_array() failed to kmalloc ls_node_array... */

	node = (hash >> 16) % ls->ls_num_nodes;

	list_for_each(tmp, &ls->ls_nodes) {
		if (n++ != node)
			continue;
		memb = list_entry(tmp, struct dlm_member, list);
		break;
	}

	DLM_ASSERT(memb , printk("num_nodes=%u n=%u node=%u\n",
				 ls->ls_num_nodes, n, node););
	nodeid = memb->nodeid;
 out:
	return nodeid;
}

int dlm_dir_nodeid(struct dlm_rsb *r)
{
	return dlm_hash2nodeid(r->res_ls, r->res_hash);
}

static inline uint32_t dir_hash(struct dlm_ls *ls, char *name, int len)
{
	uint32_t val;

	val = jhash(name, len, 0);
	val &= (ls->ls_dirtbl_size - 1);

	return val;
}

static void add_entry_to_hash(struct dlm_ls *ls, struct dlm_direntry *de)
{
	uint32_t bucket;

	bucket = dir_hash(ls, de->name, de->length);
	list_add_tail(&de->list, &ls->ls_dirtbl[bucket].list);
}

static struct dlm_direntry *search_bucket(struct dlm_ls *ls, char *name,
					  int namelen, uint32_t bucket)
{
	struct dlm_direntry *de;

	list_for_each_entry(de, &ls->ls_dirtbl[bucket].list, list) {
		if (de->length == namelen && !memcmp(name, de->name, namelen))
			goto out;
	}
	de = NULL;
 out:
	return de;
}

void dlm_dir_remove_entry(struct dlm_ls *ls, int nodeid, char *name, int namelen)
{
	struct dlm_direntry *de;
	uint32_t bucket;

	bucket = dir_hash(ls, name, namelen);

	write_lock(&ls->ls_dirtbl[bucket].lock);

	de = search_bucket(ls, name, namelen, bucket);

	if (!de) {
		log_error(ls, "remove fr %u none", nodeid);
		goto out;
	}

	if (de->master_nodeid != nodeid) {
		log_error(ls, "remove fr %u ID %u", nodeid, de->master_nodeid);
		goto out;
	}

	list_del(&de->list);
	kfree(de);
 out:
	write_unlock(&ls->ls_dirtbl[bucket].lock);
}

void dlm_dir_clear(struct dlm_ls *ls)
{
	struct list_head *head;
	struct dlm_direntry *de;
	int i;

	DLM_ASSERT(list_empty(&ls->ls_recover_list), );

	for (i = 0; i < ls->ls_dirtbl_size; i++) {
		write_lock(&ls->ls_dirtbl[i].lock);
		head = &ls->ls_dirtbl[i].list;
		while (!list_empty(head)) {
			de = list_entry(head->next, struct dlm_direntry, list);
			list_del(&de->list);
			put_free_de(ls, de);
		}
		write_unlock(&ls->ls_dirtbl[i].lock);
	}
}

int dlm_recover_directory(struct dlm_ls *ls)
{
	struct dlm_member *memb;
	struct dlm_direntry *de;
	char *b, *last_name = NULL;
	int error = -ENOMEM, last_len, count = 0;
	uint16_t namelen;

	log_debug(ls, "dlm_recover_directory");

	if (dlm_no_directory(ls))
		goto out_status;

	dlm_dir_clear(ls);

	last_name = kmalloc(DLM_RESNAME_MAXLEN, GFP_KERNEL);
	if (!last_name)
		goto out;

	list_for_each_entry(memb, &ls->ls_nodes, list) {
		memset(last_name, 0, DLM_RESNAME_MAXLEN);
		last_len = 0;

		for (;;) {
			int left;
			error = dlm_recovery_stopped(ls);
			if (error)
				goto out_free;

			error = dlm_rcom_names(ls, memb->nodeid,
					       last_name, last_len);
			if (error)
				goto out_free;

			schedule();

			/*
			 * pick namelen/name pairs out of received buffer
			 */

			b = ls->ls_recover_buf->rc_buf;
			left = ls->ls_recover_buf->rc_header.h_length;
			left -= sizeof(struct dlm_rcom);

			for (;;) {
				__be16 v;

				error = -EINVAL;
				if (left < sizeof(__be16))
					goto out_free;

				memcpy(&v, b, sizeof(__be16));
				namelen = be16_to_cpu(v);
				b += sizeof(__be16);
				left -= sizeof(__be16);

				/* namelen of 0xFFFFF marks end of names for
				   this node; namelen of 0 marks end of the
				   buffer */

				if (namelen == 0xFFFF)
					goto done;
				if (!namelen)
					break;

				if (namelen > left)
					goto out_free;

				if (namelen > DLM_RESNAME_MAXLEN)
					goto out_free;

				error = -ENOMEM;
				de = get_free_de(ls, namelen);
				if (!de)
					goto out_free;

				de->master_nodeid = memb->nodeid;
				de->length = namelen;
				last_len = namelen;
				memcpy(de->name, b, namelen);
				memcpy(last_name, b, namelen);
				b += namelen;
				left -= namelen;

				add_entry_to_hash(ls, de);
				count++;
			}
		}
         done:
		;
	}

 out_status:
	error = 0;
	dlm_set_recover_status(ls, DLM_RS_DIR);
	log_debug(ls, "dlm_recover_directory %d entries", count);
 out_free:
	kfree(last_name);
 out:
	dlm_clear_free_entries(ls);
	return error;
}

static int get_entry(struct dlm_ls *ls, int nodeid, char *name,
		     int namelen, int *r_nodeid)
{
	struct dlm_direntry *de, *tmp;
	uint32_t bucket;

	bucket = dir_hash(ls, name, namelen);

	write_lock(&ls->ls_dirtbl[bucket].lock);
	de = search_bucket(ls, name, namelen, bucket);
	if (de) {
		*r_nodeid = de->master_nodeid;
		write_unlock(&ls->ls_dirtbl[bucket].lock);
		if (*r_nodeid == nodeid)
			return -EEXIST;
		return 0;
	}

	write_unlock(&ls->ls_dirtbl[bucket].lock);

	de = kzalloc(sizeof(struct dlm_direntry) + namelen, GFP_KERNEL);
	if (!de)
		return -ENOMEM;

	de->master_nodeid = nodeid;
	de->length = namelen;
	memcpy(de->name, name, namelen);

	write_lock(&ls->ls_dirtbl[bucket].lock);
	tmp = search_bucket(ls, name, namelen, bucket);
	if (tmp) {
		kfree(de);
		de = tmp;
	} else {
		list_add_tail(&de->list, &ls->ls_dirtbl[bucket].list);
	}
	*r_nodeid = de->master_nodeid;
	write_unlock(&ls->ls_dirtbl[bucket].lock);
	return 0;
}

int dlm_dir_lookup(struct dlm_ls *ls, int nodeid, char *name, int namelen,
		   int *r_nodeid)
{
	return get_entry(ls, nodeid, name, namelen, r_nodeid);
}

static struct dlm_rsb *find_rsb_root(struct dlm_ls *ls, char *name, int len)
{
	struct dlm_rsb *r;

	down_read(&ls->ls_root_sem);
	list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
		if (len == r->res_length && !memcmp(name, r->res_name, len)) {
			up_read(&ls->ls_root_sem);
			return r;
		}
	}
	up_read(&ls->ls_root_sem);
	return NULL;
}

/* Find the rsb where we left off (or start again), then send rsb names
   for rsb's we're master of and whose directory node matches the requesting
   node.  inbuf is the rsb name last sent, inlen is the name's length */

void dlm_copy_master_names(struct dlm_ls *ls, char *inbuf, int inlen,
 			   char *outbuf, int outlen, int nodeid)
{
	struct list_head *list;
	struct dlm_rsb *r;
	int offset = 0, dir_nodeid;
	uint16_t be_namelen;

	down_read(&ls->ls_root_sem);

	if (inlen > 1) {
		r = find_rsb_root(ls, inbuf, inlen);
		if (!r) {
			inbuf[inlen - 1] = '\0';
			log_error(ls, "copy_master_names from %d start %d %s",
				  nodeid, inlen, inbuf);
			goto out;
		}
		list = r->res_root_list.next;
	} else {
		list = ls->ls_root_list.next;
	}

	for (offset = 0; list != &ls->ls_root_list; list = list->next) {
		r = list_entry(list, struct dlm_rsb, res_root_list);
		if (r->res_nodeid)
			continue;

		dir_nodeid = dlm_dir_nodeid(r);
		if (dir_nodeid != nodeid)
			continue;

		/*
		 * The block ends when we can't fit the following in the
		 * remaining buffer space:
		 * namelen (uint16_t) +
		 * name (r->res_length) +
		 * end-of-block record 0x0000 (uint16_t)
		 */

		if (offset + sizeof(uint16_t)*2 + r->res_length > outlen) {
			/* Write end-of-block record */
			be_namelen = 0;
			memcpy(outbuf + offset, &be_namelen, sizeof(uint16_t));
			offset += sizeof(uint16_t);
			goto out;
		}

		be_namelen = cpu_to_be16(r->res_length);
		memcpy(outbuf + offset, &be_namelen, sizeof(uint16_t));
		offset += sizeof(uint16_t);
		memcpy(outbuf + offset, r->res_name, r->res_length);
		offset += r->res_length;
	}

	/*
	 * If we've reached the end of the list (and there's room) write a
	 * terminating record.
	 */

	if ((list == &ls->ls_root_list) &&
	    (offset + sizeof(uint16_t) <= outlen)) {
		be_namelen = 0xFFFF;
		memcpy(outbuf + offset, &be_namelen, sizeof(uint16_t));
		offset += sizeof(uint16_t);
	}

 out:
	up_read(&ls->ls_root_sem);
}
Commit	Line	Data
e7fd4179 DT	1	/******************************************************************************
	2	*******************************************************************************
	3	**
	4	** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
	5	** Copyright (C) 2004-2005 Red Hat, Inc. All rights reserved.
	6	**
	7	** This copyrighted material is made available to anyone wishing to use,
	8	** modify, copy, or redistribute it subject to the terms and conditions
	9	** of the GNU General Public License v.2.
	10	**
	11	*******************************************************************************
	12	******************************************************************************/
	13
	14	#include "dlm_internal.h"
	15	#include "lockspace.h"
	16	#include "member.h"
	17	#include "lowcomms.h"
	18	#include "rcom.h"
	19	#include "config.h"
	20	#include "memory.h"
	21	#include "recover.h"
	22	#include "util.h"
	23	#include "lock.h"
	24	#include "dir.h"
	25
	26
	27	static void put_free_de(struct dlm_ls ls, struct dlm_direntry de)
	28	{
	29	spin_lock(&ls->ls_recover_list_lock);
	30	list_add(&de->list, &ls->ls_recover_list);
	31	spin_unlock(&ls->ls_recover_list_lock);
	32	}
	33
	34	static struct dlm_direntry get_free_de(struct dlm_ls ls, int len)
	35	{
90135925	36	int found = 0;
e7fd4179 DT	37	struct dlm_direntry *de;
	38
	39	spin_lock(&ls->ls_recover_list_lock);
	40	list_for_each_entry(de, &ls->ls_recover_list, list) {
	41	if (de->length == len) {
	42	list_del(&de->list);
	43	de->master_nodeid = 0;
	44	memset(de->name, 0, len);
90135925	45	found = 1;
e7fd4179 DT	46	break;
	47	}
	48	}
	49	spin_unlock(&ls->ls_recover_list_lock);
	50
	51	if (!found)
52bda2b5	52	de = kzalloc(sizeof(struct dlm_direntry) + len, GFP_KERNEL);
e7fd4179 DT	53	return de;
	54	}
	55
	56	void dlm_clear_free_entries(struct dlm_ls *ls)
	57	{
	58	struct dlm_direntry *de;
	59
	60	spin_lock(&ls->ls_recover_list_lock);
	61	while (!list_empty(&ls->ls_recover_list)) {
	62	de = list_entry(ls->ls_recover_list.next, struct dlm_direntry,
	63	list);
	64	list_del(&de->list);
52bda2b5	65	kfree(de);
e7fd4179 DT	66	}
	67	spin_unlock(&ls->ls_recover_list_lock);
	68	}
	69
	70	/*
	71	* We use the upper 16 bits of the hash value to select the directory node.
	72	* Low bits are used for distribution of rsb's among hash buckets on each node.
	73	*
	74	* To give the exact range wanted (0 to num_nodes-1), we apply a modulus of
	75	* num_nodes to the hash value. This value in the desired range is used as an
	76	* offset into the sorted list of nodeid's to give the particular nodeid.
	77	*/
	78
	79	int dlm_hash2nodeid(struct dlm_ls *ls, uint32_t hash)
	80	{
	81	struct list_head *tmp;
	82	struct dlm_member *memb = NULL;
	83	uint32_t node, n = 0;
	84	int nodeid;
	85
	86	if (ls->ls_num_nodes == 1) {
	87	nodeid = dlm_our_nodeid();
	88	goto out;
	89	}
	90
	91	if (ls->ls_node_array) {
	92	node = (hash >> 16) % ls->ls_total_weight;
	93	nodeid = ls->ls_node_array[node];
	94	goto out;
	95	}
	96
	97	/* make_member_array() failed to kmalloc ls_node_array... */
	98
	99	node = (hash >> 16) % ls->ls_num_nodes;
	100
	101	list_for_each(tmp, &ls->ls_nodes) {
	102	if (n++ != node)
	103	continue;
	104	memb = list_entry(tmp, struct dlm_member, list);
	105	break;
	106	}
	107
	108	DLM_ASSERT(memb , printk("num_nodes=%u n=%u node=%u\n",
	109	ls->ls_num_nodes, n, node););
	110	nodeid = memb->nodeid;
	111	out:
	112	return nodeid;
	113	}
	114
	115	int dlm_dir_nodeid(struct dlm_rsb *r)
	116	{
	117	return dlm_hash2nodeid(r->res_ls, r->res_hash);
	118	}
	119
	120	static inline uint32_t dir_hash(struct dlm_ls ls, char name, int len)
	121	{
	122	uint32_t val;
	123
	124	val = jhash(name, len, 0);
	125	val &= (ls->ls_dirtbl_size - 1);
	126
	127	return val;
	128	}
	129
130	static void add_entry_to_hash(struct dlm_ls ls, struct dlm_direntry de)
131	{
132	uint32_t bucket;
133
134	bucket = dir_hash(ls, de->name, de->length);
135	list_add_tail(&de->list, &ls->ls_dirtbl[bucket].list);
136	}
137
138	static struct dlm_direntry search_bucket(struct dlm_ls ls, char *name,
139	int namelen, uint32_t bucket)
140	{
141	struct dlm_direntry *de;
142
143	list_for_each_entry(de, &ls->ls_dirtbl[bucket].list, list) {
144	if (de->length == namelen && !memcmp(name, de->name, namelen))
145	goto out;
146	}
147	de = NULL;
148	out:
149	return de;
150	}
151
152	void dlm_dir_remove_entry(struct dlm_ls ls, int nodeid, char name, int namelen)
153	{
154	struct dlm_direntry *de;
155	uint32_t bucket;
156
157	bucket = dir_hash(ls, name, namelen);
158
159	write_lock(&ls->ls_dirtbl[bucket].lock);
160
161	de = search_bucket(ls, name, namelen, bucket);
162
163	if (!de) {
164	log_error(ls, "remove fr %u none", nodeid);
165	goto out;
166	}
167
168	if (de->master_nodeid != nodeid) {
169	log_error(ls, "remove fr %u ID %u", nodeid, de->master_nodeid);
170	goto out;
171	}
172
173	list_del(&de->list);
52bda2b5	174	kfree(de);
e7fd4179 DT	175	out:
	176	write_unlock(&ls->ls_dirtbl[bucket].lock);
	177	}
	178
	179	void dlm_dir_clear(struct dlm_ls *ls)
	180	{
	181	struct list_head *head;
	182	struct dlm_direntry *de;
	183	int i;
	184
	185	DLM_ASSERT(list_empty(&ls->ls_recover_list), );
	186
	187	for (i = 0; i < ls->ls_dirtbl_size; i++) {
	188	write_lock(&ls->ls_dirtbl[i].lock);
	189	head = &ls->ls_dirtbl[i].list;
	190	while (!list_empty(head)) {
	191	de = list_entry(head->next, struct dlm_direntry, list);
	192	list_del(&de->list);
	193	put_free_de(ls, de);
	194	}
	195	write_unlock(&ls->ls_dirtbl[i].lock);
	196	}
	197	}
	198
	199	int dlm_recover_directory(struct dlm_ls *ls)
	200	{
	201	struct dlm_member *memb;
	202	struct dlm_direntry *de;
	203	char b, last_name = NULL;
	204	int error = -ENOMEM, last_len, count = 0;
	205	uint16_t namelen;
	206
	207	log_debug(ls, "dlm_recover_directory");
	208
	209	if (dlm_no_directory(ls))
	210	goto out_status;
	211
	212	dlm_dir_clear(ls);
	213
	214	last_name = kmalloc(DLM_RESNAME_MAXLEN, GFP_KERNEL);
	215	if (!last_name)
	216	goto out;
	217
	218	list_for_each_entry(memb, &ls->ls_nodes, list) {
	219	memset(last_name, 0, DLM_RESNAME_MAXLEN);
	220	last_len = 0;
	221
	222	for (;;) {
cd9df1aa	223	int left;
e7fd4179 DT	224	error = dlm_recovery_stopped(ls);
	225	if (error)
	226	goto out_free;
	227
	228	error = dlm_rcom_names(ls, memb->nodeid,
	229	last_name, last_len);
	230	if (error)
	231	goto out_free;
	232
	233	schedule();
	234
	235	/*
	236	* pick namelen/name pairs out of received buffer
	237	*/
	238
4007685c	239	b = ls->ls_recover_buf->rc_buf;
cd9df1aa AV	240	left = ls->ls_recover_buf->rc_header.h_length;
cd9df1aa AV	241	left -= sizeof(struct dlm_rcom);
e7fd4179 DT	242
e7fd4179 DT	243	for (;;) {
cd9df1aa AV	244	__be16 v;
	245
	246	error = -EINVAL;
	247	if (left < sizeof(__be16))
	248	goto out_free;
	249
	250	memcpy(&v, b, sizeof(__be16));
	251	namelen = be16_to_cpu(v);
	252	b += sizeof(__be16);
	253	left -= sizeof(__be16);
e7fd4179 DT	254
	255	/* namelen of 0xFFFFF marks end of names for
	256	this node; namelen of 0 marks end of the
	257	buffer */
	258
	259	if (namelen == 0xFFFF)
	260	goto done;
	261	if (!namelen)
	262	break;
	263
cd9df1aa AV	264	if (namelen > left)
	265	goto out_free;
	266
	267	if (namelen > DLM_RESNAME_MAXLEN)
	268	goto out_free;
	269
e7fd4179 DT	270	error = -ENOMEM;
	271	de = get_free_de(ls, namelen);
	272	if (!de)
	273	goto out_free;
	274
	275	de->master_nodeid = memb->nodeid;
	276	de->length = namelen;
	277	last_len = namelen;
	278	memcpy(de->name, b, namelen);
	279	memcpy(last_name, b, namelen);
	280	b += namelen;
cd9df1aa	281	left -= namelen;
e7fd4179 DT	282
	283	add_entry_to_hash(ls, de);
	284	count++;
	285	}
	286	}
	287	done:
	288	;
	289	}
	290
	291	out_status:
	292	error = 0;
	293	dlm_set_recover_status(ls, DLM_RS_DIR);
	294	log_debug(ls, "dlm_recover_directory %d entries", count);
	295	out_free:
	296	kfree(last_name);
	297	out:
	298	dlm_clear_free_entries(ls);
	299	return error;
	300	}
	301
	302	static int get_entry(struct dlm_ls ls, int nodeid, char name,
	303	int namelen, int *r_nodeid)
	304	{
	305	struct dlm_direntry de, tmp;
	306	uint32_t bucket;
	307
	308	bucket = dir_hash(ls, name, namelen);
	309
	310	write_lock(&ls->ls_dirtbl[bucket].lock);
	311	de = search_bucket(ls, name, namelen, bucket);
	312	if (de) {
	313	*r_nodeid = de->master_nodeid;
	314	write_unlock(&ls->ls_dirtbl[bucket].lock);
	315	if (*r_nodeid == nodeid)
	316	return -EEXIST;
	317	return 0;
	318	}
	319
	320	write_unlock(&ls->ls_dirtbl[bucket].lock);
	321
52bda2b5	322	de = kzalloc(sizeof(struct dlm_direntry) + namelen, GFP_KERNEL);
e7fd4179 DT	323	if (!de)
	324	return -ENOMEM;
	325
	326	de->master_nodeid = nodeid;
	327	de->length = namelen;
	328	memcpy(de->name, name, namelen);
	329
	330	write_lock(&ls->ls_dirtbl[bucket].lock);
	331	tmp = search_bucket(ls, name, namelen, bucket);
	332	if (tmp) {
52bda2b5	333	kfree(de);
e7fd4179 DT	334	de = tmp;
	335	} else {
	336	list_add_tail(&de->list, &ls->ls_dirtbl[bucket].list);
	337	}
	338	*r_nodeid = de->master_nodeid;
	339	write_unlock(&ls->ls_dirtbl[bucket].lock);
	340	return 0;
	341	}
	342
	343	int dlm_dir_lookup(struct dlm_ls ls, int nodeid, char name, int namelen,
	344	int *r_nodeid)
	345	{
	346	return get_entry(ls, nodeid, name, namelen, r_nodeid);
	347	}
	348
85f0379a DT	349	static struct dlm_rsb find_rsb_root(struct dlm_ls ls, char *name, int len)
	350	{
	351	struct dlm_rsb *r;
	352
	353	down_read(&ls->ls_root_sem);
	354	list_for_each_entry(r, &ls->ls_root_list, res_root_list) {
	355	if (len == r->res_length && !memcmp(name, r->res_name, len)) {
	356	up_read(&ls->ls_root_sem);
	357	return r;
	358	}
	359	}
	360	up_read(&ls->ls_root_sem);
	361	return NULL;
	362	}
	363
	364	/* Find the rsb where we left off (or start again), then send rsb names
	365	for rsb's we're master of and whose directory node matches the requesting
	366	node. inbuf is the rsb name last sent, inlen is the name's length */
e7fd4179 DT	367
	368	void dlm_copy_master_names(struct dlm_ls ls, char inbuf, int inlen,
	369	char *outbuf, int outlen, int nodeid)
	370	{
	371	struct list_head *list;
85f0379a DT	372	struct dlm_rsb *r;
85f0379a DT	373	int offset = 0, dir_nodeid;
e7fd4179 DT	374	uint16_t be_namelen;
e7fd4179 DT	375
e7fd4179	376	down_read(&ls->ls_root_sem);
85f0379a DT	377
	378	if (inlen > 1) {
	379	r = find_rsb_root(ls, inbuf, inlen);
	380	if (!r) {
	381	inbuf[inlen - 1] = '\0';
	382	log_error(ls, "copy_master_names from %d start %d %s",
	383	nodeid, inlen, inbuf);
	384	goto out;
	385	}
	386	list = r->res_root_list.next;
	387	} else {
e7fd4179	388	list = ls->ls_root_list.next;
85f0379a	389	}
e7fd4179 DT	390
	391	for (offset = 0; list != &ls->ls_root_list; list = list->next) {
	392	r = list_entry(list, struct dlm_rsb, res_root_list);
	393	if (r->res_nodeid)
	394	continue;
	395
	396	dir_nodeid = dlm_dir_nodeid(r);
	397	if (dir_nodeid != nodeid)
	398	continue;
	399
	400	/*
	401	* The block ends when we can't fit the following in the
	402	* remaining buffer space:
	403	* namelen (uint16_t) +
	404	* name (r->res_length) +
	405	* end-of-block record 0x0000 (uint16_t)
	406	*/
	407
	408	if (offset + sizeof(uint16_t)*2 + r->res_length > outlen) {
	409	/* Write end-of-block record */
	410	be_namelen = 0;
	411	memcpy(outbuf + offset, &be_namelen, sizeof(uint16_t));
	412	offset += sizeof(uint16_t);
	413	goto out;
	414	}
	415
	416	be_namelen = cpu_to_be16(r->res_length);
	417	memcpy(outbuf + offset, &be_namelen, sizeof(uint16_t));
	418	offset += sizeof(uint16_t);
	419	memcpy(outbuf + offset, r->res_name, r->res_length);
	420	offset += r->res_length;
	421	}
	422
	423	/*
	424	* If we've reached the end of the list (and there's room) write a
	425	* terminating record.
	426	*/
	427
	428	if ((list == &ls->ls_root_list) &&
	429	(offset + sizeof(uint16_t) <= outlen)) {
	430	be_namelen = 0xFFFF;
	431	memcpy(outbuf + offset, &be_namelen, sizeof(uint16_t));
	432	offset += sizeof(uint16_t);
	433	}
	434
	435	out:
	436	up_read(&ls->ls_root_sem);
	437	}
	438