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

cache: use checksums instead of commit_id

Browse Source 
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 <[email protected]>
This commit is contained in:
Ruslan Kuprieiev
2017-10-17 14:06:43 +03:00
parent b0e1d3374f
commit 8bb51f5efc
8 changed files with 56 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/command/import_file.py
+2 -3
View File
@@ -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))
dvc/command/init.py
+4 -5
View File
@@ -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=[],
dvc/command/run.py
+2 -2
View File
@@ -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))
dvc/data_cloud.py
+17 -34
View File
@@ -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
dvc/executor.py
+1 -1
View File
@@ -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
dvc/git_wrapper.py
+6 -3
View File
@@ -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)
dvc/path/data_item.py
+13 -7
View File
@@ -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)
dvc/state_file.py
+11 -3
View File
@@ -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: