diff --git a/dvc/cli.py b/dvc/cli.py index 972e22959..5dd6d7371 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -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 diff --git a/dvc/command/gc.py b/dvc/command/gc.py index fa05faf86..f0b25ae49 100644 --- a/dvc/command/gc.py +++ b/dvc/command/gc.py @@ -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 diff --git a/dvc/project.py b/dvc/project.py index 05aedc62a..8682e62c0 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -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, diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 44dc4c2a3..a6ad7e681 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -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]) diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index 32b07dd48..2dbf967fe 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -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