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

remote: properly collect dir cache

Browse Source 
Fixes #1293

Signed-off-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
2018-11-06 14:19:38 +02:00
parent 7398a174a8
commit f00d08d46d
5 changed files with 121 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/cli.py
+6
View File
@@ -578,6 +578,12 @@ def parse_args(argv=None):
action='store_true',
default=False,
help='Force garbage collection.')
gc_parser.add_argument(
'-j',
'--jobs',
type=int,
default=None,
help='Number of jobs to run simultaneously.')
gc_parser.set_defaults(func=CmdGC)
# Config
dvc/command/gc.py
+3 -1
View File
@@ -22,5 +22,7 @@ class CmdGC(CmdBase):
self.project.gc(all_branches=self.args.all_branches,
all_tags=self.args.all_tags,
cloud=self.args.cloud,
remote=self.args.remote)
remote=self.args.remote,
force=self.args.force,
jobs=self.args.jobs)
return 0
dvc/project.py
+96 -13
View File
@@ -491,12 +491,82 @@ class Project(object):
return ret
def _collect_dir_cache(self,
out,
branch=None,
remote=None,
force=False,
jobs=None):
info = out.dumpd()
ret = [info]
r = out.remote
md5 = info[r.PARAM_MD5]
if self.cache.local.changed_cache_file(md5):
try:
self.cloud.pull(ret,
jobs=jobs,
remote=remote,
show_checksums=False)
except DvcException as exc:
msg = "Failed to pull cache for '{}': {}"
self.logger.debug(msg.format(out, exc))
if self.cache.local.changed_cache_file(md5):
msg = "Missing cache for directory '{}'. " \
"Cache for files inside will be lost. " \
"Would you like to continue? Use '-f' to force. "
if not (force or self.prompt.prompt(msg, False)):
raise DvcException("Unable to fully collect "
"used cache without cache "
"for directory '{}'".format(out))
else:
return ret
for i in self.cache.local.load_dir_cache(md5):
i['branch'] = branch
i[r.PARAM_PATH] = os.path.join(info[r.PARAM_PATH],
i[r.PARAM_RELPATH])
ret.append(i)
return ret
def _collect_used_cache(self,
out,
branch=None,
remote=None,
force=False,
jobs=None):
if not out.use_cache or not out.info:
return []
info = out.dumpd()
info['branch'] = branch
ret = [info]
if out.path_info['scheme'] != 'local':
return ret
md5 = info[out.remote.PARAM_MD5]
cache = self.cache.local.get(md5)
if not out.remote.is_dir_cache(cache):
return ret
return self._collect_dir_cache(out,
branch=branch,
remote=remote,
force=force,
jobs=jobs)
def _used_cache(self,
target=None,
all_branches=False,
active=True,
with_deps=False,
all_tags=False):
all_tags=False,
remote=None,
force=False,
jobs=None):
cache = {}
cache['local'] = []
cache['s3'] = []
@@ -522,13 +592,12 @@ class Project(object):
self.logger.warn(msg.format(stage.relpath))
for out in stage.outs:
if not out.use_cache or not out.info:
continue
info = out.dumpd()
info['branch'] = branch
cache[out.path_info['scheme']] += [info]
scheme = out.path_info['scheme']
cache[scheme] += self._collect_used_cache(out,
branch=branch,
remote=remote,
force=force,
jobs=jobs)
return cache
@@ -542,13 +611,18 @@ class Project(object):
cloud=False,
remote=None,
with_deps=False,
all_tags=False):
all_tags=False,
force=False,
jobs=None):
with self.state:
clist = self._used_cache(target=None,
all_branches=all_branches,
active=False,
with_deps=with_deps,
all_tags=all_tags)
all_tags=all_tags,
remote=remote,
force=force,
jobs=jobs)
self._do_gc('local', self.cache.local.gc, clist)
if self.cache.s3:
@@ -582,7 +656,10 @@ class Project(object):
used = self._used_cache(target,
all_branches=all_branches,
all_tags=all_tags,
with_deps=with_deps)['local']
with_deps=with_deps,
force=True,
remote=remote,
jobs=jobs)['local']
self.cloud.push(used,
jobs,
remote=remote,
@@ -600,7 +677,10 @@ class Project(object):
used = self._used_cache(target,
all_branches=all_branches,
all_tags=all_tags,
with_deps=with_deps)['local']
with_deps=with_deps,
force=True,
remote=remote,
jobs=jobs)['local']
self.cloud.pull(used,
jobs,
remote=remote,
@@ -654,7 +734,10 @@ class Project(object):
used = self._used_cache(target,
all_branches=all_branches,
all_tags=all_tags,
with_deps=with_deps)['local']
with_deps=with_deps,
force=True,
remote=remote,
jobs=jobs)['local']
status = {}
for md5, ret in self.cloud.status(used,
dvc/remote/local.py
+6 -54
View File
@@ -470,30 +470,6 @@ class RemoteLOCAL(RemoteBase):
os.rename(tmp_file, to_info['path'])
def _collect(self, checksum_infos, push=False):
missing = []
collected = []
for info in checksum_infos:
md5 = info[self.PARAM_MD5]
cache = self.get(md5)
if not self.is_dir_cache(info[self.PARAM_MD5]):
continue
if not os.path.exists(cache):
missing.append(info)
continue
for i in self.load_dir_cache(md5):
if info.get('branch'):
i['branch'] = info['branch']
i[self.PARAM_PATH] = os.path.join(info[self.PARAM_PATH],
i[self.PARAM_RELPATH])
collected.append(i.copy())
collected.extend(checksum_infos)
return collected, missing
def _group(self, checksum_infos, show_checksums=False):
by_md5 = {}
@@ -519,7 +495,7 @@ class RemoteLOCAL(RemoteBase):
return list(by_md5.keys()), list(by_md5.values())
def gc(self, checksum_infos):
checksum_infos = self._collect(checksum_infos['local'])[0]
checksum_infos = checksum_infos['local']
used_md5s = [info[self.PARAM_MD5] for info in checksum_infos]
removed = False
@@ -538,9 +514,6 @@ class RemoteLOCAL(RemoteBase):
progress.set_n_total(1)
progress.update_target(title, 0, 100)
checksum_infos, missing = self._collect(checksum_infos)
checksum_infos += missing
progress.update_target(title, 10, 100)
md5s, names = self._group(checksum_infos,
@@ -564,11 +537,11 @@ class RemoteLOCAL(RemoteBase):
local_exists,
remote_exists)]
def _do_pull(self,
checksum_infos,
remote,
jobs=None,
show_checksums=False):
def pull(self,
checksum_infos,
remote,
jobs=None,
show_checksums=False):
title = "Collecting information"
progress.set_n_total(1)
@@ -631,23 +604,6 @@ class RemoteLOCAL(RemoteBase):
for f in futures:
f.result()
def pull(self, checksum_infos, remote, jobs=None, show_checksums=False):
Logger.info("Preparing to pull data from {}".format(remote.url))
# NOTE: try fetching missing dir info
checksum_infos, missing = self._collect(checksum_infos)
if len(missing) > 0:
self._do_pull(missing,
remote,
jobs,
show_checksums=show_checksums)
checksum_infos += self._collect(missing)[0]
self._do_pull(checksum_infos,
remote,
jobs,
show_checksums=show_checksums)
def push(self, checksum_infos, remote, jobs=None, show_checksums=False):
Logger.info("Preparing to push data to {}".format(remote.url))
title = "Collecting information"
@@ -655,10 +611,6 @@ class RemoteLOCAL(RemoteBase):
progress.set_n_total(1)
progress.update_target(title, 0, 100)
checksum_infos = self._collect(checksum_infos)[0]
progress.update_target(title, 10, 100)
# NOTE: verifying that our cache is not corrupted
def func(info):
return not self.changed_cache_file(info[self.PARAM_MD5])
tests/test_data_cloud.py
+10
View File
@@ -414,6 +414,16 @@ class TestDataCloudCLIBase(TestDvc):
self.main(['status', '-c', '--show-checksums'] + args)
# NOTE: check if remote gc works correctly on directories
self.main(['gc', '-c', '-f'] + args)
shutil.rmtree(self.dvc.cache.local.cache_dir)
self.main(['pull'] + args)
self.assertTrue(os.path.exists(cache))
self.assertTrue(os.path.isfile(cache))
self.assertTrue(os.path.isfile(cache_dir))
self.assertTrue(os.path.isfile(self.FOO))
self.assertTrue(os.path.isdir(self.DATA_DIR))
def _should_test(self):
return True