cache: added clear_expired_cache (#1055)

fsspec/implementations/cached.py

@@ -159,7 +159,7 @@ def save_cache(self):
         if os.path.exists(fn):
             with open(fn, "rb") as f:
                 cached_files = pickle.load(f)
-            for k, c in cached_files.items():
+            for k, c in cached_files.copy().items():
                 if k in cache:
                     if c["blocks"] is True or cache[k]["blocks"] is True:
                         c["blocks"] = True
@@ -173,7 +173,8 @@ def save_cache(self):
                         c["blocks"] = blocks
                     c["time"] = max(c["time"], cache[k]["time"])
                     c["uid"] = cache[k]["uid"]
-
+                else:
+                    cached_files.pop(k)
             # Files can be added to cache after it was written once
             for k, c in cache.items():
                 if k not in cached_files:
@@ -232,6 +233,41 @@ def clear_cache(self):
         rmtree(self.storage[-1])
         self.load_cache()
 
+
+    def clear_expired_cache(self, expiry_time=None):
+        """Remove all expired files and metadata from the cache
+        Parameters
+        ----------
+        expiry_time: int
+            The time in seconds after which a local copy is considered useless.
+            If not defined the default is equivalent to the attribute from the
+            file caching instantiation.
+        In the case of multiple cache locations, this clears only the last one,
+        which is assumed to be the read/write one.
+        """
+        if not expiry_time:
+            expiry_time = self.expiry
+
+        self._check_cache()
+
+        for path, detail in self.cached_files[-1].copy().items():
+            if time.time() - detail["time"] > expiry_time:
+                if self.same_names:
+                    basename = os.path.basename(detail["original"])
+                    fn = os.path.join(self.storage[-1], basename)
+                else:
+                    fn = os.path.join(self.storage[-1], detail["fn"])
+                if os.path.exists(fn):
+                    os.remove(fn)
+                    self.cached_files[-1].pop(path)
+
+        if self.cached_files[-1]:
+            self.save_cache()
+        else:
+            rmtree(self.storage[-1])
+            self.load_cache()
+
+
     def pop_from_cache(self, path):
         """Remove cached version of given file
 
@@ -389,6 +425,7 @@ def __getattribute__(self, item):
             "_check_cache",
             "_mkcache",
             "clear_cache",
+            "clear_expired_cache",
             "pop_from_cache",
             "_mkcache",
             "local_file",

fsspec/implementations/tests/test_cached.py

@@ -153,6 +153,76 @@ def test_clear():
     assert len(os.listdir(cache1)) < 2
 
 
+def test_clear_expired():
+    import tempfile
+    import time
+
+    origin = tempfile.mkdtemp()
+    cache1 = tempfile.mkdtemp()
+    cache2 = tempfile.mkdtemp()
+    cache3 = tempfile.mkdtemp()
+
+    data = b"test data"
+    f1 = os.path.join(origin, "afile")
+    f2 = os.path.join(origin, "bfile")
+    f3 = os.path.join(origin, "cfile")
+    f4 = os.path.join(origin, "dfile")
+
+    with open(f1, "wb") as f:
+        f.write(data)
+    with open(f2, "wb") as f:
+        f.write(data)
+    with open(f3, "wb") as f:
+        f.write(data)
+    with open(f4, "wb") as f:
+        f.write(data)
+
+    # populates first cache
+    fs = fsspec.filesystem("filecache", target_protocol="file", cache_storage=cache1)
+    assert fs.cat(f1) == data
+
+    # populates last cache if file not found in first cache
+    fs = fsspec.filesystem(
+        "filecache",
+        target_protocol="file",
+        cache_storage=[cache1, cache2],
+        expiry_time=60,
+    )
+
+    assert fs.cat(f2) == data
+
+    # let f2 expire
+    time.sleep(60)
+    assert fs.cat(f3) == data
+    assert len(os.listdir(cache2)) == 3
+
+    fs.clear_expired_cache()
+    assert len(os.listdir(cache2)) == 2
+
+    # check complete cleanup
+    time.sleep(60)
+    fs.clear_expired_cache()
+    assert not fs._check_file(f2)
+    assert not fs._check_file(f3)
+    assert len(os.listdir(cache2)) < 2
+
+    # check cache1 to be untouched after cleaning
+    assert len(os.listdir(cache1)) == 2
+
+    # check cleaning with 'same_name' option enabled
+    fs = fsspec.filesystem(
+        "filecache",
+        target_protocol="file",
+        cache_storage=[cache1, cache2, cache3],
+        same_names=True,
+        expiry_time=30,
+    )
+    assert fs.cat(f4) == data
+    time.sleep(30)
+    fs.clear_expired_cache()
+    assert not fs._check_file(f4)
+
+
 def test_pop():
     import tempfile