Parsl

1 The API

Snakemake uses its own DSL (Domain-Specific Language) to configure jobs. I hated that. It broke IDE support and linting, was an unnecessary abstraction over Python, and cemented my side-eyed distrust of DSLs.

Parsl avoids that by being pure python.

In Parsl, we define workflows by annotating standard Python functions with decorators. We call these annotated functions “Apps”.

import parsl
from parsl import python_app, bash_app

# A Parsl App that executes Python code
@python_app
def process_data(input_data):
    # ... complex processing ...
    return result

# A Parsl App that executes a shell command
@bash_app
def run_simulation(inputs, outputs):
    return f"my-simulator --in {inputs[0]} --out {outputs[0]}"

When we call a Parsl App, it doesn’t execute immediately. Instead, it returns a Future—an object representing the eventual result of the computation. This is a relatively modern approach in Python and is probably the ‘right’ way to do things.

If we pass the Future returned by one App as an argument to another App, Parsl automatically recognizes the dependency.

# Call the first app
future1 = process_data(initial_value)

# Call the second app, passing the future from the first
# Parsl knows this task depends on the completion of the first
future2 = process_data(future1)

# Execution happens asynchronously. We only block when we ask for the result.
print(future2.result())

Parsl constructs the DAG from these implicit data flows. When all inputs (Futures) for a task are ready, Parsl schedules the task on available resources. This pure-Python approach feels intuitive, integrates well with modern development tools, and lets us write complex, dynamic logic that’s impossible in a static DSL.

2 How Cluster Execution Works

The main reason I tolerate workflow managers at all is that they handle the nightmare campus cluster horrors. Parsl has tight integration with batch schedulers like Slurm, PBS, SGE, and HTCondor, as well as with cloud providers and Kubernetes.

In Snakemake, we combine rules (with resource hints) and a separate YAML “profile” to map those hints to scheduler flags.

In Parsl, we handle configuration entirely within the Python script using a Config object. This object defines where and how tasks should run by combining Executors and Providers.

1. Providers: These handle the interaction with the resource manager (e.g., SlurmProvider). They’re responsible for requesting, scaling, and terminating “blocks” of resources (like N nodes on a cluster).
2. Executors: These manage task execution on the resources acquired by a Provider. The HighThroughputExecutor (HTEX) is commonly used for HPC scenarios, efficiently distributing tasks across many workers.

Here is a simple example of a configuration for a Slurm cluster:

from parsl import Config
from parsl.executors import HighThroughputExecutor
from parsl.providers import SlurmProvider
import os

# Get dynamic values from the environment
slurm_account = os.getenv("SLURM_ACCOUNT", "default_account")
partition = os.getenv("SLURM_PARTITION", "standard")

config = Config(
    executors=[
        HighThroughputExecutor(
            label="my_hpc_cluster",
            max_workers_per_node=48,
            provider=SlurmProvider(
                account=slurm_account,
                partition=partition,
                nodes_per_block=10,  # Request 10 nodes per Slurm job
                init_blocks=1,
                max_blocks=5,        # Scale up to 5 blocks (50 nodes total)
                walltime="02:00:00",
                # Optional: Add specific scheduler options
                # scheduler_options="#SBATCH --gpus-per-node=4"
            ),
        )
    ]
)

# Load the configuration before executing the workflow
# import parsl
# parsl.load(config)

When the script runs, Parsl uses the SlurmProvider to submit sbatch jobs (the “blocks”). Once those jobs start, the HighThroughputExecutor connects to the allocated nodes and begins distributing the workflow tasks across them.

3 Python Configuration

In my experience with Snakemake (especially version 8+), configuring cluster execution was frustrating. Snakemake’s YAML-based executor profiles didn’t let us use environment variables.

This rigidity forced me to hard-code site-specific details, producing a proliferation of near-identical config files and terrible portability. Ultimately, I was generating YAML files programmatically just to work around the limitations.

Parsl’s configuration system is a massive improvement because it’s just Python. See the Parsl example above. We can use os.getenv() to dynamically pull the Slurm account or partition. We can use conditional logic, loops, and functions to construct the configuration object. We can easily integrate it with automatic config tools.

This flexibility is neat for writing portable workflows. We can define a single script that intelligently adapts to the environment it’s running in, whether that’s a local machine or a different HPC cluster, without the configuration-boilerplate nightmare that plagued my Snakemake setup.

4 Local execution

The oldest local executor is the ThreadPoolExecutor. Threads are almost always more trouble than they’re worth in Python, IMO. If I wanted to spend time debugging non-deterministic segfaults, I’d switch to writing C.

The HighThroughputExecutor, however, seems to support multiprocessing backends.

5 Providers

Parsl supports various Providers for public cloud backends (AWS, Google Cloud, Azure) and Kubernetes. Because the workflow logic (the Apps and their dependencies) is completely decoupled from the execution configuration (the Config object), running the same analysis on-prem or in the cloud often just means loading a different configuration object. It probably gets less seamless when I need to manage massive data assets, but I’ve managed to avoid that so far. TBC

6 Debugging Parsl workers

When Parsl tasks fail, the default error reporting is often opaque and unhelpful:

[1/4] ✗ FAILED: tgt_deaf1a769193/hmc/f4567256
  Error: Dependency failure for task 1. The representative cause is via task 0

What’s happening:

• Parsl workers run in separate processes (via HTEX)
• When a worker crashes, the exception is wrapped in Parsl’s internal classes
• The actual Python traceback and error message are hidden inside the wrapper
• We only see “Dependency failure” without knowing the root cause

This is infuriating.

• No visibility into what actually went wrong
• We can’t distinguish between import errors, type errors, or logic bugs
• We have to manually inspect worker log files (if they exist)
• Errors propagate through dependency chains, hiding the original failure

What follows are some design patterns and code snippets to improve error visibility in Parsl workflows.

def unwrap_parsl_future(future, name: str):
    """Extract and surface exceptions from Parsl futures with full diagnostics.

    Args:
        future: Parsl AppFuture to unwrap
        name: Descriptive name for logging (e.g., "build_tgt_abc123")

    Returns:
        Future result if successful

    Raises:
        Original exception with enhanced logging of remote stdout/stderr
    """
    try:
        return future.result()
    except Exception as e:
        import traceback
        log.error(f"[{name}] FAILED in worker:")
        log.error("".join(traceback.format_exception(type(e), e, e.__traceback__)))

        # Dump remote debug info if available
        if hasattr(e, 'stdout') and e.stdout:
            log.error("---- WORKER STDOUT ----\n%s", e.stdout)
        if hasattr(e, 'stderr') and e.stderr:
            log.error("---- WORKER STDERR ----\n%s", e.stderr)

        # Log exception attributes for debugging Parsl wrappers
        log.error("Exception type: %s", type(e).__name__)
        log.error("Exception attributes: %s", dir(e))

        raise

for i, (future, record) in enumerate(zip(run_futures, run_records), 1):
    try:
        future.result()  # ❌ Hides worker exceptions
        log.info("  [%d/%d] ✓ %s", i, len(run_futures), record['run_id'])
    except Exception as e:
        log.error("  [%d/%d] ✗ FAILED: %s", i, len(run_futures), record['run_id'])
        log.error("    Error: %s", str(e))  # Only sees "Dependency failure"

for i, (future, record) in enumerate(zip(run_futures, run_records), 1):
    try:
        name = f"{record['target_id']}_{record['sampler']}_{record['run_id']}"
        unwrap_parsl_future(future, name)  # ✅ Extracts full traceback
        log.info("  [%d/%d] ✓ %s", i, len(run_futures), record['run_id'])
    except Exception as e:
        log.error("  [%d/%d] ✗ FAILED: %s", i, len(run_futures), record['run_id'])
        log.error("    Error: %s", str(e))

executors:
  local_htex:
    worker_logdir_root: /path/to/artifacts/run_dir/parsl_workers
    max_workers: 4

@python_app
def build_target_app(cfg_yaml, target_id, experiment):
    """Build a target via direct command call."""
    import logging
    log = logging.getLogger(__name__)

    log.info("[WORKER START] build_target_app(target_id=%s)", target_id)

    from lambda_hat.commands.build_cmd import build_entry
    result = build_entry(cfg_yaml, target_id, experiment)

    log.info("[WORKER END] build_target_app(target_id=%s)", target_id)
    return result

7 Parsl Worker Error Zoo

I extracted this list of failures from the logs of my recent Parsl runs and had the LLM summarise them.

# config/parsl/local.yaml
executors:
  local_htex:
    worker_init: |
      source /path/to/venv/bin/activate
      export PYTHONPATH=/path/to/project

failures = []

for i, (future, record) in enumerate(zip(run_futures, run_records), 1):
    try:
        unwrap_parsl_future(future, record['run_id'])
        log.info("  [%d/%d] ✓ %s", i, len(run_futures), record['run_id'])
    except Exception as e:
        log.error("  [%d/%d] ✗ FAILED: %s", i, len(run_futures), record['run_id'])
        failures.append({'record': record, 'error': str(e)})

# Report all failures at end
if failures:
    log.error("⚠ FAILURE SUMMARY: %d of %d runs failed", len(failures), len(run_futures))
    for f in failures:
        log.error("  • %s: %s", f['record']['run_id'], f['error'])

def unwrap_with_retry(future, name: str, max_retries: int = 3):
    """Extract error and optionally retry on transient failures."""
    for attempt in range(max_retries):
        try:
            return unwrap_parsl_future(future, name)
        except (TimeoutError, ConnectionError) as e:
            if attempt < max_retries - 1:
                log.warning(f"[{name}] Retry {attempt+1}/{max_retries} after: {e}")
                continue
            raise

import json

def unwrap_to_json(future, name: str, output_path: Path):
    """Extract error and write structured JSON for analysis."""
    try:
        result = future.result()
        output_path.write_text(json.dumps({
            'name': name,
            'status': 'success',
            'result': str(result),
        }))
        return result
    except Exception as e:
        import traceback
        output_path.write_text(json.dumps({
            'name': name,
            'status': 'failed',
            'error_type': type(e).__name__,
            'error_message': str(e),
            'traceback': traceback.format_exception(type(e), e, e.__traceback__),
        }, indent=2))
        raise