From 8bb51f5efc988778d91bdd3aeb67c4bb8edad05d Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 16 Oct 2017 02:43:26 +0300 Subject: [PATCH] cache: use checksums instead of commit_id From now on, we identify our cache files by their md5sum, so that we can avoid cache collission in branching scenario and also dedup same data at the same time. Fixes #250, #251 Signed-off-by: Ruslan Kuprieiev --- dvc/command/import_file.py | 5 ++-- dvc/command/init.py | 9 +++---- dvc/command/run.py | 4 +-- dvc/data_cloud.py | 51 +++++++++++++------------------------- dvc/executor.py | 2 +- dvc/git_wrapper.py | 9 ++++--- dvc/path/data_item.py | 20 +++++++++------ dvc/state_file.py | 14 ++++++++--- 8 files changed, 56 insertions(+), 58 deletions(-) diff --git a/dvc/command/import_file.py b/dvc/command/import_file.py index 55e552633..2ff486398 100644 --- a/dvc/command/import_file.py +++ b/dvc/command/import_file.py @@ -107,9 +107,8 @@ class CmdImportFile(CmdBase): raise ImportFileError('Output file directory "{}" does not exists'.format( os.path.dirname(data_item.data.relative))) - cache_dir = os.path.dirname(data_item.cache.relative) - if not os.path.exists(cache_dir): - os.makedirs(cache_dir) + if not os.path.exists(data_item.cache_dir): + os.makedirs(data_item.cache_dir) data_targets.append((input, data_item)) diff --git a/dvc/command/init.py b/dvc/command/init.py index 529c1a035..2e3568cdc 100644 --- a/dvc/command/init.py +++ b/dvc/command/init.py @@ -129,14 +129,13 @@ ProjectName = def create_empty_file(self): empty_data_path = os.path.join(self.parsed_args.data_dir, self.EMPTY_FILE_NAME) - cache_file_suffix = self.EMPTY_FILE_NAME + '_' + self.EMPTY_FILE_CHECKSUM - empty_cache_path = os.path.join(Config.CONFIG_DIR, Config.CACHE_DIR, cache_file_suffix) - open(empty_cache_path, 'w').close() - System.symlink(os.path.join('..', empty_cache_path), empty_data_path) + data_item = self.settings.path_factory.data_item(empty_data_path) + open(empty_data_path, 'w').close() + data_item.move_data_to_cache() StateFile(StateFile.COMMAND_EMPTY_FILE, - self.settings.path_factory.data_item(empty_data_path), + data_item, self.settings, input_files=[], output_files=[], diff --git a/dvc/command/run.py b/dvc/command/run.py index 662d4499c..9c8a94de7 100644 --- a/dvc/command/run.py +++ b/dvc/command/run.py @@ -112,12 +112,12 @@ class CmdRun(CmdBase): input_set = set(data_items_from_args + input_data_items) - output_set input_files_dvc = [x.data.dvc for x in input_set] - code_dependencies_dvc = self.git.abs_paths_to_dvc(code_dependencies + not_data_items_from_args) + code_dependencies_dvc = self.git.abs_paths_to_dvc(set(code_dependencies + not_data_items_from_args)) result = [] for data_item in repo_change.changed_data_items: Logger.debug('Move output file "{}" to cache dir "{}" and create a symlink'.format( - data_item.data.relative, data_item.cache.relative)) + data_item.data.relative, data_item.cache_dir)) data_item.move_data_to_cache() Logger.debug('Create state file "{}"'.format(data_item.state.relative)) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index a88b2de10..5a84692d0 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -154,7 +154,7 @@ class DataCloudBase(object): """ pass - def _import(self, bucket, path, fname, item): + def _import(self, bucket, path, item): """ Cloud-specific method for importing data file. """ @@ -168,14 +168,13 @@ class DataCloudBase(object): """ Generic method for pulling data from the cloud """ fname = item.resolved_cache.dvc key_name = self.cache_file_key(fname) - self.create_directory(item) - return self._import(self.storage_bucket, key_name, fname, item) + return self._import(self.storage_bucket, key_name, item) def import_data(self, url, item): """ Generic method for importing data """ parsed = urlparse(url) - return self._import(parsed.netloc, parsed.path, item.cache.relative, item) + return self._import(parsed.netloc, parsed.path, item) def sync(self, fname): """ @@ -189,23 +188,6 @@ class DataCloudBase(object): else: return self.pull(item) - def create_directory(self, item): - """ Create local directories for data item cache """ - self._lock.acquire() - try: - directory = os.path.dirname(item.cache.relative) - if not os.path.exists(directory): - Logger.debug(u'Creating directory {}'.format(directory)) - try: - os.makedirs(directory) - except OSError as ex: - raise DataCloudError(u'Cannot create directory {}: {}'.format(directory, ex)) - elif not os.path.isdir(directory): - msg = u'File {} cannot be synced because {} is not a directory' - raise DataCloudError(msg.format(item.cache.relative, directory)) - finally: - self._lock.release() - def remove(self, item): """ Cloud-specific method for removing data item from the cloud. @@ -233,11 +215,11 @@ class DataCloudLOCAL(DataCloudBase): copyfile(item.resolved_cache.dvc, self.storage_path) return item - def _import(self, bucket, i, out, item): - tmp_file = self.tmp_file(out) + def _import(self, bucket, i, item): + tmp_file = self.tmp_file(item.data.dvc) try: copyfile(i, tmp_file) - os.rename(tmp_file, out) + item.import_cache(tmp_file) except Exception as exc: Logger.error('Failed to copy "{}": {}'.format(i, exc)) return None @@ -246,15 +228,15 @@ class DataCloudLOCAL(DataCloudBase): def pull(self, item): Logger.debug('sync from cloud ' + self.storage_path + " " + item.resolved_cache.dvc) - return self._import(None, self.storage_path, item.resolved_cache.dvc, item) + return self._import(None, self.storage_path, item) def remove(self, item): Logger.debug('rm from cloud ' + item.resolved_cache.dvc) os.remove(item.resolved_cache.dvc) def import_data(self, path, item): - Logger.debug('import from cloud ' + path + " " + item.cache.relative) - return self._import(None, path, item.cache.relative, item) + Logger.debug('import from cloud ' + path + " " + item.data.dvc) + return self._import(None, path, item) def _status(self, data_item): local = data_item.resolved_cache.relative @@ -366,8 +348,7 @@ class DataCloudHTTP(DataCloudBase): Download single file from url. """ - to_file = item.cache.relative - tmp_file = self.tmp_file(to_file) + tmp_file = self.tmp_file(item.data.dvc) downloaded, header = self._downloaded_size(tmp_file) req = requests.get(url, stream=True, headers=header) @@ -382,7 +363,7 @@ class DataCloudHTTP(DataCloudBase): if not self._verify_md5(req, tmp_file): return None - os.rename(tmp_file, to_file) + item.import_cache(tmp_file) return item @@ -471,10 +452,11 @@ class DataCloudAWS(DataCloudBase): """ return fname + '.download' - def _import(self, bucket_name, key_name, fname, data_item): + def _import(self, bucket_name, key_name, data_item): bucket = self._get_bucket_aws(bucket_name) + fname = data_item.data.dvc tmp_file = self.tmp_file(fname) name = os.path.basename(fname) key = bucket.get_key(key_name) @@ -494,7 +476,7 @@ class DataCloudAWS(DataCloudBase): num_retries=10) try: key.get_contents_to_filename(tmp_file, cb=create_cb(name), res_download_handler=res_h) - os.rename(tmp_file, fname) + data_item.import_cache(tmp_file) except Exception as exc: Logger.error('Failed to download "{}": {}'.format(key_name, exc)) return None @@ -714,10 +696,11 @@ class DataCloudGCP(DataCloudBase): return False - def _import(self, bucket_name, key, fname, data_item): + def _import(self, bucket_name, key, data_item): bucket = self._get_bucket_gc(bucket_name) + fname = data_item.data.dvc name = os.path.basename(fname) tmp_file = self.tmp_file(fname) @@ -738,7 +721,7 @@ class DataCloudGCP(DataCloudBase): try: blob.download_to_filename(tmp_file) - os.rename(tmp_file, fname) + data_item.import_cache(tmp_file) except Exception as exc: Logger.error('Failed to download "{}": {}'.format(key, exc)) return None diff --git a/dvc/executor.py b/dvc/executor.py index 528d8cd8c..09dcfe38a 100644 --- a/dvc/executor.py +++ b/dvc/executor.py @@ -58,5 +58,5 @@ class Executor(object): code, out, err = Executor.exec_cmd(cmd, stdout_file=stdout_file, stderr_file=stderr_file, cwd=cwd, shell=shell) if code != 0: - raise ExecutorError('Git command error ({}):\n{}'.format(' '.join(cmd), err)) + raise ExecutorError('Git command error ({}):\n{}\n{}'.format(' '.join(cmd), out, err)) return out diff --git a/dvc/git_wrapper.py b/dvc/git_wrapper.py index 2e307159e..9cc120476 100644 --- a/dvc/git_wrapper.py +++ b/dvc/git_wrapper.py @@ -179,10 +179,13 @@ class GitWrapper(GitWrapperI): Executor.exec_cmd_only_success(['git', 'add', '--all']) out_status = Executor.exec_cmd_only_success(['git', 'status', '--porcelain']) - Executor.exec_cmd_only_success(['git', 'commit', '-m', message]) - Logger.debug('[dvc-git] Commit all changes. Success.') + changes = GitWrapper.parse_porcelain_files(out_status) + if len(changes) == 0: + return [] - return GitWrapper.parse_porcelain_files(out_status) + Executor.exec_cmd_only_success(['git', 'commit', '-m', '\'{}\''.format(message)]) + Logger.debug('[dvc-git] Commit all changes. Success.') + return changes def commit_all_changes_and_log_status(self, message): statuses = self.commit_all_changes(message) diff --git a/dvc/path/data_item.py b/dvc/path/data_item.py index 90e2fbff4..6719a75a5 100644 --- a/dvc/path/data_item.py +++ b/dvc/path/data_item.py @@ -5,6 +5,7 @@ from dvc.path.path import Path from dvc.exceptions import DvcException from dvc.system import System from dvc.utils import cached_property +from dvc.data_cloud import file_md5 class DataItemError(DvcException): @@ -81,14 +82,19 @@ class DataItem(object): state_file = os.path.join(state_dir, self.data_dvc_short + self.STATE_FILE_SUFFIX) return Path(state_file, self._git) + @cached_property + def cache_dir(self): + return os.path.join(self._git.git_dir_abs, self._config.cache_dir) + @cached_property def cache(self): - cache_dir = os.path.join(self._git.git_dir_abs, self._config.cache_dir) + cache_dir = self.cache_dir if self._cache_file: file_name = os.path.relpath(os.path.realpath(self._cache_file), cache_dir) else: - file_name = self.data_dvc_short + self.CACHE_FILE_SEP + self._git.curr_commit + from dvc.state_file import StateFile + file_name = StateFile.load(self, self._git).md5 cache_file = os.path.join(cache_dir, file_name) return Path(cache_file, self._git) @@ -125,9 +131,9 @@ class DataItem(object): return os.path.relpath(self.cache.relative, data_file_dir) def move_data_to_cache(self): - cache_dir = os.path.dirname(self.cache.relative) - if not os.path.isdir(cache_dir): - os.makedirs(cache_dir) - - shutil.move(self.data.relative, self.cache.relative) + self.import_cache(self.data.relative) System.symlink(self.symlink_file, self.data.relative) + + def import_cache(self, fname): + self._cache_file = os.path.join(self.cache_dir, file_md5(fname)[0]) + os.rename(fname, self._cache_file) diff --git a/dvc/state_file.py b/dvc/state_file.py index 0b8fadd90..0b45bc625 100644 --- a/dvc/state_file.py +++ b/dvc/state_file.py @@ -42,6 +42,7 @@ class StateFile(object): PARAM_SHELL = "Shell" PARAM_TARGET_METRICS = 'TargetMetrics' TARGET_METRICS_SINGLE_METRIC = 'SingleMetric' + PARAM_MD5 = 'Md5' def __init__(self, command, @@ -56,7 +57,8 @@ class StateFile(object): stderr=None, cwd=None, shell=False, - target_metrics={}): + target_metrics={}, + md5=None): self.data_item = data_item self.settings = settings @@ -66,6 +68,10 @@ class StateFile(object): self.code_dependencies = code_dependencies self.shell = shell + self.md5 = md5 + if not md5 and data_item: + self.md5 = os.path.basename(data_item.cache.relative) + if command not in self.ACCEPTED_COMMANDS: raise StateFileError('Args error: unknown command %s' % command) self.command = command @@ -182,7 +188,8 @@ class StateFile(object): StateFile.decode_path(json.get(StateFile.PARAM_STDERR)), StateFile.decode_path(json.get(StateFile.PARAM_CWD)), json.get(StateFile.PARAM_SHELL, False), - json.get(StateFile.PARAM_TARGET_METRICS, {})) + json.get(StateFile.PARAM_TARGET_METRICS, {}), + json.get(StateFile.PARAM_MD5, '')) @staticmethod def load(data_item, settings): @@ -213,7 +220,8 @@ class StateFile(object): self.PARAM_STDOUT: self.encode_path(self.stdout), self.PARAM_STDERR: self.encode_path(self.stderr), self.PARAM_SHELL: self.shell, - self.PARAM_TARGET_METRICS: self.target_metrics + self.PARAM_TARGET_METRICS: self.target_metrics, + self.PARAM_MD5: self.md5 } if self.locked: