Commit f66241cb authored by Douglas Fuller's avatar Douglas Fuller Committed by Ilya Dryomov
Browse files

libceph: support for advisory locking on RADOS objects 



This patch adds support for rados lock, unlock and break lock.

Based heavily on code by Mike Christie <michaelc@cs.wisc.edu>.

Signed-off-by: default avatarDouglas Fuller <dfuller@redhat.com>
Signed-off-by: default avatarIlya Dryomov <idryomov@gmail.com>
Reviewed-by: default avatarMike Christie <mchristi@redhat.com>
Reviewed-by: default avatarAlex Elder <elder@linaro.org>
parent 428a7158

include/linux/ceph/cls_lock_client.h

0 → 100644
+27 −0
Original line number Diff line number Diff line 
#ifndef _LINUX_CEPH_CLS_LOCK_CLIENT_H

#define _LINUX_CEPH_CLS_LOCK_CLIENT_H



#include <linux/ceph/osd_client.h>



enum ceph_cls_lock_type {

	CEPH_CLS_LOCK_NONE = 0,

	CEPH_CLS_LOCK_EXCLUSIVE = 1,

	CEPH_CLS_LOCK_SHARED = 2,

};



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 ceph_cls_unlock(struct ceph_osd_client *osdc,

		    struct ceph_object_id *oid,

		    struct ceph_object_locator *oloc,

		    char *lock_name, char *cookie);

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



#endif

net/ceph/Makefile

+1 −0
Original line number Diff line number Diff line
@@ -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_client.o \

	osd_client.o osdmap.o crush/crush.o crush/mapper.o crush/hash.o \

	debugfs.o \

	auth.o auth_none.o \

net/ceph/cls_lock_client.c

0 → 100644
+180 −0
Original line number Diff line number Diff line 
#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 | CEPH_OSD_FLAG_ONDISK,

			     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 | CEPH_OSD_FLAG_ONDISK,

			     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 | CEPH_OSD_FLAG_ONDISK,

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

CRA Git | Maintained and supported by SUSTech CRA and CCSE