@task
def task_one():
    return 1


@task
def task_two(results):
    logger = get_run_logger()
    <http://logger.info|logger.info>(results)
    return 2


@flow
def test_flow():
    results = {}

    t1_result = task_one()
    results["t1_result"] = t1_result
    t2_result = task_two(results, wait_for=[t1_result])
    results["t2_result"] = t2_result

appears to cause the flow to run indefinitely and i'm unclear why..

since i'm not actually using

t1_result

in

task_two

in the above example, i wouldn't expect it to be stuck. fully agree if i planned to use it, i would have to do something like:

def task_two(results):
    do_something_with(results["t1_result"].result())

raising it up to a more generic level, the case i'm trying to solve for is:

given a list of prefect tasks, execute each one while passing the results of all previously completed tasks to the next

so given that, an example would be:

@task def t1
@task def t2
@task def t3
...

@flow
def my_flow():
    tasks = [t1, t2, t3, t4]

    results = {}
    for task in tasks:
        task_result = task(results)
        results[task.name] = task_result

or, is the expectation that you can't do that, and you have to instead do

task_result = task(**results)

because prior task results (futures) need to be passed in as args/kwargs

Oh gosh

You might have made t2 depend on itself by adding its future to the dictionary you passed to it

Since the dictionary is mutable

I’m not entirely sure we can guard against this

You could nest a mutable object arbitrarily deeply in there

We can exclude the simple cycle of self-dependency though

unless there's a point where you ask the question "have all futures that need to resolve to run this task resolved?" in which case...you could validate that the future isn't related to the currently in progress task, and if it is then throw an error and exit? would need you to pre-process and pull out all futures arbitrarily deep though i'd guess

We can easily exclude the case where a task requires its own future, but we’d have to set the value in your object to

None

or something. I’d probably just throw an exception if we find that.

The issue is that the cycle could be deeper, e.g. task C relies on task B which relies on task A which ries on C.

We could add that to futures though it starts getting complicated. All very interesting 🙂

You might as well switch to the

SequentialTaskRunner

if you’re not using concurrency.