distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Caching utilities — ByteCache, SimpleByteCache, LHMCache.
2
3Ported from net.i2p.util.{ByteCache, SimpleByteCache, LHMCache}.
4"""
5
6import threading
7from collections import OrderedDict
8from typing import Dict, Optional
9
10
11class LHMCache(OrderedDict):
12 """LRU cache backed by OrderedDict.
13
14 When the cache exceeds max_size, the oldest entry is evicted.
15 Not thread-safe — caller must synchronize.
16 """
17
18 def __init__(self, max_size: int) -> None:
19 super().__init__()
20 self._max_size = max_size
21
22 def __setitem__(self, key, value):
23 if key in self:
24 self.move_to_end(key)
25 super().__setitem__(key, value)
26 while len(self) > self._max_size:
27 self.popitem(last=False)
28
29 def __getitem__(self, key):
30 value = super().__getitem__(key)
31 self.move_to_end(key)
32 return value
33
34 def get(self, key, default=None):
35 if key in self:
36 return self[key]
37 return default
38
39
40class SimpleByteCache:
41 """Per-size byte array cache for reuse. Reduces allocation pressure.
42
43 In Python, this is less critical than Java but maintains API compatibility.
44 """
45
46 _instances: Dict[int, "SimpleByteCache"] = {}
47 _lock = threading.Lock()
48
49 def __init__(self, cache_size: int, size: int) -> None:
50 self._size = size
51 self._cache_size = cache_size
52 self._pool: list[bytes] = []
53 self._pool_lock = threading.Lock()
54
55 @classmethod
56 def get_instance(cls, size: int, cache_size: int = 32) -> "SimpleByteCache":
57 with cls._lock:
58 if size not in cls._instances:
59 cls._instances[size] = SimpleByteCache(cache_size, size)
60 return cls._instances[size]
61
62 def acquire(self) -> bytearray:
63 """Get a byte array from cache or create a new one."""
64 with self._pool_lock:
65 if self._pool:
66 return bytearray(self._pool.pop())
67 return bytearray(self._size)
68
69 def release(self, data: bytes | bytearray) -> None:
70 """Return a byte array to the cache."""
71 with self._pool_lock:
72 if len(self._pool) < self._cache_size:
73 # Zero it out
74 if isinstance(data, bytearray):
75 for i in range(len(data)):
76 data[i] = 0
77 self._pool.append(bytes(data))
78
79 @classmethod
80 def clear_all(cls) -> None:
81 with cls._lock:
82 for cache in cls._instances.values():
83 with cache._pool_lock:
84 cache._pool.clear()
85
86
87class ByteCache:
88 """Byte array cache using ByteArray wrappers.
89
90 Simplified for Python — wraps SimpleByteCache.
91 """
92
93 _instances: Dict[tuple, "ByteCache"] = {}
94 _lock = threading.Lock()
95
96 def __init__(self, cache_size: int, size: int) -> None:
97 self._inner = SimpleByteCache.get_instance(size, cache_size)
98 self._size = size
99
100 @classmethod
101 def get_instance(cls, cache_size: int, size: int) -> "ByteCache":
102 key = (cache_size, size)
103 with cls._lock:
104 if key not in cls._instances:
105 cls._instances[key] = ByteCache(cache_size, size)
106 return cls._instances[key]
107
108 def acquire(self) -> bytearray:
109 return self._inner.acquire()
110
111 def release(self, data: bytes | bytearray, should_zero: bool = True) -> None:
112 self._inner.release(data)
113
114 @classmethod
115 def clear_all(cls) -> None:
116 SimpleByteCache.clear_all()