From 40c61b3aaa5a29fa69842fa3a8fe2e73734a4cae Mon Sep 17 00:00:00 2001 From: "Mr. Outis" Date: Wed, 28 Aug 2019 22:56:42 -0500 Subject: [PATCH] remote: abstract link logic to base --- dvc/remote/base.py | 51 +++----- dvc/remote/hdfs.py | 2 - dvc/remote/local/__init__.py | 108 ++++------------ dvc/remote/{local => }/slow_link_detection.py | 10 +- dvc/remote/ssh/__init__.py | 118 +++--------------- dvc/remote/ssh/connection.py | 6 +- dvc/repo/__init__.py | 2 + dvc/utils/http.py | 2 - tests/unit/remote/test_slow_link_detection.py | 15 +-- 9 files changed, 67 insertions(+), 247 deletions(-) rename dvc/remote/{local => }/slow_link_detection.py (85%) diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 2eaa3a4c7..72b174c3d 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -12,6 +12,8 @@ from operator import itemgetter from multiprocessing import cpu_count from functools import partial from concurrent.futures import ThreadPoolExecutor +from copy import copy +from dvc.remote.slow_link_detection import slow_link_guard import dvc.prompt as prompt from dvc.config import Config @@ -364,44 +366,10 @@ class RemoteBASE(object): self._link(from_info, to_info, self.cache_types) def _link(self, from_info, to_info, link_types): - # XXX: This is a fallback until the following methods are - # implemented on all remotes: - # * `isfile` - # * `makedirs` - # * `getsize` - # - # Another thing we can do is to delegate the edge case handling - # to hardlink, ommit the `isfile` check, and let the link - # implementation to create a directory before copying - if link_types == ["copy"]: - return self.copy(from_info, to_info) - assert self.isfile(from_info) self.makedirs(to_info.parent) - # 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` - # - # This is because all those empty files will have the same checksum - # (i.e. 68b329da9893e34099c7d8ad5cb9c940), therfore, they will be - # linked to the same file in the cache. - # - # From https://en.wikipedia.org/wiki/Hard_link - # * ext4 limits the number of hard links on a file to 65,000 - # * 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() - - logger.debug( - "Created empty file: {src} -> {dest}" - .format(src=str(from_path), dest=str(to_path)) - ) - return - self._try_links(from_info, to_info, link_types) @slow_link_guard @@ -422,6 +390,7 @@ class RemoteBASE(object): raise DvcException("no possible cache types left to try out.") def _do_link(self, from_info, to_info, link_method): + # XXX: We are testing if file exists rather than if file is a link if self.exists(to_info): raise DvcException("Link '{}' already exists!".format(to_info)) else: @@ -475,9 +444,15 @@ class RemoteBASE(object): return False def isfile(self, path_info): - raise NotImplementedError + """Optional: Overwrite only if the remote has a way to distinguish + between a directory and a file. + """ + return True def isdir(self, path_info): + """Optional: Overwrite only if the remote has a way to distinguish + between a directory and a file. + """ return False def walk(self, path_info): @@ -775,7 +750,10 @@ class RemoteBASE(object): progress_callback(str(path_info)) def makedirs(self, path_info): - raise NotImplementedError + """Optional: Implement only if the remote needs to create + directories before copying/linking/moving data + """ + pass def _checkout_dir( self, path_info, checksum, force, progress_callback=None @@ -799,6 +777,7 @@ class RemoteBASE(object): if self.changed(entry_info, entry_checksum_info): if self.exists(entry_info): self.safe_remove(entry_info, force=force) + self.makedirs(entry_info.parent) self.link(entry_cache_info, entry_info) self.state.save(entry_info, entry_checksum) if progress_callback: diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index efdbfbba4..ec37ace63 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -143,8 +143,6 @@ class RemoteHDFS(RemoteBASE): @contextmanager def open(self, path_info, mode="r", encoding=None): - assert mode in {"r", "rt", "rb"} - try: with self.hdfs(path_info) as hdfs, closing( hdfs.open(path_info.path, mode="rb") diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py index 5c60d650b..ebec6b85e 100644 --- a/dvc/remote/local/__init__.py +++ b/dvc/remote/local/__init__.py @@ -1,9 +1,6 @@ from __future__ import unicode_literals -from copy import copy - from dvc.scheme import Schemes -from dvc.remote.local.slow_link_detection import slow_link_guard from dvc.utils.compat import str, fspath_py35, open import os @@ -118,87 +115,6 @@ class RemoteLOCAL(RemoteBASE): def makedirs(self, path_info): makedirs(path_info, exist_ok=True, mode=self._dir_mode) - def link(self, from_info, to_info): - self._link(from_info, to_info, self.cache_types) - - def _link(self, from_info, to_info, link_types): - assert self.isfile(from_info) - - self.makedirs(to_info.parent) - - # 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` - # - # This is because all those empty files will have the same checksum - # (i.e. 68b329da9893e34099c7d8ad5cb9c940), therfore, they will be - # linked to the same file in the cache. - # - # From https://en.wikipedia.org/wiki/Hard_link - # * ext4 limits the number of hard links on a file to 65,000 - # * 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() - - logger.debug( - "Created empty file: {src} -> {dest}" - .format(src=str(from_path), dest=str(to_path)) - ) - return - - self._try_links(from_info, to_info, link_types) - - def _get_link_method(self, link_type): - CACHE_TYPE_MAP = { - "copy": self.copy, - "symlink": self.symlink, - "hardlink": self.hardlink, - "reflink": self.reflink, - } - - try: - return CACHE_TYPE_MAP[link_type] - except KeyError: - raise DvcException( - "Cache type: '{}' not supported!".format(link_type) - ) - - def _do_link(self, from_info, to_info, link_method): - if self.exists(to_info): - raise DvcException("Link '{}' already exists!".format(to_info)) - else: - link_method(from_info, to_info) - - if self.protected: - self.protect(to_info) - - msg = "Created {}'{}': {} -> {}".format( - "protected " if self.protected else "", - self.cache_types[0], - from_info, - to_info, - ) - logger.debug(msg) - - @slow_link_guard - def _try_links(self, from_info, to_info, link_types): - i = len(link_types) - while i > 0: - link_method = self._get_link_method(link_types[0]) - try: - self._do_link(from_info, to_info, link_method) - return - - except DvcException as exc: - msg = "Cache type '{}' is not supported: {}" - logger.debug(msg.format(link_types[0], str(exc))) - del link_types[0] - i -= 1 - - raise DvcException("no possible cache types left to try out.") - def already_cached(self, path_info): assert path_info.scheme in ["", "local"] @@ -262,6 +178,29 @@ class RemoteLOCAL(RemoteBASE): System.symlink(from_info.fspath, to_info.fspath) def hardlink(self, from_info, to_info): + # 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` + # + # This is because all those empty files will have the same checksum + # (i.e. 68b329da9893e34099c7d8ad5cb9c940), therfore, they will be + # linked to the same file in the cache. + # + # From https://en.wikipedia.org/wiki/Hard_link + # * ext4 limits the number of hard links on a file to 65,000 + # * 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() + + logger.debug( + "Created empty file: {src} -> {dest}".format( + src=str(from_info), dest=str(to_info) + ) + ) + return + System.hardlink(from_info.fspath, to_info.fspath) def reflink(self, from_info, to_info): @@ -296,7 +235,6 @@ class RemoteLOCAL(RemoteBASE): ) def open(self, path_info, mode="r", encoding=None): - assert mode in {"r", "rt", "rb"} return open(fspath_py35(path_info), mode=mode, encoding=encoding) def _group(self, checksum_infos, show_checksums=False): diff --git a/dvc/remote/local/slow_link_detection.py b/dvc/remote/slow_link_detection.py similarity index 85% rename from dvc/remote/local/slow_link_detection.py rename to dvc/remote/slow_link_detection.py index 7bd3bb143..2e51413ba 100644 --- a/dvc/remote/local/slow_link_detection.py +++ b/dvc/remote/slow_link_detection.py @@ -46,17 +46,15 @@ class SlowLinkDetectorDecorator(object): def slow_link_guard(method): - def call(remote_local, *args, **kwargs): - cache_config = remote_local.repo.config.config.get( - Config.SECTION_CACHE, {} - ) + def call(remote, *args, **kwargs): + cache_config = remote.repo.config.config.get(Config.SECTION_CACHE, {}) should_warn = cache_config.get( Config.SECTION_CACHE_SLOW_LINK_WARNING, True ) and not cache_config.get(Config.SECTION_CACHE_TYPE, None) if should_warn: decorated = SlowLinkDetectorDecorator(method) - return decorated(remote_local, *args, **kwargs) - return method(remote_local, *args, **kwargs) + return decorated(remote, *args, **kwargs) + return method(remote, *args, **kwargs) return call diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py index e03154fd3..12d95e381 100644 --- a/dvc/remote/ssh/__init__.py +++ b/dvc/remote/ssh/__init__.py @@ -4,13 +4,11 @@ import errno import itertools import io import os -import posixpath import getpass import logging import threading from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager, closing -from copy import copy try: import paramiko @@ -22,11 +20,9 @@ from dvc.config import Config from dvc.utils import to_chunks from dvc.utils.compat import urlparse, StringIO from dvc.remote.base import RemoteBASE -from dvc.remote.local.slow_link_detection import slow_link_guard from dvc.scheme import Schemes from dvc.remote.pool import get_connection from dvc.progress import Tqdm -from dvc.exceptions import DvcException from .connection import SSHConnection @@ -51,19 +47,10 @@ class RemoteSSH(RemoteBASE): # We use conservative setting of 4 instead to not exhaust max sessions. CHECKSUM_JOBS = 4 - DEFAULT_CACHE_TYPES = ["reflink, copy"] + DEFAULT_CACHE_TYPES = ["copy"] def __init__(self, repo, config): super(RemoteSSH, self).__init__(repo, config) - - types = config.get(Config.SECTION_CACHE_TYPE, None) - if types: - if isinstance(types, str): - types = [t.strip() for t in types.split(",")] - self.cache_types = types - else: - self.cache_types = copy(self.DEFAULT_CACHE_TYPES) - url = config.get(Config.SECTION_REMOTE_URL) if url: parsed = urlparse(url) @@ -184,28 +171,35 @@ class RemoteSSH(RemoteBASE): with self.ssh(path_info) as ssh: return ssh.getsize(path_info.path) - def open(self, path_info, mode="r", encoding=None): - with self.ssh(path_info) as ssh: - return ssh.open(path_info.path, mode=mode, encoding=encoding) - def copy(self, from_info, to_info): - if from_info.scheme != self.scheme or to_info.scheme != self.scheme: + if not from_info.scheme == to_info.scheme == self.scheme: raise NotImplementedError with self.ssh(from_info) as ssh: - ssh.cp(from_info.path, to_info.path) + ssh.copy(from_info.path, to_info.path) def symlink(self, from_info, to_info): - if from_info.scheme != self.scheme or to_info.scheme != self.scheme: + if not from_info.scheme == to_info.scheme == self.scheme: raise NotImplementedError with self.ssh(from_info) as ssh: ssh.symlink(from_info.path, to_info.path) def hardlink(self, from_info, to_info): - if from_info.scheme != self.scheme or to_info.scheme != self.scheme: + if not from_info.scheme == to_info.scheme == self.scheme: raise NotImplementedError + # See dvc/remote/local/__init__.py - hardlink() + if self.getsize(from_info) == 0: + self.open(to_info, "w").close() + + logger.debug( + "Created empty file: {src} -> {dest}".format( + src=str(from_info), dest=str(to_info) + ) + ) + return + with self.ssh(from_info) as ssh: ssh.hardlink(from_info.path, to_info.path) @@ -252,8 +246,6 @@ class RemoteSSH(RemoteBASE): @contextmanager def open(self, path_info, mode="r", encoding=None): - assert mode in {"r", "rt", "rb"} - with self.ssh(path_info) as ssh, closing( ssh.sftp.file(path_info.path, mode="r") ) as fd: @@ -325,81 +317,3 @@ class RemoteSSH(RemoteBASE): ret = list(itertools.compress(checksums, in_remote)) pbar.update_desc("", 0) # clear path name description return ret - - def link(self, from_info, to_info): - with self.ssh(from_info) as ssh: - self._link(from_info, to_info, self.cache_types, ssh) - - def _link(self, from_info, to_info, link_types, ssh): - from_path = from_info.path - to_path = to_info.path - - assert ssh.isfile(from_path) - - dname = posixpath.dirname(to_path) - - # XXX: We can remove the `exists` call, since `makedirs` - # will ignore it - if not ssh.exists(dname): - ssh.makedirs(dname) - - # NOTE: just create an empty file for an empty cache - if ssh.getsize(from_path) == 0: - ssh.open(to_path, "w+").close() - - msg = "Created empty file: {} -> {}".format(from_path, to_path) - logger.debug(msg) - return - - self._try_links(from_info, to_info, link_types, ssh) - - @classmethod - def _get_link_method(cls, link_type, ssh): - CACHE_TYPE_MAP = { - "copy": ssh.cp, - "symlink": ssh.symlink, - "reflink": ssh.reflink, - "hardlink": ssh.hardlink, - } - - try: - return CACHE_TYPE_MAP[link_type] - except KeyError: - raise DvcException( - "Cache type: '{}' not supported!".format(link_type) - ) - - def _do_link(self, from_info, to_info, link_method, ssh): - # XXX: We are testing if file exists rather than if file is a link - if ssh.exists(to_info.path): - raise DvcException("Link '{}' already exists!".format(to_info)) - else: - link_method(from_info.path, to_info.path) - - if self.protected: - self.protect(to_info) - - msg = "Created {}'{}': {} -> {}".format( - "protected " if self.protected else "", - self.cache_types[0], - from_info, - to_info, - ) - logger.debug(msg) - - @slow_link_guard - def _try_links(self, from_info, to_info, link_types, ssh): - i = len(link_types) - while i > 0: - link_method = self._get_link_method(link_types[0], ssh) - try: - self._do_link(from_info, to_info, link_method, ssh) - return - - except DvcException as exc: - msg = "Cache type '{}' is not supported: {}" - logger.debug(msg.format(link_types[0], str(exc))) - del link_types[0] - i -= 1 - - raise DvcException("no possible cache types left to try out.") diff --git a/dvc/remote/ssh/connection.py b/dvc/remote/ssh/connection.py index bfb058187..0ef7b4bb1 100644 --- a/dvc/remote/ssh/connection.py +++ b/dvc/remote/ssh/connection.py @@ -273,8 +273,7 @@ class SSHConnection: assert len(md5) == 32 return md5 - def cp(self, src, dest): - self.makedirs(posixpath.dirname(dest)) + def copy(self, src, dest): self.execute("cp {} {}".format(src, dest)) def open_max_sftp_channels(self): @@ -293,13 +292,10 @@ class SSHConnection: return self.sftp.open(*args, **kwargs) def symlink(self, src, dest): - self.makedirs(posixpath.dirname(dest)) self.sftp.symlink(src, dest) def reflink(self, src, dest): - self.makedirs(posixpath.dirname(dest)) self.execute("cp --reflink {} {}".format(src, dest)) def hardlink(self, src, dest): - self.makedirs(posixpath.dirname(dest)) self.execute("ln {} {}".format(src, dest)) diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index e1139d90a..4b83eda67 100644 --- a/dvc/repo/__init__.py +++ b/dvc/repo/__init__.py @@ -459,6 +459,8 @@ class Repo(object): @contextmanager def open(self, path, remote=None, mode="r", encoding=None): """Opens a specified resource as a file descriptor""" + assert mode in {"r", "rt", "rb"} + try: with self._open(path, remote, mode, encoding) as fd: yield fd diff --git a/dvc/utils/http.py b/dvc/utils/http.py index b2ecf4bb3..7d8c6f55a 100644 --- a/dvc/utils/http.py +++ b/dvc/utils/http.py @@ -13,8 +13,6 @@ def open_url(url, mode="r", encoding=None): Resumes on connection error. Url could be a string or a callable returning a string. """ - assert mode in {"r", "rt", "rb"} - with iter_url(url) as (response, it): bytes_stream = IterStream(it) diff --git a/tests/unit/remote/test_slow_link_detection.py b/tests/unit/remote/test_slow_link_detection.py index 3b981e69e..791c8a945 100644 --- a/tests/unit/remote/test_slow_link_detection.py +++ b/tests/unit/remote/test_slow_link_detection.py @@ -3,7 +3,7 @@ import logging import colorama import pytest from dvc.config import Config -from dvc.remote.local.slow_link_detection import ( +from dvc.remote.slow_link_detection import ( slow_link_guard, SlowLinkDetectorDecorator, ) @@ -11,18 +11,15 @@ from mock import Mock, patch @patch( - "dvc.remote.local.slow_link_detection.SlowLinkDetectorDecorator", - autospec=True, + "dvc.remote.slow_link_detection.SlowLinkDetectorDecorator", autospec=True ) class TestSlowLinkGuard(object): @pytest.fixture(autouse=True) def setUp(self): self.method = Mock() - self.remote_local_mock = Mock() + self.remote_mock = Mock() self.config_mock = Mock() - self.remote_local_mock.repo.config.config.get.return_value = ( - self.config_mock - ) + self.remote_mock.repo.config.config.get.return_value = self.config_mock def _cache_config(self, slow_link_warning, cache_type): def config_side_effect(section_name, _): @@ -38,7 +35,7 @@ class TestSlowLinkGuard(object): ): self._cache_config(slow_link_warning=True, cache_type=None) - slow_link_guard(self.method)(self.remote_local_mock) + slow_link_guard(self.method)(self.remote_mock) assert 1 == SlowLinkDetectorDecoratorClassMock.call_count assert 0 == self.method.call_count @@ -54,7 +51,7 @@ class TestSlowLinkGuard(object): slow_link_warning=slow_link_warning, cache_type=cache_type ) - slow_link_guard(self.method)(self.remote_local_mock) + slow_link_guard(self.method)(self.remote_mock) assert 0 == SlowLinkDetectorDecoratorClassMock.call_count assert 1 == self.method.call_count