Introduce ocf_mngt_cache_set_no_dirty()

inc/ocf_mngt.h

@@ -771,6 +771,18 @@ bool ocf_mngt_core_is_dirty(ocf_core_t core);
  */
 bool ocf_mngt_cache_is_dirty(ocf_cache_t cache);
 
+/**
+ * @brief Enable or disable new dirty data in cache
+ *
+ * When disabled, all new write requests are handled as if cache were in
+ * pass-through mode with respect to dirty data - no new dirty cachelines
+ * are introduced. Existing dirty data is not affected.
+ *
+ * @param[in] cache Cache handle
+ * @param[in] no_dirty true to prevent new dirty data, false to allow it
+ */
+void ocf_mngt_cache_set_no_dirty(ocf_cache_t cache, bool no_dirty);
+
 /**
  * @brief Completion callback of core flush operation
  *

src/mngt/ocf_mngt_cache.c

@@ -2414,6 +2414,11 @@ static void ocf_mngt_cache_stop_finish(ocf_pipeline_t pipeline,
 	ocf_mngt_cache_stop_end_t pipeline_cmpl;
 	void *completion_priv;
 
+	if (cache->no_dirty) {
+		env_refcnt_unfreeze(&cache->refcnt.dirty);
+		cache->no_dirty = false;
+	}
+
 	if (!error) {
 		ocf_mngt_cache_remove(context->ctx, cache);
 	} else {

src/mngt/ocf_mngt_flush.c

@@ -151,6 +151,21 @@ bool ocf_mngt_cache_is_dirty(ocf_cache_t cache)
 	return false;
 }
 
+void ocf_mngt_cache_set_no_dirty(ocf_cache_t cache, bool no_dirty)
+{
+	OCF_CHECK_NULL(cache);
+
+	if (no_dirty == cache->no_dirty)
+		return;
+
+	cache->no_dirty = no_dirty;
+
+	if (no_dirty)
+		env_refcnt_freeze(&cache->refcnt.dirty);
+	else
+		env_refcnt_unfreeze(&cache->refcnt.dirty);
+}
+
 /************************FLUSH CORE CODE**************************************/
 /* Returns:
  * 0 if OK and tbl & num is filled:

src/ocf_cache_priv.h

@@ -110,6 +110,8 @@ struct ocf_cache {
 	env_atomic flush_in_progress;
 	env_mutex flush_mutex;
 
+	bool no_dirty;
+
 	env_atomic attach_pt;
 
 	struct ocf_cleaner cleaner;

tests/functional/pyocf/types/cache.py

@@ -371,6 +371,9 @@ def standby_activate(self, device, open_cores=True):
             self.device = None
             raise OcfError("Failed to activate standby cache", c.results["error"])
 
+    def set_no_dirty(self, no_dirty: bool):
+        self.owner.lib.ocf_mngt_cache_set_no_dirty(self.cache_handle, no_dirty)
+
     def change_cache_mode(self, cache_mode: CacheMode):
         self.write_lock()
         status = self.owner.lib.ocf_mngt_cache_set_mode(self.cache_handle, cache_mode)

tests/functional/tests/management/test_no_dirty.py

@@ -0,0 +1,86 @@
+#
+# Copyright(c) 2026 Unvertical
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+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
+from pyocf.types.volume_core import CoreVolume
+from pyocf.rio import Rio, ReadWrite
+
+
+def test_no_dirty_prevents_new_dirty_data(pyocf_ctx):
+    cache = Cache.start_on_device(
+        RamVolume(S.from_MiB(50)), cache_mode=CacheMode.WB, metadata_volatile=True
+    )
+    core = Core.using_device(RamVolume(S.from_MiB(100)))
+    cache.add_core(core)
+    cache.set_cleaning_policy(CleaningPolicy.NOP)
+
+    cfv = CoreVolume(core)
+    queue = cache.get_default_queue()
+    r = Rio().target(cfv).bs(S.from_KiB(4))
+
+    r.copy().readwrite(ReadWrite.WRITE).size(S.from_MiB(1)).run([queue])
+
+    stats = cache.get_stats()
+    dirty_before = stats["usage"]["dirty"]["value"]
+    assert dirty_before > 0
+
+    cache.set_no_dirty(True)
+
+    r.copy().readwrite(ReadWrite.WRITE).size(S.from_MiB(1)).offset(S.from_MiB(1)).run([queue])
+
+    stats = cache.get_stats()
+    dirty_after = stats["usage"]["dirty"]["value"]
+    assert dirty_after == dirty_before
+
+
+def test_no_dirty_can_be_unset(pyocf_ctx):
+    cache = Cache.start_on_device(
+        RamVolume(S.from_MiB(50)), cache_mode=CacheMode.WB, metadata_volatile=True
+    )
+    core = Core.using_device(RamVolume(S.from_MiB(100)))
+    cache.add_core(core)
+    cache.set_cleaning_policy(CleaningPolicy.NOP)
+
+    cfv = CoreVolume(core)
+    queue = cache.get_default_queue()
+    r = Rio().target(cfv).bs(S.from_KiB(4))
+
+    cache.set_no_dirty(True)
+
+    r.copy().readwrite(ReadWrite.WRITE).size(S.from_MiB(1)).run([queue])
+
+    stats = cache.get_stats()
+    dirty_while_set = stats["usage"]["dirty"]["value"]
+    assert dirty_while_set == 0
+
+    cache.set_no_dirty(False)
+
+    r.copy().readwrite(ReadWrite.WRITE).size(S.from_MiB(1)).offset(S.from_MiB(1)).run([queue])
+
+    stats = cache.get_stats()
+    dirty_after_unset = stats["usage"]["dirty"]["value"]
+    assert dirty_after_unset > 0
+
+
+def test_no_dirty_stop_cache(pyocf_ctx):
+    cache = Cache.start_on_device(
+        RamVolume(S.from_MiB(50)), cache_mode=CacheMode.WB, metadata_volatile=True
+    )
+    core = Core.using_device(RamVolume(S.from_MiB(100)))
+    cache.add_core(core)
+    cache.set_cleaning_policy(CleaningPolicy.NOP)
+
+    cfv = CoreVolume(core)
+    queue = cache.get_default_queue()
+    r = Rio().target(cfv).bs(S.from_KiB(4))
+
+    r.copy().readwrite(ReadWrite.WRITE).size(S.from_MiB(1)).run([queue])
+
+    cache.set_no_dirty(True)
+    cache.flush()
+    cache.stop()