Short answer: it means the Dask worker’s (or scheduler’s) Tornado event loop isn’t getting CPU time for multiple seconds. That almost always happens when something in the worker process is blocking the interpreter or starving the loop. Common causes when using DaskTaskRunner (Prefect 3.x) - CPU-bound Python code that holds the GIL - Tight Python loops, heavy JSON/regex/pandas ops, large compressions/serializations can monopolize the GIL so the event loop can’t run. - Blocking I/O or sleeps inside “async” code - Using

time.sleep()

, synchronous DB/HTTP clients, or other blocking calls inside tasks (especially in async tasks) pauses the loop. - Too many threads per worker (oversubscription) - With

processes=True

, each worker process still runs a thread pool (

threads_per_worker

). If CPU-bound tasks occupy all threads, the loop can be starved. - Large object serialization or result transfer - Pickling/compressing 100s of MB/GB can block CPU long enough to trigger the warning. - Memory pressure, GC pauses, or spill-to-disk - High memory churn can cause stop-the-world GC or disk spill thrashing; both stall the loop. - Container/VM CPU throttling - Strict CPU limits in Kubernetes/containers can starve the loop under load. - Scheduler/worker overload or excessive logging - A very busy scheduler or per-task log storms can delay loop callbacks. - Network hiccups (less common for this exact message) - Usually shows as missed heartbeats; can appear together with unresponsive-loop warnings. Why you can see this even with parallel “processes” - Each worker is a process, but within that process tasks run on a thread pool; the worker’s event loop lives in the same process and still needs to acquire the GIL and CPU slices. If your task threads monopolize the GIL, the loop goes quiet. Practical mitigations - Match execution model to the workload - CPU-bound: prefer processes and limit threads per worker

from prefect_dask.task_runners import DaskTaskRunner

    task_runner = DaskTaskRunner(
      cluster_kwargs={"processes": True, "threads_per_worker": 1, "n_workers": 4}
    )

- I/O-bound: threads are fine; avoid blocking calls and use async libraries where possible. - Avoid blocking in async tasks - Replace

time.sleep()

with

await asyncio.sleep()

- Use non-blocking clients (e.g.,

aiohttp

instead of

requests

), or make the task sync and let Dask handle the threads. - Set task timeouts to prevent indefinite blocking

from prefect import task

  @task(timeout_seconds=300)
  def do_work(...):
      ...

- Reduce serialization pressure - Don’t pass huge in-memory objects between tasks; pass file/object-store paths instead. - Right-size resources - Keep

threads_per_worker

low for CPU-bound work; give workers enough CPU so the event loop isn’t constantly throttled; avoid container CPU limits that are too tight. - Monitor with the Dask dashboard - Look for long-running tasks, saw-tooth memory (GC/spill), or missing/slow workers. If you can share: - Your

DaskTaskRunner

config (

processes

,

threads_per_worker

,

n_workers

) - Whether tasks are CPU-bound or I/O-bound - Typical input/output sizes - The exact log line (including “unresponsive for X s”) …I can help pinpoint which of the above is most likely in your setup and suggest concrete tweaks. References - Dask Distributed docs: Dask Distributed documentation - Prefect Dask integration: prefect-dask repository

Thanks for the details — this behavior is expected with CPU‑bound workloads on a single machine as you increase concurrency. Even if the tasks are independent, per‑task latency often grows with higher

n_workers

due to resource contention and runtime effects. Why individual tasks get slower as n_workers increases - CPU oversubscription and context switching - If

n_workers

exceeds the number of physical cores, the OS time-slices processes. Context switching + loss of CPU cache locality slows each task. - Even at/below core count, enabling more cores typically lowers per‑core turbo frequency and can reduce single‑task speed. - Nested threading in numeric libraries - If tasks use NumPy/pandas/BLAS (MKL/OpenBLAS/numexpr), each process may spin up its own thread pool. With 30 workers this can massively oversubscribe the CPU and slow everything down. - Memory bandwidth and cache contention - Many simultaneous CPU‑heavy processes fight for LLC and memory bandwidth. Individual tasks slow down even without dependencies. - Serialization and data movement overhead - More in-flight tasks means more pickling/unpickling, result transfers, and GC activity, which can hurt per‑task latency. - Possible misalignment with process model - Make sure you’re actually using processes for CPU‑bound work. If you ended up with threads (GIL), throughput and latency will degrade. What to change for CPU‑bound performance 1) Be explicit about using processes and 1 thread per worker - You already set

threads_per_worker=1

. Make

processes=True

explicit to guarantee process-based parallelism:

from prefect_dask.task_runners import DaskTaskRunner

task_runner = DaskTaskRunner(
    cluster_kwargs={
        "processes": True,
        "n_workers": 30,          # tune based on physical cores (see below)
        "threads_per_worker": 1,
        "memory_limit": "auto",   # or a realistic per-worker value
        "silence_logs": False,
    }
)

2) Disable nested threading in numeric libs - If you use NumPy/pandas/numexpr/MKL/OpenBLAS, set these before creating the cluster:

import os
os.environ["OMP_NUM_THREADS"] = "1"
os.environ["MKL_NUM_THREADS"] = "1"
os.environ["OPENBLAS_NUM_THREADS"] = "1"
os.environ["NUMEXPR_MAX_THREADS"] = "1"

- This avoids each process spawning multiple BLAS threads (a very common cause of slowdowns at higher worker counts). 3) Match n_workers to physical cores (then tune) - Start with

n_workers = number_of_physical_cores

(not hyperthreads). Then measure. - It’s common to find the sweet spot slightly below physical cores for heavy vectorized math (to reduce bandwidth/cache contention). 4) Set a realistic per‑worker memory_limit - With

memory_limit='20GiB'

and 30 workers, you’re “allowing” 600 GiB. If the machine doesn’t have that, you can hit spill/GC behavior that hurts performance. - Prefer

"auto"

or set something like total_memory * 0.6 / n_workers. 5) Verify with the Dask dashboard - Check Task Stream for increased per‑task runtime as concurrency rises. - Watch the system CPU graph (all cores pegged), memory use (spill events), and worker logs. 6) If you still see event loop unresponsive warnings - That’s typically due to 100% CPU saturation — the worker loop can’t heartbeat. The fixes above (processes, BLAS threads, right-sized workers) usually eliminate it. Avoid “fixing” with longer heartbeat timeouts. Optional, advanced - Pin workers to cores (taskset/numactl) if you need predictable per‑task latencies. - If tasks are very heavy in NumPy/BLAS, you may get better stability with fewer workers than cores. If you can share your CPU model, OS, whether the task uses NumPy/pandas/scikit‑learn/numexpr, and the machine’s physical core count, I can suggest a concrete

n_workers

and env var setup. Useful links - Dask Distributed docs: Dask Distributed documentation

- Diagnosing performance: Diagnosing performance (Dask) - Prefect Dask integration: prefect-dask repository

Short answer: no —

threads_per_worker=1

only limits Dask’s Python worker threads. It does not cap native threads spawned by libraries like NumPy/BLAS (MKL/OpenBLAS) or numexpr. Those libraries create their own thread pools inside each worker process and can easily oversubscribe your CPU (e.g., 30 processes × 8 MKL threads = 240 runnable threads), which slows each task as you increase

n_workers

. What to do - Keep

threads_per_worker=1

for CPU-bound work. - Also cap native library threads per process:

import os
os.environ["OMP_NUM_THREADS"] = "1"        # OpenMP default
os.environ["MKL_NUM_THREADS"] = "1"       # MKL
os.environ["OPENBLAS_NUM_THREADS"] = "1"  # OpenBLAS
os.environ["NUMEXPR_MAX_THREADS"] = "1"   # numexpr
os.environ["MKL_DYNAMIC"] = "FALSE"       # optional: prevent MKL from ramping threads

- Set these before importing NumPy/pandas/scikit‑learn, or configure them in the worker environment (recommended for deployments). How to verify at runtime - Use threadpoolctl to see what libraries are loaded and their thread limits:

from threadpoolctl import threadpool_info, threadpool_limits

print(threadpool_info())               # inspect current lib/thread settings
threadpool_limits(1)                   # programmatically force 1 thread per lib
print(threadpool_info())               # confirm limits applied

Notes - If your tasks are pure-Python tight loops (no NumPy/BLAS/numexpr), these env vars won’t change much — but they’re safe to set and prevent surprises if any array ops sneak in. - Also make sure you’re using processes for CPU-bound work:

from prefect_dask.task_runners import DaskTaskRunner

task_runner = DaskTaskRunner(
  cluster_kwargs={"processes": True, "n_workers": <~physical_cores>, "threads_per_worker": 1}
)

Related refs - Dask Distributed docs: Dask Distributed documentation - Prefect Dask integration: prefect-dask repository If you can share whether your tasks use NumPy/pandas/scikit‑learn and your CPU’s physical core count, I can suggest concrete

n_workers

and env settings.

Short answer: threadpoolctl can both inspect and enforce limits, but only inside the process where you call it. With Dask’s process-based workers, that means you must run it in every worker process. Environment variables remain the simplest way to guarantee the limit is applied to all workers from the moment they start. How to use threadpoolctl to enforce limits - It works for MKL, OpenBLAS, BLIS, Apple Accelerate, OpenMP runtimes, and numexpr. - Call it early in each worker process and don’t restore the limits to keep them for the worker’s lifetime:

from threadpoolctl import threadpool_limits
threadpool_limits(1)  # set limit to 1 for this process until restored or exit

Applying it across all Dask workers - Run once on all workers right after the cluster is up:

def _limit_blas_threads(n=1):
    from threadpoolctl import threadpool_limits
    threadpool_limits(n)

# after you have a Dask client
client.run(_limit_blas_threads, n=1)

- In Prefect, you can wrap that in a small “init” task that runs before the heavy tasks. When you might still prefer env vars - You don’t control worker startup code (Kubernetes, Docker images, remote workers). - You want guarantees before any library imports and thread pools are created. - Simpler deployment: set vars in the container/worker env and they apply to every spawned worker process automatically. Practical recommendation - If you can reliably run

threadpool_limits(1)

in every worker at startup (and before heavy libs are used), you don’t strictly need the env vars. - Otherwise, set env vars (and optionally keep

threadpoolctl

as a sanity check):

OMP_NUM_THREADS=1
MKL_NUM_THREADS=1
OPENBLAS_NUM_THREADS=1
NUMEXPR_MAX_THREADS=1
MKL_DYNAMIC=FALSE   # optional, prevents MKL from ramping up

Reference: threadpoolctl docs threadpoolctl

Short answer: yes, but only if that function actually runs inside each Dask worker process. Putting it at the top of your flow script (the orchestrator) won’t affect worker subprocesses. Ways to apply limits correctly - In each worker process once (recommended)

from prefect import flow
  from prefect_dask import DaskTaskRunner
  from prefect_dask.utils import get_dask_client

  def _limit_blas_threads(n=1):
      from threadpoolctl import threadpool_limits
      threadpool_limits(n)  # stays in effect for this worker process

  @flow(task_runner=DaskTaskRunner(cluster_kwargs={"processes": True, "n_workers": 30, "threads_per_worker": 1}))
  def my_flow():
      client = get_dask_client()
      client.run(_limit_blas_threads, n=1)  # run once on all workers
      # submit tasks...

This ensures every worker sets the limit early and only once. - At the top of the task function (works, but redundant)

from threadpoolctl import threadpool_limits
  from prefect import task

  @task
  def my_task(...):
      threadpool_limits(1)  # applies to this worker process; persists for later tasks too
      ...

This will work, but you’ll run it on every task invocation. It’s safe, just a bit of overhead. - Via environment variables (simplest for deployments) Set before workers start so it applies automatically to all worker processes:

OMP_NUM_THREADS=1
  MKL_NUM_THREADS=1
  OPENBLAS_NUM_THREADS=1
  NUMEXPR_MAX_THREADS=1
  MKL_DYNAMIC=FALSE

Notes - Make sure you’re using processes for CPU-bound work: include

"processes": True

in

cluster_kwargs

. - threadpoolctl changes take effect even after libs are imported, but it’s still best to apply early in the worker lifecycle. - If different tasks need different limits, prefer

client.run

once with a value that fits the bulk of your workload. Reference: threadpoolctl