Commit ddd4603c authored by Yankun Li's avatar Yankun Li Committed by Konrad Sztyber
Browse files

bdev/compress: Supports the specified compression algorithm 



compress bdev Indicates that compression uses the synchronization
mechanism and only deflate is supported. However, for some business
scenarios, it is more concerned with compression performance than
compression rate. So we provide extensions that support multiple
compression algorithms

Change-Id: I36b93875eed3069aa0c74843d98e4a0c90412b9e
Signed-off-by: default avatarYankun Li <845245370@qq.com>
Reviewed-on: https://review.spdk.io/gerrit/c/spdk/spdk/+/24452


Reviewed-by: default avatarGangCao <gang.cao@intel.com>
Community-CI: Broadcom CI <spdk-ci.pdl@broadcom.com>
Community-CI: Mellanox Build Bot
Reviewed-by: default avatarKonrad Sztyber <konrad.sztyber@intel.com>
Reviewed-by: default avatarAleksey Marchuk <alexeymar@nvidia.com>
Tested-by: default avatarSPDK CI Jenkins <sys_sgci@intel.com>
parent a649646c

doc/jsonrpc.md

+5 −1
Original line number Diff line number Diff line
@@ -2915,6 +2915,8 @@ Name | Optional | Type | Description 
base_bdev_name          | Required | string      | Name of the base bdev

pm_path                 | Required | string      | Path to persistent memory

lb_size                 | Optional | int         | Compressed vol logical block size (512 or 4096)

comp_algo               | Optional | string      | Compression algorithm for the compressed vol. Default is deflate

comp_level              | Optional | int         | Compression algorithm level for the compressed vol. Default is 1



#### Result
@@ -2929,7 +2931,9 @@ Example request: 
  "params": {

    "base_bdev_name": "Nvme0n1",

    "pm_path": "/pm_files",

    "lb_size": 4096

    "lb_size": 4096,

    "comp_algo": "deflate",

    "comp_level": 1

  },

  "jsonrpc": "2.0",

  "method": "bdev_compress_create",

module/bdev/compress/vbdev_compress.c

+60 −14
Original line number Diff line number Diff line
@@ -21,7 +21,6 @@ 


#include "spdk/accel_module.h"





#define CHUNK_SIZE (1024 * 16)

#define COMP_BDEV_NAME "compress"

#define BACKING_IO_SZ (4 * 1024)
@@ -58,6 +57,8 @@ struct vbdev_compress { 
	TAILQ_HEAD(, vbdev_comp_op)	queued_comp_ops;

	TAILQ_ENTRY(vbdev_compress)	link;

	struct spdk_thread		*thread;	/* thread where base device is opened */

	enum spdk_accel_comp_algo       comp_algo;      /* compression algorithm for compress bdev */

	uint32_t                        comp_level;     /* compression algorithm level */

};

static TAILQ_HEAD(, vbdev_compress) g_vbdev_comp = TAILQ_HEAD_INITIALIZER(g_vbdev_comp);
@@ -80,7 +81,8 @@ struct comp_bdev_io { 
static void vbdev_compress_examine(struct spdk_bdev *bdev);

static int vbdev_compress_claim(struct vbdev_compress *comp_bdev);

static void vbdev_compress_queue_io(struct spdk_bdev_io *bdev_io);

struct vbdev_compress *_prepare_for_load_init(struct spdk_bdev_desc *bdev_desc, uint32_t lb_size);

struct vbdev_compress *_prepare_for_load_init(struct spdk_bdev_desc *bdev_desc, uint32_t lb_size,

		uint8_t comp_algo, uint32_t comp_level);

static void vbdev_compress_submit_request(struct spdk_io_channel *ch, struct spdk_bdev_io *bdev_io);

static void comp_bdev_ch_destroy_cb(void *io_device, void *ctx_buf);

static void vbdev_compress_delete_done(void *cb_arg, int bdeverrno);
@@ -134,13 +136,16 @@ _compress_operation(struct spdk_reduce_backing_dev *backing_dev, struct iovec *s 


	if (compress) {

		assert(dst_iovcnt == 1);

		rc = spdk_accel_submit_compress(comp_bdev->accel_channel, dst_iovs[0].iov_base, dst_iovs[0].iov_len,

						src_iovs, src_iovcnt, &reduce_cb_arg->output_size,

						reduce_cb_arg->cb_fn, reduce_cb_arg->cb_arg);

		rc = spdk_accel_submit_compress_ext(comp_bdev->accel_channel, dst_iovs[0].iov_base,

						    dst_iovs[0].iov_len, src_iovs, src_iovcnt,

						    comp_bdev->comp_algo, comp_bdev->comp_level,

						    &reduce_cb_arg->output_size, reduce_cb_arg->cb_fn,

						    reduce_cb_arg->cb_arg);

	} else {

		rc = spdk_accel_submit_decompress(comp_bdev->accel_channel, dst_iovs, dst_iovcnt,

						  src_iovs, src_iovcnt, &reduce_cb_arg->output_size,

						  reduce_cb_arg->cb_fn, reduce_cb_arg->cb_arg);

		rc = spdk_accel_submit_decompress_ext(comp_bdev->accel_channel, dst_iovs, dst_iovcnt,

						      src_iovs, src_iovcnt, comp_bdev->comp_algo,

						      &reduce_cb_arg->output_size, reduce_cb_arg->cb_fn,

						      reduce_cb_arg->cb_arg);

	}



	return rc;
@@ -775,7 +780,8 @@ vbdev_compress_base_bdev_event_cb(enum spdk_bdev_event_type type, struct spdk_bd 
 * information for reducelib to init or load.

 */

struct vbdev_compress *

_prepare_for_load_init(struct spdk_bdev_desc *bdev_desc, uint32_t lb_size)

_prepare_for_load_init(struct spdk_bdev_desc *bdev_desc, uint32_t lb_size, uint8_t comp_algo,

		       uint32_t comp_level)

{

	struct vbdev_compress *comp_bdev;

	struct spdk_bdev *bdev;
@@ -799,6 +805,10 @@ _prepare_for_load_init(struct spdk_bdev_desc *bdev_desc, uint32_t lb_size) 


	comp_bdev->backing_dev.user_ctx_size = sizeof(struct spdk_bdev_io_wait_entry);



	comp_bdev->comp_algo = comp_algo;

	comp_bdev->comp_level = comp_level;

	comp_bdev->params.comp_algo = comp_algo;

	comp_bdev->params.comp_level = comp_level;

	comp_bdev->params.chunk_size = CHUNK_SIZE;

	if (lb_size == 0) {

		comp_bdev->params.logical_block_size = bdev->blocklen;
@@ -812,8 +822,8 @@ _prepare_for_load_init(struct spdk_bdev_desc *bdev_desc, uint32_t lb_size) 


/* Call reducelib to initialize a new volume */

static int

vbdev_init_reduce(const char *bdev_name, const char *pm_path, uint32_t lb_size,

		  bdev_compress_create_cb cb_fn, void *cb_arg)

vbdev_init_reduce(const char *bdev_name, const char *pm_path, uint32_t lb_size, uint8_t comp_algo,

		  uint32_t comp_level, bdev_compress_create_cb cb_fn, void *cb_arg)

{

	struct spdk_bdev_desc *bdev_desc = NULL;

	struct vbdev_init_reduce_ctx *init_ctx;
@@ -837,7 +847,7 @@ vbdev_init_reduce(const char *bdev_name, const char *pm_path, uint32_t lb_size, 
		return rc;

	}



	comp_bdev = _prepare_for_load_init(bdev_desc, lb_size);

	comp_bdev = _prepare_for_load_init(bdev_desc, lb_size, comp_algo, comp_level);

	if (comp_bdev == NULL) {

		free(init_ctx);

		spdk_bdev_close(bdev_desc);
@@ -930,13 +940,40 @@ comp_bdev_ch_destroy_cb(void *io_device, void *ctx_buf) 
	pthread_mutex_unlock(&comp_bdev->reduce_lock);

}



static int

_check_compress_bdev_comp_algo(enum spdk_accel_comp_algo algo, uint32_t comp_level)

{

	uint32_t min_level, max_level;

	int rc;



	rc = spdk_accel_get_compress_level_range(algo, &min_level, &max_level);

	if (rc != 0) {

		return rc;

	}



	/* If both min_level and max_level are 0, the compression level can be ignored.

	 * The back-end implementation hardcodes the compression level.

	 */

	if (min_level == 0 && max_level == 0) {

		return 0;

	}



	if (comp_level > max_level || comp_level < min_level) {

		return -EINVAL;

	}



	return 0;

}



/* RPC entry point for compression vbdev creation. */

int

create_compress_bdev(const char *bdev_name, const char *pm_path, uint32_t lb_size,

		     uint8_t comp_algo, uint32_t comp_level,

		     bdev_compress_create_cb cb_fn, void *cb_arg)

{

	struct vbdev_compress *comp_bdev = NULL;

	struct stat info;

	int rc;



	if (stat(pm_path, &info) != 0) {

		SPDK_ERRLOG("PM path %s does not exist.\n", pm_path);
@@ -951,13 +988,20 @@ create_compress_bdev(const char *bdev_name, const char *pm_path, uint32_t lb_siz 
		return -EINVAL;

	}



	rc = _check_compress_bdev_comp_algo(comp_algo, comp_level);

	if (rc != 0) {

		SPDK_ERRLOG("Compress bdev doesn't support compression algo(%u) or level(%u)\n",

			    comp_algo, comp_level);

		return rc;

	}



	TAILQ_FOREACH(comp_bdev, &g_vbdev_comp, link) {

		if (strcmp(bdev_name, comp_bdev->base_bdev->name) == 0) {

			SPDK_ERRLOG("Bass bdev %s already being used for a compress bdev\n", bdev_name);

			return -EBUSY;

		}

	}

	return vbdev_init_reduce(bdev_name, pm_path, lb_size, cb_fn, cb_arg);

	return vbdev_init_reduce(bdev_name, pm_path, lb_size, comp_algo, comp_level, cb_fn, cb_arg);

}



static int
@@ -1243,6 +1287,8 @@ vbdev_reduce_load_cb(void *cb_arg, struct spdk_reduce_vol *vol, int reduce_errno 
		comp_bdev->vol = vol;

		memcpy(&comp_bdev->params, spdk_reduce_vol_get_params(vol),

		       sizeof(struct spdk_reduce_vol_params));

		comp_bdev->comp_algo = comp_bdev->params.comp_algo;

		comp_bdev->comp_level = comp_bdev->params.comp_level;

	}



	comp_bdev->reduce_errno = reduce_errno;
@@ -1279,7 +1325,7 @@ vbdev_compress_examine(struct spdk_bdev *bdev) 
		return;

	}



	comp_bdev = _prepare_for_load_init(bdev_desc, 0);

	comp_bdev = _prepare_for_load_init(bdev_desc, 0, SPDK_ACCEL_COMP_ALGO_DEFLATE, 1);

	if (comp_bdev == NULL) {

		spdk_bdev_close(bdev_desc);

		spdk_bdev_module_examine_done(&compress_if);

module/bdev/compress/vbdev_compress.h

+3 −0
Original line number Diff line number Diff line
@@ -55,11 +55,14 @@ typedef void (*spdk_delete_compress_complete)(void *cb_arg, int bdeverrno); 
 * \param bdev_name Bdev on which compression bdev will be created.

 * \param pm_path Path to persistent memory.

 * \param lb_size Logical block size for the compressed volume in bytes. Must be 4K or 512.

 * \param comp_algo compression algorithm for the compressed volume.

 * \param comp_level compression algorithm level for the compressed volume.

 * \param cb_fn Function to call after creation.

 * \param cb_arg Argument to pass to cb_fn.

 * \return 0 on success, other on failure.

 */

int create_compress_bdev(const char *bdev_name, const char *pm_path, uint32_t lb_size,

			 uint8_t comp_algo, uint32_t comp_level,

			 bdev_compress_create_cb cb_fn, void *cb_arg);



/**

module/bdev/compress/vbdev_compress_rpc.c

+33 −3
Original line number Diff line number Diff line
@@ -76,8 +76,35 @@ struct rpc_construct_compress { 
	char *base_bdev_name;

	char *pm_path;

	uint32_t lb_size;

	enum spdk_accel_comp_algo comp_algo;

	uint32_t comp_level;

};



static int

rpc_decode_comp_algo(const struct spdk_json_val *val, void *out)

{

	enum spdk_accel_comp_algo *algo = out;

	char *name = NULL;

	int rc;



	rc = spdk_json_decode_string(val, &name);

	if (rc != 0) {

		return rc;

	}



	if (strcmp(name, "deflate") == 0) {

		*algo = SPDK_ACCEL_COMP_ALGO_DEFLATE;

	} else if (strcmp(name, "lz4") == 0) {

		*algo = SPDK_ACCEL_COMP_ALGO_LZ4;

	} else {

		rc = -EINVAL;

	}



	free(name);



	return rc;

}



struct rpc_bdev_compress_create_ctx {

	struct rpc_construct_compress req;

	struct spdk_jsonrpc_request *request;
@@ -104,6 +131,8 @@ static const struct spdk_json_object_decoder rpc_construct_compress_decoders[] = 
	{"base_bdev_name", offsetof(struct rpc_construct_compress, base_bdev_name), spdk_json_decode_string},

	{"pm_path", offsetof(struct rpc_construct_compress, pm_path), spdk_json_decode_string},

	{"lb_size", offsetof(struct rpc_construct_compress, lb_size), spdk_json_decode_uint32, true},

	{"comp_algo", offsetof(struct rpc_construct_compress, comp_algo), rpc_decode_comp_algo, true},

	{"comp_level", offsetof(struct rpc_construct_compress, comp_level), spdk_json_decode_uint32, true},

};



static void
@@ -147,7 +176,8 @@ rpc_bdev_compress_create(struct spdk_jsonrpc_request *request, 
	}



	req = &ctx->req;



	req->comp_algo = SPDK_ACCEL_COMP_ALGO_DEFLATE;

	req->comp_level = 1;

	if (spdk_json_decode_object(params, rpc_construct_compress_decoders,

				    SPDK_COUNTOF(rpc_construct_compress_decoders),

				    req)) {
@@ -157,8 +187,8 @@ rpc_bdev_compress_create(struct spdk_jsonrpc_request *request, 
		goto cleanup;

	}



	rc = create_compress_bdev(req->base_bdev_name, req->pm_path, req->lb_size,

				  rpc_bdev_compress_create_cb, ctx);

	rc = create_compress_bdev(req->base_bdev_name, req->pm_path, req->lb_size, req->comp_algo,

				  req->comp_level, rpc_bdev_compress_create_cb, ctx);

	if (rc != 0) {

		if (rc == -EBUSY) {

			spdk_jsonrpc_send_error_response(request, rc, "Base bdev already in use for compression.");

python/spdk/rpc/bdev.py

+7 −1
Original line number Diff line number Diff line
@@ -47,12 +47,14 @@ def bdev_wait_for_examine(client): 
    return client.call('bdev_wait_for_examine')





def bdev_compress_create(client, base_bdev_name, pm_path, lb_size=None):

def bdev_compress_create(client, base_bdev_name, pm_path, lb_size=None, comp_algo=None, comp_level=None):

    """Construct a compress virtual block device.

    Args:

        base_bdev_name: name of the underlying base bdev

        pm_path: path to persistent memory

        lb_size: logical block size for the compressed vol in bytes.  Must be 4K or 512.

        comp_algo: compression algorithm for the compressed vol. Default is deflate.

        comp_level: compression algorithm level for the compressed vol. Default is 1.

    Returns:

        Name of created virtual block device.

    """
@@ -61,6 +63,10 @@ def bdev_compress_create(client, base_bdev_name, pm_path, lb_size=None): 
    params['pm_path'] = pm_path

    if lb_size is not None:

        params['lb_size'] = lb_size

    if comp_algo is not None:

        params['comp_algo'] = comp_algo

    if comp_level is not None:

        params['comp_level'] = comp_level

    return client.call('bdev_compress_create', params)