Have a chance to terminate all flow's (concurrent) tasks on flow exit / canceling futures

[hnykda]

Bug summary

Consider this simple flow:

from prefect import flow, task
import time


@task
def long_task():
    time.sleep(100)


@task
def short_task():
    time.sleep(5)
    return True


@flow
def f():
    lt = long_task.submit()
    st = short_task.submit()
    if st.result():
        return "I finished"
    return "This will never happen"


if __name__ == "__main__":
    f.serve(name="f")

I want to trigger some stuff and I don't mind that I might cancel them mid-flight. What matters to me is that they all run as soon as possible just in case I will need them. However, if the flow above finishes (and even shows as Completed, confusingly), the future of long_task still seems to be still running.

I understand why, but I am surprised there doesn't seem to be a way to cancel all existing running tasks. I don't want to wait nor result (those would block other parts from the flow), and I don't see any .cancel or anything. In my ideal world, the warning would be raised and the task killed/terminated.

Another variant of this is when a concurrent task raises - again, the other submitted task is not cancelled. Again, I realize that "terminate all on any task failure" is not a "universal" behavior everyone wants all the time, but I couldn't find any mention about this behavior and its adjustment in the docs.

Version info

Version:             3.3.5
API version:         0.8.4
Python version:      3.13.0
Git commit:          db4b7a33
Built:               Thu, Apr 17, 2025 09:25 PM
OS/Arch:             darwin/arm64
Profile:             local
Server type:         server
Pydantic version:    2.11.3

Additional context

No response

[desertaxle]

Thanks for the issue @hnykda! To cancel in-flight execution, we're bound by what the underlying executor offers. In this case, the tasks run in separate threads via a ThreadPoolExecutor. While the futures returned after submitting to the thread pool have a cancel method, that cancellation only works if cancel is called before execution starts, which sounds like it wouldn't help in your case.

If you want to cancel work that has already started, you'll need to run the work either in a coroutine or a subprocess. I think if Prefect offered a ProcessTaskRunner (it's been requested before), then we could probably expose a cancel method to the corresponding PrefectFuture implementation.

[hnykda]

I see, thanks