// SPDX-License-Identifier: GPL-2.0-or-later

#include "cache.h"
#include "backing_dev.h"
#include "cache_dev.h"
#include "dm_pcache.h"

static int cache_data_head_init(struct pcache_cache *cache)
{
	struct pcache_cache_segment *next_seg;
	struct pcache_cache_data_head *data_head;

	data_head = get_data_head(cache);
	next_seg = get_cache_segment(cache);
	if (!next_seg)
		return -EBUSY;

	cache_seg_get(next_seg);
	data_head->head_pos.cache_seg = next_seg;
	data_head->head_pos.seg_off = 0;

	return 0;
}

/**
 * cache_data_alloc - Allocate data for a cache key.
 * @cache: Pointer to the cache structure.
 * @key: Pointer to the cache key to allocate data for.
 *
 * This function tries to allocate space from the cache segment specified by the
 * data head. If the remaining space in the segment is insufficient to allocate
 * the requested length for the cache key, it will allocate whatever is available
 * and adjust the key's length accordingly. This function does not allocate
 * space that crosses segment boundaries.
 */
static int cache_data_alloc(struct pcache_cache *cache, struct pcache_cache_key *key)
{
	struct pcache_cache_data_head *data_head;
	struct pcache_cache_pos *head_pos;
	struct pcache_cache_segment *cache_seg;
	u32 seg_remain;
	u32 allocated = 0, to_alloc;
	int ret = 0;

	preempt_disable();
	data_head = get_data_head(cache);
again:
	to_alloc = key->len - allocated;
	if (!data_head->head_pos.cache_seg) {
		seg_remain = 0;
	} else {
		cache_pos_copy(&key->cache_pos, &data_head->head_pos);
		key->seg_gen = key->cache_pos.cache_seg->gen;

		head_pos = &data_head->head_pos;
		cache_seg = head_pos->cache_seg;
		seg_remain = cache_seg_remain(head_pos);
	}

	if (seg_remain > to_alloc) {
		/* If remaining space in segment is sufficient for the cache key, allocate it. */
		cache_pos_advance(head_pos, to_alloc);
		allocated += to_alloc;
		cache_seg_get(cache_seg);
	} else if (seg_remain) {
		/* If remaining space is not enough, allocate the remaining space and adjust the cache key length. */
		cache_pos_advance(head_pos, seg_remain);
		key->len = seg_remain;

		/* Get for key: obtain a reference to the cache segment for the key. */
		cache_seg_get(cache_seg);
		/* Put for head_pos->cache_seg: release the reference for the current head's segment. */
		cache_seg_put(head_pos->cache_seg);
		head_pos->cache_seg = NULL;
	} else {
		/* Initialize a new data head if no segment is available. */
		ret = cache_data_head_init(cache);
		if (ret)
			goto out;

		goto again;
	}

out:
	preempt_enable();

	return ret;
}

static int cache_copy_from_req_bio(struct pcache_cache *cache, struct pcache_cache_key *key,
				struct pcache_request *pcache_req, u32 bio_off)
{
	struct pcache_cache_pos *pos = &key->cache_pos;
	struct pcache_segment *segment;

	segment = &pos->cache_seg->segment;

	return segment_copy_from_bio(segment, pos->seg_off, key->len, pcache_req->bio, bio_off);
}

static int cache_copy_to_req_bio(struct pcache_cache *cache, struct pcache_request *pcache_req,
			    u32 bio_off, u32 len, struct pcache_cache_pos *pos, u64 key_gen)
{
	struct pcache_cache_segment *cache_seg = pos->cache_seg;
	struct pcache_segment *segment = &cache_seg->segment;
	int ret;

	spin_lock(&cache_seg->gen_lock);
	if (key_gen < cache_seg->gen) {
		spin_unlock(&cache_seg->gen_lock);
		return -EINVAL;
	}

	ret = segment_copy_to_bio(segment, pos->seg_off, len, pcache_req->bio, bio_off);
	spin_unlock(&cache_seg->gen_lock);

	return ret;
}

/**
 * miss_read_end_req - Handle the end of a miss read request.
 * @backing_req: Pointer to the request structure.
 * @read_ret: Return value of read.
 *
 * This function is called when a backing request to read data from
 * the backing_dev is completed. If the key associated with the request
 * is empty (a placeholder), it allocates cache space for the key,
 * copies the data read from the bio into the cache, and updates
 * the key's status. If the key has been overwritten by a write
 * request during this process, it will be deleted from the cache
 * tree and no further action will be taken.
 */
static void miss_read_end_req(struct pcache_backing_dev_req *backing_req, int read_ret)
{
	void *priv_data = backing_req->priv_data;
	struct pcache_request *pcache_req = backing_req->req.upper_req;
	struct pcache_cache *cache = backing_req->backing_dev->cache;
	int ret;

	if (priv_data) {
		struct pcache_cache_key *key;
		struct pcache_cache_subtree *cache_subtree;

		key = (struct pcache_cache_key *)priv_data;
		cache_subtree = key->cache_subtree;

		/* if this key was deleted from cache_subtree by a write, key->flags should be cleared,
		 * so if cache_key_empty() return true, this key is still in cache_subtree
		 */
		spin_lock(&cache_subtree->tree_lock);
		if (cache_key_empty(key)) {
			/* Check if the backing request was successful. */
			if (read_ret) {
				cache_key_delete(key);
				goto unlock;
			}

			/* Allocate cache space for the key and copy data from the backing_dev. */
			ret = cache_data_alloc(cache, key);
			if (ret) {
				cache_key_delete(key);
				goto unlock;
			}

			ret = cache_copy_from_req_bio(cache, key, pcache_req, backing_req->req.bio_off);
			if (ret) {
				cache_seg_put(key->cache_pos.cache_seg);
				cache_key_delete(key);
				goto unlock;
			}
			key->flags &= ~PCACHE_CACHE_KEY_FLAGS_EMPTY;
			key->flags |= PCACHE_CACHE_KEY_FLAGS_CLEAN;

			/* Append the key to the cache. */
			ret = cache_key_append(cache, key, false);
			if (ret) {
				cache_seg_put(key->cache_pos.cache_seg);
				cache_key_delete(key);
				goto unlock;
			}
		}
unlock:
		spin_unlock(&cache_subtree->tree_lock);
		cache_key_put(key);
	}
}

/**
 * submit_cache_miss_req - Submit a backing request when cache data is missing
 * @cache: The cache context that manages cache operations
 * @backing_req: The cache request containing information about the read request
 *
 * This function is used to handle cases where a cache read request cannot locate
 * the required data in the cache. When such a miss occurs during `cache_subtree_walk`,
 * it triggers a backing read request to fetch data from the backing storage.
 *
 * If `pcache_req->priv_data` is set, it points to a `pcache_cache_key`, representing
 * a new cache key to be inserted into the cache. The function calls `cache_key_insert`
 * to attempt adding the key. On insertion failure, it releases the key reference and
 * clears `priv_data` to avoid further processing.
 */
static void submit_cache_miss_req(struct pcache_cache *cache, struct pcache_backing_dev_req *backing_req)
{
	if (backing_req->priv_data) {
		struct pcache_cache_key *key;

		/* Attempt to insert the key into the cache if priv_data is set */
		key = (struct pcache_cache_key *)backing_req->priv_data;
		cache_key_insert(&cache->req_key_tree, key, true);
	}
	backing_dev_req_submit(backing_req, false);
}

static void cache_miss_req_free(struct pcache_backing_dev_req *backing_req)
{
	struct pcache_cache_key *key;

	if (backing_req->priv_data) {
		key = backing_req->priv_data;
		backing_req->priv_data = NULL;
		cache_key_put(key); /* for ->priv_data */
		cache_key_put(key); /* for init ref in alloc */
	}

	backing_dev_req_end(backing_req);
}

static struct pcache_backing_dev_req *cache_miss_req_alloc(struct pcache_cache *cache,
							   struct pcache_request *parent,
							   gfp_t gfp_mask)
{
	struct pcache_backing_dev *backing_dev = cache->backing_dev;
	struct pcache_backing_dev_req *backing_req;
	struct pcache_cache_key *key = NULL;
	struct pcache_backing_dev_req_opts req_opts = { 0 };

	req_opts.type = BACKING_DEV_REQ_TYPE_REQ;
	req_opts.gfp_mask = gfp_mask;
	req_opts.req.upper_req = parent;

	backing_req = backing_dev_req_alloc(backing_dev, &req_opts);
	if (!backing_req)
		return NULL;

	key = cache_key_alloc(&cache->req_key_tree, gfp_mask);
	if (!key)
		goto free_backing_req;

	cache_key_get(key);
	backing_req->priv_data = key;

	return backing_req;

free_backing_req:
	cache_miss_req_free(backing_req);
	return NULL;
}

static void cache_miss_req_init(struct pcache_cache *cache,
				struct pcache_backing_dev_req *backing_req,
				struct pcache_request *parent,
				u32 off, u32 len, bool insert_key)
{
	struct pcache_cache_key *key;
	struct pcache_backing_dev_req_opts req_opts = { 0 };

	req_opts.type = BACKING_DEV_REQ_TYPE_REQ;
	req_opts.req.upper_req = parent;
	req_opts.req.req_off = off;
	req_opts.req.len = len;
	req_opts.end_fn = miss_read_end_req;

	backing_dev_req_init(backing_req, &req_opts);

	if (insert_key) {
		key = backing_req->priv_data;
		key->off = parent->off + off;
		key->len = len;
		key->flags |= PCACHE_CACHE_KEY_FLAGS_EMPTY;
	} else {
		key = backing_req->priv_data;
		backing_req->priv_data = NULL;
		cache_key_put(key);
		cache_key_put(key);
	}
}

static struct pcache_backing_dev_req *get_pre_alloc_req(struct pcache_cache_subtree_walk_ctx *ctx)
{
	struct pcache_cache *cache = ctx->cache_tree->cache;
	struct pcache_request *pcache_req = ctx->pcache_req;
	struct pcache_backing_dev_req *backing_req;

	if (ctx->pre_alloc_req) {
		backing_req = ctx->pre_alloc_req;
		ctx->pre_alloc_req = NULL;

		return backing_req;
	}

	return cache_miss_req_alloc(cache, pcache_req, GFP_NOWAIT);
}

/*
 * In the process of walking the cache tree to locate cached data, this
 * function handles the situation where the requested data range lies
 * entirely before an existing cache node (`key_tmp`). This outcome
 * signifies that the target data is absent from the cache (cache miss).
 *
 * To fulfill this portion of the read request, the function creates a
 * backing request (`backing_req`) for the missing data range represented
 * by `key`. It then appends this request to the submission list in the
 * `ctx`, which will later be processed to retrieve the data from backing
 * storage. After setting up the backing request, `req_done` in `ctx` is
 * updated to reflect the length of the handled range, and the range
 * in `key` is adjusted by trimming off the portion that is now handled.
 *
 * The scenario handled here:
 *
 *	  |--------|			  key_tmp (existing cached range)
 * |====|					   key (requested range, preceding key_tmp)
 *
 * Since `key` is before `key_tmp`, it signifies that the requested data
 * range is missing in the cache (cache miss) and needs retrieval from
 * backing storage.
 */
static int read_before(struct pcache_cache_key *key, struct pcache_cache_key *key_tmp,
		struct pcache_cache_subtree_walk_ctx *ctx)
{
	struct pcache_backing_dev_req *backing_req;
	struct pcache_cache *cache = ctx->cache_tree->cache;

	/*
	 * In this scenario, `key` represents a range that precedes `key_tmp`,
	 * meaning the requested data range is missing from the cache tree
	 * and must be retrieved from the backing_dev.
	 */
	backing_req = get_pre_alloc_req(ctx);
	if (!backing_req)
		return SUBTREE_WALK_RET_NEED_REQ;

	cache_miss_req_init(cache, backing_req, ctx->pcache_req, ctx->req_done, key->len, true);

	list_add(&backing_req->node, ctx->submit_req_list);
	ctx->req_done += key->len;
	cache_key_cutfront(key, key->len);

	return SUBTREE_WALK_RET_OK;
}

/*
 * During cache_subtree_walk, this function manages a scenario where part of the
 * requested data range overlaps with an existing cache node (`key_tmp`).
 *
 *	 |----------------|  key_tmp (existing cached range)
 * |===========|		   key (requested range, overlapping the tail of key_tmp)
 */
static int read_overlap_tail(struct pcache_cache_key *key, struct pcache_cache_key *key_tmp,
		struct pcache_cache_subtree_walk_ctx *ctx)
{
	struct pcache_cache *cache = ctx->cache_tree->cache;
	struct pcache_backing_dev_req *backing_req;
	u32 io_len;
	int ret;

	/*
	 * Calculate the length of the non-overlapping portion of `key`
	 * before `key_tmp`, representing the data missing in the cache.
	 */
	io_len = cache_key_lstart(key_tmp) - cache_key_lstart(key);
	if (io_len) {
		backing_req = get_pre_alloc_req(ctx);
		if (!backing_req)
			return SUBTREE_WALK_RET_NEED_REQ;

		cache_miss_req_init(cache, backing_req, ctx->pcache_req, ctx->req_done, io_len, true);

		list_add(&backing_req->node, ctx->submit_req_list);
		ctx->req_done += io_len;
		cache_key_cutfront(key, io_len);
	}

	/*
	 * Handle the overlapping portion by calculating the length of
	 * the remaining data in `key` that coincides with `key_tmp`.
	 */
	io_len = cache_key_lend(key) - cache_key_lstart(key_tmp);
	if (cache_key_empty(key_tmp)) {
		backing_req = get_pre_alloc_req(ctx);
		if (!backing_req)
			return SUBTREE_WALK_RET_NEED_REQ;

		cache_miss_req_init(cache, backing_req, ctx->pcache_req, ctx->req_done, io_len, false);
		submit_cache_miss_req(cache, backing_req);
	} else {
		ret = cache_copy_to_req_bio(ctx->cache_tree->cache, ctx->pcache_req, ctx->req_done,
					io_len, &key_tmp->cache_pos, key_tmp->seg_gen);
		if (ret) {
			if (ret == -EINVAL) {
				cache_key_delete(key_tmp);
				return SUBTREE_WALK_RET_RESEARCH;
			}

			ctx->ret = ret;
			return SUBTREE_WALK_RET_ERR;
		}
	}

	ctx->req_done += io_len;
	cache_key_cutfront(key, io_len);

	return SUBTREE_WALK_RET_OK;
}

/*
 *    |----|          key_tmp (existing cached range)
 * |==========|       key (requested range)
 */
static int read_overlap_contain(struct pcache_cache_key *key, struct pcache_cache_key *key_tmp,
		struct pcache_cache_subtree_walk_ctx *ctx)
{
	struct pcache_cache *cache = ctx->cache_tree->cache;
	struct pcache_backing_dev_req *backing_req;
	u32 io_len;
	int ret;

	/*
	 * Calculate the non-overlapping part of `key` before `key_tmp`
	 * to identify the missing data length.
	 */
	io_len = cache_key_lstart(key_tmp) - cache_key_lstart(key);
	if (io_len) {
		backing_req = get_pre_alloc_req(ctx);
		if (!backing_req)
			return SUBTREE_WALK_RET_NEED_REQ;

		cache_miss_req_init(cache, backing_req, ctx->pcache_req, ctx->req_done, io_len, true);

		list_add(&backing_req->node, ctx->submit_req_list);

		ctx->req_done += io_len;
		cache_key_cutfront(key, io_len);
	}

	/*
	 * Handle the overlapping portion between `key` and `key_tmp`.
	 */
	io_len = key_tmp->len;
	if (cache_key_empty(key_tmp)) {
		backing_req = get_pre_alloc_req(ctx);
		if (!backing_req)
			return SUBTREE_WALK_RET_NEED_REQ;

		cache_miss_req_init(cache, backing_req, ctx->pcache_req, ctx->req_done, io_len, false);
		submit_cache_miss_req(cache, backing_req);
	} else {
		ret = cache_copy_to_req_bio(ctx->cache_tree->cache, ctx->pcache_req, ctx->req_done,
					io_len, &key_tmp->cache_pos, key_tmp->seg_gen);
		if (ret) {
			if (ret == -EINVAL) {
				cache_key_delete(key_tmp);
				return SUBTREE_WALK_RET_RESEARCH;
			}

			ctx->ret = ret;
			return SUBTREE_WALK_RET_ERR;
		}
	}

	ctx->req_done += io_len;
	cache_key_cutfront(key, io_len);

	return SUBTREE_WALK_RET_OK;
}

/*
 *	 |-----------|		key_tmp (existing cached range)
 *	   |====|			key (requested range, fully within key_tmp)
 *
 * If `key_tmp` contains valid cached data, this function copies the relevant
 * portion to the request's bio. Otherwise, it sends a backing request to
 * fetch the required data range.
 */
static int read_overlap_contained(struct pcache_cache_key *key, struct pcache_cache_key *key_tmp,
		struct pcache_cache_subtree_walk_ctx *ctx)
{
	struct pcache_cache *cache = ctx->cache_tree->cache;
	struct pcache_backing_dev_req *backing_req;
	struct pcache_cache_pos pos;
	int ret;

	/*
	 * Check if `key_tmp` is empty, indicating a miss. If so, initiate
	 * a backing request to fetch the required data for `key`.
	 */
	if (cache_key_empty(key_tmp)) {
		backing_req = get_pre_alloc_req(ctx);
		if (!backing_req)
			return SUBTREE_WALK_RET_NEED_REQ;

		cache_miss_req_init(cache, backing_req, ctx->pcache_req, ctx->req_done, key->len, false);
		submit_cache_miss_req(cache, backing_req);
	} else {
		cache_pos_copy(&pos, &key_tmp->cache_pos);
		cache_pos_advance(&pos, cache_key_lstart(key) - cache_key_lstart(key_tmp));

		ret = cache_copy_to_req_bio(ctx->cache_tree->cache, ctx->pcache_req, ctx->req_done,
					key->len, &pos, key_tmp->seg_gen);
		if (ret) {
			if (ret == -EINVAL) {
				cache_key_delete(key_tmp);
				return SUBTREE_WALK_RET_RESEARCH;
			}

			ctx->ret = ret;
			return SUBTREE_WALK_RET_ERR;
		}
	}

	ctx->req_done += key->len;
	cache_key_cutfront(key, key->len);

	return SUBTREE_WALK_RET_OK;
}

/*
 *	 |--------|		  key_tmp (existing cached range)
 *	   |==========|	  key (requested range, overlapping the head of key_tmp)
 */
static int read_overlap_head(struct pcache_cache_key *key, struct pcache_cache_key *key_tmp,
		struct pcache_cache_subtree_walk_ctx *ctx)
{
	struct pcache_cache *cache = ctx->cache_tree->cache;
	struct pcache_backing_dev_req *backing_req;
	struct pcache_cache_pos pos;
	u32 io_len;
	int ret;

	io_len = cache_key_lend(key_tmp) - cache_key_lstart(key);

	if (cache_key_empty(key_tmp)) {
		backing_req = get_pre_alloc_req(ctx);
		if (!backing_req)
			return SUBTREE_WALK_RET_NEED_REQ;

		cache_miss_req_init(cache, backing_req, ctx->pcache_req, ctx->req_done, io_len, false);
		submit_cache_miss_req(cache, backing_req);
	} else {
		cache_pos_copy(&pos, &key_tmp->cache_pos);
		cache_pos_advance(&pos, cache_key_lstart(key) - cache_key_lstart(key_tmp));

		ret = cache_copy_to_req_bio(ctx->cache_tree->cache, ctx->pcache_req, ctx->req_done,
					io_len, &pos, key_tmp->seg_gen);
		if (ret) {
			if (ret == -EINVAL) {
				cache_key_delete(key_tmp);
				return SUBTREE_WALK_RET_RESEARCH;
			}

			ctx->ret = ret;
			return SUBTREE_WALK_RET_ERR;
		}
	}

	ctx->req_done += io_len;
	cache_key_cutfront(key, io_len);

	return SUBTREE_WALK_RET_OK;
}

/**
 * read_walk_finally - Finalizes the cache read tree walk by submitting any
 *					 remaining backing requests
 * @ctx:	Context structure holding information about the cache,
 *		read request, and submission list
 * @ret:	the return value after this walk.
 *
 * This function is called at the end of the `cache_subtree_walk` during a
 * cache read operation. It completes the walk by checking if any data
 * requested by `key` was not found in the cache tree, and if so, it sends
 * a backing request to retrieve that data. Then, it iterates through the
 * submission list of backing requests created during the walk, removing
 * each request from the list and submitting it.
 *
 * The scenario managed here includes:
 * - Sending a backing request for the remaining length of `key` if it was
 *   not fulfilled by existing cache entries.
 * - Iterating through `ctx->submit_req_list` to submit each backing request
 *   enqueued during the walk.
 *
 * This ensures all necessary backing requests for cache misses are submitted
 * to the backing storage to retrieve any data that could not be found in
 * the cache.
 */
static int read_walk_finally(struct pcache_cache_subtree_walk_ctx *ctx, int ret)
{
	struct pcache_cache *cache = ctx->cache_tree->cache;
	struct pcache_backing_dev_req *backing_req, *next_req;
	struct pcache_cache_key *key = ctx->key;

	list_for_each_entry_safe(backing_req, next_req, ctx->submit_req_list, node) {
		list_del_init(&backing_req->node);
		submit_cache_miss_req(ctx->cache_tree->cache, backing_req);
	}

	if (ret != SUBTREE_WALK_RET_OK)
		return ret;

	if (key->len) {
		backing_req = get_pre_alloc_req(ctx);
		if (!backing_req)
			return SUBTREE_WALK_RET_NEED_REQ;

		cache_miss_req_init(cache, backing_req, ctx->pcache_req, ctx->req_done, key->len, true);
		submit_cache_miss_req(cache, backing_req);
		ctx->req_done += key->len;
	}

	return SUBTREE_WALK_RET_OK;
}

/*
 * This function is used within `cache_subtree_walk` to determine whether the
 * read operation has covered the requested data length. It compares the
 * amount of data processed (`ctx->req_done`) with the total data length
 * specified in the original request (`ctx->pcache_req->data_len`).
 *
 * If `req_done` meets or exceeds the required data length, the function
 * returns `true`, indicating the walk is complete. Otherwise, it returns `false`,
 * signaling that additional data processing is needed to fulfill the request.
 */
static bool read_walk_done(struct pcache_cache_subtree_walk_ctx *ctx)
{
	return (ctx->req_done >= ctx->pcache_req->data_len);
}

/**
 * cache_read - Process a read request by traversing the cache tree
 * @cache:	 Cache structure holding cache trees and related configurations
 * @pcache_req:   Request structure with information about the data to read
 *
 * This function attempts to fulfill a read request by traversing the cache tree(s)
 * to locate cached data for the requested range. If parts of the data are missing
 * in the cache, backing requests are generated to retrieve the required segments.
 *
 * The function operates by initializing a key for the requested data range and
 * preparing a context (`walk_ctx`) to manage the cache tree traversal. The context
 * includes pointers to functions (e.g., `read_before`, `read_overlap_tail`) that handle
 * specific conditions encountered during the traversal. The `walk_finally` and `walk_done`
 * functions manage the end stages of the traversal, while the `delete_key_list` and
 * `submit_req_list` lists track any keys to be deleted or requests to be submitted.
 *
 * The function first calculates the requested range and checks if it fits within the
 * current cache tree (based on the tree's size limits). It then locks the cache tree
 * and performs a search to locate any matching keys. If there are outdated keys,
 * these are deleted, and the search is restarted to ensure accurate data retrieval.
 *
 * If the requested range spans multiple cache trees, the function moves on to the
 * next tree once the current range has been processed. This continues until the
 * entire requested data length has been handled.
 */
static int cache_read(struct pcache_cache *cache, struct pcache_request *pcache_req)
{
	struct pcache_cache_key key_data = { .off = pcache_req->off, .len = pcache_req->data_len };
	struct pcache_cache_subtree *cache_subtree;
	struct pcache_cache_key *key_tmp = NULL, *key_next;
	struct rb_node *prev_node = NULL;
	struct pcache_cache_key *key = &key_data;
	struct pcache_cache_subtree_walk_ctx walk_ctx = { 0 };
	struct pcache_backing_dev_req *backing_req, *next_req;
	LIST_HEAD(delete_key_list);
	LIST_HEAD(submit_req_list);
	int ret;

	walk_ctx.cache_tree = &cache->req_key_tree;
	walk_ctx.req_done = 0;
	walk_ctx.pcache_req = pcache_req;
	walk_ctx.before = read_before;
	walk_ctx.overlap_tail = read_overlap_tail;
	walk_ctx.overlap_head = read_overlap_head;
	walk_ctx.overlap_contain = read_overlap_contain;
	walk_ctx.overlap_contained = read_overlap_contained;
	walk_ctx.walk_finally = read_walk_finally;
	walk_ctx.walk_done = read_walk_done;
	walk_ctx.delete_key_list = &delete_key_list;
	walk_ctx.submit_req_list = &submit_req_list;

next:
	key->off = pcache_req->off + walk_ctx.req_done;
	key->len = pcache_req->data_len - walk_ctx.req_done;
	if (key->len > PCACHE_CACHE_SUBTREE_SIZE - (key->off & PCACHE_CACHE_SUBTREE_SIZE_MASK))
		key->len = PCACHE_CACHE_SUBTREE_SIZE - (key->off & PCACHE_CACHE_SUBTREE_SIZE_MASK);

	cache_subtree = get_subtree(&cache->req_key_tree, key->off);
	spin_lock(&cache_subtree->tree_lock);
search:
	prev_node = cache_subtree_search(cache_subtree, key, NULL, NULL, &delete_key_list);
	if (!list_empty(&delete_key_list)) {
		list_for_each_entry_safe(key_tmp, key_next, &delete_key_list, list_node) {
			list_del_init(&key_tmp->list_node);
			cache_key_delete(key_tmp);
		}
		goto search;
	}

	walk_ctx.start_node = prev_node;
	walk_ctx.key = key;

	ret = cache_subtree_walk(&walk_ctx);
	if (ret == SUBTREE_WALK_RET_RESEARCH)
		goto search;
	spin_unlock(&cache_subtree->tree_lock);

	if (ret == SUBTREE_WALK_RET_ERR) {
		ret = walk_ctx.ret;
		goto out;
	}

	if (ret == SUBTREE_WALK_RET_NEED_REQ) {
		walk_ctx.pre_alloc_req = cache_miss_req_alloc(cache, pcache_req, GFP_NOIO);
		pcache_dev_debug(CACHE_TO_PCACHE(cache), "allocate pre_alloc_req with GFP_NOIO");
	}

	if (walk_ctx.req_done < pcache_req->data_len)
		goto next;
	ret = 0;
out:
	if (walk_ctx.pre_alloc_req)
		cache_miss_req_free(walk_ctx.pre_alloc_req);

	list_for_each_entry_safe(backing_req, next_req, &submit_req_list, node) {
		list_del_init(&backing_req->node);
		backing_dev_req_end(backing_req);
	}

	return ret;
}

static int cache_write(struct pcache_cache *cache, struct pcache_request *pcache_req)
{
	struct pcache_cache_subtree *cache_subtree;
	struct pcache_cache_key *key;
	u64 offset = pcache_req->off;
	u32 length = pcache_req->data_len;
	u32 io_done = 0;
	int ret;

	while (true) {
		if (io_done >= length)
			break;

		key = cache_key_alloc(&cache->req_key_tree, GFP_NOIO);
		key->off = offset + io_done;
		key->len = length - io_done;
		if (key->len > PCACHE_CACHE_SUBTREE_SIZE - (key->off & PCACHE_CACHE_SUBTREE_SIZE_MASK))
			key->len = PCACHE_CACHE_SUBTREE_SIZE - (key->off & PCACHE_CACHE_SUBTREE_SIZE_MASK);

		ret = cache_data_alloc(cache, key);
		if (ret) {
			cache_key_put(key);
			goto err;
		}

		ret = cache_copy_from_req_bio(cache, key, pcache_req, io_done);
		if (ret) {
			cache_seg_put(key->cache_pos.cache_seg);
			cache_key_put(key);
			goto err;
		}

		cache_subtree = get_subtree(&cache->req_key_tree, key->off);
		spin_lock(&cache_subtree->tree_lock);
		cache_key_insert(&cache->req_key_tree, key, true);
		ret = cache_key_append(cache, key, pcache_req->bio->bi_opf & REQ_FUA);
		if (ret) {
			cache_seg_put(key->cache_pos.cache_seg);
			cache_key_delete(key);
			goto unlock;
		}

		io_done += key->len;
		spin_unlock(&cache_subtree->tree_lock);
	}

	return 0;
unlock:
	spin_unlock(&cache_subtree->tree_lock);
err:
	return ret;
}

/**
 * pcache_cache_flush - Flush all ksets to persist any pending cache data
 * @cache: Pointer to the cache structure
 *
 * This function iterates through all ksets associated with the provided `cache`
 * and ensures that any data marked for persistence is written to media. For each
 * kset, it acquires the kset lock, then invokes `cache_kset_close`, which handles
 * the persistence logic for that kset.
 *
 * If `cache_kset_close` encounters an error, the function exits immediately with
 * the respective error code, preventing the flush operation from proceeding to
 * subsequent ksets.
 */
int pcache_cache_flush(struct pcache_cache *cache)
{
	struct pcache_cache_kset *kset;
	int ret;
	u32 i;

	for (i = 0; i < cache->n_ksets; i++) {
		kset = get_kset(cache, i);

		spin_lock(&kset->kset_lock);
		ret = cache_kset_close(cache, kset);
		spin_unlock(&kset->kset_lock);

		if (ret)
			return ret;
	}

	return 0;
}

int pcache_cache_handle_req(struct pcache_cache *cache, struct pcache_request *pcache_req)
{
	struct bio *bio = pcache_req->bio;

	if (unlikely(bio->bi_opf & REQ_PREFLUSH))
		return pcache_cache_flush(cache);

	if (bio_data_dir(bio) == READ)
		return cache_read(cache, pcache_req);

	return cache_write(cache, pcache_req);
}