[linux-block.git] / net / ceph / cls_lock_client.c

// SPDX-License-Identifier: GPL-2.0
#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>
#include <linux/ceph/libceph.h>

/**
 * ceph_cls_lock - grab rados lock for object
 * @osdc: OSD client instance
 * @oid: object to lock
 * @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 timespec64 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_timespec64(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
 * @osdc: OSD client instance
 * @oid: object to lock
 * @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
 * @osdc: OSD client instance
 * @oid: object to lock
 * @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 */

	ret = ceph_decode_entity_addr(p, end, &locker->info.addr);
	if (ret)
		return ret;

	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));
	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);

int ceph_cls_assert_locked(struct ceph_osd_request *req, int which,
			   char *lock_name, u8 type, char *cookie, char *tag)
{
	int assert_op_buf_size;
	int name_len = strlen(lock_name);
	int cookie_len = strlen(cookie);
	int tag_len = strlen(tag);
	struct page **pages;
	void *p, *end;
	int ret;

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

	ret = osd_req_op_cls_init(req, which, "lock", "assert_locked");
	if (ret)
		return ret;

	pages = ceph_alloc_page_vector(1, GFP_NOIO);
	if (IS_ERR(pages))
		return PTR_ERR(pages);

	p = page_address(pages[0]);
	end = p + assert_op_buf_size;

	/* encode cls_lock_assert_op struct */
	ceph_start_encoding(&p, 1, 1,
			    assert_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);
	WARN_ON(p != end);

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