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

Merge pull request #338 from efiop/master

Browse Source 
State(stage) files design changes
This commit is contained in:
Ruslan Kuprieiev
2017-12-07 23:18:24 +02:00
committed by GitHub
parent 2ee71f8687 927d58cfad
commit 84789d334d
12 changed files with 181 additions and 209 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/cli.py
+4
View File
@@ -113,6 +113,10 @@ def parse_args(argv=None):
action='store_true',
default=False,
help='Lock data item - disable reproduction.')
run_parser.add_argument('-f',
'--file',
default='stage.dvc',
help='Specify name of the state file')
run_parser.add_argument(
'command',
nargs=argparse.REMAINDER,
dvc/command/add.py
+4 -4
View File
@@ -3,6 +3,7 @@ import os
from dvc.command.common.base import CmdBase
from dvc.logger import Logger
from dvc.state_file import StateFile
from dvc.path.data_item import DataItem
class CmdAdd(CmdBase):
@@ -39,13 +40,12 @@ class CmdAdd(CmdBase):
for data_item in targets:
Logger.debug('Creating state file for {}'.format(data_item.data.relative))
state_file = StateFile(data_item=data_item,
state_file = StateFile(fname=data_item.data.relative + StateFile.STATE_FILE_SUFFIX,
cmd=None,
out=[data_item.data.dvc],
out=StateFile.parse_deps_state(self.settings, [data_item.data.relative], currdir=data_item.data.dirname),
out_git=[],
deps=[],
locked=True,
md5=os.path.basename(data_item.cache.relative))
locked=True)
state_file.save()
Logger.debug('State file "{}" was created'.format(data_item.state.relative))
dvc/command/checkout.py
+2 -26
View File
@@ -35,30 +35,6 @@ class CmdCheckout(CmdBase):
Logger.info('Checkout \'{}\''.format(item.data.relative))
def run(self):
try:
states = self.git.state_files_for_previous_commit()
prev_items = self.settings.path_factory.data_items_from_states(states, existing=False)
except Exception as ex:
Logger.error(u'Unable to get data files from previous commit: {}'.format(ex))
return 1
curr_items = self.settings.path_factory.all_existing_data_items()
self.checkout(curr_items)
self.remove_files(list(set(prev_items) - set(curr_items)))
items = self.settings.path_factory.all_existing_data_items()
self.checkout(items)
return 0
@staticmethod
def remove_files(removed_items_set):
for item in removed_items_set:
path = item.data.relative
if not os.path.exists(path):
Logger.debug(u'File \'{}\' does not exist'.format(path))
else:
Logger.info(u'Remove \'{}\''.format(path))
os.remove(path)
dir = os.path.dirname(path)
if not os.listdir(dir):
Logger.info(u'Remove directory \'{}\''.format(dir))
os.removedirs(dir)
pass
dvc/command/gc.py
+6 -32
View File
@@ -1,49 +1,23 @@
import os
from dvc.command.common.traverse import Traverse
from dvc.command.common.base import CmdBase
from dvc.logger import Logger
from dvc.path.data_item import DataItemError
from dvc.config import ConfigI
from dvc.state_file import StateFile
class CmdGC(Traverse):
class CmdGC(CmdBase):
def __init__(self, settings):
super(CmdGC, self).__init__(settings, "garbage collect", do_not_start_from_root=False)
self.clist = []
super(CmdGC, self).__init__(settings)
def run(self):
if not self._traverse(self.git.git_dir_abs):
Logger.error('Failed to collect used cache')
return 1
clist = [str(x) for x in StateFile.find_all_cache_files(self.git)]
for cache in os.listdir(ConfigI.CACHE_DIR):
fname = os.path.join(ConfigI.CACHE_DIR, cache)
if fname in self.clist:
if os.path.basename(fname) in clist:
continue
os.remove(fname)
self._remove_cloud_cache(self.settings.path_factory.data_item('.empty', fname))
Logger.info('Cache \'{}\' was removed'.format(fname))
return 0
def process_file(self, target):
Logger.debug(u'[Cmd-GC] GC file {}.'.format(target))
try:
data_item = self.settings.path_factory.existing_data_item(target)
except DataItemError:
return
self.clist.append(data_item.cache.relative)
Logger.debug(u'[Cmd-GC] GC data item {}. Success.'.format(data_item.data.relative))
def is_recursive(self):
return self.parsed_args.recursive
@property
def no_git_actions(self):
return True
@staticmethod
def not_committed_changes_warning():
pass
dvc/command/merge.py
+17 -32
View File
@@ -14,50 +14,35 @@ class CmdMerge(CmdBase):
super(CmdMerge, self).__init__(settings)
def print_info(self, targets):
for item in targets:
for fname in targets:
Logger.info('Restored original data after merge:')
Logger.info(' {}'.format(item.data.relative))
def collect_data(self):
dlist = []
flist = self.git.get_last_merge_changed_files()
for fname in flist:
if fname.endswith(DataItem.STATE_FILE_SUFFIX):
data = fname[:-len(DataItem.STATE_FILE_SUFFIX)]
dlist.append(data)
return dlist
Logger.info(' {}'.format(fname))
def collect_targets(self):
targets = []
flist = self.collect_data()
items = self.settings.path_factory.to_data_items(flist)[0]
for item in items:
command = StateFile.load(item)
for fname in self.git.get_last_merge_changed_files():
if not StateFile._is_state_file(fname):
continue
if not command.cmd and command.locked:
targets.append(item)
state = StateFile.load(fname)
if not state.cmd and state.locked:
targets.append(fname)
return targets
def checkout_targets(self, targets):
data = []
for item in targets:
prev_state = StateFile.loads(self.git.get_file_content_before_last_merge(item.state.relative))
curr_state = StateFile.load(item)
items = []
for fname in targets:
self.git.checkout_file_before_last_merge(fname)
state = StateFile.load(fname)
for out in state.out:
item = self.settings.path_factory.data_item(os.path.join(state.cwd, out))
items.append(item)
state = StateFile(data_item=item,
cmd=curr_state.cmd,
out=curr_state.out,
out_git=curr_state.out_git,
locked=curr_state.locked,
deps=curr_state.deps,
md5=prev_state.md5)
state.save()
CmdCheckout.checkout(items)
CmdCheckout.checkout([item])
msg = 'DVC merge files: {}'.format(' '.join(data))
msg = 'DVC merge files: {}'.format(' '.join(targets))
self.commit_if_needed(msg)
def run(self):
dvc/command/repro.py
+8 -10
View File
@@ -88,7 +88,7 @@ class ReproChange(object):
self._recursive = recursive
self._force = force
self.command = StateFile.load(data_item)
self.command = StateFile.find(data_item)
if not self.command.cmd and not self.command.locked:
msg = 'Error: data item "{}" is not locked, but has no command for reproduction'
@@ -105,15 +105,15 @@ class ReproChange(object):
return self._cmd_obj
def remove_output_files(self):
for output_dvc in self.command.out:
Logger.debug('Removing output file {} before reproduction.'.format(output_dvc))
for out in self.command.out:
path = os.path.join(self.command.cwd, out)
Logger.debug('Removing output file {} before reproduction.'.format(path))
try:
data_item = self.cmd_obj.settings.path_factory.data_item_from_dvc_path(output_dvc)
data_item = self.cmd_obj.settings.path_factory.data_item(path)
os.remove(data_item.data.relative)
except Exception as ex:
msg = 'Data item {} cannot be removed before reproduction: {}'
Logger.debug(msg.format(output_dvc, ex))
Logger.debug(msg.format(path, ex))
def reproduce_run(self):
Logger.info('Reproducing run command for data item {}. Args: {}'.format(
@@ -169,10 +169,8 @@ class ReproChange(object):
def reproduce_deps(self, data_item_dvc, recursive):
result = False
for dep in self.command.deps:
path = dep[StateFile.PARAM_PATH]
md5 = dep[StateFile.PARAM_MD5]
for name,md5 in self.command.deps.items():
path = os.path.join(self.command.cwd, name)
if self.reproduce_dep(path, md5, recursive):
result = True
dvc/command/run.py
+26 -37
View File
@@ -19,51 +19,40 @@ class CmdRun(CmdBase):
def run(self):
cmd = ' '.join(self.parsed_args.command)
state = StateFile(data_item=None,
cmd=cmd,
out=self.parsed_args.out,
out_git=self.parsed_args.out_git,
deps=self.parsed_args.deps,
locked=self.parsed_args.lock,
md5=None)
if os.path.isfile(self.parsed_args.file):
Logger.error("Stage file {} already exists".format(self.parsed_args.file))
return 1
state = StateFile(fname=self.parsed_args.file,
cmd=cmd,
out=self.parsed_args.out,
out_git=self.parsed_args.out_git,
deps=self.parsed_args.deps,
locked=self.parsed_args.lock)
self.run_command(self.settings, state)
return self.commit_if_needed('DVC run: {}'.format(state.cmd))
@staticmethod
def run_command(settings, state):
Executor.exec_cmd_only_success(state.cmd, shell=True)
Executor.exec_cmd_only_success(state.cmd, cwd=state.cwd, shell=True)
CmdRun.apply_to_files(state.out, state, CmdRun._create_cache_and_state_files, settings)
CmdRun.apply_to_files(state.out_git, state, CmdRun._create_state_file, settings)
CmdRun.move_output_to_cache(settings, state)
CmdRun.update_state_file(settings, state)
@staticmethod
def apply_to_files(files, state, func, settings):
items = settings.path_factory.to_data_items(files)[0]
[func(i, state, settings) for i in items]
def update_state_file(settings, state):
Logger.debug('Update state file "{}"'.format(state.path))
state.out = StateFile.parse_deps_state(settings, state.out)
state.out_git = StateFile.parse_deps_state(settings, state.out_git)
state.deps = StateFile.parse_deps_state(settings, state.deps)
state.save()
@staticmethod
def _create_cache_and_state_files(data_item, state, settings):
Logger.debug('Move output file "{}" to cache dir "{}" and create a hardlink'.format(
data_item.data.relative, data_item.cache_dir_abs))
data_item.move_data_to_cache()
return CmdRun._create_state_file(data_item, state, settings, cache_file_exists = True)
@staticmethod
def _create_state_file(data_item, state, settings, cache_file_exists=False):
Logger.debug('Create state file "{}"'.format(data_item.state.relative))
if cache_file_exists:
md5 = os.path.basename(data_item.cache.relative)
else:
md5 = file_md5(data_item.data.relative)[0]
state_file = StateFile(data_item,
state.cmd,
state.out,
state.out_git,
locked=state.locked,
deps=StateFile.parse_deps_state(settings, state.deps),
md5=md5)
state_file.save()
return state_file
def move_output_to_cache(settings, state):
items = settings.path_factory.to_data_items(state.out)[0]
for item in items:
Logger.debug('Move output file "{}" to cache dir "{}" and create a hardlink'.format(
item.data.relative, item.cache_dir_abs))
item.move_data_to_cache()
dvc/path/data_item.py
+2 -4
View File
@@ -33,7 +33,6 @@ class NotInGitDirError(DataDirError):
class DataItem(object):
STATE_FILE_SUFFIX = '.dvc'
LOCAL_STATE_FILE_PREFIX = '.'
LOCAL_STATE_FILE_SUFFIX = '.dvc_local_state'
CACHE_FILE_SEP = '_'
@@ -78,7 +77,7 @@ class DataItem(object):
@cached_property
def state(self):
return self._state('', self.STATE_FILE_SUFFIX)
return Path(StateFile.find(self).path, self._git)
@cached_property
def cache_dir_abs(self):
@@ -99,7 +98,7 @@ class DataItem(object):
if self._cache_file:
file_name = os.path.relpath(os.path.realpath(self._cache_file), cache_dir)
else:
file_name = StateFile.load(self).md5
file_name = str(StateFile.find_md5(self))
cache_file = os.path.join(cache_dir, file_name)
return Path(cache_file, self._git)
@@ -112,5 +111,4 @@ class DataItem(object):
System.hardlink(self.data.relative, self.cache.relative)
os.chmod(self.data.relative, stat.S_IREAD)
StateFile(self, md5=md5).save()
LocalStateFile(self).save()
dvc/path/factory.py
+5 -25
View File
@@ -6,6 +6,7 @@ from dvc.path.data_item import DataItem, DataItemError, DataDirError
from dvc.path.path import Path
from dvc.path.stated_data_item import StatedDataItem
from dvc.system import System
from dvc.state_file import StateFile
class PathFactory(object):
@@ -31,7 +32,8 @@ class PathFactory(object):
return self.data_item(path.relative)
def is_data_item(self, fname):
return os.path.isfile(fname + DataItem.STATE_FILE_SUFFIX)
item = DataItem(fname, self._git, self._config)
return StateFile.find(item) != None
def existing_data_item(self, fname):
if not self.is_data_item(fname):
@@ -53,27 +55,5 @@ class PathFactory(object):
return result, externally_created_files
def all_existing_data_items(self, subdir='.', cache_exists=True):
states = []
for root, dirs, files in os.walk(os.path.join(self._git.git_dir_abs, subdir)):
for fname in files:
path = os.path.join(root, fname)
if not fname.endswith(DataItem.STATE_FILE_SUFFIX):
continue
states.append(path)
data_items = self.data_items_from_states(states)
if cache_exists:
data_items = filter(lambda i: os.path.exists(i.state.relative), data_items)
return data_items
def data_items_from_states(self, states, existing=True):
return [self.data_item_from_dvc_path(self.state_path_to_dvc_path(s), existing)
for s in states]
@staticmethod
def state_path_to_dvc_path(state):
return re.sub(DataItem.STATE_FILE_SUFFIX + '$', '', state)
files = StateFile.find_all_data_files(self._git, subdir)
return self.to_data_items(files)[0]
dvc/state_file.py
+102 -34
View File
@@ -3,6 +3,7 @@ import yaml
from dvc.exceptions import DvcException
from dvc.data_cloud import file_md5
from dvc.path.path import Path
class StateFileError(DvcException):
@@ -24,68 +25,140 @@ class StateFileBase(object):
class StateFile(StateFileBase):
MAGIC = 'DVC-State'
VERSION = '0.1'
STATE_FILE_SUFFIX = '.dvc'
PARAM_CMD = 'cmd'
PARAM_DEPS = 'deps'
PARAM_PATH = 'path'
PARAM_MD5 = 'md5'
PARAM_LOCKED = 'locked'
PARAM_OUT = 'out'
PARAM_OUT_GIT = 'out-git'
def __init__(self,
data_item=None,
fname=None,
cmd=None,
out=None,
out_git=None,
deps=None,
locked=None,
md5=None):
locked=None):
super(StateFile, self).__init__()
self.data_item = data_item
self.cmd = cmd
self.out = out
self.out_git = out_git
self.deps = deps
self.locked = locked
self.md5 = md5
self.path = os.path.abspath(fname) if fname else None
self.cwd = os.path.dirname(self.path) if self.path else None
@staticmethod
def parse_deps_state(settings, deps):
state = []
def parse_deps_state(settings, deps, currdir=None):
state = {}
for dep in deps:
if isinstance(dep, dict):
dep = dep[StateFile.PARAM_PATH]
if settings.path_factory.is_data_item(dep):
item = settings.path_factory.data_item(dep)
state.append({StateFile.PARAM_PATH: item.data.dvc,
StateFile.PARAM_MD5: StateFile.load(item).md5})
md5 = StateFile.find_md5(item)
else:
state.append({StateFile.PARAM_PATH: dep,
StateFile.PARAM_MD5: file_md5(os.path.join(settings.git.git_dir_abs, dep))[0]})
md5 = file_md5(os.path.join(settings.git.git_dir_abs, dep))[0]
if currdir:
name = os.path.relpath(dep, currdir)
else:
name = dep
state[name] = md5
return state
@property
def file(self):
return self.data_item.state.relative
@staticmethod
def loadd(data):
return StateFile(data_item=None,
def loadd(data, fname=None):
return StateFile(fname=fname,
cmd=data.get(StateFile.PARAM_CMD, None),
out=data.get(StateFile.PARAM_OUT, None),
out_git=data.get(StateFile.PARAM_OUT_GIT, None),
deps=data.get(StateFile.PARAM_DEPS, []),
md5=data.get(StateFile.PARAM_MD5, []),
locked=data.get(StateFile.PARAM_LOCKED, None))
@staticmethod
def load(data_item):
return StateFile._load(data_item.state.relative, StateFile)
def load(fname):
return StateFile._load(fname, StateFile, fname)
@staticmethod
def _is_state_file(path):
return path.endswith(StateFile.STATE_FILE_SUFFIX) and os.path.isfile(path)
@staticmethod
def _find_state(fname, dname):
name = os.path.relpath(fname, dname)
for entry in os.listdir(dname):
state_file = os.path.join(dname, entry)
if not StateFile._is_state_file(state_file):
# if not state_file.endswith(StateFile.STATE_FILE_SUFFIX) or not os.path.isfile(state_file):
continue
state = StateFile.load(os.path.join(dname, state_file))
if name in state.out:
return state
return None
@staticmethod
def _find(name, start, finish):
dname = start
fname = name
while True:
state = StateFile._find_state(fname, dname)
if state:
return state
if dname == finish:
break
dname = os.path.dirname(dname)
return None
@staticmethod
def find(data_item):
return StateFile._find(data_item.data.abs, data_item.data.dirname, data_item._git.git_dir_abs)
@staticmethod
def find_all_state_files(git, subdir='.'):
states = []
for root, dirs, files in os.walk(os.path.join(git.git_dir_abs, subdir)):
for fname in files:
path = os.path.join(root, fname)
if not StateFile._is_state_file(path):
continue
states.append(path)
return states
@staticmethod
def find_all_states(git, subdir='.'):
state_files = StateFile.find_all_state_files(git, subdir)
return [StateFile.load(state_file) for state_file in state_files]
@staticmethod
def find_all_data_files(git, subdir='.'):
states = StateFile.find_all_states(git, subdir)
files = []
for state in states:
for out in state.out:
files.append(os.path.join(state.cwd, out))
return files
@staticmethod
def find_all_cache_files(git, subdir='.'):
states = StateFile.find_all_states(git, subdir)
cache_files = []
for state in states:
for out,md5 in state.out.items():
cache_files.append(md5)
return cache_files
@staticmethod
def find_md5(data_item):
state = StateFile.find(data_item)
name = os.path.relpath(data_item.data.abs, os.path.dirname(state.path))
return state.out[name]
@staticmethod
def loads(content):
@@ -98,15 +171,10 @@ class StateFile(StateFileBase):
self.PARAM_OUT: self.out,
self.PARAM_OUT_GIT: self.out_git,
self.PARAM_DEPS: self.deps,
self.PARAM_LOCKED: self.locked,
self.PARAM_MD5: self.md5
self.PARAM_LOCKED: self.locked
}
file_dir = os.path.dirname(self.file)
if file_dir != '' and not os.path.isdir(file_dir):
os.makedirs(file_dir)
self._save(self.file, res)
self._save(self.path, res)
class LocalStateFile(StateFileBase):
dvc/utils.py
+1 -1
View File
@@ -79,7 +79,7 @@ def map_progress(func, targets, n_threads):
try:
ret = pool.map(func, targets)
except Exception as exc:
Logger.error('Unexpected exception while processing targets: {}'.format(exc))
Logger.error('Unexpected exception while processing targets: {}'.format(exc), exc_info=True)
finally:
progress.finish()
functest/test-repro-code.sh
+4 -4
View File
@@ -10,7 +10,7 @@ function test_generic() {
dvc_create_repo
dvc_info 'Copy foo into foo1'
dvc run -d code/code.sh -d data/foo -o data/foo1 bash code/code.sh data/foo data/foo1
dvc run -f copy_foo_foo1.dvc -d code/code.sh -d data/foo -o data/foo1 bash code/code.sh data/foo data/foo1
dvc_info 'Modify code'
echo " " >> code/code.sh
@@ -52,9 +52,9 @@ function test_partial() {
git commit -m "copy code"
dvc_info "Create repro chain foo -> foo1 -> foo2 -> foo3"
dvc run -d code/code1.sh -d data/foo -o data/foo1 cp data/foo data/foo1
dvc run -d code/code2.sh -d data/foo1 -o data/foo2 cp data/foo1 data/foo2
dvc run -d code/code3.sh -d data/foo2 -o data/foo3 cp data/foo2 data/foo3
dvc run -f copy_foo_foo1.dvc -d code/code1.sh -d data/foo -o data/foo1 cp data/foo data/foo1
dvc run -f copy_foo1_foo2.dvc -d code/code2.sh -d data/foo1 -o data/foo2 cp data/foo1 data/foo2
dvc run -f copy_foo2_foo3.dvc -d code/code3.sh -d data/foo2 -o data/foo3 cp data/foo2 data/foo3
dvc_info "Save original timestamps"
FOO_TS=$(dvc_timestamp data/foo)