> ## Documentation Index
> Fetch the complete documentation index at: /docs/llms.txt
> Use this file to discover all available pages before exploring further.

# task_runners

# `prefect.task_runners`

## Classes

### `TaskRunner` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L64](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L64)</sup>

Abstract base class for task runners.

A task runner is responsible for submitting tasks to the task run engine running
in an execution environment. Submitted tasks are non-blocking and return a future
object that can be used to wait for the task to complete and retrieve the result.

Task runners are context managers and should be used in a `with` block to ensure
proper cleanup of resources.

**Methods:**

#### `duplicate` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L86](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L86)</sup>

```python theme={null}
duplicate(self) -> Self
```

Return a new instance of this task runner with the same configuration.

#### `map` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L119](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L119)</sup>

```python theme={null}
map(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any | unmapped[Any] | allow_failure[Any]], wait_for: Iterable[PrefectFuture[R]] | None = None) -> PrefectFutureList[F]
```

Submit multiple tasks to the task run engine.

**Args:**

* `task`: The task to submit.
* `parameters`: The parameters to use when running the task.
* `wait_for`: A list of futures that the task depends on.

**Returns:**

* An iterable of future objects that can be used to wait for the tasks to
* complete and retrieve the results.

#### `name` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L81](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L81)</sup>

```python theme={null}
name(self) -> str
```

The name of this task runner

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L92](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L92)</sup>

```python theme={null}
submit(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> F
```

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L102](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L102)</sup>

```python theme={null}
submit(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> F
```

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L111](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L111)</sup>

```python theme={null}
submit(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> F
```

### `ThreadPoolTaskRunner` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L243](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L243)</sup>

A task runner that executes tasks in a separate thread pool.

**Attributes:**

* `max_workers`: The maximum number of threads to use for executing tasks.
  Defaults to `PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS` or `sys.maxsize`.

**Examples:**

Use a thread pool task runner with a flow:

```python theme={null}
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def some_io_bound_task(x: int) -> int:
    # making a query to a database, reading a file, etc.
    return x * 2

@flow(task_runner=ThreadPoolTaskRunner(max_workers=3)) # use at most 3 threads at a time
def my_io_bound_flow():
    futures = []
    for i in range(10):
        future = some_io_bound_task.submit(i * 100)
        futures.append(future)

    return [future.result() for future in futures]
```

Use a thread pool task runner as a context manager:

```python theme={null}
from prefect.task_runners import ThreadPoolTaskRunner

@task
def some_io_bound_task(x: int) -> int:
    # making a query to a database, reading a file, etc.
    return x * 2

# Use the runner directly
with ThreadPoolTaskRunner(max_workers=2) as runner:
    future1 = runner.submit(some_io_bound_task, {"x": 1})
    future2 = runner.submit(some_io_bound_task, {"x": 2})

    result1 = future1.result()  # 2
    result2 = future2.result()  # 4
```

Configure max workers via settings:

```python theme={null}
# Set via environment variable
# export PREFECT_TASKS_RUNNER_THREAD_POOL_MAX_WORKERS=8

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner

@flow(task_runner=ThreadPoolTaskRunner())  # Uses 8 workers from setting
def my_flow():
    ...
```

**Methods:**

#### `cancel_all` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L516](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L516)</sup>

```python theme={null}
cancel_all(self) -> None
```

#### `duplicate` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L339](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L339)</sup>

```python theme={null}
duplicate(self) -> 'ThreadPoolTaskRunner[R]'
```

#### `map` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L493](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L493)</sup>

```python theme={null}
map(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]
```

#### `map` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L501](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L501)</sup>

```python theme={null}
map(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]
```

#### `map` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L508](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L508)</sup>

```python theme={null}
map(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]
```

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L343](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L343)</sup>

```python theme={null}
submit(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]
```

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L352](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L352)</sup>

```python theme={null}
submit(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]
```

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L360](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L360)</sup>

```python theme={null}
submit(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]
```

Submit a task to the task run engine running in a separate thread.

**Args:**

* `task`: The task to submit.
* `parameters`: The parameters to use when running the task.
* `wait_for`: A list of futures that the task depends on.

**Returns:**

* A future object that can be used to wait for the task to complete and
* retrieve the result.

### `ProcessPoolTaskRunner` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L769](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L769)</sup>

A task runner that executes tasks in a separate process pool.

This task runner uses `ProcessPoolExecutor` to run tasks in separate processes,
providing true parallelism for CPU-bound tasks and process isolation. Tasks
are executed with proper context propagation and error handling.

**Attributes:**

* `max_workers`: The maximum number of processes to use for executing tasks.
  Defaults to `multiprocessing.cpu_count()` if `PREFECT_TASKS_RUNNER_PROCESS_POOL_MAX_WORKERS` is not set.
* `initializer`: An optional picklable callable run once in each worker
  subprocess before it executes any tasks, mirroring
  `ProcessPoolExecutor`'s `initializer`.
* `initargs`: A tuple of arguments passed to `initializer`.

**Examples:**

Use a process pool task runner with a flow:

```python theme={null}
from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner

@task
def compute_heavy_task(n: int) -> int:
    # CPU-intensive computation that benefits from process isolation
    return sum(i ** 2 for i in range(n))

@flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
def my_flow():
    futures = []
    for i in range(10):
        future = compute_heavy_task.submit(i * 1000)
        futures.append(future)

    return [future.result() for future in futures]
```

Use a process pool task runner as a context manager:

```python theme={null}
from prefect.task_runners import ProcessPoolTaskRunner

@task
def my_task(x: int) -> int:
    return x * 2

# Use the runner directly
with ProcessPoolTaskRunner(max_workers=2) as runner:
    future1 = runner.submit(my_task, {"x": 1})
    future2 = runner.submit(my_task, {"x": 2})

    result1 = future1.result()  # 2
    result2 = future2.result()  # 4
```

Configure max workers via settings:

```python theme={null}
# Set via environment variable
# export PREFECT_TASKS_RUNNER_PROCESS_POOL_MAX_WORKERS=8

from prefect import flow
from prefect.task_runners import ProcessPoolTaskRunner

@flow(task_runner=ProcessPoolTaskRunner())  # Uses 8 workers from setting
def my_flow():
    ...
```

Run custom setup in each worker subprocess:

```python theme={null}
import sys
from prefect.task_runners import ProcessPoolTaskRunner

def add_to_path(*paths):
    sys.path[:0] = paths

runner = ProcessPoolTaskRunner(
    initializer=add_to_path,
    initargs=("/opt/my_app/libs",),
)
```

**Methods:**

#### `cancel_all` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1288](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1288)</sup>

```python theme={null}
cancel_all(self) -> None
```

#### `duplicate` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L897](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L897)</sup>

```python theme={null}
duplicate(self) -> Self
```

#### `map` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1265](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1265)</sup>

```python theme={null}
map(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]
```

#### `map` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1273](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1273)</sup>

```python theme={null}
map(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]
```

#### `map` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1280](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1280)</sup>

```python theme={null}
map(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectConcurrentFuture[R]]
```

#### `set_subprocess_message_processor_factories` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L928](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L928)</sup>

```python theme={null}
set_subprocess_message_processor_factories(self, subprocess_message_processor_factories: Iterable[_SubprocessMessageProcessorFactory] | None = None) -> None
```

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1160](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1160)</sup>

```python theme={null}
submit(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]
```

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1169](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1169)</sup>

```python theme={null}
submit(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]
```

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1177](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1177)</sup>

```python theme={null}
submit(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectConcurrentFuture[R]
```

Submit a task to the task run engine running in a separate process.

**Args:**

* `task`: The task to submit.
* `parameters`: The parameters to use when running the task.
* `wait_for`: A list of futures that the task depends on.
* `dependencies`: A dictionary of dependencies for the task.

**Returns:**

* A future object that can be used to wait for the task to complete and
* retrieve the result.

#### `subprocess_message_processor_factories` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L908](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L908)</sup>

```python theme={null}
subprocess_message_processor_factories(self) -> tuple[_SubprocessMessageProcessorFactory, ...]
```

#### `subprocess_message_processor_factories` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L914](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L914)</sup>

```python theme={null}
subprocess_message_processor_factories(self, subprocess_message_processor_factories: Iterable[_SubprocessMessageProcessorFactory] | None = None) -> None
```

### `PrefectTaskRunner` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1362](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1362)</sup>

**Methods:**

#### `duplicate` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1366](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1366)</sup>

```python theme={null}
duplicate(self) -> 'PrefectTaskRunner[R]'
```

#### `map` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1425](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1425)</sup>

```python theme={null}
map(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectDistributedFuture[R]]
```

#### `map` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1433](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1433)</sup>

```python theme={null}
map(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectDistributedFuture[R]]
```

#### `map` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1440](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1440)</sup>

```python theme={null}
map(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None) -> PrefectFutureList[PrefectDistributedFuture[R]]
```

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1370](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1370)</sup>

```python theme={null}
submit(self, task: 'Task[P, CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectDistributedFuture[R]
```

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1379](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1379)</sup>

```python theme={null}
submit(self, task: 'Task[Any, R]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectDistributedFuture[R]
```

#### `submit` <sup>[https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task\_runners.py#L1387](https://github.com/PrefectHQ/prefect/blob/main/src/prefect/task_runners.py#L1387)</sup>

```python theme={null}
submit(self, task: 'Task[P, R | CoroutineType[Any, Any, R]]', parameters: dict[str, Any], wait_for: Iterable[PrefectFuture[Any]] | None = None, dependencies: dict[str, set[RunInput]] | None = None) -> PrefectDistributedFuture[R]
```

Submit a task to the task run engine running in a separate thread.

**Args:**

* `task`: The task to submit.
* `parameters`: The parameters to use when running the task.
* `wait_for`: A list of futures that the task depends on.

**Returns:**

* A future object that can be used to wait for the task to complete and
* retrieve the result.