Merge pull request #65 from Open-CAS/add-set-flush-queue

Add ocf_cache_set_flush_queue() function

inc/ocf_cache.h

@@ -266,5 +266,12 @@ 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__ */

src/mngt/ocf_mngt_cache.c

@@ -1060,6 +1060,9 @@ static void _ocf_mngt_init_handle_error(ocf_cache_t cache,
 
 	env_mutex_lock(&ctx->lock);
 
+	if (cache->flush_queue)
+		ocf_queue_put(cache->flush_queue);
+
 	list_for_each_entry_safe(queue, tmp_queue, &cache->io_queues, list)
 		ocf_queue_put(queue);
 
@@ -1573,6 +1576,9 @@ static int _ocf_mngt_cache_stop(ocf_cache_t cache)
 	if (env_atomic_read(&cache->attached))
 		result = _ocf_mngt_cache_unplug(cache, true);
 
+	if (cache->flush_queue)
+		ocf_queue_put(cache->flush_queue);
+
 	list_for_each_entry_safe(queue, tmp_queue, &cache->io_queues, list)
 		ocf_queue_put(queue);
 

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 = 0;
+		fctbl[i].attribs.io_queue = cache->flush_queue;
 		fctbl[i].cache = cache;
 		fctbl[i].progress = &progress;
 		fctbl[i].error = &error;
@@ -479,6 +479,12 @@ int ocf_mngt_cache_flush(ocf_cache_t cache, bool interruption)
 		return -OCF_ERR_CACHE_IN_INCOMPLETE_STATE;
 	}
 
+	if (!cache->flush_queue) {
+		ocf_cache_log(cache, log_err,
+				"Cannot flush cache - no flush queue set\n");
+		return -OCF_ERR_INVAL;
+	}
+
 	ocf_cache_log(cache, log_info, "Flushing cache\n");
 
 	_ocf_mngt_begin_flush(cache);
@@ -536,6 +542,12 @@ int ocf_mngt_core_flush(ocf_core_t core, bool interruption)
 		return -OCF_ERR_CORE_IN_INACTIVE_STATE;
 	}
 
+	if (!cache->flush_queue) {
+		ocf_core_log(core, log_err,
+				"Cannot flush core - no flush queue set\n");
+		return -OCF_ERR_INVAL;
+	}
+
 	ocf_core_log(core, log_info, "Flushing\n");
 
 	_ocf_mngt_begin_flush(cache);
@@ -562,6 +574,12 @@ int ocf_mngt_core_purge(ocf_core_t core, bool interruption)
 	cache = ocf_core_get_cache(core);
 	core_id = ocf_core_get_id(core);
 
+	if (!cache->flush_queue) {
+		ocf_core_log(core, log_err,
+				"Cannot purge core - no flush queue set\n");
+		return -OCF_ERR_INVAL;
+	}
+
 	core_size = ocf_volume_get_length(&cache->core[core_id].volume);
 	core_size = core_size ?: ~0ULL;
 
@@ -591,6 +609,12 @@ int ocf_mngt_cache_purge(ocf_cache_t cache, bool interruption)
 
 	OCF_CHECK_NULL(cache);
 
+	if (!cache->flush_queue) {
+		ocf_cache_log(cache, log_err,
+				"Cannot purge cache - no flush queue set\n");
+		return -OCF_ERR_INVAL;
+	}
+
 	_ocf_mngt_begin_flush(cache);
 
 	ocf_cache_log(cache, log_info, "Purging\n");

src/ocf_cache.c

@@ -242,3 +242,17 @@ 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,6 +186,7 @@ struct ocf_cache {
 	env_atomic pending_eviction_clines;
 
 	struct list_head io_queues;
+	ocf_queue_t flush_queue;
 
 	uint16_t ocf_core_inactive_count;
 	struct ocf_core core[OCF_CORE_MAX];