[mirror_ubuntu-artful-kernel.git] / net / ceph / cls_lock_client.c

#include <linux/ceph/ceph_debug.h>

#include <linux/types.h>
#include <linux/slab.h>

#include <linux/ceph/cls_lock_client.h>
#include <linux/ceph/decode.h>

/**
 * ceph_cls_lock - grab rados lock for object
 * @oid, @oloc: object to lock
 * @lock_name: the name of the lock
 * @type: lock type (CEPH_CLS_LOCK_EXCLUSIVE or CEPH_CLS_LOCK_SHARED)
 * @cookie: user-defined identifier for this instance of the lock
 * @tag: user-defined tag
 * @desc: user-defined lock description
 * @flags: lock flags
 *
 * All operations on the same lock should use the same tag.
 */
int ceph_cls_lock(struct ceph_osd_client *osdc,
		  struct ceph_object_id *oid,
		  struct ceph_object_locator *oloc,
		  char *lock_name, u8 type, char *cookie,
		  char *tag, char *desc, u8 flags)
{
	int lock_op_buf_size;
	int name_len = strlen(lock_name);
	int cookie_len = strlen(cookie);
	int tag_len = strlen(tag);
	int desc_len = strlen(desc);
	void *p, *end;
	struct page *lock_op_page;
	struct timespec mtime;
	int ret;

	lock_op_buf_size = name_len + sizeof(__le32) +
			   cookie_len + sizeof(__le32) +
			   tag_len + sizeof(__le32) +
			   desc_len + sizeof(__le32) +
			   sizeof(struct ceph_timespec) +
			   /* flag and type */
			   sizeof(u8) + sizeof(u8) +
			   CEPH_ENCODING_START_BLK_LEN;
	if (lock_op_buf_size > PAGE_SIZE)
		return -E2BIG;

	lock_op_page = alloc_page(GFP_NOIO);
	if (!lock_op_page)
		return -ENOMEM;

	p = page_address(lock_op_page);
	end = p + lock_op_buf_size;

	/* encode cls_lock_lock_op struct */
	ceph_start_encoding(&p, 1, 1,
			    lock_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
	ceph_encode_string(&p, end, lock_name, name_len);
	ceph_encode_8(&p, type);
	ceph_encode_string(&p, end, cookie, cookie_len);
	ceph_encode_string(&p, end, tag, tag_len);
	ceph_encode_string(&p, end, desc, desc_len);
	/* only support infinite duration */
	memset(&mtime, 0, sizeof(mtime));
	ceph_encode_timespec(p, &mtime);
	p += sizeof(struct ceph_timespec);
	ceph_encode_8(&p, flags);

	dout("%s lock_name %s type %d cookie %s tag %s desc %s flags 0x%x\n",
	     __func__, lock_name, type, cookie, tag, desc, flags);
	ret = ceph_osdc_call(osdc, oid, oloc, "lock", "lock",
			     CEPH_OSD_FLAG_WRITE, lock_op_page,
			     lock_op_buf_size, NULL, NULL);

	dout("%s: status %d\n", __func__, ret);
	__free_page(lock_op_page);
	return ret;
}
EXPORT_SYMBOL(ceph_cls_lock);

/**
 * ceph_cls_unlock - release rados lock for object
 * @oid, @oloc: object to lock
 * @lock_name: the name of the lock
 * @cookie: user-defined identifier for this instance of the lock
 */
int ceph_cls_unlock(struct ceph_osd_client *osdc,
		    struct ceph_object_id *oid,
		    struct ceph_object_locator *oloc,
		    char *lock_name, char *cookie)
{
	int unlock_op_buf_size;
	int name_len = strlen(lock_name);
	int cookie_len = strlen(cookie);
	void *p, *end;
	struct page *unlock_op_page;
	int ret;

	unlock_op_buf_size = name_len + sizeof(__le32) +
			     cookie_len + sizeof(__le32) +
			     CEPH_ENCODING_START_BLK_LEN;
	if (unlock_op_buf_size > PAGE_SIZE)
		return -E2BIG;

	unlock_op_page = alloc_page(GFP_NOIO);
	if (!unlock_op_page)
		return -ENOMEM;

	p = page_address(unlock_op_page);
	end = p + unlock_op_buf_size;

	/* encode cls_lock_unlock_op struct */
	ceph_start_encoding(&p, 1, 1,
			    unlock_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
	ceph_encode_string(&p, end, lock_name, name_len);
	ceph_encode_string(&p, end, cookie, cookie_len);

	dout("%s lock_name %s cookie %s\n", __func__, lock_name, cookie);
	ret = ceph_osdc_call(osdc, oid, oloc, "lock", "unlock",
			     CEPH_OSD_FLAG_WRITE, unlock_op_page,
			     unlock_op_buf_size, NULL, NULL);

	dout("%s: status %d\n", __func__, ret);
	__free_page(unlock_op_page);
	return ret;
}
EXPORT_SYMBOL(ceph_cls_unlock);

/**
 * ceph_cls_break_lock - release rados lock for object for specified client
 * @oid, @oloc: object to lock
 * @lock_name: the name of the lock
 * @cookie: user-defined identifier for this instance of the lock
 * @locker: current lock owner
 */
int ceph_cls_break_lock(struct ceph_osd_client *osdc,
			struct ceph_object_id *oid,
			struct ceph_object_locator *oloc,
			char *lock_name, char *cookie,
			struct ceph_entity_name *locker)
{
	int break_op_buf_size;
	int name_len = strlen(lock_name);
	int cookie_len = strlen(cookie);
	struct page *break_op_page;
	void *p, *end;
	int ret;

	break_op_buf_size = name_len + sizeof(__le32) +
			    cookie_len + sizeof(__le32) +
			    sizeof(u8) + sizeof(__le64) +
			    CEPH_ENCODING_START_BLK_LEN;
	if (break_op_buf_size > PAGE_SIZE)
		return -E2BIG;

	break_op_page = alloc_page(GFP_NOIO);
	if (!break_op_page)
		return -ENOMEM;

	p = page_address(break_op_page);
	end = p + break_op_buf_size;

	/* encode cls_lock_break_op struct */
	ceph_start_encoding(&p, 1, 1,
			    break_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
	ceph_encode_string(&p, end, lock_name, name_len);
	ceph_encode_copy(&p, locker, sizeof(*locker));
	ceph_encode_string(&p, end, cookie, cookie_len);

	dout("%s lock_name %s cookie %s locker %s%llu\n", __func__, lock_name,
	     cookie, ENTITY_NAME(*locker));
	ret = ceph_osdc_call(osdc, oid, oloc, "lock", "break_lock",
			     CEPH_OSD_FLAG_WRITE, break_op_page,
			     break_op_buf_size, NULL, NULL);

	dout("%s: status %d\n", __func__, ret);
	__free_page(break_op_page);
	return ret;
}
EXPORT_SYMBOL(ceph_cls_break_lock);

int ceph_cls_set_cookie(struct ceph_osd_client *osdc,
			struct ceph_object_id *oid,
			struct ceph_object_locator *oloc,
			char *lock_name, u8 type, char *old_cookie,
			char *tag, char *new_cookie)
{
	int cookie_op_buf_size;
	int name_len = strlen(lock_name);
	int old_cookie_len = strlen(old_cookie);
	int tag_len = strlen(tag);
	int new_cookie_len = strlen(new_cookie);
	void *p, *end;
	struct page *cookie_op_page;
	int ret;

	cookie_op_buf_size = name_len + sizeof(__le32) +
			     old_cookie_len + sizeof(__le32) +
			     tag_len + sizeof(__le32) +
			     new_cookie_len + sizeof(__le32) +
			     sizeof(u8) + CEPH_ENCODING_START_BLK_LEN;
	if (cookie_op_buf_size > PAGE_SIZE)
		return -E2BIG;

	cookie_op_page = alloc_page(GFP_NOIO);
	if (!cookie_op_page)
		return -ENOMEM;

	p = page_address(cookie_op_page);
	end = p + cookie_op_buf_size;

	/* encode cls_lock_set_cookie_op struct */
	ceph_start_encoding(&p, 1, 1,
			    cookie_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
	ceph_encode_string(&p, end, lock_name, name_len);
	ceph_encode_8(&p, type);
	ceph_encode_string(&p, end, old_cookie, old_cookie_len);
	ceph_encode_string(&p, end, tag, tag_len);
	ceph_encode_string(&p, end, new_cookie, new_cookie_len);

	dout("%s lock_name %s type %d old_cookie %s tag %s new_cookie %s\n",
	     __func__, lock_name, type, old_cookie, tag, new_cookie);
	ret = ceph_osdc_call(osdc, oid, oloc, "lock", "set_cookie",
			     CEPH_OSD_FLAG_WRITE, cookie_op_page,
			     cookie_op_buf_size, NULL, NULL);

	dout("%s: status %d\n", __func__, ret);
	__free_page(cookie_op_page);
	return ret;
}
EXPORT_SYMBOL(ceph_cls_set_cookie);

void ceph_free_lockers(struct ceph_locker *lockers, u32 num_lockers)
{
	int i;

	for (i = 0; i < num_lockers; i++)
		kfree(lockers[i].id.cookie);
	kfree(lockers);
}
EXPORT_SYMBOL(ceph_free_lockers);

static int decode_locker(void **p, void *end, struct ceph_locker *locker)
{
	u8 struct_v;
	u32 len;
	char *s;
	int ret;

	ret = ceph_start_decoding(p, end, 1, "locker_id_t", &struct_v, &len);
	if (ret)
		return ret;

	ceph_decode_copy(p, &locker->id.name, sizeof(locker->id.name));
	s = ceph_extract_encoded_string(p, end, NULL, GFP_NOIO);
	if (IS_ERR(s))
		return PTR_ERR(s);

	locker->id.cookie = s;

	ret = ceph_start_decoding(p, end, 1, "locker_info_t", &struct_v, &len);
	if (ret)
		return ret;

	*p += sizeof(struct ceph_timespec); /* skip expiration */
	ceph_decode_copy(p, &locker->info.addr, sizeof(locker->info.addr));
	ceph_decode_addr(&locker->info.addr);
	len = ceph_decode_32(p);
	*p += len; /* skip description */

	dout("%s %s%llu cookie %s addr %s\n", __func__,
	     ENTITY_NAME(locker->id.name), locker->id.cookie,
	     ceph_pr_addr(&locker->info.addr.in_addr));
	return 0;
}

static int decode_lockers(void **p, void *end, u8 *type, char **tag,
			  struct ceph_locker **lockers, u32 *num_lockers)
{
	u8 struct_v;
	u32 struct_len;
	char *s;
	int i;
	int ret;

	ret = ceph_start_decoding(p, end, 1, "cls_lock_get_info_reply",
				  &struct_v, &struct_len);
	if (ret)
		return ret;

	*num_lockers = ceph_decode_32(p);
	*lockers = kcalloc(*num_lockers, sizeof(**lockers), GFP_NOIO);
	if (!*lockers)
		return -ENOMEM;

	for (i = 0; i < *num_lockers; i++) {
		ret = decode_locker(p, end, *lockers + i);
		if (ret)
			goto err_free_lockers;
	}

	*type = ceph_decode_8(p);
	s = ceph_extract_encoded_string(p, end, NULL, GFP_NOIO);
	if (IS_ERR(s)) {
		ret = PTR_ERR(s);
		goto err_free_lockers;
	}

	*tag = s;
	return 0;

err_free_lockers:
	ceph_free_lockers(*lockers, *num_lockers);
	return ret;
}

/*
 * On success, the caller is responsible for:
 *
 *     kfree(tag);
 *     ceph_free_lockers(lockers, num_lockers);
 */
int ceph_cls_lock_info(struct ceph_osd_client *osdc,
		       struct ceph_object_id *oid,
		       struct ceph_object_locator *oloc,
		       char *lock_name, u8 *type, char **tag,
		       struct ceph_locker **lockers, u32 *num_lockers)
{
	int get_info_op_buf_size;
	int name_len = strlen(lock_name);
	struct page *get_info_op_page, *reply_page;
	size_t reply_len = PAGE_SIZE;
	void *p, *end;
	int ret;

	get_info_op_buf_size = name_len + sizeof(__le32) +
			       CEPH_ENCODING_START_BLK_LEN;
	if (get_info_op_buf_size > PAGE_SIZE)
		return -E2BIG;

	get_info_op_page = alloc_page(GFP_NOIO);
	if (!get_info_op_page)
		return -ENOMEM;

	reply_page = alloc_page(GFP_NOIO);
	if (!reply_page) {
		__free_page(get_info_op_page);
		return -ENOMEM;
	}

	p = page_address(get_info_op_page);
	end = p + get_info_op_buf_size;

	/* encode cls_lock_get_info_op struct */
	ceph_start_encoding(&p, 1, 1,
			    get_info_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
	ceph_encode_string(&p, end, lock_name, name_len);

	dout("%s lock_name %s\n", __func__, lock_name);
	ret = ceph_osdc_call(osdc, oid, oloc, "lock", "get_info",
			     CEPH_OSD_FLAG_READ, get_info_op_page,
			     get_info_op_buf_size, reply_page, &reply_len);

	dout("%s: status %d\n", __func__, ret);
	if (ret >= 0) {
		p = page_address(reply_page);
		end = p + reply_len;

		ret = decode_lockers(&p, end, type, tag, lockers, num_lockers);
	}

	__free_page(get_info_op_page);
	__free_page(reply_page);
	return ret;
}
EXPORT_SYMBOL(ceph_cls_lock_info);
Commit	Line	Data
f66241cb DF	1	#include <linux/ceph/ceph_debug.h>
	2
	3	#include <linux/types.h>
	4	#include <linux/slab.h>
	5
	6	#include <linux/ceph/cls_lock_client.h>
	7	#include <linux/ceph/decode.h>
	8
	9	/**
	10	* ceph_cls_lock - grab rados lock for object
	11	* @oid, @oloc: object to lock
	12	* @lock_name: the name of the lock
	13	* @type: lock type (CEPH_CLS_LOCK_EXCLUSIVE or CEPH_CLS_LOCK_SHARED)
	14	* @cookie: user-defined identifier for this instance of the lock
	15	* @tag: user-defined tag
	16	* @desc: user-defined lock description
	17	* @flags: lock flags
	18	*
	19	* All operations on the same lock should use the same tag.
	20	*/
	21	int ceph_cls_lock(struct ceph_osd_client *osdc,
	22	struct ceph_object_id *oid,
	23	struct ceph_object_locator *oloc,
	24	char lock_name, u8 type, char cookie,
	25	char tag, char desc, u8 flags)
	26	{
	27	int lock_op_buf_size;
	28	int name_len = strlen(lock_name);
	29	int cookie_len = strlen(cookie);
	30	int tag_len = strlen(tag);
	31	int desc_len = strlen(desc);
	32	void p, end;
	33	struct page *lock_op_page;
	34	struct timespec mtime;
	35	int ret;
	36
	37	lock_op_buf_size = name_len + sizeof(__le32) +
	38	cookie_len + sizeof(__le32) +
	39	tag_len + sizeof(__le32) +
	40	desc_len + sizeof(__le32) +
	41	sizeof(struct ceph_timespec) +
	42	/* flag and type */
	43	sizeof(u8) + sizeof(u8) +
	44	CEPH_ENCODING_START_BLK_LEN;
	45	if (lock_op_buf_size > PAGE_SIZE)
	46	return -E2BIG;
	47
	48	lock_op_page = alloc_page(GFP_NOIO);
	49	if (!lock_op_page)
	50	return -ENOMEM;
	51
	52	p = page_address(lock_op_page);
	53	end = p + lock_op_buf_size;
	54
	55	/* encode cls_lock_lock_op struct */
	56	ceph_start_encoding(&p, 1, 1,
	57	lock_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
	58	ceph_encode_string(&p, end, lock_name, name_len);
	59	ceph_encode_8(&p, type);
	60	ceph_encode_string(&p, end, cookie, cookie_len);
	61	ceph_encode_string(&p, end, tag, tag_len);
	62	ceph_encode_string(&p, end, desc, desc_len);
	63	/* only support infinite duration */
	64	memset(&mtime, 0, sizeof(mtime));
65	ceph_encode_timespec(p, &mtime);
66	p += sizeof(struct ceph_timespec);
67	ceph_encode_8(&p, flags);
68
69	dout("%s lock_name %s type %d cookie %s tag %s desc %s flags 0x%x\n",
70	__func__, lock_name, type, cookie, tag, desc, flags);
71	ret = ceph_osdc_call(osdc, oid, oloc, "lock", "lock",
54ea0046 ID	72	CEPH_OSD_FLAG_WRITE, lock_op_page,
54ea0046 ID	73	lock_op_buf_size, NULL, NULL);
f66241cb DF	74
	75	dout("%s: status %d\n", __func__, ret);
	76	__free_page(lock_op_page);
	77	return ret;
	78	}
	79	EXPORT_SYMBOL(ceph_cls_lock);
	80
	81	/**
	82	* ceph_cls_unlock - release rados lock for object
	83	* @oid, @oloc: object to lock
	84	* @lock_name: the name of the lock
	85	* @cookie: user-defined identifier for this instance of the lock
	86	*/
	87	int ceph_cls_unlock(struct ceph_osd_client *osdc,
	88	struct ceph_object_id *oid,
	89	struct ceph_object_locator *oloc,
	90	char lock_name, char cookie)
	91	{
	92	int unlock_op_buf_size;
	93	int name_len = strlen(lock_name);
	94	int cookie_len = strlen(cookie);
	95	void p, end;
	96	struct page *unlock_op_page;
	97	int ret;
	98
	99	unlock_op_buf_size = name_len + sizeof(__le32) +
	100	cookie_len + sizeof(__le32) +
	101	CEPH_ENCODING_START_BLK_LEN;
	102	if (unlock_op_buf_size > PAGE_SIZE)
	103	return -E2BIG;
	104
	105	unlock_op_page = alloc_page(GFP_NOIO);
	106	if (!unlock_op_page)
	107	return -ENOMEM;
	108
	109	p = page_address(unlock_op_page);
	110	end = p + unlock_op_buf_size;
	111
	112	/* encode cls_lock_unlock_op struct */
	113	ceph_start_encoding(&p, 1, 1,
	114	unlock_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
	115	ceph_encode_string(&p, end, lock_name, name_len);
	116	ceph_encode_string(&p, end, cookie, cookie_len);
	117
	118	dout("%s lock_name %s cookie %s\n", __func__, lock_name, cookie);
	119	ret = ceph_osdc_call(osdc, oid, oloc, "lock", "unlock",
54ea0046 ID	120	CEPH_OSD_FLAG_WRITE, unlock_op_page,
54ea0046 ID	121	unlock_op_buf_size, NULL, NULL);
f66241cb DF	122
	123	dout("%s: status %d\n", __func__, ret);
	124	__free_page(unlock_op_page);
	125	return ret;
	126	}
	127	EXPORT_SYMBOL(ceph_cls_unlock);
	128
	129	/**
	130	* ceph_cls_break_lock - release rados lock for object for specified client
	131	* @oid, @oloc: object to lock
	132	* @lock_name: the name of the lock
	133	* @cookie: user-defined identifier for this instance of the lock
	134	* @locker: current lock owner
	135	*/
	136	int ceph_cls_break_lock(struct ceph_osd_client *osdc,
	137	struct ceph_object_id *oid,
	138	struct ceph_object_locator *oloc,
	139	char lock_name, char cookie,
	140	struct ceph_entity_name *locker)
	141	{
	142	int break_op_buf_size;
	143	int name_len = strlen(lock_name);
	144	int cookie_len = strlen(cookie);
	145	struct page *break_op_page;
	146	void p, end;
	147	int ret;
	148
	149	break_op_buf_size = name_len + sizeof(__le32) +
	150	cookie_len + sizeof(__le32) +
	151	sizeof(u8) + sizeof(__le64) +
	152	CEPH_ENCODING_START_BLK_LEN;
	153	if (break_op_buf_size > PAGE_SIZE)
	154	return -E2BIG;
	155
	156	break_op_page = alloc_page(GFP_NOIO);
	157	if (!break_op_page)
	158	return -ENOMEM;
	159
	160	p = page_address(break_op_page);
	161	end = p + break_op_buf_size;
	162
	163	/* encode cls_lock_break_op struct */
	164	ceph_start_encoding(&p, 1, 1,
	165	break_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
	166	ceph_encode_string(&p, end, lock_name, name_len);
	167	ceph_encode_copy(&p, locker, sizeof(*locker));
	168	ceph_encode_string(&p, end, cookie, cookie_len);
	169
	170	dout("%s lock_name %s cookie %s locker %s%llu\n", __func__, lock_name,
	171	cookie, ENTITY_NAME(*locker));
	172	ret = ceph_osdc_call(osdc, oid, oloc, "lock", "break_lock",
54ea0046 ID	173	CEPH_OSD_FLAG_WRITE, break_op_page,
54ea0046 ID	174	break_op_buf_size, NULL, NULL);
f66241cb DF	175
	176	dout("%s: status %d\n", __func__, ret);
	177	__free_page(break_op_page);
	178	return ret;
	179	}
	180	EXPORT_SYMBOL(ceph_cls_break_lock);
d4ed4a53	181
14bb211d ID	182	int ceph_cls_set_cookie(struct ceph_osd_client *osdc,
	183	struct ceph_object_id *oid,
	184	struct ceph_object_locator *oloc,
	185	char lock_name, u8 type, char old_cookie,
	186	char tag, char new_cookie)
	187	{
	188	int cookie_op_buf_size;
	189	int name_len = strlen(lock_name);
	190	int old_cookie_len = strlen(old_cookie);
	191	int tag_len = strlen(tag);
	192	int new_cookie_len = strlen(new_cookie);
	193	void p, end;
	194	struct page *cookie_op_page;
	195	int ret;
	196
	197	cookie_op_buf_size = name_len + sizeof(__le32) +
	198	old_cookie_len + sizeof(__le32) +
	199	tag_len + sizeof(__le32) +
	200	new_cookie_len + sizeof(__le32) +
	201	sizeof(u8) + CEPH_ENCODING_START_BLK_LEN;
	202	if (cookie_op_buf_size > PAGE_SIZE)
	203	return -E2BIG;
	204
	205	cookie_op_page = alloc_page(GFP_NOIO);
	206	if (!cookie_op_page)
	207	return -ENOMEM;
	208
	209	p = page_address(cookie_op_page);
	210	end = p + cookie_op_buf_size;
	211
	212	/* encode cls_lock_set_cookie_op struct */
	213	ceph_start_encoding(&p, 1, 1,
	214	cookie_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
	215	ceph_encode_string(&p, end, lock_name, name_len);
	216	ceph_encode_8(&p, type);
	217	ceph_encode_string(&p, end, old_cookie, old_cookie_len);
	218	ceph_encode_string(&p, end, tag, tag_len);
	219	ceph_encode_string(&p, end, new_cookie, new_cookie_len);
	220
	221	dout("%s lock_name %s type %d old_cookie %s tag %s new_cookie %s\n",
	222	__func__, lock_name, type, old_cookie, tag, new_cookie);
	223	ret = ceph_osdc_call(osdc, oid, oloc, "lock", "set_cookie",
	224	CEPH_OSD_FLAG_WRITE, cookie_op_page,
	225	cookie_op_buf_size, NULL, NULL);
	226
	227	dout("%s: status %d\n", __func__, ret);
	228	__free_page(cookie_op_page);
	229	return ret;
	230	}
	231	EXPORT_SYMBOL(ceph_cls_set_cookie);
	232
d4ed4a53 DF	233	void ceph_free_lockers(struct ceph_locker *lockers, u32 num_lockers)
	234	{
	235	int i;
	236
	237	for (i = 0; i < num_lockers; i++)
	238	kfree(lockers[i].id.cookie);
	239	kfree(lockers);
	240	}
	241	EXPORT_SYMBOL(ceph_free_lockers);
	242
	243	static int decode_locker(void *p, void end, struct ceph_locker *locker)
	244	{
	245	u8 struct_v;
	246	u32 len;
	247	char *s;
	248	int ret;
	249
	250	ret = ceph_start_decoding(p, end, 1, "locker_id_t", &struct_v, &len);
	251	if (ret)
	252	return ret;
	253
	254	ceph_decode_copy(p, &locker->id.name, sizeof(locker->id.name));
	255	s = ceph_extract_encoded_string(p, end, NULL, GFP_NOIO);
	256	if (IS_ERR(s))
	257	return PTR_ERR(s);
	258
	259	locker->id.cookie = s;
	260
	261	ret = ceph_start_decoding(p, end, 1, "locker_info_t", &struct_v, &len);
	262	if (ret)
	263	return ret;
	264
	265	p += sizeof(struct ceph_timespec); / skip expiration */
	266	ceph_decode_copy(p, &locker->info.addr, sizeof(locker->info.addr));
	267	ceph_decode_addr(&locker->info.addr);
	268	len = ceph_decode_32(p);
	269	p += len; / skip description */
	270
	271	dout("%s %s%llu cookie %s addr %s\n", __func__,
	272	ENTITY_NAME(locker->id.name), locker->id.cookie,
	273	ceph_pr_addr(&locker->info.addr.in_addr));
	274	return 0;
	275	}
	276
	277	static int decode_lockers(void *p, void end, u8 type, char *tag,
	278	struct ceph_locker *lockers, u32 num_lockers)
	279	{
	280	u8 struct_v;
	281	u32 struct_len;
	282	char *s;
	283	int i;
	284	int ret;
	285
	286	ret = ceph_start_decoding(p, end, 1, "cls_lock_get_info_reply",
	287	&struct_v, &struct_len);
	288	if (ret)
	289	return ret;
	290
	291	*num_lockers = ceph_decode_32(p);
	292	lockers = kcalloc(num_lockers, sizeof(**lockers), GFP_NOIO);
	293	if (!*lockers)
	294	return -ENOMEM;
	295
	296	for (i = 0; i < *num_lockers; i++) {
297	ret = decode_locker(p, end, *lockers + i);
298	if (ret)
299	goto err_free_lockers;
300	}
301
302	*type = ceph_decode_8(p);
303	s = ceph_extract_encoded_string(p, end, NULL, GFP_NOIO);
304	if (IS_ERR(s)) {
305	ret = PTR_ERR(s);
306	goto err_free_lockers;
307	}
308
309	*tag = s;
310	return 0;
311
312	err_free_lockers:
313	ceph_free_lockers(lockers, num_lockers);
314	return ret;
315	}
316
317	/*
318	* On success, the caller is responsible for:
319	*
320	* kfree(tag);
321	* ceph_free_lockers(lockers, num_lockers);
322	*/
323	int ceph_cls_lock_info(struct ceph_osd_client *osdc,
324	struct ceph_object_id *oid,
325	struct ceph_object_locator *oloc,
326	char lock_name, u8 type, char **tag,
327	struct ceph_locker *lockers, u32 num_lockers)
328	{
329	int get_info_op_buf_size;
330	int name_len = strlen(lock_name);
331	struct page get_info_op_page, reply_page;
2544a020	332	size_t reply_len = PAGE_SIZE;
d4ed4a53 DF	333	void p, end;
	334	int ret;
	335
	336	get_info_op_buf_size = name_len + sizeof(__le32) +
	337	CEPH_ENCODING_START_BLK_LEN;
	338	if (get_info_op_buf_size > PAGE_SIZE)
	339	return -E2BIG;
	340
	341	get_info_op_page = alloc_page(GFP_NOIO);
	342	if (!get_info_op_page)
	343	return -ENOMEM;
	344
	345	reply_page = alloc_page(GFP_NOIO);
	346	if (!reply_page) {
	347	__free_page(get_info_op_page);
	348	return -ENOMEM;
	349	}
	350
	351	p = page_address(get_info_op_page);
	352	end = p + get_info_op_buf_size;
	353
	354	/* encode cls_lock_get_info_op struct */
	355	ceph_start_encoding(&p, 1, 1,
	356	get_info_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
	357	ceph_encode_string(&p, end, lock_name, name_len);
	358
	359	dout("%s lock_name %s\n", __func__, lock_name);
	360	ret = ceph_osdc_call(osdc, oid, oloc, "lock", "get_info",
	361	CEPH_OSD_FLAG_READ, get_info_op_page,
	362	get_info_op_buf_size, reply_page, &reply_len);
	363
	364	dout("%s: status %d\n", __func__, ret);
	365	if (ret >= 0) {
	366	p = page_address(reply_page);
	367	end = p + reply_len;
	368
	369	ret = decode_lockers(&p, end, type, tag, lockers, num_lockers);
	370	}
	371
	372	__free_page(get_info_op_page);
	373	__free_page(reply_page);
	374	return ret;
	375	}
	376	EXPORT_SYMBOL(ceph_cls_lock_info);