From 4107cb34f969cf8cee99cbebb97cfdfd09788372 Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Tue, 10 Nov 2026 22:22:16 +0100 Subject: [PATCH 1/3] concurrency: Add ocf_hb_id_prot_lock_rd() Signed-off-by: Robert Baldyga --- src/concurrency/ocf_metadata_concurrency.c | 14 ++++++++++++++ src/concurrency/ocf_metadata_concurrency.h | 5 +++++ 2 files changed, 19 insertions(+) diff --git a/src/concurrency/ocf_metadata_concurrency.c b/src/concurrency/ocf_metadata_concurrency.c index da7d6114..7d3ee6a5 100644 --- a/src/concurrency/ocf_metadata_concurrency.c +++ b/src/concurrency/ocf_metadata_concurrency.c @@ -358,6 +358,20 @@ void ocf_hb_cline_prot_unlock_rd(struct ocf_metadata_lock *metadata_lock, hash, OCF_METADATA_RD); } +void ocf_hb_id_prot_lock_rd(struct ocf_metadata_lock *metadata_lock, + unsigned lock_idx, ocf_cache_line_t hash) +{ + ocf_hb_id_prot_lock_common(metadata_lock, lock_idx, hash, + OCF_METADATA_RD); +} + +void ocf_hb_id_prot_unlock_rd(struct ocf_metadata_lock *metadata_lock, + unsigned lock_idx, ocf_cache_line_t hash) +{ + ocf_hb_id_prot_unlock_common(metadata_lock, lock_idx, hash, + OCF_METADATA_RD); +} + void ocf_hb_id_prot_lock_wr(struct ocf_metadata_lock *metadata_lock, unsigned lock_idx, ocf_cache_line_t hash) { diff --git a/src/concurrency/ocf_metadata_concurrency.h b/src/concurrency/ocf_metadata_concurrency.h index 71cffcbf..ec114894 100644 --- a/src/concurrency/ocf_metadata_concurrency.h +++ b/src/concurrency/ocf_metadata_concurrency.h @@ -116,6 +116,11 @@ void ocf_hb_cline_prot_lock_wr(struct ocf_metadata_lock *metadata_lock, void ocf_hb_cline_prot_unlock_wr(struct ocf_metadata_lock *metadata_lock, uint32_t lock_idx, uint32_t core_id, uint64_t core_line); +void ocf_hb_id_prot_lock_rd(struct ocf_metadata_lock *metadata_lock, + unsigned lock_idx, ocf_cache_line_t hash); +void ocf_hb_id_prot_unlock_rd(struct ocf_metadata_lock *metadata_lock, + unsigned lock_idx, ocf_cache_line_t hash); + void ocf_hb_id_prot_lock_wr(struct ocf_metadata_lock *metadata_lock, unsigned lock_idx, ocf_cache_line_t hash); void ocf_hb_id_prot_unlock_wr(struct ocf_metadata_lock *metadata_lock, From 983e6035558fb0904325d4a4cff7eb5d7a6e2554 Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Tue, 10 Nov 2026 17:26:18 +0100 Subject: [PATCH 2/3] flush: Replace sort with sweep through the hash table The flush collected all dirty cache lines of a core into one array up front and sorted it whole, so its memory grew with the amount of dirty data - about 16 bytes per dirty line, ~12 GB for a dirty 3 TB cache with 4 KiB lines - and the flush failed with -OCF_ERR_NO_MEM when that could not be allocated. Sweep the hash table instead: collect the dirty lines of the flushed core (or of all cores) in the collision lists of consecutive hash buckets, up to a portion, sort the portion by core address and clean it, then go on from the next bucket. The portion size is adapted as before, between 4 MiB and 100 MiB of cache lines, which also bounds the buffer. The hash function assigns consecutive core lines consecutive buckets, so a portion containing lines from consecutive buckets is likely to contain several consecutive core line ranges, interleaved by a collision list length (typically a few lines at most), thus the portion needs to be sorted before being passed to the cleaner. To keep more than one portion in flight, the hash table is split into slices swept concurrently, one per core with dirty data, at most 16. A line is in the list of exactly one bucket, so slices never collect the same line. Signed-off-by: Robert Baldyga --- src/mngt/ocf_mngt_flush.c | 645 +++++++++++++++++--------------------- src/utils/utils_cleaner.c | 11 - src/utils/utils_cleaner.h | 34 +- 3 files changed, 281 insertions(+), 409 deletions(-) diff --git a/src/mngt/ocf_mngt_flush.c b/src/mngt/ocf_mngt_flush.c index b0078bd6..e81a52d6 100644 --- a/src/mngt/ocf_mngt_flush.c +++ b/src/mngt/ocf_mngt_flush.c @@ -26,25 +26,42 @@ struct ocf_mngt_cache_flush_context; typedef void (*ocf_flush_complete_t)(struct ocf_mngt_cache_flush_context *, int); -struct flush_containers_context -{ - /* array of container descriptors */ - struct flush_container *fctbl; - /* fctbl array size */ - uint32_t fcnum; - /* shared error for all concurrent container flushes */ +struct flush_sweep { + struct ocf_mngt_cache_flush_context *context; + ocf_cache_t cache; + /* next hash bucket to examine and the end of the slice */ + ocf_cache_line_t bucket; + ocf_cache_line_t end; + /* portion being cleaned */ + struct flush_data *data; + uint32_t count; + struct ocf_cleaner_attribs attribs; + struct ocf_request *req; + /* portion size, adapted to the time the last portion took */ + uint64_t flush_portion; + uint64_t ticks1; + uint64_t ticks2; +}; + +struct flush_sweeps_context { + /* range of cache lines to flush */ + ocf_cache_line_t begin; + ocf_cache_line_t end; + /* slices of the hash table, swept concurrently */ + struct flush_sweep *sweeps; + uint32_t num; + /* shared error for all concurrent sweeps */ env_atomic error; - /* number of outstanding container flushes */ + /* number of sweeps in progress, plus one while starting them */ env_atomic count; - /* first container flush to notice interrupt sets this to 1 */ + /* first sweep to notice interrupt sets this to 1 */ env_atomic interrupt_seen; - /* completion to be called after all containers are flushed */ + /* completion to be called after all sweeps are done */ ocf_flush_complete_t complete; }; /* common struct for cache/core flush/purge pipeline priv */ -struct ocf_mngt_cache_flush_context -{ +struct ocf_mngt_cache_flush_context { /* pipeline for flush / purge */ ocf_pipeline_t pipeline; /* target cache */ @@ -87,8 +104,8 @@ struct ocf_mngt_cache_flush_context uint64_t core_id; } purge; - /* context for flush containers logic */ - struct flush_containers_context fcs; + /* context for the flush sweeps */ + struct flush_sweeps_context fsc; }; static void _ocf_mngt_begin_flush(ocf_pipeline_t pipeline, void *priv, @@ -168,234 +185,140 @@ void ocf_mngt_cache_set_no_dirty(ocf_cache_t cache, bool no_dirty) } /************************FLUSH CORE CODE**************************************/ -/* Returns: - * 0 if OK and tbl & num is filled: - * * tbl - table with blocks&cacheline - * * num - number of items in this table. - * other value means error. - * NOTE: - * Table is not sorted. - */ -static int _ocf_mngt_get_blocks(ocf_cache_t cache, ocf_core_id_t core_id, - struct flush_data **tbl, uint32_t *num) -{ - ocf_core_t core = ocf_cache_get_core(cache, core_id); - uint64_t core_line; - ocf_core_id_t i_core_id; - struct flush_data *elem; - uint32_t line, dirty_found = 0, dirty_total = 0; - unsigned ret = 0; - - ocf_metadata_start_exclusive_access(&cache->metadata_lock); - - dirty_total = env_atomic_read(&core->runtime_meta->dirty_clines); - if (!dirty_total) { - *num = 0; - *tbl = NULL; - goto unlock; - } - - *tbl = env_vmalloc(dirty_total * sizeof(**tbl)); - if (*tbl == NULL) { - ret = -OCF_ERR_NO_MEM; - goto unlock; - } - - for (line = 0, elem = *tbl; - line < ocf_metadata_line_count(&cache->metadata); - line++) { - ocf_metadata_get_core_info(cache, line, &i_core_id, - &core_line); - - if (i_core_id == core_id && - metadata_test_valid_any(cache, line) && - metadata_test_dirty(cache, line)) { - /* It's valid and dirty target core cacheline */ - elem->cache_line = line; - elem->core_line = core_line; - elem->core_id = i_core_id; - elem++; - dirty_found++; - - /* stop if all cachelines were found */ - if (dirty_found == dirty_total) - break; - } - - if ((line + 1) % 131072 == 0) { - ocf_metadata_end_exclusive_access( - &cache->metadata_lock); - env_cond_resched(); - ocf_metadata_start_exclusive_access( - &cache->metadata_lock); - } - } - ocf_core_log(core, log_debug, - "%u dirty cache lines to clean\n", dirty_found); +#define OCF_MNG_FLUSH_MIN (4*MiB / ocf_line_size(cache)) +#define OCF_MNG_FLUSH_MAX (100*MiB / ocf_line_size(cache)) +#define OCF_MNG_FLUSH_SLICES 16 - *num = dirty_found; +/* Hash buckets examined between reschedules */ +#define OCF_MNG_FLUSH_RESCHED_BUCKETS 131072 -unlock: - ocf_metadata_end_exclusive_access(&cache->metadata_lock); +static bool _ocf_mngt_flush_target(struct ocf_mngt_cache_flush_context *context, + ocf_core_id_t core_id) +{ + if (context->core) + return core_id == ocf_core_get_id(context->core); - return ret; + return core_id < OCF_CORE_NUM; } -static int _ocf_mngt_get_flush_containers(ocf_cache_t cache, - struct flush_container **fctbl, uint32_t *fcnum, - ocf_cache_line_t begin, ocf_cache_line_t end) +static uint32_t _ocf_mngt_flush_target_dirty( + struct ocf_mngt_cache_flush_context *context) { - struct flush_container *fc; - struct flush_container *curr; - uint32_t *core_revmap; - uint32_t core_count; - uint64_t core_line; - ocf_core_id_t core_id; + ocf_cache_t cache = context->cache; ocf_core_t core; - uint32_t i, j = 0, line; - uint32_t dirty_found = 0, dirty_total = 0, dirty_cores = 0; - int ret = 0; - - end = OCF_MIN(ocf_metadata_line_count(&cache->metadata), end); - - if (begin > end) - return -OCF_ERR_INVAL; - - if (begin == end) { - ocf_cache_log(cache, log_warn, "Flush with 0 length\n"); - return 0; - } - - ocf_metadata_start_exclusive_access(&cache->metadata_lock); - - core_count = cache->conf_meta->core_count; - if (core_count == 0) - goto unlock; - - core_revmap = env_vzalloc(sizeof(*core_revmap) * OCF_CORE_NUM); - if (!core_revmap) { - ret = -OCF_ERR_NO_MEM; - goto unlock; - } + ocf_core_id_t core_id; + uint32_t dirty = 0; - /* TODO: Alloc fcs and data tables in single allocation */ - fc = env_vzalloc(sizeof(**fctbl) * core_count); - if (!fc) { - ret = -OCF_ERR_NO_MEM; - goto free_core_revmap; + if (context->core) { + return env_atomic_read( + &context->core->runtime_meta->dirty_clines); } - for_each_core(cache, core, core_id) { - fc[j].core_id = core_id; - core_revmap[core_id] = j; + for_each_core(cache, core, core_id) + dirty += env_atomic_read(&core->runtime_meta->dirty_clines); - /* Check for dirty blocks */ - fc[j].count = env_atomic_read( - &core->runtime_meta->dirty_clines); - dirty_total += fc[j].count; - - if (fc[j].count == 0) - continue; + return dirty; +} - fc[j].flush_data = env_vmalloc(fc[j].count * - sizeof(*fc[j].flush_data)); - if (!fc[j].flush_data) { - ret = -OCF_ERR_NO_MEM; - goto free_flush_data; +static void _ocf_mngt_flush_collect(struct flush_sweep *sweep, uint32_t max) +{ + struct ocf_mngt_cache_flush_context *context = sweep->context; + struct flush_sweeps_context *fsc = &context->fsc; + ocf_cache_t cache = sweep->cache; + ocf_cache_line_t terminator = + ocf_metadata_terminator_line(&cache->metadata); + ocf_cache_line_t line, next; + ocf_core_id_t core_id; + uint64_t core_line; + uint32_t scanned = 0, bucket_start; + unsigned lock_idx; + bool bucket_done; + + sweep->count = 0; + + while (sweep->bucket < sweep->end && sweep->count < max) { + bucket_start = sweep->count; + + lock_idx = ocf_metadata_concurrency_next_idx(cache->mngt_queue); + ocf_hb_id_prot_lock_rd(&cache->metadata_lock, lock_idx, + sweep->bucket); + + bucket_done = true; + next = ocf_metadata_get_hash(cache, sweep->bucket); + while (next != terminator) { + line = next; + next = ocf_metadata_get_collision_next(cache, line); + + ocf_metadata_get_core_info(cache, line, &core_id, + &core_line); + + if (line < fsc->begin || line >= fsc->end) + continue; + if (!_ocf_mngt_flush_target(context, core_id)) + continue; + if (!metadata_test_valid_any(cache, line)) + continue; + if (!metadata_test_dirty(cache, line)) + continue; + if (sweep->count == max) { + bucket_done = false; + break; + } + sweep->data[sweep->count].cache_line = line; + sweep->data[sweep->count].core_line = core_line; + sweep->data[sweep->count].core_id = core_id; + sweep->count++; } - if (++j == core_count) + ocf_hb_id_prot_unlock_rd(&cache->metadata_lock, lock_idx, + sweep->bucket); + + if (!bucket_done) { + if (bucket_start > 0) + sweep->count = bucket_start; break; - } + } - /* Set number of dirty cores found to flush. */ - dirty_cores = j; + sweep->bucket++; - if (!dirty_total) { - *fcnum = 0; - goto free_fc; + if (++scanned % OCF_MNG_FLUSH_RESCHED_BUCKETS == 0) + env_cond_resched(); } +} - ocf_cache_log(cache, log_debug, - "Range of cache lines to flush %u - %u\n", - begin, end); - - for (line = begin; line < end; line++) { - ocf_metadata_get_core_info(cache, line, &core_id, &core_line); - - if (metadata_test_valid_any(cache, line) && - metadata_test_dirty(cache, line)) { - curr = &fc[core_revmap[core_id]]; - - ENV_BUG_ON(curr->iter >= curr->count); - - curr->flush_data[curr->iter].cache_line = line; - curr->flush_data[curr->iter].core_line = core_line; - curr->flush_data[curr->iter].core_id = core_id; - curr->iter++; - dirty_found++; - - if (dirty_found == dirty_total) - break; - } +static void _ocf_mngt_flush_account(struct flush_sweep *sweep) +{ + ocf_cache_t cache = sweep->cache; + uint32_t i, run = 0; - if ((line + 1) % 131072 == 0) { - ocf_metadata_end_exclusive_access( - &cache->metadata_lock); - env_cond_resched(); - ocf_metadata_start_exclusive_access( - &cache->metadata_lock); + for (i = 0; i < sweep->count; i++) { + run++; + if (i + 1 < sweep->count && sweep->data[i + 1].core_id == + sweep->data[i].core_id) { + continue; } - } - if (dirty_total != dirty_found) { - for (i = 0; i < dirty_cores; i++) - fc[i].count = fc[i].iter; + env_atomic_add(run, &ocf_cache_get_core(cache, + sweep->data[i].core_id)->flushed); + run = 0; } - - for (i = 0; i < dirty_cores; i++) - fc[i].iter = 0; - - *fctbl = fc; - *fcnum = dirty_cores; - goto free_core_revmap; - -free_flush_data: - for (i = 0; i < j; i++) - env_vfree(fc[i].flush_data); -free_fc: - env_vfree(fc); -free_core_revmap: - env_vfree(core_revmap); -unlock: - ocf_metadata_end_exclusive_access(&cache->metadata_lock); - return ret; } -static void _ocf_mngt_free_flush_containers(struct flush_container *fctbl, - uint32_t num) -{ - int i; - - for (i = 0; i < num; i++) - env_vfree(fctbl[i].flush_data); - env_vfree(fctbl); -} +static void _ocf_mngt_flush_sweep_complete(struct flush_sweep *sweep); static void _ocf_mngt_flush_portion_end(void *private_data, int error) { - struct flush_container *fc = private_data; - struct ocf_mngt_cache_flush_context *context = fc->context; - struct flush_containers_context *fsc = &context->fcs; + struct flush_sweep *sweep = private_data; + struct ocf_mngt_cache_flush_context *context = sweep->context; + struct flush_sweeps_context *fsc = &context->fsc; ocf_cache_t cache = context->cache; - ocf_core_t core = ocf_cache_get_core(cache, fc->core_id); bool first_interrupt; - env_atomic_set(&core->flushed, fc->iter); + _ocf_mngt_flush_account(sweep); - fc->ticks2 = env_get_tick_count(); + sweep->ticks2 = env_get_tick_count(); env_atomic_cmpxchg(&fsc->error, 0, error); @@ -410,205 +333,213 @@ static void _ocf_mngt_flush_portion_end(void *private_data, int error) } } - if (fc->iter == fc->count || env_atomic_read(&fsc->error)) { - ocf_req_put(fc->req); - fc->end(context); + if (env_atomic_read(&fsc->error)) { + _ocf_mngt_flush_sweep_complete(sweep); return; } - ocf_queue_push_req(fc->req, 0); + ocf_queue_push_req(sweep->req, 0); } -/* - * OCF will try to guess disk speed etc. and adjust flushing block - * size accordingly, however these bounds shall be respected regardless - * of disk speed, cache line size configured etc. - */ -#define OCF_MNG_FLUSH_MIN (4*MiB / ocf_line_size(cache)) -#define OCF_MNG_FLUSH_MAX (100*MiB / ocf_line_size(cache)) - -static void _ocf_mngt_flush_portion(struct flush_container *fc) +static int _ocf_mngt_flush_sweep_step(struct ocf_request *req) { - ocf_cache_t cache = fc->cache; - ocf_core_t core = ocf_cache_get_core(cache, fc->core_id); + struct flush_sweep *sweep = req->priv; + ocf_cache_t cache = sweep->cache; uint64_t flush_portion_div; - uint32_t flush_data_offset = fc->iter; - uint32_t curr_flush_portion_size; - uint32_t core_dirty_count = - env_atomic_read(&core->runtime_meta->dirty_clines); - - if (core_dirty_count == 0) { - fc->iter = fc->count; - _ocf_mngt_flush_portion_end(fc, 0); - return; - } - flush_portion_div = env_ticks_to_msecs(fc->ticks2 - fc->ticks1); + /* Nothing left to flush - skip the rest of the slice */ + if (_ocf_mngt_flush_target_dirty(sweep->context) == 0) + sweep->bucket = sweep->end; + + flush_portion_div = env_ticks_to_msecs(sweep->ticks2 - sweep->ticks1); if (unlikely(!flush_portion_div)) flush_portion_div = 1; - fc->flush_portion = fc->flush_portion * 1000 / flush_portion_div; - fc->flush_portion &= ~0x3ffULL; + sweep->flush_portion = sweep->flush_portion * 1000 / flush_portion_div; + sweep->flush_portion &= ~0x3ffULL; - /* regardless those calculations, limit flush portion to be + /* + * regardless those calculations, limit flush portion to be * between OCF_MNG_FLUSH_MIN and OCF_MNG_FLUSH_MAX */ - fc->flush_portion = OCF_MIN(fc->flush_portion, OCF_MNG_FLUSH_MAX); - fc->flush_portion = OCF_MAX(fc->flush_portion, OCF_MNG_FLUSH_MIN); - - curr_flush_portion_size = OCF_MIN(core_dirty_count, fc->flush_portion); - curr_flush_portion_size = OCF_MIN(curr_flush_portion_size, - fc->count - fc->iter); - - ENV_BUG_ON(curr_flush_portion_size == 0); + sweep->flush_portion = OCF_MIN(sweep->flush_portion, OCF_MNG_FLUSH_MAX); + sweep->flush_portion = OCF_MAX(sweep->flush_portion, OCF_MNG_FLUSH_MIN); - fc->iter += curr_flush_portion_size; + _ocf_mngt_flush_collect(sweep, sweep->flush_portion); - fc->ticks1 = env_get_tick_count(); + if (sweep->count == 0) { + _ocf_mngt_flush_sweep_complete(sweep); + return 0; + } - ocf_cleaner_do_flush_data_async(fc->cache, - &fc->flush_data[flush_data_offset], - curr_flush_portion_size, &fc->attribs); -} + /* The consecutive ranges are interleaved, we need to sort first */ + ocf_cleaner_sort_flush_data(sweep->data, sweep->count); -static int _ofc_flush_container_step(struct ocf_request *req) -{ - struct flush_container *fc = req->priv; - ocf_cache_t cache = fc->cache; + sweep->ticks1 = env_get_tick_count(); ocf_metadata_start_exclusive_access(&cache->metadata_lock); - _ocf_mngt_flush_portion(fc); + ocf_cleaner_do_flush_data_async(cache, sweep->data, sweep->count, + &sweep->attribs); ocf_metadata_end_exclusive_access(&cache->metadata_lock); return 0; } -static void _ocf_mngt_flush_container( - struct ocf_mngt_cache_flush_context *context, - struct flush_container *fc, ocf_flush_containter_coplete_t end) +static void _ocf_mngt_flush_sweeps_free(struct flush_sweeps_context *fsc) { - ocf_cache_t cache = context->cache; - struct ocf_request *req; - int error = 0; + uint32_t i; - if (!fc->count) - goto finish; - - fc->end = end; - fc->context = context; - - req = ocf_req_new_mngt(cache, cache->mngt_queue); - if (!req) { - error = OCF_ERR_NO_MEM; - goto finish; + for (i = 0; i < fsc->num; i++) { + if (fsc->sweeps[i].req) + ocf_req_put(fsc->sweeps[i].req); + env_vfree(fsc->sweeps[i].data); } - - req->info.internal = true; - req->engine_handler = _ofc_flush_container_step; - req->priv = fc; - - fc->req = req; - fc->attribs.lock_cacheline = true; - fc->attribs.cmpl_context = fc; - fc->attribs.cmpl_fn = _ocf_mngt_flush_portion_end; - fc->attribs.io_queue = cache->mngt_queue; - fc->cache = cache; - fc->flush_portion = OCF_MNG_FLUSH_MIN; - fc->ticks1 = 0; - fc->ticks2 = UINT_MAX; - - ocf_queue_push_req(fc->req, OCF_QUEUE_ALLOW_SYNC); - return; - -finish: - env_atomic_cmpxchg(&context->fcs.error, 0, error); - end(context); + env_vfree(fsc->sweeps); + fsc->sweeps = NULL; + fsc->num = 0; } -static void _ocf_flush_container_complete(void *ctx) +static void _ocf_mngt_flush_sweeps_put( + struct ocf_mngt_cache_flush_context *context) { - struct ocf_mngt_cache_flush_context *context = ctx; + struct flush_sweeps_context *fsc = &context->fsc; - if (env_atomic_dec_return(&context->fcs.count)) { + if (env_atomic_dec_return(&fsc->count)) return; - } - _ocf_mngt_free_flush_containers(context->fcs.fctbl, - context->fcs.fcnum); + _ocf_mngt_flush_sweeps_free(fsc); - context->fcs.complete(context, - env_atomic_read(&context->fcs.error)); + fsc->complete(context, env_atomic_read(&fsc->error)); } -static void _ocf_mngt_flush_containers( - struct ocf_mngt_cache_flush_context *context, - struct flush_container *fctbl, - uint32_t fcnum, ocf_flush_complete_t complete) +static void _ocf_mngt_flush_sweep_complete(struct flush_sweep *sweep) { - int i; - - if (fcnum == 0) { - complete(context, 0); - return; - } + _ocf_mngt_flush_sweeps_put(sweep->context); +} - /* Sort data. Smallest blocks first (0...n). */ - ocf_cleaner_sort_flush_containers(fctbl, fcnum); +static uint32_t _ocf_mngt_flush_dirty_cores( + struct ocf_mngt_cache_flush_context *context) +{ + ocf_cache_t cache = context->cache; + ocf_core_t core; + ocf_core_id_t core_id; + uint32_t num = 0; - env_atomic_set(&context->fcs.error, 0); - env_atomic_set(&context->fcs.count, 1); - context->fcs.complete = complete; - context->fcs.fctbl = fctbl; - context->fcs.fcnum = fcnum; + if (context->core) + return 1; - for (i = 0; i < fcnum; i++) { - if (fctbl[i].count == 0) - continue; - env_atomic_inc(&context->fcs.count); - _ocf_mngt_flush_container(context, &fctbl[i], - _ocf_flush_container_complete); + for_each_core(cache, core, core_id) { + if (env_atomic_read(&core->runtime_meta->dirty_clines)) + num++; } - _ocf_flush_container_complete(context); + return num; } - -static void _ocf_mngt_flush_core( - struct ocf_mngt_cache_flush_context *context, - ocf_flush_complete_t complete) +static void _ocf_mngt_flush_sweeps( + struct ocf_mngt_cache_flush_context *context, + ocf_cache_line_t begin, ocf_cache_line_t end, + ocf_flush_complete_t complete) { + struct flush_sweeps_context *fsc = &context->fsc; ocf_cache_t cache = context->cache; - ocf_core_t core = context->core; - ocf_core_id_t core_id = ocf_core_get_id(core); - struct flush_container *fc; - int ret; + struct flush_sweep *sweep; + uint32_t i, num; + uint64_t buckets = ocf_metadata_hash_entries(&cache->metadata); + + end = OCF_MIN(ocf_metadata_line_count(&cache->metadata), end); - fc = env_vzalloc(sizeof(*fc)); - if (!fc) { - complete(context, -OCF_ERR_NO_MEM); + if (begin > end) { + complete(context, -OCF_ERR_INVAL); return; } - ret = _ocf_mngt_get_blocks(cache, core_id, - &fc->flush_data, &fc->count); - if (ret) { - ocf_core_log(core, log_err, "Flushing operation aborted, " - "no memory\n"); - env_vfree(fc); - complete(context, -OCF_ERR_NO_MEM); + if (begin == end) { + ocf_cache_log(cache, log_warn, "Flush with 0 length\n"); + complete(context, 0); return; } - if (fc->count == 0) { - env_vfree(fc); + if (_ocf_mngt_flush_target_dirty(context) == 0) { complete(context, 0); return; } - fc->core_id = core_id; - fc->iter = 0; + fsc->begin = begin; + fsc->end = end; + + num = _ocf_mngt_flush_dirty_cores(context); + num = OCF_MIN(num, OCF_MNG_FLUSH_SLICES); + num = OCF_MIN(num, buckets); + num = OCF_MAX(num, 1); + + fsc->sweeps = env_vzalloc(sizeof(*fsc->sweeps) * num); + if (!fsc->sweeps) + goto err_nomem; + fsc->num = num; + + for (i = 0; i < num; i++) { + sweep = &fsc->sweeps[i]; + + sweep->data = env_vmalloc(sizeof(*sweep->data) * + OCF_MNG_FLUSH_MAX); + if (!sweep->data) + goto err_free; + + sweep->req = ocf_req_new_mngt(cache, cache->mngt_queue); + if (!sweep->req) + goto err_free; + + sweep->req->info.internal = true; + sweep->req->engine_handler = _ocf_mngt_flush_sweep_step; + sweep->req->priv = sweep; + + sweep->context = context; + sweep->cache = cache; + sweep->bucket = buckets * i / num; + sweep->end = buckets * (i + 1) / num; + + sweep->attribs.lock_cacheline = true; + sweep->attribs.cmpl_context = sweep; + sweep->attribs.cmpl_fn = _ocf_mngt_flush_portion_end; + sweep->attribs.io_queue = cache->mngt_queue; + + sweep->flush_portion = OCF_MNG_FLUSH_MIN; + sweep->ticks1 = 0; + sweep->ticks2 = UINT_MAX; + } + + env_atomic_set(&fsc->error, 0); + env_atomic_set(&fsc->interrupt_seen, 0); + env_atomic_set(&fsc->count, num + 1); + fsc->complete = complete; + + for (i = 0; i < num; i++) + ocf_queue_push_req(fsc->sweeps[i].req, OCF_QUEUE_ALLOW_SYNC); + + _ocf_mngt_flush_sweeps_put(context); + return; + +err_free: + _ocf_mngt_flush_sweeps_free(fsc); +err_nomem: + if (context->core) { + ocf_core_log(context->core, log_err, + "Flushing operation aborted, no memory\n"); + } else { + ocf_cache_log(cache, log_err, + "Flushing operation aborted, no memory\n"); + } + complete(context, -OCF_ERR_NO_MEM); +} - _ocf_mngt_flush_containers(context, fc, 1, complete); +static void _ocf_mngt_flush_core( + struct ocf_mngt_cache_flush_context *context, + ocf_flush_complete_t complete) +{ + _ocf_mngt_flush_sweeps(context, 0, + ocf_metadata_line_count(&context->cache->metadata), + complete); } static void _ocf_mngt_flush_all_cores( @@ -616,9 +547,6 @@ static void _ocf_mngt_flush_all_cores( ocf_flush_complete_t complete) { ocf_cache_t cache = context->cache; - struct flush_container *fctbl = NULL; - uint32_t fcnum = 0; - int ret; if (context->op == flush_cache) ocf_cache_log(cache, log_info, "Flushing cache\n"); @@ -627,17 +555,7 @@ static void _ocf_mngt_flush_all_cores( env_atomic_set(&cache->flush_in_progress, 1); - /* Get all 'dirty' blocks for all cores */ - ret = _ocf_mngt_get_flush_containers(cache, &fctbl, &fcnum, - context->begin, context->end); - if (ret) { - ocf_cache_log(cache, log_err, "Flushing operation aborted, " - "no memory\n"); - complete(context, ret); - return; - } - - _ocf_mngt_flush_containers(context, fctbl, fcnum, complete); + _ocf_mngt_flush_sweeps(context, context->begin, context->end, complete); } static void _ocf_mngt_flush_all_cores_complete( @@ -667,9 +585,6 @@ static void _ocf_mngt_flush_all_cores_complete( ocf_pipeline_next(context->pipeline); } -/** - * Flush all the dirty data stored on cache (all the cores attached to it) - */ static void _ocf_mngt_cache_flush(ocf_pipeline_t pipeline, void *priv, ocf_pipeline_arg_t arg) { diff --git a/src/utils/utils_cleaner.c b/src/utils/utils_cleaner.c index c2f79870..7d5bce1f 100644 --- a/src/utils/utils_cleaner.c +++ b/src/utils/utils_cleaner.c @@ -838,17 +838,6 @@ void ocf_cleaner_sort_flush_data(struct flush_data *flush_data, uint32_t count) } } -void ocf_cleaner_sort_flush_containers(struct flush_container *fctbl, - uint32_t num) -{ - int i; - - for (i = 0; i < num; i++) { - ocf_cleaner_sort_flush_data(fctbl[i].flush_data, - fctbl[i].count); - } -} - void ocf_cleaner_refcnt_freeze(ocf_cache_t cache) { struct ocf_user_part *curr_part; diff --git a/src/utils/utils_cleaner.h b/src/utils/utils_cleaner.h index 9f87cfe1..488a14dd 100644 --- a/src/utils/utils_cleaner.h +++ b/src/utils/utils_cleaner.h @@ -1,6 +1,7 @@ /* * Copyright(c) 2012-2021 Intel Corporation * Copyright(c) 2024 Huawei Technologies + * Copyright(c) 2026 Unvertical * SPDX-License-Identifier: BSD-3-Clause */ @@ -54,30 +55,6 @@ struct flush_data { ocf_core_id_t core_id; }; -typedef void (*ocf_flush_containter_coplete_t)(void *ctx); - -/** - * @brief Flush table container - */ -struct flush_container { - ocf_core_id_t core_id; - struct flush_data *flush_data; - uint32_t count; - uint32_t iter; - - struct ocf_cleaner_attribs attribs; - ocf_cache_t cache; - - struct ocf_request *req; - - uint64_t flush_portion; - uint64_t ticks1; - uint64_t ticks2; - - ocf_flush_containter_coplete_t end; - struct ocf_mngt_cache_flush_context *context; -}; - typedef void (*ocf_cleaner_refcnt_zero_cb_t)(void *priv); /** @@ -121,15 +98,6 @@ int ocf_cleaner_do_flush_data_async(struct ocf_cache *cache, */ void ocf_cleaner_sort_flush_data(struct flush_data *flush_data, uint32_t count); -/** - * @brief Sort flush data in all flush containters - * - * @param fctbl Flush containers to sort - * @param num Number of entries in fctbl - */ -void ocf_cleaner_sort_flush_containers(struct flush_container *fctbl, - uint32_t num); - /** * @brief Disable incrementing of cleaner reference counters * From 3370c2db5d7bf36af30fb8f416e9e41c4cc57cb9 Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Tue, 10 Nov 2026 22:22:33 +0100 Subject: [PATCH 3/3] pyocf: Add more tests for flush Signed-off-by: Robert Baldyga --- .../functional/tests/management/test_flush.py | 155 +++++++++++++++++- 1 file changed, 154 insertions(+), 1 deletion(-) diff --git a/tests/functional/tests/management/test_flush.py b/tests/functional/tests/management/test_flush.py index 88b933ce..18b33084 100644 --- a/tests/functional/tests/management/test_flush.py +++ b/tests/functional/tests/management/test_flush.py @@ -1,15 +1,21 @@ # # Copyright(c) 2024 Huawei Technologies Co., Ltd. +# Copyright(c) 2026 Unvertical # SPDX-License-Identifier: BSD-3-Clause # +import random + import pytest from pyocf.utils import Size as S from pyocf.types.cache import Cache, CacheMode, CleaningPolicy from pyocf.types.core import Core -from pyocf.types.volume import RamVolume, Volume +from pyocf.types.volume import RamVolume, TraceDevice, Volume from pyocf.types.volume_core import CoreVolume +from pyocf.types.data import Data +from pyocf.types.io import IoDir +from pyocf.types.shared import CacheLineSize from pyocf.rio import Rio, ReadWrite @@ -90,3 +96,150 @@ def test_flush_core(pyocf_ctx): core2_stats = core2.get_stats() assert S(core2_stats["usage"]["dirty"]["value"] * 4096) == S.from_MiB(0) + + +def write_random(vol, queue, offset, size): + content = random.randbytes(size) + data = Data(size) + data.write(content, size) + vol.sync_io(queue, offset, data, IoDir.WRITE) + return content + + +def dirty_bytes(stats): + return S(stats["usage"]["dirty"]["value"] * 4096) + + +@pytest.mark.parametrize("cls", CacheLineSize) +def test_flush_more_than_one_portion(pyocf_ctx, cls): + """ + A flush cleans the dirty data in portions of at most 100 MiB, collected while + sweeping the cache lines. With more dirty data than that, every portion has to be + flushed for the core to end up with the data held in the cache. + """ + cache = Cache.start_on_device( + RamVolume(S.from_MiB(400)), + cache_mode=CacheMode.WB, + cache_line_size=cls, + metadata_volatile=True, + ) + core_device = RamVolume(S.from_MiB(300)) + core = Core.using_device(core_device) + cache.add_core(core) + cache.set_cleaning_policy(CleaningPolicy.NOP) + vol = CoreVolume(core) + vol.open() + queue = cache.get_default_queue() + + for mib in range(250): + write_random(vol, queue, mib * S.from_MiB(1).B, S.from_MiB(1).B) + assert dirty_bytes(cache.get_stats()) == S.from_MiB(250) + + cache.flush() + + assert dirty_bytes(cache.get_stats()) == S.from_MiB(0) + assert cache.get_stats()["usage"]["occupancy"]["value"] > 0, "flush does not invalidate" + # md5() of the exported object closes it - last + assert core_device.md5() == vol.md5() + + +def test_flush_interleaved_cores(pyocf_ctx): + """ + With the dirty lines of several cores interleaved in the cache, a core flush cleans + only the lines of its core, and a cache flush - swept in slices concurrently, one + per core with dirty data - cleans all of them. + """ + cache = Cache.start_on_device( + RamVolume(S.from_MiB(300)), cache_mode=CacheMode.WB, metadata_volatile=True + ) + core_devices = [RamVolume(S.from_MiB(100)) for _ in range(4)] + cores = [Core.using_device(dev, name=f"core{i}") for i, dev in enumerate(core_devices)] + for core in cores: + cache.add_core(core) + cache.set_cleaning_policy(CleaningPolicy.NOP) + vols = [CoreVolume(core) for core in cores] + for vol in vols: + vol.open() + queue = cache.get_default_queue() + + for mib in range(40): + for vol in vols: + write_random(vol, queue, mib * S.from_MiB(1).B, S.from_MiB(1).B) + assert dirty_bytes(cache.get_stats()) == S.from_MiB(160) + + cores[2].flush() + + for i, core in enumerate(cores): + expected = S.from_MiB(0) if i == 2 else S.from_MiB(40) + assert dirty_bytes(core.get_stats()) == expected + assert dirty_bytes(cache.get_stats()) == S.from_MiB(120) + + cache.flush() + + assert dirty_bytes(cache.get_stats()) == S.from_MiB(0) + for core in cores: + assert dirty_bytes(core.get_stats()) == S.from_MiB(0) + # md5() of the exported object closes it - last + for dev, vol in zip(core_devices, vols): + assert dev.md5() == vol.md5() + + +@pytest.mark.parametrize("cls", CacheLineSize) +def test_flush_contiguous_dirty_data_in_runs(pyocf_ctx, cls): + """ + Contiguous dirty data is written to the core in runs of consecutive addresses, even if + its cache lines are in a different order: the flush sweeps the hash table, where + consecutive core lines are in consecutive buckets. A portion of buckets holds one range + of core addresses per wrap of the core around the hash table (a quarter of the cache + lines), so the writes are ascending runs, not one ascending sequence. Sparse dirty + data elsewhere is flushed too. + """ + pyocf_ctx.register_volume_type(TraceDevice) + cache = Cache.start_on_device( + RamVolume(S.from_MiB(200)), + cache_mode=CacheMode.WB, + cache_line_size=cls, + metadata_volatile=True, + ) + flushing = False + writes = [] + + def trace(vol, io_type, rw, addr, nbytes, flags): + if flushing and io_type == TraceDevice.IoType.Data and rw == IoDir.WRITE: + writes.append(addr) + return True + + core_device = TraceDevice(RamVolume(S.from_MiB(300)), trace_fcn=trace) + core = Core.using_device(core_device) + cache.add_core(core) + cache.set_cleaning_policy(CleaningPolicy.NOP) + vol = CoreVolume(core) + vol.open() + queue = cache.get_default_queue() + line = int(cls) + + # 60 MiB written line by line in random order, so that the cache lines are allocated in + # an order unrelated to the core addresses + written = {} + dense = list(range(0, S.from_MiB(60).B, line)) + random.shuffle(dense) + for addr in dense: + written[addr] = write_random(vol, queue, addr, line) + # a few lines elsewhere + sparse = random.sample(range(S.from_MiB(200).B, S.from_MiB(300).B, line), 16) + for addr in sparse: + written[addr] = write_random(vol, queue, addr, line) + + flushing = True + cache.flush() + flushing = False + + assert dirty_bytes(cache.get_stats()) == S.from_MiB(0) + dense_writes = [addr for addr in writes if addr < S.from_MiB(100).B] + assert sorted(dense_writes) == list(range(0, S.from_MiB(60).B, line)), "every dense line written once" + adjacent = sum(1 for a, b in zip(dense_writes, dense_writes[1:]) if b == a + line) + assert adjacent >= 0.9 * (len(dense_writes) - 1), "contiguous data written in runs" + assert sorted(addr for addr in writes if addr >= S.from_MiB(200).B) == sorted(sparse) + core_bytes = core_device.vol.get_bytes() + assert all(core_bytes[addr : addr + line] == content for addr, content in written.items()) +