mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-10 15:28:34 +01:00
Peter Rowlands (변기호)
389efd7ae6
* experiments: refactor regular/checkpoint experiment separation * fix scm_checkout pull bug * experiments: return stages from executor.reproduce * since executor has to move to threadpool execution anyways, we no longer have to work around the pickle repro result limitation * refactor new/resume behavior * refactor checkpoints tests * use -f to force checkpoints reset/restart * checkpoints: use 'exp run' and 'exp res[ume]' commands * stage: all outs can be missing for checkpoint stages * regular outputs may not be written at the time a checkpoint is generated, but subsequent checkouts of that stage should not fail due to the missing outs * commit stage on checkpoint generation otherwise non-persist cached outs are lost * checkpoints: wait for stage proc to end after killing via terminate() * stage: fix issues with killing checkpoint stage * refactor signaling (use threading.Event instead of Condition) * fix windows bug (kill all child processes via taskkill) * experiments: refactor executor collection * tests: move checkpoint tests into separate module * refactor experiment tests to use stage fixture
120 lines
3.4 KiB
Python
120 lines
3.4 KiB
Python
import logging
|
|
from textwrap import dedent
|
|
|
|
import pytest
|
|
from funcy import first
|
|
|
|
from dvc.exceptions import DvcException
|
|
from dvc.repo.experiments import Experiments
|
|
|
|
CHECKPOINT_SCRIPT_FORMAT = dedent(
|
|
"""\
|
|
import os
|
|
import sys
|
|
import shutil
|
|
from time import sleep
|
|
|
|
from dvc.api import make_checkpoint
|
|
|
|
checkpoint_file = {}
|
|
checkpoint_iterations = int({})
|
|
if os.path.exists(checkpoint_file):
|
|
with open(checkpoint_file) as fobj:
|
|
try:
|
|
value = int(fobj.read())
|
|
except ValueError:
|
|
value = 0
|
|
else:
|
|
with open(checkpoint_file, "w"):
|
|
pass
|
|
value = 0
|
|
|
|
shutil.copyfile({}, {})
|
|
|
|
if os.getenv("DVC_CHECKPOINT"):
|
|
for _ in range(checkpoint_iterations):
|
|
value += 1
|
|
with open(checkpoint_file, "w") as fobj:
|
|
fobj.write(str(value))
|
|
make_checkpoint()
|
|
"""
|
|
)
|
|
CHECKPOINT_SCRIPT = CHECKPOINT_SCRIPT_FORMAT.format(
|
|
"sys.argv[1]", "sys.argv[2]", "sys.argv[3]", "sys.argv[4]"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def checkpoint_stage(tmp_dir, scm, dvc):
|
|
tmp_dir.gen("checkpoint.py", CHECKPOINT_SCRIPT)
|
|
tmp_dir.gen("params.yaml", "foo: 1")
|
|
stage = dvc.run(
|
|
cmd="python checkpoint.py foo 5 params.yaml metrics.yaml",
|
|
metrics_no_cache=["metrics.yaml"],
|
|
params=["foo"],
|
|
checkpoints=["foo"],
|
|
no_exec=True,
|
|
name="checkpoint-file",
|
|
)
|
|
scm.add(["dvc.yaml", "checkpoint.py", "params.yaml"])
|
|
scm.commit("init")
|
|
return stage
|
|
|
|
|
|
def test_new_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, mocker):
|
|
new_mock = mocker.spy(dvc.experiments, "new")
|
|
dvc.experiments.run(checkpoint_stage.addressing, params=["foo=2"])
|
|
|
|
new_mock.assert_called_once()
|
|
assert (tmp_dir / "foo").read_text() == "5"
|
|
assert (
|
|
tmp_dir / ".dvc" / "experiments" / "metrics.yaml"
|
|
).read_text().strip() == "foo: 2"
|
|
|
|
|
|
@pytest.mark.parametrize("last", [True, False])
|
|
def test_resume_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, mocker, last):
|
|
with pytest.raises(DvcException):
|
|
if last:
|
|
dvc.experiments.run(
|
|
checkpoint_stage.addressing,
|
|
checkpoint_resume=Experiments.LAST_CHECKPOINT,
|
|
)
|
|
else:
|
|
dvc.experiments.run(
|
|
checkpoint_stage.addressing, checkpoint_resume="foo"
|
|
)
|
|
|
|
results = dvc.experiments.run(
|
|
checkpoint_stage.addressing, params=["foo=2"]
|
|
)
|
|
if last:
|
|
exp_rev = Experiments.LAST_CHECKPOINT
|
|
else:
|
|
exp_rev = first(results)
|
|
|
|
dvc.experiments.run(checkpoint_stage.addressing, checkpoint_resume=exp_rev)
|
|
|
|
assert (tmp_dir / "foo").read_text() == "10"
|
|
assert (
|
|
tmp_dir / ".dvc" / "experiments" / "metrics.yaml"
|
|
).read_text().strip() == "foo: 2"
|
|
|
|
|
|
def test_reset_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, mocker, caplog):
|
|
dvc.experiments.run(checkpoint_stage.addressing)
|
|
scm.repo.git.reset(hard=True)
|
|
scm.repo.git.clean(force=True)
|
|
|
|
with caplog.at_level(logging.ERROR):
|
|
results = dvc.experiments.run(checkpoint_stage.addressing)
|
|
assert len(results) == 0
|
|
assert "already exists" in caplog.text
|
|
|
|
dvc.experiments.run(checkpoint_stage.addressing, force=True)
|
|
|
|
assert (tmp_dir / "foo").read_text() == "5"
|
|
assert (
|
|
tmp_dir / ".dvc" / "experiments" / "metrics.yaml"
|
|
).read_text().strip() == "foo: 1"
|