Support & Contribution

Support & Contribution#

The executorlib open-source software package is developed by scientists for scientists. We are open for any contribution, from feedback about spelling mistakes in the documentation, to raising issues about functionality which is insufficiently explained in the documentation or simply requesting support to suggesting new features or opening pull requests. Our Github repository is the easiest way to get in contact with the developers.

Issues#

The easiest way for us as developers to help in solving an issue is to provide us with sufficient information about how to reproduce the issue. The simpler the test case which causes the issue the easier it is to identify the part of the code which is causing the issue. As a general rule of thumb, everything that works with the ProcessPoolExecutor or the ThreadPoolExecutor should also work with the different Executor classes provided by executorlib. If this is not the case, then it is most likely a bug and worth reporting.

Pull Requests#

Reviewing a pull request is easier when the changes are clearly lined out, covered by tests and following the automated formatting using black. Still when you decide to work on a new feature it can also be helpful to open a pull request early on and mark it as draft, this gives other developers the opportunity to see what you are working on.

License#

BSD 3-Clause License

Copyright (c) 2022, Jan Janssen
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:

* Redistributions of source code must retain the above copyright notice, this
  list of conditions and the following disclaimer.

* Redistributions in binary form must reproduce the above copyright notice,
  this list of conditions and the following disclaimer in the documentation
  and/or other materials provided with the distribution.

* Neither the name of the copyright holder nor the names of its
  contributors may be used to endorse or promote products derived from
  this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

Modules#

While it is not recommended to link to specific internal components of executorlib in external Python packages but rather only the Executor classes should be used as central interfaces to executorlib, the internal architecture is briefly outlined below.

backend - the backend module contains the functionality for the Python processes created by executorlib to execute the submitted Python functions.
executor - the executor module defines the different Executor classes, namely SingleNodeExecutor, SlurmClusterExecutor, SlurmJobExecutor, FluxClusterExecutor and FluxJobExecutor. These are the interfaces the user interacts with.
standalone - the standalone module contains a number of utility functions which only depend on external libraries and do not have any internal dependency to other parts of executorlib. This includes the functionality to generate executable commands, the h5py based interface for caching, a number of input checks, routines to plot the dependencies of a number of future objects, functionality to interact with the queues defined in the Python standard library, the interface for serialization based on cloudpickle and finally an extension to the threading of the Python standard library.
task_scheduler - the internal task scheduler module defines the task schedulers, namely BlockAllocationTaskScheduler, DependencyTaskScheduler, FileTaskScheduler and OneProcessTaskScheduler. They are divided into two sub modules:
- file - the file based task scheduler module defines the file based communication for the HPC Cluster Executor.
- interactive - the interactive task scheduler module defines the zero message queue based communication for the Single Node Executor and the HPC Job Executor.

Given the level of separation the integration of submodules from the standalone module in external software packages should be the easiest way to benefit from the developments in executorlib beyond just using the Executor class.

Interface Class Hierarchy#

executorlib provides five different interfaces, namely SingleNodeExecutor, SlurmClusterExecutor, SlurmJobExecutor, FluxClusterExecutor and FluxJobExecutor, internally these are mapped to four types of task schedulers, namely BlockAllocationTaskScheduler, DependencyTaskScheduler, FileTaskScheduler and OneProcessTaskScheduler depending on which options are selected. Finally, the task schedulers are connected to spawners to start new processes, namely the MpiExecSpawner, SrunSpawner and FluxPythonSpawner. The dependence is illustrated in the following table:

	`BlockAllocationTaskScheduler`	`DependencyTaskScheduler`	`FileTaskScheduler`	`OneProcessTaskScheduler`
`SingleNodeExecutor(disable_dependencies=False)`		with `MpiExecSpawner`
`SingleNodeExecutor(disable_dependencies=True, block_allocation=False)`				with `MpiExecSpawner`
`SingleNodeExecutor(disable_dependencies=True, block_allocation=True)`	with `MpiExecSpawner`
`SlurmClusterExecutor(plot_dependency_graph=False)`			with `pysqa`
`SlurmClusterExecutor(plot_dependency_graph=True)`		with `SrunSpawner`
`SlurmJobExecutor(disable_dependencies=False)`		with `SrunSpawner`
`SlurmJobExecutor(disable_dependencies=True, block_allocation=False)`				with `SrunSpawner`
`SlurmJobExecutor(disable_dependencies=True, block_allocation=True)`	with `SrunSpawner`
`FluxClusterExecutor(plot_dependency_graph=False)`			with `pysqa`
`FluxClusterExecutor(plot_dependency_graph=True)`		with `FluxPythonSpawner`
`FluxJobExecutor(disable_dependencies=False)`		with `FluxPythonSpawner`
`FluxJobExecutor(disable_dependencies=True, block_allocation=False)`				with `FluxPythonSpawner`
`FluxJobExecutor(disable_dependencies=True, block_allocation=True)`	with `FluxPythonSpawner`

The following diagram illustrates the relationship between the main classes in executorlib.

        classDiagram
    class FutureExecutor {
        <<interface>>
    }
    class BaseExecutor {
        -_task_scheduler: TaskSchedulerBase
        +submit(fn, *args, **kwargs) Future
        +shutdown(wait)
    }
    class TaskSchedulerBase {
        -_future_queue: Queue
        -_process: Thread
        +submit(fn, *args, **kwargs) Future
    }
    class BaseSpawner {
        <<interface>>
        +bootup(command_lst)
        +shutdown(wait)
    }
    class SocketInterface {
        +send_dict(input_dict)
        +receive_dict() dict
    }

    FutureExecutor <|-- BaseExecutor
    BaseExecutor o-- TaskSchedulerBase
    TaskSchedulerBase <|-- OneProcessTaskScheduler
    TaskSchedulerBase <|-- BlockAllocationTaskScheduler
    TaskSchedulerBase <|-- DependencyTaskScheduler
    TaskSchedulerBase <|-- FileTaskScheduler

    OneProcessTaskScheduler o-- BaseSpawner
    BaseSpawner <|-- MpiExecSpawner
    BaseSpawner <|-- SrunSpawner
    BaseSpawner <|-- FluxPythonSpawner

    OneProcessTaskScheduler ..> SocketInterface : uses

Execution Flow#

When a user submits a function to an executor, several steps occur in the background to ensure the task is executed with the requested resources and the result is returned.

        sequenceDiagram
    participant User
    participant Executor
    participant TaskScheduler
    participant Spawner
    participant Backend

    User->>Executor: submit(fn, args, resource_dict)
    Executor->>TaskScheduler: submit(fn, args, resource_dict)
    TaskScheduler->>TaskScheduler: Add to _future_queue
    TaskScheduler-->>User: Return Future object

    Note over TaskScheduler, Spawner: Task loop in background thread

    TaskScheduler->>Spawner: bootup(command)
    Spawner->>Backend: Start worker process
    TaskScheduler->>Backend: Send function and arguments (ZMQ/File)
    Backend->>Backend: Execute function
    Backend->>TaskScheduler: Send result (ZMQ/File)
    TaskScheduler->>User: Update Future with result

Test Environment#

The test environment of the executorlib library consists of three components - they are all available in the executorlib Github repository:

The Jupyter Notebooks in the executorlib Github repository demonstrate the usage of executorlib. These notebooks are used as examples for new users, as documentation available on readthedocs.org and as integration tests.
The likelihood benchmark to compare the performance on a single compute node to the built-in interfaces in the standard library. The benchmark can be run with the following parameters python llh.py static. Here static refers to single process execution, process refers to the ProcessPoolExecutor from the standard library, thread refers to the ThreadPoolExecutor from the standard library, executorlib refers to the SingleNodeExecutor in executorlib and block_allocation to the SingleNodeExecutor in executorlib with block allocation enabled. Finally, for comparison to mpi4py the test can be executed with mpiexec -n 4 python -m mpi4py.futures llh.py mpi4py.
The unit tests these can be executed with python -m unittest discover . in the tests directory. The tests are structured based on the internal structure of executorlib. Tests for the SingleNodeExecutor are named test_singlenodeexecutor_*.py and correspondingly for the other modules.

Communication#

Interactive Communication#

The key functionality of the executorlib package is the up-scaling of python functions with thread based parallelism, MPI based parallelism or by assigning GPUs to individual python functions. In the background this is realized using a combination of the zero message queue and cloudpickle to communicate binary python objects. The executorlib.standalone.interactive.communication.SocketInterface is an abstraction of this interface, which is used in the other classes inside executorlib and might also be helpful for other projects. It comes with a series of utility functions:

executorlib.standalone.interactive.communication.interface_bootup(): To initialize the interface
executorlib.standalone.interactive.communication.interface_connect(): To connect the interface to another instance
executorlib.standalone.interactive.communication.interface_send(): To send messages via this interface
executorlib.standalone.interactive.communication.interface_receive(): To receive messages via this interface
executorlib.standalone.interactive.communication.interface_shutdown(): To shutdown the interface

While executorlib was initially designed for up-scaling python functions for HPC, the same functionality can be leveraged to up-scale any executable independent of the programming language it is developed in.

File-based Communication#

Used by FluxClusterExecutor, SlurmClusterExecutor and TestClusterExecutor. It uses the filesystem to communicate between the main process and the individual HPC jobs. This mode is necessary when tasks are submitted as independent jobs to a scheduler like SLURM or Flux, where direct network communication between the login node and compute nodes might be restricted.

External Libraries#

For external libraries executorlib provides a standardized interface for a subset of its internal functionality, which is designed to remain stable with minor version updates. Developers can import the following functionality from executorlib.api:

cancel_items_in_queue() - Cancel items which are still waiting in the Python standard library queue - queue.queue.
cloudpickle_register() - Cloudpickle can either pickle by value or pickle by reference. The functions which are communicated have to be pickled by value rather than by reference, so the module which calls the map function is pickled by value.
get_command_path() - Get path of the backend executable script executorlib.backend.
interface_bootup() - Start interface for ZMQ communication.
interface_connect() - Connect to an existing SocketInterface instance by providing the hostname and the port as strings.
interface_receive() - Receive instructions from a SocketInterface instance.
interface_send() - Send results to a SocketInterface instance.
interface_shutdown() - Close the connection to a SocketInterface instance.
MpiExecSpawner - Subprocess interface to start mpi4py parallel process.
SocketInterface - The SocketInterface is an abstraction layer on top of the zero message queue.
SubprocessSpawner - Subprocess interface to start serial Python process.

It is not recommended to import components from other parts of executorlib in other libraries, only the interfaces in executorlib and executorlib.api are designed to be stable. All other classes and functions are considered for internal use only.

External Executables#

On extension beyond the submission of Python functions is the communication with an external executable. This could be any kind of program written in any programming language which does not provide Python bindings so it cannot be represented in Python functions.

Subprocess#

If the external executable is called only once, then the call to the external executable can be represented in a Python function with the subprocess module of the Python standard library. In the example below the shell command echo test is submitted to the execute_shell_command() function, which itself is submitted to the Executor class.

from executorlib import SingleNodeExecutor

def execute_shell_command(
    command: list, universal_newlines: bool = True, shell: bool = False
):
    import subprocess

    return subprocess.check_output(
        command, universal_newlines=universal_newlines, shell=shell
    )

with SingleNodeExecutor() as exe:
    future = exe.submit(
        execute_shell_command,
        ["echo", "test"],
        universal_newlines=True,
        shell=False,
    )
    print(future.result())

test

Interactive#

The more complex case is the interaction with an external executable during the run time of the executable. This can be implemented with executorlib using the block allocation block_allocation=True feature. The external executable is started as part of the initialization function init_function and then the indivdual functions submitted to the Executor class interact with the process which is connected to the external executable.

Starting with the definition of the executable, in this example it is a simple script which just increases a counter. The script is written in the file count.py so it behaves like an external executable, which could also use any other progamming language.

count_script = """\
def count(iterations):
    for i in range(int(iterations)):
        print(i)
    print("done")


if __name__ == "__main__":
    while True:
        user_input = input()
        if "shutdown" in user_input:
            break
        else:
            count(iterations=int(user_input))
"""

with open("count.py", "w") as f:
    f.writelines(count_script)

The connection to the external executable is established in the initialization function init_function of the Executor class. By using the subprocess module from the standard library two process pipes are created to communicate with the external executable. One process pipe is connected to the standard input stdin and the other is connected to the standard output stdout.

def init_process():
    import subprocess

    return {
        "process": subprocess.Popen(
            ["python", "count.py"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            universal_newlines=True,
            shell=False,
        )
    }

The interaction function handles the data conversion from the Python datatypes to the strings which can be communicated to the external executable. It is important to always add a new line \n to each command send via the standard input stdin to the external executable and afterwards flush the pipe by calling flush() on the standard input pipe stdin.

def interact(shell_input, process, lines_to_read=None, stop_read_pattern=None):
    process.stdin.write(shell_input)
    process.stdin.flush()
    lines_count = 0
    output = ""
    while True:
        output_current = process.stdout.readline()
        output += output_current
        lines_count += 1
        if stop_read_pattern is not None and stop_read_pattern in output_current:
            break
        elif lines_to_read is not None and lines_to_read == lines_count:
            break
    return output

Finally, to close the process after the external executable is no longer required it is recommended to define a shutdown function, which communicates to the external executable that it should shutdown. In the case of the count.py script defined above this is achieved by sending the keyword shutdown.

def shutdown(process):
    process.stdin.write("shutdown\n")
    process.stdin.flush()

With these utility functions is to possible to communicate with any kind of external executable. Still for the specific implementation of the external executable it might be necessary to adjust the corresponding Python functions. Therefore this functionality is currently limited to developers and not considered a general feature of executorlib.

with SingleNodeExecutor(
    max_workers=1,
    init_function=init_process,
    block_allocation=True,
) as exe:
    future = exe.submit(
        interact, shell_input="4\n", lines_to_read=5, stop_read_pattern=None
    )
    print(future.result())
    future_shutdown = exe.submit(shutdown)
    print(future_shutdown.result())

0
1
2
3
done

None