Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-08 23:09:48 +01:00
Code Issues Packages Projects Releases Wiki Activity

remote: abstract link logic to base

Browse Source
This commit is contained in:
Mr. Outis
2019-08-30 23:51:20 -05:00
parent 83a9a6362b
commit 40c61b3aaa
9 changed files with 67 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/remote/base.py
+15 -36
View File
@@ -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:
dvc/remote/hdfs.py
-2
View File
@@ -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")
dvc/remote/local/__init__.py
+23 -85
View File
@@ -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):
dvc/remote/local/slow_link_detection.py → dvc/remote/slow_link_detection.py
+4 -6
View File
@@ -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
dvc/remote/ssh/__init__.py
+16 -102
View File
@@ -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.")
dvc/remote/ssh/connection.py
+1 -5
View File
@@ -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))
dvc/repo/__init__.py
+2
View File
@@ -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
dvc/utils/http.py
-2
View File
@@ -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)
tests/unit/remote/test_slow_link_detection.py
+6 -9
View File
@@ -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