Limit read lock holders

src/utils/utils_alock.c

@@ -33,9 +33,10 @@
 #define OCF_DEBUG_CACHE(cache, format, ...)
 #endif
 
-#define OCF_CACHE_LINE_ACCESS_WR	((unsigned char)(-1))
 #define OCF_CACHE_LINE_ACCESS_IDLE	0
 #define OCF_CACHE_LINE_ACCESS_ONE_RD	1
+#define OCF_CACHE_LINE_ACCESS_MAX_RD	254
+#define OCF_CACHE_LINE_ACCESS_WR	255
 
 #define _WAITERS_LIST_SIZE	(16UL * MiB)
 #define _WAITERS_LIST_ENTRIES \
@@ -291,12 +292,28 @@ bool ocf_alock_trylock_entry_rd_idle(struct ocf_alock *alock,
 	return (prev == OCF_CACHE_LINE_ACCESS_IDLE);
 }
 
+/*
+ * The access counter is 8 bit wide and its top value is reserved for the
+ * write lock, so no more than OCF_CACHE_LINE_ACCESS_MAX_RD readers can hold
+ * the entry at the same time. Incrementing the counter beyond that limit
+ * would make the entry indistinguishable from a write locked one, so the
+ * excess readers have to be put on the waiters list instead.
+ */
 static inline bool ocf_alock_trylock_entry_rd(struct ocf_alock *alock,
 		ocf_cache_line_t entry)
 {
 	env_atomic8 *access = &alock->access[entry];
+	uint8_t old, v = env_atomic8_read(access);
+
+	while (v != OCF_CACHE_LINE_ACCESS_WR &&
+			v != OCF_CACHE_LINE_ACCESS_MAX_RD) {
+		old = env_atomic8_cmpxchg(access, v, v + 1);
+		if (likely(old == v))
+			return true;
+		v = old;
+	}
 
-	return !!env_atomic8_add_unless(access, 1, OCF_CACHE_LINE_ACCESS_WR);
+	return false;
 }
 
 static inline void ocf_alock_unlock_entry_wr(struct ocf_alock *alock,

tests/functional/tests/engine/test_concurrent_readers.py

@@ -0,0 +1,191 @@
+#
+# Copyright(c) 2026 Unvertical
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import time
+from ctypes import c_int
+from threading import Lock
+
+from pyocf.types.cache import Cache, CacheMode
+from pyocf.types.core import Core
+from pyocf.types.data import Data
+from pyocf.types.io import IoDir
+from pyocf.types.shared import CacheLineSize, OcfCompletion, SeqCutOffPolicy
+from pyocf.types.volume import RamVolume
+from pyocf.types.volume_core import CoreVolume
+from pyocf.utils import Size
+
+# Highest reader count that still fits in the 8 bit access counter without
+# colliding with the OCF_CACHE_LINE_ACCESS_WR (0xFF) sentinel
+MAX_CONCURRENT_READERS = 254
+
+PATTERN = b"\xa5"
+
+
+class HoldDevice(RamVolume):
+    """RamVolume which is able to suspend completion of forwarded reads"""
+
+    def __init__(self, size, uuid=None):
+        super().__init__(size, uuid)
+        self._lock = Lock()
+        self._armed = False
+        self._held = []
+
+    def arm(self):
+        with self._lock:
+            self._armed = True
+
+    def disarm(self):
+        with self._lock:
+            self._armed = False
+
+    def held_count(self):
+        with self._lock:
+            return len(self._held)
+
+    def release(self, count=None):
+        with self._lock:
+            if count is None:
+                count = len(self._held)
+            released, self._held = self._held[:count], self._held[count:]
+
+        for io in released:
+            super().do_forward_io(*io)
+
+        return len(released)
+
+    def do_forward_io(self, token, rw, addr, nbytes, offset):
+        with self._lock:
+            if self._armed and rw == IoDir.READ:
+                self._held.append((token, rw, addr, nbytes, offset))
+                return
+
+        super().do_forward_io(token, rw, addr, nbytes, offset)
+
+
+def _submit_read(vol, queue, addr, size):
+    data = Data(size)
+    io = vol.new_io(queue, addr, size, IoDir.READ, 0, 0)
+    io.set_data(data)
+    cmpl = OcfCompletion([("err", c_int)])
+    io.callback = cmpl.callback
+    io.submit()
+
+    return cmpl, data
+
+
+def _wait_for(predicate, timeout=30):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return True
+        time.sleep(0.005)
+
+    return predicate()
+
+
+def _wait_stable(value_fn, queue, quiet=0.2, timeout=30):
+    """Wait until value_fn() stops changing and the queue goes idle"""
+    deadline = time.monotonic() + timeout
+    value = value_fn()
+    changed = time.monotonic()
+
+    while time.monotonic() < deadline:
+        time.sleep(0.005)
+        current = value_fn()
+        if current != value:
+            value, changed = current, time.monotonic()
+            continue
+        if time.monotonic() - changed < quiet:
+            continue
+        # I/O submission may still be in progress in the queue thread
+        if queue.settle():
+            continue
+        return value
+
+    return value_fn()
+
+
+def test_max_concurrent_readers(pyocf_ctx):
+    """
+    Submit more read hits on a single cache line than the 8 bit alock access
+    counter is able to track and verify that:
+      * no more than 254 readers hold the cache line at the same time,
+      * the excess readers are queued on the waiters list and are served once
+        an active reader releases the line,
+      * releasing the readers doesn't trip ENV_BUG_ON() in
+        ocf_alock_unlock_one_rd().
+    """
+    readers = MAX_CONCURRENT_READERS + 1
+
+    pyocf_ctx.register_volume_type(HoldDevice)
+
+    cache_device = HoldDevice(Size.from_MiB(50))
+    core_device = RamVolume(Size.from_MiB(50))
+
+    cache = Cache.start_on_device(
+        cache_device, cache_mode=CacheMode.WT, cache_line_size=CacheLineSize.LINE_4KiB
+    )
+    core = Core.using_device(core_device)
+    cache.add_core(core)
+    cache.set_seq_cut_off_policy(SeqCutOffPolicy.NEVER)
+
+    vol = CoreVolume(core)
+    vol.open()
+    queue = cache.get_default_queue()
+
+    cl = int(CacheLineSize.LINE_4KiB)
+
+    # Map exactly one cache line, so that every subsequent read of the first
+    # cache line worth of core data is a hit on that very same cache line
+    vol.sync_io(queue, 0, Data.from_bytes(PATTERN * cl), IoDir.WRITE)
+    queue.settle()
+
+    # From now on reads issued to the cache device are parked, so every read
+    # hit holds its cache line read lock until released
+    cache_device.arm()
+
+    try:
+        ios = [_submit_read(vol, queue, 0, cl) for _ in range(readers)]
+
+        held = _wait_stable(cache_device.held_count, queue)
+
+        assert held <= MAX_CONCURRENT_READERS, (
+            "{} readers hold a single cache line at the same time - the 8 bit alock "
+            "access counter reached 0x{:02X} (OCF_CACHE_LINE_ACCESS_WR), making the "
+            "cache line indistinguishable from write locked".format(held, held)
+        )
+        assert held == MAX_CONCURRENT_READERS, (
+            "only {} readers acquired the cache line - the access counter is not "
+            "saturated, so the test no longer exercises the bug".format(held)
+        )
+
+        assert not any(cmpl.completed() for cmpl, _ in ios), (
+            "no read may complete while the cache device holds it"
+        )
+
+        # The reader which didn't get the lock is parked on the alock waiters
+        # list. Releasing a single reader must hand the lock over to it, which
+        # then issues its own read to the cache device.
+        cache_device.release(1)
+
+        handed_over = _wait_stable(cache_device.held_count, queue)
+        assert handed_over == held, (
+            "reader waiting for the cache line was not resumed after an active "
+            "reader released it ({} readers in flight, expected {})".format(handed_over, held)
+        )
+    finally:
+        # Reads must never be left parked on the cache device - cache.stop()
+        # waits for I/O in flight, so a failed assertion would hang the fixture
+        # teardown instead of reporting the failure
+        cache_device.disarm()
+        cache_device.release()
+
+    assert _wait_for(lambda: all(cmpl.completed() for cmpl, _ in ios)), "I/O not completed"
+    queue.settle()
+
+    assert sorted({int(cmpl.results["err"]) for cmpl, _ in ios}) == [0]
+    assert sorted({data.get_bytes() for _, data in ios}) == [PATTERN * cl]
+
+    cache.stop()