Merge remote-tracking branch 'upstream/master' into remove-deprecated-filesystem

Author: jorisvandenbossche

fsspec/asyn.py

@@ -6,7 +6,7 @@
 import sys
 import threading
 
-from .utils import other_paths
+from .utils import other_paths, is_exception
 from .spec import AbstractFileSystem
 
 # this global variable holds whether this thread is running async or not
@@ -16,7 +16,8 @@
 
 def _run_until_done(coro):
     """execute coroutine, when already in the event loop"""
-    loop = asyncio.get_running_loop()
+    loop = asyncio.get_event_loop()
+    assert loop.is_running()
     task = asyncio.current_task()
     asyncio.tasks._unregister_task(task)
     current_task = asyncio.tasks._current_tasks.get(loop)
@@ -72,7 +73,7 @@ async def f():
 def maybe_sync(func, self, *args, **kwargs):
     """Make function call into coroutine or maybe run
 
-    If we are running async, returns the coroutine object so that it can be awaited;
+    If we are running async, run coroutine on current loop until done;
     otherwise runs it on the loop (if is a coroutine already) or directly. Will guess
     we are running async if either "self" has an attribute asynchronous which is True,
     or thread_state does (this gets set in ``sync()`` itself, to avoid nesting loops).
@@ -105,7 +106,7 @@ async def _run_as_coroutine(func, *args, **kwargs):
 
 
 def sync_wrapper(func, obj=None):
-    """Given a function, make so can be called in async or bocking contexts
+    """Given a function, make so can be called in async or blocking contexts
 
     Leave obj=None if defining within a class. Pass the instance if attaching
     as an attribute of the instance.
@@ -189,7 +190,7 @@ async def _rm(self, path, recursive=False, **kwargs):
 
     def rm(self, path, recursive=False, **kwargs):
         path = self.expand_path(path, recursive=recursive)
-        sync(self.loop, self._rm, path, **kwargs)
+        maybe_sync(self._rm, self, path, **kwargs)
 
     async def _copy(self, paths, path2, **kwargs):
         await asyncio.gather(
@@ -199,7 +200,7 @@ async def _copy(self, paths, path2, **kwargs):
     def copy(self, path1, path2, recursive=False, **kwargs):
         paths = self.expand_path(path1, recursive=recursive)
         path2 = other_paths(paths, path2)
-        sync(self.loop, self._copy, paths, path2, **kwargs)
+        maybe_sync(self._copy, self, paths, path2)
 
     async def _pipe(self, path, value=None, **kwargs):
         if isinstance(path, str):
@@ -213,18 +214,27 @@ async def _cat(self, paths, **kwargs):
             *[
                 asyncio.ensure_future(self._cat_file(path, **kwargs), loop=self.loop)
                 for path in paths
-            ]
+            ],
+            return_exceptions=True
         )
 
-    def cat(self, path, recursive=False, **kwargs):
+    def cat(self, path, recursive=False, on_error="raise", **kwargs):
         paths = self.expand_path(path, recursive=recursive)
-        out = sync(self.loop, self._cat, paths, **kwargs)
+        out = maybe_sync(self._cat, self, paths, **kwargs)
+        if on_error == "raise":
+            ex = next(filter(is_exception, out), False)
+            if ex:
+                raise ex
         if (
             len(paths) > 1
             or isinstance(path, list)
             or paths[0] != self._strip_protocol(path)
         ):
-            return {k: v for k, v in zip(paths, out)}
+            return {
+                k: v
+                for k, v in zip(paths, out)
+                if on_error != "omit" or not is_exception(v)
+            }
         else:
             return out[0]
 
@@ -245,9 +255,11 @@ def put(self, lpath, rpath, recursive=False, **kwargs):
         fs = LocalFileSystem()
         lpaths = fs.expand_path(lpath, recursive=recursive)
         rpaths = other_paths(lpaths, rpath)
-        sync(self.loop, self._put, lpaths, rpaths, **kwargs)
+        maybe_sync(self._put, self, lpaths, rpaths, **kwargs)
 
     async def _get(self, rpaths, lpaths, **kwargs):
+        dirs = [os.path.dirname(lp) for lp in lpaths]
+        [os.makedirs(d, exist_ok=True) for d in dirs]
         return await asyncio.gather(
             *[
                 self._get_file(rpath, lpath, **kwargs)

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/cached.py

@@ -294,7 +294,7 @@ def _open(
             # TODO: action where partial file exists in read-only cache
             logger.debug("Opening partially cached copy of %s" % path)
         else:
-            hash = hash_name(path, self.same_names)
+            hash = self.hash_name(path, self.same_names)
             fn = os.path.join(self.storage[-1], hash)
             blocks = set()
             detail = {
@@ -339,6 +339,9 @@ def _open(
         self.save_cache()
         return f
 
+    def hash_name(self, path, same_name):
+        return hash_name(path, same_name=same_name)
+
     def close_and_update(self, f, close):
         """Called when a file is closing, so store the set of blocks"""
         path = self._strip_protocol(f.path)
@@ -376,6 +379,7 @@ def __getattribute__(self, item):
             "_paths_from_path",
             "open_many",
             "commit_many",
+            "hash_name",
         ]:
             # all the methods defined in this class. Note `open` here, since
             # it calls `_open`, but is actually in superclass
@@ -442,7 +446,7 @@ def open_many(self, open_files):
         downpath = [p for p, d in zip(paths, details) if not d]
         downstore = [p for p, d in zip(store_paths, details) if not d]
         downfn0 = [
-            os.path.join(self.storage[-1], hash_name(p, self.same_names))
+            os.path.join(self.storage[-1], self.hash_name(p, self.same_names))
             for p, d in zip(paths, details)
         ]  # keep these path names for opening later
         downfn = [fn for fn, d in zip(downfn0, details) if not d]
@@ -453,7 +457,7 @@ def open_many(self, open_files):
             # update metadata - only happens when downloads are successful
             newdetail = [
                 {
-                    "fn": hash_name(path, self.same_names),
+                    "fn": self.hash_name(path, self.same_names),
                     "blocks": True,
                     "time": time.time(),
                     "uid": self.fs.ukey(path),
@@ -504,7 +508,7 @@ def _open(self, path, mode="rb", **kwargs):
                     "as a wholly cached file" % path
                 )
         else:
-            hash = hash_name(path, self.same_names)
+            hash = self.hash_name(path, self.same_names)
             fn = os.path.join(self.storage[-1], hash)
             detail = {
                 "fn": hash,
@@ -576,7 +580,7 @@ def __init__(self, **kwargs):
 
     def _check_file(self, path):
         self._check_cache()
-        sha = hash_name(path, self.same_names)
+        sha = self.hash_name(path, self.same_names)
         for storage in self.storage:
             fn = os.path.join(storage, sha)
             if os.path.exists(fn):
@@ -602,7 +606,7 @@ def _open(self, path, mode="rb", **kwargs):
         if fn:
             return open(fn, mode)
 
-        sha = hash_name(path, self.same_names)
+        sha = self.hash_name(path, self.same_names)
         fn = os.path.join(self.storage[-1], sha)
         logger.debug("Copying %s to local cache" % path)
         kwargs["mode"] = mode

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/implementations/memory.py

@@ -2,6 +2,7 @@
 
 from io import BytesIO
 from datetime import datetime
+from errno import ENOTEMPTY
 from fsspec import AbstractFileSystem
 import logging
 
@@ -102,7 +103,7 @@ def rmdir(self, path):
             if not self.ls(path):
                 self.pseudo_dirs.remove(path)
             else:
-                raise OSError("Directory %s not empty" % path)
+                raise OSError(ENOTEMPTY, "Directory not empty", path)
         else:
             raise FileNotFoundError(path)
 

fsspec/implementations/tests/test_http.py

@@ -12,6 +12,7 @@
 data = b"\n".join([b"some test data"] * 1000)
 realfile = "http://localhost:%i/index/realfile" % port
 index = b'<a href="%s">Link</a>' % realfile.encode()
+win = os.name == "nt"
 
 
 class HTTPTestHandler(BaseHTTPRequestHandler):
@@ -240,9 +241,12 @@ async def _():
         loop = asyncio.get_event_loop()
         fs = fsspec.filesystem("http", asynchronous=True, loop=loop)
 
+        # fails because client creation has not yet been awaited
+        assert isinstance(
+            (await fs._cat([server + "/index/realfile"]))[0], RuntimeError
+        )
         with pytest.raises(RuntimeError):
-            # fails because client creation has not yet been awaited
-            await fs._cat([server + "/index/realfile"])
+            fs.cat([server + "/index/realfile"])
 
         await fs.set_session()  # creates client
 
@@ -251,3 +255,30 @@ async def _():
         assert out == [data]
 
     asyncio.run(_())
+
+
+def _inner_pass(fs, q, fn):
+    # pass the s3 instance, but don't use it; in new process, the instance
+    # cache should be skipped to make a new instance
+    fs = fsspec.filesystem("http")
+    q.put(fs.cat(fn))
+
+
+@pytest.mark.skipif(
+    bool(os.environ.get("TRAVIS", "")), reason="Travis is weird in many ways"
+)
+@pytest.mark.parametrize("method", ["spawn", "forkserver", "fork"])
+def test_processes(server, method):
+    import multiprocessing as mp
+
+    if win and method != "spawn":
+        pytest.skip("Windows can only spawn")
+    ctx = mp.get_context(method)
+    fn = server + "/index/realfile"
+    fs = fsspec.filesystem("http")
+
+    q = ctx.Queue()
+    p = ctx.Process(target=_inner_pass, args=(fs, q, fn))
+    p.start()
+    assert q.get() == fs.cat(fn)
+    p.join()