From 23e62a9ff7970de697f6a1ebbf55e6fc1538dee7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Saugat=20Pachhai=20=28=E0=A4=B8=E0=A5=8C=E0=A4=97=E0=A4=BE?= =?UTF-8?q?=E0=A4=A4=29?= Date: Wed, 20 Apr 2022 23:32:16 +0545 Subject: [PATCH] fs: make localfs fsspec-compliant It is not completely fsspec compliant, but it's upto the level required by our FSSpecWrapper. Next thing to tackle would be to merging `fs.base.FileSystem` and `fs.fsspec_wrapper.FSSpecWrapper` into one filesystem that is lazily-loaded and is itself fsspec-compliant. --- dvc/fs/fsspec_wrapper.py | 42 +++++++++- dvc/fs/local.py | 177 +++++++++++++++++++++------------------ tests/func/test_add.py | 2 +- 3 files changed, 136 insertions(+), 85 deletions(-) diff --git a/dvc/fs/fsspec_wrapper.py b/dvc/fs/fsspec_wrapper.py index 09a43fbd2..2adcc4d09 100644 --- a/dvc/fs/fsspec_wrapper.py +++ b/dvc/fs/fsspec_wrapper.py @@ -4,7 +4,7 @@ from typing import IO, TYPE_CHECKING, Any, Dict, Iterator, Optional, overload from funcy import cached_property from ._callback import DEFAULT_CALLBACK -from .base import FileSystem +from .base import FileSystem, RemoteActionNotImplemented FSPath = str AnyFSPath = str @@ -66,7 +66,7 @@ class FSSpecWrapper(FileSystem): encoding: Optional[str] = None, **kwargs, ) -> "IO": # pylint: disable=arguments-differ - return self.fs.open(path, mode=mode) + return self.fs.open(path, mode=mode, encoding=encoding, **kwargs) def checksum(self, path: AnyFSPath) -> str: return self.fs.checksum(path) @@ -78,6 +78,42 @@ class FSSpecWrapper(FileSystem): def exists(self, path: AnyFSPath) -> bool: return self.fs.exists(path) + def lexists(self, path: AnyFSPath) -> bool: + return self.fs.lexists(path) + + def symlink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: + try: + return self.fs.symlink(from_info, to_info) + except AttributeError: + raise RemoteActionNotImplemented("symlink", self.scheme) + + def hardlink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: + try: + return self.fs.hardlink(from_info, to_info) + except AttributeError: + raise RemoteActionNotImplemented("hardlink", self.scheme) + + def reflink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: + try: + return self.fs.reflink(from_info, to_info) + except AttributeError: + raise RemoteActionNotImplemented("reflink", self.scheme) + + def is_symlink(self, path: AnyFSPath) -> bool: + try: + return self.fs.is_symlink(path) + except AttributeError: + return False + + def is_hardlink(self, path: AnyFSPath) -> bool: + try: + return self.fs.is_hardlink(path) + except AttributeError: + return False + + def iscopy(self, path: AnyFSPath) -> bool: + return self.is_symlink(path) or self.is_hardlink(path) + @overload def ls( self, path: AnyFSPath, detail: "Literal[True]" @@ -88,7 +124,7 @@ class FSSpecWrapper(FileSystem): def ls(self, path: AnyFSPath, detail: "Literal[False]") -> Iterator[str]: ... - def ls(self, path, detail=False): + def ls(self, path, detail=False, **kwargs): return self.fs.ls(path, detail=detail) def find(self, path, prefix=None): diff --git a/dvc/fs/local.py b/dvc/fs/local.py index 7b5e2423b..658f2f90c 100644 --- a/dvc/fs/local.py +++ b/dvc/fs/local.py @@ -1,43 +1,45 @@ import logging import os +import threading -from funcy import cached_property +from fsspec import AbstractFileSystem +from funcy import cached_property, wrap_prop from dvc.scheme import Schemes from dvc.system import System from dvc.utils import tmp_fname from dvc.utils.fs import copy_fobj_to_file, copyfile, makedirs, move, remove -from ._callback import DEFAULT_CALLBACK -from .base import FileSystem +from .fsspec_wrapper import FSSpecWrapper logger = logging.getLogger(__name__) -class LocalFileSystem(FileSystem): +# pylint:disable=abstract-method, arguments-differ +class FsspecLocalFileSystem(AbstractFileSystem): sep = os.sep - scheme = Schemes.LOCAL - PARAM_CHECKSUM = "md5" - PARAM_PATH = "path" - TRAVERSE_PREFIX_LEN = 2 - - def __init__(self, **config): + def __init__(self, *args, **kwargs): from fsspec.implementations.local import LocalFileSystem as LocalFS - super().__init__(**config) + super().__init__(*args, **kwargs) self.fs = LocalFS() - @cached_property - def path(self): - from .path import Path + def makedirs(self, path, exist_ok=False): + makedirs(path, exist_ok=exist_ok) - return Path(self.sep, os.getcwd) + def mkdir(self, path, create_parents=True, **kwargs): + if self.exists(path): + raise FileExistsError(path) + if create_parents: + self.makedirs(path, exist_ok=True) + else: + os.mkdir(path, **kwargs) - def open(self, path, mode="r", encoding=None, **kwargs): - return open(path, mode=mode, encoding=encoding) + def lexists(self, path, **kwargs): + return os.path.lexists(path) - def exists(self, path) -> bool: + def exists(self, path, **kwargs): # TODO: replace this with os.path.exists once the problem is fixed on # the fsspec https://github.com/intake/filesystem_spec/issues/742 return os.path.lexists(path) @@ -46,82 +48,89 @@ class LocalFileSystem(FileSystem): from fsspec.utils import tokenize st = os.stat(path) - return str(int(tokenize([st.st_ino, st.st_mtime, st.st_size]), 16)) + def info(self, path, **kwargs): + return self.fs.info(path) + + def ls(self, path, **kwargs): + return self.fs.ls(path, **kwargs) + def isfile(self, path) -> bool: return os.path.isfile(path) def isdir(self, path) -> bool: return os.path.isdir(path) - def iscopy(self, path): - return not (System.is_symlink(path) or System.is_hardlink(path)) - - def walk(self, top, topdown=True, **kwargs): + def walk(self, path, maxdepth=None, topdown=True, **kwargs): """Directory fs generator. See `os.walk` for the docs. Differences: - no support for symlinks """ for root, dirs, files in os.walk( - top, + path, topdown=topdown, ): yield os.path.normpath(root), dirs, files - def find(self, path, prefix=None): - for root, _, files in self.walk(path): + def find(self, path, **kwargs): + for root, _, files in self.walk(path, **kwargs): for file in files: # NOTE: os.path.join is ~5.5 times slower yield f"{root}{os.sep}{file}" - def is_empty(self, path): - if self.isfile(path) and os.path.getsize(path) == 0: - return True + @classmethod + def _parent(cls, path): + return os.path.dirname(path) - if self.isdir(path) and len(os.listdir(path)) == 0: - return True + def put_file(self, lpath, rpath, callback=None, **kwargs): + parent = self._parent(rpath) + makedirs(parent, exist_ok=True) + tmp_file = os.path.join(parent, tmp_fname()) + copyfile(lpath, tmp_file, callback=callback) + os.replace(tmp_file, rpath) - return False + def get_file(self, rpath, lpath, callback=None, **kwargs): + copyfile(rpath, lpath, callback=callback) - def remove(self, path): + def mv(self, path1, path2, **kwargs): + self.makedirs(self._parent(path2), exist_ok=True) + move(path1, path2) + + def rmdir(self, path): + os.rmdir(path) + + def rm_file(self, path): remove(path) - def makedirs(self, path, **kwargs): - makedirs(path, exist_ok=kwargs.pop("exist_ok", True)) + def rm(self, path, recursive=False, maxdepth=None): + remove(path) - def move(self, from_info, to_info): - self.makedirs(self.path.parent(to_info)) - move(from_info, to_info) - - def copy(self, from_info, to_info): - tmp_info = self.path.join(self.path.parent(to_info), tmp_fname("")) + def copy(self, path1, path2, recursive=False, on_error=None, **kwargs): + tmp_info = os.path.join(self._parent(path2), tmp_fname("")) try: - copyfile(from_info, tmp_info) - os.rename(tmp_info, to_info) + copyfile(path1, tmp_info) + os.rename(tmp_info, path2) except Exception: - self.remove(tmp_info) + self.rm_file(tmp_info) raise - def upload_fobj(self, fobj, to_info, **kwargs): - self.makedirs(self.path.parent(to_info)) - tmp_info = self.path.join(self.path.parent(to_info), tmp_fname("")) - try: - copy_fobj_to_file(fobj, tmp_info) - os.rename(tmp_info, to_info) - except Exception: - self.remove(tmp_info) - raise + def open(self, path, mode="r", encoding=None, **kwargs): + return open(path, mode=mode, encoding=encoding) - def symlink(self, from_info, to_info): - System.symlink(from_info, to_info) + def symlink(self, path1, path2): + return System.symlink(path1, path2) @staticmethod def is_symlink(path): return System.is_symlink(path) - def hardlink(self, from_info, to_info): + @staticmethod + def is_hardlink(path): + return System.is_hardlink(path) + + def hardlink(self, path1, path2): # If there are a lot of empty files (which happens a lot in datasets), # and the cache type is `hardlink`, we might reach link limits and # will get something like: `too many links error` @@ -135,40 +144,46 @@ class LocalFileSystem(FileSystem): # * Windows with NTFS has a limit of 1024 hard links on a file # # That's why we simply create an empty file rather than a link. - if self.getsize(from_info) == 0: - self.open(to_info, "w").close() + if self.size(path1) == 0: + self.open(path2, "w").close() - logger.debug("Created empty file: %s -> %s", from_info, to_info) + logger.debug("Created empty file: %s -> %s", path1, path2) return - System.hardlink(from_info, to_info) + return System.hardlink(path1, path2) - @staticmethod - def is_hardlink(path): - return System.is_hardlink(path) + def reflink(self, path1, path2): + return System.reflink(path1, path2) - def reflink(self, from_info, to_info): - System.reflink(from_info, to_info) - def info(self, path): - return self.fs.info(path) +class LocalFileSystem(FSSpecWrapper): + sep = os.sep - def ls(self, path, **kwargs): - return self.fs.ls(path, **kwargs) + scheme = Schemes.LOCAL + PARAM_CHECKSUM = "md5" + PARAM_PATH = "path" + TRAVERSE_PREFIX_LEN = 2 - def put_file( - self, from_file, to_info, callback=DEFAULT_CALLBACK, **kwargs - ): - parent = self.path.parent(to_info) - makedirs(parent, exist_ok=True) - tmp_file = self.path.join(parent, tmp_fname()) - copyfile(from_file, tmp_file, callback=callback) - os.replace(tmp_file, to_info) + @wrap_prop(threading.Lock()) + @cached_property + def fs(self): + return FsspecLocalFileSystem(**self.config) - def get_file( - self, from_info, to_file, callback=DEFAULT_CALLBACK, **kwargs - ): - copyfile(from_info, to_file, callback=callback) + @cached_property + def path(self): + from .path import Path + + return Path(self.sep, os.getcwd) + + def upload_fobj(self, fobj, to_info, **kwargs): + self.makedirs(self.path.parent(to_info)) + tmp_info = self.path.join(self.path.parent(to_info), tmp_fname("")) + try: + copy_fobj_to_file(fobj, tmp_info) + os.rename(tmp_info, to_info) + except Exception: + self.remove(tmp_info) + raise localfs = LocalFileSystem() diff --git a/tests/func/test_add.py b/tests/func/test_add.py index 9bc660dd9..610710b48 100644 --- a/tests/func/test_add.py +++ b/tests/func/test_add.py @@ -783,7 +783,7 @@ def test_add_optimization_for_hardlink_on_empty_files(tmp_dir, dvc, mocker): m = mocker.spy(LocalFileSystem, "is_hardlink") stages = dvc.add(["foo", "bar", "lorem", "ipsum"]) - assert m.call_count == 4 + assert m.call_count == 8 assert m.call_args != call(tmp_dir / "foo") assert m.call_args != call(tmp_dir / "bar")