Short answer: In Prefect 3.x you don’t need to do anything special. With a ProcessPoolTaskRunner, a task’s

timeout_seconds

is enforced inside the worker process. When the timeout hits: - The task raises a cancellation/timeout error inside the subprocess - The task unwinds and returns a failed state - The worker process remains healthy for reuse by the pool - When the flow ends, the runner shuts the pool down cleanly (

shutdown(cancel_futures=True, wait=True)

), so there are no zombie processes Why this works: - For sync tasks in a ProcessPoolTaskRunner, the task runs on the subprocess’s main thread; Prefect uses a signal-based timeout on Unix-like systems that can interrupt many blocking calls (e.g.,

time.sleep

), allowing the subprocess to unwind normally. - The runner’s lifecycle management ensures processes are reaped on exit. Best practices to ensure everything stays clean - Use the runner via the flow config (recommended):

from prefect import flow, task
  from prefect.task_runners import ProcessPoolTaskRunner
  import time

  @task(timeout_seconds=5)
  def do_work():
      time.sleep(10)  # Will be interrupted by timeout in a subprocess
      return "done"

  @flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
  def my_flow():
      fut = do_work.submit()
      return fut.result()

Prefect manages the pool and cleans up processes automatically when the flow finishes. - If your task spawns its own children (subprocesses, threads, DB connections, etc.), close them in

finally

so they’re cleaned up when the timeout exception is raised:

import subprocess
  from prefect import task

  @task(timeout_seconds=10)
  def run_external():
      p = subprocess.Popen(["sleep", "60"])
      try:
          return p.wait()
      finally:
          if p.poll() is None:
              p.terminate()  # or p.kill() if needed

- Prefer libraries with native timeouts for long I/O (e.g.,

requests.get(..., timeout=5)

) so they return promptly when Prefect cancels. Notes and edge cases - Windows: signals aren’t available; timeouts rely on exception injection which can’t interrupt some blocking C calls. If you have uninterruptible native code, wrap it in a child process you can kill in

finally

as shown above. - If you truly need to hard-kill the work on timeout (e.g., uninterruptible native code), run that work in a subprocess you control and terminate it when you catch `CancelledError`/`TaskRunTimeoutError`. If you’re seeing lingering processes despite this, can you share: - OS and Python version - A minimal task example (especially if it calls native/C extensions or spawns children) - How you’re configuring the runner in your flow I can help tailor a strict-cleanup pattern for your case.

Great question. For async tasks in Prefect 3.x, timeouts are enforced with anyio cancel scopes and are cooperative. Key points and considerations: - Timeouts are cooperative at await points - Cancellation is delivered at the next

await

. Long CPU-bound sections without awaits won’t be interrupted. - Add periodic

await asyncio.sleep(0)

(or yield to the loop another way) inside long loops to keep them cancellable. - Cleanup on timeout - An async timeout will raise

CancelledError

inside your task; Prefect will record a failed state and surface a

TaskRunTimeoutError

when you resolve the future. - Use

try/finally

(or

async with

) to clean up resources when cancellation happens and re-raise the cancellation:

import asyncio
    from prefect import task

    @task(timeout_seconds=5)
    async def do_io():
        conn = await open_conn()
        try:
            await conn.long_call()
        except asyncio.CancelledError:
            # clean up quickly, then re-raise
            await conn.aclose()
            raise
        else:
            await conn.aclose()

- Don’t block the event loop inside async tasks - Avoid blocking sync calls like

time.sleep(10)

in an async task; they prevent cancellation from being delivered. - If you must call sync code, offload it so it becomes cancellable at await points:

import anyio, time

    @task(timeout_seconds=5)
    async def async_wrapper():
        await anyio.to_thread.run_sync(time.sleep, 10)

- Use native library timeouts - Prefer built-in timeouts on async clients (e.g., aiohttp’s

ClientTimeout

, async DB client timeouts) so they fail promptly when cancelled. - ProcessPoolTaskRunner vs ThreadPoolTaskRunner - Async timeouts behave the same from your perspective: cancellation is delivered at awaits inside the (sub)process. - For primarily async/IO-bound workloads, prefer an in-process async runner (e.g.,

PrefectTaskRunner

) to maximize concurrency and cancellation responsiveness. A process pool doesn’t help async I/O and adds overhead. - If you do use

ProcessPoolTaskRunner

, the async task runs in a subprocess with its own event loop; on timeout, the task cancels cooperatively and the process remains healthy for reuse. The pool is shut down cleanly when the flow ends. - Avoid swallowing cancellation - If you catch

asyncio.CancelledError

, make sure to re-raise it after cleanup; swallowing it will make the task appear to succeed while background work may keep running. - If you spawn background tasks inside your task - Manage them explicitly. On timeout, cancel any

asyncio.create_task(...)

children in `finally`:

t = asyncio.create_task(worker())
    try:
        await do_work()
    finally:
        t.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await t

- Windows support - Async timeouts work cross‑platform (anyio cancellation). The sync signal-based behavior has platform nuances, but async is fine on Windows. If you share a small snippet of your async task and runner choice, I can suggest concrete tweaks to make it maximally cancellable.