Here are practical ways to investigate and narrow down BrokenProcessPool errors. Assuming you’re running Prefect 3.x with ProcessPoolTaskRunner. Quick triage checklist - Check for non-picklable objects in task inputs/returns (lambdas, local functions, open file handles, DB connections, locks) - Reduce parallelism temporarily: ProcessPoolTaskRunner(max_workers=1 or 2) - Await all futures before flow ends: call .result(), wait(), or return futures from the flow - Try ThreadPoolTaskRunner or SequentialTaskRunner to isolate process-only issues - Look for OOM/segfaults: large payloads, big maps, or heavy native libs - If on Windows/macOS or notebooks: ensure tasks/flows are defined at top-level (no nested defs); avoid Jupyter for ProcessPool where possible Why this happens (most common causes) - Serialization failures: cloudpickle can’t serialize a parameter/return (e.g., lambda, local function, file handle) - OOM or native crash in a worker: big data, returning huge results, or C-extensions (NumPy/TF/PyTorch/OpenMP) crashing - Too many small tasks at once: overwhelming memory or IPC queues - Abrupt shutdown: killing the parent process or ending the flow before futures resolve - Start method/context issues: Prefect uses “spawn”; forcing “fork” elsewhere in your code can destabilize workers Experiments to pinpoint the cause 1) Swap task runners - If process-only: ThreadPoolTaskRunner works but ProcessPoolTaskRunner breaks → pickling or multi-process incompatibility - If both break → logic bug or resource limits

from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner, ThreadPoolTaskRunner, SequentialTaskRunner

@task
def work(x): ...
@flow(task_runner=ProcessPoolTaskRunner(max_workers=2))
def with_process(): return work.map(range(10))
@flow(task_runner=ThreadPoolTaskRunner(max_workers=10))
def with_threads(): return work.map(range(10))
@flow(task_runner=SequentialTaskRunner())
def sequential(): return [work.submit(i).result() for i in range(10)]

2) Validate pickling early

import cloudpickle as cp
cp.dumps(your_inputs)         # should not raise
# If you return complex objects, also test their serialization

3) Minimize concurrency and batch - If it works with max_workers=1 and fails >1, it’s often a race/native lib crash or resource exhaustion - If mapping over large iterables, batch your inputs:

@task
def process_batch(batch): ...
@flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
def run():
    data = list(range(10000))
    batch_size = 100
    batches = [data[i:i+batch_size] for i in range(0, len(data), batch_size)]
    futures = process_batch.map(batches)
    return [f.result() for f in futures]

4) Ensure futures are resolved before pool shutdown

from prefect import flow, task
from prefect.futures import wait

@task
def t(x): return x*2

@flow
def ok():
    futs = [t.submit(i) for i in range(10)]
    wait(futs)                      # or [f.result() for f in futs] or return futs
    return [f.result() for f in futs]

Diagnostics and logging - Turn up logging to see worker lifecycle: - Set environment var: PREFECT_LOGGING_LEVEL=DEBUG - Catch native crashes: - Set PYTHONFAULTHANDLER=1 to dump tracebacks on segfaults - If running in containers/Kubernetes: - Check for OOMKill events (kubectl describe pod) or container exit codes - Reduce memory footprint and/or increase container memory - Libraries that oversubscribe threads (can cause OOM/instability): limit threads - OMP_NUM_THREADS=1, MKL_NUM_THREADS=1, NUMEXPR_NUM_THREADS=1 Gotchas to avoid - Don’t pass lambdas, local functions, or partially applied callables as task args/returns - Don’t pass open connections/sessions into tasks (create them inside the task) - Don’t force multiprocessing start method elsewhere (let Prefect use spawn) - Avoid returning massive payloads; persist to storage and return references instead Minimal repro template

Use this to isolate whether the function itself or Prefect/multiprocessing is at fault:

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

def pure_func(x):
    # Only use picklable inputs/outputs; no global mutable state
    return x * x

@task
def do_work(x):
    return pure_func(x)

@flow(task_runner=ProcessPoolTaskRunner(max_workers=2))
def demo():
    futs = do_work.map(range(100))
    return [f.result() for f in futs]

if __name__ == "__main__":
    demo()

Useful Prefect docs - Task runners concept: Task runners in Prefect 3.x - ProcessPoolTaskRunner API: ProcessPoolTaskRunner reference - Run work concurrently (submit/map best practices): Run work concurrently If you can share: - A minimal code snippet of the task(s) and how you invoke them - The task runner you’re using and your platform (OS, local vs. Docker/K8s) - Whether you see OOM/segfault hints in system/pod logs I can help pinpoint the likely root cause and suggest a targeted fix.