mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-13 23:10:15 +01:00
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 <[email protected]>
This commit is contained in:
+184
-15
@@ -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)
|
||||
|
||||
+5
-6
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
+2
-2
@@ -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
|
||||
|
||||
+14
-175
@@ -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)
|
||||
|
||||
+9
-8
@@ -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)
|
||||
|
||||
+19
-11
@@ -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
|
||||
|
||||
|
||||
+22
-17
@@ -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 = '<SYMLINK>'
|
||||
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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -25,6 +25,7 @@ install_requires = [
|
||||
"paramiko>=2.4.1",
|
||||
"binaryornot==0.4.4",
|
||||
"jsonpath-rw==1.4.0",
|
||||
"reflink==0.2.0",
|
||||
]
|
||||
|
||||
setup(
|
||||
|
||||
+2
-3
@@ -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, '<clearly>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))
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
+2
-2
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user