From 6ebc81f9d05398387d22696fb3f114c12b11ea15 Mon Sep 17 00:00:00 2001 From: Daniel Madej Date: Tue, 22 Apr 2025 08:26:17 +0200 Subject: [PATCH 1/6] Repart cache lines to default IO class Add ocf_user_part_move_all() moving all cache lines of a user partition to another one. It's meant to be called under exclusive metadata access, which keeps out every concurrent metadata update, so cache line locks are not needed and lines held by in-flight requests are moved as well. Moved lines are inserted at the cold end of the destination LRU lists (new ocf_lru_repart_cold()), preserving their relative order, so they neither displace hot data of the destination partition nor trigger its hot/cold rebalancing. The per-line part (cleaning policy and per-core counters update) is factored out of ocf_user_part_move() and shared by both paths. Signed-off-by: Daniel Madej Signed-off-by: Robert Baldyga --- src/ocf_lru.c | 47 ++++++++++++++ src/ocf_lru.h | 2 + src/utils/utils_user_part.c | 124 +++++++++++++++++++++++++----------- src/utils/utils_user_part.h | 3 + 4 files changed, 139 insertions(+), 37 deletions(-) diff --git a/src/ocf_lru.c b/src/ocf_lru.c index 27aadebc..d0433f34 100644 --- a/src/ocf_lru.c +++ b/src/ocf_lru.c @@ -141,6 +141,34 @@ static void add_lru_head(ocf_cache_t cache, struct ocf_lru_list *list, balance_lru_list(cache, list); } +/* Adds the given collision_index to the _tail_ of the LRU list as cold */ +static void add_lru_tail(ocf_cache_t cache, struct ocf_lru_list *list, + ocf_cache_line_t collision_index) +{ + struct ocf_lru_meta *node; + + ENV_BUG_ON(collision_index == end_marker); + + node = ocf_metadata_get_lru(cache, collision_index); + node->hot = false; + node->next = end_marker; + + if (!list->num_nodes) { + list->head = collision_index; + node->prev = end_marker; + } else { + ENV_BUG_ON(list->tail == end_marker); + + node->prev = list->tail; + ocf_metadata_get_lru(cache, list->tail)->next = collision_index; + } + + list->tail = collision_index; + ++list->num_nodes; + + balance_lru_list(cache, list); +} + /* update list global pointers and node neighbours to reflect removal */ static inline void remove_update_ptrs(ocf_cache_t cache, struct ocf_lru_list *list, ocf_cache_line_t collision_index, @@ -308,6 +336,25 @@ void ocf_lru_repart(ocf_cache_t cache, ocf_cache_line_t cline, OCF_METADATA_LRU_UNLOCK(lru_list); } +/* the caller must hold exclusive metadata access */ +void ocf_lru_repart_cold(ocf_cache_t cache, ocf_cache_line_t cline, + struct ocf_part *src_part, struct ocf_part *dst_part) +{ + uint32_t lru_list = OCF_LRU_GET_LIST_INDEX(cline); + struct ocf_lru_list *src_list, *dst_list; + bool clean; + + clean = !metadata_test_dirty(cache, cline); + src_list = ocf_lru_get_list(src_part, lru_list, clean); + dst_list = ocf_lru_get_list(dst_part, lru_list, clean); + + ocf_lru_remove_locked(cache, src_list, cline); + add_lru_tail(cache, dst_list, cline); + ocf_metadata_set_partition_id(cache, cline, dst_part->id); + env_atomic_dec(&src_part->runtime->curr_size); + env_atomic_inc(&dst_part->runtime->curr_size); +} + /* the caller must hold the metadata lock */ void ocf_lru_rm_cline(ocf_cache_t cache, ocf_cache_line_t cline) { diff --git a/src/ocf_lru.h b/src/ocf_lru.h index b42e4d33..2ee27870 100644 --- a/src/ocf_lru.h +++ b/src/ocf_lru.h @@ -34,6 +34,8 @@ void ocf_lru_clean(ocf_cache_t cache, struct ocf_user_part *user_part, ocf_queue_t io_queue, uint32_t count); void ocf_lru_repart(ocf_cache_t cache, ocf_cache_line_t cline, struct ocf_part *src_upart, struct ocf_part *dst_upart); +void ocf_lru_repart_cold(ocf_cache_t cache, ocf_cache_line_t cline, + struct ocf_part *src_part, struct ocf_part *dst_part); void ocf_lru_add_free(ocf_cache_t cache, ocf_cache_line_t cline); uint32_t ocf_lru_num_free(ocf_cache_t cache); struct ocf_lru_list *ocf_lru_get_list(struct ocf_part *part, diff --git a/src/utils/utils_user_part.c b/src/utils/utils_user_part.c index 26fea182..fb7257d9 100644 --- a/src/utils/utils_user_part.c +++ b/src/utils/utils_user_part.c @@ -94,6 +94,54 @@ void ocf_user_part_init(struct ocf_cache *cache) cache->user_parts[i].part.id = i; } +static void _ocf_user_part_move_line(struct ocf_cache *cache, + struct ocf_core *core, ocf_cache_line_t line, + ocf_part_id_t id_old, ocf_part_id_t id_new, bool cold) +{ + struct ocf_part *src_part = &cache->user_parts[id_old].part; + struct ocf_part *dst_part = &cache->user_parts[id_new].part; + + if (metadata_test_dirty(cache, line)) { + /* + * Remove cline from cleaning - this if for ioclass + * oriented cleaning policy (e.g. ALRU). + * TODO: Consider adding update_cache_line() ops + * to cleaning policy to let policies handle this. + */ + ocf_cleaning_purge_cache_block(cache, line); + } + + if (cold) + ocf_lru_repart_cold(cache, line, src_part, dst_part); + else + ocf_lru_repart(cache, line, src_part, dst_part); + + /* Check if cache line is dirty. If yes then need to change + * cleaning policy and update partition dirty clines + * statistics. + */ + if (metadata_test_dirty(cache, line)) { + /* Add cline back to cleaning policy */ + ocf_cleaning_set_hot_cache_line(cache, line); + + env_atomic_inc(&core->runtime_meta-> + part_counters[id_new].dirty_clines); + env_atomic_dec(&core->runtime_meta-> + part_counters[id_old].dirty_clines); + } + + /* Only valid cache lines are accounted. A line that is mapped, + * but not valid yet, gets accounted to its current partition + * once it becomes valid. + */ + if (metadata_test_valid_any(cache, line)) { + env_atomic_inc(&core->runtime_meta-> + part_counters[id_new].cached_clines); + env_atomic_dec(&core->runtime_meta-> + part_counters[id_old].cached_clines); + } +} + void ocf_user_part_move(struct ocf_request *req) { struct ocf_cache *cache = req->cache; @@ -130,45 +178,47 @@ void ocf_user_part_move(struct ocf_request *req) continue; } - if (metadata_test_dirty(cache, line)) { - /* - * Remove cline from cleaning - this if for ioclass - * oriented cleaning policy (e.g. ALRU). - * TODO: Consider adding update_cache_line() ops - * to cleaning policy to let policies handle this. - */ - ocf_cleaning_purge_cache_block(cache, line); - } - - ocf_lru_repart(cache, line, &cache->user_parts[id_old].part, - &cache->user_parts[id_new].part); - - /* Check if cache line is dirty. If yes then need to change - * cleaning policy and update partition dirty clines - * statistics. - */ - if (metadata_test_dirty(cache, line)) { - /* Add cline back to cleaning policy */ - ocf_cleaning_set_hot_cache_line(cache, line); - - env_atomic_inc(&req->core->runtime_meta-> - part_counters[id_new].dirty_clines); - env_atomic_dec(&req->core->runtime_meta-> - part_counters[id_old].dirty_clines); - } + _ocf_user_part_move_line(cache, req->core, line, + id_old, id_new, false); + } +} - /* Only valid cache lines are accounted. A line that is mapped, - * but not valid yet, gets accounted to its current partition - * once it becomes valid. - */ - if (metadata_test_valid_any(cache, line)) { - env_atomic_inc(&req->core->runtime_meta-> - part_counters[id_new].cached_clines); - env_atomic_dec(&req->core->runtime_meta-> - part_counters[id_old].cached_clines); +/* + * Move all cache lines of user partition id_old to user partition id_new. + * Lines are inserted at the cold end of destination LRU lists, preserving + * their relative order. Cache line locks are not taken, so the caller must + * hold exclusive metadata access to keep out any concurrent metadata update. + */ +void ocf_user_part_move_all(struct ocf_cache *cache, ocf_part_id_t id_old, + ocf_part_id_t id_new) +{ + struct ocf_part *src_part = &cache->user_parts[id_old].part; + struct ocf_lru_list *list; + struct ocf_core *core; + ocf_cache_line_t line; + ocf_core_id_t core_id; + uint32_t lru_idx; + uint32_t step = 0; + int clean; + + ENV_BUG_ON(id_old >= OCF_USER_IO_CLASS_MAX || + id_new >= OCF_USER_IO_CLASS_MAX); + + if (id_old == id_new) + return; + + for (lru_idx = 0; lru_idx < OCF_NUM_LRU_LISTS; lru_idx++) { + for (clean = 0; clean <= 1; clean++) { + list = ocf_lru_get_list(src_part, lru_idx, clean); + while (list->num_nodes) { + line = list->head; + core_id = ocf_metadata_get_core_id(cache, line); + core = ocf_cache_get_core(cache, core_id); + _ocf_user_part_move_line(cache, core, line, + id_old, id_new, true); + OCF_COND_RESCHED_DEFAULT(step); + } } - - /* DONE */ } } diff --git a/src/utils/utils_user_part.h b/src/utils/utils_user_part.h index 481e53c8..6eb1e110 100644 --- a/src/utils/utils_user_part.h +++ b/src/utils/utils_user_part.h @@ -88,6 +88,9 @@ static inline uint32_t ocf_user_part_get_max_size(ocf_cache_t cache, void ocf_user_part_move(struct ocf_request *req); +void ocf_user_part_move_all(struct ocf_cache *cache, ocf_part_id_t id_old, + ocf_part_id_t id_new); + #define for_each_user_part(cache, user_part, id) \ for_each_lst_entry(&cache->user_part_list, user_part, id, \ struct ocf_user_part, lst_valid) From db4e474c0a7d759c1ee8475ce3f2fe7216138216 Mon Sep 17 00:00:00 2001 From: Daniel Madej Date: Tue, 22 Apr 2025 08:26:17 +0200 Subject: [PATCH 2/6] Optional repartitioning when reconfiguring IO classes Add a per IO class `repart` flag to struct ocf_mngt_io_class_config. When set, ocf_mngt_cache_io_classes_configure() moves cache lines of that IO class to the unclassified IO class, both when the class is updated and when it's removed. The lines get reclassified on the next access. Only the user knows whether a modified classification rule changes the meaning of the IO class, so the decision is left to them per IO class. Repartitioning is done after the whole configuration is applied, as it can't be rolled back. Update the pyocf mirror of the structure accordingly. Signed-off-by: Daniel Madej Signed-off-by: Robert Baldyga --- inc/ocf_mngt.h | 10 ++++++++++ src/mngt/ocf_mngt_io_class.c | 23 +++++++++++++++++++++++ tests/functional/pyocf/types/ioclass.py | 3 ++- 3 files changed, 35 insertions(+), 1 deletion(-) diff --git a/inc/ocf_mngt.h b/inc/ocf_mngt.h index 27917b26..2674cdda 100644 --- a/inc/ocf_mngt.h +++ b/inc/ocf_mngt.h @@ -1127,6 +1127,16 @@ struct ocf_mngt_io_class_config { * @brief IO class eviction priority */ int16_t prio; + + /** + * @brief Move cache lines currently assigned to this IO class + * to the unclassified IO class + * + * Applies both to an IO class being updated and to one being removed + * (name set to NULL). Lines will be reclassified on next access. + * Ignored for the unclassified and prefetch IO classes. + */ + bool repart; }; struct ocf_mngt_io_classes_config { diff --git a/src/mngt/ocf_mngt_io_class.c b/src/mngt/ocf_mngt_io_class.c index a356eb18..8557a58c 100644 --- a/src/mngt/ocf_mngt_io_class.c +++ b/src/mngt/ocf_mngt_io_class.c @@ -283,6 +283,7 @@ int ocf_mngt_cache_io_classes_configure(ocf_cache_t cache, const struct ocf_mngt_io_classes_config *cfg) { struct ocf_user_part *old_config; + ocf_part_id_t part_id; int result; int i; @@ -318,6 +319,28 @@ int ocf_mngt_cache_io_classes_configure(ocf_cache_t cache, } } + /* Repartitioning can't be rolled back, so do it only once the whole + * configuration has been applied successfully. + */ + for (i = 0; i < OCF_USER_IO_CLASS_MAX; i++) { + part_id = cfg->config[i].class_id; + + /* Classification rules of the default IO classes can't change, + * so their cache lines can't become misclassified. + */ + if (!cfg->config[i].repart || + part_id == OCF_IO_CLASS_UNCLASSIFIED || + part_id == OCF_IO_CLASS_PREFETCH) { + continue; + } + + ocf_user_part_move_all(cache, part_id, + OCF_IO_CLASS_UNCLASSIFIED); + ocf_cache_log(cache, log_info, "Moved cache lines of IO class, " + "id: %u to the unclassified IO class\n", + part_id); + } + ocf_user_part_sort(cache); out_edit: diff --git a/tests/functional/pyocf/types/ioclass.py b/tests/functional/pyocf/types/ioclass.py index 434d351f..3f6c4c6f 100644 --- a/tests/functional/pyocf/types/ioclass.py +++ b/tests/functional/pyocf/types/ioclass.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD-3-Clause # -from ctypes import c_uint8, c_uint32, c_int, c_int16, c_uint16, c_char, c_char_p, Structure +from ctypes import c_bool, c_uint8, c_uint32, c_int, c_int16, c_uint16, c_char, c_char_p, Structure class IoClassInfo(Structure): @@ -27,6 +27,7 @@ class IoClassConfig(Structure): ("_name", c_char_p), ("_cache_mode", c_int), ("_priority", c_uint16), + ("_repart", c_bool), ] From 54ca0289b222a5280bcb864254653762eb22aa06 Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Fri, 16 Oct 2026 11:36:33 +0200 Subject: [PATCH 3/6] pyocf: rio: Add io_class option Signed-off-by: Robert Baldyga --- tests/functional/pyocf/rio.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/tests/functional/pyocf/rio.py b/tests/functional/pyocf/rio.py index c811eea5..fa904810 100644 --- a/tests/functional/pyocf/rio.py +++ b/tests/functional/pyocf/rio.py @@ -1,6 +1,7 @@ # # Copyright(c) 2022 Intel Corporation # Copyright(c) 2024 Huawei Technologies +# Copyright(c) 2026 Unvertical # SPDX-License-Identifier: BSD-3-Clause # @@ -72,6 +73,7 @@ class JobSpec: time_based: bool = False time: timedelta = None continue_on_error: bool = False + io_class: int = 0 def merge(self, other): # TODO implement @@ -155,7 +157,8 @@ def run(self): data = Data(self.jobspec.bs) # TODO pattern and verify io = self.jobspec.target.new_io( - self.queue, next(iogen), self.jobspec.bs, iodir, 0, 0, + self.queue, next(iogen), self.jobspec.bs, iodir, + self.jobspec.io_class, 0, ) io.set_data(data) io.callback = self.get_io_cb() @@ -248,6 +251,10 @@ def time(self, time: timedelta): self.global_jobspec.time = time return self + def io_class(self, io_class: int): + self.global_jobspec.io_class = io_class + return self + def continue_on_error(self): self.global_jobspec.continue_on_error = True return self From 6b9225af9c28b0419dedfcef4f7387cd784d2e16 Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Fri, 16 Oct 2026 14:42:37 +0200 Subject: [PATCH 4/6] pyocf: Add core ioclass stats Signed-off-by: Robert Baldyga --- tests/functional/pyocf/types/core.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/tests/functional/pyocf/types/core.py b/tests/functional/pyocf/types/core.py index 021d15f3..073981dc 100644 --- a/tests/functional/pyocf/types/core.py +++ b/tests/functional/pyocf/types/core.py @@ -151,6 +151,26 @@ def get_stats(self): "errors": struct_to_dict(errors), } + def get_ioclass_stats(self, io_class_id): + usage = UsageStats() + req = RequestsStats() + blocks = BlocksStats() + + self.cache.read_lock() + status = self.cache.owner.lib.ocf_stats_collect_part_core( + self.handle, io_class_id, byref(usage), byref(req), byref(blocks) + ) + self.cache.read_unlock() + + if status: + raise OcfError(f"Failed to retrieve core ioclass {io_class_id} stats", status) + + return { + "usage": struct_to_dict(usage), + "req": struct_to_dict(req), + "blocks": struct_to_dict(blocks), + } + def set_seq_cut_off_policy(self, policy: SeqCutOffPolicy): self.cache.write_lock() @@ -218,6 +238,8 @@ def reset_stats(self): lib.ocf_mngt_core_set_seq_detect_promotion_count.restype = c_int lib.ocf_stats_collect_core.argtypes = [c_void_p, c_void_p, c_void_p, c_void_p, c_void_p] lib.ocf_stats_collect_core.restype = c_int +lib.ocf_stats_collect_part_core.argtypes = [c_void_p, c_uint16, c_void_p, c_void_p, c_void_p] +lib.ocf_stats_collect_part_core.restype = c_int lib.ocf_core_get_info.argtypes = [c_void_p, c_void_p] lib.ocf_core_get_info.restype = c_int lib.ocf_core_stats_initialize.argtypes = [c_void_p] From 970b4d4c5cc748df03e676f6d27977d5e2aea290 Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Sat, 17 Oct 2026 00:39:33 +0200 Subject: [PATCH 5/6] pyocf: Add IO class repart and removal to Cache Signed-off-by: Robert Baldyga --- tests/functional/pyocf/types/cache.py | 42 +++++++++++++++++++-------- 1 file changed, 30 insertions(+), 12 deletions(-) diff --git a/tests/functional/pyocf/types/cache.py b/tests/functional/pyocf/types/cache.py index 0517a36a..56eef742 100644 --- a/tests/functional/pyocf/types/cache.py +++ b/tests/functional/pyocf/types/cache.py @@ -612,14 +612,7 @@ def add_partition( if status: raise OcfError("Error adding partition to cache", status) - def configure_partition( - self, - part_id: int, - name: str, - max_size: int, - priority: int, - cache_mode=CACHE_MODE_NONE, - ): + def _get_ioclasses_config(self): ioclasses_info = IoClassesInfo() self.read_lock() @@ -641,11 +634,9 @@ def configure_partition( self.read_unlock() - ioclasses_info._config[part_id]._name = name.encode("utf-8") - ioclasses_info._config[part_id]._cache_mode = int(cache_mode) - ioclasses_info._config[part_id]._priority = priority - ioclasses_info._config[part_id]._max_size = max_size + return ioclasses_info + def _set_ioclasses_config(self, ioclasses_info): self.write_lock() status = self.owner.lib.ocf_mngt_cache_io_classes_configure( @@ -657,6 +648,33 @@ def configure_partition( if status: raise OcfError("Error adding partition to cache", status) + def configure_partition( + self, + part_id: int, + name: str, + max_size: int, + priority: int, + cache_mode=CACHE_MODE_NONE, + repart: bool = False, + ): + ioclasses_info = self._get_ioclasses_config() + + ioclasses_info._config[part_id]._name = name.encode("utf-8") + ioclasses_info._config[part_id]._cache_mode = int(cache_mode) + ioclasses_info._config[part_id]._priority = priority + ioclasses_info._config[part_id]._max_size = max_size + ioclasses_info._config[part_id]._repart = repart + + self._set_ioclasses_config(ioclasses_info) + + def remove_partition(self, part_id: int, repart: bool = False): + ioclasses_info = self._get_ioclasses_config() + + ioclasses_info._config[part_id]._name = None + ioclasses_info._config[part_id]._repart = repart + + self._set_ioclasses_config(ioclasses_info) + def alloc_device_config(self, device, perform_test=True): if not device.handle: uuid = Uuid( From f6e78a2663dee35914cd25e2152db5495490560f Mon Sep 17 00:00:00 2001 From: Robert Baldyga Date: Sat, 17 Oct 2026 00:39:33 +0200 Subject: [PATCH 6/6] pyocf: Add IO class repart tests Signed-off-by: Robert Baldyga --- .../tests/management/test_ioclass_repart.py | 552 ++++++++++++++++++ 1 file changed, 552 insertions(+) create mode 100644 tests/functional/tests/management/test_ioclass_repart.py diff --git a/tests/functional/tests/management/test_ioclass_repart.py b/tests/functional/tests/management/test_ioclass_repart.py new file mode 100644 index 00000000..7aec3382 --- /dev/null +++ b/tests/functional/tests/management/test_ioclass_repart.py @@ -0,0 +1,552 @@ +# +# Copyright(c) 2026 Unvertical +# SPDX-License-Identifier: BSD-3-Clause +# + +import time +from threading import Thread + +import pytest + +from pyocf.types.cache import ( + Cache, + CacheMode, + CleaningPolicy, + PrefetchPolicy, + ReadaheadParams, +) +from pyocf.types.core import Core +from pyocf.types.data import Data +from pyocf.types.io import Io, IoDir +from pyocf.rio import ReadWrite, Rio +from pyocf.types.shared import CacheLineSize, OcfError, OcfErrorCode, SeqCutOffPolicy +from pyocf.types.volume import RamVolume +from pyocf.types.volume_core import CoreVolume +from pyocf.utils import Size + +UNCLASSIFIED = 0 +PREFETCH = 33 +LINES = 128 + + +def fill(vol, ioclass, first_line=0, rw=ReadWrite.WRITE): + rio = ( + Rio() + .target(vol) + .bs(Size.from_KiB(4)) + .offset(Size.from_KiB(4 * first_line)) + .size(Size.from_KiB(4 * LINES)) + .readwrite(rw) + .io_class(ioclass) + ) + rio.run([vol.parent.get_default_queue()]) + + assert rio.error_count == 0 + + +def usage(cache, ioclass): + stats = cache.get_ioclass_stats(ioclass)["usage"] + return stats["occupancy"]["value"], stats["dirty"]["value"] + + +def prepare(cache_mode, core_device=None): + cache_device = RamVolume(Size.from_MiB(50)) + if core_device is None: + core_device = RamVolume(Size.from_MiB(50)) + cache = Cache.start_on_device( + cache_device, cache_mode=cache_mode, cache_line_size=CacheLineSize.LINE_4KiB + ) + core = Core.using_device(core_device) + cache.add_core(core) + cache.set_seq_cut_off_policy(SeqCutOffPolicy.NEVER) + + return cache_device, cache, CoreVolume(core) + + +class HoldVolume(RamVolume): + """RAM volume able to hold IOs in flight until released""" + + def __init__(self, size): + super().__init__(size) + self.hold = False + self.held = [] + + def do_forward_io(self, token, rw, addr, nbytes, offset): + if self.hold: + self.held.append((token, rw, addr, nbytes, offset)) + else: + super().do_forward_io(token, rw, addr, nbytes, offset) + + def release(self, error=0): + self.hold = False + held, self.held = self.held, [] + for args in held: + if error: + Io.forward_end(args[0], error) + else: + super().do_forward_io(*args) + + +def wait_for(condition, timeout=10): + deadline = time.monotonic() + timeout + while not condition(): + assert time.monotonic() < deadline, "Timed out waiting for condition" + time.sleep(0.01) + + +def submit_async(vol, addr, io_dir, ioclass, results): + io = vol.new_io(vol.parent.get_default_queue(), addr, 4096, io_dir, ioclass, 0) + data = Data(4096) + io.set_data(data) + io.callback = lambda err: results.append(err) + io.submit() + # Keep IO and its data alive until completion + return io, data + + +def configure_in_thread(func, timeout=10): + """Run configuration in a thread to detect a deadlock instead of hanging""" + errors = [] + + def run(): + try: + func() + except Exception as e: # noqa E722 + errors.append(e) + + thread = Thread(target=run) + thread.start() + thread.join(timeout) + assert not thread.is_alive(), "IO class configuration deadlocked" + assert not errors, errors + + +def check_totals(cache): + """Occupancy and dirty of all IO classes must sum up to cache totals""" + stats = cache.get_stats()["usage"] + occupancy, dirty = 0, 0 + for ioclass in range(PREFETCH + 1): + try: + o, d = usage(cache, ioclass) + except OcfError: + continue + assert o <= stats["occupancy"]["value"], f"IO class {ioclass} occupancy {o}" + assert d <= stats["dirty"]["value"], f"IO class {ioclass} dirty {d}" + occupancy += o + dirty += d + assert occupancy == stats["occupancy"]["value"] + assert dirty == stats["dirty"]["value"] + + +@pytest.mark.parametrize("cache_mode", [CacheMode.WT, CacheMode.WB]) +def test_ioclass_repart_update(pyocf_ctx, cache_mode): + """ + Verify that cache lines of an updated IO class are moved to the unclassified + IO class only if requested, that dirty data stays consistent and survives + flush and cache load, and that lines get reclassified on next access. + """ + cache_device, cache, vol = prepare(cache_mode) + dirty = LINES if cache_mode == CacheMode.WB else 0 + + cache.configure_partition(part_id=1, name="class1", max_size=100, priority=1) + fill(vol, 1) + assert usage(cache, 1) == (LINES, dirty) + assert usage(cache, UNCLASSIFIED) == (0, 0) + + cache.configure_partition(part_id=1, name="class1_v2", max_size=100, priority=1) + assert usage(cache, 1) == (LINES, dirty) + assert usage(cache, UNCLASSIFIED) == (0, 0) + + cache.configure_partition( + part_id=1, name="class1_v3", max_size=100, priority=1, repart=True + ) + assert usage(cache, 1) == (0, 0) + assert usage(cache, UNCLASSIFIED) == (LINES, dirty) + + # Hits reclassify lines back to the IO class of the request + fill(vol, 1, rw=ReadWrite.READ) + assert usage(cache, 1) == (LINES, dirty) + assert usage(cache, UNCLASSIFIED) == (0, 0) + + cache.configure_partition( + part_id=1, name="class1_v3", max_size=100, priority=1, repart=True + ) + cache.flush() + assert usage(cache, UNCLASSIFIED) == (LINES, 0) + assert cache.get_stats()["usage"]["dirty"]["value"] == 0 + + cache.stop() + cache = Cache.load_from_device(cache_device) + assert usage(cache, 1) == (0, 0) + assert usage(cache, UNCLASSIFIED) == (LINES, 0) + + +@pytest.mark.parametrize("cache_mode", [CacheMode.WT, CacheMode.WB]) +def test_ioclass_repart_remove(pyocf_ctx, cache_mode): + """ + Verify that removing an IO class moves its cache lines to the unclassified + IO class only if requested. + """ + cache_device, cache, vol = prepare(cache_mode) + dirty = LINES if cache_mode == CacheMode.WB else 0 + + cache.configure_partition(part_id=1, name="class1", max_size=100, priority=1) + cache.configure_partition(part_id=2, name="class2", max_size=100, priority=1) + fill(vol, 1, first_line=0) + fill(vol, 2, first_line=LINES) + assert usage(cache, 1) == (LINES, dirty) + assert usage(cache, 2) == (LINES, dirty) + + cache.remove_partition(1, repart=True) + cache.remove_partition(2, repart=False) + assert usage(cache, UNCLASSIFIED) == (LINES, dirty) + + # Re-added IO class 1 starts empty, IO class 2 keeps its old lines + cache.configure_partition(part_id=1, name="class1", max_size=100, priority=1) + cache.configure_partition(part_id=2, name="class2", max_size=100, priority=1) + assert usage(cache, 1) == (0, 0) + assert usage(cache, 2) == (LINES, dirty) + assert usage(cache, UNCLASSIFIED) == (LINES, dirty) + + cache.flush() + assert cache.get_stats()["usage"]["dirty"]["value"] == 0 + + +@pytest.mark.parametrize("cache_mode", [CacheMode.WT, CacheMode.WB]) +def test_ioclass_repart_inactive(pyocf_ctx, cache_mode): + """ + Verify that repart of an already removed IO class still holding cache + lines moves them to the unclassified IO class, and that repart of an + empty IO class doesn't change anything. + """ + cache_device, cache, vol = prepare(cache_mode) + dirty = LINES if cache_mode == CacheMode.WB else 0 + + cache.configure_partition(part_id=1, name="class1", max_size=100, priority=1) + cache.configure_partition(part_id=2, name="class2", max_size=100, priority=1) + fill(vol, 1, first_line=0) + fill(vol, 2, first_line=LINES) + + # Removed IO class keeps its cache lines + cache.remove_partition(1, repart=False) + assert cache.get_stats()["usage"]["occupancy"]["value"] == 2 * LINES + assert usage(cache, UNCLASSIFIED) == (0, 0) + + # Repart of the already removed IO class moves them + cache.remove_partition(1, repart=True) + assert usage(cache, UNCLASSIFIED) == (LINES, dirty) + assert usage(cache, 2) == (LINES, dirty) + check_totals(cache) + + # Repart of empty IO classes, both removed and active, changes nothing + cache.configure_partition(part_id=3, name="class3", max_size=100, priority=1) + cache.remove_partition(1, repart=True) + cache.configure_partition(part_id=3, name="class3", max_size=100, priority=1, repart=True) + cache.remove_partition(4, repart=True) + assert usage(cache, UNCLASSIFIED) == (LINES, dirty) + assert usage(cache, 2) == (LINES, dirty) + assert usage(cache, 3) == (0, 0) + check_totals(cache) + + # Re-added IO class 1 starts empty + cache.configure_partition(part_id=1, name="class1", max_size=100, priority=1) + assert usage(cache, 1) == (0, 0) + + cache.flush() + assert cache.get_stats()["usage"]["dirty"]["value"] == 0 + + +@pytest.mark.parametrize("ioclass", [UNCLASSIFIED, PREFETCH]) +def test_ioclass_repart_default_ioclass(pyocf_ctx, ioclass): + """ + Verify that the repart flag is ignored for the default IO classes, whose + classification rules can't change. + """ + cache_device, cache, vol = prepare(CacheMode.WT) + + cache.set_prefetch_policy(1 << PrefetchPolicy.READAHEAD) + cache.set_prefetch_param(PrefetchPolicy.READAHEAD, ReadaheadParams.THRESHOLD, 0) + + # Sequential reads populate the unclassified IO class and trigger + # readahead populating the prefetch IO class + fill(vol, UNCLASSIFIED, rw=ReadWrite.READ) + cache.settle() + + unclassified = usage(cache, UNCLASSIFIED) + prefetch = usage(cache, PREFETCH) + assert unclassified[0] > 0 + assert prefetch[0] > 0 + + info = cache.get_partition_info(ioclass) + cache.configure_partition( + part_id=ioclass, + name=info["_name"], + max_size=info["_max_size"], + priority=info["_priority"], + cache_mode=info["_cache_mode"], + repart=True, + ) + + assert usage(cache, UNCLASSIFIED) == unclassified + assert usage(cache, PREFETCH) == prefetch + + +@pytest.mark.parametrize("cache_mode", [CacheMode.WT, CacheMode.WB]) +def test_ioclass_repart_inflight_io(pyocf_ctx, cache_mode): + """ + Verify that repartitioning moves cache lines held by in-flight requests, + including lines mapped by a read miss that aren't valid yet, and that + accounting and data stay consistent once the requests complete. + """ + pyocf_ctx.register_volume_type(HoldVolume) + core_device = HoldVolume(Size.from_MiB(10)) + cache_device, cache, vol = prepare(cache_mode, core_device) + dirty = LINES if cache_mode == CacheMode.WB else 0 + + cache.configure_partition(part_id=1, name="class1", max_size=100, priority=1) + fill(vol, 1) + + # Read misses of IO class 1 get held on core read, with their cache lines + # mapped to IO class 1, but not valid yet. In WT mode write hits get held + # on core write as well. + core_device.hold = True + vol.open() + results, ios = [], [] + for i in range(LINES, 2 * LINES): + ios.append(submit_async(vol, i * 4096, IoDir.READ, 1, results)) + expected_held = LINES + if cache_mode == CacheMode.WT: + for i in range(LINES // 2): + ios.append(submit_async(vol, i * 4096, IoDir.WRITE, 1, results)) + expected_held += LINES // 2 + wait_for(lambda: len(core_device.held) == expected_held) + + configure_in_thread( + lambda: cache.configure_partition( + part_id=1, name="class1", max_size=100, priority=1, repart=True + ) + ) + + core_device.release() + wait_for(lambda: len(results) == len(ios)) + vol.close() + cache.settle() + assert results == [0] * len(ios) + + assert usage(cache, 1) == (0, 0) + assert usage(cache, UNCLASSIFIED) == (2 * LINES, dirty) + check_totals(cache) + + cache.flush() + assert usage(cache, UNCLASSIFIED) == (2 * LINES, 0) + check_totals(cache) + + cache.stop() + cache = Cache.load_from_device(cache_device) + assert usage(cache, 1) == (0, 0) + assert usage(cache, UNCLASSIFIED) == (2 * LINES, 0) + + # Reading the whole exported object inserts it into the cache, so check + # data as the last step + vol = CoreVolume(cache.get_core_by_name("core")) + assert core_device.md5() == vol.md5() + + +def test_ioclass_lazy_repart_invalid_line(pyocf_ctx): + """ + Verify accounting when a request reclassifies a cache line that is mapped, + but not valid: a read miss fails, leaving its cache line mapped for the + request of another IO class waiting for it. + """ + pyocf_ctx.register_volume_type(HoldVolume) + core_device = HoldVolume(Size.from_MiB(10)) + cache_device, cache, vol = prepare(CacheMode.WT, core_device) + + cache.configure_partition(part_id=1, name="class1", max_size=100, priority=1) + cache.configure_partition(part_id=2, name="class2", max_size=100, priority=1) + + core_device.hold = True + vol.open() + first, second, ios = [], [], [] + for i in range(LINES): + ios.append(submit_async(vol, i * 4096, IoDir.READ, 1, first)) + wait_for(lambda: len(core_device.held) == LINES) + + # Requests of IO class 2 wait for the cache lines locked by IO class 1 + for i in range(LINES): + ios.append(submit_async(vol, i * 4096, IoDir.READ, 2, second)) + time.sleep(0.5) + assert len(second) == 0 + + core_device.release(error=-OcfErrorCode.OCF_ERR_IO) + wait_for(lambda: len(first) == LINES and len(second) == LINES) + vol.close() + cache.settle() + assert all(err != 0 for err in first) + assert second == [0] * LINES + + assert usage(cache, 1) == (0, 0) + assert usage(cache, 2) == (LINES, 0) + check_totals(cache) + + +def test_ioclass_repart_during_cleaning(pyocf_ctx): + """ + Verify that repartitioning dirty cache lines that are being cleaned keeps + dirty accounting consistent. Cleaning policy runs hold the cache + management lock, so they can't overlap with IO class configuration, but + cleaning triggered by eviction runs without it. + """ + pyocf_ctx.register_volume_type(HoldVolume) + core_device = HoldVolume(Size.from_MiB(20)) + cache_device, cache, vol = prepare(CacheMode.WB, core_device) + cache.set_cleaning_policy(CleaningPolicy.NOP) + cache_lines = cache.get_stats()["conf"]["size"].blocks_4k + + # Fill the whole cache with dirty data, so that eviction requires cleaning + cache.configure_partition(part_id=1, name="class1", max_size=100, priority=1) + rio = ( + Rio() + .target(vol) + .bs(Size.from_KiB(4)) + .size(Size.from_KiB(4 * cache_lines)) + .readwrite(ReadWrite.WRITE) + .io_class(1) + ) + rio.run([vol.parent.get_default_queue()]) + assert rio.error_count == 0 + assert usage(cache, 1) == (cache_lines, cache_lines) + + # Write a new block, triggering cleaning, held on core write + core_device.hold = True + vol.open() + results = [] + io = submit_async(vol, cache_lines * 4096, IoDir.WRITE, 1, results) + wait_for(lambda: len(core_device.held) > 0) + time.sleep(0.5) + + configure_in_thread( + lambda: cache.configure_partition( + part_id=1, name="class1", max_size=100, priority=1, repart=True + ) + ) + assert usage(cache, 1) == (0, 0) + assert usage(cache, UNCLASSIFIED) == (cache_lines, cache_lines) + + core_device.release() + wait_for(lambda: len(results) == 1) + vol.close() + cache.settle() + assert results == [0] + + occupancy, dirty = usage(cache, UNCLASSIFIED) + assert dirty < cache_lines, "Some cache lines should have been cleaned" + assert usage(cache, 1)[1] == 0 + check_totals(cache) + + cache.flush() + assert usage(cache, UNCLASSIFIED)[1] == 0 + check_totals(cache) + assert core_device.md5() == vol.md5() + + +@pytest.mark.parametrize("cache_mode", [CacheMode.WT, CacheMode.WB]) +def test_ioclass_repart_multiple_cores(pyocf_ctx, cache_mode): + """ + Verify per core accounting of repartitioned cache lines, including removal + of a core afterwards. + """ + cache_device, cache, vol1 = prepare(cache_mode) + core2 = Core.using_device(RamVolume(Size.from_MiB(50)), name="core2") + cache.add_core(core2) + vol2 = CoreVolume(core2) + core1 = vol1.core + dirty = LINES if cache_mode == CacheMode.WB else 0 + + def core_usage(core, ioclass): + stats = core.get_ioclass_stats(ioclass)["usage"] + return stats["occupancy"]["value"], stats["dirty"]["value"] + + cache.configure_partition(part_id=1, name="class1", max_size=100, priority=1) + fill(vol1, 1) + fill(vol2, 1, first_line=LINES) + fill(vol2, UNCLASSIFIED, first_line=2 * LINES) + + cache.configure_partition( + part_id=1, name="class1", max_size=100, priority=1, repart=True + ) + assert core_usage(core1, 1) == (0, 0) + assert core_usage(core2, 1) == (0, 0) + assert core_usage(core1, UNCLASSIFIED) == (LINES, dirty) + assert core_usage(core2, UNCLASSIFIED) == (2 * LINES, 2 * dirty) + check_totals(cache) + + cache.remove_core(core1) + assert usage(cache, 1) == (0, 0) + assert usage(cache, UNCLASSIFIED) == (2 * LINES, 2 * dirty) + check_totals(cache) + + cache.flush() + assert usage(cache, UNCLASSIFIED) == (2 * LINES, 0) + check_totals(cache) + + +@pytest.mark.parametrize("cache_mode", [CacheMode.WT, CacheMode.WB]) +def test_ioclass_repart_partial_lines(pyocf_ctx, cache_mode): + """ + Verify repartitioning of partially valid and partially dirty cache lines. + """ + cache_device = RamVolume(Size.from_MiB(50)) + core_device = RamVolume(Size.from_MiB(50)) + cache = Cache.start_on_device( + cache_device, cache_mode=cache_mode, cache_line_size=CacheLineSize.LINE_64KiB + ) + core = Core.using_device(core_device) + cache.add_core(core) + cache.set_seq_cut_off_policy(SeqCutOffPolicy.NEVER) + vol = CoreVolume(core) + line = int(CacheLineSize.LINE_64KiB) + blocks = line // 4096 + + cache.configure_partition(part_id=1, name="class1", max_size=100, priority=1) + + # A single 4KiB block in each cache line, at varying positions + vol.open() + for i in range(LINES): + vol.sync_io( + vol.parent.get_default_queue(), i * line + (i % blocks) * 4096, + Data(4096), IoDir.WRITE, io_class=1 + ) + vol.close() + + # Usage is accounted in whole cache lines + full = LINES * blocks + dirty = full if cache_mode == CacheMode.WB else 0 + before = usage(cache, 1) + assert before == (full, dirty) + + cache.configure_partition( + part_id=1, name="class1", max_size=100, priority=1, repart=True + ) + assert usage(cache, 1) == (0, 0) + assert usage(cache, UNCLASSIFIED) == before + check_totals(cache) + + # Fill rest of the cache lines from IO class 1, getting them reclassified + rio = ( + Rio() + .target(vol) + .bs(Size.from_KiB(4)) + .size(Size(LINES * line)) + .readwrite(ReadWrite.WRITE) + .io_class(1) + ) + rio.run([vol.parent.get_default_queue()]) + assert rio.error_count == 0 + assert usage(cache, 1) == (full, dirty) + assert usage(cache, UNCLASSIFIED) == (0, 0) + check_totals(cache) + + cache.flush() + check_totals(cache) + assert core_device.md5() == vol.md5()