Cylc-flow: cylc clean and numbered/named run dirs

[MetRonnie]

Say you have a workflow foo with numbered run dirs, i.e. ~/cylc-run/foo/runN/.

Currently if you do cylc clean foo it will just remove ~/cylc-run/foo/, without removing symlink dir targets, or cleaning on remote install targets. (I think the rationale behind this is to be able to clean any partially cleaned dirs.)

The same problem exists for hiearchical workflow names, e.g. workflow alpha/bravo and you do cylc clean alpha. (Named/numbered run dirs is basically a special case of hiearchical workflow names)

How to address this?

1. Raise an error if cylc clean REG is run and ~/cylc-run/REG is not a run dir (i.e. it does not contain a .service dir?). Suggested by @dpmatthews. Perhaps include a --force option to simply remove ~/cylc-run/REG?
2. Clean each run dir in turn (~/cylc-run/REG/run1, ~/cylc-run/REG/run2 etc) and finally remove ~/cylc-run/REG (which would be in line with #3895¹). However this still runs the risk of a user accidentally deleting more than they meant to.
3. Do (1) by default, but include an --all option to do (2).

---

¹ Except whereas cylc play foo would be the same as cylc play foo/runN, cylc clean foo would be the same as cylc clean foo/run1 && cylc clean foo/run2 ...

[hjoliver]

It'd be good to be able to clean multiple runN-dirs at once, so I'd be fine with option 2 or 3. Probably 3, for safety, I guess.

[oliver-sanders]

Universal ID gives us the glob-like patterns e.g:

# stop one flow
cylc stop foo
# stop all running flows
cylc stop '*'
# stop all flows registered under bar/
cylc stop 'bar/*'

Once implemented that functionality will trickle down to all subcommands (i.e. they will get reinvoked for each workflow) so for the following example:

~/cylc-run/
   one/
       run1/
       run2/
       runN/

The cylc clean command would work like this:

$ cylc clean one
deleting one/run1
$ cylc clean one/run1
deleting one/run1
$ cylc clean one/*
deleting one/run1
deleting one/run2

This behaviour should be standard to all commands and implemented at a higher level than any one command.

[MetRonnie]

Is #3931 going to close this then?

[hjoliver]

Wondering if we need an interim solution for cylc clean though, to tide us over.

[oliver-sanders]

Should be relatively simple to implement (famous last words).

There's a bit of centralisation to do as part of the "implicit run dirs" work. Once that's done it should be fairly simple to do the rest:

• The tokeniser has already been written, PR up.
• Next, plug the tokeniser output into the scan interface for workflow listing.
• Finally, invoke the commands one at a time in a for loop (can improve this later).

[MetRonnie]

The cylc clean command would work like this:

$ cylc clean one
deleting one/run1
$ cylc clean one/run1
deleting one/run1
$ cylc clean one/*
deleting one/run1
deleting one/run2

This behaviour should be standard to all commands and implemented at a higher level than any one command.

Can we go over this example in more detail? Say there's a workflow "base" run dir that looks like this

~/cylc-run/one
|-- _cylc-install
|-- run1
|    `-- # etc
|-- run2
|    `-- # etc
|-- run3
|    `-- # etc
`-- runN -> ./run3

As I understand it, cylc play, cylc stop etc will behave like this:

$ cylc stop one
Stopping one/run3
$ cylc play one
Restarting one/run3

And this actually the inferred run name behaviour #3895, which is actually kind of separate to the global identifier, right?

But cylc clean needs its own, distinct behaviour for inferring run names. But what is that behaviour?

$ cylc clean one

Is it to:

1. Delete ~/cylc-run/one/run1 only?
2. Delete ~/cylc-run/one/run1 and ~/cylc-run/one/run2 (everything apart from the most recent one)?

[oliver-sanders]

Good point, we wouldn't want cylc clean one to default to the most recent run of one. Consistent behaviour, but fundamentally unhelpful.

Ok, how about this, don't infer run names for cylc clean.

In the event a users tries to clean a base registration (e.g. fails to specify the run name) use cylc scan --name '<name>/*' -t name --states=all to list workflows contained within this base reg:

$ cylc install one
INSTALLED: ~/cylc-src/one -> one/run1
$ cylc install one
INSTALLED: ~/cylc-src/one -> one/run2
$ cylc install one
INSTALLED: ~/cylc-src/one -> one/run3
$ cylc clean one
ERROR: One contains multiple workflows:
* one/run1
* one/run2
* one/run3
$ cylc clean one/run1
DELETING: one/run1
$ cylc clean one/*
DELETING: one/run2
DELETING: one/run3

Note: The glob expansion behaviour would come from the universal identifier.

[MetRonnie]

In the event a users tries to clean a base registration (e.g. fails to specify the run name) use cylc scan --name '<name>/*' -t name --states=all to list workflows contained within this base reg:

I'm trying to incorporate this into _clean_check()
https://github.com/cylc/cylc-flow/blob/6e287849e5a632855afdaa12f9725c224b01333a/cylc/flow/workflow_files.py#L603

but struggling due to the asyncio-ness of the scan code.

Aside from running subprocess.run(cylc_scan_cmd), do you have any pointers for how I can do this?

[oliver-sanders]

I think something along these lines should do it:

async def whatever(base_flow_path):
    ret = []
    async for flow in scan(scan_dir=base_flow_path, run_dir=base_flow_path):
        ret.append(flow)
    return ret

asyncio.run(whatever(base_flow_path))

There are some examples in the integration tests which create a dir of flows somewhere on the filesystem then use scan to run through them (and only them):

https://github.com/cylc/cylc-flow/blob/master/tests/integration/test_scan.py

There is also the potential to call the CLI logic using an Options object, however, that's not as nice and was only introduced for testing:

https://github.com/cylc/cylc-flow/blob/master/tests/integration/test_scan_api.py

[oliver-sanders]

The scan stuff is pretty flexible, you can add filters onto the pipe using | syntax. So you can filter for running workflows, etc.

Here's the pipe used by the UIS for example:

          self._scan_pipe = (    
              # all flows on the filesystem                                                                                                 
              scan(run_dir)    
              # stop here is the flow is stopped, else...    
              | is_active(True, filter_stop=False)    
              # extract info from the contact file    
              | contact_info    
              # only flows which are using the same api version    
              | api_version(f'=={API}')   

Note the filter_stop argument (which all filters can use). This stops the flow from being passed further through the pipe, but rather than filtering the item out (default behaviour) it yields it early.

So I think this pipe might be the one you are looking for, you can tell whether a flow is running or not by looking at the fields of the yielded dicts (might be the contact field or something like that):

          pipe = (    
              scan(run_dir)    
              # stop here is the flow is stopped, else...    
              | is_active(True, filter_stop=False)    

[MetRonnie]

One problem I had was trying to return early from the coroutine

async def contains_workflows(path) -> bool:
    async for flow in scan(path):
        return True
    return False

unhandled exception during asyncio.run() shutdown
task: <Task finished coro=<<async_generator_athrow without __name__>()>
       exception=RuntimeError("can't send non-None value to a just-started coroutine")>
RuntimeError: can't send non-None value to a just-started coroutine

Actually, listing the workflows would be more user-friendly so I've switched to that approach anyway, but out of curiosity, what is the matter with returning early like that?

[MetRonnie]

Another problem is that scan(run_dir) thinks run_dir/log/workflow is a workflow because of the log dir inside. I.e. it doesn't stop if the run_dir or scan_dir passed to it is a workflow run dir

I think I can fix this with

diff --git a/cylc/flow/network/scan.py b/cylc/flow/network/scan.py
@@ -145,7 +146,12 @@ async def scan(run_dir=None, scan_dir=None, max_depth=MAX_SCAN_DEPTH):
         return path, depth, contents

     # perform the first directory listing
-    for subdir in await scandir(scan_dir):
+    scan_dir_listing = await scandir(scan_dir)
+    if dir_is_flow(scan_dir_listing):
+        # If the scan_dir itself is a workflow run dir, yield nothing
+        return
+
+    for subdir in scan_dir_listing:
         if subdir.is_dir():
             running.append(
                 asyncio.create_task(

[oliver-sanders]

but out of curiosity, what is the matter with returning early like that?

I don't think the @pipe thinggy was designed to handle the abort case, perhaps try a break statement.

I think I can fix this with

Ah, ok, it's not expecting to be provided with a flow directory as a run directory, could pick the next level up, otherwise will have to patch the logic, should get it to return the regular workflow info.

[oliver-sanders]

what is the matter with returning early like that?

I've not been able to replicate your error (Python3.7), my test script:

import asyncio    

from cylc.flow.network.scan import scan    


async def foo():    
    async for flow in scan():    
        print('$', flow)    
        return True           
    return False              


asyncio.run(foo())