Merge pull request #243 from martindurant/caches

instance- and dir-caches

Author: martindurant

docs/source/api.rst

@@ -32,8 +32,10 @@ Base Classes
    fsspec.core.OpenFile
    fsspec.core.BaseCache
    fsspec.core.get_fs_token_paths
+   fsspec.dircache.DirCache
 
 .. autoclass:: fsspec.spec.AbstractFileSystem
+   :members:
 
 .. autoclass:: fsspec.spec.Transaction
    :members:
@@ -52,6 +54,9 @@ Base Classes
 
 .. autofunction:: fsspec.core.get_fs_token_paths
 
+.. autoclass:: fsspec.dircache.DirCache
+   :members: __init__
+
 .. _implementations:
 
 Built-in Implementations
@@ -84,7 +89,7 @@ Built-in Implementations
    :members: __init__
 
 .. autoclass:: fsspec.implementations.local.LocalFileSystem
-   :members:
+   :members: __init__
 
 .. autoclass:: fsspec.implementations.memory.MemoryFileSystem
    :members: __init__
@@ -105,6 +110,7 @@ Built-in Implementations
    :members: __init__
 
 .. autoclass:: fsspec.implementations.cached.WholeFileCacheFileSystem
+   :members: __init__
 
 Other Known Implementations
 ---------------------------

docs/source/features.rst

@@ -166,6 +166,28 @@ the instance cache may cause excessive memory usage in some situations; but norm
 will get ``close``d, and the data discarded. Only when there is also an unfinalised transaction or
 captured traceback might this be anticipated becoming a problem.
 
+To disable instance caching, i.e., get a fresh instance which is not in the cache
+even for a cachable class, pass ``skip_instance_cache=True``.
+
+Listings Caching
+----------------
+
+For some implementations, getting file listigns (i.e., ``ls`` and anything that
+depends on it) is expensive. These implementations use dict-like instances of
+:class:`fsspec.dircache.DirCache` to manage the listings.
+
+The cache allows for time-based expiry of entries with the ``listings_expiry_time``
+parameter, or or LRU expiry with the ``max_paths`` parameter. These can be
+set on any implementation instance that uses listings caching; or to skip the
+caching altogether, use ``use_listings_cache=False``. That would be appropriate
+when the target location is known to be volatile because it is being written
+to from other sources.
+
+When the ``fsspec`` instance writes to the backend, the method ``invalidate_cache``
+is called, so that subsequent listing of the given paths will force a refresh. In
+addition, some methods like ``ls`` have a ``refresh`` parameter to force fetching
+the listing again.
+
 File Buffering
 --------------
 

fsspec/dircache.py

@@ -0,0 +1,96 @@
+from functools import lru_cache
+import time
+from collections.abc import MutableMapping
+
+
+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
+    """
+
+    def __init__(
+        self,
+        use_listings_cache=True,
+        listings_expiry_time=None,
+        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 (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
+
+    def __getitem__(self, item):
+        if self.listings_expiry_time:
+            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
+
+    def clear(self):
+        self._cache.clear()
+
+    def __len__(self):
+        return len(self._cache)
+
+    def __contains__(self, item):
+        try:
+            self[item]
+            return True
+        except KeyError:
+            return False
+
+    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:
+            self._times[key] = time.time()
+
+    def __delitem__(self, key):
+        del self._cache[key]
+
+    def __iter__(self):
+        return (k for k in self._cache if k in self)
+
+    def __reduce__(self):
+        return (
+            DirCache,
+            (self.use_listings_cache, self.listings_expiry_time, self.max_paths),
+        )

fsspec/implementations/ftp.py

@@ -9,6 +9,7 @@ class FTPFileSystem(AbstractFileSystem):
 
     root_marker = "/"
     cachable = False
+    protocol = "ftp"
 
     def __init__(
         self,
@@ -74,12 +75,6 @@ def _get_kwargs_from_urls(urlpath):
         out.pop("protocol", None)
         return out
 
-    def invalidate_cache(self, path=None):
-        if path is not None:
-            self.dircache.pop(path, None)
-        else:
-            self.dircache.clear()
-
     def ls(self, path, detail=True):
         path = self._strip_protocol(path)
         out = []

fsspec/implementations/local.py

@@ -21,6 +21,7 @@ class LocalFileSystem(AbstractFileSystem):
     """
 
     root_marker = "/"
+    protocol = "file"
 
     def __init__(self, auto_mkdir=None, **kwargs):
         super().__init__(**kwargs)

fsspec/implementations/sftp.py

@@ -10,6 +10,10 @@ class SFTPFileSystem(AbstractFileSystem):
     """Files over SFTP/SSH
 
     Peer-to-peer filesystem over SSH using paramiko.
+
+    Note: if using this with the ``open`` or ``open_files``, with full URLs,
+    there is no way to tell if a path is relative, so all paths are assumed
+    to be absolute.
     """
 
     protocol = "sftp", "ssh"

fsspec/spec.py

@@ -6,6 +6,7 @@
 
 from .transaction import Transaction
 from .utils import read_block, tokenize, stringify_path
+from .dircache import DirCache
 
 logger = logging.getLogger("fsspec")
 
@@ -42,11 +43,12 @@ def __init__(cls, *args, **kwargs):
         cls._cache = {}
 
     def __call__(cls, *args, **kwargs):
+        skip = kwargs.get("skip_instance_cache", False)
         extra_tokens = tuple(
             getattr(cls, attr, None) for attr in cls._extra_tokenize_attributes
         )
         token = tokenize(cls, *args, *extra_tokens, **kwargs)
-        if cls.cachable and token in cls._cache:
+        if not skip and cls.cachable and token in cls._cache:
             return cls._cache[token]
         else:
             obj = super().__call__(*args, **kwargs)
@@ -55,7 +57,7 @@ def __call__(cls, *args, **kwargs):
             obj.storage_args = args
             obj.storage_options = kwargs
 
-            if cls.cachable:
+            if cls.cachable and not skip:
                 cls._cache[token] = obj
             return obj
 
@@ -97,17 +99,24 @@ def __init__(self, *args, **storage_options):
 
         Subclasses should call this method.
 
-        Magic kwargs that affect functionality here:
-        add_docs: if True, will append docstrings from this spec to the
-            specific implementation
+        Parameters
+        ----------
+        use_listings_cache, listings_expiry_time, max_paths:
+            passed to ``DirCache``, if the implementation supports
+            directory listing caching. Pass use_listings_cache=False
+            to disable such caching.
+        skip_instance_cache: bool
+            If this is a cachable implementation, pass True here to force
+            creating a new instance even if a matching instance exists, and prevent
+            storing this instance.
         """
         if self._cached:
             # reusing instance, don't change
             return
         self._cached = True
         self._intrans = False
         self._transaction = None
-        self.dircache = {}
+        self.dircache = DirCache(**storage_options)
 
         if storage_options.pop("add_docs", None):
             warnings.warn("add_docs is no longer supported.", FutureWarning)
@@ -290,14 +299,18 @@ def _ls_from_cache(self, path):
         but contains nothing), None if not in cache.
         """
         parent = self._parent(path)
-        if path in self.dircache:
+        try:
             return self.dircache[path]
-        elif parent in self.dircache:
+        except KeyError:
+            pass
+        try:
             files = [f for f in self.dircache[parent] if f["name"] == path]
             if len(files) == 0:
                 # parent dir was listed but did not contain this file
                 raise FileNotFoundError(path)
             return files
+        except KeyError:
+            pass
 
     def walk(self, path, maxdepth=None, **kwargs):
         """ Return all files belows path

fsspec/tests/test_utils.py

@@ -4,7 +4,7 @@
 from fsspec.utils import infer_storage_options, seek_delimiter, read_block
 
 
-WIN = "win" in sys.platform
+WIN = sys.platform.startswith("win")
 
 
 def test_read_block():