Running parallel batch jobs is not performant due to all Python subprocesses running on the same master node

[Andrew-S-Rosen]

We have had pretty good success with the batch job feature, but we have not been able to carry out parallel batch jobs successfully.

We are running VASP with Custodian, where each VASP job uses one full CPU node and we are requesting multiple CPU nodes per Slurm allocation. Unfortunately, some VASP jobs are killed with SIGTERM calls even though there is no indication anything is going wrong with them, perhaps because Custodian is sending the SIGTERM to the wrong node? I should note that the jobs are not all SIGTERM-ing at the same time; a rare one or two might complete, and others SIGTERM later on, so it is not that Custodian is sending the SIGTERM to all the processes. We have the same issue across multiple machines, so it is not related to the VASP build or node architecture.

The following is a typical worker configuration that @blaked8619 is using. We tried adding the --exclusive flag, but it did not help. Note that even though we are using quacc here, it is calling Custodian in exactly the same way as Atomate2, so I do not think that should be related to the issue. Of course, we could try to generate an Atomate2 example if absolutely needed.

stellar_vasp:
    type: remote
    host: <REDACTED>
    user: <REDACTED>
    scheduler_type: slurm
    work_dir: /scratch/gpfs/ROSENGROUP/bd8619/path/to/my/stuff
    max_jobs: 5
    pre_run: |
      source ~/.bashrc
      module load anaconda3/2024.10
      conda activate jobflow
      module load vasp/6.5.1
      export QUACC_VASP_PARALLEL_CMD="srun -N 1 --ntasks-per-node 96 --exclusive"
      export QUACC_WORKFLOW_ENGINE=jobflow
      export QUACC_CREATE_UNIQUE_DIR=False
      export QUACC_GZIP_FILES=True
    timeout_execute: 60
    resources:
      nodes: 5
      ntasks_per_node: 96
      cpus_per_task: 1
      mem: 700G
      time: 48:00:00
      account: cbe
    batch:
      jobs_handle_dir: /scratch/gpfs/ROSENGROUP/bd8619/path/to/my/stuff/jfr_handle_dir
      work_dir: /scratch/gpfs/ROSENGROUP/bd8619/path/to/my/stuff/jf_batch_jobs
      parallel_jobs: 5
      max_time: 144000

I am also attaching the failed VASP run (failed.zip), minus the POTCAR which is proprietary. The job is going along fine and then the stderr hits us with the following SIGTERM. It is not related to this particular system.

forrtl: error (78): process killed (SIGTERM)
Image              PC                Routine            Line        Source             
libpthread-2.28.s  0000152CCC12A990  Unknown               Unknown  Unknown
libmkl_avx512.so.  00001529FA5B234B  mkl_blas_avx512_d     Unknown  Unknown

Please let us know if you have any suggestions or what other information would be helpful for debugging. This is, of course, a challenging one to debug given how many components are involved.

[gpetretto]

Hi @Andrew-S-Rosen,

this looks indeed a difficult problem to debug, also seeing that the error happened several hours into the job.
Custodian definitely looks like a potential culprit, since in the terminate method it does not check the PID but just kills all the processes with vasp in the name and with a vasprun.xml in the folder:
https://github.com/materialsproject/custodian/blob/0c1b6a562dd84bfeae66181bf25db72ca9b1c39e/src/custodian/vasp/jobs.py#L718
However this indeed leaves some questions open:

• If this is the case, why don't all the jobs get killed simultaneously?
• I don't expect that psutil will find processes from other nodes, so which process can really be killed?
• Can the kills be traced back to some action of the Custodian handlers?

One way to determine if Custodian is the responsible would be to run the same jobs without custodian, or even just without any handler that is also a monitor. Do you think you can run this test?

Looking for alternative sources of errors, at the end of the std_err.txt that you uploaded there are these lines:

srun: Terminating StepId=2127309.0
slurmstepd: error: *** STEP 2127309.0 ON stellar-i06n11 CANCELLED AT 2025-07-28T07:42:49 ***

In my experience the error messages are often dependent on the slurm version and configuration, but I often find similar logs when slurm is part in actively canceling the job. Altough in those cases it usually also reports the cause. Still it may be worth investigating. Maybe scontrol can give some additional information about the reason for the failure.
To check this I made a simple test to see if killing a job executed with srun gave the same kind of error message. I have compiled this fortran code with sleep:

program sleep_example
    implicit none

    print *, 'init sleep'
    call sleep(3)
    print *, 'finished'

end program sleep_example

and submitted with:

#!/bin/bash

#SBATCH --partition=batch
#SBATCH --job-name=test
#SBATCH --nodes=1
#SBATCH --ntasks-per-node=2
#SBATCH --mem=1000
#SBATCH --time=00:03:00

srun --nodes 1 -n 1 --exclusive ./a.out > stdout 2> stderr &

sleep 1
killall a.out
sleep 2

This could mimic the effect of an external process killing the code running with srun. It may be interesting to see if this gives a message similar to slurmstepd: error: *** STEP 2127309.0 ON stellar-i06n11 CANCELLED AT 2025-07-28T07:42:49 *** in stderr. In my case I did not have it. Can you maybe test this as well?

[Andrew-S-Rosen]

Thank you so much for sharing your thoughts! We will dig into this some more and report back since this is a feature we are very interested in.

@blaked8619 tested things with just VASP and no Python wrapper and it worked okay, but these are some other good tests we should try out that may help us isolate what is going on.

@blaked8619: if you share with me via Slack a representative Python script to reproduce the problem in full, I can start debugging this. I know you have a bunch of runs going, so I'm happy to give it a go in the meantime as well.

[Andrew-S-Rosen]

From the day's debugging, I can report the following thus far.

1. It is very likely related to Custodian. Running the exact same code but without using Custodian to launch and manage the srun calls results in completion of the parallel batch jobs.
2. Running the toy fortran code you shared does not have the CANCELLED text. It shows

srun: error: tiger-i04g13: task 0: Terminated
srun: Terminating StepId=1318345.0
srun: Force Terminated StepId=1318345.0

I will continue to debug things.

[Andrew-S-Rosen]

Sadly, no luck here and I'll have to postpone further troubleshooting until the middle of next week. Because the issue is seemingly non-deterministic, trying to figure out the precise cause has proven particularly challenging.

[Andrew-S-Rosen]

I may have gotten closer to the answer.

In the queue.err file for one of the failed jobs, it shows the following:

WARNING:custodian.vasp.jobs:Killing VASP processes in work_dir='/scratch/gpfs/ROSENGROUP/asrosen/jobflow/vasp/80/6b/fd/806bfdfc-b19a-4520-b5b2-b783246fdb93_1/tmp-quacc-2025-08-14-05-57-16-998502-51543' failed with subprocess.Popen.terminate(). Resorting to 'killall'.
vasp_gam: no process found

and then later

WARNING:custodian.vasp.jobs:Killing VASP processes in work_dir='/scratch/gpfs/ROSENGROUP/asrosen/jobflow/vasp/80/6b/fd/806bfdfc-b19a-4520-b5b2-b783246fdb93_1/tmp-quacc-2025-08-14-05-57-16-998502-51543' failed with subprocess.Popen.terminate(). Resorting to 'killall'.
vasp_std: no process found
vasp_gam: no process found

This seems potentially concerning. There is no SIGTERM in this job.

The other failed job fails with SIGTERM calls in its VASP stderr file and no such warning messages. I don't know how Custodian would be able to kill a job on another node though, so it still doesn't add up.

From the Custodian source:

There is also a safety that kills all VASP processes if none of the processes can be killed (This is bad if more than one VASP runs are simultaneously executed on the same node). However, this should never happen.

[Andrew-S-Rosen]

Yup, I have further timestamps to support that this is what is likely what is happening.

Job 1 is a VASP calculation that Custodian restarts at 10:00:20 PM. The queue.err shows the following:

INFO:custodian.custodian:Starting job no. 1 (VaspJob) attempt no. 1. Total errors and errors in job thus far = 0, 0.
INFO:custodian.vasp.jobs:Running srun --nodes 1 --ntasks-per-node 112 --exclusive vasp_std
ERROR:VaspErrorHandler:WARNING in EDDRMM: call to ZHEGV failed
INFO:custodian.custodian:Terminating job
INFO:custodian.vasp.jobs:Killing VASP processes in work_dir='/scratch/gpfs/ROSENGROUP/asrosen/jobflow/vasp/d4/c7/ab/d4c7ab7b-fd84-4253-a869-f88d823b6c8b_1/tmp-quacc-2025-08-15-01-50-19-361195-29810'.
WARNING:custodian.vasp.jobs:Killing VASP processes in work_dir='/scratch/gpfs/ROSENGROUP/asrosen/jobflow/vasp/d4/c7/ab/d4c7ab7b-fd84-4253-a869-f88d823b6c8b_1/tmp-quacc-2025-08-15-01-50-19-361195-29810' failed with subprocess.Popen.terminate(). Resorting to 'killall'.
vasp_gam: no process found
INFO:root:Backing up run to /scratch/gpfs/ROSENGROUP/asrosen/jobflow/vasp/d4/c7/ab/d4c7ab7b-fd84-4253-a869-f88d823b6c8b_1/tmp-quacc-2025-08-15-01-50-19-361195-29810/error.1.tar.gz.

Job 1's VASP calculation then restarts as usual. The std_err.txt is empty. The job proceeds as usual.

Exactly 10 seconds later at 10:00:30 PM, Job 2's VASP calculation fails unexpectedly and the entire job is failed. The std_err.txt shows

forrtl: error (78): process killed (SIGTERM)
Image              PC                Routine            Line        Source             
libc.so.6          000014EA34C3EBF0  Unknown               Unknown  Unknown
libmkl_avx512.so.  000014E6EE8F63D7  mkl_blas_avx512_d     Unknown  Unknown

and the queue.err shows the expected VasprunXMLValidator error because the job was spontaneously aborted.

This suggests that when Custodian tried to kill Job 1, it also ended up killing Job 2. This is alluded to in the Custodian warning. Any thoughts, @gpetretto?

[gpetretto]

Hi @Andrew-S-Rosen, thanks for the detailed investigation. Given your last findings (especially the fact that the first job does not find the VASP process to kill) I think the explanation may be clear: the python code is executed only on the master node. The VASP execution is sent to the other node when Custodian runs srun, but custodian itself still lives in the main node. Does this match your finding? Do you have a way of checkeking if the Job that is being killed is the one running on the main node?

If this is the reason it may be challenging to address it. There are a few points that clash:

• The destination node is chosen by slurm when executin srun
• srun is executed by custodian, so custodian is already assigned to a node when VASP starts.

I can imagine that we have two different options to address this:

• when killing VASP, custodian should be able to target the node where VASP is being executed
• custodian is itself started as another python process directly on the node where VASP will be executed.
  In both cases this seems technically challenging and I don't have a clear idea of how this could be done easily.

An obvious alternative would be to run custodian without the monitors and just rely on the handlers, but I don't know if this will be acceptable in your case.

[Andrew-S-Rosen]

Thanks for the clarity @gpetretto! That description matches my understanding of the situation. I don't know how to further test where Custodian is running easily, but I think you're right. Unfortunately, we need to rely on Custodian but of course disabling the handlers is one route.

Since this was done many times with Fireworks at NERSC, I share the job packing external utility we used below just in case it provides inspiration. I don't know if it's relevant.

#!/usr/bin/env python
import os
import subprocess
import json
import click
import time
import random
import datetime

try:
    # Python 3
    from itertools import zip_longest
except ImportError:
    # Python 2
    from itertools import izip_longest as zip_longest

def dump_slurm_env():
    """
    Dump the SLURM environments
    """
    slurm_vars = {k: v for k, v in os.environ.items() if "SLURM" in k}
    with open(".slurm_env", "w") as f:
        json.dump(slurm_vars, f)


def load_slurm_env(wd):
    """
    Load the SLURM environments
    """
    with open(wd+"/.slurm_env", "r") as f:
        slurm_vars = json.load(f)
    return slurm_vars


def grouper(iterable, n, fillvalue=None):
    """
    Collect data into fixed-length chunks or blocks
    grouper('ABCDEFG', 3, 'x') --> ABC DEF Gxx
    """
    args = [iter(iterable)] * n
    return zip_longest(fillvalue=fillvalue, *args)


def get_nodes():
    """
    Get the node list from SLURM
    """
    nodes = (
        subprocess.Popen("scontrol show hostname".split(), stdout=subprocess.PIPE)
        .stdout.read()
        .decode("utf-8")
    )
    nodes = set(nodes.split())
    print("Node list from scontrol: {}".format(nodes))
    return nodes


def create_datestamp_dir(root_dir, prefix="block_", max_tries=10):
    """
    Internal method to create a new block or launcher directory.
    The dir name is based on the time and the FW_BLOCK_FORMAT
    Args:
        root_dir: directory to create the new dir in
        l_logger: the logger to use
        prefix: the prefix for the new dir, default="block_"
    """

    def get_path():
        time_now = datetime.datetime.utcnow().strftime("%Y-%m-%d-%H-%M-%S-%f")
        block_path = prefix + time_now
        return os.path.join(root_dir, block_path)

    for i in range(max_tries):
        full_path = get_path()
        if os.path.exists(full_path):
            time.sleep(random.random() / 3 + 0.1)
        else:
            try:
                os.mkdir(full_path)
                return full_path
            except OSError as e:
                pass
    return None


@click.group()
def main():
    """
    Master script to control running multi-sub jobs on
    the Cori Super Computer at Cluster
    Use this to manage packaging thousands of jobs into large
    blocks
    """
    pass


@main.command("master")
@click.option(
    "-s",
    "--seperate_head_node",
    default=False,
    is_flag=True,
    help="Run the head process on its own node",
)
@click.option(
    "-n", "--nodes_per_job", default=1, help="Number of nodes for each sub job"
)
@click.option("--delay", default=2, type=int, help="Delay between sequential starts")
@click.option("--make_block", default=True, is_flag=True)
@click.option("--script", default="singleshot")
def master(seperate_head_node, nodes_per_job, delay, make_block, script):
    """
    Master process for a multi-job
    """
    hostname = "{}".format(os.uname()[1])
    cwd = os.getcwd()
    cmd_file = os.path.abspath(__file__)

    print(
        "Running on master node: {} in {} with a delay of {}".format(hostname,cwd,delay)
    )

    if make_block:
        cwd = create_datestamp_dir(cwd)

    os.chdir(cwd)

    nodes = get_nodes()

    dump_slurm_env()

    if seperate_head_node:
        nodes = nodes - {hostname}

    num_srun_launches = 1

    procs = []
    for i, sub_nodes in enumerate(grouper(nodes, nodes_per_job, None)):
        if None in set(sub_nodes):
            print("Ignoring {} due to incomplete set".format(sub_nodes))
        else:
            sub_head_node = sub_nodes[0]
            sub_nodes_str = ",".join(sub_nodes)
            worker_cmd = "{} worker {} --script {} --cwd {} --delay {}".format(cmd_file,sub_nodes_str,script,cwd,i)
            ssh_cmd = 'ssh {} -q "python {}"\n'.format(sub_head_node,worker_cmd)

        procs.append(subprocess.Popen(ssh_cmd, stdout=subprocess.PIPE, shell=True))

    for p in procs:
        p.wait()


@main.command("worker")
@click.argument("nodes", required=True, type=str)
@click.option(
    "--env_variable",
    default="MULTI_NODES",
    help="Environment varibale to store the node list in",
)
@click.option("--script", default="singleshot")
@click.option("--cwd", default=None)
@click.option("--delay", default=2, type=int, help="Delay between sequential starts")
def worker(nodes, env_variable, script, cwd, delay):
    """
    Worker process for a multi-job
    You don't directly run this.
    """

    slurm_vars = load_slurm_env(cwd)

    # Change directory if given
    if cwd:
        os.chdir(cwd)

    # Always delay by a small bit of time
    time.sleep(delay)

    #Ensure even single nodes are formatted as a list
    nodes = nodes.split()

    hostname = "{}".format(os.uname()[1])
    cwd = os.getcwd()
    cmd_file = os.path.abspath(__file__)
    cmd_dir = os.path.dirname(cmd_file)

    print("Running worker on node: {} in {}".format(hostname,cwd))

    temp_env = os.environ.copy()
    temp_env[env_variable] = " ".join(nodes)
    temp_env["PATH"] += ":{}".format(cmd_dir)
    temp_env.update(slurm_vars)


    p = subprocess.Popen(os.path.join(cmd_dir, script), env=temp_env)

    p.wait()


if __name__ == "__main__":
    main()

[gpetretto]

Thanks for sharing this! Indeed it can be useful to improve the handling of parallel jobs in jobflow-remote.
Just one question: can you clarify how is the MULTI_NODES environmental variable used inside the Job?

I think that that something like this could be implemented but the implementation seems somewhat specific to the use case (i.e. slurm based, not possible to allocate more than one job per node, available passwordless ssh connection between the nodes). These are probably reasonable options, but I would prefer to allow a broader range of options.
Since providing a one-fits-all solution may prove difficult it may be interesting to just implement some basic functionality that could be then adapted by the users through a plug-in system.

[Andrew-S-Rosen]

Thanks for sharing this! Indeed it can be useful to improve the handling of parallel jobs in jobflow-remote. Just one question: can you clarify how is the MULTI_NODES environmental variable used inside the Job?

It was used in the VASP command it seems: srun -N 1 --ntasks-per-node 32 -c 8 -w ${MULTI_NODES} --cpu_bind=cores vasp_std.

I am attaching the full set of scripts and YAMLs that enabled parallel batch jobs ("job packing") with FireWorks:

production.zip

I think that that something like this could be implemented but the implementation seems somewhat specific to the use case (i.e. slurm based, not possible to allocate more than one job per node, available passwordless ssh connection between the nodes). These are probably reasonable options, but I would prefer to allow a broader range of options. Since providing a one-fits-all solution may prove difficult it may be interesting to just implement some basic functionality that could be then adapted by the users through a plug-in system.

Yes, this is likely true. At one point, PSI/J was going to be designed to allow for a general-purpose, modular approach to this kind of job submission. I don't think that was ever realized.

[Andrew-S-Rosen]

I have further evidence that all Custodian processes are running on the master node and also have some additional information.

I submitted a 3-node parallel batch job and submitted 3 one-node jobs. The Slurm queue indicates that there are three nodes reserved:

 JOBID PARTITION     NAME     USER ST       TIME  NODES NODELIST(REASON)
           1417760       cpu batch_c3  asrosen  R       3:02      3 tiger-i02c12n4,tiger-i02c13n[1,3]

However, opening the queue.err within each of the three Jobflow jobs shows:

INFO:custodian.custodian:Hostname: tiger-i02c12n4, Cluster: unknown

In other words, all three Custodian processes are operating on tiger-i02c12n4.

If the VASP process that is on the master node (tiger-i02c12n4) fails, Custodian works as expected. However, if the VASP processes that are on the two other nodes fails, Custodian isn't able to kill the desired process and instead falls back to killall, which kills everything.

WARNING:custodian.vasp.jobs:Killing VASP processes in work_dir='/scratch/gpfs/ROSENGROUP/asrosen/jobflow/vasp/d4/c7/ab/d4c7ab7b-fd84-4253-a869-f88d823b6c8b_1/tmp-quacc-2025-08-15-01-50-19-361195-29810' failed with subprocess.Popen.terminate(). Resorting to 'killall'.

This is the reason why it has been hard to debug: because sometimes the failing job is on the master node and sometimes it is on the other nodes, the latter of which is the problem.