Second question, I have a mapped flow pattern like below. How can I ensure this kicks off all sub-flows simultaneously before waiting for any individual run to complete? I am using a synchronous LocalDaskExecutor for this.

child_ids_result = create_flow_run.map(
    flow_name=unmapped(...),
    project_name=unmapped(...),
    parameters=parameters,
    run_name=run_names,
)
wait_for_flow_run_result = wait_for_flow_run.map(
    flow_run_id=child_ids_result,
    stream_states=unmapped(True),
    stream_logs=unmapped(True),
    raise_final_state=unmapped(True),
)
final_task(..., upstream_tasks=[wait_for_flow_run_result])

It is the wait here that is occupying a thread. Your waits are the one that will limit here because these are not async unfortuantely.

Is there not a way to say run all these create flow tasks, then run all the wait for flow tasks?

Oh that’s news to me but that’s clever!