Merge pull request #142 from robertbaldyga/pipeline-return-cleanup

Cleanup, fixes & improvements

src/metadata/metadata.c

@@ -251,44 +251,37 @@ static void ocf_metadata_load_properties_cmpl(
 
 	if (superblock->magic_number != CACHE_MAGIC_NUMBER) {
 		ocf_log(ctx, log_info, "Cannot detect pre-existing metadata\n");
-		cmpl(priv, -ENODATA, NULL);
+		OCF_CMPL_RET(priv, -ENODATA, NULL);
-		return;
 	}
 
 	if (METADATA_VERSION() != superblock->metadata_version) {
 		ocf_log(ctx, log_err, "Metadata version mismatch!\n");
-		cmpl(priv, -EBADF, NULL);
+		OCF_CMPL_RET(priv, -EBADF, NULL);
-		return;
 	}
 
 	if (!ocf_cache_line_size_is_valid(superblock->line_size)) {
 		ocf_log(ctx, log_err, "ERROR: Invalid cache line size!\n");
-		cmpl(priv, -EINVAL, NULL);
+		OCF_CMPL_RET(priv, -EINVAL, NULL);
-		return;
 	}
 
 	if ((unsigned)superblock->metadata_layout >= ocf_metadata_layout_max) {
 		ocf_log(ctx, log_err, "ERROR: Invalid metadata layout!\n");
-		cmpl(priv, -EINVAL, NULL);
+		OCF_CMPL_RET(priv, -EINVAL, NULL);
-		return;
 	}
 
 	if (superblock->cache_mode >= ocf_cache_mode_max) {
 		ocf_log(ctx, log_err, "ERROR: Invalid cache mode!\n");
-		cmpl(priv, -EINVAL, NULL);
+		OCF_CMPL_RET(priv, -EINVAL, NULL);
-		return;
 	}
 
 	if (superblock->clean_shutdown > ocf_metadata_clean_shutdown) {
 		ocf_log(ctx, log_err, "ERROR: Invalid shutdown status!\n");
-		cmpl(priv, -EINVAL, NULL);
+		OCF_CMPL_RET(priv, -EINVAL, NULL);
-		return;
 	}
 
 	if (superblock->dirty_flushed > DIRTY_FLUSHED) {
 		ocf_log(ctx, log_err, "ERROR: Invalid flush status!\n");
-		cmpl(priv, -EINVAL, NULL);
+		OCF_CMPL_RET(priv, -EINVAL, NULL);
-		return;
 	}
 
 	properties.line_size = superblock->line_size;
@@ -297,7 +290,7 @@ static void ocf_metadata_load_properties_cmpl(
 	properties.shutdown_status = superblock->clean_shutdown;
 	properties.dirty_flushed = superblock->dirty_flushed;
 
-	cmpl(priv, 0, &properties);
+	OCF_CMPL_RET(priv, 0, &properties);
 }
 
 void ocf_metadata_load_properties(ocf_volume_t volume,
@@ -310,7 +303,7 @@ void ocf_metadata_load_properties(ocf_volume_t volume,
 	result = ocf_metadata_read_sb(volume->cache->owner, volume,
 			ocf_metadata_load_properties_cmpl, cmpl, priv);
 	if (result)
-		cmpl(priv, result, NULL);
+		OCF_CMPL_RET(priv, result, NULL);
 }
 
 static void ocf_metadata_probe_cmpl(struct ocf_metadata_read_sb_ctx *context)
@@ -320,31 +313,23 @@ static void ocf_metadata_probe_cmpl(struct ocf_metadata_read_sb_ctx *context)
 	ocf_metadata_probe_end_t cmpl = context->priv1;
 	void *priv = context->priv2;
 
-	if (superblock->magic_number != CACHE_MAGIC_NUMBER) {
+	if (superblock->magic_number != CACHE_MAGIC_NUMBER)
-		cmpl(priv, -ENODATA, NULL);
+		OCF_CMPL_RET(priv, -ENODATA, NULL);
-		return;
-	}
 
-	if (METADATA_VERSION() != superblock->metadata_version) {
+	if (METADATA_VERSION() != superblock->metadata_version)
-		cmpl(priv, -EBADF, NULL);
+		OCF_CMPL_RET(priv, -EBADF, NULL);
-		return;
-	}
 
-	if (superblock->clean_shutdown > ocf_metadata_clean_shutdown) {
+	if (superblock->clean_shutdown > ocf_metadata_clean_shutdown)
-		cmpl(priv, -EINVAL, NULL);
+		OCF_CMPL_RET(priv, -EINVAL, NULL);
-		return;
-	}
 
-	if (superblock->dirty_flushed > DIRTY_FLUSHED) {
+	if (superblock->dirty_flushed > DIRTY_FLUSHED)
-		cmpl(priv, -EINVAL, NULL);
+		OCF_CMPL_RET(priv, -EINVAL, NULL);
-		return;
-	}
 
 	status.clean_shutdown = (superblock->clean_shutdown !=
 			ocf_metadata_dirty_shutdown);
 	status.cache_dirty = (superblock->dirty_flushed == DIRTY_NOT_FLUSHED);
 
-	cmpl(priv, 0, &status);
+	OCF_CMPL_RET(priv, 0, &status);
 }
 
 void ocf_metadata_probe(ocf_ctx_t ctx, ocf_volume_t volume,
@@ -358,7 +343,7 @@ void ocf_metadata_probe(ocf_ctx_t ctx, ocf_volume_t volume,
 	result = ocf_metadata_read_sb(ctx, volume, ocf_metadata_probe_cmpl,
 			cmpl, priv);
 	if (result)
-		cmpl(priv, result, NULL);
+		OCF_CMPL_RET(priv, result, NULL);
 }
 
 /* completion context for query_cores */
@@ -385,10 +370,8 @@ void ocf_metadata_probe_cores(ocf_ctx_t ctx, ocf_volume_t volume,
 	const struct ocf_metadata_iface *iface;
 
 	context = env_vzalloc(sizeof(*context));
-	if (!context) {
+	if (!context)
-		cmpl(priv, -OCF_ERR_NO_MEM, 0);
+		OCF_CMPL_RET(priv, -OCF_ERR_NO_MEM, 0);
-		return;
-	}
 
 	context->cmpl = cmpl;
 	context->priv = priv;

src/metadata/metadata_hash.c

@@ -12,6 +12,7 @@
 #include "../utils/utils_cache_line.h"
 #include "../utils/utils_pipeline.h"
 #include "../ocf_def_priv.h"
+#include "../ocf_priv.h"
 
 #define OCF_METADATA_HASH_DEBUG 0
 
@@ -794,17 +795,14 @@ void ocf_metadata_hash_query_cores(ocf_ctx_t owner, ocf_volume_t volume,
 	struct query_cores_context *context;
 	int err;
 
-	if (count > OCF_CORE_MAX) {
+	if (count > OCF_CORE_MAX)
-		cmpl(priv, -EINVAL, 0);
+		OCF_CMPL_RET(priv, -EINVAL, 0);
-		return;
-	}
 
 	/* intialize query context */
 	context = env_secure_alloc(sizeof(*context));
-	if (!context) {
+	if (!context)
-		cmpl(priv, -ENOMEM, 0);
+		OCF_CMPL_RET(priv, -ENOMEM, 0);
-		return;
+
-	}
 	ENV_BUG_ON(env_memset(context, sizeof(*context), 0));
 	context->ctx = owner;
 	context->params.cmpl = cmpl;
@@ -1199,10 +1197,7 @@ static void ocf_metadata_hash_generic_complete(void *priv, int error)
 {
 	struct ocf_metadata_hash_context *context = priv;
 
-	if (error)
+	OCF_PL_NEXT_ON_SUCCESS_RET(context->pipeline, error);
-		ocf_pipeline_finish(context->pipeline, error);
-	else
-		ocf_pipeline_next(context->pipeline);
 }
 
 static void ocf_medatata_hash_load_segment(ocf_pipeline_t pipeline,
@@ -1240,8 +1235,7 @@ static void ocf_medatata_hash_check_crc_sb_config(ocf_pipeline_t pipeline,
 		ocf_cache_log(cache, log_err,
 				"Loading %s ERROR, invalid checksum",
 				ocf_metadata_hash_raw_names[segment]);
-		ocf_pipeline_finish(pipeline, -OCF_ERR_INVAL);
+		OCF_PL_FINISH_RET(pipeline, -OCF_ERR_INVAL);
-		return;
 	}
 
 	ocf_pipeline_next(pipeline);
@@ -1267,8 +1261,7 @@ static void ocf_medatata_hash_check_crc(ocf_pipeline_t pipeline,
 		ocf_cache_log(cache, log_err,
 				"Loading %s ERROR, invalid checksum",
 				ocf_metadata_hash_raw_names[segment]);
-		ocf_pipeline_finish(pipeline, -OCF_ERR_INVAL);
+		OCF_PL_FINISH_RET(pipeline, -OCF_ERR_INVAL);
-		return;
 	}
 
 	ocf_pipeline_next(pipeline);
@@ -1308,15 +1301,13 @@ static void ocf_medatata_hash_load_superblock_post(ocf_pipeline_t pipeline,
 	if (sb_config->core_count > OCF_CORE_MAX) {
 		ocf_cache_log(cache, log_err,
 			"Loading cache state ERROR, invalid cores count\n");
-		ocf_pipeline_finish(pipeline, -OCF_ERR_INVAL);
+		OCF_PL_FINISH_RET(pipeline, -OCF_ERR_INVAL);
-		return;
 	}
 
 	if (sb_config->valid_parts_no > OCF_IO_CLASS_MAX) {
 		ocf_cache_log(cache, log_err,
 			"Loading cache state ERROR, invalid partition count\n");
-		ocf_pipeline_finish(pipeline, -OCF_ERR_INVAL);
+		OCF_PL_FINISH_RET(pipeline, -OCF_ERR_INVAL);
-		return;
 	}
 
 	ocf_pipeline_next(pipeline);
@@ -1392,10 +1383,8 @@ static void ocf_metadata_hash_load_superblock(ocf_cache_t cache,
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&ocf_metadata_hash_load_sb_pipeline_props);
-	if (result) {
+	if (result)
-		cmpl(priv, result);
+		OCF_CMPL_RET(priv, result);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 
@@ -1527,10 +1516,8 @@ static void ocf_metadata_hash_flush_superblock(ocf_cache_t cache,
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&ocf_metadata_hash_flush_sb_pipeline_props);
-	if (result) {
+	if (result)
-		cmpl(priv, result);
+		OCF_CMPL_RET(priv, result);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 
@@ -1602,12 +1589,7 @@ static void ocf_medatata_hash_flush_all_set_status_complete(
 {
 	struct ocf_metadata_hash_context *context = priv;
 
-	if (error) {
+	OCF_PL_NEXT_ON_SUCCESS_RET(context->pipeline, error);
-		ocf_pipeline_finish(context->pipeline, error);
-		return;
-	}
-
-	ocf_pipeline_next(context->pipeline);
 }
 
 static void ocf_medatata_hash_flush_all_set_status(ocf_pipeline_t pipeline,
@@ -1683,10 +1665,8 @@ static void ocf_metadata_hash_flush_all(ocf_cache_t cache,
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&ocf_metadata_hash_flush_all_pipeline_props);
-	if (result) {
+	if (result)
-		cmpl(priv, result);
+		OCF_CMPL_RET(priv, result);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 
@@ -1807,10 +1787,8 @@ static void ocf_metadata_hash_load_all(ocf_cache_t cache,
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&ocf_metadata_hash_load_all_pipeline_props);
-	if (result) {
+	if (result)
-		cmpl(priv, result);
+		OCF_CMPL_RET(priv, result);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 
@@ -1968,10 +1946,8 @@ static void _ocf_metadata_hash_load_recovery_legacy(ocf_cache_t cache,
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&ocf_metadata_hash_load_recovery_legacy_pl_props);
-	if (result) {
+	if (result)
-		cmpl(priv, result);
+		OCF_CMPL_RET(priv, result);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 
@@ -2003,12 +1979,7 @@ static void ocf_metadata_hash_load_atomic_metadata_complete(
 {
 	struct ocf_metadata_hash_context *context = priv;
 
-	if (error) {
+	OCF_PL_NEXT_ON_SUCCESS_RET(context->pipeline, error);
-		ocf_pipeline_finish(context->pipeline, error);
-		return;
-	}
-
-	ocf_pipeline_next(context->pipeline);
 }
 
 static int ocf_metadata_hash_load_atomic_metadata_drain(void *priv,
@@ -2072,7 +2043,7 @@ static void ocf_medatata_hash_load_atomic_metadata(
 		ocf_metadata_error(cache);
 		ocf_cache_log(cache, log_err,
 				"Metadata read for recovery FAILURE\n");
-		ocf_pipeline_finish(pipeline, result);
+		OCF_PL_FINISH_RET(pipeline, result);
 	}
 }
 
@@ -2117,10 +2088,8 @@ static void _ocf_metadata_hash_load_recovery_atomic(ocf_cache_t cache,
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&ocf_metadata_hash_load_recovery_atomic_pl_props);
-	if (result) {
+	if (result)
-		cmpl(priv, result);
+		OCF_CMPL_RET(priv, result);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 

src/metadata/metadata_raw.c

@@ -9,6 +9,7 @@
 #include "metadata_io.h"
 #include "metadata_raw_atomic.h"
 #include "../ocf_def_priv.h"
+#include "../ocf_priv.h"
 
 #define OCF_METADATA_RAW_DEBUG 0
 
@@ -250,10 +251,8 @@ static void _raw_ram_load_all(ocf_cache_t cache, struct ocf_metadata_raw *raw,
 	OCF_DEBUG_TRACE(cache);
 
 	context = env_vmalloc(sizeof(*context));
-	if (!context) {
+	if (!context)
-		cmpl(priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	context->raw = raw;
 	context->cmpl = cmpl;
@@ -319,10 +318,8 @@ static void _raw_ram_flush_all(ocf_cache_t cache, struct ocf_metadata_raw *raw,
 	OCF_DEBUG_TRACE(cache);
 
 	context = env_vmalloc(sizeof(*context));
-	if (!context) {
+	if (!context)
-		cmpl(priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	context->raw = raw;
 	context->cmpl = cmpl;

src/metadata/metadata_raw_dynamic.c

@@ -13,6 +13,7 @@
 #include "../utils/utils_io.h"
 #include "../utils/utils_req.h"
 #include "../ocf_def_priv.h"
+#include "../ocf_priv.h"
 
 #define OCF_METADATA_RAW_DEBUG  0
 
@@ -434,10 +435,8 @@ void raw_dynamic_load_all(ocf_cache_t cache, struct ocf_metadata_raw *raw,
 	OCF_DEBUG_TRACE(cache);
 
 	context = env_vzalloc(sizeof(*context));
-	if (!context) {
+	if (!context)
-		cmpl(priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	context->raw = raw;
 	context->cache = cache;
@@ -475,7 +474,7 @@ err_zpage:
 	ctx_data_free(cache->owner, context->data);
 err_data:
 	env_vfree(context);
-	cmpl(priv, result);
+	OCF_CMPL_RET(priv, result);
 }
 
 /*
@@ -534,10 +533,8 @@ void raw_dynamic_flush_all(ocf_cache_t cache, struct ocf_metadata_raw *raw,
 	OCF_DEBUG_TRACE(cache);
 
 	context = env_vmalloc(sizeof(*context));
-	if (!context) {
+	if (!context)
-		cmpl(priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	context->raw = raw;
 	context->cmpl = cmpl;
@@ -548,7 +545,7 @@ void raw_dynamic_flush_all(ocf_cache_t cache, struct ocf_metadata_raw *raw,
 			raw_dynamic_flush_all_fill,
 			raw_dynamic_flush_all_complete);
 	if (result)
-		cmpl(priv, result);
+		OCF_CMPL_RET(priv, result);
 }
 
 /*

src/metadata/metadata_superblock.h

@@ -79,12 +79,6 @@ static inline void ocf_metadata_load_superblock(ocf_cache_t cache,
 static inline void ocf_metadata_flush_superblock(ocf_cache_t cache,
 		ocf_metadata_end_t cmpl, void *priv)
 {
-	/* TODO: Shouldn't it be checked by the caller? */
-	if (!cache->device) {
-		cmpl(priv, 0);
-		return;
-	}
-
 	cache->metadata.iface.flush_superblock(cache, cmpl, priv);
 }
 

src/mngt/ocf_mngt_cache.c

@@ -475,9 +475,7 @@ void _ocf_mngt_init_instance_load_complete(void *priv, int error)
 	if (error) {
 		ocf_cache_log(cache, log_err,
 				"Cannot read cache metadata\n");
-		ocf_pipeline_finish(context->pipeline,
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_START_CACHE_FAIL);
-				-OCF_ERR_START_CACHE_FAIL);
-		return;
 	}
 
 	cleaning_policy = cache->conf_meta->cleaning_policy_type;
@@ -533,10 +531,8 @@ static void _ocf_mngt_init_instance_load(
 	OCF_ASSERT_PLUGGED(cache);
 
 	ret = _ocf_mngt_init_instance_add_cores(context);
-	if (ret) {
+	if (ret)
-		ocf_pipeline_finish(context->pipeline, ret);
+		OCF_PL_FINISH_RET(context->pipeline, ret);
-		return;
-	}
 
 	if (context->metadata.shutdown_status == ocf_metadata_clean_shutdown)
 		_ocf_mngt_init_instance_clean_load(context);
@@ -589,10 +585,9 @@ static void _ocf_mngt_attach_cache_device(ocf_pipeline_t pipeline,
 	int ret;
 
 	cache->device = env_vzalloc(sizeof(*cache->device));
-	if (!cache->device) {
+	if (!cache->device)
-		ret = -OCF_ERR_NO_MEM;
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_NO_MEM);
-		goto err;
+
-	}
 	context->flags.device_alloc = true;
 
 	cache->device->init_mode = context->init_mode;
@@ -600,14 +595,14 @@ static void _ocf_mngt_attach_cache_device(ocf_pipeline_t pipeline,
 	/* Prepare UUID of cache volume */
 	type = ocf_ctx_get_volume_type(cache->owner, context->cfg.volume_type);
 	if (!type) {
-		ret = -OCF_ERR_INVAL_VOLUME_TYPE;
+		OCF_PL_FINISH_RET(context->pipeline,
-		goto err;
+				-OCF_ERR_INVAL_VOLUME_TYPE);
 	}
 
 	ret = ocf_volume_init(&cache->device->volume, type,
 			&context->cfg.uuid, true);
 	if (ret)
-		goto err;
+		OCF_PL_FINISH_RET(context->pipeline, ret);
 
 	cache->device->volume.cache = cache;
 	context->flags.volume_inited = true;
@@ -620,7 +615,7 @@ static void _ocf_mngt_attach_cache_device(ocf_pipeline_t pipeline,
 			context->cfg.volume_params);
 	if (ret) {
 		ocf_cache_log(cache, log_err, "ERROR: Cache not available\n");
-		goto err;
+		OCF_PL_FINISH_RET(context->pipeline, ret);
 	}
 	context->flags.device_opened = true;
 
@@ -630,15 +625,10 @@ static void _ocf_mngt_attach_cache_device(ocf_pipeline_t pipeline,
 	if (context->volume_size < OCF_CACHE_SIZE_MIN) {
 		ocf_cache_log(cache, log_err, "ERROR: Cache cache size must "
 			"be at least %llu [MiB]\n", OCF_CACHE_SIZE_MIN / MiB);
-		ret = -OCF_ERR_START_CACHE_FAIL;
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_START_CACHE_FAIL);
-		goto err;
 	}
 
 	ocf_pipeline_next(pipeline);
-	return;
-
-err:
-	ocf_pipeline_finish(context->pipeline, ret);
 }
 
 /**
@@ -719,12 +709,7 @@ static void _ocf_mngt_test_volume_initial_write_complete(void *priv, int error)
 {
 	struct ocf_cache_attach_context *context = priv;
 
-	if (error) {
+	OCF_PL_NEXT_ON_SUCCESS_RET(context->test.pipeline, error);
-		ocf_pipeline_finish(context->test.pipeline, error);
-		return;
-	}
-
-	ocf_pipeline_next(context->test.pipeline);
 }
 
 static void _ocf_mngt_test_volume_initial_write(
@@ -750,29 +735,23 @@ static void _ocf_mngt_test_volume_first_read_complete(void *priv, int error)
 	ocf_cache_t cache = context->cache;
 	int ret, diff;
 
-	if (error) {
+	if (error)
-		ocf_pipeline_finish(context->test.pipeline, error);
+		OCF_PL_FINISH_RET(context->test.pipeline, error);
-		return;
-	}
 
 	ret = env_memcmp(context->test.rw_buffer, PAGE_SIZE,
 			context->test.cmp_buffer, PAGE_SIZE, &diff);
-	if (ret) {
+	if (ret)
-		ocf_pipeline_finish(context->test.pipeline, ret);
+		OCF_PL_FINISH_RET(context->test.pipeline, ret);
-		return;
-	}
 
 	if (diff) {
 		/* we read back different data than what we had just
 		   written - this is fatal error */
-		ocf_pipeline_finish(context->test.pipeline, -EIO);
+		OCF_PL_FINISH_RET(context->test.pipeline, -EIO);
-		return;
 	}
 
 	if (!ocf_volume_is_atomic(&cache->device->volume)) {
 		/* If not atomic, stop testing here */
-		ocf_pipeline_finish(context->test.pipeline, 0);
+		OCF_PL_FINISH_RET(context->test.pipeline, 0);
-		return;
 	}
 
 	ocf_pipeline_next(context->test.pipeline);
@@ -800,12 +779,7 @@ static void _ocf_mngt_test_volume_discard_complete(void *priv, int error)
 {
 	struct ocf_cache_attach_context *context = priv;
 
-	if (error) {
+	OCF_PL_NEXT_ON_SUCCESS_RET(context->test.pipeline, error);
-		ocf_pipeline_finish(context->test.pipeline, error);
-		return;
-	}
-
-	ocf_pipeline_next(context->test.pipeline);
 }
 
 static void _ocf_mngt_test_volume_discard(
@@ -829,17 +803,13 @@ static void _ocf_mngt_test_volume_second_read_complete(void *priv, int error)
 	ocf_cache_t cache = context->cache;
 	int ret, diff;
 
-	if (error) {
+	if (error)
-		ocf_pipeline_finish(context->test.pipeline, error);
+		OCF_PL_FINISH_RET(context->test.pipeline, error);
-		return;
-	}
 
 	ret = env_memcmp(context->test.rw_buffer, PAGE_SIZE,
 			context->test.cmp_buffer, PAGE_SIZE, &diff);
-	if (ret) {
+	if (ret)
-		ocf_pipeline_finish(context->test.pipeline, ret);
+		OCF_PL_FINISH_RET(context->test.pipeline, ret);
-		return;
-	}
 
 	if (diff) {
 		/* discard does not cause target adresses to return 0 on
@@ -876,12 +846,9 @@ static void _ocf_mngt_test_volume_finish(ocf_pipeline_t pipeline,
 	env_free(context->test.rw_buffer);
 	env_free(context->test.cmp_buffer);
 
-	if (error)
-		ocf_pipeline_finish(context->pipeline, error);
-	else
-		ocf_pipeline_next(context->pipeline);
-
 	ocf_pipeline_destroy(context->test.pipeline);
+
+	OCF_PL_NEXT_ON_SUCCESS_RET(context->pipeline, error);
 }
 
 struct ocf_pipeline_properties _ocf_mngt_test_volume_pipeline_properties = {
@@ -906,18 +873,14 @@ static void _ocf_mngt_test_volume(ocf_pipeline_t pipeline,
 
 	cache->device->volume.features.discard_zeroes = 1;
 
-	if (!context->cfg.perform_test) {
+	if (!context->cfg.perform_test)
-		ocf_pipeline_next(pipeline);
+		OCF_PL_NEXT_RET(pipeline);
-		return;
-	}
 
 	context->test.reserved_lba_addr = ocf_metadata_get_reserved_lba(cache);
 
 	context->test.rw_buffer = env_malloc(PAGE_SIZE, ENV_MEM_NORMAL);
-	if (!context->test.rw_buffer) {
+	if (!context->test.rw_buffer)
-		ocf_pipeline_finish(context->pipeline, -OCF_ERR_NO_MEM);
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	context->test.cmp_buffer = env_malloc(PAGE_SIZE, ENV_MEM_NORMAL);
 	if (!context->test.cmp_buffer)
@@ -932,14 +895,13 @@ static void _ocf_mngt_test_volume(ocf_pipeline_t pipeline,
 
 	context->test.pipeline = test_pipeline;
 
-	ocf_pipeline_next(test_pipeline);
+	OCF_PL_NEXT_RET(test_pipeline);
-	return;
 
 err_pipeline:
 	env_free(context->test.rw_buffer);
 err_buffer:
 	env_free(context->test.cmp_buffer);
-	ocf_pipeline_finish(context->pipeline, -OCF_ERR_NO_MEM);
+	OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_NO_MEM);
 }
 
 /**
@@ -954,10 +916,8 @@ static void _ocf_mngt_attach_load_properties_end(void *priv, int error,
 
 	context->metadata.status = error;
 
-	if (error) {
+	if (error)
-		ocf_pipeline_next(context->pipeline);
+		OCF_PL_NEXT_RET(context->pipeline);
-		return;
-	}
 
 	context->metadata.shutdown_status = properties->shutdown_status;
 	context->metadata.dirty_flushed = properties->dirty_flushed;
@@ -983,10 +943,8 @@ static void _ocf_mngt_attach_load_properties(ocf_pipeline_t pipeline,
 	context->metadata.dirty_flushed = DIRTY_FLUSHED;
 	context->metadata.line_size = context->cfg.cache_line_size;
 
-	if (cache->device->init_mode == ocf_init_mode_metadata_volatile) {
+	if (cache->device->init_mode == ocf_init_mode_metadata_volatile)
-		ocf_pipeline_next(context->pipeline);
+		OCF_PL_NEXT_RET(context->pipeline);
-		return;
-	}
 
 	ocf_metadata_load_properties(&cache->device->volume,
 			_ocf_mngt_attach_load_properties_end, context);
@@ -1001,9 +959,7 @@ static void _ocf_mngt_attach_prepare_metadata(ocf_pipeline_t pipeline,
 
 	if (context->init_mode == ocf_init_mode_load &&
 			context->metadata.status) {
-		ocf_pipeline_finish(context->pipeline,
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_START_CACHE_FAIL);
-				-OCF_ERR_START_CACHE_FAIL);
-		return;
 	}
 
 	context->metadata.line_size = context->metadata.line_size ?:
@@ -1015,9 +971,7 @@ static void _ocf_mngt_attach_prepare_metadata(ocf_pipeline_t pipeline,
 	if (ocf_metadata_init_variable_size(cache, context->volume_size,
 			context->metadata.line_size,
 			cache->conf_meta->metadata_layout)) {
-		ocf_pipeline_finish(context->pipeline,
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_START_CACHE_FAIL);
-				-OCF_ERR_START_CACHE_FAIL);
-		return;
 	}
 
 	ocf_cache_log(cache, log_debug, "Cache attached\n");
@@ -1031,10 +985,8 @@ static void _ocf_mngt_attach_prepare_metadata(ocf_pipeline_t pipeline,
 	cache->device->freelist_part = &cache->device->runtime_meta->freelist_part;
 
 	ret = ocf_concurrency_init(cache);
-	if (ret) {
+	if (ret)
-		ocf_pipeline_finish(context->pipeline, ret);
+		OCF_PL_FINISH_RET(context->pipeline, ret);
-		return;
-	}
 
 	context->flags.concurrency_inited = 1;
 
@@ -1055,17 +1007,16 @@ static void _ocf_mngt_init_instance_init(struct ocf_cache_attach_context *contex
 		if (context->metadata.shutdown_status !=
 				ocf_metadata_clean_shutdown) {
 			ocf_cache_log(cache, log_err, DIRTY_SHUTDOWN_ERROR_MSG);
-			ocf_pipeline_finish(context->pipeline,
+			OCF_PL_FINISH_RET(context->pipeline,
 					-OCF_ERR_DIRTY_SHUTDOWN);
-			return;
 		}
 
 		if (context->metadata.dirty_flushed == DIRTY_NOT_FLUSHED) {
 			ocf_cache_log(cache, log_err,
 					DIRTY_NOT_FLUSHED_ERROR_MSG);
-			ocf_pipeline_finish(context->pipeline,
+			OCF_PL_FINISH_RET(context->pipeline,
 					-OCF_ERR_DIRTY_EXISTS);
-			return;
+
 		}
 	}
 
@@ -1351,7 +1302,7 @@ static void _ocf_mngt_attach_check_ram(ocf_pipeline_t pipeline,
 				"Available RAM: %" ENV_PRIu64 " B\n", free_ram);
 		ocf_cache_log(cache, log_err, "Needed RAM: %" ENV_PRIu64 " B\n",
 				min_free_ram);
-		ocf_pipeline_finish(pipeline, -OCF_ERR_NO_FREE_RAM);
+		OCF_PL_FINISH_RET(pipeline, -OCF_ERR_NO_FREE_RAM);
 	}
 
 	ocf_pipeline_next(pipeline);
@@ -1366,9 +1317,8 @@ static void _ocf_mngt_attach_load_superblock_complete(void *priv, int error)
 	if (error) {
 		ocf_cache_log(cache, log_err,
 				"ERROR: Cannot load cache state\n");
-		ocf_pipeline_finish(context->pipeline,
+		OCF_PL_FINISH_RET(context->pipeline,
 				-OCF_ERR_START_CACHE_FAIL);
-		return;
 	}
 
 	ocf_pipeline_next(context->pipeline);
@@ -1380,10 +1330,8 @@ static void _ocf_mngt_attach_load_superblock(ocf_pipeline_t pipeline,
 	struct ocf_cache_attach_context *context = priv;
 	ocf_cache_t cache = context->cache;
 
-	if (cache->device->init_mode != ocf_init_mode_load) {
+	if (cache->device->init_mode != ocf_init_mode_load)
-		ocf_pipeline_next(context->pipeline);
+		OCF_PL_NEXT_RET(context->pipeline);
-		return;
-	}
 
 	ocf_cache_log(cache, log_info, "Loading cache state...\n");
 	ocf_metadata_load_superblock(cache,
@@ -1405,7 +1353,7 @@ static void _ocf_mngt_attach_init_instance(ocf_pipeline_t pipeline,
 		_ocf_mngt_init_instance_load(context);
 		return;
 	default:
-		ocf_pipeline_finish(context->pipeline, -OCF_ERR_INVAL);
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_INVAL);
 	}
 }
 
@@ -1419,10 +1367,8 @@ static void _ocf_mngt_attach_clean_pol(ocf_pipeline_t pipeline,
 	/* TODO: Should this even be here? */
 	if (cache->device->init_mode != ocf_init_mode_load) {
 		result = _ocf_mngt_cache_add_cores_t_clean_pol(cache);
-		if (result) {
+		if (result)
-			ocf_pipeline_finish(context->pipeline, result);
+			OCF_PL_FINISH_RET(context->pipeline, result);
-			return;
-		}
 	}
 
 	ocf_pipeline_next(context->pipeline);
@@ -1436,9 +1382,7 @@ static void _ocf_mngt_attach_flush_metadata_complete(void *priv, int error)
 	if (error) {
 		ocf_cache_log(cache, log_err,
 				"ERROR: Cannot save cache state\n");
-		ocf_pipeline_finish(context->pipeline,
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_WRITE_CACHE);
-				-OCF_ERR_WRITE_CACHE);
-		return;
 	}
 
 	ocf_pipeline_next(context->pipeline);
@@ -1468,8 +1412,7 @@ static void _ocf_mngt_attach_discard_complete(void *priv, int error)
 		if (ocf_volume_is_atomic(&cache->device->volume)) {
 			ocf_cache_log(cache, log_err, "This step is required"
 					" for atomic mode!\n");
-			ocf_pipeline_finish(context->pipeline, error);
+			OCF_PL_FINISH_RET(context->pipeline, error);
-			return;
 		}
 
 		ocf_cache_log(cache, log_warn, "This may impact cache"
@@ -1488,15 +1431,11 @@ static void _ocf_mngt_attach_discard(ocf_pipeline_t pipeline,
 	uint64_t length = ocf_volume_get_length(&cache->device->volume) - addr;
 	bool discard = cache->device->volume.features.discard_zeroes;
 
-	if (cache->device->init_mode == ocf_init_mode_load) {
+	if (cache->device->init_mode == ocf_init_mode_load)
-		ocf_pipeline_next(context->pipeline);
+		OCF_PL_NEXT_RET(context->pipeline);
-		return;
-	}
 
-	if (!context->cfg.discard_on_start) {
+	if (!context->cfg.discard_on_start)
-		ocf_pipeline_next(context->pipeline);
+		OCF_PL_NEXT_RET(context->pipeline);
-		return;
-	}
 
 	if (!discard && ocf_volume_is_atomic(&cache->device->volume)) {
 		/* discard doesn't zero data - need to explicitly write zeros */
@@ -1513,10 +1452,7 @@ static void _ocf_mngt_attach_flush_complete(void *priv, int error)
 {
 	struct ocf_cache_attach_context *context = priv;
 
-	if (error)
+	OCF_PL_NEXT_ON_SUCCESS_RET(context->pipeline, error);
-		ocf_pipeline_finish(context->pipeline, error);
-	else
-		ocf_pipeline_next(context->pipeline);
 }
 
 static void _ocf_mngt_attach_flush(ocf_pipeline_t pipeline,
@@ -1541,9 +1477,7 @@ static void _ocf_mngt_attach_shutdown_status_complete(void *priv, int error)
 
 	if (error) {
 		ocf_cache_log(cache, log_err, "Cannot flush shutdown status\n");
-		ocf_pipeline_finish(context->pipeline,
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_WRITE_CACHE);
-				-OCF_ERR_WRITE_CACHE);
-		return;
 	}
 
 	ocf_pipeline_next(context->pipeline);
@@ -1572,8 +1506,7 @@ static void _ocf_mngt_attach_post_init(ocf_pipeline_t pipeline,
 		if (result) {
 			ocf_cache_log(cache, log_err,
 					"Error while starting cleaner\n");
-			ocf_pipeline_finish(context->pipeline, result);
+			OCF_PL_FINISH_RET(context->pipeline, result);
-			return;
 		}
 		context->flags.cleaner_started = true;
 	}
@@ -1631,10 +1564,8 @@ static void _ocf_mngt_cache_attach(ocf_cache_t cache,
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&_ocf_mngt_cache_attach_pipeline_properties);
-	if (result) {
+	if (result)
-		cmpl(cache, priv1, priv2, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(cache, priv1, priv2, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 
@@ -1668,14 +1599,13 @@ static void _ocf_mngt_cache_attach(ocf_cache_t cache,
 
 	_ocf_mngt_init_attached_nonpersistent(cache);
 
-	ocf_pipeline_next(pipeline);
+	OCF_PL_NEXT_RET(pipeline);
-	return;
 
 err_uuid:
 	env_vfree(data);
 err_pipeline:
 	ocf_pipeline_destroy(pipeline);
-	cmpl(cache, priv1, priv2, result);
+	OCF_CMPL_RET(cache, priv1, priv2, result);
 }
 
 static int _ocf_mngt_cache_validate_cfg(struct ocf_mngt_cache_config *cfg)
@@ -1794,7 +1724,7 @@ static void _ocf_mngt_cache_attach_complete(ocf_cache_t cache, void *priv1,
 			       "failed\n");
 	}
 
-	cmpl(cache, priv2, error);
+	OCF_CMPL_RET(cache, priv2, error);
 }
 
 void ocf_mngt_cache_attach(ocf_cache_t cache,
@@ -1806,11 +1736,12 @@ void ocf_mngt_cache_attach(ocf_cache_t cache,
 	OCF_CHECK_NULL(cache);
 	OCF_CHECK_NULL(cfg);
 
+	if (!cache->mngt_queue)
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_INVAL);
+
 	result = _ocf_mngt_cache_validate_device_cfg(cfg);
-	if (result) {
+	if (result)
-		cmpl(cache, priv, result);
+		OCF_CMPL_RET(cache, priv, result);
-		return;
-	}
 
 	_ocf_mngt_cache_attach(cache, cfg, false,
 			_ocf_mngt_cache_attach_complete, cmpl, priv);
@@ -1927,15 +1858,13 @@ static void _ocf_mngt_cache_load_complete(ocf_cache_t cache, void *priv1,
 {
 	ocf_mngt_cache_load_end_t cmpl = priv1;
 
-	if (error) {
+	if (error)
-		cmpl(cache, priv2, error);
+		OCF_CMPL_RET(cache, priv2, error);
-		return;
-	}
 
 	_ocf_mng_cache_set_valid(cache);
 	_ocf_mngt_cache_load_log(cache);
 
-	cmpl(cache, priv2, 0);
+	OCF_CMPL_RET(cache, priv2, 0);
 }
 
 void ocf_mngt_cache_load(ocf_cache_t cache,
@@ -1947,15 +1876,16 @@ void ocf_mngt_cache_load(ocf_cache_t cache,
 	OCF_CHECK_NULL(cache);
 	OCF_CHECK_NULL(cfg);
 
+	if (!cache->mngt_queue)
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_INVAL);
+
 	/* Load is not allowed in volatile metadata mode */
 	if (cache->metadata.is_volatile)
-		cmpl(cache, priv, -EINVAL);
+		OCF_CMPL_RET(cache, priv, -EINVAL);
 
 	result = _ocf_mngt_cache_validate_device_cfg(cfg);
-	if (result) {
+	if (result)
-		cmpl(cache, priv, result);
+		OCF_CMPL_RET(cache, priv, result);
-		return;
-	}
 
 	_ocf_mngt_cache_attach(cache, cfg, true,
 			_ocf_mngt_cache_load_complete, cmpl, priv);
@@ -2030,10 +1960,8 @@ static void ocf_mngt_cache_stop_unplug(ocf_pipeline_t pipeline,
 	struct ocf_mngt_cache_stop_context *context = priv;
 	ocf_cache_t cache = context->cache;
 
-	if (!env_atomic_read(&cache->attached)) {
+	if (!env_atomic_read(&cache->attached))
-		ocf_pipeline_next(pipeline);
+		OCF_PL_NEXT_RET(pipeline);
-		return;
-	}
 
 	_ocf_mngt_cache_unplug(cache, true, &context->unplug_context,
 			ocf_mngt_cache_stop_unplug_complete, context);
@@ -2117,12 +2045,18 @@ void ocf_mngt_cache_stop(ocf_cache_t cache,
 
 	OCF_CHECK_NULL(cache);
 
+	/*
+	 * FIXME: What if creating/setting management queue failed?
+	 * In such case we will be unable to use pipeline, and thus
+	 * perform cache stop procedure.
+	 */
+	if (!cache->mngt_queue)
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_INVAL);
+
 	result = ocf_pipeline_create(&pipeline, cache,
 			&ocf_mngt_cache_stop_pipeline_properties);
-	if (result) {
+	if (result)
-		cmpl(cache, priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 
@@ -2136,8 +2070,7 @@ void ocf_mngt_cache_stop(ocf_cache_t cache,
 			ocf_cache_get_name(cache), sizeof(context->cache_name));
 	if (result) {
 		ocf_pipeline_destroy(pipeline);
-		cmpl(cache, priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_NO_MEM);
-		return;
 	}
 
 	ocf_cache_log(cache, log_info, "Stopping cache\n");
@@ -2182,8 +2115,7 @@ static void ocf_mngt_cache_save_flush_sb_complete(void *priv, int error)
 		ocf_cache_log(cache, log_err,
 				"Failed to flush superblock! Changes "
 				"in cache config are not persistent!\n");
-		ocf_pipeline_finish(context->pipeline, -OCF_ERR_WRITE_CACHE);
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_WRITE_CACHE);
-		return;
 	}
 
 	ocf_pipeline_next(context->pipeline);
@@ -2198,12 +2130,13 @@ void ocf_mngt_cache_save(ocf_cache_t cache,
 
 	OCF_CHECK_NULL(cache);
 
+	if (!cache->mngt_queue)
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_INVAL);
+
 	result = ocf_pipeline_create(&pipeline, cache,
 			&ocf_mngt_cache_save_pipeline_properties);
-	if (result) {
+	if (result)
-		cmpl(cache, priv, result);
+		OCF_CMPL_RET(cache, priv, result);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 
@@ -2349,12 +2282,7 @@ static void ocf_mngt_cache_detach_flush_cmpl(ocf_cache_t cache,
 {
 	struct ocf_mngt_cache_detach_context *context = priv;
 
-	if (error) {
+	OCF_PL_NEXT_ON_SUCCESS_RET(context->pipeline, error);
-		ocf_pipeline_finish(context->pipeline, error);
-		return;
-	}
-
-	ocf_pipeline_next(context->pipeline);
 }
 
 static void ocf_mngt_cache_detach_flush(ocf_pipeline_t pipeline,
@@ -2475,17 +2403,16 @@ void ocf_mngt_cache_detach(ocf_cache_t cache,
 
 	OCF_CHECK_NULL(cache);
 
-	if (!env_atomic_read(&cache->attached)) {
+	if (!cache->mngt_queue)
-		cmpl(cache, priv, -OCF_ERR_INVAL);
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_INVAL);
-		return;
+
-	}
+	if (!env_atomic_read(&cache->attached))
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_INVAL);
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&ocf_mngt_cache_detach_pipeline_properties);
-	if (result) {
+	if (result)
-		cmpl(cache, priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 

src/mngt/ocf_mngt_core.c

@@ -125,10 +125,8 @@ static void _ocf_mngt_cache_add_core_flush_sb_complete(void *priv, int error)
 {
 	struct ocf_cache_add_core_context *context = priv;
 
-	if (error) {
+	if (error)
-		ocf_pipeline_finish(context->pipeline, -OCF_ERR_WRITE_CACHE);
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_WRITE_CACHE);
-		return;
-	}
 
 	/* Increase value of added cores */
 	context->cache->conf_meta->core_count++;
@@ -157,48 +155,41 @@ static void _ocf_mngt_cache_add_core(ocf_cache_t cache,
 
 	/* Set uuid */
 	result = ocf_metadata_set_core_uuid(core, &cfg->uuid, &new_uuid);
-	if (result) {
+	if (result)
-		ocf_pipeline_finish(context->pipeline, result);
+		OCF_PL_FINISH_RET(context->pipeline, result);
-		return;
+
-	}
 	context->flags.uuid_set = true;
 
 	type = ocf_ctx_get_volume_type(cache->owner, cfg->volume_type);
 	if (!type) {
-		ocf_pipeline_finish(context->pipeline,
+		OCF_PL_FINISH_RET(context->pipeline,
 				-OCF_ERR_INVAL_VOLUME_TYPE);
-		return;
 	}
 
 	result = ocf_volume_init(volume, type, &new_uuid, false);
-	if (result) {
+	if (result)
-		ocf_pipeline_finish(context->pipeline, result);
+		OCF_PL_FINISH_RET(context->pipeline, result);
-		return;
+
-	}
 	context->flags.volume_inited = true;
 
 	if (cfg->user_metadata.data && cfg->user_metadata.size > 0) {
 		result = ocf_mngt_core_set_user_metadata(core,
 				cfg->user_metadata.data,
 				cfg->user_metadata.size);
-		if (result) {
+		if (result)
-			ocf_pipeline_finish(context->pipeline, result);
+			OCF_PL_FINISH_RET(context->pipeline, result);
-			return;
-		}
 	}
 
 	result = ocf_volume_open(volume, NULL);
-	if (result) {
+	if (result)
-		ocf_pipeline_finish(context->pipeline, result);
+		OCF_PL_FINISH_RET(context->pipeline, result);
-		return;
+
-	}
 	context->flags.volume_opened = true;
 
 	length = ocf_volume_get_length(volume);
-	if (!length) {
+	if (!length)
-		ocf_pipeline_finish(context->pipeline, -OCF_ERR_CORE_NOT_AVAIL);
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_CORE_NOT_AVAIL);
-		return;
+
-	}
 	cache->core_conf_meta[cfg->core_id].length = length;
 
 	clean_type = cache->conf_meta->cleaning_policy_type;
@@ -206,20 +197,18 @@ static void _ocf_mngt_cache_add_core(ocf_cache_t cache,
 			cleaning_policy_ops[clean_type].add_core) {
 		result = cleaning_policy_ops[clean_type].add_core(cache,
 				cfg->core_id);
-		if (result) {
+		if (result)
-			ocf_pipeline_finish(context->pipeline, result);
+			OCF_PL_FINISH_RET(context->pipeline, result);
-			return;
+
-		}
 		context->flags.clean_pol_added = true;
 	}
 
 	/* When adding new core to cache, allocate stat counters */
 	core->counters =
 		env_zalloc(sizeof(*core->counters), ENV_MEM_NORMAL);
-	if (!core->counters) {
+	if (!core->counters)
-		ocf_pipeline_finish(context->pipeline, -OCF_ERR_NO_MEM);
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_NO_MEM);
-		return;
+
-	}
 	context->flags.counters_allocated = true;
 
 	/* When adding new core to cache, reset all core/cache statistics */
@@ -244,10 +233,9 @@ static void _ocf_mngt_cache_add_core(ocf_cache_t cache,
 
 	/* Add core sequence number for atomic metadata matching */
 	core_sequence_no = _ocf_mngt_get_core_seq_no(cache);
-	if (core_sequence_no == OCF_SEQ_NO_INVALID) {
+	if (core_sequence_no == OCF_SEQ_NO_INVALID)
-		ocf_pipeline_finish(context->pipeline, -OCF_ERR_TOO_MANY_CORES);
+		OCF_PL_FINISH_RET(context->pipeline, -OCF_ERR_TOO_MANY_CORES);
-		return;
+
-	}
 	cache->core_conf_meta[cfg->core_id].seq_no = core_sequence_no;
 
 	/* Update super-block with core device addition */
@@ -429,33 +417,25 @@ static void ocf_mngt_cache_add_core_prepare(ocf_pipeline_t pipeline,
 	int result;
 
 	result = _ocf_mngt_find_core_id(cache, &context->cfg);
-	if (result) {
+	if (result)
-		ocf_pipeline_finish(context->pipeline, result);
+		OCF_PL_FINISH_RET(context->pipeline, result);
-		return;
-	}
 
 	if (context->cfg.name) {
 		result = env_strncpy(core_name, sizeof(context->core_name),
 				context->cfg.name, sizeof(context->core_name));
-		if (result) {
+		if (result)
-			ocf_pipeline_finish(context->pipeline, result);
+			OCF_PL_FINISH_RET(context->pipeline, result);
-			return;
-		}
 	} else {
 		result = snprintf(core_name, sizeof(context->core_name),
 				"core%hu", context->cfg.core_id);
-		if (result < 0) {
+		if (result < 0)
-			ocf_pipeline_finish(context->pipeline, result);
+			OCF_PL_FINISH_RET(context->pipeline, result);
-			return;
-		}
 	}
 
 	result = ocf_core_set_name(&cache->core[context->cfg.core_id],
 			core_name, sizeof(context->core_name));
-	if (result) {
+	if (result)
-		ocf_pipeline_finish(context->pipeline, result);
+		OCF_PL_FINISH_RET(context->pipeline, result);
-		return;
-	}
 
 	ocf_pipeline_next(context->pipeline);
 }
@@ -473,13 +453,8 @@ static void ocf_mngt_cache_add_core_insert(ocf_pipeline_t pipeline,
 	if (context->cfg.try_add) {
 		result = _ocf_mngt_cache_try_add_core(cache, &context->core,
 				&context->cfg);
-		if (result) {
-			ocf_pipeline_finish(context->pipeline, result);
-			return;
-		}
 
-		ocf_pipeline_next(context->pipeline);
+		OCF_PL_NEXT_ON_SUCCESS_RET(context->pipeline, result);
-		return;
 	}
 
 	_ocf_mngt_cache_add_core(cache, context);
@@ -492,10 +467,8 @@ static void ocf_mngt_cache_add_core_init_front_volume(ocf_pipeline_t pipeline,
 	int result;
 
 	result = ocf_mngt_core_init_front_volume(context->core);
-	if (result) {
+	if (result)
-		ocf_pipeline_finish(context->pipeline, result);
+		OCF_PL_FINISH_RET(context->pipeline, result);
-		return;
-	}
 
 	ocf_pipeline_next(context->pipeline);
 }
@@ -549,12 +522,13 @@ void ocf_mngt_cache_add_core(ocf_cache_t cache,
 
 	OCF_CHECK_NULL(cache);
 
+	if (!cache->mngt_queue)
+		OCF_CMPL_RET(cache, NULL, priv, -OCF_ERR_INVAL);
+
 	result = ocf_pipeline_create(&pipeline, cache,
 			&ocf_mngt_cache_add_core_pipeline_properties);
-	if (result) {
+	if (result)
-		cmpl(cache, NULL, priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(cache, NULL, priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 
@@ -577,15 +551,13 @@ void ocf_mngt_cache_add_core(ocf_cache_t cache,
 
 	context->cfg.uuid.data = data;
 
-
+	OCF_PL_NEXT_RET(pipeline);
-	ocf_pipeline_next(pipeline);
-	return;
 
 err_uuid:
 	env_vfree(data);
 err_pipeline:
 	ocf_pipeline_destroy(context->pipeline);
-	cmpl(cache, NULL, priv, result);
+	OCF_CMPL_RET(cache, NULL, priv, result);
 }
 
 /*
@@ -647,13 +619,8 @@ static void ocf_mngt_cache_remove_core_flush_sb_complete(void *priv, int error)
 {
 	struct ocf_mngt_cache_remove_core_context *context = priv;
 
-	if (error) {
+	OCF_PL_NEXT_ON_SUCCESS_RET(context->pipeline,
-		ocf_pipeline_finish(context->pipeline,
+			error ? -OCF_ERR_WRITE_CACHE : 0);
-				-OCF_ERR_WRITE_CACHE);
-		return;
-	}
-
-	ocf_pipeline_next(context->pipeline);
 }
 
 void ocf_mngt_cache_remove_core(ocf_core_t core,
@@ -670,18 +637,17 @@ void ocf_mngt_cache_remove_core(ocf_core_t core,
 	cache = ocf_core_get_cache(core);
 	core_id = ocf_core_get_id(core);
 
+	if (!cache->mngt_queue)
+		OCF_CMPL_RET(cache, -OCF_ERR_INVAL);
+
 	/* TODO: Make this asynchronous */
-	if (_ocf_cleaning_wait_for_finish(cache, 60 * 1000)) {
+	if (_ocf_cleaning_wait_for_finish(cache, 60 * 1000))
-		cmpl(priv, -OCF_ERR_CACHE_IN_USE);
+		OCF_CMPL_RET(priv, -OCF_ERR_CACHE_IN_USE);
-		return;
-	}
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&ocf_mngt_cache_remove_core_pipeline_props);
-	if (result) {
+	if (result)
-		cmpl(priv, result);
+		OCF_CMPL_RET(priv, result);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 
@@ -737,11 +703,12 @@ void ocf_mngt_cache_detach_core(ocf_core_t core,
 	cache = ocf_core_get_cache(core);
 	core_name = ocf_core_get_name(core);
 
+	if (!cache->mngt_queue)
+		OCF_CMPL_RET(cache, -OCF_ERR_INVAL);
+
 	/* TODO: Make this asynchronous */
-	if (_ocf_cleaning_wait_for_finish(cache, 60 * 1000)) {
+	if (_ocf_cleaning_wait_for_finish(cache, 60 * 1000))
-		cmpl(priv, -OCF_ERR_CACHE_IN_USE);
+		OCF_CMPL_RET(priv, -OCF_ERR_CACHE_IN_USE);
-		return;
-	}
 
 	ocf_core_log(core, log_debug, "Detaching core\n");
 
@@ -754,7 +721,7 @@ void ocf_mngt_cache_detach_core(ocf_core_t core,
 				core_name);
 	}
 
-	cmpl(priv, result);
+	OCF_CMPL_RET(priv, result);
 }
 
 int ocf_mngt_core_set_uuid(ocf_core_t core, const struct ocf_volume_uuid *uuid)

src/mngt/ocf_mngt_flush.c

@@ -582,10 +582,8 @@ static void _ocf_mngt_flush_all_cores_complete(
 			break;
 	}
 
-	if (error) {
+	if (error)
-		ocf_pipeline_finish(context->pipeline, error);
+		OCF_PL_FINISH_RET(context->pipeline, error);
-		return;
-	}
 
 	if (context->op == flush_cache)
 		ocf_cache_log(cache, log_info, "Flushing cache completed\n");
@@ -671,30 +669,26 @@ void ocf_mngt_cache_flush(ocf_cache_t cache, bool interruption,
 	if (!ocf_cache_is_device_attached(cache)) {
 		ocf_cache_log(cache, log_err, "Cannot flush cache - "
 				"cache device is detached\n");
-		cmpl(cache, priv, -OCF_ERR_INVAL);
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_INVAL);
-		return;
 	}
 
 	if (ocf_cache_is_incomplete(cache)) {
 		ocf_cache_log(cache, log_err, "Cannot flush cache - "
 				"cache is in incomplete state\n");
-		cmpl(cache, priv, -OCF_ERR_CACHE_IN_INCOMPLETE_STATE);
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_CACHE_IN_INCOMPLETE_STATE);
-		return;
 	}
 
 	if (!cache->mngt_queue) {
 		ocf_cache_log(cache, log_err,
 				"Cannot flush cache - no flush queue set\n");
-		cmpl(cache, priv, -OCF_ERR_INVAL);
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_INVAL);
-		return;
 	}
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&_ocf_mngt_cache_flush_pipeline_properties);
-	if (result) {
+	if (result)
-		cmpl(cache, priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_NO_MEM);
-		return;
+
-	}
 	context = ocf_pipeline_get_priv(pipeline);
 
 	context->pipeline = pipeline;
@@ -715,10 +709,8 @@ static void _ocf_mngt_flush_core_complete(
 
 	env_atomic_set(&core->flushed, 0);
 
-	if (error) {
+	if (error)
-		ocf_pipeline_finish(context->pipeline, error);
+		OCF_PL_FINISH_RET(context->pipeline, error);
-		return;
-	}
 
 	if (context->op == flush_core)
 		ocf_cache_log(cache, log_info, "Flushing completed\n");
@@ -767,30 +759,26 @@ void ocf_mngt_core_flush(ocf_core_t core, bool interruption,
 	if (!ocf_cache_is_device_attached(cache)) {
 		ocf_cache_log(cache, log_err, "Cannot flush core - "
 				"cache device is detached\n");
-		cmpl(core, priv, -OCF_ERR_INVAL);
+		OCF_CMPL_RET(core, priv, -OCF_ERR_INVAL);
-		return;
 	}
 
 	if (!core->opened) {
 		ocf_core_log(core, log_err, "Cannot flush - core is in "
 				"inactive state\n");
-		cmpl(core, priv, -OCF_ERR_CORE_IN_INACTIVE_STATE);
+		OCF_CMPL_RET(core, priv, -OCF_ERR_CORE_IN_INACTIVE_STATE);
-		return;
 	}
 
 	if (!cache->mngt_queue) {
 		ocf_core_log(core, log_err,
 				"Cannot flush core - no flush queue set\n");
-		cmpl(core, priv, -OCF_ERR_INVAL);
+		OCF_CMPL_RET(core, priv, -OCF_ERR_INVAL);
-		return;
 	}
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&_ocf_mngt_core_flush_pipeline_properties);
-	if (result) {
+	if (result)
-		cmpl(core, priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(core, priv, -OCF_ERR_NO_MEM);
-		return;
+
-	}
 	context = ocf_pipeline_get_priv(pipeline);
 
 	context->pipeline = pipeline;
@@ -816,10 +804,7 @@ static void _ocf_mngt_cache_invalidate(ocf_pipeline_t pipeline, void *priv,
 			context->purge.end_byte);
 	OCF_METADATA_UNLOCK_WR();
 
-	if (result)
+	OCF_PL_NEXT_ON_SUCCESS_RET(context->pipeline, result);
-		ocf_pipeline_finish(context->pipeline, result);
-	else
-		ocf_pipeline_next(context->pipeline);
 }
 
 static
@@ -846,16 +831,14 @@ void ocf_mngt_cache_purge(ocf_cache_t cache,
 	if (!cache->mngt_queue) {
 		ocf_cache_log(cache, log_err,
 				"Cannot purge cache - no flush queue set\n");
-		cmpl(cache, priv, -OCF_ERR_INVAL);
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_INVAL);
-		return;
 	}
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&_ocf_mngt_cache_purge_pipeline_properties);
-	if (result) {
+	if (result)
-		cmpl(cache, priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(cache, priv, -OCF_ERR_NO_MEM);
-		return;
+
-	}
 	context = ocf_pipeline_get_priv(pipeline);
 
 	context->pipeline = pipeline;
@@ -900,18 +883,15 @@ void ocf_mngt_core_purge(ocf_core_t core,
 	if (!cache->mngt_queue) {
 		ocf_core_log(core, log_err,
 				"Cannot purge core - no flush queue set\n");
-		cmpl(core, priv, -OCF_ERR_INVAL);
+		OCF_CMPL_RET(core, priv, -OCF_ERR_INVAL);
-		return;
 	}
 
 	core_size = ocf_volume_get_length(&cache->core[core_id].volume);
 
 	result = ocf_pipeline_create(&pipeline, cache,
 			&_ocf_mngt_core_purge_pipeline_properties);
-	if (result) {
+	if (result)
-		cmpl(core, priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(core, priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	context = ocf_pipeline_get_priv(pipeline);
 

src/ocf_priv.h

@@ -10,4 +10,9 @@
 
 #define OCF_CHECK_NULL(p) ENV_BUG_ON(!(p))
 
+#define OCF_CMPL_RET(args...) ({ \
+	cmpl(args); \
+	return; \
+})
+
 #endif /* __OCF_PRIV_H__ */

src/utils/utils_io.c

@@ -32,10 +32,8 @@ void ocf_submit_volume_flush(ocf_volume_t volume,
 	struct ocf_io *io;
 
 	io = ocf_volume_new_io(volume);
-	if (!io) {
+	if (!io)
-		cmpl(priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	ocf_io_configure(io, 0, 0, OCF_WRITE, 0, 0);
 	ocf_io_set_cmpl(io, cmpl, priv, _ocf_volume_flush_end);
@@ -68,10 +66,8 @@ void ocf_submit_volume_discard(ocf_volume_t volume, uint64_t addr,
 	struct ocf_io *io;
 
 	context = env_vzalloc(sizeof(*context));
-	if (!context) {
+	if (!context)
-		cmpl(priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	env_atomic_set(&context->req_remaining, 1);
 	context->cmpl = cmpl;
@@ -112,10 +108,8 @@ void ocf_submit_write_zeros(ocf_volume_t volume, uint64_t addr,
 	struct ocf_io *io;
 
 	context = env_vzalloc(sizeof(*context));
-	if (!context) {
+	if (!context)
-		cmpl(priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	env_atomic_set(&context->req_remaining, 1);
 	context->cmpl = cmpl;
@@ -179,10 +173,8 @@ void ocf_submit_cache_page(ocf_cache_t cache, uint64_t addr, int dir,
 	int result = 0;
 
 	context = env_vmalloc(sizeof(*context));
-	if (!context) {
+	if (!context)
-		cmpl(priv, -OCF_ERR_NO_MEM);
+		OCF_CMPL_RET(priv, -OCF_ERR_NO_MEM);
-		return;
-	}
 
 	context->cache = cache;
 	context->buffer = buffer;

src/utils/utils_pipeline.h

@@ -131,4 +131,23 @@ void ocf_pipeline_next(ocf_pipeline_t pipeline);
 
 void ocf_pipeline_finish(ocf_pipeline_t pipeline, int error);
 
+#define OCF_PL_NEXT_RET(pipeline) ({ \
+	ocf_pipeline_next(pipeline); \
+	return; \
+})
+
+#define OCF_PL_FINISH_RET(pipeline, error) ({ \
+	ocf_pipeline_finish(pipeline, error); \
+	return; \
+})
+
+#define OCF_PL_NEXT_ON_SUCCESS_RET(pipeline, error) ({ \
+	if (error) \
+		ocf_pipeline_finish(pipeline, error); \
+	else \
+		ocf_pipeline_next(pipeline); \
+	return; \
+})
+
+
 #endif /* __UTILS_PIPELINE_H__ */