Merge pull request #75 from robertbaldyga/asynchronous-cache-attach

Implement asynchronous attach, load and stop

inc/ocf_cache.h

@@ -266,12 +266,4 @@ void ocf_cache_set_priv(ocf_cache_t cache, void *priv);
  */
 void *ocf_cache_get_priv(ocf_cache_t cache);
 
-/**
- * @brief Set queue to be used during flush operation
- *
- * @param[in] cache Cache object
- * @param[in] queue Queue object
- */
-void ocf_cache_set_flush_queue(ocf_cache_t cache, ocf_queue_t queue);
-
 #endif /* __OCF_CACHE_H__ */

inc/ocf_mngt.h

@@ -356,6 +356,17 @@ struct ocf_mngt_cache_device_config {
 int ocf_mngt_cache_start(ocf_ctx_t ctx, ocf_cache_t *cache,
 		struct ocf_mngt_cache_config *cfg);
 
+/**
+ * @brief Set queue to be used during management operations
+ *
+ * @param[in] cache Cache object
+ * @param[in] queue Queue object
+ *
+ * @retval 0 Success
+ * @retval Non-zero Error occurred
+ */
+int ocf_mngt_cache_set_mngt_queue(ocf_cache_t cache, ocf_queue_t queue);
+
 /**
  * @brief Completion callback of cache stop operation
  *
@@ -376,6 +387,19 @@ typedef void (*ocf_mngt_cache_stop_end_t)(ocf_cache_t cache,
 void ocf_mngt_cache_stop(ocf_cache_t cache,
 		ocf_mngt_cache_stop_end_t cmpl, void *priv);
 
+/**
+ * @brief Get amount of free RAM needed to attach cache volume
+ *
+ * @param[in] cache Cache handle
+ * @param[in] cfg Caching device configuration
+ * @param[out] ram_needed Amount of RAM needed in bytes
+ *
+ * @retval 0 Success
+ * @retval Non-zero Error occurred
+ */
+int ocf_mngt_get_ram_needed(ocf_cache_t cache,
+		struct ocf_mngt_cache_device_config *cfg, uint64_t *ram_needed);
+
 /**
  * @brief Completion callback of cache attach operation
  *

src/engine/engine_common.c

@@ -504,9 +504,10 @@ void ocf_engine_push_req_back(struct ocf_request *req, bool allow_sync)
 
 	env_spinlock_unlock_irqrestore(&q->io_list_lock, lock_flags);
 
-	if (!req->info.internal)
+	if (!req->info.internal) {
 		env_atomic_set(&cache->last_access_ms,
 				env_ticks_to_msecs(env_get_tick_count()));
+	}
 
 	ocf_queue_kick(q, allow_sync);
 }
@@ -517,6 +518,7 @@ void ocf_engine_push_req_front(struct ocf_request *req, bool allow_sync)
 	ocf_queue_t q = NULL;
 	unsigned long lock_flags = 0;
 
+	ENV_BUG_ON(!req->io_queue);
 	INIT_LIST_HEAD(&req->list);
 
 	q = req->io_queue;
@@ -528,9 +530,10 @@ void ocf_engine_push_req_front(struct ocf_request *req, bool allow_sync)
 
 	env_spinlock_unlock_irqrestore(&q->io_list_lock, lock_flags);
 
-	if (!req->info.internal)
+	if (!req->info.internal) {
 		env_atomic_set(&cache->last_access_ms,
 				env_ticks_to_msecs(env_get_tick_count()));
+	}
 
 	ocf_queue_kick(q, allow_sync);
 }

src/metadata/metadata.c

@@ -103,20 +103,20 @@ ocf_cache_line_t ocf_metadata_get_pages_count(struct ocf_cache *cache)
 	return cache->metadata.iface.pages(cache);
 }
 
-ocf_cache_line_t
+ocf_cache_line_t ocf_metadata_get_cachelines_count(ocf_cache_t cache)
-ocf_metadata_get_cachelines_count(struct ocf_cache *cache)
 {
 	return cache->metadata.iface.cachelines(cache);
 }
 
-int ocf_metadata_flush_all(struct ocf_cache *cache)
+void ocf_metadata_flush_all(ocf_cache_t cache,
+		ocf_metadata_end_t cmpl, void *priv)
 {
 	int result;
 
 	OCF_METADATA_LOCK_WR();
 	result = cache->metadata.iface.flush_all(cache);
 	OCF_METADATA_UNLOCK_WR();
-	return result;
+	cmpl(priv, result);
 }
 
 void ocf_metadata_flush(struct ocf_cache *cache, ocf_cache_line_t line)
@@ -124,19 +124,24 @@ void ocf_metadata_flush(struct ocf_cache *cache, ocf_cache_line_t line)
 	cache->metadata.iface.flush(cache, line);
 }
 
-int ocf_metadata_load_all(struct ocf_cache *cache)
+void ocf_metadata_load_all(ocf_cache_t cache,
+		ocf_metadata_end_t cmpl, void *priv)
 {
 	int result;
 
 	OCF_METADATA_LOCK_WR();
 	result = cache->metadata.iface.load_all(cache);
 	OCF_METADATA_UNLOCK_WR();
-	return result;
+	cmpl(priv, result);
 }
 
-int ocf_metadata_load_recovery(struct ocf_cache *cache)
+void ocf_metadata_load_recovery(ocf_cache_t cache,
+		ocf_metadata_end_t cmpl, void *priv)
 {
-	return cache->metadata.iface.load_recovery(cache);
+	int result;
+
+	result = cache->metadata.iface.load_recovery(cache);
+	cmpl(priv, result);
 }
 
 void ocf_metadata_flush_mark(struct ocf_cache *cache, struct ocf_request *req,

src/metadata/metadata.h

@@ -124,6 +124,8 @@ static inline void ocf_metadata_status_bits_unlock(
 #define OCF_METADATA_FLUSH_UNLOCK() \
 		ocf_metadata_flush_unlock(cache)
 
+typedef void (*ocf_metadata_end_t)(void *priv, int error);
+
 #include "metadata_cleaning_policy.h"
 #include "metadata_eviction_policy.h"
 #include "metadata_partition.h"
@@ -224,10 +226,12 @@ ocf_cache_line_t ocf_metadata_get_pages_count(struct ocf_cache *cache);
 /**
  * @brief Flush metadata
  *
- * @param cache
+ * @param cache - Cache instance
- * @return 0 - Operation success otherwise failure
+ * @param cmpl - Completion callback
+ * @param priv - Completion context
  */
-int ocf_metadata_flush_all(struct ocf_cache *cache);
+void ocf_metadata_flush_all(ocf_cache_t cache,
+		ocf_metadata_end_t cmpl, void *priv);
 
 
 /**
@@ -263,17 +267,21 @@ void ocf_metadata_flush_do_asynch(struct ocf_cache *cache,
  * @brief Load metadata
  *
  * @param cache - Cache instance
- * @return 0 - Operation success otherwise failure
+ * @param cmpl - Completion callback
+ * @param priv - Completion context
  */
-int ocf_metadata_load_all(struct ocf_cache *cache);
+void ocf_metadata_load_all(ocf_cache_t cache,
+		ocf_metadata_end_t cmpl, void *priv);
 
 /**
  * @brief Load metadata required for recovery procedure
  *
  * @param cache Cache instance
- * @return 0 - Operation success otherwise failure
+ * @param cmpl - Completion callback
+ * @param priv - Completion context
  */
-int ocf_metadata_load_recovery(struct ocf_cache *cache);
+void ocf_metadata_load_recovery(ocf_cache_t cache,
+		ocf_metadata_end_t cmpl, void *priv);
 
 /*
  * NOTE Hash table is specific for hash table metadata service implementation

src/metadata/metadata_superblock.h

@@ -62,17 +62,24 @@ struct ocf_superblock_runtime {
 	uint32_t cleaning_thread_access;
 };
 
-static inline int ocf_metadata_set_shutdown_status(
+static inline void ocf_metadata_set_shutdown_status(ocf_cache_t cache,
-		struct ocf_cache *cache,
+		enum ocf_metadata_shutdown_status shutdown_status,
-		enum ocf_metadata_shutdown_status shutdown_status)
+		ocf_metadata_end_t cmpl, void *priv)
 {
-	return cache->metadata.iface.set_shutdown_status(cache,
+	int result;
+
+	result = cache->metadata.iface.set_shutdown_status(cache,
 			shutdown_status);
+	cmpl(priv, result);
 }
 
-static inline int ocf_metadata_load_superblock(struct ocf_cache *cache)
+static inline void ocf_metadata_load_superblock(ocf_cache_t cache,
+		ocf_metadata_end_t cmpl, void *priv)
 {
-	return cache->metadata.iface.load_superblock(cache);
+	int result;
+
+	result = cache->metadata.iface.load_superblock(cache);
+	cmpl(priv, result);
 }
 
 static inline

src/mngt/ocf_mngt_common.c

@@ -15,6 +15,7 @@
 #include "../eviction/ops.h"
 #include "../ocf_logger_priv.h"
 #include "../ocf_queue_priv.h"
+#include "../engine/engine_common.h"
 
 /* Close if opened */
 int cache_mng_core_close(ocf_cache_t cache, ocf_core_id_t core_id)
@@ -457,3 +458,83 @@ int ocf_mngt_cache_visit_reverse(ocf_ctx_t ocf_ctx,
 
 	return result;
 }
+
+struct ocf_mngt_pipeline {
+	ocf_mngt_pipeline_step_t *steps;
+	int next_step;
+	ocf_mngt_pipeline_end_t cmpl;
+	struct ocf_request *req;
+	bool finish;
+	int error;
+
+	void *priv;
+};
+
+static int _ocf_mngt_pipeline_run_step(struct ocf_request *req)
+{
+	ocf_mngt_pipeline_t pipeline = req->priv;
+
+	if (pipeline->steps[pipeline->next_step] && !pipeline->finish)
+		pipeline->steps[pipeline->next_step++](pipeline, pipeline->priv);
+	else
+		pipeline->cmpl(pipeline, pipeline->priv, pipeline->error);
+
+	return 0;
+}
+
+static const struct ocf_io_if _io_if_pipeline = {
+	.read = _ocf_mngt_pipeline_run_step,
+	.write = _ocf_mngt_pipeline_run_step,
+};
+
+int ocf_mngt_pipeline_create(ocf_mngt_pipeline_t *pipeline, ocf_cache_t cache,
+		ocf_mngt_pipeline_step_t *steps, ocf_mngt_pipeline_end_t cmpl,
+		void *priv)
+{
+	ocf_mngt_pipeline_t tmp_pipeline;
+	struct ocf_request *req;
+
+	tmp_pipeline = env_vzalloc(sizeof(struct ocf_mngt_pipeline));
+	if (!tmp_pipeline)
+		return -OCF_ERR_NO_MEM;
+
+	req = ocf_req_new(cache->mngt_queue, NULL, 0, 0, 0);
+	if (!req) {
+		env_vfree(tmp_pipeline);
+		return -OCF_ERR_NO_MEM;
+	}
+
+	tmp_pipeline->steps = steps;
+	tmp_pipeline->next_step = 0;
+	tmp_pipeline->cmpl = cmpl;
+	tmp_pipeline->req = req;
+	tmp_pipeline->finish = false;
+	tmp_pipeline->error = 0;
+	tmp_pipeline->priv = priv;
+
+	req->info.internal = true;
+	req->io_if = &_io_if_pipeline;
+	req->priv = tmp_pipeline;
+
+	*pipeline = tmp_pipeline;
+
+	return 0;
+}
+
+void ocf_mngt_pipeline_destroy(ocf_mngt_pipeline_t pipeline)
+{
+	ocf_req_put(pipeline->req);
+	env_vfree(pipeline);
+}
+
+void ocf_mngt_pipeline_next(ocf_mngt_pipeline_t pipeline)
+{
+	ocf_engine_push_req_front(pipeline->req, true);
+}
+
+void ocf_mngt_pipeline_finish(ocf_mngt_pipeline_t pipeline, int error)
+{
+	pipeline->finish = true;
+	pipeline->error = error;
+	ocf_engine_push_req_front(pipeline->req, true);
+}

src/mngt/ocf_mngt_common.h

@@ -30,4 +30,22 @@ int ocf_mngt_add_partition_to_cache(struct ocf_cache *cache,
 
 bool ocf_mngt_is_cache_locked(ocf_cache_t cache);
 
+typedef struct ocf_mngt_pipeline *ocf_mngt_pipeline_t;
+
+typedef void (*ocf_mngt_pipeline_step_t)(ocf_mngt_pipeline_t pipeline,
+		void *priv);
+
+typedef void (*ocf_mngt_pipeline_end_t)(ocf_mngt_pipeline_t pipeline,
+		void *priv, int error);
+
+int ocf_mngt_pipeline_create(ocf_mngt_pipeline_t *pipeline, ocf_cache_t cache,
+		ocf_mngt_pipeline_step_t *steps, ocf_mngt_pipeline_end_t cmpl,
+		void *priv);
+
+void ocf_mngt_pipeline_destroy(ocf_mngt_pipeline_t pipeline);
+
+void ocf_mngt_pipeline_next(ocf_mngt_pipeline_t pipeline);
+
+void ocf_mngt_pipeline_finish(ocf_mngt_pipeline_t pipeline, int error);
+
 #endif /* __OCF_MNGT_COMMON_H__ */

src/mngt/ocf_mngt_flush.c

@@ -318,7 +318,7 @@ static int _ocf_mngt_flush_containers(ocf_cache_t cache,
 		fctbl[i].attribs.cache_line_lock = true;
 		fctbl[i].attribs.cmpl_context = &fctbl[i];
 		fctbl[i].attribs.cmpl_fn = _ocf_mngt_flush_end;
-		fctbl[i].attribs.io_queue = cache->flush_queue;
+		fctbl[i].attribs.io_queue = cache->mngt_queue;
 		fctbl[i].cache = cache;
 		fctbl[i].progress = &progress;
 		fctbl[i].error = &error;
@@ -482,7 +482,7 @@ void ocf_mngt_cache_flush(ocf_cache_t cache, bool interruption,
 		return;
 	}
 
-	if (!cache->flush_queue) {
+	if (!cache->mngt_queue) {
 		ocf_cache_log(cache, log_err,
 				"Cannot flush cache - no flush queue set\n");
 		cmpl(cache, priv, -OCF_ERR_INVAL);
@@ -549,7 +549,7 @@ void ocf_mngt_core_flush(ocf_core_t core, bool interruption,
 		return;
 	}
 
-	if (!cache->flush_queue) {
+	if (!cache->mngt_queue) {
 		ocf_core_log(core, log_err,
 				"Cannot flush core - no flush queue set\n");
 		cmpl(core, priv, -OCF_ERR_INVAL);
@@ -577,7 +577,7 @@ void ocf_mngt_cache_purge(ocf_cache_t cache,
 
 	OCF_CHECK_NULL(cache);
 
-	if (!cache->flush_queue) {
+	if (!cache->mngt_queue) {
 		ocf_cache_log(cache, log_err,
 				"Cannot purge cache - no flush queue set\n");
 		cmpl(cache, priv, -OCF_ERR_INVAL);
@@ -618,7 +618,7 @@ void ocf_mngt_core_purge(ocf_core_t core,
 	cache = ocf_core_get_cache(core);
 	core_id = ocf_core_get_id(core);
 
-	if (!cache->flush_queue) {
+	if (!cache->mngt_queue) {
 		ocf_core_log(core, log_err,
 				"Cannot purge core - no flush queue set\n");
 		cmpl(core, priv, -OCF_ERR_INVAL);

src/ocf_cache.c

@@ -242,17 +242,3 @@ void *ocf_cache_get_priv(ocf_cache_t cache)
 	OCF_CHECK_NULL(cache);
 	return cache->priv;
 }
-
-void ocf_cache_set_flush_queue(ocf_cache_t cache, ocf_queue_t queue)
-{
-	OCF_CHECK_NULL(cache);
-	OCF_CHECK_NULL(queue);
-
-	if (cache->flush_queue)
-		ocf_queue_put(cache->flush_queue);
-
-	if (queue)
-		ocf_queue_get(queue);
-
-	cache->flush_queue = queue;
-}

src/ocf_cache_priv.h

@@ -186,7 +186,7 @@ struct ocf_cache {
 	env_atomic pending_eviction_clines;
 
 	struct list_head io_queues;
-	ocf_queue_t flush_queue;
+	ocf_queue_t mngt_queue;
 
 	uint16_t ocf_core_inactive_count;
 	struct ocf_core core[OCF_CORE_MAX];

src/utils/utils_io.c

@@ -11,171 +11,194 @@
 #include "utils_io.h"
 #include "utils_cache_line.h"
 
-struct ocf_submit_io_wait_context {
+struct ocf_submit_volume_context {
-	env_completion complete;
-	int error;
 	env_atomic req_remaining;
+	int error;
+	ocf_submit_end_t cmpl;
+	void *priv;
 };
 
-/*
+static void _ocf_volume_flush_end(struct ocf_io *io, int error)
- * IO discard context
- */
-struct discard_io_request {
-	void *context;
-	env_atomic req_remaining;
-	env_completion completion;
-	int error;
-};
-
-static void _ocf_volume_flush_end(struct ocf_io *io, int err)
 {
-	struct ocf_submit_io_wait_context *cntx = io->priv1;
+	ocf_submit_end_t cmpl = io->priv1;
-	cntx->error = err;
-	env_completion_complete(&cntx->complete);
 
+	cmpl(io->priv2, error);
 	ocf_io_put(io);
 }
 
-int ocf_submit_volume_flush_wait(ocf_volume_t volume)
+void ocf_submit_volume_flush(ocf_volume_t volume,
+		ocf_submit_end_t cmpl, void *priv)
 {
-	struct ocf_submit_io_wait_context cntx = { };
 	struct ocf_io *io;
 
-	env_atomic_set(&cntx.req_remaining, 1);
-	env_completion_init(&cntx.complete);
-
 	io = ocf_volume_new_io(volume);
-	if (!io)
+	if (!io) {
-		return -ENOMEM;
+		cmpl(priv, -OCF_ERR_NO_MEM);
+		return;
+	}
 
 	ocf_io_configure(io, 0, 0, OCF_WRITE, 0, 0);
-	ocf_io_set_cmpl(io, &cntx, NULL, _ocf_volume_flush_end);
+	ocf_io_set_cmpl(io, cmpl, priv, _ocf_volume_flush_end);
 
 	ocf_volume_submit_flush(io);
-
-	env_completion_wait(&cntx.complete);
-
-	return cntx.error;
 }
 
-static void ocf_submit_volume_discard_wait_io(struct ocf_io *io, int error)
+static void ocf_submit_volume_end(struct ocf_io *io, int error)
 {
-	struct ocf_submit_io_wait_context *cntx = io->priv1;
+	struct ocf_submit_volume_context *context = io->priv1;
 
 	if (error)
-		cntx->error = error;
+		context->error = error;
 
-	ocf_io_put(io); /* Release IO */
+	ocf_io_put(io);
 
-	if (env_atomic_dec_return(&cntx->req_remaining))
+	if (env_atomic_dec_return(&context->req_remaining))
 		return;
 
-	/* All discard IO handled, signal it by setting completion */
+	context->cmpl(context->priv, context->error);
-	env_completion_complete(&cntx->complete);
+	env_vfree(context);
 }
 
-int ocf_submit_volume_discard_wait(ocf_volume_t volume, uint64_t addr,
+void ocf_submit_volume_discard(ocf_volume_t volume, uint64_t addr,
-		uint64_t length)
+		uint64_t length, ocf_submit_end_t cmpl, void *priv)
 {
-	struct ocf_submit_io_wait_context cntx = { };
+	struct ocf_submit_volume_context *context;
 	uint64_t bytes;
 	uint64_t max_length = (uint32_t)~0;
+	struct ocf_io *io;
 
-	ENV_BUG_ON(env_memset(&cntx, sizeof(cntx), 0));
+	context = env_vzalloc(sizeof(*context));
-	env_atomic_set(&cntx.req_remaining, 1);
+	if (!context) {
-	env_completion_init(&cntx.complete);
+		cmpl(priv, -OCF_ERR_NO_MEM);
+		return;
+	}
+
+	env_atomic_set(&context->req_remaining, 1);
+	context->cmpl = cmpl;
+	context->priv = priv;
 
 	while (length) {
-		struct ocf_io *io = ocf_volume_new_io(volume);
+		io = ocf_volume_new_io(volume);
-
 		if (!io) {
-			cntx.error = -ENOMEM;
+			context->error = -OCF_ERR_NO_MEM;
 			break;
 		}
 
+		env_atomic_inc(&context->req_remaining);
+
 		bytes = OCF_MIN(length, max_length);
 
-		env_atomic_inc(&cntx.req_remaining);
-
 		ocf_io_configure(io, addr, bytes, OCF_WRITE, 0, 0);
-		ocf_io_set_cmpl(io, &cntx, NULL,
+		ocf_io_set_cmpl(io, context, NULL, ocf_submit_volume_end);
-				ocf_submit_volume_discard_wait_io);
 		ocf_volume_submit_discard(io);
 
 		addr += bytes;
 		length -= bytes;
 	}
 
-	if (env_atomic_dec_return(&cntx.req_remaining) == 0)
+	if (env_atomic_dec_return(&context->req_remaining))
-		env_completion_complete(&cntx.complete);
+		return;
 
-	env_completion_wait(&cntx.complete);
+	cmpl(priv, context->error);
-
+	env_vfree(context);
-	return cntx.error;
 }
 
-static void ocf_submit_volume_zeroes_wait_io(struct ocf_io *io, int error)
+void ocf_submit_write_zeros(ocf_volume_t volume, uint64_t addr,
+		uint64_t length, ocf_submit_end_t cmpl, void *priv)
 {
-	struct ocf_submit_io_wait_context *cntx = io->priv1;
+	struct ocf_submit_volume_context *context;
-
-	if (error)
-		cntx->error = error;
-
-	env_completion_complete(&cntx->complete);
-}
-
-int ocf_submit_write_zeroes_wait(ocf_volume_t volume, uint64_t addr,
-		uint64_t length)
-{
-	struct ocf_submit_io_wait_context cntx = { };
 	uint32_t bytes;
 	uint32_t max_length = ~((uint32_t)PAGE_SIZE - 1);
-	uint32_t step = 0;
 	struct ocf_io *io;
 
-	io = ocf_volume_new_io(volume);
+	context = env_vzalloc(sizeof(*context));
-	if (!io)
+	if (!context) {
-		return -ENOMEM;
+		cmpl(priv, -OCF_ERR_NO_MEM);
+		return;
+	}
+
+	env_atomic_set(&context->req_remaining, 1);
+	context->cmpl = cmpl;
+	context->priv = priv;
 
 	while (length) {
-		env_completion_init(&cntx.complete);
+		io = ocf_volume_new_io(volume);
+		if (!io) {
+			context->error = -OCF_ERR_NO_MEM;
+			break;
+		}
+
+		env_atomic_inc(&context->req_remaining);
 
 		bytes = OCF_MIN(length, max_length);
 
 		ocf_io_configure(io, addr, bytes, OCF_WRITE, 0, 0);
-		ocf_io_set_cmpl(io, &cntx, NULL,
+		ocf_io_set_cmpl(io, context, NULL, ocf_submit_volume_end);
-				ocf_submit_volume_zeroes_wait_io);
 		ocf_volume_submit_write_zeroes(io);
 
 		addr += bytes;
 		length -= bytes;
-
-		env_completion_wait(&cntx.complete);
-		if (cntx.error)
-			break;
-
-		OCF_COND_RESCHED_DEFAULT(step);
 	}
 
-	ocf_io_put(io);
+	if (env_atomic_dec_return(&context->req_remaining))
+		return;
 
-	return cntx.error;
+	cmpl(priv, context->error);
+	env_vfree(context);
 }
 
-int ocf_submit_cache_page(struct ocf_cache *cache, uint64_t addr,
+struct ocf_submit_cache_page_context {
-		int dir, void *buffer)
+	ocf_cache_t cache;
+	void *buffer;
+	ocf_submit_end_t cmpl;
+	void *priv;
+};
+
+static void ocf_submit_cache_page_end(struct ocf_io *io, int error)
 {
+	struct ocf_submit_cache_page_context *context = io->priv1;
+	ctx_data_t *data = ocf_io_get_data(io);
+
+	if (io->dir == OCF_READ) {
+		ctx_data_rd_check(context->cache->owner, context->buffer,
+				data, PAGE_SIZE);
+	}
+
+	context->cmpl(context->priv, error);
+	ctx_data_free(context->cache->owner, data);
+	env_vfree(context);
+	ocf_io_put(io);
+}
+
+void ocf_submit_cache_page(ocf_cache_t cache, uint64_t addr, int dir,
+		void *buffer, ocf_submit_end_t cmpl, void *priv)
+{
+	struct ocf_submit_cache_page_context *context;
 	ctx_data_t *data;
 	struct ocf_io *io;
 	int result = 0;
 
-	/* Allocate resources for IO */
+	context = env_vmalloc(sizeof(*context));
-	io = ocf_volume_new_io(&cache->device->volume);
+	if (!context) {
-	data = ctx_data_alloc(cache->owner, 1);
+		cmpl(priv, -OCF_ERR_NO_MEM);
+		return;
+	}
 
-	if (!io || !data) {
+	context->cache = cache;
-		result = -ENOMEM;
+	context->buffer = buffer;
-		goto end;
+	context->cmpl = cmpl;
+	context->priv = priv;
+
+	io = ocf_volume_new_io(&cache->device->volume);
+	if (!io) {
+		result = -OCF_ERR_NO_MEM;
+		goto err_io;
+	}
+
+	data = ctx_data_alloc(cache->owner, 1);
+	if (!data) {
+		result = -OCF_ERR_NO_MEM;
+		goto err_data;
 	}
 
 	if (dir == OCF_WRITE)
@@ -183,21 +206,21 @@ int ocf_submit_cache_page(struct ocf_cache *cache, uint64_t addr,
 
 	result = ocf_io_set_data(io, data, 0);
 	if (result)
-		goto end;
+		goto err_set_data;
 
 	ocf_io_configure(io, addr, PAGE_SIZE, dir, 0, 0);
+	ocf_io_set_cmpl(io, context, NULL, ocf_submit_cache_page_end);
 
-	result = ocf_submit_io_wait(io);
+	ocf_volume_submit_io(io);
-	if (result)
+	return;
-		goto end;
 
-	if (dir == OCF_READ)
+err_set_data:
-		ctx_data_rd_check(cache->owner, buffer, data, PAGE_SIZE);
-end:
-	if (io)
-		ocf_io_put(io);
 	ctx_data_free(cache->owner, data);
-	return result;
+err_data:
+	ocf_io_put(io);
+err_io:
+	env_vfree(context);
+	cmpl(priv, result);
 }
 
 static void ocf_submit_volume_req_cmpl(struct ocf_io *io, int error)
@@ -347,6 +370,12 @@ void ocf_submit_volume_req(ocf_volume_t volume, struct ocf_request *req,
 	ocf_volume_submit_io(io);
 }
 
+struct ocf_submit_io_wait_context {
+	env_completion complete;
+	int error;
+	env_atomic req_remaining;
+};
+
 static void ocf_submit_io_wait_end(struct ocf_io *io, int error)
 {
 	struct ocf_submit_io_wait_context *context = io->priv1;

src/utils/utils_io.h

@@ -43,18 +43,21 @@ static inline int ocf_io_overlaps(uint32_t start1, uint32_t count1,
 			start2 + count2 - 1);
 }
 
+typedef void (*ocf_submit_end_t)(void *priv, int error);
+
 int ocf_submit_io_wait(struct ocf_io *io);
 
-int ocf_submit_volume_flush_wait(ocf_volume_t volume);
+void ocf_submit_volume_flush(ocf_volume_t volume,
+		ocf_submit_end_t cmpl, void *priv);
 
-int ocf_submit_volume_discard_wait(ocf_volume_t volume, uint64_t addr,
+void ocf_submit_volume_discard(ocf_volume_t volume, uint64_t addr,
-		uint64_t length);
+		uint64_t length, ocf_submit_end_t cmpl, void *priv);
 
-int ocf_submit_write_zeroes_wait(ocf_volume_t volume, uint64_t addr,
+void ocf_submit_write_zeros(ocf_volume_t volume, uint64_t addr,
-		uint64_t length);
+		uint64_t length, ocf_submit_end_t cmpl, void *priv);
 
-int ocf_submit_cache_page(struct ocf_cache *cache, uint64_t addr,
+void ocf_submit_cache_page(ocf_cache_t cache, uint64_t addr, int dir,
-		int dir, void *buffer);
+		void *buffer, ocf_submit_end_t cmpl, void *priv);
 
 void ocf_submit_volume_req(ocf_volume_t volume, struct ocf_request *req,
 		ocf_req_end_t callback);

src/utils/utils_req.c

@@ -206,7 +206,9 @@ struct ocf_request *ocf_req_new(ocf_queue_t queue, ocf_core_t core,
 	req->core_id = core ? ocf_core_get_id(core) : 0;
 	req->cache = cache;
 
-	env_atomic_inc(&cache->pending_requests);
+	if (queue != cache->mngt_queue)
+		env_atomic_inc(&cache->pending_requests);
+
 	start_cache_req(req);
 
 	env_atomic_set(&req->ref_count, 1);
@@ -292,7 +294,8 @@ void ocf_req_put(struct ocf_request *req)
 		env_waitqueue_wake_up(&req->cache->pending_cache_wq);
 	}
 
-	env_atomic_dec(&req->cache->pending_requests);
+	if (req->io_queue != req->cache->mngt_queue)
+		env_atomic_dec(&req->cache->pending_requests);
 
 	allocator = _ocf_req_get_allocator(req->cache,
 			req->alloc_core_line_count);

tests/unit/tests/mngt/ocf_mngt_cache.c/_cache_mng_set_cache_mode_test.c

@@ -102,6 +102,136 @@ char *__wrap_ocf_cache_get_name(ocf_cache_t cache)
 {
 }
 
+void __wrap__ocf_mngt_test_volume_initial_write(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_test_volume_first_read(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_test_volume_discard(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_test_volume_second_read(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_cache_device(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_check_ram(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_load_properties(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_prepare_metadata(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_test_volume(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_load_superblock(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_init_instance(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_clean_pol(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_flush_metadata(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_discard(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_flush(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_shutdown_status(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_post_init(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_stop_wait_io(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_stop_remove_cores(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_stop_unplug(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_stop_put_io_queues(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_detach_flush(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_detach_wait_pending(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_detach_update_metadata(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_detach_unplug(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_test_volume_first_read(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
 static void _cache_mng_set_cache_mode_test01(void **state)
 {
 	ocf_cache_mode_t mode_old = -20;

tests/unit/tests/mngt/ocf_mngt_cache.c/ocf_mngt_cache_set_fallback_pt_error_threshold.c

@@ -59,6 +59,136 @@ char *__wrap_ocf_cache_get_name(ocf_cache_t cache)
 {
 }
 
+void __wrap__ocf_mngt_test_volume_initial_write(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_test_volume_first_read(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_test_volume_discard(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_test_volume_second_read(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_cache_device(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_check_ram(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_load_properties(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_prepare_metadata(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_test_volume(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_load_superblock(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_init_instance(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_clean_pol(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_flush_metadata(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_discard(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_flush(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_shutdown_status(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_attach_post_init(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_stop_wait_io(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_stop_remove_cores(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_stop_unplug(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_stop_put_io_queues(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_detach_flush(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_detach_wait_pending(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_detach_update_metadata(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap_ocf_mngt_cache_detach_unplug(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
+void __wrap__ocf_mngt_test_volume_first_read(
+		  ocf_mngt_pipeline_t test_pipeline, void *priv)
+{
+}
+
 static void ocf_mngt_cache_set_fallback_pt_error_threshold_test01(void **state)
 {
 	struct ocf_cache cache;