Skip to content

Commit

Permalink
Merge pull request #47 from wxtim/cylc-set-task.output-order
Browse files Browse the repository at this point in the history
Make set trigger outputs in order
  • Loading branch information
hjoliver authored Mar 11, 2024
2 parents 077075c + a286483 commit b858d06
Show file tree
Hide file tree
Showing 3 changed files with 48 additions and 0 deletions.
15 changes: 15 additions & 0 deletions cylc/flow/task_outputs.py
Original file line number Diff line number Diff line change
Expand Up @@ -299,3 +299,18 @@ def msg_sort_key(item):
except ValueError:
ind = 999
return (ind, item[_MESSAGE] or '')

@staticmethod
def output_sort_key(item):
"""Compare by output order.
Examples:
>>> this = TaskOutputs.output_sort_key
>>> sorted(['finished', 'started', 'custom'], key=this)
['started', 'custom', 'finished']
"""
if item in TASK_OUTPUTS:
return TASK_OUTPUTS.index(item)
# Sort custom outputs after started.
return TASK_OUTPUTS.index(TASK_OUTPUT_STARTED) + .5
1 change: 1 addition & 0 deletions cylc/flow/task_pool.py
Original file line number Diff line number Diff line change
Expand Up @@ -1820,6 +1820,7 @@ def _set_outputs_itask(
itask.point, itask.tdef, outputs)

changed = False
outputs = sorted(outputs, key=itask.state.outputs.output_sort_key)
for output in outputs:
if itask.state.outputs.is_completed(output):
LOG.info(f"output {itask.identity}:{output} completed already")
Expand Down
32 changes: 32 additions & 0 deletions tests/integration/test_task_pool.py
Original file line number Diff line number Diff line change
Expand Up @@ -1499,6 +1499,38 @@ async def test_set_outputs_live(
)


async def test_set_outputs_live2(
flow,
scheduler,
start,
log_filter,
):
"""Assert that optional outputs are satisfied before completion
outputs to prevent incomplete task warnings.
"""
id_ = flow(
{
'scheduler': {'allow implicit tasks': 'True'},
'scheduling': {'graph': {
'R1': """
foo:a => apple
foo:b => boat
"""}},
'runtime': {'foo': {'outputs': {
'a': 'xylophone',
'b': 'yacht'}}}
}
)
schd = scheduler(id_)

async with start(schd) as log:
schd.pool.set_prereqs_and_outputs(["1/foo"], None, None, ['all'])
assert not log_filter(
log,
contains="did not complete required outputs: ['a', 'b']"
)


async def test_set_outputs_future(
flow,
scheduler,
Expand Down

0 comments on commit b858d06

Please sign in to comment.