[PATCH RFC 0/6] Support rados locking

[PATCH RFC 0/6] Support rados locking

[Douglas Fuller]

Add decoding support for some commonly exchanged data structures.
Adapt Mike's code to serve as a generic client for cls::lock in userspace.

Discussion points:
Is this how we want to represent object classes in the kernel client?

This defines ceph_locker to contain ceph_locker_id and ceph_locker_info
from userspace. Is that what we want for adaptation of userspace data
structures (in this case std::map<locker_id_t, locker_info_t>)?

There are quite a few strings in these data structures. Do we want
to add utility functions to properly free them at some level?

Douglas Fuller (4):
  auth.c: added ceph_entity_name_encode
  osd_client: added single object method call
  cls_lock: add rados locking
  cls_lock: add support for lock_info

Mike Christie (2):
  Add support for userspace ceph DECODE_START.
  ceph: add start/finish encoding helpers

 include/linux/ceph/auth.h       |   2 +
 include/linux/ceph/cls_lock.h   |  39 +++++
 include/linux/ceph/decode.h     |  78 ++++++++++
 include/linux/ceph/osd_client.h |   6 +
 net/ceph/Makefile               |   1 +
 net/ceph/auth.c                 |  11 ++
 net/ceph/cls_lock.c             | 329 ++++++++++++++++++++++++++++++++++++++++
 net/ceph/osd_client.c           |  44 ++++++
 8 files changed, 510 insertions(+)
 create mode 100644 include/linux/ceph/cls_lock.h
 create mode 100644 net/ceph/cls_lock.c

-- 
1.9.3

[PATCH RFC 1/6] Add support for userspace ceph DECODE_START.

[Douglas Fuller]

From: Mike Christie <michaelc@cs.wisc.edu>

Signed-off-by: Mike Christie <michaelc@cs.wisc.edu>
---
 include/linux/ceph/decode.h | 23 +++++++++++++++++++++++
 1 file changed, 23 insertions(+)

diff --git a/include/linux/ceph/decode.h b/include/linux/ceph/decode.h
index a6ef9cc..b55c7fd 100644
--- a/include/linux/ceph/decode.h
+++ b/include/linux/ceph/decode.h
@@ -255,5 +255,28 @@ static inline void ceph_encode_string(void **p, void *end,
 		ceph_encode_string(p, end, s, n);		\
 	} while (0)
 
+/**
+ * ceph_start_decoding - start a decoding block
+ * @p: buffer to decode
+ * @end: end of buffer
+ * @curr_ver: current version of the encoding that the code supports/encode
+ * @len: buffer to return len of data in buffer
+ */
+static inline int ceph_start_decoding(void **p, void *end, u8 curr_ver,
+				      u32 *len)
+{
+	u8 struct_ver, struct_compat;
+
+	ceph_decode_8_safe(p, end, struct_ver, fail);
+	ceph_decode_8_safe(p, end, struct_compat, fail);
+
+	if (curr_ver < struct_compat)
+		return -EINVAL;
+
+	ceph_decode_32_safe(p, end, *len, fail);
+	return 0;
+fail:
+	return -ERANGE;
+}
 
 #endif
-- 
1.9.3

[PATCH RFC 2/6] ceph: add start/finish encoding helpers

[Douglas Fuller]

From: Mike Christie <michaelc@cs.wisc.edu>

This patch adds helpers to encode/decode the starting blocks
locking code. They are the equivalent of ENCODE_START and
DECODE_START_LEGACY_COMPAT_LEN in the userspace ceph code.

Signed-off-by: Mike Christie <michaelc@cs.wisc.edu>
[djf: added fixup from mailing list]
Signed-off-by: Douglas Fuller <dfuller@redhat.com>
---
 include/linux/ceph/decode.h | 55 +++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 55 insertions(+)

diff --git a/include/linux/ceph/decode.h b/include/linux/ceph/decode.h
index b55c7fd..c81f096 100644
--- a/include/linux/ceph/decode.h
+++ b/include/linux/ceph/decode.h
@@ -217,6 +217,61 @@ static inline void ceph_encode_string(void **p, void *end,
 	*p += len;
 }
 
+/*
+ * version and length starting block encoders/decoders
+ */
+
+/* current code version (u8) + compat code version (u8) + len of struct (u32) */
+#define CEPH_ENCODING_START_BLK_LEN 6
+
+/**
+ * ceph_start_encoding - start encoding block
+ * @p: buffer to encode data in
+ * @curr_ver: current (code) version of the encoding
+ * @compat_ver: oldest code version that can decode it
+ * @len: length of data that will be encoded in buffer
+ */
+static inline void ceph_start_encoding(void **p, u8 curr_ver, u8 compat_ver,
+				       u32 len)
+{
+	ceph_encode_8(p, curr_ver);
+	ceph_encode_8(p, compat_ver);
+	ceph_encode_32(p, len);
+}
+
+/**
+ * ceph_start_decoding_compat - decode block with legacy support for older schemes
+ * @p: buffer to decode
+ * @end: end of decode buffer
+ * @curr_ver: current version of the encoding that the code supports/encode
+ * @compat_ver: oldest version that includes a __u8 compat version field
+ * @len_ver: oldest version that includes a __u32 length wrapper
+ * @len: buffer to return len of data in buffer
+ */
+static inline int ceph_start_decoding_compat(void **p, void *end, u8 curr_ver,
+					    u8 compat_ver, u8 len_ver, u32 *len)
+{
+	u8 struct_ver, struct_compat;
+
+	ceph_decode_8_safe(p, end, struct_ver, fail);
+	if (struct_ver >= compat_ver) {
+		ceph_decode_8_safe(p, end, struct_compat, fail);
+		if (curr_ver < struct_compat)
+			return -EINVAL;
+	}
+
+	if (struct_ver >= len_ver) {
+		ceph_decode_32_safe(p, end, *len, fail);
+	} else {
+		*len = 0;
+	}
+
+	return 0;
+fail:
+	return -ERANGE;
+}
+
+
 #define ceph_encode_need(p, end, n, bad)			\
 	do {							\
 		if (!likely(ceph_has_room(p, end, n)))		\
-- 
1.9.3

[PATCH RFC 3/6] auth.c: added ceph_entity_name_encode

[Douglas Fuller]

Signed-off-by: Douglas Fuller <dfuller@redhat.com>
---
 include/linux/ceph/auth.h |  2 ++
 net/ceph/auth.c           | 11 +++++++++++
 2 files changed, 13 insertions(+)

diff --git a/include/linux/ceph/auth.h b/include/linux/ceph/auth.h
index 260d78b..7687918 100644
--- a/include/linux/ceph/auth.h
+++ b/include/linux/ceph/auth.h
@@ -104,6 +104,8 @@ extern int ceph_handle_auth_reply(struct ceph_auth_client *ac,
 				  void *buf, size_t len,
 				  void *reply_buf, size_t reply_len);
 extern int ceph_entity_name_encode(const char *name, void **p, void *end);
+extern int ceph_entity_name_decode(struct ceph_entity_name *name, void **p,
+				   void *end);
 
 extern int ceph_build_auth(struct ceph_auth_client *ac,
 		    void *msg_buf, size_t msg_len);
diff --git a/net/ceph/auth.c b/net/ceph/auth.c
index 6b923bc..50baee6 100644
--- a/net/ceph/auth.c
+++ b/net/ceph/auth.c
@@ -94,6 +94,17 @@ int ceph_entity_name_encode(const char *name, void **p, void *end)
 	return 0;
 }
 
+int ceph_entity_name_decode(struct ceph_entity_name *name, void **p, void *end)
+{
+	ceph_decode_8_safe(p, end, name->type, bad);
+	ceph_decode_64_safe(p, end, name->num, bad);
+
+	return 0;
+
+bad:
+	return -ERANGE;
+}
+
 /*
  * Initiate protocol negotiation with monitor.  Include entity name
  * and list supported protocols.
-- 
1.9.3

[PATCH RFC 4/6] osd_client: added single object method call

[Douglas Fuller]

Add a convenience function to osd_client to call class ops. The interface
assumes that request and reply data each consist of single pages.

Signed-off-by: Douglas Fuller <dfuller@redhat.com>
---
 include/linux/ceph/osd_client.h |  6 ++++++
 net/ceph/osd_client.c           | 44 +++++++++++++++++++++++++++++++++++++++++
 2 files changed, 50 insertions(+)

diff --git a/include/linux/ceph/osd_client.h b/include/linux/ceph/osd_client.h
index 7fbbd22..40561ff 100644
--- a/include/linux/ceph/osd_client.h
+++ b/include/linux/ceph/osd_client.h
@@ -362,6 +362,12 @@ extern struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *,
 				      u32 truncate_seq, u64 truncate_size,
 				      bool use_mempool);
 
+extern int ceph_osd_op_cls_call(struct ceph_osd_client *osdc, int poolid,
+				char *obj_name, char *class, char *method,
+				int flags, struct page **req_data,
+				size_t req_len, struct page **resp_data,
+				size_t *resp_len);
+
 extern void ceph_osdc_set_request_linger(struct ceph_osd_client *osdc,
 					 struct ceph_osd_request *req);
 
diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index da451eb..9db3657 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -898,6 +898,50 @@ struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
 }
 EXPORT_SYMBOL(ceph_osdc_new_request);
 
+int ceph_osd_op_cls_call(struct ceph_osd_client *osdc, int poolid,
+			 char *obj_name, char *class, char *method, int flags,
+			 struct page **req_data, size_t req_len,
+			 struct page **resp_data, size_t *resp_len)
+{
+	struct ceph_osd_request *osd_req;
+	int ret = -ENOMEM;
+	struct timespec tm = CURRENT_TIME;
+
+	osd_req = ceph_osdc_alloc_request(osdc, NULL, 1, false, GFP_NOIO);
+	if (!osd_req)
+		return ret;
+	osd_req->r_flags = flags;
+	osd_req->r_base_oloc.pool = poolid;
+	ceph_oid_set_name(&osd_req->r_base_oid, obj_name);
+	osd_req_op_cls_init(osd_req, 0, CEPH_OSD_OP_CALL, class, method);
+
+	if (req_data)
+		osd_req_op_cls_request_data_pages(osd_req, 0, req_data,
+						  req_len, 0, false, false);
+	if (resp_data)
+		osd_req_op_cls_response_data_pages(osd_req, 0, resp_data,
+						   PAGE_SIZE, 0, false, false);
+
+	ceph_osdc_build_request(osd_req, 0, NULL, CEPH_NOSNAP, &tm);
+
+	ret = ceph_osdc_start_request(osdc, osd_req, false);
+	if (ret)
+		goto out;
+
+	ret = ceph_osdc_wait_request(osdc, osd_req);
+	if (ret < 0)
+		goto out;
+
+	if (resp_data)
+		*resp_len = osd_req->r_reply_op_len[0];
+
+	ret = osd_req->r_reply_op_result[0];
+out:
+	ceph_osdc_put_request(osd_req);
+	return ret;
+}
+EXPORT_SYMBOL(ceph_osd_op_cls_call);
+
 /*
  * We keep osd requests in an rbtree, sorted by ->r_tid.
  */
-- 
1.9.3

[PATCH RFC 5/6] cls_lock: add rados locking

[Douglas Fuller]

This patch adds support for rados lock, unlock and break lock.
This will be used to sync up scsi pr info manipulation and
TMF execution.

Signed-off-by: Mike Christie <michaelc@cs.wisc.edu>
[djf: broke out lock functions and moved to new source file]
[djf: changed interface to make use of osd_client utility functions]
[djf: snipped paragraph from commit message related to rbd code not moved]
Signed-off-by: Douglas Fuller <dfuller@redhat.com>
---
 include/linux/ceph/cls_lock.h |  12 +++
 net/ceph/Makefile             |   1 +
 net/ceph/cls_lock.c           | 173 ++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 186 insertions(+)
 create mode 100644 include/linux/ceph/cls_lock.h
 create mode 100644 net/ceph/cls_lock.c

diff --git a/include/linux/ceph/cls_lock.h b/include/linux/ceph/cls_lock.h
new file mode 100644
index 0000000..2c24eb2
--- /dev/null
+++ b/include/linux/ceph/cls_lock.h
@@ -0,0 +1,12 @@
+#ifndef _NET_CEPH_RADOS_LOCK_H
+#define _NET_CEPH_RADOS_LOCK_H
+
+int ceph_cls_lock(struct ceph_osd_client *osdc, int poolid, char *obj_name,
+		  char *lock_name, u8 type, char *cookie, char *tag, char *desc,
+		  u8 flags);
+int ceph_cls_unlock(struct ceph_osd_client *osdc, int poolid, char *obj_name,
+		    char *lock_name, char *cookie);
+int ceph_cls_break_lock(struct ceph_osd_client *osdc, int poolid,
+			char *obj_name, char *lock_name, u8 entity_type,
+			u64 num, char *cookie);
+#endif
diff --git a/net/ceph/Makefile b/net/ceph/Makefile
index 958d9856..f76ee18 100644
--- a/net/ceph/Makefile
+++ b/net/ceph/Makefile
@@ -5,6 +5,7 @@ obj-$(CONFIG_CEPH_LIB) += libceph.o
 
 libceph-y := ceph_common.o messenger.o msgpool.o buffer.o pagelist.o \
 	mon_client.o \
+	cls_lock.o \
 	osd_client.o osdmap.o crush/crush.o crush/mapper.o crush/hash.o \
 	debugfs.o \
 	auth.o auth_none.o \
diff --git a/net/ceph/cls_lock.c b/net/ceph/cls_lock.c
new file mode 100644
index 0000000..4eff868
--- /dev/null
+++ b/net/ceph/cls_lock.c
@@ -0,0 +1,173 @@
+#include <linux/ceph/ceph_debug.h>
+
+#include <linux/types.h>
+#include <linux/slab.h>
+
+#include <linux/ceph/decode.h>
+#include <linux/ceph/messenger.h>
+#include <linux/ceph/msgpool.h>
+#include <linux/ceph/osd_client.h>
+
+/**
+ * ceph_cls_lock - grab rados lock for object
+ *  @osdc, @poolid, @obj_name: object to lock
+ *  @lock_name: the name of the lock
+ *  @type: lock type (RADOS_LOCK_EXCLUSIVE or RADOS_LOCK_SHARED)
+ *  @cookie: user-defined identifier for this instance of the lock
+ *  @tag: if RADOS_LOCK_SHARED, tag of the lock. NULL if non shared.
+ *  @desc: user-defined lock description
+ *  @flags: lock flags
+ */
+int ceph_cls_lock(struct ceph_osd_client *osdc, int poolid, char *obj_name,
+		  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(mtime) +
+				/* flag and type */
+				sizeof(u8) + sizeof(u8) +
+				CEPH_ENCODING_START_BLK_LEN;
+	BUG_ON(lock_op_buf_size > PAGE_SIZE);
+	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;
+
+	ceph_start_encoding(&p, 1, 1,
+			    lock_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
+	/* encode cls_lock_lock_op struct */
+	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: %s %d %s %s %s %d\n", __func__,
+	     lock_name, type, cookie, tag, desc, flags);
+
+	ret = ceph_osd_op_cls_call(osdc, poolid, obj_name, "lock", "lock",
+				   CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
+				   &lock_op_page, lock_op_buf_size,
+				   NULL, 0);
+	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, @poolid, @obj_name: 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, int poolid, char *obj_name,
+		    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;
+	BUG_ON(unlock_op_buf_size > PAGE_SIZE);
+	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;
+
+	ceph_start_encoding(&p, 1, 1,
+			    unlock_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
+	/* encode cls_lock_unlock_op struct */
+	ceph_encode_string(&p, end, lock_name, name_len);
+	ceph_encode_string(&p, end, cookie, cookie_len);
+
+	dout("%s: %s %s\n", __func__, lock_name, cookie);
+	ret = ceph_osd_op_cls_call(osdc, poolid, obj_name, "lock", "unlock",
+				   CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
+				   &unlock_op_page, unlock_op_buf_size,
+				   NULL, 0);
+
+	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, @poolid, @obj_name: object to lock
+ *  @lock_name: the name of the lock
+ *  @entity_type: ceph entity type (CEPH_ENTITY_TYPE_*)
+ *  @num: ceph entity id
+ *  @cookie: user-defined identifier for this instance of the lock
+ */
+int ceph_cls_break_lock(struct ceph_osd_client *osdc, int poolid,
+			char *obj_name, char *lock_name, u8 entity_type,
+			u64 num, char *cookie)
+{
+	int break_lock_op_buf_size;
+	int name_len = strlen(lock_name);
+	int cookie_len = strlen(cookie);
+	struct page *break_lock_op_page;
+	void *p, *end;
+	int ret;
+
+	break_lock_op_buf_size = name_len + sizeof(__le32) +
+				 cookie_len + sizeof(__le32) +
+				 sizeof(u8) + sizeof(__le64) +
+				 CEPH_ENCODING_START_BLK_LEN;
+	BUG_ON(break_lock_op_buf_size > PAGE_SIZE);
+	break_lock_op_page = alloc_page(GFP_NOIO);
+	if (!break_lock_op_page)
+		return -ENOMEM;
+
+	p = page_address(break_lock_op_page);
+	end = p + break_lock_op_buf_size;
+
+	ceph_start_encoding(&p, 1, 1,
+			break_lock_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
+	/* encode cls_lock_break_op struct */
+	ceph_encode_string(&p, end, lock_name, name_len);
+	ceph_encode_8(&p, entity_type);
+	ceph_encode_64(&p, num);
+	ceph_encode_string(&p, end, cookie, cookie_len);
+
+	dout("%s: lock %s entity_type %hu id %llu cookie %s\n",
+	     __func__, lock_name, entity_type, num, cookie);
+
+	ret = ceph_osd_op_cls_call(osdc, poolid, obj_name, "lock", "break_lock",
+				   CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
+				   &break_lock_op_page, break_lock_op_buf_size,
+				   NULL, 0);
+	dout("%s: status %d\n", __func__, ret);
+	__free_page(break_lock_op_page);
+	return ret;
+}
+EXPORT_SYMBOL(ceph_cls_break_lock);
-- 
1.9.3

[PATCH RFC 6/6] cls_lock: add support for lock_info

[Douglas Fuller]

Add an interface for the lock.lock_info method and associated data
structures.

Based heavily on Mike Christie's code originally authored for the previous
commit.

Signed-off-by: Douglas Fuller <dfuller@redhat.com>
---
 include/linux/ceph/cls_lock.h |  27 ++++++++
 net/ceph/cls_lock.c           | 156 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 183 insertions(+)

diff --git a/include/linux/ceph/cls_lock.h b/include/linux/ceph/cls_lock.h
index 2c24eb2..d6d3636 100644
--- a/include/linux/ceph/cls_lock.h
+++ b/include/linux/ceph/cls_lock.h
@@ -1,6 +1,33 @@
 #ifndef _NET_CEPH_RADOS_LOCK_H
 #define _NET_CEPH_RADOS_LOCK_H
 
+#include <linux/ceph/types.h>
+#include <linux/ceph/msgpool.h>
+#include <linux/ceph/messenger.h>
+#include <linux/ceph/osdmap.h>
+#include <linux/ceph/osd_client.h>
+
+struct ceph_locker_id {
+	struct ceph_entity_name name;
+	size_t cookie_len;
+	char *cookie;
+};
+
+struct ceph_locker_info {
+	struct timespec ts;
+	struct ceph_entity_addr addr;
+	size_t desc_len;
+	char *desc;
+};
+
+struct ceph_locker {
+	struct ceph_locker_id id;
+	struct ceph_locker_info info;
+};
+
+int ceph_cls_lock_info(struct ceph_osd_client *osdc, int poolid, char *obj_name,
+		       char * lock_name, int *num_lockers,
+		       struct ceph_locker **lockers, u8 *lock_type, char **tag);
 int ceph_cls_lock(struct ceph_osd_client *osdc, int poolid, char *obj_name,
 		  char *lock_name, u8 type, char *cookie, char *tag, char *desc,
 		  u8 flags);
diff --git a/net/ceph/cls_lock.c b/net/ceph/cls_lock.c
index 4eff868..415a41b 100644
--- a/net/ceph/cls_lock.c
+++ b/net/ceph/cls_lock.c
@@ -3,11 +3,167 @@
 #include <linux/types.h>
 #include <linux/slab.h>
 
+#include <linux/ceph/cls_lock.h>
+#include <linux/ceph/auth.h>
 #include <linux/ceph/decode.h>
 #include <linux/ceph/messenger.h>
 #include <linux/ceph/msgpool.h>
 #include <linux/ceph/osd_client.h>
 
+static int __decode_locker(struct ceph_locker *locker, void **p, void *end)
+{
+	/*
+	 * struct cls_lock_get_info_reply {
+	 * 	map {
+	 * 		struct locker_id_t {
+	 * 			struct entity_name_t {
+	 * 				__u8 type;
+	 * 				int64_t num;
+	 * 			}
+	 * 			string cookie;
+	 * 		}
+	 * 		struct locker_info_t {
+	 * 			struct timespec ts;
+	 *			struct ceph_entity_addr addr;
+	 *			string description;
+	 *		}
+	 *	}
+	 *	uint8_t lock_type;
+	 *	string tag;
+	 * }
+	 */
+	int ret;
+	u32 len;
+
+	/* locker_id_t */
+	ret = ceph_start_decoding_compat(p, end, 1, 1, 1, &len);
+	if (ret)
+		return ret;
+
+	ret = ceph_entity_name_decode(&locker->id.name, p, end);
+	if (ret)
+		return ret;
+
+	locker->id.cookie = ceph_extract_encoded_string(p, end,
+						        &locker->id.cookie_len,
+						        GFP_NOIO);
+	if (IS_ERR(locker->id.cookie))
+		return PTR_ERR(locker->id.cookie);
+
+	/* locker_info_t */
+	ret = ceph_start_decoding_compat(p, end, 1, 1, 1, &len);
+	if (ret)
+		goto free_cookie;
+
+	ceph_decode_timespec(&locker->info.ts, *p);
+	*p += sizeof(struct ceph_timespec);
+
+	ret = -ERANGE;
+	ceph_decode_copy_safe(p, end, &locker->info.addr,
+			      sizeof(locker->info.addr), free_cookie);
+	ceph_decode_addr(&locker->info.addr);
+
+	locker->info.desc = ceph_extract_encoded_string(p, end,
+							&locker->info.desc_len,
+							GFP_NOIO);
+	if (IS_ERR(locker->info.desc)) {
+		ret = PTR_ERR(locker->info.desc);
+		goto free_cookie;
+	}
+
+	return 0;
+
+free_cookie:
+	kfree(locker->id.cookie);
+	return ret;
+}
+	
+int ceph_cls_lock_info(struct ceph_osd_client *osdc, int poolid, char *obj_name,
+		       char *lock_name, int *num_lockers,
+		       struct ceph_locker **lockers, u8 *lock_type, char **tag)
+{
+	int get_info_op_buf_size;
+	int name_len = strlen(lock_name);
+	struct page *get_info_page;
+	struct page *reply_page;
+	size_t reply_len;
+	int len;
+	u32 num;
+	void *p, *end;
+	int ret;
+	int i;
+
+	get_info_op_buf_size = name_len + sizeof(__le32) +
+			       CEPH_ENCODING_START_BLK_LEN;
+	BUG_ON(get_info_op_buf_size > PAGE_SIZE);
+
+	get_info_page = alloc_page(GFP_NOIO);
+	if (!get_info_page)
+		return -ENOMEM;
+
+	reply_page = alloc_page(GFP_NOIO);
+	if (!reply_page) {
+		__free_page(get_info_page);
+		return -ENOMEM;
+	}
+
+	p = page_address(get_info_page);
+	end = p + get_info_op_buf_size;
+
+	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 info for %s on object %s\n",
+	     __func__, lock_name, obj_name);
+
+	ret = ceph_osd_op_cls_call(osdc, poolid, obj_name, "lock", "get_info",
+				   CEPH_OSD_FLAG_READ | CEPH_OSD_FLAG_ACK,
+				   &get_info_page, get_info_op_buf_size,
+				   &reply_page, &reply_len);
+
+	dout("%s: status %d\n", __func__, ret);
+	if (ret < 0)
+		goto err;
+
+	p = page_address(reply_page);
+	end = p + reply_len;
+
+	ret = ceph_start_decoding_compat(&p, end, 1, 1, 1, &len);
+	if (ret)
+		goto err;
+
+	ret = -ERANGE;
+	ceph_decode_32_safe(&p, end, num, err);
+	*num_lockers = (int)num;
+
+	*lockers = kzalloc(num * sizeof(**lockers), GFP_NOIO);
+	if (!*lockers) {
+		ret = -ENOMEM;
+		goto err;
+	}
+
+	for (i = 0; i < num; i++) {
+		ret = __decode_locker(*lockers + i, &p, end);
+		if (ret)
+			goto free_lockers;
+	}
+
+	ceph_decode_8_safe(&p, end, *lock_type, free_lockers);
+	*tag = ceph_extract_encoded_string(&p, end, NULL, GFP_NOIO);
+
+	return 0;
+
+free_lockers:
+	kfree(*lockers);
+err:
+	__free_page(get_info_page);
+	__free_page(reply_page);
+	return ret;
+}
+EXPORT_SYMBOL(ceph_cls_lock_info);
+
 /**
  * ceph_cls_lock - grab rados lock for object
  *  @osdc, @poolid, @obj_name: object to lock
-- 
1.9.3