Got it — you want to immediately stop a local DaskTaskRunner, kill all its worker processes, and ensure queued tasks don’t start. If you’re in a Prefect 3.x flow using DaskTaskRunner with a local cluster, the most reliable way is to close the underlying Dask cluster. That tears down the scheduler and all workers, so any queued tasks won’t run. Do it from inside the running flow

from prefect import flow
from prefect.context import get_run_context
from prefect_dask import DaskTaskRunner

@flow(task_runner=DaskTaskRunner())
def stop_everything_now():
    ctx = get_run_context()
    runner = ctx.task_runner

    # If using a local cluster created by DaskTaskRunner, this kills all worker processes
    cluster = getattr(runner, "_cluster", None)
    if cluster is not None:
        cluster.close()            # terminates all local workers and the scheduler
    else:
        # If connected to an external address/cluster, at least detach our client
        runner.client.close()

    # After this point, queued tasks won’t start and running tasks are killed with the workers

Notes - Closing the cluster is the key step for “kill all processes.” It stops workers and the scheduler immediately; queued work cannot start. - If you only close the client, the external cluster may continue running; use

cluster.close()

for a local cluster created by the task runner. - When the flow exits normally, Prefect’s context manager already closes the Dask client/cluster; the snippet above is for forcing an early shutdown. If you need to clean up orphaned workers from your shell - macOS/Linux:

pkill -f 'dask-worker|dask-scheduler' || true

- Windows (PowerShell):

Get-CimInstance Win32_Process | Where-Object { $_.CommandLine -match 'dask-worker|dask-scheduler' } | ForEach-Object { Stop-Process -Id $_.ProcessId -Force }

Best practice going forward - Prefer letting the flow finish (Prefect will auto-close the cluster). - If you must abort mid-flow, call

cluster.close()

as above. That guarantees no queued work will start and kills all workers. References - Prefect Dask integration: Prefect Dask docs - Task runners overview: Task runners in Prefect 3 - Dask Cluster/Client lifecycle: Dask Client API Quick check: are you trying to do this from inside a currently running flow, or do you just need to clean up leftover local Dask processes from a previous run?