Skip to main content

Run Tasks Locally, Remotely, and in Hybrid Mode

Developers executing tasks in flyte-sdk need the flexibility to debug workflows on their local machines, submit orchestrated runs to remote clusters, or run driver actions locally while dispatching heavy compute actions remotely. If execution modes and task signatures are mismatched—such as trying to execute a remote-only task reference locally or calling synchronous tasks improperly inside async workflows—tasks fail with errors like RemoteTaskUsageError or thread deadlocks.

flyte-sdk resolves this by offering three execution modes (local, remote, and hybrid), providing the _Runner engine to manage execution contexts, and utilizing LocalController and RemoteController to handle dispatch, caching, serialization, and retries.

Execution Modes and Mode Selection​

flyte-sdk provides three distinct execution modes:

  • local: Executes the task and any sub-tasks directly in the local Python process. Intermediate metadata and raw data are written to local scratch paths.
  • remote: Packages code into bundles, builds or resolves container images, serializes inputs, and submits an execution run to the remote Flyte cluster via gRPC.
  • hybrid: Runs the parent task function locally while dispatching nested sub-tasks as remote actions on the Flyte cluster.

Configuring and Forcing Modes​

The public entrypoint for executing tasks is flyte.run or flyte.with_runcontext(...). When invoking flyte.run, flyte-sdk instantiates a _Runner (defined in flyte._run) that determines the effective execution mode:

import flyte

env = flyte.TaskEnvironment("example-env")

@env.task
async def compute(x: int) -> int:
    return x * 2

# Automatic selection: defaults to "remote" if flyte.init() configured a client, else "local"
run = flyte.run(compute, x=5)

# Explicitly forcing local mode
local_run = flyte.with_runcontext(mode="local").run(compute, x=10)

# Explicitly forcing remote mode with execution overrides
remote_run = flyte.with_runcontext(
    mode="remote",
    name="custom-run-001",
    env_vars={"STAGE": "production"},
    disable_run_cache=True,
).run(compute, x=20)

Internal Mode Selection Mechanism​

Inside _Runner.__init__, mode resolution inspects global client configuration via _get_init_config():

init_config = _get_init_config()
client = init_config.client if init_config else None
if not force_mode and client is not None:
    force_mode = "remote"
force_mode = force_mode or "local"
self._mode = force_mode

When _Runner.run() executes, it sets the context variable _run_mode_var to self._mode. This allows offloaded storage types (such as remote files and directories) to check whether they should perform auto-uploading (only active in remote mode) or remain local:

_run_mode_var.set(self._mode)
try:
    if self._mode == "remote":
        return await self._run_remote(task, *args, **kwargs)
    task = cast(TaskTemplate, task)
    if self._mode == "hybrid":
        return await self._run_hybrid(task, *args, **kwargs)

    with internal_ctx().new_raw_data_path(
        raw_data_path=RawDataPath.from_local_folder(local_folder=self._raw_data_path)
    ):
        return await self._run_local(task, *args, **kwargs)
finally:
    _run_mode_var.set(None)

Attempting to run a remote task representation (LazyEntity or TaskDetails) when mode is not "remote" immediately raises a ValueError("Remote task can only be run in remote mode.").

Synchronous and Asynchronous Task Behaviors​

flyte-sdk supports both standard synchronous functions (def) and asynchronous coroutine functions (async def) decorated with @env.task.

import asyncio
from typing import List
import flyte

env = flyte.TaskEnvironment("task-behaviors")

@env.task
def sync_worker(x: int) -> int:
    return x + 1

@env.task
async def async_worker(x: int) -> int:
    await asyncio.sleep(0.01)
    return x * 2

@env.task
async def orchestrator(n: int) -> List[int]:
    # Using .aio() to run synchronous tasks concurrently in asyncio
    coros = [sync_worker.aio(i) for i in range(n)]
    return await asyncio.gather(*coros)

Task Templates and _call_as_synchronous​

When @env.task decorates a function, it constructs an AsyncFunctionTaskTemplate (subclassing TaskTemplate from flyte._task). In AsyncFunctionTaskTemplate.__post_init__, the task inspects the underlying function:

if not iscoroutinefunction(self.func):
    self._call_as_synchronous = True

The execution behavior depends on whether the task is invoked within an active TaskContext:

┌─────────────────────────────────────────────────────────────┐
│                       Task Invocation                       │
└──────────────────────────────┬──────────────────────────────┘
                               │
                Is TaskContext active?
               /                       \
             No                         Yes
            /                             \
┌───────────────────────┐       ┌───────────────────────────────┐
│     Unit Test /       │       │    Active Run Orchestration   │
│ Direct Function Call  │       │                               │
│  (calls .forward())   │       │  Is inside `flyte.trace`?     │
└───────────────────────┘       └───────┬───────────────┬───────┘
                                       Yes              No
                                        │                │
                        ┌───────────────┴───┐   ┌────────┴──────────┐
                        │ Raises            │   │ Submits to active │
                        │ TraceDoesNotAllow │   │ controller        │
                        │ NestedTasksError  │   └───────────────────┘
                        └───────────────────┘
  1. Direct execution / Unit Tests (Outside TaskContext): Calling sync_worker(1) or await async_worker(1) invokes TaskTemplate.__call__, which delegates directly to TaskTemplate.forward(). The function runs as standard native Python with zero framework overhead, serialization, or controller involvement.
  2. Inside TaskContext: If invoked inside a running task, TaskTemplate.__call__ checks ctx.is_in_trace(). If inside a trace, it raises TraceDoesNotAllowNestedTasksError (tasks should use task.forward(...) if intended as regular helper calls inside traces). Otherwise, it delegates to the active controller.
  3. Synchronous Invocation via .aio(): When calling synchronous tasks inside an async parent task, calling sync_worker(x) blocks execution via controller.submit_sync(). Calling sync_worker.aio(x) submits the sync task to the controller via controller.submit_sync() and wraps the resulting concurrent.futures.Future into an asyncio.Future using asyncio.wrap_future, allowing non-blocking concurrency with asyncio.gather.

Local Controller Architecture​

Local execution is managed by LocalController in flyte._internal.controllers._local_controller. It provides in-process task execution, caching, retry loops, and terminal output recording.

Task Context Setup in _run_local​

When _Runner._run_local runs, it establishes local scratch directories and metadata paths:

metadata_path = pathlib.Path("/") / "tmp" / "flyte" / "metadata" / action.name
output_path = metadata_path / "a0"
raw_data_path = RawDataPath(path="/tmp/flyte/raw_data/" + action.name)

tctx = TaskContext(
    action=action,
    checkpoint_paths=CheckpointPaths(
        prev_checkpoint_path=storage_join(rd_base, "prev_checkpoint"),
        checkpoint_path=storage_join(rd_base, "checkpoint"),
    ),
    code_bundle=None,
    output_path=str(output_path),
    run_base_dir=str(metadata_path),
    version="na",
    raw_data_path=raw_data_path,
    compiled_image_cache=None,
    report=Report(name=action.name),
    mode="local",
    custom_context=self._custom_context,
    disable_run_cache=self._disable_run_cache,
)

Local Dispatch Lifecycle​

When LocalController.submit(_task, *args, **kwargs) is called:

  1. Deterministic Sub-Action IDs: TaskCallSequencer computes a sequence number for (_task, parent_action_name). Combined with the hash of the inputs, generate_sub_action_id_and_output_path creates deterministic sub-action identifiers and output directories.
  2. Local Cache Evaluation: If task.cache is enabled and tctx.disable_run_cache is False, the controller computes a cache key using task name, input hash, interface, and cache version. It queries LocalTaskCache.get(cache_key). On a cache hit, execution of the task body is skipped.
  3. Recording and Progress: RunRecorder registers the action start, capturing native input values, rendered log links, and parent action IDs for the terminal UI and persistence layer.
  4. Retry Loop with Checkpoint Staging: If the task fails, LocalController iterates up to max_attempts = retries + 1. On retry attempts (attempt_num > 1), _stage_prev_checkpoint_for_local_retry copies the current attempt's checkpoint to prev_checkpoint so the task can resume from state. If the error is non-recoverable (err.recoverable is False), retries terminate immediately.
  5. Execution Dispatch: Tasks are executed via direct_dispatch, which invokes _task.execute(*args, **kwargs).
  6. Thread Isolation via _TaskRunner: For synchronous submissions via submit_sync, LocalController maintains a thread map (_runner_map) where each calling thread is paired with a dedicated _TaskRunner running an asyncio event loop on a background daemon thread.

Remote Controller Architecture​

Remote execution is handled by RemoteController in flyte._internal.controllers.remote._controller. It interfaces with the Flyte control plane and data plane services.

Remote Run Submission in _run_remote​

When _Runner._run_remote is called:

  1. Code Bundling: Packages project code into a tarball (tgz) or pickle bundle (pkl) depending on copy_style and interactive_mode.
  2. Image Resolution: Calls build_images.aio(parent_env) unless dry-run is requested.
  3. Wire Translation & Upload: Converts the task into a Protobuf TaskSpec via translate_task_to_wire and serializes inputs.
  4. DataProxy & Run Creation: Uploads inputs using dataproxy_service.upload_inputs and creates the run on the cluster using run_service.create_run.
  5. Run Object: Returns a flyte.remote.Run instance representing the remote execution.

Sub-Action Dispatch and Remote Lifecycles​

Inside remote parent tasks or workflows running on the cluster, child tasks invoke RemoteController.submit(_task, *args, **kwargs):

┌───────────────────────────────────────────────────────────┐
│                 RemoteController.submit                   │
└─────────────────────────────┬─────────────────────────────┘
                              │
               Acquire Semaphore (_F_P_CNC)
                              │
               Upload Inputs with Retry
                              │
               Submit Action to Queue Service
                              │
              ┌───────────────┴───────────────┐
              │  Handle Execution Lifecycle   │
              └───────────────┬───────────────┘
                              │
        ┌─────────────────────┼─────────────────────┐
        ▼                     ▼                     ▼
┌──────────────┐      ┌──────────────┐      ┌──────────────┐
│  Cancelled   │      │   Aborted    │      │  Timed Out   │
│ Cancel action│      │    Raise     │      │    Raise     │
│  on server   │      │ActionAborted │      │ TaskTimeout  │
└──────────────┘      └──────────────┘      └──────────────┘
                              │
                     Download Realized
                     Outputs & Convert
  • Concurrency Limiting: Submissions are constrained per parent action by _parent_action_semaphore, configured by the environment variable _F_P_CNC (defaults to 1000).
  • Inline IO Bounds: Inputs and outputs must fit within max_inline_io_bytes (defaulting to MAX_INLINE_IO_BYTES = 10 MB). Larger datasets must use offloaded types (flyte.io.File, flyte.io.Dir, or DataFrames).
  • Action Cancellation: If the Python task coroutine is cancelled, RemoteController traps asyncio.CancelledError and explicitly cancels the action on the server.
  • Phase Handling: Terminal non-success phases raise specific exceptions:
    • ACTION_PHASE_ABORTED raises flyte.errors.ActionAbortedError.
    • ACTION_PHASE_TIMED_OUT raises flyte.errors.TaskTimeoutError.
    • ACTION_PHASE_FAILED invokes handle_action_failure to extract the underlying remote traceback and raise the corresponding error.

Hybrid Mode Architecture​

Hybrid execution bridges local and remote execution by running the entrypoint task function locally while executing all spawned child tasks remotely on the Flyte cluster.

import flyte

env = flyte.TaskEnvironment("hybrid-env")

@env.task
async def remote_child(x: int) -> int:
    return x * 10

@env.task
async def local_driver(n: int) -> int:
    total = 0
    for i in range(n):
        # Dispatched to remote cluster by RemoteController
        total += await remote_child(i)
    return total

# Requires remote storage configured (S3, GCS, or ABFS) and a run base directory
result = flyte.with_runcontext(
    mode="hybrid",
    run_base_dir="s3://my-flyte-bucket/hybrid-output",
).run(local_driver, n=5)

Hybrid Requirements and Execution Flow​

Under _Runner._run_hybrid:

  1. Remote Storage Validation: Hybrid execution requires a cloud storage backend (S3, GCS, or ABFS). If a local filesystem is detected, a ValueError is raised.
  2. Base Directory Enforcement: run_base_dir must be provided (e.g., s3://bucket/path) to store intermediate action inputs, outputs, and checkpoints.
  3. Remote Controller Injection: The runner initializes a RemoteController pointing to the cluster endpoint.
  4. Local Task Context: The entrypoint task is executed locally within run_task, while any nested task invocations are routed to RemoteController.submit(), dispatching them as actions onto the remote cluster.
  5. Output Extraction: When the local driver function returns, _run_hybrid returns the computed output values directly to the caller.