feat(Metadata cache): Unified metadata cache for both list and info caching

fsspec/dircache.py

@@ -1,27 +1,37 @@
+import functools
+import threading
 import time
+from collections import OrderedDict, defaultdict
 from collections.abc import MutableMapping
-from functools import lru_cache
+
+
+def _locked(func):
+    @functools.wraps(func)
+    def wrapper(self, *args, **kwargs):
+        with self._lock:
+            return func(self, *args, **kwargs)
+
+    return wrapper
 
 
 class DirCache(MutableMapping):
     """
-    Caching of directory listings, in a structure like::
-
-        {"path0": [
-            {"name": "path0/file0",
-             "size": 123,
-             "type": "file",
-             ...
-            },
-            {"name": "path0/file1",
-            },
-            ...
-            ],
-         "path1": [...]
-        }
-
-    Parameters to this class control listing expiry or indeed turn
-    caching off
+    Thread-safe Unified Entry-Index Caching of directory listings and file metadata.
+
+    Decouples single object metadata storage (`_entries`) from directory tree
+    indexing (`_children`) and listing completeness (`_fully_cached_dirs`).
+
+    Parameters
+    ----------
+    use_listings_cache: bool
+        If False, this cache never returns items, but always reports KeyError,
+        and setting items has no effect.
+    listings_expiry_time: int or float (optional)
+        Time in seconds that a listing is considered valid. If None,
+        listings do not expire.
+    max_paths: int (optional)
+        The maximum number of path entries retained in cache; 'recent'
+        refers to when the entry was set or accessed.
     """
 
     def __init__(
@@ -31,65 +41,177 @@ def __init__(
         max_paths=None,
         **kwargs,
     ):
-        """
-
-        Parameters
-        ----------
-        use_listings_cache: bool
-            If False, this cache never returns items, but always reports KeyError,
-            and setting items has no effect
-        listings_expiry_time: int or float (optional)
-            Time in seconds that a listing is considered valid. If None,
-            listings do not expire.
-        max_paths: int (optional)
-            The number of most recent listings that are considered valid; 'recent'
-            refers to when the entry was set.
-        """
-        self._cache = {}
-        self._times = {}
-        if max_paths:
-            self._q = lru_cache(max_paths + 1)(lambda key: self._cache.pop(key, None))
         self.use_listings_cache = use_listings_cache
         self.listings_expiry_time = listings_expiry_time
         self.max_paths = max_paths
 
+        self._lock = threading.RLock()
+        self._entries = OrderedDict()
+        self._children = defaultdict(set)
+        self._fully_cached_dirs = {}
+
+    @staticmethod
+    def _parent(path: str) -> str:
+        clean = path.rstrip("/")
+        if "/" not in clean:
+            return ""
+        return clean.rsplit("/", 1)[0]
+
+    def _calc_expiry(self) -> float:
+        return (
+            time.time() + self.listings_expiry_time
+            if self.listings_expiry_time is not None
+            else float("inf")
+        )
+
+    @_locked
+    def get_info(self, path: str):
+        """O(1) thread-safe lookup for single item metadata."""
+        if not self.use_listings_cache:
+            return None
+
+        path = path.rstrip("/")
+        if path not in self._entries:
+            return None
+
+        info, expiry = self._entries[path]
+        if time.time() > expiry:
+            self._evict_entry(path)
+            return None
+
+        self._entries.move_to_end(path)
+        return info
+
+    @_locked
+    def save_info(self, path: str, info: dict, expiry: float | None = None):
+        """Thread-safe cache for single item info."""
+        if not self.use_listings_cache:
+            return
+
+        path = path.rstrip("/")
+        parent = self._parent(path)
+        expiry = expiry if expiry is not None else self._calc_expiry()
+
+        self._entries[path] = (info, expiry)
+        self._entries.move_to_end(path)
+        self._children[parent].add(path)
+        self._enforce_capacity()
+
+    @_locked
     def __getitem__(self, item):
-        if self.listings_expiry_time is not None:
-            if self._times.get(item, 0) - time.time() < -self.listings_expiry_time:
-                del self._cache[item]
-        if self.max_paths:
-            self._q(item)
-        return self._cache[item]  # maybe raises KeyError
+        if not self.use_listings_cache:
+            raise KeyError(item)
 
-    def clear(self):
-        self._cache.clear()
+        path = item.rstrip("/")
 
-    def __len__(self):
-        return len(self._cache)
+        # Check full directory listing
+        if path in self._fully_cached_dirs:
+            if time.time() > self._fully_cached_dirs[path]:
+                self._invalidate_dir(path)
+                raise KeyError(item)
 
-    def __contains__(self, item):
-        try:
-            self[item]
-            return True
-        except KeyError:
-            return False
+            res = []
+            for child in list(self._children.get(path, set())):
+                info = self.get_info(child)
+                if info is None:
+                    self._fully_cached_dirs.pop(path, None)
+                    raise KeyError(item)
+                res.append(info)
+            return sorted(res, key=lambda x: x.get("name", ""))
 
+        # Fallback to single item info lookup
+        info = self.get_info(path)
+        if info is not None:
+            return [info]
+
+        raise KeyError(item)
+
+    @_locked
     def __setitem__(self, key, value):
         if not self.use_listings_cache:
             return
-        if self.max_paths:
-            self._q(key)
-        self._cache[key] = value
-        if self.listings_expiry_time is not None:
-            self._times[key] = time.time()
 
+        dir_path = key.rstrip("/")
+        if isinstance(value, list):
+            expiry = self._calc_expiry()
+            child_paths = set()
+            for item in value:
+                child_path = item.get("name", "").rstrip("/")
+                if child_path:
+                    self.save_info(child_path, item, expiry=expiry)
+                    child_paths.add(child_path)
+
+            self._children[dir_path] = child_paths
+            self._fully_cached_dirs[dir_path] = expiry
+        elif isinstance(value, dict):
+            self.save_info(dir_path, value)
+
+    @_locked
     def __delitem__(self, key):
-        del self._cache[key]
+        path = key.rstrip("/")
+        found = False
 
-    def __iter__(self):
-        entries = list(self._cache)
+        if path in self._fully_cached_dirs:
+            self._invalidate_dir(path)
+            found = True
+
+        if path in self._entries:
+            self._evict_entry(path)
+            found = True
+
+        if not found:
+            raise KeyError(key)
+
+    @_locked
+    def _invalidate_dir(self, dir_path: str):
+        self._fully_cached_dirs.pop(dir_path, None)
+        children = self._children.pop(dir_path, set())
+        for child in children:
+            if child in self._fully_cached_dirs:
+                self._invalidate_dir(child)
+            if child in self._entries:
+                del self._entries[child]
 
-        return (k for k in entries if k in self)
+    @_locked
+    def _evict_entry(self, path: str):
+        self._entries.pop(path, None)
+        parent = self._parent(path)
+        if parent in self._children:
+            self._children[parent].discard(path)
+            if not self._children[parent]:
+                del self._children[parent]
+        self._fully_cached_dirs.pop(parent, None)
+
+    @_locked
+    def _enforce_capacity(self):
+        if self.max_paths is not None:
+            while len(self._entries) > self.max_paths:
+                oldest_path, _ = self._entries.popitem(last=False)
+                self._evict_entry(oldest_path)
+
+    @_locked
+    def clear(self):
+        self._entries.clear()
+        self._children.clear()
+        self._fully_cached_dirs.clear()
+
+    @_locked
+    def __len__(self):
+        return len(self._entries)
+
+    @_locked
+    def __contains__(self, item):
+        path = item.rstrip("/")
+        if path in self._fully_cached_dirs:
+            return True
+        return self.get_info(path) is not None
+
+    @_locked
+    def __iter__(self):
+        keys = list(self._fully_cached_dirs) + [
+            k for k in self._entries if k not in self._fully_cached_dirs
+        ]
+        return iter(keys)
 
     def __reduce__(self):
         return (

fsspec/spec.py

@@ -744,19 +744,42 @@ def info(self, path, **kwargs):
         directory, or something else) and other FS-specific keys.
         """
         path = self._strip_protocol(path)
+        if not kwargs.get("refresh", False):
+            try:
+                cached_info = self.dircache.get_info(path)
+                if cached_info is not None:
+                    return cached_info
+            except AttributeError:
+                pass
+
         out = self.ls(self._parent(path), detail=True, **kwargs)
         out = [o for o in out if o["name"].rstrip("/") == path]
         if out:
-            return out[0]
+            res = out[0]
+            try:
+                self.dircache.save_info(path, res)
+            except AttributeError:
+                pass
+            return res
         out = self.ls(path, detail=True, **kwargs)
         path = path.rstrip("/")
         out1 = [o for o in out if o["name"].rstrip("/") == path]
         if len(out1) == 1:
             if "size" not in out1[0]:
                 out1[0]["size"] = None
-            return out1[0]
+            res = out1[0]
+            try:
+                self.dircache.save_info(path, res)
+            except AttributeError:
+                pass
+            return res
         elif len(out1) > 1 or out:
-            return {"name": path, "size": 0, "type": "directory"}
+            res = {"name": path, "size": 0, "type": "directory"}
+            try:
+                self.dircache.save_info(path, res)
+            except AttributeError:
+                pass
+            return res
         else:
             raise FileNotFoundError(path)