From 404b976109e023c135d90512679c4415ab3e9b6f Mon Sep 17 00:00:00 2001
From: Robert Baldyga <robert.baldyga@intel.com>
Date: Fri, 15 Feb 2019 18:39:26 +0100
Subject: [PATCH] Make cleaner asynchronous

Signed-off-by: Robert Baldyga <robert.baldyga@intel.com>
---
 inc/ocf_cleaner.h                             |  31 ++-
 src/cleaning/acp.c                            | 187 +++++++-------
 src/cleaning/acp.h                            |  24 +-
 src/cleaning/alru.c                           | 229 ++++++++++--------
 src/cleaning/alru.h                           |  20 +-
 src/cleaning/cleaning.c                       |  54 +++--
 src/cleaning/cleaning.h                       |  44 ++--
 src/mngt/ocf_mngt_flush.c                     |   1 -
 src/ocf_cache_priv.h                          |   2 -
 src/utils/utils_cleaner.c                     |  45 ----
 src/utils/utils_cleaner.h                     |  16 --
 .../cleaning.c/ocf_cleaner_run_test.c         |  24 +-
 12 files changed, 341 insertions(+), 336 deletions(-)

diff --git a/inc/ocf_cleaner.h b/inc/ocf_cleaner.h
index 85c0df7..42c218b 100644
--- a/inc/ocf_cleaner.h
+++ b/inc/ocf_cleaner.h
@@ -12,15 +12,38 @@
  *
  */
 
+/**
+ * @brief OCF Cleaner completion
+ *
+ * @note Completion function for cleaner
+ *
+ * @param[in] cleaner Cleaner instance
+ * @param[in] interval Time to sleep before next cleaner iteration
+ */
+typedef void (*ocf_cleaner_end_t)(ocf_cleaner_t cleaner, uint32_t interval);
+
+/**
+ * @brief Set cleaner completion function
+ *
+ * @param[in] cleaner Cleaner instance
+ * @param[in] fn Completion function
+ */
+void ocf_cleaner_set_cmpl(ocf_cleaner_t cleaner, ocf_cleaner_end_t fn);
+
+/**
+ * @brief Set cleaner queue
+ *
+ * @param[in] cleaner Cleaner instance
+ * @param[in] io_queue Queue number
+ */
+void ocf_cleaner_set_io_queue(ocf_cleaner_t cleaner, uint32_t io_queue);
+
 /**
  * @brief Run cleaner
  *
  * @param[in] c Cleaner instance to run
- * @param[in] io_queue I/O queue to which cleaner requests should be submitted
- *
- * @retval Hint when to run cleaner next time. Value expressed in miliseconds.
  */
-uint32_t ocf_cleaner_run(ocf_cleaner_t c, uint32_t io_queue);
+void ocf_cleaner_run(ocf_cleaner_t c);
 
 /**
  * @brief Set cleaner private data
diff --git a/src/cleaning/acp.c b/src/cleaning/acp.c
index 7a44cb3..2f3f528 100644
--- a/src/cleaning/acp.c
+++ b/src/cleaning/acp.c
@@ -123,6 +123,12 @@ struct acp_context {
 	 perform_cleaning */
 	struct acp_state state;
 
+	/* cache handle */
+	ocf_cache_t cache;
+
+	/* cleaner completion callback */
+	ocf_cleaner_end_t cmpl;
+
 #if 1 == OCF_ACP_DEBUG
 	/* debug only */
 	uint64_t checksum;
@@ -146,7 +152,7 @@ struct acp_core_line_info
 
 static struct acp_context *_acp_get_ctx_from_cache(struct ocf_cache *cache)
 {
-	return cache->cleaning_policy_context;
+	return cache->cleaner.cleaning_policy_context;
 }
 
 static struct acp_cleaning_policy_meta* _acp_meta_get(
@@ -227,8 +233,8 @@ void cleaning_policy_acp_deinitialize(struct ocf_cache *cache)
 {
 	_acp_remove_cores(cache);
 
-	env_vfree(cache->cleaning_policy_context);
-	cache->cleaning_policy_context = NULL;
+	env_vfree(cache->cleaner.cleaning_policy_context);
+	cache->cleaner.cleaning_policy_context = NULL;
 }
 
 static void _acp_rebuild(struct ocf_cache *cache)
@@ -282,14 +288,15 @@ int cleaning_policy_acp_initialize(struct ocf_cache *cache,
 			1U << (sizeof(acp->chunk_info[0][0].num_dirty) * 8));
 #endif
 
-	ENV_BUG_ON(cache->cleaning_policy_context);
+	ENV_BUG_ON(cache->cleaner.cleaning_policy_context);
 
-	cache->cleaning_policy_context = env_vzalloc(sizeof(struct acp_context));
-	if (!cache->cleaning_policy_context) {
+	acp = env_vzalloc(sizeof(*acp));
+	if (!acp) {
 		ocf_cache_log(cache, log_err, "acp context allocation error\n");
 		return -OCF_ERR_NO_MEM;
 	}
-	acp = cache->cleaning_policy_context;
+	cache->cleaner.cleaning_policy_context = acp;
+	acp->cache = cache;
 
 	env_rwsem_init(&acp->chunks_lock);
 
@@ -412,40 +419,6 @@ static void _acp_handle_flush_error(struct ocf_cache *cache,
 	}
 }
 
-/* called after flush request completed */
-static void _acp_flush_end(
-		struct ocf_cache *cache,
-		struct acp_context *acp)
-{
-	struct acp_flush_context *flush = &acp->flush;
-	int i;
-
-	for (i = 0; i < flush->size; i++) {
-		ocf_cache_line_unlock_rd(cache, flush->data[i].cache_line);
-		ACP_DEBUG_END(acp, flush->data[i].cache_line);
-	}
-
-	if (flush->error)
-		_acp_handle_flush_error(cache, acp);
-}
-
-/* flush data  */
-static void _acp_flush(struct ocf_cache *cache, struct acp_context *acp,
-		uint32_t io_queue, struct acp_flush_context *flush)
-{
-	struct ocf_cleaner_attribs attribs = {
-		.cache_line_lock = false,
-		.metadata_locked = false,
-		.do_sort = false,
-		.io_queue = io_queue,
-	};
-
-	flush->error = ocf_cleaner_do_flush_data(cache, flush->data,
-			flush->size, &attribs);
-
-	_acp_flush_end(cache, acp);
-}
-
 static inline bool _acp_can_clean_chunk(struct ocf_cache *cache,
 		struct acp_chunk_info *chunk)
 {
@@ -456,12 +429,11 @@ static inline bool _acp_can_clean_chunk(struct ocf_cache *cache,
 					!chunk->next_cleaning_timestamp));
 }
 
-static struct acp_chunk_info *_acp_get_cleaning_candidate(
-		struct ocf_cache *cache)
+static struct acp_chunk_info *_acp_get_cleaning_candidate(ocf_cache_t cache)
 {
 	int i;
 	struct acp_chunk_info *cur;
-	struct acp_context *acp = cache->cleaning_policy_context;
+	struct acp_context *acp = cache->cleaner.cleaning_policy_context;
 
 	ACP_LOCK_CHUNKS_RD();
 
@@ -480,72 +452,107 @@ static struct acp_chunk_info *_acp_get_cleaning_candidate(
 	return NULL;
 }
 
-#define CHUNK_FINISHED -1
+/* called after flush request completed */
+static void _acp_flush_end(void *priv, int error)
+{
+	struct acp_cleaning_policy_config *config;
+	struct acp_context *acp = priv;
+	struct acp_flush_context *flush = &acp->flush;
+	ocf_cache_t cache = acp->cache;
+	int i;
 
-/* clean at most 'flush_max_buffers' cache lines from given chunk, starting
- * at given cache line */
-static int _acp_clean(struct ocf_cache *cache, uint32_t io_queue,
-		struct acp_chunk_info *chunk, unsigned start,
+	config = (void *)&cache->conf_meta->cleaning[ocf_cleaning_acp].data;
+
+	for (i = 0; i < flush->size; i++) {
+		ocf_cache_line_unlock_rd(cache, flush->data[i].cache_line);
+		ACP_DEBUG_END(acp, flush->data[i].cache_line);
+	}
+
+	if (error) {
+		flush->error = error;
+		_acp_handle_flush_error(cache, acp);
+	}
+
+	ACP_DEBUG_CHECK(acp);
+
+	acp->cmpl(&cache->cleaner, config->thread_wakeup_time);
+}
+
+/* flush data  */
+static void _acp_flush(struct acp_context *acp)
+{
+	ocf_cache_t cache = acp->cache;
+	struct ocf_cleaner_attribs attribs = {
+		.cmpl_context = acp,
+		.cmpl_fn = _acp_flush_end,
+		.cache_line_lock = false,
+		.do_sort = false,
+		.io_queue = cache->cleaner.io_queue,
+	};
+
+	ocf_cleaner_do_flush_data_async(cache, acp->flush.data,
+				acp->flush.size, &attribs);
+}
+
+static bool _acp_prepare_flush_data(struct acp_context *acp,
 		uint32_t flush_max_buffers)
 {
-	struct acp_context *acp = _acp_get_ctx_from_cache(cache);
-	size_t lines_per_chunk = ACP_CHUNK_SIZE /
-			ocf_line_size(cache);
+	ocf_cache_t cache = acp->cache;
+	struct acp_state *state = &acp->state;
+	struct acp_chunk_info *chunk = state->chunk;
+	size_t lines_per_chunk = ACP_CHUNK_SIZE / ocf_line_size(cache);
 	uint64_t first_core_line = chunk->chunk_id * lines_per_chunk;
-	unsigned i;
 
 	OCF_DEBUG_PARAM(cache, "lines per chunk %llu chunk %llu "
-			"first_core_line %llu\n",
-			(uint64_t)lines_per_chunk,
-			chunk->chunk_id,
-			first_core_line);
-
-	ACP_DEBUG_INIT(acp);
+			"first_core_line %llu\n", (uint64_t)lines_per_chunk,
+			chunk->chunk_id, first_core_line);
 
 	acp->flush.size = 0;
 	acp->flush.chunk = chunk;
-	for (i = start; i < lines_per_chunk && acp->flush.size < flush_max_buffers ; i++) {
-		uint64_t core_line = first_core_line + i;
+	for (; state->iter < lines_per_chunk &&
+			acp->flush.size < flush_max_buffers; state->iter++) {
+		uint64_t core_line = first_core_line + state->iter;
 		ocf_cache_line_t cache_line;
 
 		cache_line = _acp_trylock_dirty(cache, chunk->core_id, core_line);
 		if (cache_line == cache->device->collision_table_entries)
 			continue;
 
-		acp->flush.data[acp->flush.size].core_id =  chunk->core_id;
-		acp->flush.data[acp->flush.size].core_line =  core_line;
+		ACP_DEBUG_BEGIN(acp, cache_line);
+
+		acp->flush.data[acp->flush.size].core_id = chunk->core_id;
+		acp->flush.data[acp->flush.size].core_line = core_line;
 		acp->flush.data[acp->flush.size].cache_line = cache_line;
 		acp->flush.size++;
-		ACP_DEBUG_BEGIN(acp, cache_line);
 	}
 
-	if (acp->flush.size > 0) {
-		_acp_flush(cache, acp, io_queue, &acp->flush);
+	if (state->iter == lines_per_chunk) {
+		/* reached end of chunk - reset state */
+		state->in_progress = false;
 	}
 
-	ACP_DEBUG_CHECK(acp);
-
-	return (i == lines_per_chunk) ? CHUNK_FINISHED : i;
+	return (acp->flush.size > 0);
 }
 
-#define NOTHING_TO_CLEAN 0
-#define MORE_TO_CLEAN 1
-
 /* Clean at most 'flush_max_buffers' cache lines from current or newly
  * selected chunk */
-static int _acp_clean_iteration(struct ocf_cache *cache, uint32_t io_queue,
-		uint32_t flush_max_buffers)
+void cleaning_policy_acp_perform_cleaning(struct ocf_cache *cache,
+		ocf_cleaner_end_t cmpl)
 {
+	struct acp_cleaning_policy_config *config;
 	struct acp_context *acp = _acp_get_ctx_from_cache(cache);
 	struct acp_state *state = &acp->state;
 
+	acp->cmpl = cmpl;
+
 	if (!state->in_progress) {
 		/* get next chunk to clean */
 		state->chunk = _acp_get_cleaning_candidate(cache);
 
 		if (!state->chunk) {
 			/* nothing co clean */
-			return  NOTHING_TO_CLEAN;
+			cmpl(&cache->cleaner, ACP_BACKOFF_TIME_MS);
+			return;
 		}
 
 		/* new cleaning cycle - reset state */
@@ -553,34 +560,14 @@ static int _acp_clean_iteration(struct ocf_cache *cache, uint32_t io_queue,
 		state->in_progress = true;
 	}
 
-	state->iter = _acp_clean(cache, io_queue, state->chunk, state->iter,
-					flush_max_buffers);
-
-	if (state->iter == CHUNK_FINISHED) {
-		/* reached end of chunk - reset state */
-		state->in_progress = false;
-	}
-
-
-	return MORE_TO_CLEAN;
-}
-
-int cleaning_policy_acp_perform_cleaning(struct ocf_cache *cache,
-		uint32_t io_queue)
-{
-	struct acp_cleaning_policy_config *config;
-	int ret;
+	ACP_DEBUG_INIT(acp);
 
 	config = (void *)&cache->conf_meta->cleaning[ocf_cleaning_acp].data;
 
-	if (NOTHING_TO_CLEAN == _acp_clean_iteration(cache, io_queue,
-					config->flush_max_buffers)) {
-		ret = ACP_BACKOFF_TIME_MS;
-	} else {
-		ret = config->thread_wakeup_time;
-	}
-
-	return ret;
+	if (_acp_prepare_flush_data(acp, config->flush_max_buffers))
+		_acp_flush(acp);
+	else
+		_acp_flush_end(acp, 0);
 }
 
 static void _acp_update_bucket(struct acp_context *acp,
diff --git a/src/cleaning/acp.h b/src/cleaning/acp.h
index f2b6c9d..1693b93 100644
--- a/src/cleaning/acp.h
+++ b/src/cleaning/acp.h
@@ -8,32 +8,30 @@
 
 #include "cleaning.h"
 
-void cleaning_policy_acp_setup(struct ocf_cache *cache);
+void cleaning_policy_acp_setup(ocf_cache_t cache);
 
-int cleaning_policy_acp_initialize(struct ocf_cache *cache,
-		int init_metadata);
+int cleaning_policy_acp_initialize(ocf_cache_t cache, int init_metadata);
 
-void cleaning_policy_acp_deinitialize(struct ocf_cache *cache);
+void cleaning_policy_acp_deinitialize(ocf_cache_t cache);
 
-int cleaning_policy_acp_perform_cleaning(struct ocf_cache *cache,
-		uint32_t io_queue);
+void cleaning_policy_acp_perform_cleaning(ocf_cache_t cache,
+		ocf_cleaner_end_t cmpl);
 
-void cleaning_policy_acp_init_cache_block(struct ocf_cache *cache,
+void cleaning_policy_acp_init_cache_block(ocf_cache_t cache,
 		uint32_t cache_line);
 
-void cleaning_policy_acp_set_hot_cache_line(struct ocf_cache *cache,
+void cleaning_policy_acp_set_hot_cache_line(ocf_cache_t cache,
 		uint32_t cache_line);
 
-void cleaning_policy_acp_purge_block(struct ocf_cache *cache,
-		uint32_t cache_line);
+void cleaning_policy_acp_purge_block(ocf_cache_t cache, uint32_t cache_line);
 
-int cleaning_policy_acp_purge_range(struct ocf_cache *cache,
+int cleaning_policy_acp_purge_range(ocf_cache_t cache,
 		int core_id, uint64_t start_byte, uint64_t end_byte);
 
-int cleaning_policy_acp_set_cleaning_param(struct ocf_cache *cache,
+int cleaning_policy_acp_set_cleaning_param(ocf_cache_t cache,
 		uint32_t param_id, uint32_t param_value);
 
-int cleaning_policy_acp_get_cleaning_param(struct ocf_cache *cache,
+int cleaning_policy_acp_get_cleaning_param(ocf_cache_t cache,
 		uint32_t param_id, uint32_t *param_value);
 
 int cleaning_policy_acp_add_core(ocf_cache_t cache, ocf_core_id_t core_id);
diff --git a/src/cleaning/alru.c b/src/cleaning/alru.c
index 45a36ef..6d0373e 100644
--- a/src/cleaning/alru.c
+++ b/src/cleaning/alru.c
@@ -49,6 +49,17 @@ struct flush_merge_struct {
 	uint64_t core_sector;
 };
 
+struct alru_flush_ctx {
+	struct ocf_cleaner_attribs attribs;
+	struct ocf_user_part *parts[OCF_IO_CLASS_MAX];
+	int part_id;
+	uint32_t clines_no;
+	ocf_cache_t cache;
+	ocf_cleaner_end_t cmpl;
+	struct flush_data *flush_data;
+	size_t flush_data_limit;
+};
+
 /* -- Start of ALRU functions -- */
 
 
@@ -412,7 +423,6 @@ static void _alru_rebuild(struct ocf_cache *cache)
 static int cleaning_policy_alru_initialize_part(struct ocf_cache *cache,
 		struct ocf_user_part *part, int init_metadata)
 {
-
 	if (init_metadata) {
 		/* ALRU initialization */
 		env_atomic_set(&part->runtime->cleaning.policy.alru.size, 0);
@@ -443,6 +453,15 @@ int cleaning_policy_alru_initialize(struct ocf_cache *cache, int init_metadata)
 {
 	struct ocf_user_part *part;
 	ocf_part_id_t part_id;
+	struct alru_flush_ctx *fctx;
+
+	fctx = env_vzalloc(sizeof(*fctx));
+	if (!fctx) {
+		ocf_cache_log(cache, log_err, "alru ctx allocation error\n");
+		return -OCF_ERR_NO_MEM;
+	}
+
+	cache->cleaner.cleaning_policy_context = fctx;
 
 	for_each_part(cache, part, part_id) {
 		cleaning_policy_alru_initialize_part(cache,
@@ -455,6 +474,12 @@ int cleaning_policy_alru_initialize(struct ocf_cache *cache, int init_metadata)
 	return 0;
 }
 
+void cleaning_policy_alru_deinitialize(struct ocf_cache *cache)
+{
+	env_vfree(cache->cleaner.cleaning_policy_context);
+	cache->cleaner.cleaning_policy_context = NULL;
+}
+
 int cleaning_policy_alru_set_cleaning_param(ocf_cache_t cache,
 		uint32_t param_id, uint32_t param_value)
 {
@@ -590,7 +615,7 @@ static void get_parts_sorted(struct ocf_user_part **parts,
 			cmp_ocf_user_parts, swp_ocf_user_part);
 }
 
-static int clean_later(ocf_cache_t cache, uint32_t *delta)
+static bool clean_later(ocf_cache_t cache, uint32_t *delta)
 {
 	struct alru_cleaning_policy_config *config;
 
@@ -604,6 +629,33 @@ static int clean_later(ocf_cache_t cache, uint32_t *delta)
 	return false;
 }
 
+static bool is_cleanup_possible(ocf_cache_t cache)
+{
+	struct alru_cleaning_policy_config *config;
+	uint32_t delta;
+
+	config = (void *)&cache->conf_meta->cleaning[ocf_cleaning_alru].data;
+
+	if (check_for_io_activity(cache, config)) {
+		OCF_DEBUG_PARAM(cache, "IO activity detected");
+		return false;
+	}
+
+	if (clean_later(cache, &delta)) {
+		OCF_DEBUG_PARAM(cache,
+			"Cleaning policy configured to clean later "
+			"delta=%u wake_up=%u", delta,
+			config->thread_wakeup_time);
+		return false;
+	}
+
+	//Cleaning policy configured to not clean anything
+	if (config->flush_max_buffers == 0)
+		return false;
+
+	return true;
+}
+
 static void get_block_to_flush(struct flush_data* dst,
 		ocf_cache_line_t cache_line, struct ocf_cache* cache)
 {
@@ -618,7 +670,7 @@ static void get_block_to_flush(struct flush_data* dst,
 	dst->core_line = core_line;
 }
 
-static int more_blocks_to_flush(struct ocf_cache *cache,
+static bool more_blocks_to_flush(struct ocf_cache *cache,
 		ocf_cache_line_t cache_line, uint32_t last_access)
 {
 	struct cleaning_policy_meta policy;
@@ -634,7 +686,7 @@ static int more_blocks_to_flush(struct ocf_cache *cache,
 	return true;
 }
 
-static int block_is_busy(struct ocf_cache *cache,
+static bool block_is_busy(struct ocf_cache *cache,
 		ocf_cache_line_t cache_line)
 {
 	ocf_core_id_t core_id;
@@ -687,116 +739,99 @@ static int get_data_to_flush(struct flush_data *dst, uint32_t clines_no,
 	return to_flush;
 }
 
-static int perform_flushing(int clines_no, struct ocf_cache *cache, uint32_t io_queue,
-		struct flush_data *flush_data, struct ocf_user_part *part)
+static bool alru_do_clean(ocf_cache_t cache, struct alru_flush_ctx *fctx)
 {
-	int to_clean = get_data_to_flush(flush_data, clines_no, cache, part);
+	struct ocf_user_part *part = fctx->parts[fctx->part_id];
+	int to_clean;
 
+	if (!is_cleanup_possible(cache))
+		return false;
+
+	if (OCF_METADATA_LOCK_WR_TRY())
+		return false;
+
+	OCF_REALLOC(&fctx->flush_data, sizeof(fctx->flush_data[0]),
+			fctx->clines_no, &fctx->flush_data_limit);
+	if (!fctx->flush_data) {
+		OCF_METADATA_UNLOCK_WR();
+		ocf_cache_log(cache, log_warn, "No memory to allocate flush "
+				"data for ALRU cleaning policy");
+		return false;
+	}
+
+	to_clean = get_data_to_flush(fctx->flush_data, fctx->clines_no,
+			cache, part);
 	if (to_clean > 0) {
-		struct ocf_cleaner_attribs attribs = {
-			.cache_line_lock = true,
-			.metadata_locked = true,
-			.do_sort = true,
-			.io_queue = io_queue
-		};
-
-		ocf_cleaner_do_flush_data(cache, flush_data,
-				to_clean, &attribs);
+		fctx->clines_no -= to_clean;
+		ocf_cleaner_do_flush_data_async(cache, fctx->flush_data,
+				to_clean, &fctx->attribs);
 	} else {
 		/* Update timestamp only if there are no items to be cleaned */
 		cache->device->runtime_meta->cleaning_thread_access =
 				env_ticks_to_secs(env_get_tick_count());
 	}
 
-	return to_clean;
-}
-
-static int is_cleanup_possible(ocf_cache_t cache)
-{
-	struct alru_cleaning_policy_config *config;
-	uint32_t delta;
-
-	config = (void *)&cache->conf_meta->cleaning[ocf_cleaning_alru].data;
-
-	if (check_for_io_activity(cache, config)) {
-		OCF_DEBUG_PARAM(cache, "IO activity detected");
-		return false;
-	}
-
-	if (clean_later(cache, &delta)) {
-		OCF_DEBUG_PARAM(cache,
-			"Cleaning policy configured to clean later "
-			"delta=%u wake_up=%u", delta,
-			config->thread_wakeup_time);
-		return false;
-	}
-
-	//Cleaning policy configured to not clean anything
-	if (config->flush_max_buffers == 0)
-		return false;
-
-	return true;
-}
-
-static int cleanup(struct ocf_cache *cache, uint32_t clines_no,
-		struct ocf_user_part *part, uint32_t io_queue)
-{
-	struct flush_data *flush_data;
-	size_t flush_data_limit;
-	int flushed_blocks = 0;
-
-	if (!is_cleanup_possible(cache))
-		return flushed_blocks;
-
-	if (OCF_METADATA_LOCK_WR_TRY())
-		return flushed_blocks;
-
-	OCF_REALLOC_INIT(&flush_data, &flush_data_limit);
-	OCF_REALLOC(&flush_data, sizeof(flush_data[0]), clines_no,
-			&flush_data_limit);
-
-	if (!flush_data) {
-		OCF_METADATA_UNLOCK_WR();
-		ocf_cache_log(cache, log_warn, "No memory to allocate flush "
-				"data for ALRU cleaning policy");
-		return flushed_blocks;
-	}
-
-	flushed_blocks = perform_flushing(clines_no, cache, io_queue,
-			flush_data, part);
-
 	OCF_METADATA_UNLOCK_WR();
 
-	OCF_REALLOC_DEINIT(&flush_data, &flush_data_limit);
-
-	return flushed_blocks;
+	return to_clean > 0;
 }
 
-int cleaning_alru_perform_cleaning(ocf_cache_t cache, uint32_t io_queue)
+static void alru_clean(void *priv, int error)
 {
-	struct ocf_user_part *parts[OCF_IO_CLASS_MAX];
-	int part_id = OCF_IO_CLASS_MAX - 1;
 	struct alru_cleaning_policy_config *config;
-	uint32_t clines_no;
+	struct alru_flush_ctx *fctx = priv;
+	ocf_cache_t cache = fctx->cache;
+	int interval;
+
+	while (fctx->clines_no > 0 && --fctx->part_id >= 0) {
+		/*
+		 * The alru_do_clean() function returns true when submitting
+		 * flush request for the io class succeeded. In such case we
+		 * return and wait until flush finishes - then this function
+		 * will be called as completion callback and iteration over
+		 * io classes will continue.
+		 *
+		 * If the processed io class contains nothing to clean, the
+		 * alru_do_clean() function returns false, and then we try to
+		 * clean another io class until we reach last io class or until
+		 * requested number of cache lines will be flushed - then we
+		 * call the completion and finish.
+		 */
+		if (alru_do_clean(cache, fctx))
+			return;
+	}
+
+	OCF_REALLOC_DEINIT(&fctx->flush_data, &fctx->flush_data_limit);
 
 	config = (void *)&cache->conf_meta->cleaning[ocf_cleaning_alru].data;
 
-	get_parts_sorted(parts, cache);
+	interval = (fctx->clines_no > 0) ?
+			config->thread_wakeup_time * 1000 : 0;
 
-	clines_no = config->flush_max_buffers;
-
-	while (part_id >= 0) {
-		clines_no -= cleanup(cache, clines_no,
-				parts[part_id], io_queue);
-
-		if (clines_no > 0)
-			part_id--;
-		else
-			break;
-	}
-
-	if (clines_no > 0)
-		return config->thread_wakeup_time * 1000;
-
-	return 0;
+	fctx->cmpl(&fctx->cache->cleaner, interval);
+}
+
+void cleaning_alru_perform_cleaning(ocf_cache_t cache, ocf_cleaner_end_t cmpl)
+{
+	struct alru_flush_ctx *fctx = cache->cleaner.cleaning_policy_context;
+	struct alru_cleaning_policy_config *config;
+
+	config = (void *)&cache->conf_meta->cleaning[ocf_cleaning_alru].data;
+
+	OCF_REALLOC_INIT(&fctx->flush_data, &fctx->flush_data_limit);
+
+	fctx->attribs.cmpl_context = fctx;
+	fctx->attribs.cmpl_fn = alru_clean;
+	fctx->attribs.cache_line_lock = true;
+	fctx->attribs.do_sort = true;
+	fctx->attribs.io_queue = cache->cleaner.io_queue;
+
+	fctx->clines_no = config->flush_max_buffers;
+	fctx->cache = cache;
+	fctx->cmpl = cmpl;
+
+	get_parts_sorted(fctx->parts, cache);
+	fctx->part_id = OCF_IO_CLASS_MAX;
+
+	alru_clean(fctx, 0);
 }
diff --git a/src/cleaning/alru.h b/src/cleaning/alru.h
index fc48ac4..ce0eb9e 100644
--- a/src/cleaning/alru.h
+++ b/src/cleaning/alru.h
@@ -9,22 +9,22 @@
 #include "cleaning.h"
 #include "alru_structs.h"
 
-void cleaning_policy_alru_setup(struct ocf_cache *cache);
-int cleaning_policy_alru_initialize(struct ocf_cache *cache,
-		int init_metadata);
-void cleaning_policy_alru_init_cache_block(struct ocf_cache *cache,
+void cleaning_policy_alru_setup(ocf_cache_t cache);
+int cleaning_policy_alru_initialize(ocf_cache_t cache, int init_metadata);
+void cleaning_policy_alru_deinitialize(ocf_cache_t cache);
+void cleaning_policy_alru_init_cache_block(ocf_cache_t cache,
 		uint32_t cache_line);
-void cleaning_policy_alru_purge_cache_block(struct ocf_cache *cache,
+void cleaning_policy_alru_purge_cache_block(ocf_cache_t cache,
 		uint32_t cache_line);
-int cleaning_policy_alru_purge_range(struct ocf_cache *cache, int core_id,
+int cleaning_policy_alru_purge_range(ocf_cache_t cache, int core_id,
 		uint64_t start_byte, uint64_t end_byte);
-void cleaning_policy_alru_set_hot_cache_line(struct ocf_cache *cache,
+void cleaning_policy_alru_set_hot_cache_line(ocf_cache_t cache,
 		uint32_t cache_line);
-int cleaning_policy_alru_set_cleaning_param(struct ocf_cache *cache,
+int cleaning_policy_alru_set_cleaning_param(ocf_cache_t cache,
 		uint32_t param_id, uint32_t param_value);
-int cleaning_policy_alru_get_cleaning_param(struct ocf_cache *cache,
+int cleaning_policy_alru_get_cleaning_param(ocf_cache_t cache,
 		uint32_t param_id, uint32_t *param_value);
-int cleaning_alru_perform_cleaning(struct ocf_cache *cache, uint32_t io_queue);
+void cleaning_alru_perform_cleaning(ocf_cache_t cache, ocf_cleaner_end_t cmpl);
 
 #endif
 
diff --git a/src/cleaning/cleaning.c b/src/cleaning/cleaning.c
index 559fcd6..3997e71 100644
--- a/src/cleaning/cleaning.c
+++ b/src/cleaning/cleaning.c
@@ -25,6 +25,7 @@ struct cleaning_policy_ops cleaning_policy_ops[ocf_cleaning_max] = {
 		.purge_range = cleaning_policy_alru_purge_range,
 		.set_hot_cache_line = cleaning_policy_alru_set_hot_cache_line,
 		.initialize = cleaning_policy_alru_initialize,
+		.deinitialize = cleaning_policy_alru_deinitialize,
 		.set_cleaning_param = cleaning_policy_alru_set_cleaning_param,
 		.get_cleaning_param = cleaning_policy_alru_get_cleaning_param,
 		.perform_cleaning = cleaning_alru_perform_cleaning,
@@ -47,16 +48,26 @@ struct cleaning_policy_ops cleaning_policy_ops[ocf_cleaning_max] = {
 	},
 };
 
-int ocf_start_cleaner(struct ocf_cache *cache)
+int ocf_start_cleaner(ocf_cache_t cache)
 {
 	return ctx_cleaner_init(cache->owner, &cache->cleaner);
 }
 
-void ocf_stop_cleaner(struct ocf_cache *cache)
+void ocf_stop_cleaner(ocf_cache_t cache)
 {
 	ctx_cleaner_stop(cache->owner, &cache->cleaner);
 }
 
+void ocf_cleaner_set_cmpl(ocf_cleaner_t cleaner, ocf_cleaner_end_t fn)
+{
+	cleaner->end = fn;
+}
+
+void ocf_cleaner_set_io_queue(ocf_cleaner_t cleaner, uint32_t io_queue)
+{
+	cleaner->io_queue = io_queue;
+}
+
 void ocf_cleaner_set_priv(ocf_cleaner_t c, void *priv)
 {
 	OCF_CHECK_NULL(c);
@@ -75,7 +86,7 @@ ocf_cache_t ocf_cleaner_get_cache(ocf_cleaner_t c)
 	return container_of(c, struct ocf_cache, cleaner);
 }
 
-static int _ocf_cleaner_run_check_dirty_inactive(struct ocf_cache *cache)
+static int _ocf_cleaner_run_check_dirty_inactive(ocf_cache_t cache)
 {
 	int i;
 
@@ -95,29 +106,37 @@ static int _ocf_cleaner_run_check_dirty_inactive(struct ocf_cache *cache)
 	return 1;
 }
 
-uint32_t ocf_cleaner_run(ocf_cleaner_t c, uint32_t io_queue)
+static void ocf_cleaner_run_complete(ocf_cleaner_t cleaner, uint32_t interval)
 {
-	struct ocf_cache *cache;
-	ocf_cleaning_t clean_type;
-	int sleep = SLEEP_TIME_MS;
+	ocf_cache_t cache = ocf_cleaner_get_cache(cleaner);
 
-	cache = ocf_cleaner_get_cache(c);
+	env_rwsem_up_write(&cache->lock);
+	cleaner->end(cleaner, interval);
+}
+
+void ocf_cleaner_run(ocf_cleaner_t cleaner)
+{
+	ocf_cache_t cache = ocf_cleaner_get_cache(cleaner);
+	ocf_cleaning_t clean_type;
 
 	/* Do not involve cleaning when cache is not running
 	 * (error, etc.).
 	 */
 	if (!env_bit_test(ocf_cache_state_running, &cache->cache_state) ||
 			ocf_mngt_is_cache_locked(cache)) {
-		return SLEEP_TIME_MS;
+		cleaner->end(cleaner, SLEEP_TIME_MS);
+		return;
 	}
 
 	/* Sleep in case there is management operation in progress. */
-	if (env_rwsem_down_write_trylock(&cache->lock) == 0)
-		return SLEEP_TIME_MS;
+	if (env_rwsem_down_write_trylock(&cache->lock) == 0) {
+		cleaner->end(cleaner, SLEEP_TIME_MS);
+		return;
+	}
 
 	if (_ocf_cleaner_run_check_dirty_inactive(cache)) {
-		env_rwsem_up_write(&cache->lock);
-		return SLEEP_TIME_MS;
+		cleaner->end(cleaner, SLEEP_TIME_MS);
+		return;
 	}
 
 	clean_type = cache->conf_meta->cleaning_policy_type;
@@ -126,12 +145,7 @@ uint32_t ocf_cleaner_run(ocf_cleaner_t c, uint32_t io_queue)
 
 	/* Call cleaning. */
 	if (cleaning_policy_ops[clean_type].perform_cleaning) {
-		sleep = cleaning_policy_ops[clean_type].
-			perform_cleaning(cache, io_queue);
+		cleaning_policy_ops[clean_type].perform_cleaning(cache,
+				ocf_cleaner_run_complete);
 	}
-
-	env_rwsem_up_write(&cache->lock);
-
-	return sleep;
 }
-
diff --git a/src/cleaning/cleaning.h b/src/cleaning/cleaning.h
index dc2039b..d625e26 100644
--- a/src/cleaning/cleaning.h
+++ b/src/cleaning/cleaning.h
@@ -9,6 +9,7 @@
 #include "alru_structs.h"
 #include "nop_structs.h"
 #include "acp_structs.h"
+#include "ocf/ocf_cleaner.h"
 
 #define CLEANING_POLICY_CONFIG_BYTES 256
 #define CLEANING_POLICY_TYPE_MAX 4
@@ -17,7 +18,6 @@ struct ocf_request;
 
 struct cleaning_policy_config {
 	uint8_t data[CLEANING_POLICY_CONFIG_BYTES];
-	struct acp_cleaning_policy_config acp;
 };
 
 struct cleaning_policy {
@@ -37,39 +37,35 @@ struct cleaning_policy_meta {
 };
 
 struct cleaning_policy_ops {
-	void (*setup)(struct ocf_cache *cache);
-	int (*initialize)(struct ocf_cache *cache, int init_metadata);
-	void (*deinitialize)(struct ocf_cache *cache);
-	int (*add_core)(struct ocf_cache *cache, ocf_core_id_t core_id);
-	void (*remove_core)(struct ocf_cache *cache, ocf_core_id_t core_id);
-	void (*init_cache_block)(struct ocf_cache *cache, uint32_t cache_line);
-	void (*purge_cache_block)(struct ocf_cache *cache,
-			uint32_t cache_line);
-	int (*purge_range)(struct ocf_cache *cache, int core_id,
+	void (*setup)(ocf_cache_t cache);
+	int (*initialize)(ocf_cache_t cache, int init_metadata);
+	void (*deinitialize)(ocf_cache_t cache);
+	int (*add_core)(ocf_cache_t cache, ocf_core_id_t core_id);
+	void (*remove_core)(ocf_cache_t cache, ocf_core_id_t core_id);
+	void (*init_cache_block)(ocf_cache_t cache, uint32_t cache_line);
+	void (*purge_cache_block)(ocf_cache_t cache, uint32_t cache_line);
+	int (*purge_range)(ocf_cache_t cache, int core_id,
 			uint64_t start_byte, uint64_t end_byte);
-	void (*set_hot_cache_line)(struct ocf_cache *cache,
-			uint32_t cache_line);
-	int (*set_cleaning_param)(struct ocf_cache *cache,
-			uint32_t param_id, uint32_t param_value);
-	int (*get_cleaning_param)(struct ocf_cache *cache,
-			uint32_t param_id, uint32_t *param_value);
-	/**
-	 * @brief Performs cleaning.
-	 * @return requested time (in ms) of next call
-	 */
-	int (*perform_cleaning)(struct ocf_cache *cache,
-			uint32_t io_queue);
+	void (*set_hot_cache_line)(ocf_cache_t cache, uint32_t cache_line);
+	int (*set_cleaning_param)(ocf_cache_t cache, uint32_t param_id,
+			uint32_t param_value);
+	int (*get_cleaning_param)(ocf_cache_t cache, uint32_t param_id,
+			uint32_t *param_value);
+	void (*perform_cleaning)(ocf_cache_t cache, ocf_cleaner_end_t cmpl);
 	const char *name;
 };
 
 extern struct cleaning_policy_ops cleaning_policy_ops[ocf_cleaning_max];
 
 struct ocf_cleaner {
+	void *cleaning_policy_context;
+	uint32_t io_queue;
+	ocf_cleaner_end_t end;
 	void *priv;
 };
 
-int ocf_start_cleaner(struct ocf_cache *cache);
+int ocf_start_cleaner(ocf_cache_t cache);
 
-void ocf_stop_cleaner(struct ocf_cache *cache);
+void ocf_stop_cleaner(ocf_cache_t cache);
 
 #endif
diff --git a/src/mngt/ocf_mngt_flush.c b/src/mngt/ocf_mngt_flush.c
index a8612f4..7ba8807 100644
--- a/src/mngt/ocf_mngt_flush.c
+++ b/src/mngt/ocf_mngt_flush.c
@@ -316,7 +316,6 @@ static int _ocf_mngt_flush_containers(ocf_cache_t cache,
 
 	for (i = 0; i < fcnum; i++) {
 		fctbl[i].attribs.cache_line_lock = true;
-		fctbl[i].attribs.metadata_locked = true;
 		fctbl[i].attribs.cmpl_context = &fctbl[i];
 		fctbl[i].attribs.cmpl_fn = _ocf_mngt_flush_end;
 		fctbl[i].attribs.io_queue = 0;
diff --git a/src/ocf_cache_priv.h b/src/ocf_cache_priv.h
index 637e366..819025c 100644
--- a/src/ocf_cache_priv.h
+++ b/src/ocf_cache_priv.h
@@ -223,8 +223,6 @@ struct ocf_cache {
 
 	bool use_submit_io_fast;
 
-	void *cleaning_policy_context;
-
 	struct ocf_trace trace;
 
 	void *priv;
diff --git a/src/utils/utils_cleaner.c b/src/utils/utils_cleaner.c
index f599890..eedc92e 100644
--- a/src/utils/utils_cleaner.c
+++ b/src/utils/utils_cleaner.c
@@ -31,12 +31,6 @@
 #define OCF_DEBUG_PARAM(cache, format, ...)
 #endif
 
-
-struct ocf_cleaner_sync {
-	env_completion cmpl;
-	int error;
-};
-
 /*
  * Allocate cleaning request
  */
@@ -948,17 +942,6 @@ void ocf_cleaner_fire(struct ocf_cache *cache,
 	ocf_req_put(master);
 }
 
-static void ocf_cleaner_sync_end(void *private_data, int error)
-{
-	struct ocf_cleaner_sync *sync = private_data;
-
-	OCF_DEBUG_TRACE(req->cache);
-	if (error)
-		sync->error = error;
-
-	env_completion_complete(&sync->cmpl);
-}
-
 static int _ocf_cleaner_do_flush_data_getter(struct ocf_cache *cache,
 		void *context, uint32_t item, ocf_cache_line_t *line)
 {
@@ -972,34 +955,6 @@ static int _ocf_cleaner_do_flush_data_getter(struct ocf_cache *cache,
 	}
 }
 
-int ocf_cleaner_do_flush_data(struct ocf_cache *cache,
-		struct flush_data *flush, uint32_t count,
-		struct ocf_cleaner_attribs *attribs)
-{
-	struct ocf_cleaner_sync sync;
-
-	env_completion_init(&sync.cmpl);
-	sync.error = 0;
-	attribs->cmpl_context = &sync;
-	attribs->cmpl_fn = ocf_cleaner_sync_end;
-	attribs->getter = _ocf_cleaner_do_flush_data_getter;
-	attribs->getter_context = flush;
-	attribs->count = count;
-
-	ocf_cleaner_fire(cache, attribs);
-
-	if (attribs->metadata_locked)
-		OCF_METADATA_UNLOCK_WR();
-
-	env_completion_wait(&sync.cmpl);
-
-	if (attribs->metadata_locked)
-		OCF_METADATA_LOCK_WR();
-
-	attribs->cmpl_context = NULL;
-	return sync.error;
-}
-
 int ocf_cleaner_do_flush_data_async(struct ocf_cache *cache,
 		struct flush_data *flush, uint32_t count,
 		struct ocf_cleaner_attribs *attribs)
diff --git a/src/utils/utils_cleaner.h b/src/utils/utils_cleaner.h
index 78d9636..8e09fe5 100644
--- a/src/utils/utils_cleaner.h
+++ b/src/utils/utils_cleaner.h
@@ -27,8 +27,6 @@ typedef int (*ocf_cleaner_get_item)(struct ocf_cache *cache,
 struct ocf_cleaner_attribs {
 	uint8_t  cache_line_lock : 1;	/*!< Clean under cache line lock */
 
-	uint8_t metadata_locked : 1;  /*< true if caller holds metadata lock */
-
 	uint8_t  do_sort : 1;	/*!< Sort cache lines which will be cleaned */
 
 	uint32_t count; /*!< max number of cache lines to be cleaned */
@@ -87,20 +85,6 @@ struct flush_container {
 void ocf_cleaner_fire(struct ocf_cache *cache,
 		const struct ocf_cleaner_attribs *attribs);
 
-/**
- * @brief Perform cleaning procedure for specified flush data synchronously.
- * Only dirty cache lines will be cleaned.
- *
- * @param cache - Cache instance
- * @param flush - flush data to be cleaned
- * @param count - Count of cache lines to be cleaned
- * @param attribs - Cleaning attributes
- * @return - Cleaning result. 0 - no errors, non zero errors occurred
- */
-int ocf_cleaner_do_flush_data(struct ocf_cache *cache,
-		struct flush_data *flush, uint32_t count,
-		struct ocf_cleaner_attribs *attribs);
-
 /**
  * @brief Perform cleaning procedure for specified flush data. Only dirty
  * cache lines will be cleaned.
diff --git a/tests/unit/tests/cleaning/cleaning.c/ocf_cleaner_run_test.c b/tests/unit/tests/cleaning/cleaning.c/ocf_cleaner_run_test.c
index 9125ddc..eaf32a9 100644
--- a/tests/unit/tests/cleaning/cleaning.c/ocf_cleaner_run_test.c
+++ b/tests/unit/tests/cleaning/cleaning.c/ocf_cleaner_run_test.c
@@ -13,6 +13,9 @@
 
 //<tested_file_path>src/cleaning/cleaning.c</tested_file_path>
 //<tested_function>ocf_cleaner_run</tested_function>
+//<functions_to_leave>
+//ocf_cleaner_set_cmpl
+//</functions_to_leave>
 
 
 #include <stdarg.h>
@@ -133,6 +136,11 @@ int __wrap_cleaning_policy_alru_initialize(struct ocf_cache *cache, int partitio
 
 }
 
+void __wrap_cleaning_policy_alru_deinitialize(ocf_cache_t cache)
+{
+
+}
+
 int __wrap_cleaning_policy_alru_flush_block(struct ocf_cache *cache,
 		                 uint32_t io_queue, uint32_t count, uint32_t *cache_lines,
 				                                  int partition_id, int core_id, uint8_t do_lock)
@@ -152,7 +160,7 @@ void __wrap_cleaning_policy_alru_get_cleaning_parameters(ocf_cache_t cache,
 
 }
 
-int __wrap_cleaning_alru_perform_cleaning(struct ocf_cache *cache, uint32_t io_queue)
+int __wrap_cleaning_alru_perform_cleaning(struct ocf_cache *cache, ocf_cleaner_end_t cmpl)
 {
 	function_called();
 	return mock();
@@ -178,6 +186,11 @@ int __wrap__ocf_cleaner_run_check_dirty_inactive(struct ocf_cache *cache)
 	return mock();
 }
 
+void __wrap_ocf_cleaner_run_complete(ocf_cleaner_t cleaner, uint32_t interval)
+{
+	function_called();
+}
+
 int __wrap_env_bit_test(int nr, const void *addr)
 {
 	function_called();
@@ -195,6 +208,10 @@ void __wrap_env_rwsem_up_write(env_rwsem *s)
 	function_called();
 }
 
+static void cleaner_complete(ocf_cleaner_t cleaner, uint32_t interval)
+{
+	function_called();
+}
 
 /*
  * Tests of functions. Every test name must be written to tests array in main().
@@ -233,10 +250,9 @@ static void ocf_cleaner_run_test01(void **state)
 	expect_function_call(__wrap_cleaning_alru_perform_cleaning);
 	will_return(__wrap_cleaning_alru_perform_cleaning, 0);
 
-	expect_function_call(__wrap_env_rwsem_up_write);
+	ocf_cleaner_set_cmpl(&cache.cleaner, cleaner_complete);
 
-	result = ocf_cleaner_run(&cache.cleaner, io_queue);
-	assert_int_equal(result, 0);
+	ocf_cleaner_run(&cache.cleaner);
 
 	/* Release allocated memory if allocated with test_* functions */