feat(lru_queue): add LRUQueue

Author: Rizhiy

class_cache/core.py

@@ -14,13 +14,15 @@ def __init__(
         self,
         id_: IdType = None,
         backend_type: type[CacheInterface] | Callable[[IdType], CacheInterface] = DEFAULT_BACKEND_TYPE,
+        max_items=128,
     ) -> None:
         super().__init__(id_)
         self._backend = backend_type(id_)
         # TODO: Implement max_size logic
         self._data: dict[KeyType, ValueType] = {}
         self._to_write = set()
         self._to_delete = set()
+        self._max_items = max_items
 
     @property
     def backend(self) -> CacheInterface:

class_cache/lru_queue.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Generic, Iterator
+
+from .types import KeyType
+
+
+@dataclass(slots=True, eq=False)
+class Link(Generic[KeyType]):
+    prev: Link
+    next: Link
+    key: KeyType
+
+
+# Adapted from https://github.com/python/cpython/blob/5592399313c963c110280a7c98de974889e1d353/Lib/functools.py#L542
+class LRUQueue(Generic[KeyType]):
+    def __init__(self):
+        self._links = {}
+        self._root = Link(None, None, None)  # type: ignore
+        self._root.prev = self._root
+        self._root.next = self._root
+
+    def __contains__(self, key: KeyType) -> bool:
+        if result := key in self._links:
+            self.update(key)
+        return result
+
+    def _move_to_front(self, link: Link) -> None:
+        next_ = self._root.next
+
+        next_.prev = link
+        link.next = next_
+
+        self._root.next = link
+        link.prev = self._root
+
+    @staticmethod
+    def _connect_neighbours(link: Link) -> None:
+        link.prev.next = link.next
+        link.next.prev = link.prev
+
+    def update(self, key: KeyType) -> None:
+        if key in self._links:
+            link = self._links[key]
+            # Connect neighbouring keys
+            self._connect_neighbours(link)
+        else:
+            link = Link(None, None, key)  # type: ignore
+            self._links[key] = link
+        self._move_to_front(link)
+
+    def peek(self) -> KeyType:
+        if self._root.prev == self._root:
+            raise IndexError("peeking into an empty queue")
+
+        return self._root.prev.key
+
+    def pop(self) -> KeyType:
+        if self._root.prev == self._root:
+            raise IndexError("pop from an empty queue")
+
+        last = self._root.prev
+        self._connect_neighbours(last)
+        key = last.key
+        del self._links[key]
+        return key
+
+    def __delitem__(self, key: KeyType) -> None:
+        link = self._links[key]
+        del self._links[key]
+        self._connect_neighbours(link)
+
+    def __len__(self) -> int:
+        return len(self._links)
+
+    def __iter__(self) -> Iterator[KeyType]:
+        current = self._root
+        while current.next != self._root:
+            yield current.next.key
+            current = current.next
+
+    def __str__(self) -> str:
+        result = ""
+        for key in self:
+            result += f"{key} -> "
+        return result[:-4]
+
+
+__all__ = ["LRUQueue"]

pyproject.toml

@@ -15,7 +15,7 @@ dependencies = ["Pympler", "brotli", "marisa-trie", "replete>=2.3.1"]
 Home = "https://github.com/Rizhiy/class-cache"
 
 [project.optional-dependencies]
-test = ["pytest", "pytest-cov", "replete[testing]"]
+test = ["flaky", "pytest", "pytest-cov", "replete[testing]"]
 dev = ["black", "class-cache[test]", "numpy", "pre-commit", "ruff"]
 
 [tool.flit.sdist]
@@ -27,7 +27,7 @@ version_variables = ["class_cache/__init__.py:__version__"]
 
 [tool.pytest.ini_options]
 minversion = "6.0"
-addopts = "--doctest-modules --ignore=benchmark"
+addopts = "--doctest-modules --ignore=benchmark --no-flaky-report"
 
 [tool.black]
 line-length = 120

tests/test_lru_queue.py

@@ -0,0 +1,76 @@
+from flaky import flaky
+from replete import Timer
+
+from class_cache.lru_queue import LRUQueue
+
+NUM_ITERS = 10_000
+
+
+def get_queue(size=4) -> LRUQueue:
+    queue = LRUQueue()
+    for i in range(size):
+        queue.update(i)
+    return queue
+
+
+def test_basic_queue():
+    queue = get_queue()
+
+    assert len(queue) == 4
+    for i in range(4):
+        assert i in queue
+    assert queue.peek() == 0
+    queue.update(0)
+    assert queue.peek() == 1
+    assert queue.pop() == 1
+    assert queue.peek() == 2
+    del queue[2]
+    assert len(queue) == 2
+    assert queue.peek() == 3
+    assert list(queue) == [0, 3]
+
+
+def test_next():
+    queue = get_queue()
+    assert next(queue) == 3  # type: ignore FP
+
+
+def test_str():
+    queue = get_queue()
+    queue.update(0)
+
+    assert str(queue) == "0 -> 3 -> 2 -> 1"
+
+
+@flaky(max_runs=3, min_passes=1)  # This is very noisy
+def test_contains_speed():
+    small_queue = get_queue()
+    with Timer(process_only=True) as base_timer:
+        for _ in range(NUM_ITERS):
+            assert 0 in small_queue
+
+    large_queue = get_queue(1024)
+    with Timer(base_timer.time, process_only=True) as timer:
+        for _ in range(NUM_ITERS):
+            assert 0 in large_queue
+
+    # Some noise is allowed
+    assert timer.base_time_ratio < 1.05
+
+
+@flaky(max_runs=3, min_passes=1)  # This is very noisy
+def test_pop_update_speed():
+    small_queue = get_queue()
+    with Timer(process_only=True) as base_timer:
+        for _ in range(NUM_ITERS):
+            key = small_queue.pop()
+            small_queue.update(key)
+
+    large_queue = get_queue(1024)
+    with Timer(base_timer.time, process_only=True) as timer:
+        for _ in range(NUM_ITERS):
+            key = large_queue.pop()
+            large_queue.update(key)
+
+    # Some noise is allowed
+    assert timer.base_time_ratio < 1.05