taps.executor.parsl

taps/executor/parsl.py

ParslLocalConfig

Bases: ExecutorConfig

Local ParslPoolExecutor plugin configuration.

Simple Parsl configuration that uses the HighThroughputExecutor on the local node.

Attributes:

• workers (Optional[int]) –

  Maximum number of Parsl workers.

• run_dir (str) –

  Parsl run directory.

get_executor()

get_executor() -> ParslPoolExecutor

Create an executor instance from the config.

Source code in taps/executor/parsl.py

def get_executor(self) -> ParslPoolExecutor:
    """Create an executor instance from the config."""
    workers = (
        self.workers
        if self.workers is not None
        else multiprocessing.cpu_count()
    )
    executor = HighThroughputExecutor(
        label='taps-htex-local',
        max_workers_per_node=workers,
        address=address_by_hostname(),
        cores_per_worker=1,
        provider=LocalProvider(
            channel=LocalChannel(),
            init_blocks=1,
            max_blocks=1,
        ),
    )
    config = Config(
        executors=[executor],
        run_dir=self.run_dir,
        initialize_logging=False,
    )
    return ParslPoolExecutor(config)

ParslHTExConfig

Bases: ExecutorConfig

HTEx ParslPoolExecutor plugin configuration.

These parameters are a subset of parsl.config.Config.

For simple, single-node HTEx deployments, prefer ParslLocalConfig.

Attributes:

• htex (HTExConfig) –

  HTEx configuration.

• app_cache (Optional[bool]) –

  Enable app caching.

• retries (int) –

  Number of retries in case of task failure.

• strategy (Optional[str]) –

  Block scaling strategy.

• max_idletime (Optional[float]) –

  Max idle time before strategy can shutdown unused blocks.

• run_dir (str) –

  Parsl run directory.

get_executor()

get_executor() -> ParslPoolExecutor

Create an executor instance from the config.

Source code in taps/executor/parsl.py

def get_executor(self) -> ParslPoolExecutor:
    """Create an executor instance from the config."""
    options = self.model_dump(exclude={'name', 'htex'}, exclude_none=True)

    config = Config(
        executors=[self.htex.get_executor()],
        initialize_logging=False,
        **options,
    )
    return ParslPoolExecutor(config)

HTExConfig

Bases: BaseModel

Configuration for Parl's parsl.executors.HighThroughputExecutor.

Note

Optional attributes will default to Parsl's default values.

Note

Extra options passed to this model will be provided as keyword arguments to parsl.executors.HighThroughputExecutor.

Attributes:

• provider (Optional[ProviderConfig]) –

  Configuration for the compute resource provider.

• address (Optional[Union[str, AddressConfig]]) –

  Address to connect to the main Parsl process.

• worker_ports (Optional[Tuple[int, int]]) –

  Ports used by workers to connect to Parsl.

• worker_port_range (Optional[Tuple[int, int]]) –

  Range of ports to choose worker ports from.

• interchange_port_range (Optional[Tuple[int, int]]) –

  Ports used by Parsl to connect to the interchange.

get_executor()

get_executor() -> HighThroughputExecutor

Create an executor instance from the config.

Source code in taps/executor/parsl.py

def get_executor(self) -> HighThroughputExecutor:
    """Create an executor instance from the config."""
    options = self.model_dump(exclude_none=True)
    if self.model_extra is not None:  # pragma: no branch
        options.update(self.model_extra)

    if self.provider is not None:
        options['provider'] = self.provider.get_provider()
    if self.address is not None and isinstance(
        self.address,
        AddressConfig,
    ):
        options['address'] = self.address.get_address()

    return HighThroughputExecutor(label='taps-htex', **options)

AddressConfig

Bases: BaseModel

Parsl address configuration.

Example

from parsl.addresses import address_by_interface
from taps.executor.config import AddressConfig

config = AddressConfig(kind='address_by_interface', ifname='bond0')
assert config.get_address() == address_by_interface(ifname='bond0')

get_address()

get_address() -> str

Get the address according to the configuration.

Source code in taps/executor/parsl.py

def get_address(self) -> str:
    """Get the address according to the configuration."""
    address_fn = getattr(parsl.addresses, self.kind)
    options = self.model_extra if self.model_extra is not None else {}
    return address_fn(**options)

ProviderConfig

Bases: BaseModel

Parsl execution provider configuration.

Example

Create a provider configuration and call get_provider().

from taps.executor.config import ProviderConfig

config = ProviderConfig(
    kind='PBSProProvider',
    account='my-account',
    cpus_per_node=32,
    init_blocks=1,
    max_blocks=1,
    min_blocks=0,
    nodes_per_block=1,
    queue='debug',
    select_options='ngpus=4',
    walltime='00:30:00',
    worker_init='module load conda',
)
config.get_provider()

The resulting provider is equivalent to creating it manually.

from parsl.providers import PBSProProvider

PBSProProvider(
    account='my-account',
    cpus_per_node=32,
    init_blocks=1,
    max_blocks=1,
    min_blocks=0,
    nodes_per_block=1,
    queue='debug',
    select_options='ngpus=4',
    walltime='00:30:00',
    worker_init='module load conda',
),

get_provider()

get_provider() -> ExecutionProvider

Create a provider from the configuration.

Source code in taps/executor/parsl.py

def get_provider(self) -> ExecutionProvider:
    """Create a provider from the configuration."""
    options = self.model_extra if self.model_extra is not None else {}

    if self.launcher is not None:
        options['launcher'] = self.launcher.get_launcher()

    provider_cls = getattr(parsl.providers, self.kind)
    return provider_cls(**options)

LauncherConfig

Bases: BaseModel

Parsl launcher configuration.

Example

Create a launcher configuration and call get_launcher().

from taps.executor.config import LauncherConfig

config = LauncherConfig(
    kind='MpiExecLauncher',
    bind_cmd='--cpu-bind',
    overrides='--depth=64 --ppn=1,
)
config.get_launcher()

The resulting launcher is equivalent to creating it manually.

from parsl.launchers import MpiExecLauncher

MpiExecLauncher(bind_cmd='--cpu-bind', overrides='--depth=64 --ppn 1')

get_launcher()

get_launcher() -> Launcher

Create a launcher from the configuration.

Source code in taps/executor/parsl.py

def get_launcher(self) -> Launcher:
    """Create a launcher from the configuration."""
    launcher_cls = getattr(parsl.launchers, self.kind)
    options = self.model_extra if self.model_extra is not None else {}
    return launcher_cls(**options)