Fixes for LVS export and import

lib/blob/blobstore.c

@@ -3402,37 +3402,25 @@ bs_channel_create(void *io_device, void *ctx_buf)
 	struct spdk_blob_store		*bs = io_device;
 	struct spdk_bs_channel		*channel = ctx_buf;
 	struct spdk_bs_dev		*dev;
-	uint32_t			max_ops = bs->max_channel_ops;
-	uint32_t			i;
 
 	dev = bs->dev;
 
-	channel->req_mem = calloc(max_ops, sizeof(struct spdk_bs_request_set));
-	if (!channel->req_mem) {
-		return -1;
-	}
-
-	TAILQ_INIT(&channel->reqs);
-
-	for (i = 0; i < max_ops; i++) {
-		TAILQ_INSERT_TAIL(&channel->reqs, &channel->req_mem[i], link);
-	}
-
 	channel->bs = bs;
 	channel->dev = dev;
+	bs_request_set_pool_init(&channel->req_pool);
 	channel->dev_channel = dev->create_channel(dev);
 
 	if (!channel->dev_channel) {
 		SPDK_ERRLOG("Failed to create device channel.\n");
-		free(channel->req_mem);
+		bs_request_set_pool_free(&channel->req_pool);
 		return -1;
 	}
 
 	channel->new_cluster_page = spdk_zmalloc(SPDK_BS_PAGE_SIZE, 0, NULL, SPDK_ENV_SOCKET_ID_ANY,
 				    SPDK_MALLOC_DMA);
 	if (!channel->new_cluster_page) {
 		SPDK_ERRLOG("Failed to allocate new cluster page\n");
-		free(channel->req_mem);
+		bs_request_set_pool_free(&channel->req_pool);
 		channel->dev->destroy_channel(channel->dev, channel->dev_channel);
 		return -1;
 	}
@@ -3464,7 +3452,7 @@ bs_channel_destroy(void *io_device, void *ctx_buf)
 
 	blob_esnap_destroy_bs_channel(channel);
 
-	free(channel->req_mem);
+	bs_request_set_pool_free(&channel->req_pool);
 	spdk_free(channel->new_cluster_page);
 	channel->dev->destroy_channel(channel->dev, channel->dev_channel);
 }

lib/blob/blobstore.h

@@ -203,8 +203,7 @@ struct spdk_blob_store {
 };
 
 struct spdk_bs_channel {
-	struct spdk_bs_request_set	*req_mem;
-	TAILQ_HEAD(, spdk_bs_request_set) reqs;
+	struct spdk_bs_request_set_pool	req_pool;
 
 	struct spdk_blob_store		*bs;
 

lib/blob/request.c

@@ -14,6 +14,105 @@
 
 #include "spdk/log.h"
 
+void
+bs_request_set_pool_init(struct spdk_bs_request_set_pool *req_pool)
+{
+	req_pool->req_mem = NULL;
+	TAILQ_INIT(&req_pool->reqs);
+}
+
+void
+bs_request_set_pool_free(struct spdk_bs_request_set_pool *req_pool)
+{
+	struct spdk_bs_request_set **ps = req_pool->req_mem;
+
+	if (ps != NULL) {
+		while (*ps != NULL) {
+			free(*ps);
+			++ps;
+		}
+		free(req_pool->req_mem);
+		req_pool->req_mem = NULL;
+	}
+}
+
+static int
+bs_request_set_pool_grow(struct spdk_bs_request_set_pool *req_pool,
+			 uint32_t max_ops)
+{
+	struct spdk_bs_request_set	**ps;
+	struct spdk_bs_request_set	*new_set;
+	uint32_t			i;
+
+	/* Allocate a new chunk of spdk_bs_request_set elements */
+	new_set = calloc(max_ops, sizeof(struct spdk_bs_request_set));
+	if (!new_set) {
+		return -ENOMEM;
+	}
+
+	/* Count the number of existing chunks and grow it by 1 */
+	i = 0;
+	if (req_pool->req_mem != NULL) {
+		ps = req_pool->req_mem;
+		while (*ps++ != NULL) {
+			++i;
+		}
+	}
+	++i;
+
+	ps = realloc(req_pool->req_mem, (i + 1) * sizeof(struct spdk_bs_request_set *));
+	if (!ps) {
+		free(new_set);
+		return -ENOMEM;
+	}
+	ps[i - 1] = new_set;
+	ps[i] = NULL;		/* Final NULL element indicates end of the list */
+	req_pool->req_mem = ps;
+
+	/* Append the elements from the new chunk to the list of free requests */
+	for (i = 0; i < max_ops; i++) {
+		TAILQ_INSERT_TAIL(&req_pool->reqs, &new_set[i], link);
+	}
+
+	return 0;
+}
+
+static struct spdk_bs_request_set *
+bs_request_set_alloc(struct spdk_bs_channel *channel, struct spdk_bs_cpl *cpl,
+		     struct spdk_io_channel *back_channel)
+{
+	struct spdk_bs_request_set *set = NULL;
+
+	if (channel->req_pool.req_mem != NULL) {
+		set = TAILQ_FIRST(&channel->req_pool.reqs);
+	}
+
+	if (!set) {
+		if (bs_request_set_pool_grow(&channel->req_pool,
+					     channel->bs->max_channel_ops) < 0) {
+			SPDK_ERRLOG("Failed to allocate per-channel request set pool\n");
+			return NULL;
+		}
+	}
+
+	set = TAILQ_FIRST(&channel->req_pool.reqs);
+	assert(set);
+	TAILQ_REMOVE(&channel->req_pool.reqs, set, link);
+
+	set->cpl = *cpl;
+	set->bserrno = 0;
+	set->channel = channel;
+	set->back_channel = back_channel;
+
+	return set;
+}
+
+static inline void
+bs_request_set_dealloc(struct spdk_bs_request_set *set)
+{
+	TAILQ_INSERT_TAIL(&set->channel->req_pool.reqs, set, link);
+}
+
 void
 bs_call_cpl(struct spdk_bs_cpl *cpl, int bserrno)
 {
@@ -58,8 +157,7 @@ bs_request_set_complete(struct spdk_bs_request_set *set)
 	struct spdk_bs_cpl cpl = set->cpl;
 	int bserrno = set->bserrno;
 
-	TAILQ_INSERT_TAIL(&set->channel->reqs, set, link);
-
+	bs_request_set_dealloc(set);
 	bs_call_cpl(&cpl, bserrno);
 }
 
@@ -81,16 +179,11 @@ bs_sequence_start(struct spdk_io_channel *_channel, struct spdk_bs_cpl *cpl,
 
 	channel = spdk_io_channel_get_ctx(_channel);
 	assert(channel != NULL);
-	set = TAILQ_FIRST(&channel->reqs);
+
+	set = bs_request_set_alloc(channel, cpl, back_channel);
 	if (!set) {
 		return NULL;
 	}
-	TAILQ_REMOVE(&channel->reqs, set, link);
-
-	set->cpl = *cpl;
-	set->bserrno = 0;
-	set->channel = channel;
-	set->back_channel = back_channel;
 
 	set->cb_args.cb_fn = bs_sequence_completion;
 	set->cb_args.cb_arg = set;
@@ -340,16 +433,11 @@ bs_batch_open(struct spdk_io_channel *_channel, struct spdk_bs_cpl *cpl, struct
 
 	channel = spdk_io_channel_get_ctx(_channel);
 	assert(channel != NULL);
-	set = TAILQ_FIRST(&channel->reqs);
+
+	set = bs_request_set_alloc(channel, cpl, back_channel);
 	if (!set) {
 		return NULL;
 	}
-	TAILQ_REMOVE(&channel->reqs, set, link);
-
-	set->cpl = *cpl;
-	set->bserrno = 0;
-	set->channel = channel;
-	set->back_channel = back_channel;
 
 	set->u.batch.cb_fn = NULL;
 	set->u.batch.cb_arg = NULL;
@@ -477,15 +565,11 @@ bs_user_op_alloc(struct spdk_io_channel *_channel, struct spdk_bs_cpl *cpl,
 
 	channel = spdk_io_channel_get_ctx(_channel);
 	assert(channel != NULL);
-	set = TAILQ_FIRST(&channel->reqs);
+
+	set = bs_request_set_alloc(channel, cpl, NULL);
 	if (!set) {
 		return NULL;
 	}
-	TAILQ_REMOVE(&channel->reqs, set, link);
-
-	set->cpl = *cpl;
-	set->channel = channel;
-	set->back_channel = NULL;
 	set->ext_io_opts = NULL;
 
 	args = &set->u.user_op;
@@ -541,7 +625,7 @@ bs_user_op_execute(spdk_bs_user_op_t *op)
 					set->ext_io_opts);
 		break;
 	}
-	TAILQ_INSERT_TAIL(&set->channel->reqs, set, link);
+	bs_request_set_dealloc(set);
 }
 
 void
@@ -552,7 +636,7 @@ bs_user_op_abort(spdk_bs_user_op_t *op, int bserrno)
 	set = (struct spdk_bs_request_set *)op;
 
 	set->cpl.u.blob_basic.cb_fn(set->cpl.u.blob_basic.cb_arg, bserrno);
-	TAILQ_INSERT_TAIL(&set->channel->reqs, set, link);
+	bs_request_set_dealloc(set);
 }
 
 SPDK_LOG_REGISTER_COMPONENT(blob_rw)

lib/blob/request.h

@@ -127,6 +127,16 @@ struct spdk_bs_request_set {
 	TAILQ_ENTRY(spdk_bs_request_set) link;
 };
 
+/* A pool of pre-allocated requested sets. */
+struct spdk_bs_request_set_pool {
+	struct spdk_bs_request_set		**req_mem;
+	TAILQ_HEAD(, spdk_bs_request_set)	reqs;
+};
+
+void bs_request_set_pool_init(struct spdk_bs_request_set_pool *req_pool);
+
+void bs_request_set_pool_free(struct spdk_bs_request_set_pool *req_pool);
+
 void bs_call_cpl(struct spdk_bs_cpl *cpl, int bserrno);
 
 spdk_bs_sequence_t *bs_sequence_start_bs(struct spdk_io_channel *channel,

module/bdev/lvol/vbdev_lvol.c

@@ -1735,12 +1735,35 @@ vbdev_lvs_examine_config(struct spdk_bdev *bdev)
 	spdk_bdev_module_examine_done(&g_lvol_if);
 }
 
+static void _vbdev_lvs_open_next_lvol(struct spdk_lvol *lvol, void *cb_arg);
+
+static void
+_vbdev_lvs_open_next_lvol_cb(void *cb_arg, struct spdk_lvol *lvol, int lvolerrno)
+{
+	struct spdk_lvol *next_lvol = TAILQ_NEXT(lvol, link);
+
+	if (lvolerrno != 0) {
+		SPDK_ERRLOG("Failed to open lvol %s (blob 0x%" PRIx64 ")\n", lvol->name, lvol->blob_id);
+	}
+
+	_vbdev_lvs_examine_finish(cb_arg, lvol, lvolerrno);
+
+	if (next_lvol) {
+		_vbdev_lvs_open_next_lvol(next_lvol, cb_arg);
+	}
+}
+
+static void
+_vbdev_lvs_open_next_lvol(struct spdk_lvol *lvol, void *cb_arg)
+{
+	spdk_lvol_open(lvol, _vbdev_lvs_open_next_lvol_cb, cb_arg);
+}
+
 static void
 _vbdev_lvs_examine_cb(void *arg, struct spdk_lvol_store *lvol_store, int lvserrno)
 {
 	struct lvol_store_bdev *lvs_bdev;
 	struct spdk_lvs_with_handle_req *req = (struct spdk_lvs_with_handle_req *)arg;
-	struct spdk_lvol *lvol, *tmp;
 	struct spdk_lvs_req *ori_req = req->cb_arg;
 
 	if (lvserrno == -EEXIST) {
@@ -1790,9 +1813,7 @@ _vbdev_lvs_examine_cb(void *arg, struct spdk_lvol_store *lvol_store, int lvserrn
 		_vbdev_lvs_examine_done(ori_req, 0);
 	} else {
 		/* Open all lvols */
-		TAILQ_FOREACH_SAFE(lvol, &lvol_store->lvols, link, tmp) {
-			spdk_lvol_open(lvol, _vbdev_lvs_examine_finish, ori_req);
-		}
+		_vbdev_lvs_open_next_lvol(TAILQ_FIRST(&lvol_store->lvols), ori_req);
 	}
 
 end:

test/unit/lib/blob/blob.c/blob_ut.c

@@ -1536,7 +1536,7 @@ bs_channel_get_req_count(struct spdk_io_channel *_channel)
 	struct spdk_bs_request_set *set;
 	uint32_t count = 0;
 
-	TAILQ_FOREACH(set, &channel->reqs, link) {
+	TAILQ_FOREACH(set, &channel->req_pool.reqs, link) {
 		count++;
 	}