Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
""" Launch scattered jobs: one job = one pbs script. """
from copy import deepcopy
import os
import re
import subprocess
from os.path import exists, basename
from os import remove
import pylada
from .. import get_shell
from ...misc import local_path, testValidProgram
from ... import pbs_string, default_pbs, qsub_exe, default_comm
from . import get_walltime, get_mppalloc, get_queues, scattered_script
from .. import logger
if not hasattr(pylada, 'ipython_qstat'):
logger.warning("Missing ipython_qstat function: cannot check for jobs already in queue")
qstat = lambda x: []
else:
qstat = lambda x: self.qstat(x)
logger.info("launch/scattered: event: %s" % event)
shell = get_shell(self)
pbsargs = deepcopy(dict(default_comm))
pbsargs.update(default_pbs)
pbsargs['ppn'] = event.ppn
mppalloc = get_mppalloc(shell, event)
if mppalloc is None:
return
# Set pbsargs['walltime'] to a string like '03:59:59'