Architectural guidance: waiting on some remote background tasks to finish inside the flow

[asmhack]

Hi all,

I have a prefect flow and at some point it submits a lot of tasks into Google Cloud Tasks, so part of work is just delegated to the cloud functions that can be run in parallel and store some data into cloud storage.

Before I start with prefect I just had a sleep() in an infinite while loop to pause the whole linear logic and wait until all task queues would be empty.

Could you please suggest best solution for prefect? Should I just pause the flow like I did or there is better solution for this situation.

Thanks in advance.

[joshmeek]

Hi @asmhack if you have tasks that are shipping work out to external solutions then I would suggest having your task (or a subsequent downstream task) wait on that function call and only continue based on its success. For example the StartFlowRun task in the task library has a kwarg wait that when toggled will wait for the created flow run to finish before the task proceeds.

prefect/src/prefect/tasks/prefect/flow_run.py

Lines 181 to 191 in 38ec61f

	if not self.wait:
	return flow_run_id
	while True:
	time.sleep(10)
	flow_run_state = client.get_flow_run_info(flow_run_id).state
	if flow_run_state.is_finished():
	exc = signal_from_state(flow_run_state)(
	f"{flow_run_id} finished in state {flow_run_state}"
	)
	raise exc

If you want to set up your tasks to require manual approval that the Google Cloud Tasks finished then you can also optionally look into the pause/resume functionality in prefect.
https://docs.prefect.io/core/idioms/pause-for-approval.html
https://docs.prefect.io/api/latest/engine/signals.html#pause

[asmhack]

Thank you for answer @joshmeek .
Is there any pitfalls when use the time.sleep() inside the task? I mean I can set the timeout value for the @task but what about the heartbeat option in flow settings? Would it work properly and not mark task as zombie if it holds the execution for a long time, like 20-30 minutes.

Maybe it's better to retry the task with a retry delay? Or use both techniques in a mix. Sleep for some reasonable time, like 30 seconds and then retry and again sleep and so on.
What is the best approach here?

And question about resources:

Imagine situation when task A and B are not dependant and should be executed in parallel.

if I would retry task (or sleep inside), then would be agent able to run another tasks that have to be ran in parallel? In case of retrying it should do this, since it's waiting for scheduled execution of task A after retry, so it can do some job in parallel task B.

Thanks in advance.

[joshmeek]

Task run heart beats were actually removed in a recent release so heartbeats shouldn't be a a problem at all here! (even before that the heartbeat ran in a subprocess so if your task was sleeping it had no effect anyway)

It's honestly up to you on how you want to mix sleeps and retries. Sleeping is a perfectly fine technique because retries generally mean a task entered some state (such as an error occurred) where you want it to retry. If one task is sleeping or in the process of retrying other tasks will still be able to execute in parallel because this is dependent on the flow's executor choice.