Merge pull request #420 from martindurant/fix_blockcache

Fix blockcache

Author: martindurant

fsspec/caching.py

@@ -218,7 +218,7 @@ def _fetch(self, start, end):
 
         # these are cached, so safe to do multiple calls for the same start and end.
         for block_number in range(start_block_number, end_block_number + 1):
-            self._fetch_block(block_number)
+            self._fetch_block_cached(block_number)
 
         return self._read_cache(
             start,

fsspec/implementations/http.py

@@ -2,6 +2,7 @@
 
 import aiohttp
 import asyncio
+import logging
 import re
 import requests
 import weakref
@@ -14,6 +15,7 @@
 # https://stackoverflow.com/a/15926317/3821154
 ex = re.compile(r"""<a\s+(?:[^>]*?\s+)?href=(["'])(.*?)\1""")
 ex2 = re.compile(r"""(http[s]?://[-a-zA-Z0-9@:%_+.~#?&/=]+)""")
+logger = logging.getLogger("fsspec.http")
 
 
 async def get_client(**kwargs):
@@ -101,6 +103,7 @@ async def _ls(self, url, detail=True, **kwargs):
         # ignoring URL-encoded arguments
         kw = self.kwargs.copy()
         kw.update(kwargs)
+        logger.debug(url)
         async with self.session.get(url, **self.kwargs) as r:
             r.raise_for_status()
             text = await r.text()
@@ -145,6 +148,7 @@ async def _ls(self, url, detail=True, **kwargs):
     async def _cat_file(self, url, **kwargs):
         kw = self.kwargs.copy()
         kw.update(kwargs)
+        logger.debug(url)
         async with self.session.get(url, **kw) as r:
             if r.status == 404:
                 raise FileNotFoundError(url)
@@ -155,6 +159,7 @@ async def _cat_file(self, url, **kwargs):
     async def _get_file(self, rpath, lpath, chunk_size=5 * 2 ** 20, **kwargs):
         kw = self.kwargs.copy()
         kw.update(kwargs)
+        logger.debug(rpath)
         async with self.session.get(rpath, **self.kwargs) as r:
             if r.status == 404:
                 raise FileNotFoundError(rpath)
@@ -169,6 +174,7 @@ async def _exists(self, path, **kwargs):
         kw = self.kwargs.copy()
         kw.update(kwargs)
         try:
+            logger.debug(path)
             r = await self.session.get(path, **kw)
             async with r:
                 return r.status < 400
@@ -370,6 +376,7 @@ async def async_fetch_range(self, start, end):
         kwargs = self.kwargs.copy()
         headers = kwargs.pop("headers", {}).copy()
         headers["Range"] = "bytes=%i-%i" % (start, end - 1)
+        logger.debug(self.url + " : " + headers["Range"])
         r = await self.session.get(self.url, headers=headers, **kwargs)
         async with r:
             if r.status == 416:

fsspec/tests/test_caches.py

@@ -16,25 +16,21 @@ def test_block_cache_lru():
     cache = BlockCache(4, letters_fetcher, len(string.ascii_letters), maxblocks=2)
     # miss
     cache._fetch(0, 2)
-    assert cache.cache_info().hits == 0
     assert cache.cache_info().misses == 1
     assert cache.cache_info().currsize == 1
 
     # hit
     cache._fetch(0, 2)
-    assert cache.cache_info().hits == 1
     assert cache.cache_info().misses == 1
     assert cache.cache_info().currsize == 1
 
     # miss
     cache._fetch(4, 6)
-    assert cache.cache_info().hits == 1
     assert cache.cache_info().misses == 2
     assert cache.cache_info().currsize == 2
 
     # miss & evict
     cache._fetch(12, 13)
-    assert cache.cache_info().hits == 1
     assert cache.cache_info().misses == 3
     assert cache.cache_info().currsize == 2
 

fsspec/tests/test_utils.py

@@ -7,6 +7,7 @@
     read_block,
     common_prefix,
     other_paths,
+    setup_logger,
 )
 
 
@@ -284,3 +285,10 @@ def test_common_prefix(paths, out):
 )
 def test_other_paths(paths, other, is_dir, expected):
     assert other_paths(paths, other, is_dir) == expected
+
+
+def test_log():
+    import logging
+
+    logger = setup_logger("fsspec.test")
+    assert logger.level == logging.DEBUG

fsspec/utils.py

@@ -363,3 +363,17 @@ def other_paths(paths, path2, is_dir=None):
 
 def is_exception(obj):
     return isinstance(obj, BaseException)
+
+
+def setup_logger(logname, level="DEBUG"):
+    import logging
+
+    logger = logging.getLogger(logname)
+    handle = logging.StreamHandler()
+    formatter = logging.Formatter(
+        "%(asctime)s - %(name)s - %(levelname)s " "- %(message)s"
+    )
+    handle.setFormatter(formatter)
+    logger.addHandler(handle)
+    logger.setLevel(level)
+    return logger