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

experiments: add checkpoint output type (#4845)

Browse Source 
* output: add checkpoint output type

* stage: handle checkpoint outs

* run/repro: handle checkpoint outs

* experiments: use checkpoint output type in tests

* tests: update run command tests
This commit is contained in:
Peter Rowlands (변기호)
2020-11-06 14:44:21 +09:00
committed by GitHub
parent 89bf7fce13
commit c85a46cdda
11 changed files with 97 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/command/run.py
+11 -1
View File
@@ -21,6 +21,7 @@ class CmdRun(CmdBase):
self.args.plots_no_cache,
self.args.outs_persist,
self.args.outs_persist_no_cache,
self.args.checkpoints,
self.args.params,
self.args.command,
]
@@ -45,12 +46,13 @@ class CmdRun(CmdBase):
params=self.args.params,
fname=self.args.file,
wdir=self.args.wdir,
no_exec=self.args.no_exec,
no_exec=(self.args.no_exec or bool(self.args.checkpoints)),
force=self.args.force,
run_cache=not self.args.no_run_cache,
no_commit=self.args.no_commit,
outs_persist=self.args.outs_persist,
outs_persist_no_cache=self.args.outs_persist_no_cache,
checkpoints=self.args.checkpoints,
always_changed=self.args.always_changed,
name=self.args.name,
single_stage=self.args.single_stage,
@@ -212,6 +214,14 @@ def add_parser(subparsers, parent_parser):
"removed upon repro (do not put into DVC cache).",
metavar="<filename>",
)
run_parser.add_argument(
"-c",
"--checkpoints",
action="append",
default=[],
help=argparse.SUPPRESS,
metavar="<filename>",
).complete = completion.FILE
run_parser.add_argument(
"--always-changed",
action="store_true",
dvc/output/__init__.py
+30 -3
View File
@@ -60,12 +60,20 @@ SCHEMA[BaseOutput.PARAM_CACHE] = bool
SCHEMA[BaseOutput.PARAM_METRIC] = BaseOutput.METRIC_SCHEMA
SCHEMA[BaseOutput.PARAM_PLOT] = bool
SCHEMA[BaseOutput.PARAM_PERSIST] = bool
SCHEMA[BaseOutput.PARAM_CHECKPOINT] = bool
SCHEMA[HashInfo.PARAM_SIZE] = int
SCHEMA[HashInfo.PARAM_NFILES] = int
def _get(
stage, p, info=None, cache=True, metric=False, plot=False, persist=False
stage,
p,
info=None,
cache=True,
metric=False,
plot=False,
persist=False,
checkpoint=False,
):
parsed = urlparse(p)
@@ -80,6 +88,7 @@ def _get(
metric=metric,
plot=plot,
persist=persist,
checkpoint=checkpoint,
)
for o in OUTS:
@@ -93,6 +102,7 @@ def _get(
metric=metric,
plot=plot,
persist=persist,
checkpoint=checkpoint,
)
return LocalOutput(
stage,
@@ -103,6 +113,7 @@ def _get(
metric=metric,
plot=plot,
persist=persist,
checkpoint=checkpoint,
)
@@ -114,6 +125,7 @@ def loadd_from(stage, d_list):
metric = d.pop(BaseOutput.PARAM_METRIC, False)
plot = d.pop(BaseOutput.PARAM_PLOT, False)
persist = d.pop(BaseOutput.PARAM_PERSIST, False)
checkpoint = d.pop(BaseOutput.PARAM_CHECKPOINT, False)
ret.append(
_get(
stage,
@@ -123,13 +135,20 @@ def loadd_from(stage, d_list):
metric=metric,
plot=plot,
persist=persist,
checkpoint=checkpoint,
)
)
return ret
def loads_from(
stage, s_list, use_cache=True, metric=False, plot=False, persist=False
stage,
s_list,
use_cache=True,
metric=False,
plot=False,
persist=False,
checkpoint=False,
):
return [
_get(
@@ -140,6 +159,7 @@ def loads_from(
metric=metric,
plot=plot,
persist=persist,
checkpoint=checkpoint,
)
for s in s_list
]
@@ -183,5 +203,12 @@ def load_from_pipeline(stage, s_list, typ="outs"):
from dvc.schema import PLOT_PROPS
plt_d, flags = _split_dict(flags, keys=PLOT_PROPS.keys())
extra = project(flags, ["cache", "persist"])
extra = project(
flags,
[
BaseOutput.PARAM_CACHE,
BaseOutput.PARAM_PERSIST,
BaseOutput.PARAM_CHECKPOINT,
],
)
yield _get(stage, path, {}, plot=plt_d or plot, metric=metric, **extra)
dvc/output/base.py
+7 -1
View File
@@ -61,6 +61,7 @@ class BaseOutput:
PARAM_PATH = "path"
PARAM_CACHE = "cache"
PARAM_CHECKPOINT = "checkpoint"
PARAM_METRIC = "metric"
PARAM_METRIC_TYPE = "type"
PARAM_METRIC_XPATH = "xpath"
@@ -100,6 +101,7 @@ class BaseOutput:
metric=False,
plot=False,
persist=False,
checkpoint=False,
):
self._validate_output_path(path, stage)
# This output (and dependency) objects have too many paths/urls
@@ -124,6 +126,7 @@ class BaseOutput:
self.metric = False if self.IS_DEPENDENCY else metric
self.plot = False if self.IS_DEPENDENCY else plot
self.persist = persist
self.checkpoint = checkpoint
self.path_info = self._parse_path(tree, path)
if self.use_cache and self.cache is None:
@@ -315,6 +318,9 @@ class BaseOutput:
if self.persist:
ret[self.PARAM_PERSIST] = self.persist
if self.checkpoint:
ret[self.PARAM_CHECKPOINT] = self.checkpoint
return ret
def verify_metric(self):
@@ -350,7 +356,7 @@ class BaseOutput:
**kwargs,
)
except CheckoutError:
if allow_missing:
if allow_missing or self.checkpoint:
return None
raise
dvc/repo/reproduce.py
+10 -2
View File
@@ -13,11 +13,19 @@ logger = logging.getLogger(__name__)
def _reproduce_stage(stage, **kwargs):
def _run_callback(repro_callback):
_dump_stage(stage)
logger.debug(f"{repro_callback} ([{stage}])")
repro_callback([stage])
checkpoint_func = kwargs.pop("checkpoint_func", None)
if checkpoint_func:
kwargs["checkpoint_func"] = partial(_run_callback, checkpoint_func)
if stage.is_checkpoint:
if checkpoint_func:
kwargs["checkpoint_func"] = partial(_run_callback, checkpoint_func)
else:
logger.warning(
"Checkpoint stages are not fully supported in 'dvc repro'. "
"Checkpoint stages should be reproduced with 'dvc exp run' "
"or 'dvc exp resume'."
)
if stage.frozen and not stage.is_import:
logger.warning(
dvc/repo/run.py
+1
View File
@@ -48,6 +48,7 @@ def _get_file_path(kwargs):
kwargs.get("plots_no_cache", []),
kwargs.get("outs_persist", []),
kwargs.get("outs_persist_no_cache", []),
kwargs.get("checkpoints", []),
)
)
dvc/schema.py
+5 -1
View File
@@ -34,7 +34,11 @@ LOCK_FILE_STAGE_SCHEMA = {
LOCKFILE_SCHEMA = {str: LOCK_FILE_STAGE_SCHEMA}
OUT_PSTAGE_DETAILED_SCHEMA = {
str: {BaseOutput.PARAM_CACHE: bool, BaseOutput.PARAM_PERSIST: bool}
str: {
BaseOutput.PARAM_CACHE: bool,
BaseOutput.PARAM_PERSIST: bool,
BaseOutput.PARAM_CHECKPOINT: bool,
}
}
PLOT_PROPS = {
dvc/stage/__init__.py
+13 -5
View File
@@ -216,6 +216,14 @@ class Stage(params.StageParams):
return isinstance(self.deps[0], dependency.RepoDependency)
@property
def is_checkpoint(self):
"""
A stage containing checkpoint outs is always considered as changed
since the checkpoint out is a circular dependency.
"""
return any(out.checkpoint for out in self.outs)
def changed_deps(self):
if self.frozen:
return False
@@ -229,7 +237,7 @@ class Stage(params.StageParams):
)
return True
if self.always_changed:
if self.always_changed or self.is_checkpoint:
return True
return self._changed_deps()
@@ -284,7 +292,7 @@ class Stage(params.StageParams):
def remove_outs(self, ignore_remove=False, force=False):
"""Used mainly for `dvc remove --outs` and :func:`Stage.reproduce`."""
for out in self.outs:
if out.persist and not force:
if (out.persist or out.checkpoint) and not force:
out.unprotect()
continue
@@ -402,7 +410,7 @@ class Stage(params.StageParams):
try:
out.save()
except OutputDoesNotExistError:
if not allow_missing:
if not (allow_missing or out.checkpoint):
raise
def ignore_outs(self):
@@ -432,7 +440,7 @@ class Stage(params.StageParams):
try:
out.commit()
except OutputDoesNotExistError:
if not allow_missing:
if not (allow_missing or out.checkpoint):
raise
except CacheLinkError:
link_failures.append(out.path_info)
@@ -527,7 +535,7 @@ class Stage(params.StageParams):
ret.append({"changed outs": outs_status})
def _status_always_changed(self, ret):
if self.is_callback or self.always_changed:
if self.is_callback or self.always_changed or self.is_checkpoint:
ret.append("always changed")
def _status_stage(self, ret):
dvc/stage/serialize.py
+3
View File
@@ -26,6 +26,7 @@ PARAM_CACHE = BaseOutput.PARAM_CACHE
PARAM_METRIC = BaseOutput.PARAM_METRIC
PARAM_PLOT = BaseOutput.PARAM_PLOT
PARAM_PERSIST = BaseOutput.PARAM_PERSIST
PARAM_CHECKPOINT = BaseOutput.PARAM_CHECKPOINT
DEFAULT_PARAMS_FILE = ParamsDependency.DEFAULT_PARAMS_FILE
@@ -37,6 +38,8 @@ sort_by_path = partial(sorted, key=attrgetter("def_path"))
def _get_flags(out):
if not out.use_cache:
yield PARAM_CACHE, False
if out.checkpoint:
yield PARAM_CHECKPOINT, True
if out.persist:
yield PARAM_PERSIST, True
if out.plot and isinstance(out.plot, dict):
dvc/stage/utils.py
+3
View File
@@ -53,6 +53,7 @@ def fill_stage_outputs(stage, **kwargs):
"plots",
"outs_no_cache",
"outs",
"checkpoints",
]
stage.outs = []
@@ -64,6 +65,7 @@ def fill_stage_outputs(stage, **kwargs):
persist="persist" in key,
metric="metrics" in key,
plot="plots" in key,
checkpoint="checkpoints" in key,
)
@@ -174,6 +176,7 @@ def compute_md5(stage):
stage.PARAM_FROZEN,
BaseOutput.PARAM_METRIC,
BaseOutput.PARAM_PERSIST,
BaseOutput.PARAM_CHECKPOINT,
HashInfo.PARAM_SIZE,
HashInfo.PARAM_NFILES,
],
tests/func/experiments/test_experiments.py
+9 -33
View File
@@ -320,19 +320,11 @@ def test_new_checkpoint(tmp_dir, scm, dvc, mocker):
cmd="python checkpoint.py foo 5 params.yaml metrics.yaml",
metrics_no_cache=["metrics.yaml"],
params=["foo"],
outs_persist=["foo"],
always_changed=True,
checkpoints=["foo"],
no_exec=True,
name="checkpoint-file",
)
scm.add(
[
"dvc.yaml",
"dvc.lock",
"checkpoint.py",
"params.yaml",
"metrics.yaml",
]
)
scm.add(["dvc.yaml", "checkpoint.py", "params.yaml"])
scm.commit("init")
new_mock = mocker.spy(dvc.experiments, "new")
@@ -353,19 +345,11 @@ def test_continue_checkpoint(tmp_dir, scm, dvc, mocker, last):
cmd="python checkpoint.py foo 5 params.yaml metrics.yaml",
metrics_no_cache=["metrics.yaml"],
params=["foo"],
outs_persist=["foo"],
always_changed=True,
checkpoints=["foo"],
no_exec=True,
name="checkpoint-file",
)
scm.add(
[
"dvc.yaml",
"dvc.lock",
"checkpoint.py",
"params.yaml",
"metrics.yaml",
]
)
scm.add(["dvc.yaml", "checkpoint.py", "params.yaml"])
scm.commit("init")
results = dvc.experiments.run(
@@ -395,19 +379,11 @@ def test_reset_checkpoint(tmp_dir, scm, dvc, mocker):
cmd="python checkpoint.py foo 5 params.yaml metrics.yaml",
metrics_no_cache=["metrics.yaml"],
params=["foo"],
outs_persist=["foo"],
always_changed=True,
checkpoints=["foo"],
no_exec=True,
name="checkpoint-file",
)
scm.add(
[
"dvc.yaml",
"dvc.lock",
"checkpoint.py",
"params.yaml",
"metrics.yaml",
]
)
scm.add(["dvc.yaml", "checkpoint.py", "params.yaml"])
scm.commit("init")
dvc.experiments.run(stage.addressing, checkpoint=True)
tests/unit/command/test_run.py
+5
View File
@@ -34,6 +34,8 @@ def test_run(mocker, dvc):
"outs-persist",
"--outs-persist-no-cache",
"outs-persist-no-cache",
"--checkpoints",
"checkpoints",
"--always-changed",
"--params",
"file:param1,param2",
@@ -60,6 +62,7 @@ def test_run(mocker, dvc):
plots_no_cache=["plots-no-cache"],
outs_persist=["outs-persist"],
outs_persist_no_cache=["outs-persist-no-cache"],
checkpoints=["checkpoints"],
params=["file:param1,param2", "param3"],
fname="file",
wdir="wdir",
@@ -90,6 +93,7 @@ def test_run_args_from_cli(mocker, dvc):
plots_no_cache=[],
outs_persist=[],
outs_persist_no_cache=[],
checkpoints=[],
params=[],
fname=None,
wdir=None,
@@ -120,6 +124,7 @@ def test_run_args_with_spaces(mocker, dvc):
plots_no_cache=[],
outs_persist=[],
outs_persist_no_cache=[],
checkpoints=[],
params=[],
fname=None,
wdir=None,