Merge pull request #805 from mmichal10/locking-fixes

Locking and queue improvements

example/simple/src/main.c

@@ -1,5 +1,6 @@
 /*
  * Copyright(c) 2019-2022 Intel Corporation
+ * Copyright(c) 2024 Huawei Technologies
  * SPDX-License-Identifier: BSD-3-Clause
  */
 
@@ -129,23 +130,19 @@ int initialize_cache(ocf_ctx_t ctx, ocf_cache_t *cache)
 	/*
 	 * Create management queue. It will be used for performing various
 	 * asynchronous management operations, such as attaching cache volume
-	 * or adding core object.
+	 * or adding core object. This has to be done before any other
+	 * management operation. Management queue is treated specially,
+	 * and it may not be used for submitting IO requests. It also will not
+	 * be put on the cache stop - we have to put it manually at the end.
 	 */
-	ret = ocf_queue_create(*cache, &cache_priv->mngt_queue, &queue_ops);
+	ret = ocf_queue_create_mngt(*cache, &cache_priv->mngt_queue,
+			&queue_ops);
 	if (ret) {
 		ocf_mngt_cache_stop(*cache, simple_complete, &context);
 		sem_wait(&context.sem);
 		goto err_priv;
 	}
 
-	/*
-	 * Assign management queue to cache. This has to be done before any
-	 * other management operation. Management queue is treated specially,
-	 * and it may not be used for submitting IO requests. It also will not
-	 * be put on the cache stop - we have to put it manually at the end.
-	 */
-	ocf_mngt_cache_set_mngt_queue(*cache, cache_priv->mngt_queue);
-
 	/* Create queue which will be used for IO submission. */
 	ret = ocf_queue_create(*cache, &cache_priv->io_queue, &queue_ops);
 	if (ret)

inc/ocf_mngt.h

@@ -1,5 +1,6 @@
 /*
  * Copyright(c) 2012-2022 Intel Corporation
+ * Copyright(c) 2024 Huawei Technologies
  * SPDX-License-Identifier: BSD-3-Clause
  */
 
@@ -325,17 +326,6 @@ static inline void ocf_mngt_cache_config_set_default(
 int ocf_mngt_cache_start(ocf_ctx_t ctx, ocf_cache_t *cache,
 		struct ocf_mngt_cache_config *cfg, void *priv);
 
-/**
- * @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
  *

inc/ocf_queue.h

@@ -1,5 +1,6 @@
 /*
  * Copyright(c) 2012-2021 Intel Corporation
+ * Copyright(c) 2024 Huawei Technologies
  * SPDX-License-Identifier: BSD-3-Clause
  */
 
@@ -59,6 +60,40 @@ struct ocf_queue_ops {
 int ocf_queue_create(ocf_cache_t cache, ocf_queue_t *queue,
 		const struct ocf_queue_ops *ops);
 
+/**
+ * @brief Allocate mngt queue and assign it to cache
+ *
+ * @param[in] cache Handle to cache instance
+ * @param[out] queue Handle to created queue
+ * @param[in] ops Queue operations
+ *
+ * @return Zero on success, otherwise error code
+ */
+int ocf_queue_create_mngt(ocf_cache_t cache, ocf_queue_t *queue,
+		const struct ocf_queue_ops *ops);
+
+/**
+ * @brief Queue visitor function
+ * @param[in] queue Queue handle
+ * @param[in] ctx Visitor function context
+ *
+ * @return Zero on success, otherwise error code
+ */
+typedef int (*ocf_cache_queue_visitor_t)(ocf_queue_t queue, void *ctx);
+
+/**
+ * @brief Call @visitor for every IO queue
+ *
+ * @param[in] cache Cache instance
+ * @param[in] visitor Function to be called on every queue.
+ *		The visitor function is called in atomic context
+ * @param[in] ctx Context to be passes to @visitor
+ *
+ * @return Zero on success, otherwise error code
+ */
+int ocf_queue_visit(ocf_cache_t cache, ocf_cache_queue_visitor_t visitor,
+		void *ctx);
+
 /**
  * @brief Increase reference counter in queue
  *

src/mngt/ocf_mngt_cache.c

@@ -850,6 +850,11 @@ static int _ocf_mngt_init_new_cache(struct ocf_cache_mngt_init_params *params)
 		goto lock_err;
 	}
 
+	INIT_LIST_HEAD(&cache->io_queues);
+	result = env_spinlock_init(&cache->io_queues_lock);
+	if (result)
+		goto mutex_err;
+
 	ENV_BUG_ON(!ocf_refcnt_inc(&cache->refcnt.cache));
 
 	/* start with freezed metadata ref counter to indicate detached device*/
@@ -865,7 +870,10 @@ static int _ocf_mngt_init_new_cache(struct ocf_cache_mngt_init_params *params)
 
 	return 0;
 
+mutex_err:
+	env_mutex_destroy(&cache->flush_mutex);
 lock_err:
+	ocf_mngt_cache_unlock(cache);
 	ocf_mngt_cache_lock_deinit(cache);
 alloc_err:
 	env_vfree(cache);
@@ -1437,6 +1445,12 @@ static void _ocf_mngt_init_handle_error(ocf_ctx_t ctx,
 	if (!params->flags.cache_alloc)
 		return;
 
+	env_spinlock_destroy(&cache->io_queues_lock);
+
+	env_mutex_destroy(&cache->flush_mutex);
+
+	ocf_mngt_cache_lock_deinit(cache);
+
 	if (params->flags.metadata_inited)
 		ocf_metadata_deinit(cache);
 
@@ -1461,8 +1475,6 @@ static void _ocf_mngt_cache_init(ocf_cache_t cache,
 	cache->conf_meta->promotion_policy_type = params->metadata.promotion_policy;
 	__set_cleaning_policy(cache, ocf_cleaning_default);
 
-	INIT_LIST_HEAD(&cache->io_queues);
-
 	/* Init Partitions */
 	ocf_user_part_init(cache);
 	__init_free(cache);
@@ -2192,6 +2204,9 @@ static void ocf_mngt_cache_remove(ocf_ctx_t ctx, ocf_cache_t cache)
 
 	/* Deinitialize locks */
 	ocf_mngt_cache_lock_deinit(cache);
+
+	env_spinlock_destroy(&cache->io_queues_lock);
+
 	env_mutex_destroy(&cache->flush_mutex);
 
 	/* Remove cache from the list */
@@ -3034,20 +3049,6 @@ int ocf_mngt_cache_start(ocf_ctx_t ctx, ocf_cache_t *cache,
 	return result;
 }
 
-int ocf_mngt_cache_set_mngt_queue(ocf_cache_t cache, ocf_queue_t queue)
-{
-	OCF_CHECK_NULL(cache);
-	OCF_CHECK_NULL(queue);
-
-	if (cache->mngt_queue)
-		return -OCF_ERR_INVAL;
-
-	ocf_queue_get(queue);
-	cache->mngt_queue = queue;
-
-	return 0;
-}
-
 static void _ocf_mngt_cache_attach_complete(ocf_cache_t cache, void *priv1,
 		void *priv2, int error)
 {

src/mngt/ocf_mngt_common.c

@@ -1,5 +1,6 @@
 /*
  * Copyright(c) 2012-2021 Intel Corporation
+ * Copyright(c) 2024 Huawei Technologies
  * SPDX-License-Identifier: BSD-3-Clause
  */
 
@@ -275,10 +276,10 @@ static int _ocf_mngt_cache_trylock(ocf_cache_t cache,
 	if (env_bit_test(ocf_cache_state_stopping, &cache->cache_state)) {
 		/* Cache already stopping, do not allow any operation */
 		unlock_fn(&cache->lock);
-		return -OCF_ERR_CACHE_NOT_EXIST;
+		result = -OCF_ERR_CACHE_NOT_EXIST;
 	}
 
-	return 0;
+	return result;
 }
 
 static void _ocf_mngt_cache_unlock(ocf_cache_t cache,

src/ocf_cache_priv.h

@@ -107,6 +107,8 @@ struct ocf_cache {
 	struct ocf_cleaner cleaner;
 
 	struct list_head io_queues;
+	env_spinlock io_queues_lock;
+
 	ocf_promotion_policy_t promotion_policy;
 
 	struct {

src/ocf_queue.c

@@ -14,28 +14,20 @@
 #include "engine/cache_engine.h"
 #include "ocf_def_priv.h"
 
-int ocf_queue_create(ocf_cache_t cache, ocf_queue_t *queue,
+int _ocf_queue_create(ocf_cache_t cache, ocf_queue_t *queue,
 		const struct ocf_queue_ops *ops)
 {
 	ocf_queue_t tmp_queue;
 	int result;
 
-	OCF_CHECK_NULL(cache);
-
-	result = ocf_mngt_cache_get(cache);
-	if (result)
-		return result;
-
 	tmp_queue = env_zalloc(sizeof(*tmp_queue), ENV_MEM_NORMAL);
 	if (!tmp_queue) {
-		ocf_mngt_cache_put(cache);
 		return -OCF_ERR_NO_MEM;
 	}
 
 	env_atomic_set(&tmp_queue->io_no, 0);
 	result = env_spinlock_init(&tmp_queue->io_list_lock);
 	if (result) {
-		ocf_mngt_cache_put(cache);
 		env_free(tmp_queue);
 		return result;
 	}
@@ -46,14 +38,88 @@ int ocf_queue_create(ocf_cache_t cache, ocf_queue_t *queue,
 	tmp_queue->cache = cache;
 	tmp_queue->ops = ops;
 
+	*queue = tmp_queue;
+
+	return 0;
+}
+
+int ocf_queue_create(ocf_cache_t cache, ocf_queue_t *queue,
+		const struct ocf_queue_ops *ops)
+{
+	ocf_queue_t tmp_queue;
+	int result;
+	unsigned long flags = 0;
+
+	OCF_CHECK_NULL(cache);
+
+	result = ocf_mngt_cache_get(cache);
+	if (result)
+		return result;
+
+	result = _ocf_queue_create(cache, &tmp_queue, ops);
+	if (result) {
+		ocf_mngt_cache_put(cache);
+		return result;
+	}
+
 	result = ocf_queue_seq_cutoff_init(tmp_queue);
 	if (result) {
 		ocf_mngt_cache_put(cache);
 		env_free(tmp_queue);
 		return result;
 	}
 
+	env_spinlock_lock_irqsave(&cache->io_queues_lock, flags);
 	list_add(&tmp_queue->list, &cache->io_queues);
+	env_spinlock_unlock_irqrestore(&cache->io_queues_lock, flags);
+
+	*queue = tmp_queue;
+
+	return 0;
+}
+
+int ocf_queue_visit(ocf_cache_t cache, ocf_cache_queue_visitor_t visitor,
+		void *ctx)
+{
+	ocf_queue_t queue;
+	int result = 0;
+	unsigned long flags = 0;
+
+	env_spinlock_lock_irqsave(&cache->io_queues_lock, flags);
+
+	list_for_each_entry(queue, &cache->io_queues, list) {
+		result = visitor(queue, ctx);
+		if (result)
+			break;
+	}
+
+	env_spinlock_unlock_irqrestore(&cache->io_queues_lock, flags);
+
+	return result;
+}
+
+int ocf_queue_create_mngt(ocf_cache_t cache, ocf_queue_t *queue,
+		const struct ocf_queue_ops *ops)
+{
+	ocf_queue_t tmp_queue;
+	int result;
+
+	OCF_CHECK_NULL(cache);
+
+	if (cache->mngt_queue)
+		return -OCF_ERR_INVAL;
+
+	result = ocf_mngt_cache_get(cache);
+	if (result)
+		return result;
+
+	result = _ocf_queue_create(cache, &tmp_queue, ops);
+	if (result) {
+		ocf_mngt_cache_put(cache);
+		return result;
+	}
+
+	cache->mngt_queue = tmp_queue;
 
 	*queue = tmp_queue;
 
@@ -69,16 +135,24 @@ void ocf_queue_get(ocf_queue_t queue)
 
 void ocf_queue_put(ocf_queue_t queue)
 {
+	ocf_cache_t cache = queue->cache;
+	unsigned long flags = 0;
+
 	OCF_CHECK_NULL(queue);
 
-	if (env_atomic_dec_return(&queue->ref_count) == 0) {
+	if (env_atomic_dec_return(&queue->ref_count))
+		return;
+
+	queue->ops->stop(queue);
+	if (queue != queue->cache->mngt_queue) {
+		env_spinlock_lock_irqsave(&cache->io_queues_lock, flags);
 		list_del(&queue->list);
-		queue->ops->stop(queue);
+		env_spinlock_unlock_irqrestore(&cache->io_queues_lock, flags);
 		ocf_queue_seq_cutoff_deinit(queue);
-		ocf_mngt_cache_put(queue->cache);
-		env_spinlock_destroy(&queue->io_list_lock);
-		env_free(queue);
 	}
+	ocf_mngt_cache_put(queue->cache);
+	env_spinlock_destroy(&queue->io_list_lock);
+	env_free(queue);
 }
 
 void ocf_io_handle(struct ocf_io *io, void *opaque)

src/ocf_stats.c

@@ -1,5 +1,6 @@
 /*
  * Copyright(c) 2012-2021 Intel Corporation
+ * Copyright(c) 2024 Huawei Technologies
  * SPDX-License-Identifier: BSD-3-Clause
  */
 
@@ -415,9 +416,6 @@ void ocf_core_update_stats(ocf_core_t core, struct ocf_io *io)
 	OCF_CHECK_NULL(core);
 	OCF_CHECK_NULL(io);
 
-	core_id = ocf_core_get_id(core);
-	cache = ocf_core_get_cache(core);
-
 	stats = &core->counters->debug_stats;
 
 	idx = to_packet_idx(io->bytes);