Explanation is given in mpi4py/mpi4py#324

In addition to CPU parallelisation and GPU support it would be great to add OpenMP parallelisation.

Following the discussion mpi4py/mpi4py#370 it would be great to implement a compatible interface for pympipool.

Use the mpi4py interface to dynamically assign and remove executors in a larger network with multiple executors. Currently, the user would have to manage the network of executors on their own, but having the management part inside the MPI parallel process could be beneficial.

Allow users to define their own backend to start the parallel process. For example this could be a pysqa compatible backend.

Currently, only MPI is used for parallelization, but in principle it should also be possible to use OpenMP in addition to MPI.

Using PoolExecutor incurs strangely high CPU usage. The MWE below shows 40% cpu on both the process that runs it as well as the mpipool.py process spawned in the background. After the futures are ready the main process blocks a full core until the executor shuts down. It sounds as if processes are spinning on the ZMQ socket or there are extra messages sent back and forth all the time, because I can make the CPU usage go down while the futures are running by adding sleeps or prints into the main loop of mpipool.py. CPU usage remains high until the executor shuts down though even in that case. I haven't figured out yet how to monitor the ZMQ socket though, so I cannot confirm that hypothesis.

Using the ProcessPoolExecutor from the standard library this does not happen.

import time
from pympipool import PoolExecutor

def calc(n):
    time.sleep(30)
    return n**2

exe = PoolExecutor(max_workers=4)
f1 = exe.submit(calc, 42)
f2 = exe.submit(calc, 84)
f1.add_done_callback(print)
f2.add_done_callback(print)
time.sleep(60)  # workaround #94 
exe.shutdown(wait=True)

In particular this backend could be used for systems without MPI and the nested execution of flux executors.

Currently it is not possible to nest flux executors, still this would be a great feature to add.

Many clusters provide interactive queues, so we could use those queues to start the executor and then send functions to the executor via the zero message interface like it is done now.

How does pympipool compare to ThreadPoolExecutor and ProcessPoolExecutor ? It would be great to design benchmark studies. In addition, pympipool can be combined with the ProcessPoolExecutor when specifying the number of threads and running one pympipool process per node.

Concept:

from concurrent.futures import Executor, Future, ProcessPoolExecutor
import queue
from threading import Thread
from time import sleep

from pympipool.shared import cancel_items_in_queue


def get_callback(future_external):
    def callback(fn):
        if fn.cancelled():
            future_external.cancel()
        else:
            future_external.set_result(fn.result())
    return callback   


def refresh(running_dict, todo_lst, max_workers, exe):
    running_dict = {k: v for k, v in running_dict.items() if not k.done()}
    cores_available = max_workers - sum(running_dict.values())
    if len(todo_lst) > 0 and cores_available > todo_lst[0]["cores"]:
        task_dict = todo_lst.pop(0)
        fn_external = task_dict["future"]
        if fn_external.set_running_or_notify_cancel():
            running_dict[fn_external] = task_dict["cores"]
            fn_internal = exe.submit(task_dict["fn"], *task_dict["args"], **task_dict["kwargs"])
            fn_internal.add_done_callback(get_callback(future_external=fn_external))
        refresh(running_dict=running_dict, todo_lst=todo_lst, max_workers=max_workers, exe=exe)


def execute_task_dict(task_dict, todo_lst):
    if "fn" in task_dict.keys() or "future" in task_dict.keys():
        todo_lst.append(task_dict)
        return True
    elif "shutdown" in task_dict.keys() and task_dict["shutdown"]:
        return False
    else:
        raise ValueError("Unrecognized Task in task_dict: ", task_dict)


def background_execution(future_queue, exe_args, exe_kwargs, sleep_interval=0.1):
    todo_lst = []
    running_dict = {}
    with ProcessPoolExecutor(*exe_args, **exe_kwargs) as exe:
        while True:
            try:
                task_dict = future_queue.get_nowait()
            except queue.Empty:
                pass
            else:
                if execute_task_dict(task_dict=task_dict, todo_lst=todo_lst):
                    future_queue.task_done()
                else:
                    future_queue.task_done()
                    future_queue.join()
                    break
            refresh(running_dict=running_dict, todo_lst=todo_lst, max_workers=exe._max_workers, exe=exe)
            sleep(sleep_interval)


class MultiProcessPoolExecutor(Executor):
    def __init__(self, *args, **kwargs):
        self._future_queue = queue.Queue()
        self._process = Thread(
            target=background_execution,
            kwargs={
                "exe_args": args, 
                "exe_kwargs": kwargs,
                "future_queue": self._future_queue,
            },
        )
        self._process.start()

    def submit(self, fn, *args, cores=1, **kwargs):
        f = Future()
        self._future_queue.put({"fn": fn, "args": args, "kwargs": kwargs, "future": f, "cores": cores})
        return f

    def shutdown(self, wait=True, *, cancel_futures=False):
        if cancel_futures:
            cancel_items_in_queue(que=self._future_queue)
        self._future_queue.put({"shutdown": True, "wait": wait})
        if wait:
            self._process.join()
            self._future_queue.join()
        self._process = None
        self._future_queue = None

Example:

from multi import MultiProcessPoolExecutor

def test_addition(a, b):
    return a + b


if __name__ == "__main__":
    with MultiProcessPoolExecutor() as exe:
        fn = exe.submit(test_addition, 1, 2)
        print(fn.result())

Example:

def add_function(parameter_1, parameter_2):
    return parameter_1 + parameter_2

with Executor(max_cores=1) as exe:
    future_1 = exe.submit(add_function, 1, parameter_2=2)
    future_2 = exe.submit(add_function, 1, parameter_2=future_1)
    print(future_2.result())

It works for a single core:

lmp = LammpsBase(
    cores=1,
    oversubscribe=False,
    enable_flux_backend=False,
    working_directory=".",
)

but calling:

lmp = LammpsBase(
    cores=2,
    oversubscribe=False,
    enable_flux_backend=False,
    working_directory=".",
)

results in one process with 100% CPU load.

Found by @pmrv

I would be interested in adding explicit verification that pympipool.Executor can handle un-pickle-able stuff. I have a test suite already here that (almost!) works out-of-the box when you plug in pympipool.Executor, so the simplest thing would be to shove those into the executor tests. However, maybe there is a more generic place to do this sort of thing like in the sockets where the cloudpickling is happening?

(The "almost" is just to do with how things get handled when an exception is raised, which is test for a spec I wrote over in contrib and doesn't need to be how we do it. I am also struggling that pycharm fails to shutdown the processes and finish the tests, so it's possible that this issue goes away entirely once PyCharm plays nicely with Executor.)

The multiprocessing Pool implements more functions than just the map() function:
https://docs.python.org/3/library/multiprocessing.html#module-multiprocessing.pool

• apply()
• apply_async()
• map()
• map_async()
• imap()
• imap_unordered()
• starmap()
• starmap_async()

This does not work:

from pympipool import Pool


def my_square(i):
    return i ** 2


def calc(i):
    import numpy as np
    return np.array(my_square(i=i))


with Pool(cores=2) as p:
    print(p.map(function=calc, lst=[1, 2, 3, 4]))

But this does:

from pympipool import Pool


def calc(i):
    import numpy as np

    def my_square(i):
        return i ** 2

    return np.array(my_square(i=i))


with Pool(cores=2) as p:
    print(p.map(function=calc, lst=[1, 2, 3, 4]))

When running larger calculation, like:

import numpy as np
import pandas
from tqdm import tqdm
from ase.build import bulk
from pympipool import Pool


def body_order(n=32, b=5):
    if b == 2:
        return [[i ,n-i] for i in range(n+1)]
    else:
        lst = []
        for i in range(n+1):
            for j in body_order(n=n-i, b=b-1):
                lst.append([i] + j)
        return lst


def get_elastic_constants(input_para):
    i, conc_lst, number_atoms, element_lst, potential, structure = input_para

    # import
    import numpy as np
    import pyiron_contrib
    from pyiron_atomistics import Project, ase_to_pyiron

    # create project
    project = Project('workflow')

    # Generate structure
    if sum(np.array(conc_lst)==0) != 4:
        mole_fraction = {
            el: conc/number_atoms
            for el, conc in zip(element_lst, conc_lst)
            if conc > 0
        }
        job = project.create_job(project.job_type.SQSJobWithoutOutput, "sqs_" + str(i))
        job._interactive_disable_log_file = True
        job.structure = ase_to_pyiron(structure)
        job.input['mole_fractions'] = mole_fraction
        job.input['iterations'] = 1e6
        job.server.cores = 1
        job.run()
        structure_next = job._lst_of_struct[-1]
    else:
        structure_next = ase_to_pyiron(structure).copy()
        structure_next.symbols[:] = np.array(element_lst)[np.array(conc_lst)!=0][0]

    # Minimize strucute
    lmp_mini1 = project.create_job(project.job_type.LammpsInteractiveWithoutOutput, "lmp_mini_" + str(i), delete_existing_job=True)
    lmp_mini1.structure = structure_next
    lmp_mini1.potential = potential
    lmp_mini1.calc_minimize(pressure=0.0)
    lmp_mini1.server.run_mode.interactive = True
    lmp_mini1.interactive_mpi_communicator = MPI.COMM_SELF
    lmp_mini1._interactive_disable_log_file = True  # disable lammps.log
    lmp_mini1.run()
    lmp_mini1.interactive_close()

    # Elastic constants
    lmp_elastic = project.create_job(project.job_type.LammpsInteractiveWithoutOutput, "lmp_elastic_" + str(i), delete_existing_job=True)
    lmp_elastic.structure = lmp_mini1.get_structure()
    lmp_elastic.potential = potential
    lmp_elastic.interactive_enforce_structure_reset = True
    lmp_elastic.interactive_mpi_communicator = MPI.COMM_SELF
    lmp_elastic.server.run_mode.interactive = True
    lmp_elastic._interactive_disable_log_file = True  # disable lammps.log
    elastic = lmp_elastic.create_job(project.job_type.ElasticMatrixJobWithoutFiles, "elastic_" + str(i), delete_existing_job=True)
    elastic._interactive_disable_log_file = True  # disable lammps.log
    elastic.run()

    elastic_constants_lst = [
        elastic._data["C"][0][0],
        elastic._data["C"][0][1],
        elastic._data["C"][0][2],
        elastic._data["C"][0][3],
        elastic._data["C"][0][4],
        elastic._data["C"][0][5],
        elastic._data["C"][1][1],
        elastic._data["C"][1][2],
        elastic._data["C"][1][3],
        elastic._data["C"][1][4],
        elastic._data["C"][1][5],
        elastic._data["C"][2][2],
        elastic._data["C"][2][3],
        elastic._data["C"][2][4],
        elastic._data["C"][2][5],
        elastic._data["C"][3][3],
        elastic._data["C"][3][4],
        elastic._data["C"][3][5],
        elastic._data["C"][4][4],
        elastic._data["C"][4][5],
        elastic._data["C"][5][5],
    ]

    if "Fe" in elastic.ref_job.structure.get_species_symbols():
        conc_Fe = sum(elastic.ref_job.structure.indices==elastic.ref_job.structure.get_species_symbols().tolist().index("Fe")) / len(elastic.ref_job.structure.indices)
    else:
        conc_Fe = 0.0
    if "Ni" in elastic.ref_job.structure.get_species_symbols():
        conc_Ni = sum(elastic.ref_job.structure.indices==elastic.ref_job.structure.get_species_symbols().tolist().index("Ni")) / len(elastic.ref_job.structure.indices)
    else:
        conc_Ni = 0.0
    if "Cr" in elastic.ref_job.structure.get_species_symbols():
        conc_Cr = sum(elastic.ref_job.structure.indices==elastic.ref_job.structure.get_species_symbols().tolist().index("Cr")) / len(elastic.ref_job.structure.indices)
    else:
        conc_Cr = 0.0
    if "Co" in elastic.ref_job.structure.get_species_symbols():
        conc_Co = sum(elastic.ref_job.structure.indices==elastic.ref_job.structure.get_species_symbols().tolist().index("Co")) / len(elastic.ref_job.structure.indices)
    else:
        conc_Co = 0.0
    if "Cu" in elastic.ref_job.structure.get_species_symbols():
        conc_Cu = sum(elastic.ref_job.structure.indices==elastic.ref_job.structure.get_species_symbols().tolist().index("Cu")) / len(elastic.ref_job.structure.indices)
    else:
        conc_Cu = 0.0

    conc_lst = [conc_Fe, conc_Ni, conc_Cr, conc_Co, conc_Cu]
    return None
    # return elastic_constants_lst + conc_lst


if __name__ == "__main__": 
    structure = bulk("Al", cubic=True).repeat([2,2,2])
    element_lst = ["Fe", "Ni", "Cr", "Co", "Cu"]
    number_atoms = len(structure)
    number_species = len(element_lst)
    potential = '2021--Deluigi-O-R--Fe-Ni-Cr-Co-Cu--LAMMPS--ipr1'

    lst_32 = body_order(n=number_atoms, b=number_species)[:120]
    input_para_lst = [
        [i, conc_lst, number_atoms, element_lst, potential, structure]
        for i, conc_lst in enumerate(lst_32)
    ]

    with Pool(cores=12) as p:
        print("Pool started")
        results = p.map(function=get_elastic_constants, lst=input_para_lst)
        print("Pool stoped")

Then I get the error message:

Traceback (most recent call last):
  File "/home/janssen/pyiron/projects/2023-02-14-elastic-mpi/elastic_mpi.py", line 137, in <module>
    results = p.map(function=get_elastic_constants, lst=input_para_lst)
  File "/home/janssen/miniforge3/lib/python3.9/site-packages/pympipool/__init__.py", line 46, in map
    output = self._receive()
  File "/home/janssen/miniforge3/lib/python3.9/site-packages/pympipool/__init__.py", line 59, in _receive
    output = dill.load(self._process.stdout)
  File "/home/janssen/miniforge3/lib/python3.9/site-packages/dill/_dill.py", line 272, in load
    return Unpickler(file, ignore=ignore, **kwds).load()
  File "/home/janssen/miniforge3/lib/python3.9/site-packages/dill/_dill.py", line 419, in load
    obj = StockUnpickler.load(self)
_pickle.UnpicklingError: pickle data was truncated

As the Executors in pympipool typically use zmq to connect the executor to the process which is executing the functions it should in principle be possible to serialise the executor - store the information of the executor - and then deserialise the same object again. Still it is not clear what happens to the messages already submitted and in addition it is also not clear if the server process is started by the executor or by the process executing the python functions.

While the flux adaptor raises an import error, the SLURM initialisation fails silently.

Current solutions like dask require the user to either start each worker manually:

# If we have four GPUs on one machine
CUDA_VISIBLE_DEVICES=0 dask-worker ...
CUDA_VISIBLE_DEVICES=1 dask-worker ...
CUDA_VISIBLE_DEVICES=2 dask-worker ...
CUDA_VISIBLE_DEVICES=3 dask-worker ...

https://docs.dask.org/en/stable/gpu.html#specifying-gpus-per-machine or use a third party package like https://github.com/rapidsai/dask-cuda . With the flux backend and the recent changes #125 pympipool allows creating one Executor per GPU and then manually distribute the tasks one per GPU Executor. But managing a number of executors is still more complicated than it has to be.

When queuing some functions and then (attempting) to wait on them with the shutdown function, the executor doesn't actually wait until all futures are ready. MWE is below. Doing it inside a context manager made no difference. When adding an additional time.sleep before shutting down the done callbacks do fire, so it sounds as if it's just the executor is not checking whether there are still outstanding futures.

import time
from pympipool import PoolExecutor

def calc(n):
    time.sleep(5)
    return n**2

exe = PoolExecutor(max_workers=4)
f1 = exe.submit(calc, 42)
f2 = exe.submit(calc, 84)
f1.add_done_callback(print)
f2.add_done_callback(print)
exe.shutdown(wait=True)

Currently all the unit tests use the paradigm cd tests; python -m unittest discover ., however a perfectly reasonable alternative is python -m unittest discover tests. However, the latter fails with error messages like

Traceback (most recent call last):
  File "/home/runner/work/pympipool/pympipool/pympipool/backend/mpiexec.py", line 90, in <module>
    main()
  File "/home/runner/work/pympipool/pympipool/pympipool/backend/mpiexec.py", line 46, in main
    input_dict = interface_receive(socket=socket)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/share/miniconda3/envs/test/lib/python3.11/site-packages/pympipool/shared/communication.py", line 147, in interface_receive
    return cloudpickle.loads(socket.recv())
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
ModuleNotFoundError: No module named 'test_meta'
Error: The action has timed out.

As can be seen in #238 Unittests_by_path.

To the best of my understanding, this is a result of these lines in the backend:

Since in the latter case "." is the parent path from which discover is invoked, and not the path where the actual test files reside.

This can be fixed by adding the test directory(ies) to the path (also demonstrated in #238 Unittests_by_path_with_env and Unittests_by_path_with_pythonpath which both run fine), but it still seems to me like something that should be fixed at the source rather than patched over in github workflows. In particular, I worry that if unittest discover has this sort of problem, other ways users invoke the executors might be similarly bugged.

It's not clear to me whether or not there is a good solution to this problem.

Aside: the comment in the snippet above says this is necessary for flux, but if this is really the relevant line for this issue that would imply it is important for all the executors, or?

Both ProcessPoolExecutor and ThreadPoolExecutor take the init argument max_workers -- pympipool did too until recently, but now this is max_cores.

Are these actually fundamentally different? From my understanding these really are the same parameter, and while I feel that pympipool does indeed use the better name here, I would like to go back to max_workers for the sake of matching the standard library interface. Is that possible? The change seems to have been a tangential bit in a larger commit focused on something else.

It seems sometimes the executor is not cleanly shutdown, resulting in the tests to hang even though all tests completed successfully.

Hi - is there any way to get information such as the number of cores, number of threads, and number of gpus from the executor class? I'm imagining this would be useful for automated meta-data extraction/storage when you might have a few executors going at once.

The documentation needs to be update and it would be great to add practical examples for the different use cases.

Currently, theresult() call right after thesubmit() gives the impression that both lines should follow each other.

Python implements:

threading.excepthook(args, /)
Handle uncaught exception raised by [Thread.run()](https://docs.python.org/3/library/threading.html#threading.Thread.run).

The args argument has the following attributes:

exc_type: Exception type.
exc_value: Exception value, can be None.
exc_traceback: Exception traceback, can be None.
thread: Thread which raised the exception, can be None.
If exc_type is [SystemExit](https://docs.python.org/3/library/exceptions.html#SystemExit), the exception is silently ignored. Otherwise, the exception is printed out on [sys.stderr](https://docs.python.org/3/library/sys.html#sys.stderr).

If this function raises an exception, [sys.excepthook()](https://docs.python.org/3/library/sys.html#sys.excepthook) is called to handle it.

So most likely this is better than the current RaisingThread implementation.

Compare different kinds of executors to unify the interface:

Open Questions:

• Is there any SSH support for existing Executors? How to handle transfer of files on remote locations?
• Can shell tasks and python tasks be combined in a single submission? With flux this should not be an issue and similarly with slurm.

Currently the tests fail on the main branch.

The website is already updated but the Github readme is still out of date

Locally they seem to work fine.

Currently it is unclear which test belongs to which module.

At the moment the flux framework or SLURM workload manager are required to scale beyond a single node. We could in principle offer this for serial functions just based on mpi4py using the mpi4py.futures.MPIPoolExecutor. Still given the simple installation of flux-core via conda I do not think this is required any more.

Hi @jan-janssen - is there a way to control the distribution of cores/threads/gpus at executor.submit() time (e.g. per-"job")? I was looking over the more recent versions and it seems like the interface has gone more the way of initializing the executors with this information - but I very likely could have missed something.

Perhaps this is already known, but it is a bit buried in the docs. So thought it would be worthwhile to surface in case it may be of interest.

Within mpi4py, there is a concurrent.futures implementation of Executors. These include a MPI-2 based implementation and an MPI-1 based implementation. So can be deployed on different systems depending on what is available.

I am running into an issue where I need to specify mpi version in the srun command on one of our machines. Would it be possible to add something like an "srun_options" kwarg in the PySlurmExecutor interface (possibly others?) to allow this kind of flexibility?

I'm still playing around integrating pympipool with pyiron_workflow in this PR. I had been struggling with a couple of tests that hang local, and on the CI gave errors like ModuleNotFoundError: no module named 'unit' or no module named 'test_macro'.

I managed to whittle things down to this minimal(ish) working example:

Suppose I run a test file, e.g. test_pympipool.py, with the following contents:

from unittest import TestCase

from pympipool import PyMPIExecutor


def add_two(x):
    return x + 2


class TestPyMPIExecutor(TestCase):

    def test_local(self):
        # Works perfectly
        def add_one(x):
            return x + 1

        class Foo:
            def add(self, x):
                return add_one(x)

        foo = Foo()
        executor = PyMPIExecutor()
        fs = executor.submit(foo.add, 1)
        self.assertEqual(2, fs.result())

    def test_semi_local(self):
        # Hangs
        # CI logs make it look like cloudpickle is trying to import from this module,
        # but can't find it
        class Foo:
            def add(self, x):
                return add_two(x)

        foo = Foo()
        executor = PyMPIExecutor()
        fs = executor.submit(foo.add, 1)
        self.assertEqual(3, fs.result())

When the function and class are defined in exactly the same scope, everything works perfectly. Not shown but also working perfectly is if I imported the add_x function from some other accessible location. The only thing that seems to be breaking is when I make a local definition that relies on something else defined in a different scope but the same file.

In a very hand-wavey way this makes sense to me -- the executor sees that add_two is not defined in this scope and so decides its something that can be pickled by reference; the only problem is that the file it's defined in is not in my import path! In contrast, when add_one is defined in exactly the same scope as the object being pickled, it's clear to the code that this object also needs to be pickled by value.

My questions are:

• Should this count as a bug that can be fixed here in pympipool, or is what I'm asking for unreasonable? In the latter case, what is the recommended syntax for same-file-different-scope definitions like this, i.e. how to succinctly make sure everything is added to path, or can I flag particular stuff for by-value treatment?
• On the CI it's super helpful to get the hard crash and the import error message -- why is it only hanging on my local machine, and can we force a crash under these conditions there too?

How are the scripts executed? On the login node or on a compute node?

Currently, the executor shares a queue with the broker and the broker distributes the individual tasks to the individual workers. Still is would also be possible to have all workers directly access the queue of the user facing executor, the only thing that is unclear in that case is how to make sure the initialisation function is communicated to all workers.

In the process of getting pympipool to play well with pyiron_workflow, I have been playing around a bunch with callbacks and using executors sans the with context. (See #187 and #210 for discussion.)

After repeated narrowing of my MWE, I ended up with this:

from pympipool.mpi.executor import PyMPIExecutor as Executor

mutable = []

def slow_callable():
    from time import sleep
    sleep(1)
    return True

def callback(future):
    mutable.append("Called back")
    
def submit():
    future = Executor().submit(slow_callable)
    future.add_done_callback(callback)
    return future
    
print(mutable)

# 1) Executor is locally scoped variable
# executor = Executor()
# future = executor.submit(slow_callable)

# 2) Executor is never assigned to a variable
# future = Executor().submit(slow_callable)

# 3) Executor is in a different scope
# future = submit()

future.add_done_callback(callback)

print(mutable)
future.result()
print(mutable)

The desired output is

[]
[]
['Called back']

Where the waiting happens when we invoke future.result(). However, this is only the case for option (1) where the executor is assigned its own variable in same scope as the future object. Options (2) and (3) both give

[]
['Called back']
['Called back']

Where the waiting happens at the submit call.

This led me in the direction of garbage collection, which led me to:

(EDIT: Trying to get the preview blob to format nicely)

Changing the wait=True to wait=False resolves this issue and keeps the tests on my machine passing, but I don't know if it has other side effects.

Discussion: mpi4py/mpi4py#318

Branch: https://github.com/mpi4py/mpi4py/tree/feature/util-pool