https://prefect.io logo
Join Slack
Powered by
When DaskExecutor is used, does Prefect delegate D...
# ask-community
s
When DaskExecutor is used, does Prefect delegate DAG (of tasks) scheduling all to DaskScheduler?
k
I think the answer here is not quite. Tasks are submitted immediately using 
client.submit()
. The exception is mapping where if you have A -> B -> C, then A1 -> B1 -> C1 are submitted. The Prefect DAG still has features the Dask DAG doesn’t have. For example, Prefect is still responsible for stuff like caching across flow runs or triggers to determine whether or not to run a task.
Stuff like this happens even before submission to Dask
s
Consider a simple DAG: A -> B Who (Prefect or Dask) takes care of the dependency scheduling that ensure B only starts after A is done?
k
Prefect
Ah well, for mapping like the example above it’s Dask. But if not mapping it’s Prefect
s
Copy code
Tasks are submitted immediately using client.submit().
If tasks are submitted immediately, how is the "dependency wait" honoured?
k
Because Prefect knows the DAG so it can traverse that DAG during run time and then submit the next task
s
So prefect "blocks" until A is complete, then schedules B?
k
Yes because if A fails, or raises SKIP, then Prefect doesn’t want to execute B
s
Which line in 
flow_runner.py
's 
get_flow_run_state
is the "block" logic?
k
It’s not that simple because there are many checks and the flow runner is pretty hard to read. I can try looking for it, but is there something you are trying to do?
s
Can you give me some pointers? I can go do the reading myself. Just trying to understand prefect implementation code.
Thanks! 🙏
k
So you can start here in the FlowRunner, and then notice the tasks are sorted. I believe that’s the order of submission and then there is a bunch of logic for mapping, so it goes to be submitted here .
By that point, it’s already submitted to the DaskExecutor so you can see the DaskExecutor.run()
And then notice in the TaskRunner here , this is run recursively and checks for the conditions to run tasks, and then actually runs task in at the end after all the checks
So the answer to your question is there is a sort of the tasks in the Flow, and then the tasks are submission in order. The task runner recursively checks for conditions. The topological sort can be found here
Then to the first question, note this all happens with/without Dask. If using DaskExecutor, 
executor.submit()
= 
client.submit()
. If using LocalExecutor, 
executor.submit() = regular function call
s
Can you elaborate on "TaskRunner recursively checks for conditions"?
k
Look at all this
s
Ok, it seems recursive task run only comes into play if task loops
t
Cool thread - thanks for that!
s
@Kevin Kho Can you task runner only recursively runs a task if the task loops?
k
I don’t think so right? Because a Failed state that goes to Retry will end up in that same loop I think. I’d need to dive in more.
3 Views
Open in Slack
PreviousNext
Powered by