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/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) 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 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( 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] 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), ] 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()