parsl.executors.HighThroughputExecutor

class parsl.executors.HighThroughputExecutor(label='HighThroughputExecutor', provider=LocalProvider( channel=LocalChannel( envs={}, script_dir=None, userhome='/home/docs/checkouts/readthedocs.org/user_builds/parsl/checkouts/latest/docs' ), cmd_timeout=30, init_blocks=4, launcher=SingleNodeLauncher(), max_blocks=10, min_blocks=0, move_files=None, nodes_per_block=1, parallelism=1, walltime='00:15:00', worker_init='' ), launch_cmd=None, address='127.0.0.1', worker_ports=None, worker_port_range=(54000, 55000), interchange_port_range=(55000, 56000), storage_access=None, working_dir=None, worker_debug=False, cores_per_worker=1.0, heartbeat_threshold=120, heartbeat_period=30, managed=True)[source]

Executor designed for cluster-scale

The HighThroughputExecutor system has the following components:
  1. The HighThroughputExecutor instance which is run as part of the Parsl script.
  2. The Interchange which is acts as a load-balancing proxy between workers and Parsl
  3. The multiprocessing based worker pool which coordinates task execution over several cores on a node.
  4. ZeroMQ pipes connect the HighThroughputExecutor, Interchange and the process_worker_pool

Here is a diagram

             |  Data   |  Executor   |  Interchange  | External Process(es)
             |  Flow   |             |               |
        Task | Kernel  |             |               |
      +----->|-------->|------------>|->outgoing_q---|-> process_worker_pool
      |      |         |             | batching      |    |         |
Parsl<---Fut-|         |             | load-balancing|  result   exception
          ^  |         |             | watchdogs     |    |         |
          |  |         |   Q_mngmnt  |               |    V         V
          |  |         |    Thread<--|-incoming_q<---|--- +---------+
          |  |         |      |      |               |
          |  |         |      |      |               |
          +----update_fut-----+
Parameters:
  • provider (ExecutionProvider) –
    Provider to access computation resources. Can be one of EC2Provider,
    Cobalt, Condor, GoogleCloud, GridEngine, Jetstream, Local, GridEngine, Slurm, or Torque.
  • label (str) – Label for this executor instance.
  • launch_cmd (str) – Command line string to launch the process_worker_pool from the provider.
  • address (string) – An address to connect to the main Parsl process which is reachable from the network in which workers will be running. This can be either a hostname as returned by hostname or an IP address. Most login nodes on clusters have several network interfaces available, only some of which can be reached from the compute nodes. Some trial and error might be necessary to indentify what addresses are reachable from compute nodes.
  • worker_ports ((int, int)) – Specify the ports to be used by workers to connect to Parsl. If this option is specified, worker_port_range will not be honored.
  • worker_port_range ((int, int)) – Worker ports will be chosen between the two integers provided.
  • interchange_port_range ((int, int)) – Port range used by Parsl to communicate with the Interchange.
  • working_dir (str) – Working dir to be used by the executor.
  • worker_debug (Bool) – Enables worker debug logging.
  • managed (Bool) – If this executor is managed by the DFK or externally handled.
  • cores_per_worker (float) – cores to be assigned to each worker. Oversubscription is possible by setting cores_per_worker < 1.0. Default=1
  • heartbeat_threshold (int) – Seconds since the last message from the counterpart in the communication pair: (interchange, manager) after which the counterpart is assumed to be un-available. Default:120s
  • heartbeat_period (int) – Number of seconds after which a heartbeat message indicating liveness is sent to the counterpart (interchange, manager). Default:30s
__init__(label='HighThroughputExecutor', provider=LocalProvider( channel=LocalChannel( envs={}, script_dir=None, userhome='/home/docs/checkouts/readthedocs.org/user_builds/parsl/checkouts/latest/docs' ), cmd_timeout=30, init_blocks=4, launcher=SingleNodeLauncher(), max_blocks=10, min_blocks=0, move_files=None, nodes_per_block=1, parallelism=1, walltime='00:15:00', worker_init='' ), launch_cmd=None, address='127.0.0.1', worker_ports=None, worker_port_range=(54000, 55000), interchange_port_range=(55000, 56000), storage_access=None, working_dir=None, worker_debug=False, cores_per_worker=1.0, heartbeat_threshold=120, heartbeat_period=30, managed=True)[source]

Initialize self. See help(type(self)) for accurate signature.

Methods

__init__([label, provider, launch_cmd, …]) Initialize self.
hold_worker(worker_id) Puts the workers on hold, preventing scheduling of additional tasks to it.
scale_in(blocks) Scale in the number of active blocks by specified amount.
scale_out([blocks]) Scales out the number of blocks by “blocks”
shutdown([hub, targets, block]) Shutdown the executor, including all workers and controllers.
start() Create the Interchange process and connect to it.
status() Return status of all blocks.
submit(func, *args, **kwargs) Submits work to the the outgoing_q.
weakref_cb([q]) We do not use this yet.

Attributes

connected_workers
outstanding
run_dir Path to the run directory.
scaling_enabled Specify if scaling is enabled.