Minor changes

- call_id for logging purposes

Author: charmoniumQ

charmonium/cache/memoize.py

@@ -156,7 +156,7 @@ def __setstate__(self, state: Mapping[str, Any]) -> Any:
         )
         self._version = 0
         self._memory_lock = threading.RLock()
-        self._index_read(0)
+        self._index_read(random.randint(0, 2**64 - 1))
 
     def __init__(
         self,
@@ -245,6 +245,24 @@ def _deleter(self, item: tuple[Any, Entry]) -> None:
                 )
             )
 
+    def refresh(self) -> None:
+        """Refresh the this cache from storage
+
+        Note: This happens automatically at function import-time if
+        fine_grain_persistence is enabled, after function call.
+
+        """
+        self._index_read(random.randint(0, 2**64 - 1))
+
+    def commit(self) -> None:
+        """Save this cache into storage.
+
+        Note: This happens automatically at function import-time if
+        fine_grain_persistence is enabled, after function call.
+
+        """
+        self._index_write(random.randint(0, 2**64 - 1))
+
     def _index_read(self, call_id: int) -> None:
         with self._memory_lock, self._index_lock.reader:
             self._index_read_nolock(call_id)
@@ -263,6 +281,7 @@ def _index_read_nolock(self, call_id: int) -> None:
                     ],
                     self._pickler.loads(self._obj_store[self._index_key]),
                 )
+                # TODO: catch the case where this is unpicklable or does not exist.
                 if other_version > current_version:
                     self._version = other_version
                     self._index.update(other_index)
@@ -286,7 +305,7 @@ def _index_read_nolock(self, call_id: int) -> None:
     def _index_write(self, call_id: int) -> None:
         with perf_ctx("index_write", call_id), self._memory_lock, self._index_lock.writer:
             self._index_read_nolock(call_id)
-            self._evict()
+            self._evict(call_id)
             self._version += 1
             self._obj_store[self._index_key] = self._pickler.dumps(
                 (
@@ -320,7 +339,17 @@ def _system_state(self) -> Any:
         with self._memory_lock:
             return (__version__,) + none_tuple(self._extra_system_state())
 
-    def _evict(self) -> None:
+    def evict(self) -> None:
+        """If the size of the cache is greater than ``self._size``, use ``self._replacement_policy``.
+
+        Note: this is automatically triggered automatically either
+        when the function gets called or atexit, depending on the
+        group-level options.
+
+        """
+        self._evict(random.randint(0, 2**64 - 1))
+
+    def _evict(self, call_id: int) -> None:
         with self._memory_lock:
             total_size: bitmath.Bitmath = bitmath.Byte(0)
 
@@ -350,6 +379,7 @@ def _evict(self) -> None:
                                 "obj_key": obj_key,
                                 "entry.data_size": entry.data_size.bytes,
                                 "new_total_size": (total_size - entry.data_size).bytes,
+                                "call_id": call_id,
                             }
                         )
                     )
@@ -627,6 +657,7 @@ def __call__(
                     hit, value = self._try_unpickle(value_ser, call_id)
             else:
                 hit, value = True, cast(FuncReturn, entry.value)
+        # TODO: allow a hit if entry is None but value_ser is not.
 
         if ops_logger.isEnabledFor(logging.DEBUG):
             ops_logger.debug(
@@ -671,7 +702,7 @@ def __call__(
 
             # Update time_cost
             if self.group._fine_grain_eviction:
-                self.group._evict()
+                self.group._evict(call_id)
 
             if self.group._fine_grain_persistence:
                 self.group._index_write(call_id)
@@ -766,6 +797,17 @@ def would_hit(self, *args: FuncParams.args, **kwargs: FuncParams.kwargs) -> bool
             )
         return would_hit
 
+    def clear_entry(
+            self,
+            *args: FuncParams.args,
+            **kwargs: FuncParams.kwargs,
+    ) -> None:
+        call_id = random.randint(0, 2**64 - 1)
+        key, entry, obj_key, value_ser = self._would_hit(call_id, *args, **kwargs)
+        if entry is not None:
+            self.group._deleter((key, entry))
+        assert not self.would_hit(*args, **kwargs)
+
     def _would_hit(
         self, call_id: int, *args: FuncParams.args, **kwargs: FuncParams.kwargs
     ) -> Tuple[Tuple[Any, ...], Optional[Entry], int, Optional[bytes]]:
@@ -789,12 +831,14 @@ def _would_hit(
         with self.group._memory_lock:
             if self.group._fine_grain_persistence:
                 self.group._index_read(call_id)
-            return (
-                key,
-                self.group._index.get(key, None),
-                obj_key,
-                self.group._obj_store.get(obj_key, None) if self._use_obj_store else None,
-            )
+            entry = self.group._index.get(key, None)
+        value_ser = self.group._obj_store.get(obj_key, None) if self._use_obj_store else None
+        return (
+            key,
+            entry,
+            obj_key,
+            value_ser,
+        )
 
     def __get__(self, instance: Any, instancetype: Type[Any]) -> BoundMemoized[FuncParams, FuncReturn]:
         """Implement the descriptor protocol to make decorating instance

charmonium/cache/obj_store.py

@@ -93,22 +93,23 @@ def __setitem__(self, key: int, val: bytes) -> None:
 
     def __getitem__(self, key: int) -> bytes:
         path = self.path / self._int2str(key)
-        if not path.exists():
-            raise KeyError(key)
-        else:
+        try:
             return path.read_bytes()
+        except FileNotFoundError:
+            raise KeyError(key)
 
     def __delitem__(self, key: int) -> None:
-        if key in self:
-            # print(f"delitem {key}")
+        try:
             (self.path / self._int2str(key)).unlink()
+        except FileNotFoundError:
+            pass
 
     def get(self, key: int, default: _T) -> Union[bytes | _T]:
         path = self.path / self._int2str(key)
-        if not path.exists():
-            return default
-        else:
+        try:
             return path.read_bytes()
+        except FileNotFoundError:
+            return default
 
     def __contains__(self, key: int) -> bool:
         return (self.path / self._int2str(key)).exists()

charmonium/cache/replacement_policies.py

@@ -75,14 +75,14 @@ def access(self, key: Any, entry: Entry) -> None:
     def invalidate(
         self, key: Any, entry: Any
     ) -> None:  # pylint: disable=unused-argument
-        del self._data[key]
+        self._data.pop(key, None)
 
     def evict(self) -> tuple[Any, Entry]:
         if self._data:
             self.inflation, key, entry = min(
                 (score, key, entry) for key, (score, entry) in self._data.items()
             )
-            del self._data[key]
+            self._data.pop(key, None)
             return key, entry
         else:
             raise ValueError("No data left to evict")

tests/test_memoize_parallel.py

@@ -91,7 +91,6 @@ def square_all(lst: list[int]) -> list[int]:
 N_OVERLAP = 4
 
 
-@pytest.mark.xfail(reason="something with filesystem consistency")
 @pytest.mark.parametrize("ParallelType", [multiprocessing.Process, threading.Thread])
 def test_parallelism(ParallelType: Type[Parallel]) -> None:
     if tmp_root.exists():
@@ -154,7 +153,6 @@ def test_dask_delayed() -> None:
     assert results == tuple(x**2 for call in calls for x in call)
 
 
-@pytest.mark.xfail(reason="something with filesystem consistency")
 def test_dask_bag() -> None:
     if tmp_root.exists():
         shutil.rmtree(tmp_root)