From 0a049c232bb440a4f10a618943ace5835b75fec2 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sat, 28 Apr 2018 17:33:25 +0300 Subject: [PATCH] cache: configurable cache dir and cache type From now on we allow specifying standalone cache dirs as well as specifying linking method(copy, symlink, hardlink, reflink) for it. If 'cache type' is not specified in the config, dvc will try all of them starting from reflink all the way down to copy. Signed-off-by: Ruslan Kuprieiev --- dvc/cache.py | 199 ++++++++++++++++++++++++++++++++++++--- dvc/cloud/base.py | 11 +-- dvc/cloud/data_cloud.py | 4 +- dvc/config.py | 11 +++ dvc/lock.py | 4 +- dvc/output.py | 189 +++---------------------------------- dvc/project.py | 17 ++-- dvc/state.py | 30 +++--- dvc/system.py | 39 ++++---- dvc/utils.py | 19 ++++ requirements.txt | 1 + setup.py | 1 + tests/test_cache.py | 5 +- tests/test_data_cloud.py | 5 +- tests/test_state.py | 4 +- 15 files changed, 294 insertions(+), 245 deletions(-) diff --git a/dvc/cache.py b/dvc/cache.py index 7f0c9fdda..86dc8aa78 100644 --- a/dvc/cache.py +++ b/dvc/cache.py @@ -1,36 +1,63 @@ import os +import json +import shutil +from dvc.state import State from dvc.system import System +from dvc.logger import Logger +from dvc.utils import move, remove +from dvc.lock import Lock +from dvc.exceptions import DvcException class Cache(object): CACHE_DIR = 'cache' + CACHE_DIR_LOCK = 'cache.lock' + CACHE_TYPES = ['reflink', 'hardlink', 'symlink', 'copy'] + CACHE_TYPE_MAP = { + 'copy': shutil.copyfile, + 'symlink': System.symlink, + 'hardlink': System.hardlink, + 'reflink': System.reflink, + } + + def __init__(self, dvc_dir, cache_dir=None, cache_type=None): + self.cache_type = cache_type + + cache_dir = cache_dir if cache_dir else self.CACHE_DIR + if os.path.isabs(cache_dir): + self.cache_dir = cache_dir + else: + self.cache_dir = os.path.abspath(os.path.realpath(os.path.join(dvc_dir, cache_dir))) - def __init__(self, dvc_dir): - self.cache_dir = os.path.abspath(os.path.realpath(os.path.join(dvc_dir, self.CACHE_DIR))) if not os.path.exists(self.cache_dir): os.mkdir(self.cache_dir) + self.state = State(self.cache_dir) + self.lock = Lock(self.cache_dir, name=self.CACHE_DIR_LOCK) + @staticmethod - def init(dvc_dir): - cache_dir = os.path.join(dvc_dir, Cache.CACHE_DIR) - os.mkdir(cache_dir) - return Cache(dvc_dir) + def init(dvc_dir, cache_dir=None): + return Cache(dvc_dir, cache_dir=None) def all(self): - clist = [] - for entry in os.listdir(self.cache_dir): - subdir = os.path.join(self.cache_dir, entry) - if not os.path.isdir(subdir): - continue + with self.lock: + clist = [] + for entry in os.listdir(self.cache_dir): + subdir = os.path.join(self.cache_dir, entry) + if not os.path.isdir(subdir): + continue - for cache in os.listdir(subdir): - path = os.path.join(subdir, cache) - clist.append(path) + for cache in os.listdir(subdir): + path = os.path.join(subdir, cache) + clist.append(path) - return clist + return clist def get(self, md5): + if not md5: + return None + return os.path.join(self.cache_dir, md5[0:2], md5[2:]) def path_to_md5(self, path): @@ -46,3 +73,145 @@ class Cache(object): file_set = file_set - set(cached_files) return cached + def _changed(self, md5): + cache = self.get(md5) + if self.state.changed(cache, md5=md5): + if os.path.exists(cache): + Logger.warn('Corrupted cache file {}'.format(os.path.relpath(cache))) + remove(cache) + return True + + return False + + def changed(self, md5): + with self.lock: + return self._changed(md5) + + def link(self, src, link): + dname = os.path.dirname(link) + if not os.path.exists(dname): + os.makedirs(dname) + + if self.cache_type != None: + types = [self.cache_type] + else: + types = self.CACHE_TYPES + + for typ in types: + try: + self.CACHE_TYPE_MAP[typ](src, link) + return + except Exception as exc: + msg = 'Cache type \'{}\' is not supported'.format(typ) + Logger.debug(msg) + if typ == types[-1]: + raise DvcException(msg, cause=exc) + + @staticmethod + def load_dir_cache(path): + if os.path.isabs(path): + relpath = os.path.relpath(path) + else: + relpath = path + + try: + with open(path, 'r') as fd: + d = json.load(fd) + except Exception as exc: + msg = u'Failed to load dir cache \'{}\'' + Logger.error(msg.format(relpath), exc) + return [] + + if not isinstance(d, list): + msg = u'Dir cache file format error \'{}\': skipping the file' + Logger.error(msg.format(relpath)) + return [] + + return d + + @staticmethod + def get_dir_cache(path): + res = {} + d = Cache.load_dir_cache(path) + + for entry in d: + res[entry[State.PARAM_RELPATH]] = entry[State.PARAM_MD5] + + return res + + def dir_cache(self, cache): + res = {} + dir_cache = self.get_dir_cache(cache) + + for relpath, md5 in dir_cache.items(): + res[relpath] = self.get(md5) + + return res + + @staticmethod + def is_dir_cache(cache): + return cache.endswith(State.MD5_DIR_SUFFIX) + + def _checkout(self, path, md5): + cache = self.get(md5) + + if not cache or not os.path.exists(cache) or self._changed(md5): + if cache: + Logger.warn(u'\'{}({})\': cache file not found'.format(os.path.relpath(cache), + os.path.relpath(path))) + remove(path) + return + + if os.path.exists(path): + msg = u'Data \'{}\' exists. Removing before checkout' + Logger.debug(msg.format(os.path.relpath(path))) + remove(path) + + msg = u'Checking out \'{}\' with cache \'{}\'' + Logger.debug(msg.format(os.path.relpath(path), os.path.relpath(cache))) + + if not self.is_dir_cache(cache): + self.link(cache, path) + return + + dir_cache = self.dir_cache(cache) + for relpath, c in dir_cache.items(): + p = os.path.join(path, relpath) + self.link(c, p) + + def checkout(self, path, md5): + with self.lock: + return self._checkout(path, md5) + + def _save_file(self, path): + md5 = self.state.update(path) + cache = self.get(md5) + if self._changed(md5): + move(path, cache) + self.state.update(cache) + self._checkout(path, md5) + + def _save_dir(self, path): + md5 = self.state.update(path) + cache = self.get(md5) + dname = os.path.dirname(cache) + dir_info = self.state.collect_dir(path) + + for entry in dir_info: + relpath = entry[State.PARAM_RELPATH] + p = os.path.join(path, relpath) + + self._save_file(p) + + if not os.path.isdir(dname): + os.makedirs(dname) + + with open(cache, 'w+') as fd: + json.dump(dir_info, fd, sort_keys=True) + + def save(self, path): + with self.lock: + if os.path.isdir(path): + self._save_dir(path) + else: + self._save_file(path) diff --git a/dvc/cloud/base.py b/dvc/cloud/base.py index b23e19728..54dec99dd 100644 --- a/dvc/cloud/base.py +++ b/dvc/cloud/base.py @@ -5,7 +5,7 @@ import tempfile from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.config import Config, ConfigError -from dvc.stage import Output +from dvc.cache import Cache STATUS_UNKNOWN = 0 @@ -31,9 +31,8 @@ class DataCloudError(DvcException): class CloudSettings(object): - def __init__(self, cache=None, state=None, global_storage_path=None, cloud_config=None): + def __init__(self, cache=None, global_storage_path=None, cloud_config=None): self.cache = cache - self.state = state self.cloud_config = cloud_config self.global_storage_path = global_storage_path @@ -131,7 +130,7 @@ class DataCloudBase(object): path, local = arg ret = [path] - if not Output.is_dir_cache(path): + if not Cache.is_dir_cache(path): return ret if local: @@ -147,7 +146,7 @@ class DataCloudBase(object): self._pull_key(key, tmp, no_progress_bar=True) dir_path = tmp - for relpath, md5 in Output.get_dir_cache(dir_path).items(): + for relpath, md5 in Cache.get_dir_cache(dir_path).items(): cache = self._cloud_settings.cache.get(md5) ret.append(cache) @@ -155,7 +154,7 @@ class DataCloudBase(object): def _cmp_checksum(self, blob, fname): md5 = self._cloud_settings.cache.path_to_md5(fname) - if self._cloud_settings.state.changed(fname, md5=md5): + if self._cloud_settings.cache.state.changed(fname, md5=md5): return False return True diff --git a/dvc/cloud/data_cloud.py b/dvc/cloud/data_cloud.py index c88679934..06b23be65 100644 --- a/dvc/cloud/data_cloud.py +++ b/dvc/cloud/data_cloud.py @@ -23,10 +23,9 @@ class DataCloud(object): 'local' : DataCloudLOCAL, } - def __init__(self, cache=None, config=None, state=None): + def __init__(self, cache=None, config=None): self._cache = cache self._config = config - self._state = state remote = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_REMOTE, '') if remote == '': @@ -82,7 +81,6 @@ class DataCloud(object): Logger.warn('Using obsoleted config format. Consider updating.') cloud_settings = CloudSettings(cache=self._cache, - state=self._state, global_storage_path=global_storage_path, cloud_config=cloud_config) diff --git a/dvc/config.py b/dvc/config.py index b1832c0de..a98454d9a 100644 --- a/dvc/config.py +++ b/dvc/config.py @@ -5,6 +5,7 @@ import os import schema import configobj +from dvc.cache import Cache from dvc.exceptions import DvcException @@ -28,6 +29,15 @@ class Config(object): SECTION_CORE_LOGLEVEL_SCHEMA = schema.And(schema.Use(str.lower), lambda l: l in ('info', 'debug', 'warning', 'error')) SECTION_CORE_REMOTE = 'remote' + SECTION_CACHE = 'cache' + SECTION_CACHE_DIR = 'dir' + SECTION_CACHE_TYPE = 'type' + SECTION_CACHE_TYPE_SCHEMA = schema.And(schema.Use(str.lower), lambda t: t in Cache.CACHE_TYPE_MAP.keys()) + SECTION_CACHE_SCHEMA = { + schema.Optional(SECTION_CACHE_DIR, default=Cache.CACHE_DIR): str, + schema.Optional(SECTION_CACHE_TYPE, default=None): SECTION_CACHE_TYPE_SCHEMA, + } + # backward compatibility SECTION_CORE_CLOUD = 'cloud' SECTION_CORE_CLOUD_SCHEMA = schema.And(schema.Use(str.lower), lambda c: c in ('aws', 'gcp', 'local', '')) @@ -86,6 +96,7 @@ class Config(object): SCHEMA = { schema.Optional(SECTION_CORE, default={}): SECTION_CORE_SCHEMA, schema.Optional(schema.Regex(SECTION_REMOTE_REGEX)): SECTION_REMOTE_SCHEMA, + schema.Optional(SECTION_CACHE, default={}): SECTION_CACHE_SCHEMA, # backward compatibility schema.Optional(SECTION_AWS, default={}): SECTION_AWS_SCHEMA, diff --git a/dvc/lock.py b/dvc/lock.py index 0fbff6160..68a659562 100644 --- a/dvc/lock.py +++ b/dvc/lock.py @@ -13,8 +13,8 @@ class Lock(object): LOCK_FILE = 'lock' TIMEOUT = 5 - def __init__(self, dvc_dir): - self.lock_file = os.path.join(dvc_dir, self.LOCK_FILE) + def __init__(self, dvc_dir, name=LOCK_FILE): + self.lock_file = os.path.join(dvc_dir, name) self._lock = None @staticmethod diff --git a/dvc/output.py b/dvc/output.py index c2d032a98..d32a334f1 100644 --- a/dvc/output.py +++ b/dvc/output.py @@ -1,15 +1,11 @@ import os -import stat -import json -import shutil import schema import posixpath import ntpath -from dvc.system import System -from dvc.utils import file_md5 from dvc.exceptions import DvcException from dvc.logger import Logger +from dvc.utils import remove class OutputError(DvcException): @@ -47,7 +43,6 @@ class CmdOutputAlreadyTrackedError(CmdOutputError): class Dependency(object): - PARAM_RELPATH = 'relpath' PARAM_PATH = 'path' PARAM_MD5 = 'md5' MD5_DIR_SUFFIX = '.dir' @@ -59,7 +54,7 @@ class Dependency(object): def __init__(self, project, path, md5=None): self.project = project - self.path = os.path.abspath(os.path.realpath(path)) + self.path = os.path.abspath(os.path.normpath(path)) if not self.path.startswith(self.project.root_dir): raise CmdOutputOutsideOfRepoError(self.rel_path) @@ -96,10 +91,6 @@ class Dependency(object): return {self.rel_path: 'changed'} return {} - @staticmethod - def is_dir_cache(cache): - return cache.endswith(Output.MD5_DIR_SUFFIX) - def save(self): if not os.path.exists(self.path): raise CmdOutputDoesNotExistError(self.rel_path) @@ -160,18 +151,11 @@ class Output(Dependency): @property def cache(self): - if not self.md5: + if not self.use_cache: return None return self.project.cache.get(self.md5) - @property - def rel_cache(self): - if not self.cache: - return None - - return os.path.relpath(self.cache) - def dumpd(self, cwd): ret = super(Output, self).dumpd(cwd) ret[Output.PARAM_CACHE] = self.use_cache @@ -193,151 +177,28 @@ class Output(Dependency): def loads_from(cls, project, s_list, use_cache=False, cwd=os.curdir): return [cls.loads(project, x, use_cache=use_cache, cwd=cwd) for x in s_list] - def _changed_cache(self, cache): - md5 = self.project.cache.path_to_md5(cache) - if self.project.state.changed(cache, md5=md5): - self.project.logger.warn('Corrupted cache file {}'.format(os.path.relpath(cache))) - os.unlink(cache) + def changed(self): + if super(Output, self).changed(): + return True + + if self.use_cache and self.project.cache.changed(self.md5): return True return False - def _changed_file(self, path, cache): - if os.path.isfile(path) and \ - os.path.isfile(cache) and \ - System.samefile(path, cache) and \ - not self._changed_cache(cache): - return False - - return True - - def _changed_dir(self): - if os.path.isdir(self.path) and \ - os.path.isfile(self.cache) and \ - not self._changed_cache(self.cache): - return False - - return True - - def changed(self): - if not self.use_cache: - ret = super(Output, self).changed() - elif not self.cache: - ret = True - elif self.is_dir_cache(self.cache): - ret = self._changed_dir() - else: - ret = self._changed_file(self.path, self.cache) - - msg = u'Data file or dir \'{}\' with cache \'{}\' {}' - self.project.logger.debug(msg.format(self.rel_path, self.rel_cache, self._changed_msg(ret))) - - return ret - - def hardlink(self, src, link): - rel_src = os.path.relpath(src) - rel_link = os.path.relpath(link) - self.project.logger.debug(u'creating hardlink {} -> {}'.format(rel_src, rel_link)) - - dname = os.path.dirname(link) - if not os.path.exists(dname): - os.makedirs(dname) - - System.hardlink(src, link) - - @staticmethod - def load_dir_cache(path): - if os.path.isabs(path): - relpath = os.path.relpath(path) - else: - relpath = path - - try: - with open(path, 'r') as fd: - d = json.load(fd) - except Exception as exc: - msg = u'Failed to load dir cache \'{}\'' - Logger.error(msg.format(relpath), exc) - return [] - - if not isinstance(d, list): - msg = u'Dir cache file format error \'{}\': skipping the file' - Logger.error(msg.format(relpath)) - return [] - - return d - - @staticmethod - def get_dir_cache(path): - res = {} - d = Output.load_dir_cache(path) - - for entry in d: - res[entry[Output.PARAM_RELPATH]] = entry[Output.PARAM_MD5] - - return res - - def dir_cache(self): - res = {} - dir_cache = self.get_dir_cache(self.cache) - - for relpath, md5 in dir_cache.items(): - res[relpath] = self.project.cache.get(md5) - - return res - def checkout(self): if not self.use_cache: return msg = u'Checking out \'{}\' with cache \'{}\'' - self.project.logger.debug(msg.format(self.rel_path, self.rel_cache)) + self.project.logger.debug(msg.format(self.rel_path, self.md5)) if not self.changed(): msg = u'Data file \'{}\' with cache \'{}\' didn\'t change, skipping checkout.' - self.project.logger.debug(msg.format(self.rel_path, self.rel_cache)) + self.project.logger.debug(msg.format(self.rel_path, self.md5)) return - if not self.cache or not os.path.exists(self.cache): - if self.cache: - self.project.logger.warn(u'\'{}({})\': cache file not found'.format(self.rel_cache, self.rel_path)) - self.remove() - return - - if os.path.exists(self.path): - msg = u'Data file \'{}\' exists. Removing before checkout' - self.project.logger.debug(msg.format(self.rel_path)) - self.remove() - - if not self.is_dir_cache(self.cache): - self.hardlink(self.cache, self.path) - return - - for relpath, cache in self.dir_cache().items(): - path = os.path.join(self.path, relpath) - self.hardlink(cache, path) - - def _save_dir(self): - dname = os.path.dirname(self.cache) - dir_info = self.project.state.collect_dir(self.path) - - for entry in dir_info: - md5 = entry[self.PARAM_MD5] - relpath = entry[self.PARAM_RELPATH] - path = os.path.join(self.path, relpath) - cache = self.project.cache.get(md5) - - if os.path.exists(cache): - self._remove(path) - self.hardlink(cache, path) - else: - self.hardlink(path, cache) - - if not os.path.isdir(dname): - os.makedirs(dname) - - with open(self.cache, 'w+') as fd: - json.dump(dir_info, fd, sort_keys=True) + self.project.cache.checkout(self.path, self.md5) def save(self): super(Output, self).save() @@ -345,7 +206,7 @@ class Output(Dependency): if not self.use_cache: return - self.project.logger.debug(u'Saving \'{}\' to \'{}\''.format(self.rel_path, self.rel_cache)) + self.project.logger.debug(u'Saving \'{}\' to \'{}\''.format(self.rel_path, self.md5)) if self.project.scm.is_tracked(self.path): raise CmdOutputAlreadyTrackedError(self.rel_path) @@ -353,29 +214,7 @@ class Output(Dependency): if not self.changed(): return - if os.path.exists(self.cache): - # This means that we already have cache for this data. - # We remove data and link it to existing cache to save - # some space. - msg = u'Cache \'{}\' already exists, performing checkout for \'{}\'' - self.project.logger.debug(msg.format(self.rel_cache, self.rel_path)) - self.checkout() - return - - if os.path.isdir(self.path): - self._save_dir() - else: - self.hardlink(self.path, self.cache) - - def _remove(self, path): - if not os.path.exists(path): - return - - self.project.logger.debug(u'Removing \'{}\''.format(os.path.relpath(path))) - if os.path.isfile(path): - os.unlink(path) - else: - shutil.rmtree(path) + self.project.cache.save(self.path) def remove(self): - self._remove(self.path) + remove(self.path) diff --git a/dvc/project.py b/dvc/project.py index de3acae58..2c8e521c0 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -9,7 +9,7 @@ import dvc.cloud.base as cloud from dvc.logger import Logger from dvc.exceptions import DvcException -from dvc.stage import Stage, Output +from dvc.stage import Stage from dvc.config import Config from dvc.state import State from dvc.lock import Lock @@ -43,13 +43,14 @@ class Project(object): self.root_dir = os.path.abspath(os.path.realpath(root_dir)) self.dvc_dir = os.path.join(self.root_dir, self.DVC_DIR) + self.config = Config(self.dvc_dir) self.scm = SCM(self.root_dir) self.lock = Lock(self.dvc_dir) - self.cache = Cache(self.dvc_dir) - self.state = State(self.root_dir, self.dvc_dir) - self.config = Config(self.dvc_dir) + self.cache = Cache(self.dvc_dir, cache_dir=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_DIR, None), + cache_type=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_TYPE, None)) + self.state = State(self.dvc_dir) self.logger = Logger(self.config._config[Config.SECTION_CORE].get(Config.SECTION_CORE_LOGLEVEL, None)) - self.cloud = DataCloud(cache=self.cache, state=self.state, config=self.config._config) + self.cloud = DataCloud(cache=self.cache, config=self.config._config) @staticmethod def init(root_dir=os.curdir, no_scm=False): @@ -77,7 +78,7 @@ class Project(object): config = Config.init(dvc_dir) cache = Cache.init(dvc_dir) - state = State.init(root_dir, dvc_dir) + state = State.init(dvc_dir) lock = Lock(dvc_dir) scm.ignore_list([cache.cache_dir, @@ -211,8 +212,8 @@ class Project(object): if not out.use_cache or not out.cache: continue cache_set |= set([out.cache]) - if out.is_dir_cache(out.cache) and os.path.isfile(out.cache): - dir_cache = out.dir_cache() + if self.cache.is_dir_cache(out.cache) and os.path.isfile(out.cache): + dir_cache = self.cache.dir_cache(out.cache) cache_set |= set(dir_cache.values()) return list(cache_set) diff --git a/dvc/state.py b/dvc/state.py index 4d32bc1d6..13a29c99c 100644 --- a/dvc/state.py +++ b/dvc/state.py @@ -3,8 +3,8 @@ import json import nanotime import threading +from dvc.lock import Lock from dvc.system import System -from dvc.output import Output from dvc.utils import file_md5, dict_md5 from dvc.exceptions import DvcException @@ -40,17 +40,22 @@ class StateDuplicateError(DvcException): class State(object): STATE_FILE = 'state' + STATE_LOCK_FILE = 'state.lock' - def __init__(self, root_dir, dvc_dir): - self.root_dir = root_dir + PARAM_RELPATH = 'relpath' + PARAM_MD5 = 'md5' + MD5_DIR_SUFFIX = '.dir' + + def __init__(self, dvc_dir): self.dvc_dir = dvc_dir self.state_file = os.path.join(dvc_dir, self.STATE_FILE) self._db = self.load() self._lock = threading.Lock() + self._lock_file = Lock(dvc_dir, self.STATE_LOCK_FILE) @staticmethod - def init(root_dir, dvc_dir): - return State(root_dir, dvc_dir) + def init(dvc_dir): + return State(dvc_dir) def collect_dir(self, dname): dir_info = [] @@ -61,7 +66,7 @@ class State(object): relpath = os.path.relpath(path, dname) md5 = self.update(path, dump=False) - dir_info.append({Output.PARAM_RELPATH: relpath, Output.PARAM_MD5: md5}) + dir_info.append({self.PARAM_RELPATH: relpath, self.PARAM_MD5: md5}) self.dump() @@ -70,7 +75,7 @@ class State(object): def compute_md5(self, path): if os.path.isdir(path): dir_info = self.collect_dir(path) - return dict_md5(dir_info) + Output.MD5_DIR_SUFFIX + return dict_md5(dir_info) + self.MD5_DIR_SUFFIX else: return file_md5(path)[0] @@ -117,16 +122,19 @@ class State(object): d = state.dumpd() with self._lock: - self._db[inode] = d + with self._lock_file: + self._db[inode] = d - if dump: - self.dump() + if dump: + self.dump() return md5 def _get(self, inode, mtime): with self._lock: - d = self._db.get(inode, None) + with self._lock_file: + d = self._db.get(inode, None) + if not d: return None diff --git a/dvc/system.py b/dvc/system.py index 7a00dfa31..622e287b6 100644 --- a/dvc/system.py +++ b/dvc/system.py @@ -1,8 +1,6 @@ -import ctypes import os -import re -import subprocess -from builtins import str +import ctypes +import reflink if os.name == 'nt': import ntfsutils.hardlink as winlink @@ -10,9 +8,6 @@ if os.name == 'nt': class System(object): - SYMLINK_OUTPUT = '' - LONG_PATH_BUFFER_SIZE = 1024 - @staticmethod def is_unix(): return os.name != 'nt' @@ -24,6 +19,26 @@ class System(object): return winlink.create(source, link_name) + @staticmethod + def symlink(source, link_name): + if System.is_unix(): + return os.symlink(source, link_name) + + flags = 0 + if source is not None and os.path.isdir(source): + flags = 1 + + func = ctypes.windll.kernel32.CreateSymbolicLinkW + func.argtypes = (ctypes.c_wchar_p, ctypes.c_wchar_p, ctypes.c_uint32) + func.restype = ctypes.c_ubyte + + if func(link_name, source, flags) == 0: + raise ctypes.WinError() + + @staticmethod + def reflink(source, link_name): + return reflink.reflink(source, link_name) + @staticmethod def inode(path): if System.is_unix(): @@ -32,13 +47,3 @@ class System(object): # getdirinfo from ntfsutils works on both files and dirs info = getdirinfo(path) return hash((info.dwVolumeSerialNumber, info.nFileIndexHigh, info.nFileIndexLow)) - - @staticmethod - def samefile(path1, path2): - if not os.path.exists(path1) or not os.path.exists(path2): - return False - - if System.is_unix(): - return os.path.samefile(path1, path2) - - return winlink.samefile(path1, path2) diff --git a/dvc/utils.py b/dvc/utils.py index 7995c856b..86f8fdd4e 100644 --- a/dvc/utils.py +++ b/dvc/utils.py @@ -131,3 +131,22 @@ def map_progress(func, targets, n_threads): raise return list(zip(targets, ret)) + + +def move(src, dst): + dname = os.path.dirname(dst) + if not os.path.exists(dname): + os.makedirs(dname) + + shutil.move(src, dst) + + +def remove(path): + if not os.path.exists(path): + return + + Logger.debug(u'Removing \'{}\''.format(os.path.relpath(path))) + if os.path.isfile(path): + os.unlink(path) + else: + shutil.rmtree(path) diff --git a/requirements.txt b/requirements.txt index 3e8fb4911..2dbfd783e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -19,3 +19,4 @@ schema>=0.6.7 paramiko>=2.4.1 binaryornot==0.4.4 jsonpath-rw==1.4.0 +reflink==0.2.0 diff --git a/setup.py b/setup.py index 1417ee58d..3d5fa9e87 100644 --- a/setup.py +++ b/setup.py @@ -25,6 +25,7 @@ install_requires = [ "paramiko>=2.4.1", "binaryornot==0.4.4", "jsonpath-rw==1.4.0", + "reflink==0.2.0", ] setup( diff --git a/tests/test_cache.py b/tests/test_cache.py index 5520b28b9..be6bf0b02 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -2,7 +2,6 @@ import os from dvc.cache import Cache from dvc.system import System -from dvc.output import Output from tests.basic_env import TestDvc @@ -53,8 +52,8 @@ class TestCacheLoadBadDirCache(TestDvc): def test(self): fname = 'not-json' self.create(fname, 'not,json') - self._do_test(Output.load_dir_cache(fname)) + self._do_test(Cache.load_dir_cache(fname)) fname = 'not-list' self.create(fname, '{"a": "b"}') - self._do_test(Output.load_dir_cache(fname)) + self._do_test(Cache.load_dir_cache(fname)) diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index 171cd43a1..8a831c3b5 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -102,7 +102,7 @@ def sleep(): class TestDataCloud(TestDvc): def _test_cloud(self, config, cl): - cloud = DataCloud(cache=self.dvc.cache, state=self.dvc.state, config=config) + cloud = DataCloud(cache=self.dvc.cache, config=config) self.assertIsInstance(cloud._cloud, cl) def test(self): @@ -116,7 +116,7 @@ class TestDataCloud(TestDvc): with self.assertRaises(ConfigError) as cx: config = TEST_CONFIG config[TEST_SECTION][Config.SECTION_REMOTE_URL] = 'notsupportedscheme://a/b' - DataCloud(cache=self.dvc.cache, state=self.dvc.state, config=config) + DataCloud(cache=self.dvc.cache, config=config) class TestDataCloudBase(TestDvc): @@ -139,7 +139,6 @@ class TestDataCloudBase(TestDvc): config = TEST_CONFIG config[TEST_SECTION][Config.SECTION_REMOTE_URL] = repo cloud_settings = CloudSettings(cache=self.dvc.cache, - state=self.dvc.state, cloud_config=config[TEST_SECTION]) self.cloud = self._get_cloud_class()(cloud_settings) diff --git a/tests/test_state.py b/tests/test_state.py index 20e87cbfa..10ec64005 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -16,7 +16,7 @@ class TestState(TestDvc): mtime = os.path.getmtime(path) inode = System.inode(path) - state = State(self.dvc.root_dir, self.dvc.dvc_dir) + state = State(self.dvc.dvc_dir) entry_md5 = state.update(path) self.assertEqual(entry_md5, md5) @@ -41,6 +41,6 @@ class TestState(TestDvc): # Reload state db to make sure that it stays the same old_db = state._db - state = State(self.dvc.root_dir, self.dvc.dvc_dir) + state = State(self.dvc.dvc_dir) new_db = state._db self.assertEqual(old_db, new_db)