Implement asynchronous attach, load, detach and stop

NOTE: This is still not the real asynchronism. Metadata interfaces
are still not fully asynchronous.

Signed-off-by: Robert Baldyga <robert.baldyga@intel.com>
Signed-off-by: Michal Mielewczyk <michal.mielewczyk@intel.com>

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