zincware / dask4dvc Goto Github PK

• add

  dask4dvc/dask4dvc/dvc_repro.py

  Line 161 in c44a98c

  tasks.setup_exp(dataclasses.asdict(entry))

  to the worker

• dask4dvc/dask4dvc/dvc_repro.py

  Lines 138 to 144 in c44a98c

  	future.add_done_callback(
  	functools.partial(
  	collect_and_cleanup,
  	entry_dict=dataclasses.asdict(entry),
  	infofile=infofile,
  	)
  	)

  might be slower than starting the new experiment? Maybe it doesn't work on slurm for some reason?

There might be an issue with stages that only create git tracked outputs like the hash form zn.nodes

EDIT: they are dvc outs but the directory does not exist prior to the stage running. If that is the case, maybe create the directory before dvc stage add so the gitignore can be git tracked.

If you run dask4dvc repro will it cleanup existing files first?

To avoid

failed with Could not acquire lock after 10 tries.

E.g. with the ´ipsuite.calculators.CP2KSingelPoint´ it seems the Node is only using a single core in a LocalCluster

By default, stdout/stderr is written to e.g. slumr-xxxx.out. It would be a good addition to show that output when running dask4dvc repro

Currently there are no tests for runs that fail

If you have many Nodes but most of them haven't changed you might only want to submit the Nodes that truely changed.
You could use e.g. dvc status to check if the Node, that would be submitted actually has changed outputs or not.

When using dask4dvc clone the git apply patch fails, claiming the patch would be corrupted.

Do not rely on a dedicated Cluster but allow starting a e.g. Slurm Cluster together with dask4dvc <cmd>

Currently there is no way of setting the resources required per Node. There are ways to define this for the client.submit via https://distributed.dask.org/en/stable/resources.html.

My Idea would be, to add a dask4dvc.yaml file (or alternatively the meta key in dvc.yaml https://distributed.dask.org/en/stable/resources.html) to define the resources that should be acquired for the respective Node.

node1:
  GPU: 1
  MEMORY: 100e9

node2:
  MEMORY: 16e9

node3:
  GPU: 3
  MEMORY: 128e9

we parse this file, if it exists and use it here:

Furthermore, we might also want to have a general section e.g. for https://jobqueue.dask.org/en/latest/generated/dask_jobqueue.SLURMCluster.html

cluster:
   slurm:
      project: "my-project"

Just for a small speed up, this would be nice to have.

iterative/dvc#9245

We could e.g. use some environmental variables or don't remove previous experiements, etc.

Have some way of running custom shell scripts as a setup before running the stage. Probably through dask4dvc.yaml

You can actually use the dvc graph which makes things even easier.

@znflow.nodify
def submit_zntrack_node(name, cls, *args):

    outs = subprocess.run(["dvc", "repro", "--dry", name], capture_output=True)
    if outs.stdout.decode().startswith(f"Stage '{name}' is cached - skipping run, checking out outputs"):
        dvc.cli.main(["checkout", name])
        return

    node = cls.from_rev(name=name, results=False)
    node.run()
    node.save(parameter=False)
    # TODO retry commit, because of lock
    for _ in range(10):
        code = dvc.cli.main(["commit", "-f", name])
        if code == 0:
            break

graph = znflow.DiGraph()
mapping = {}
for node_uuid in project.graph.reverse():
    node = project.graph.nodes[node_uuid]["value"]
    predecessors = list(project.graph.predecessors(node.uuid))
    if len(predecessors) == 0:
        with graph:
            node = submit_zntrack_node(node.name, type(node))
        mapping[node_uuid] = node
    else:
        with graph:
            node = submit_zntrack_node(node.name, type(node), *[mapping[predecessor] for predecessor in predecessors])
        mapping[node_uuid] = node

deployment = znflow.deployment.Deployment(graph=graph, client=client)
deployment.submit_graph()

E.g. for following custom outputs while the experiment is running printing the path can be helpful

see https://jobqueue.dask.org/en/latest/configuration.html#configuration-files

Add dask4dvc -d to detach. This can be useful, e.g. when running on a cluster.