ProcessPoolTaskRunner produces cannot pickle error

[TropComplique]

Bug summary

The following simple flow doesn't work:

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

@task
def task1(x: int) -> int:
    return x

@task
def task2(x: int) -> int:
    return x

@task
def task3(x: int) -> int:
    return x

@flow(task_runner=ProcessPoolTaskRunner(max_workers=2))
def workflow():
    a = task1.submit(x=1)
    b = task2.submit(x=2, wait_for=[a])
    task3.submit(x=3, wait_for=[b])

if __name__ == "__main__":
    workflow()

It gives the following error:

14:30:24.443 | INFO    | Flow run 'quixotic-doberman' - Beginning flow run 'quixotic-doberman' for flow 'workflow'
14:30:24.446 | INFO    | Flow run 'quixotic-doberman' - View at http://127.0.0.1:4200/runs/flow-run/b0202b43-cd69-43e0-b8df-d27438b31b9e
14:30:24.508 | ERROR   | Flow run 'quixotic-doberman' - Encountered exception during execution: TypeError("cannot pickle '_thread.RLock' object")
Traceback (most recent call last):
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 781, in run_context
    yield self
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 1395, in run_flow_sync
    engine.call_flow_fn()
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 801, in call_flow_fn
    result = call_with_parameters(self.flow.fn, self.parameters)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/utilities/callables.py", line 210, in call_with_parameters
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/work/video_understanding/simple.py", line 19, in workflow
    b = task2.submit(x=2, wait_for=[a])
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/tasks.py", line 1321, in submit
    future = task_runner.submit(self, parameters, wait_for)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/task_runners.py", line 708, in submit
    wrapped_call = cloudpickle_wrapped_call(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/utilities/callables.py", line 224, in cloudpickle_wrapped_call
    payload = cloudpickle.dumps((__fn, args, kwargs))  # type: ignore   # no stubs available
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/cloudpickle/cloudpickle.py", line 1537, in dumps
    cp.dump(obj)
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/cloudpickle/cloudpickle.py", line 1303, in dump
    return super().dump(obj)
           ^^^^^^^^^^^^^^^^^
TypeError: cannot pickle '_thread.RLock' object
14:30:26.022 | INFO    | Task run 'task1-c3c' - Finished in state Completed()
14:30:26.073 | WARNING | EventsWorker - Still processing items: 1 items remaining...
14:30:26.274 | INFO    | Flow run 'quixotic-doberman' - Finished in state Failed("Flow run encountered an exception: TypeError: cannot pickle '_thread.RLock' object")
Traceback (most recent call last):
  File "/home/ubuntu/work/video_understanding/simple.py", line 23, in <module>
    workflow()
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/flows.py", line 1702, in __call__
    return run_flow(
           ^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 1552, in run_flow
    ret_val = run_flow_sync(**kwargs)
              ^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 1397, in run_flow_sync
    return engine.state if return_type == "state" else engine.result()
                                                       ^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 361, in result
    raise self._raised
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 781, in run_context
    yield self
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 1395, in run_flow_sync
    engine.call_flow_fn()
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/flow_engine.py", line 801, in call_flow_fn
    result = call_with_parameters(self.flow.fn, self.parameters)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/utilities/callables.py", line 210, in call_with_parameters
    return fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/work/video_understanding/simple.py", line 19, in workflow
    b = task2.submit(x=2, wait_for=[a])
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/tasks.py", line 1321, in submit
    future = task_runner.submit(self, parameters, wait_for)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/task_runners.py", line 708, in submit
    wrapped_call = cloudpickle_wrapped_call(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/prefect/utilities/callables.py", line 224, in cloudpickle_wrapped_call
    payload = cloudpickle.dumps((__fn, args, kwargs))  # type: ignore   # no stubs available
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/cloudpickle/cloudpickle.py", line 1537, in dumps
    cp.dump(obj)
  File "/home/ubuntu/default_venv/lib/python3.12/site-packages/cloudpickle/cloudpickle.py", line 1303, in dump
    return super().dump(obj)
           ^^^^^^^^^^^^^^^^^
TypeError: cannot pickle '_thread.RLock' object

Notes:

1. If I change ProcessPoolTaskRunner to ThreadPoolTaskRunner then everything works.
2. If I remove all wait_for then everything works.

Version info

Version:              3.4.22
API version:          0.8.4
Python version:       3.12.3
Git commit:           6a6d114f
Built:                Fri, Oct 03, 2025 06:15 PM
OS/Arch:              linux/x86_64
Profile:              local
Server type:          server
Pydantic version:     2.10.4
Server:
  Database:           sqlite
  SQLite version:     3.45.1

Additional context

No response

[desertaxle]

Thanks for the bug report @TropComplique! It looks like we may need to update our wait_for logic to better handle futures from the ProcessPoolTaskRunner by waiting for completion in the main process.

[TropComplique]

Flow like this also doesn't work:

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

@task
def task1(x: int) -> int:
    return x

@task
def task2(x: int) -> int:
    return x

@flow(task_runner=ProcessPoolTaskRunner(max_workers=2))
def workflow():
    a = task1.submit(x=1)
    b = task2.submit(x=a)

if __name__ == "__main__":
    workflow()

[desertaxle]

Yeah, that makes sense. a is a PrefectFuture, so passing it as input engages the same waiting mechanics.

The workaround for this right now would be to call .wait on the future before passing it in, but I know that likely won't be ideal in many scenarios.