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

Merge pull request #252 from efiop/checksum

Browse Source 
Use md5 checksum instead of commit-ids for cache files
This commit is contained in:
Ruslan Kuprieiev
2017-10-17 16:31:45 +03:00
committed by GitHub
parent b0e1d3374f db12448a6e
commit 6d03d9998b
10 changed files with 68 additions and 69 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:
tests/basic_env.py
+3 -2
View File
@@ -105,10 +105,11 @@ class DirHierarchyEnvironment(BasicEnvironment):
file_result = os.path.join('data', data_file)
state_result = os.path.join(ConfigI.CONFIG_DIR, ConfigI.STATE_DIR, data_file + DataItem.STATE_FILE_SUFFIX)
self.create_content_file(state_result, 'state content')
dummy_md5 = os.path.basename(data_file) + DataItem.CACHE_FILE_SEP + self._commit
self.create_content_file(state_result, '{"Md5" : "' + dummy_md5 + '", "Command" : "run", "Cwd" : "dir"}')
if cache_file:
cache_result = os.path.join(ConfigI.CONFIG_DIR, ConfigI.CACHE_DIR, data_file + DataItem.CACHE_FILE_SEP + self._commit)
cache_result = os.path.join(ConfigI.CONFIG_DIR, ConfigI.CACHE_DIR, dummy_md5)
self.create_content_file(cache_result, content)
relevant_dir = self.relevant_dir(data_file)
tests/test_data_file_obj.py
+9 -9
View File
@@ -16,7 +16,8 @@ class TestDataFileObjBasic(BasicEnvironment):
config = ConfigI('data')
file = os.path.join('data', 'file.txt')
self._data_path = DataItem(file, git, config)
self.cache_file_full_path = os.path.join(self._test_git_dir, ConfigI.CONFIG_DIR, ConfigI.CACHE_DIR, 'file.txt_ad45ba8')
self._data_path = DataItem(file, git, config, cache_file=self.cache_file_full_path)
pass
def test_data_file(self):
@@ -24,8 +25,7 @@ class TestDataFileObjBasic(BasicEnvironment):
self.assertEqual(self._data_path.data.abs, data_file_full_path)
def test_cache_file(self):
cache_file_full_path = os.path.join(self._test_git_dir, ConfigI.CONFIG_DIR, ConfigI.CACHE_DIR, 'file.txt_ad45ba8')
self.assertEqual(self._data_path.cache.abs, cache_file_full_path)
self.assertEqual(self._data_path.cache.abs, self.cache_file_full_path)
def test_state_file(self):
state_file_full_path = os.path.join(self._test_git_dir, ConfigI.CONFIG_DIR, ConfigI.STATE_DIR, 'file.txt.state')
@@ -156,7 +156,8 @@ class TestDataPathInDataDir(BasicEnvironment):
self.path_factory = PathFactory(git, config)
deep_path = os.path.join(self.data_dir, 'dir1', 'd2', 'file.txt')
self.data_path = self.path_factory.data_item(deep_path)
self.cache_file = os.path.join(self._test_git_dir, ConfigI.CONFIG_DIR, ConfigI.CACHE_DIR, 'dir1', 'd2', 'file.txt_eeeff8f')
self.data_path = self.path_factory.data_item(deep_path, cache_file=self.cache_file)
pass
def test_data_file(self):
@@ -164,8 +165,7 @@ class TestDataPathInDataDir(BasicEnvironment):
self.assertEqual(self.data_path.data.abs, target)
def test_cache_file(self):
target = os.path.join(self._test_git_dir, ConfigI.CONFIG_DIR, ConfigI.CACHE_DIR, 'dir1', 'd2', 'file.txt_eeeff8f')
self.assertEqual(self.data_path.cache.abs, target)
self.assertEqual(self.data_path.cache.abs, self.cache_file)
def test_state_file(self):
target = os.path.join(self._test_git_dir, ConfigI.CONFIG_DIR, ConfigI.STATE_DIR, 'dir1', 'd2', 'file.txt.state')
@@ -193,7 +193,8 @@ class TestDataFileObjLongPath(BasicEnvironment):
self._config = ConfigI('data')
self.path_factory = PathFactory(self._git, self._config)
self.data_path = self.path_factory.data_item(os.path.join('..', '..', 'data', 'file1.txt'))
self.cache_file = os.path.join(self._test_git_dir, ConfigI.CONFIG_DIR, ConfigI.CACHE_DIR, 'file1.txt_123ed8')
self.data_path = self.path_factory.data_item(os.path.join('..', '..', 'data', 'file1.txt'), cache_file=self.cache_file)
pass
def test_data_file(self):
@@ -201,8 +202,7 @@ class TestDataFileObjLongPath(BasicEnvironment):
self.assertEqual(self.data_path.data.abs, target)
def test_cache_file(self):
target = os.path.join(self._test_git_dir, ConfigI.CONFIG_DIR, ConfigI.CACHE_DIR, 'file1.txt_123ed8')
self.assertEqual(self.data_path.cache.abs, target)
self.assertEqual(self.data_path.cache.abs, self.cache_file)
def test_state_file(self):
target = os.path.join(self._test_git_dir, ConfigI.CONFIG_DIR, ConfigI.STATE_DIR, 'file1.txt.state')