##################################################################
## (c) Copyright 2015- by Jaron T. Krogel ##
##################################################################
#====================================================================#
# machines.py #
# Representations of local machine environments including #
# workstations and supercomputers and the jobs that will be #
# executed on them. #
# #
# Content summary: #
# Job #
# Class to represent a generic simulation job. #
# #
# Machine #
# Represents a generic machine. #
# Base class for specific machine types. #
# #
# Workstation #
# Represents a workstation with a fixed number of cores. #
# #
# InteractiveCluster #
# Represents a supercomputer in interactive mode. #
# Similar to a workstation with many cores. #
# #
# Supercomputer #
# Represents a generic supercomputer with a batch queue. #
# Base class for specific supercomputers. #
# See Jaguar, Kraken, Golub, OIC5, Hopper, Edison, BlueWatersXE,#
# BlueWatersXK, Titan, EOS, Vesta, Cetus, Mira, Lonestar, #
# Matisse, Komodo, and Amos #
# #
# cpu_count #
# Function to return the number of cores on the local machine. #
# #
# Options #
# Class representing command line options for a simulation job, #
# including arguments to the simulation executable, #
# the run launcher (aprun/mpirun, etc), and the job submitter #
# (e.g. qsub). #
# #
#====================================================================#
import os
from pathlib import Path
import platform
from socket import gethostname
import subprocess
from subprocess import Popen, CalledProcessError
import numpy as np
from .developer import DevBase, obj, warn
from .nexus_base import NexusCore, nexus_core
from .execute import execute
from .utilities import path_string
import importlib.util
import importlib.machinery
[docs]
def our_load_source(modname, filename):
"""" Replacement for the deprecated imp.load_source function"""
loader = importlib.machinery.SourceFileLoader(modname, filename)
spec = importlib.util.spec_from_file_location(modname, filename, loader=loader)
module = importlib.util.module_from_spec(spec)
# The module is always executed and not cached in sys.modules.
# Uncomment the following line to cache the module.
# sys.modules[module.__name__] = module
loader.exec_module(module)
return module
[docs]
def get_cpu_cores() -> int:
"""Get the number of physical CPU cores, defaulting to logical if not available.
This function makes calls to ``lscpu`` on Linux and ``sysctl`` on MacOS.
If these commands fail, it will default to :py:func:`os.cpu_count()`, which returns
the number of logical processors in the system. This may be up to twice as
much as the number of physical cores.
"""
n_cores = os.cpu_count() # Set ahead of time so it is never unbound
try:
sysname = platform.system()
match sysname:
case "Linux":
query = subprocess.run(
["lscpu", "-p=Core,Socket"],
capture_output=True,
check=True,
)
output = query.stdout.decode().strip().splitlines()
# set will automatically remove duplicate entries.
# Anything that remains is the list of physical cores.
output = set([i for i in output if not i.startswith("#")])
n_cores = len(output)
case "Darwin":
query = subprocess.run(
["sysctl", "-n", "hw.physicalcpu"],
capture_output=True,
check=True,
)
n_cores = int(query.stdout.decode().strip())
case _:
warn(
f"Operating system '{sysname}' not recognized by Nexus.\n"
"Could not get number of physical CPU cores, defaulting to logical..."
)
except CalledProcessError as err:
warn(
"Could not get number of physical CPU cores, defaulting to logical...\n"
f"Error is:\n{err}"
)
return n_cores
[docs]
class Options(DevBase):
def __init__(self,**kwargs):
self.add(**kwargs)
#end def __init__
[docs]
def add(self,**kwargs):
self.transfer_from(kwargs)
#end def add
[docs]
def read(self,options):
if isinstance(options,(dict,obj)):
self.add(**options)
elif isinstance(options,list):
for o in options:
if not isinstance(o,str):
self.error('each option must be a string')
#end if
self[str(len(self))] = o
#end for
elif isinstance(options,str):
self[str(len(self))] = options
else:
self.error('invalid type provided to Options')
#end if
#end def read
[docs]
def write(self):
s = ''
for k in sorted(self.keys()):
s += ' '+str(self[k])
#end for
return s
#end def write
#end class Options
job_defaults_assign = obj(
name = 'jobname',
type = None,
directory = None,
subdir = None,
app_name = None, # name of/path to application
app_command = None, # command used to launch application
app_props = None,
full_command = None, # custom command including e.g. mpirun
outfile = None,
errfile = None,
env = None,
user_env = True, # import user environment
presub = '', # shell text executed just prior to submission
postsub = '', # shell text executed just after submission
queue = None,
bundled_jobs = None,
relative = False,
cores = None, # number of cores for the job
nodes = None, # number of nodes for the job
threads = 1, # number of openmp threads for the job
hyperthreads = None,
ppn = None,
gpus = None, # number of gpus per node
serial = False, # run job serially, no mpi
local = False, # run job locally, no queue submission
days = 0,
hours = 0,
minutes = 0,
seconds = 0,
subfile = None,
grains = None,
procs = None,
processes = None,
processes_per_proc = None,
processes_per_node = None,
account = None,
email = None,
constraint = None, # slurm specific, Cori
core_spec = None, # slurm specific, Cori
switches = None, # slurm specific, SuperMUC-NG
alloc_flags = None, # lsf specific, Summit
filesystems = None, # pbs specific
qos = None,
group_list = None,
default_cpus_per_task = False, # optionally bypass typical nexus processing for supermucng
ntasks_per_core = None,
cpus_per_task = None,
template = None,
reservation = None,
)
# these are not assigned directly
job_defaults_nonassign = obj(
fake = False,
app = None, # name of/path to application
machine = None,
options = None,
app_flags = None,
app_options = None,
run_options = None,
sub_options = None,
skip_machine = False,
)
job_defaults = obj(job_defaults_assign,job_defaults_nonassign)
[docs]
class Job(NexusCore):
machine = None #default machine if none is specified in settings
states = obj(
none = 0,
waiting = 1,
submitted = 2,
running = 3,
finished = 4
)
state_names = states.inverse()
job_count = 0
[docs]
@staticmethod
def restore_default_settings():
Job.machine = None
#end def restore_default_settings
[docs]
@staticmethod
def generate_jobid():
Job.job_count += 1
return Job.job_count
#end def generate_jobid
[docs]
@classmethod
def zero_time(cls):
time = obj(days=0,hours=0,minutes=0,seconds=0)
return time
#end def zero_time
def __init__(self,**kwargs):
# rewrap keyword arguments
kw = obj(**kwargs)
# Ensure no pathlib.Path objects are stored
path_arguments = (
"directory",
"subdir",
"outfile",
"errfile",
"subfile",
)
for arg in path_arguments:
if arg in kw:
kw[arg] = path_string(kw[arg])
# Theoretically a bash alias can have invalid filename characters,
# so we treat app_name with some more special logic.
if "app_name" in kw:
if isinstance(kw.app_name, Path):
kw.app_name = path_string(kw.app_name)
elif not isinstance(kw.app_name, str):
self.error("app_name must be a str or Path object!")
# save information used to initialize job object
self.init_info = kw.copy()
# set defaults
kw.set_optional(**job_defaults)
# extract keywords not assigned
app = kw.delete('app')
machine = kw.delete('machine')
options = kw.delete('options')
app_flags = kw.delete('app_flags')
app_options = kw.delete('app_options')
run_options = kw.delete('run_options')
sub_options = kw.delete('sub_options')
env = kw.delete('env')
fake = kw.delete('fake')
skip_machine = kw.delete('skip_machine')
# assign keywords
self.set(**kw)
# assign fake job
self.fake_job = fake
# check template
if self.template is not None and not isinstance(self.template,str):
self.error('template must be a string\nReceived type: {}'.format(self.template.__class__.__name__))
#end if
# initialize other internal variables
self.app_options = Options()
self.run_options = Options()
self.sub_options = Options()
self.env = None
self.internal_id = None
self.system_id = None
self.tot_cores = None
self.identifier = None
self.submitted = False
self.status = self.states.none
self.crashed = False
self.overtime = False
self.successful = False
self.finished = False
self.user_app_command = self.app_command is not None
if app is not None:
self.app_name = app
#end if
if app_options is not None:
self.app_options.read(app_options)
#end if
if run_options is not None:
self.run_options.read(run_options)
#end if
if sub_options is not None:
self.sub_options.read(sub_options)
#end if
if app_flags is not None:
self.app_options.read(app_flags)
#end if
if options is not None:
self.run_options.read(options)
#end if
if self.app_props is None:
self.app_props = []
#end if
if self.serial:
self.cores = 1
self.nodes = None
#end if
if skip_machine:
self.machine = machine
else:
if machine is not None:
self.machine = machine
#end if
if env is not None:
self.set_environment(**env)
#end if
#check that the machine exists and have it complete the job info
self.process()
machine = self.get_machine()
self.batch_mode = machine.in_batch_mode()
if self.bundled_jobs is not None and not machine.batch_capable:
self.error('running batched/bundled jobs on {0} is either not possible or not yet implemented, sorry.'.format(machine.name))
#end if
#end if
self.normalize_time()
#end def __init__
[docs]
def get_machine(self):
return Machine.get(self.machine)
#end def get_machine
[docs]
def process(self,machine=None):
if self.template is not None:
return
#end if
if machine is None:
machine = self.get_machine()
#end if
machine.process_job(self)
#end def process
# test needed
[docs]
def process_options(self,machine=None):
if self.template is not None:
return
#end if
if machine is None:
machine = self.get_machine()
#end if
machine.process_job_options(self)
#end def process_options
# test needed
[docs]
def initialize(self,sim):
self.set_id()
self.identifier = sim.identifier
machine = self.get_machine()
if machine.prefixed_output:
sim.outfile = sim.identifier + machine.outfile_extension
sim.errfile = sim.identifier + machine.errfile_extension
#end if
if self.directory is None:
self.directory = sim.locdir
self.abs_dir = os.path.abspath(sim.locdir)
elif self.abs_dir is None:
self.abs_dir = os.path.abspath(self.directory)
#end if
if self.subdir is None:
if machine.local_directory is not None:
self.subdir = os.path.join(machine.local_directory,nexus_core.runs,sim.path)
self.abs_subdir = self.subdir
else:
self.subdir = self.directory
self.abs_subdir = self.abs_dir
#end if
#end if
if self.app_name is None:
app_name = sim.app_name
else:
app_name = self.app_name
#end if
if app_name is not None and '/' not in app_name:
ads = machine.app_directories
ad = machine.app_directory
new_app_name = None
if ads is not None and self.app_name in ads:
new_app_name = os.path.join(ads[self.app_name],app_name)
elif ad is not None:
new_app_name = os.path.join(ad,app_name)
#end if
if new_app_name is not None:
self.app_name = new_app_name
#end if
#end if
sim.set_app_name(app_name)
self.set(
name = sim.identifier,
simid = sim.simid,
outfile = sim.outfile,
errfile = sim.errfile
)
if self.app_command is None:
self.app_command = sim.app_command()
#end if
if self.app_props is None:
self.app_props = list(sim.app_props)
#end if
# ensure job is processed properly by this initialization stage
self.process()
#end def initialize
# test needed
[docs]
def renew_app_command(self,sim):
if not self.user_app_command:
self.app_command = sim.app_command()
#end if
#end def renew_app_command
[docs]
def set_id(self):
self.internal_id = Job.generate_jobid()
#end def set_id
# remove?
[docs]
def set_processes(self):
if self.processes is None:
self.error('processes should have been set before now\ncontact the developers and have them fix this','Developer')
self.processes = int(np.ceil(float(self.cores)/self.threads))
#end if
#end def set_processes
[docs]
def set_environment(self,limited_env=False,clear_env=False,**env):
machine = self.get_machine()
if isinstance(machine,Supercomputer):
limited_env = True
#end if
if self.env is None:
self.env = os.environ.copy()
if limited_env:
self.env.clear()
#end if
#end if
if clear_env:
self.env.clear()
#end if
for n,v in env.items():
self.env[n]=str(v)
#end for
#end def set_environment
[docs]
def divert_out_err(self):
self.identifier += '_divert'
#end def divert_out_err
[docs]
def get_time(self):
time = obj(
days = self.days,
hours = self.hours,
minutes = self.minutes,
seconds = self.seconds
)
return time
#end def get_time
[docs]
def max_time(self,time):
t = time.seconds + 60*(time.minutes+60*(time.hours+24*time.days))
ts = self.seconds + 60*(self.minutes+60*(self.hours+24*self.days))
if ts>t:
time.days = self.days
time.hours = self.hours
time.minutes = self.minutes
time.seconds = self.seconds
#end if
return time
#end def max_time
[docs]
def serial_only(self):
return 'serial' in self.app_props and len(self.app_props)==1
#end if
# remove?
[docs]
def determine_end_status(self,status):
if not nexus_core.generate_only:
self.successful = False # not really implemented yet
#end if
#end def determine_end_status
# test needed
[docs]
def write(self,file=False):
machine = self.get_machine()
return machine.write_job(self,file=file)
#end def write
# test needed
[docs]
def submit(self):
machine = self.get_machine()
machine.add_job(self)
self.submitted = True
#end def submit
# test needed
[docs]
def reenter_queue(self):
machine = self.get_machine()
machine.requeue_job(self)
#end def reenter_queue
[docs]
def run_command(self,launcher=None,redirect=False,serial=False):
if self.template is not None:
return ''
#end if
machine = self.get_machine()
if launcher is None:
launcher = machine.app_launcher
#end if
c = ''
if self.bundled_jobs is None:
if self.full_command is not None:
c = self.full_command
else:
if self.app_command is None:
self.error('app_command has not been provided')
#end if
if launcher=='runjob':
separator = ' : '
else:
separator = ' '
#end if
if self.serial and self.processes==1:
c = ''
else:
c = launcher + self.run_options.write() + separator
#end if
c+=self.app_command+self.app_options.write()
if redirect:
c+=' >'+self.outfile+' 2>'+self.errfile
if not serial:
c+='&'
#end if
elif machine.redirect_output and self.outfile is not None:
c+=' >'+self.outfile+' 2>'+self.errfile
#end if
#end if
elif machine.special_bundling:
c+='\n'
c+=machine.specialized_bundle_commands(self,launcher,serial)
elif self.relative:
cdir = self.abs_subdir
c+='\n'
for job in self.bundled_jobs:
c+='\ncd '+os.path.relpath(job.abs_subdir,cdir)+'\n'
c+=job.run_command(launcher,redirect=True,serial=serial)+'\n'
cdir = job.abs_subdir
#end for
c+='\nwait\n'
else:
c+='\n'
for job in self.bundled_jobs:
c+='\ncd '+job.abs_subdir+'\n'
c+=job.run_command(launcher,redirect=True,serial=serial)+'\n'
#end for
c+='\nwait\n'
#end if
return c
#end def run_command
[docs]
def pbs_walltime(self):
walltime=\
str(int(self.hours )).zfill(2)+':'\
+str(int(self.minutes)).zfill(2)+':'\
+str(int(self.seconds)).zfill(2)
if self.days!=0:
walltime = str(self.days)+':'+walltime
#end if
return walltime
#end def pbs_walltime
[docs]
def sbatch_walltime(self):
walltime=\
str(int(24*self.days+self.hours)).zfill(2)+':'\
+str(int(self.minutes)).zfill(2)+':'\
+str(int(self.seconds)).zfill(2)
return walltime
#end def sbatch_walltime
[docs]
def ll_walltime(self):
walltime=\
str(int(24*self.days+self.hours)).zfill(2)+':'\
+str(int(self.minutes)).zfill(2)+':'\
+str(int(self.seconds)).zfill(2)
return walltime
#end def ll_walltime
[docs]
def lsf_walltime(self):
walltime=\
str(int(24*self.days+self.hours)).zfill(2)+':'\
+str(int(self.minutes)).zfill(2)
return walltime
#end def lsf_walltime
[docs]
def normalize_time(self):
t = self.total_seconds()
d = int(t/(24*3600))
t -= d*24*3600
h = int(t/3600)
t -= h*3600
m = int(t/60)
t -= m*60
s = int(t)
self.days = d
self.hours = h
self.minutes = m
self.seconds = s
#end def normalize_time
[docs]
def total_seconds(self):
return self.seconds+60*(self.minutes+60*(self.hours+24*self.days))
#end def total_seconds
[docs]
def total_minutes(self):
return int(self.total_seconds()/60)
#end def total_minutes
[docs]
def total_hours(self):
return int(self.total_seconds()/3600)
#end def total_hours
[docs]
def total_days(self):
return int(self.total_seconds()/(24*3600))
#end def total_days
[docs]
def clone(self):
job = self.copy()
job.set_id()
return job
#end def clone
[docs]
def serial_clone(self):
kw = self.init_info.copy()
kw.serial=True
return Job(**kw)
#end def serial_clone
[docs]
def split_nodes(self,n):
run_options = self.run_options
if not isinstance(n,int):
self.error('cannot split job by nodes\nrequested split value must be an integer\nreceived type: {0}\nwith value: {1}'.format(n.__class__.__name__,n))
elif n<1 or n>=self.nodes:
self.error('cannot split job by nodes\nrequested split must be in the range [1,{0})\nrequested split: {1}'.format(self.nodes,n))
#end if
m = self.get_machine()
if m.app_launcher=='srun':
self.error('splitting jobs by nodes is not currently supported on machine "{0}" (SLURM)'.format(m.name))
#end if
job1 = self.clone()
job2 = self.clone()
job1.nodes = n
job2.nodes = self.nodes - n
m.process_job(job1)
m.process_job(job2)
return job1,job2
#end def split_nodes
#end class Job
[docs]
class Machine(NexusCore):
machines = dict()
modes = obj(
none = 0,
interactive = 1,
batch = 2
)
mode = modes.none
batch_capable = False
requires_account = False
executable_subfile = False
redirect_output = False
query_with_username = False
special_bundling = False
prefixed_output = False
outfile_extension = None
errfile_extension = None
allow_warnings = True
queue_configs = None
[docs]
@staticmethod
def get_hostname():
hostname = gethostname()
if '.' in hostname:
machine_name = hostname.split('.')[0]
else:
machine_name = hostname
#end if
return machine_name.lower()
#end def get_hostname
[docs]
@staticmethod
def exists(machine_name):
return machine_name in Machine.machines
#end def exists
[docs]
@staticmethod
def is_unique(machine):
return id(machine)==id(Machine.machines[machine.name])
#end def is_unique
[docs]
@staticmethod
def add(machine):
if not isinstance(machine,Machine):
Machine.class_error('attempted to add non-machine instance')
#end if
if 'name' not in machine:
Machine.class_error('attempted to add a machine without a name')
#end if
name = machine.name
if name not in Machine.machines:
Machine.machines[name] = machine
else:
Machine.class_error('attempted to create machine {0}, but it already exists'.format(name))
#end if
#end def add
[docs]
@staticmethod
def get(machine_name):
if isinstance(machine_name,str):
machine_name = machine_name.lower()
else:
Machine.class_error('machine name must be a string, you provided a '+machine_name.__class__.__name__)
#end if
if Machine.exists(machine_name):
machine = Machine.machines[machine_name]
else:
machs = sorted(Machine.machines.keys())
Machine.class_error('attempted to get machine '+machine_name+', but it is unknown\nknown options are '+str(machs))
#end if
return machine
#end def get
[docs]
def warn(self,*args,**kwargs):
if Machine.allow_warnings:
NexusCore.warn(self,*args,**kwargs)
#end if
#end def warn
[docs]
def validate(self):
if Machine.exists(self.name):
if not Machine.is_unique(self):
self.error('duplicate instance of machine '+self.name+' encountered\n this is either a developer error, or you have created a duplicate machine')
#end if
else:
self.error('machine {0} id {1} was created without calling Machine.__init__() and is therefore invalid'.format(self.name,id(self)))
#end if
#end def validate
[docs]
def in_batch_mode(self):
return self.mode==self.modes.batch
#end def in_batch_mode
[docs]
def query_queue(self):
self.not_implemented()
#end def query_queue
[docs]
def submit_jobs(self):
self.not_implemented()
#end def submit_jobs
# update all job information, must be idempotent
[docs]
def process_job(self,job):
self.not_implemented()
#end def process_job
[docs]
def process_job_options(self,job):
self.not_implemented()
#end def process_job_options
[docs]
def write_job(self,job,file=False):
self.not_implemented()
#end def write_job
[docs]
def submit_job(self,job):
self.not_implemented()
#end def submit_job
[docs]
def specialized_bundle_commands(self,job,launcher,serial):
self.not_implemented()
#end def specialized_bundle_commands
def __init__(self,name,queue_size=0):
# deferred type change for Machine.machines (protects testing)
if isinstance(Machine.machines,dict):
Machine.machines = obj(**Machine.machines)
self.name = name
self.queue_size = queue_size
self.processes = obj()
self.jobs = obj()
self.waiting = set()
self.running = set()
self.finished= set()
#user defined variables
self.account = None
self.user = None
self.local_directory = None
self.app_directory = None
self.app_directories = None
if not isinstance(name,str):
self.error('machine name must be a string\nyou provided '+str(name))
#end if
Machine.add(self)
#end def __init__
[docs]
def restore_default_settings(self):
self.account = None
self.user = None
self.local_directory = None
self.app_directory = None
self.app_directories = None
#end def restore_default_settings
[docs]
def add_job(self,job):
if isinstance(job,Job):
if job.template is None:
self.process_job(job)
#end if
self.write_job(job)
jid = job.internal_id
self.jobs[jid] = job
job.status = job.states.waiting
self.waiting.add(jid)
#self.write_job_states('add_job')
else:
self.error('add_job received non-Job instance '+job.__class__.__name__)
#end if
#end def add_job
[docs]
def requeue_job(self,job):
None
#end def requeue_job
allowed_user_info = set(['account','local_directory','app_directory','app_directories'])
[docs]
def incorporate_user_info(self,infoin):
info = obj(**infoin)
vars = set(info.keys())
invalid = vars-self.allowed_user_info
if len(invalid)>0:
self.error('invalid inputs encountered in incorporate_user_info\nallowed inputs: {0}\n invalid inputs: {1}'.format(list(self.allowed_user_info),list(invalid)))
#end if
if 'app_directories' in info:
ad = info.app_directories
if not isinstance(ad,dict) and not isinstance(ad,obj):
self.error('app_directories must be of type dict or obj\nyou provided '+ad.__class__.__name__)
#end if
#end if
self.transfer_from(info)
#end def incorporate_user_info
#end class Machine
[docs]
class Workstation(Machine):
mode = Machine.modes.interactive
batch_capable = False
def __init__(self,
name = 'workstation',
cores = None,
app_launcher = 'mpirun',
process_granularity = 1
):
Machine.__init__(self,name)
self.app_launcher = app_launcher
if cores is None:
self.cores = get_cpu_cores()
else:
self.cores = cores
#end if
self.queue_size = cores
self.process_granularity = process_granularity
#end def __init__
[docs]
def process_job(self,job):
if job.serial_only():
job.cores=1
elif job.cores is None:
if job.processes is not None:
job.cores = job.processes*job.threads
else:
job.cores = self.cores
#end if
#end if
job.processes = max(1,int(np.floor(float(job.cores)/job.threads)))
grains = int(np.ceil(float(job.cores)/self.process_granularity))
if abs(grains-1-float(job.cores)/self.process_granularity)<1e-6:
grains-=1
#end if
job.grains = grains
job.cores = grains*self.process_granularity
self.process_job_options(job)
#end def process_job
[docs]
def process_job_options(self,job):
job.run_options.add(np='-np '+str(job.processes))
#end def process_job_options
[docs]
def write_job_states(self,title=''):
self.log(title,n=2)
n=3
self.log('{0} {1} {2} job states'.format(self.__class__.__name__,self.name,id(self)),n=n )
self.log('processes',n=n+1)
for process in self.processes:
job = process.job
self.log('{0:>4} {1:>10} {2:>4} {3}'.format(job.internal_id,job.name,job.simid,job.directory),n=n+2)
#end for
self.log('jobs',n=n+1)
jobids = list(self.jobs.keys())
jobids.sort()
for jobid in jobids:
job = self.jobs[jobid]
self.log('{0:>4} {1:>10} {2:>4} {3}'.format(job.internal_id,job.name,job.simid,job.directory),n=n+2)
#end for
self.log('waiting',n=n+1)
jobids = list(self.waiting)
jobids.sort()
for jobid in jobids:
job = self.jobs[jobid]
self.log('{0:>4} {1:>10} {2:>4} {3}'.format(job.internal_id,job.name,job.simid,job.directory),n=n+2)
#end for
self.log('running',n=n+1)
jobids = list(self.running)
jobids.sort()
for jobid in jobids:
job = self.jobs[jobid]
self.log('{0:>4} {1:>10} {2:>4} {3}'.format(job.internal_id,job.name,job.simid,job.directory),n=n+2)
#end for
self.log('finished',n=n+1)
jobids = list(self.finished)
jobids.sort()
for jobid in jobids:
job = self.jobs[jobid]
self.log('{0:>4} {1:>10} {2:>4} {3}'.format(job.internal_id,job.name,job.simid,job.directory),n=n+2)
#end for
self.log('end job states',n=1)
#end def write_job_states
[docs]
def query_queue(self):
#self.write_job_states('query queue')
self.validate()
done = []
for pid,process in self.processes.items():
if nexus_core.generate_only or not nexus_core.monitor:
qpid,status = pid,0
else:
qpid,status = os.waitpid(pid,os.WNOHANG)
#end if
if pid==qpid:
job = process.job
job.status = job.states.finished
job.finished = True
job.determine_end_status(status)
iid = job.internal_id
self.running.remove(iid)
self.finished.add(iid)
done.append(pid)
if not nexus_core.generate_only:
job.out.close()
job.err.close()
#end if
#end if
#end for
for pid in done:
del self.processes[pid]
#end for
#end def query_queue
[docs]
def submit_jobs(self):
cores_used = 0
for process in self.processes:
cores_used += process.job.cores
#end for
cores_available = self.cores-cores_used
core_req = []
job_req = []
for iid in self.waiting:
job = self.jobs[iid]
job_req.append(job)
core_req.append(job.cores)
#end for
core_req = np.array(core_req,dtype=int)
# The following line does not work correctly under Numpy 1.10 or greater.
# It should create an ndarray of Job objects from a list of Job objects.
# Instead it creates nested ndarray's with inner type of bool
#job_req = np.array(job_req ,dtype=object)
job_req_tmp = job_req
job_req = np.empty(len(job_req_tmp) ,dtype=object)
for idx,job in enumerate(job_req_tmp):
job_req[idx] = job
#end for
order = core_req.argsort()
job_req = job_req[order]
for job in job_req:
if job.cores>self.cores and not nexus_core.generate_only:
self.error('job '+str(job.internal_id)+' is too large to run on this machine\ncores requested: '+str(job.cores)+'\nmachine cores: '+str(self.cores))
#end if
if job.cores<=cores_available:
iid = job.internal_id
self.waiting.remove(iid)
self.running.add(iid)
self.submit_job(job)
cores_available-=job.cores
elif job.cores>self.cores:
self.error('job requested more cores than are present on '+self.name+'\ncores requested: {0}\ncores present: {1}'.format(job.cores,self.cores))
else:
break
#end if
#end for
#end def submit_jobs
[docs]
def job_command(self,job,pad=None):
command = 'export OMP_NUM_THREADS='+str(job.threads)+'\n'
if len(job.presub)>0:
command += job.presub+'\n'
#end if
if job.serial is not None:
command += job.run_command(self.app_launcher,serial=job.serial)
else:
command += job.run_command(self.app_launcher)
#end if
if len(job.postsub)>0:
command += job.postsub+'\n'
#end if
if pad is not None:
command = ('\n'+command).replace('\n','\n '+pad)
#end if
return command
#end def job_command
[docs]
def write_job(self,job,file=False):
c = self.job_command(job)
return c
#end def write_job
[docs]
def submit_job(self,job):
pad = self.enter(job.directory,msg=job.simid)
command = self.job_command(job,pad=pad)
job.status = job.states.running
process = obj()
process.job = job
if nexus_core.generate_only:
self.log(pad+'Would have executed: '+command)
job.system_id = job.internal_id
else:
if nexus_core.monitor:
self.log(pad+'Executing: '+command)
job.out = open(job.outfile,'w')
job.err = open(job.errfile,'w')
p = Popen(command,env=job.env,stdout=job.out,stderr=job.err,shell=True)
process.popen = p
job.system_id = p.pid
else:
command+=' >'+job.outfile+' 2>'+job.errfile+'&'
self.log(pad+'Executing: '+command)
os.system(command)
job.system_id = job.internal_id
#end if
#end if
self.processes[job.system_id] = process
self.leave()
#end def submit_job
#end class Workstation
# test needed
[docs]
class InteractiveCluster(Workstation):
def __init__(self,*args,**kwargs):
if len(args)==0 or not isinstance(args[0],Supercomputer):
self.init_from_args(*args,**kwargs)
else:
super = args[0]
cores = args[1]
self.init_from_supercomputer(super,cores)
#end if
Machine.__init__(self,self.name,self.queue_size)
#end def __init__
[docs]
def init_from_args(self,
name = 'icluster',
nodes = None,
procs_per_node = None,
cores_per_proc = None,
process_granularity = None,
ram_per_node = None,
app_launcher = None
):
self.name = name
self.nodes = nodes
self.procs_per_node = procs_per_node
self.cores_per_proc = cores_per_proc
self.process_granularity = process_granularity
self.ram_per_node = ram_per_node
self.app_launcher = app_launcher
self.cores_per_node = self.cores_per_proc*self.procs_per_node
if process_granularity is None:
self.process_granularity = self.cores_per_node
#end if
self.procs = self.procs_per_node*self.nodes
self.cores = self.cores_per_proc*self.procs
self.ram = self.ram_per_node*self.nodes
self.queue_size = self.cores
#end def init_from_args
[docs]
def init_from_supercomputer(self,super,cores):
nodes = cores//super.cores_per_node
if cores-nodes*super.cores_per_node!=0:
self.error('interactive cores corresponds to a fractional number of nodes\n cores '+str(cores)+'\n cores per node '+str(super.cores_per_node))
#end if
self.init_from_args(super.name+'_interactive',nodes,super.procs_per_node,
super.cores_per_proc,super.cores_per_node,
super.ram_per_node,super.app_launcher)
#end def init_from_supercomputer
[docs]
def process_job(self,job):
job.cores = min(job.cores,self.cores)
Workstation.process_job(self,job)
job.nodes = job.grains
job.procs = job.nodes*self.procs_per_node
if np.mod(job.processes,job.nodes)!=0:
job.processes_per_node = None
else:
job.processes_per_node = job.processes//job.nodes
#end if
if np.mod(job.processes,job.procs)!=0:
job.processes_per_proc = None
else:
job.processes_per_proc = job.processes//job.procs
#end if
#end def process_job
#end class InteractiveCluster
[docs]
class Supercomputer(Machine):
mode = Machine.modes.batch
name = 'supercomputer'
batch_capable = False #only set to true for specific machines
aprun_options = set(['n','d'])
required_inputs = [
'nodes',
'procs_per_node',
'cores_per_proc',
'ram_per_node',
'queue_size',
'app_launcher',
'sub_launcher',
'queue_querier',
'job_remover'
]
def __init__(self,
nodes = None,
procs_per_node = None,
cores_per_proc = None,
ram_per_node = None,
queue_size = 0,
app_launcher = None,
sub_launcher = None,
queue_querier = None,
job_remover = None,
name = None,
):
if name is None:
if self.name is not None:
name = self.name
else:
name = self.__class__.__name__.lower()
#end if
#end if
Machine.__init__(self,name)
self.nodes = nodes # # of nodes
self.procs_per_node = procs_per_node # # of processors/sockets on a node
self.cores_per_proc = cores_per_proc # # of cores on a processor/socket
self.ram_per_node = ram_per_node
self.queue_size = queue_size
self.app_launcher = app_launcher
self.sub_launcher = sub_launcher
self.queue_querier = queue_querier
self.job_remover = job_remover
for var in Supercomputer.required_inputs:
if self[var] is None:
self.error('input variable '+var+' is required to initialize Supercomputer object.')
#end if
#end for
self.cores_per_node = self.cores_per_proc*self.procs_per_node
self.procs = self.procs_per_node*self.nodes
self.cores = self.cores_per_proc*self.procs
self.ram = self.ram_per_node*self.nodes
# 'complete' is the only actively used status so far
# At least one queue state should correspond to 'complete',
# though even this is not strictly necessary.
# In general if a pid is missing from the queue,
# that job is assumed to be complete.
self.system_queue = obj()
if self.queue_querier=='qstat':
self.job_states=dict(E = 'exiting',
H = 'held',
Q = 'queued',
R = 'running',
S = 'suspended',
T = 'transferring',
W = 'waiting',
C = 'complete',
F = 'complete',
B = 'has_subjob',
M = 'moved_to_another_server',
U = 'user_suspended',
X = 'subjob_completed',
)
elif self.queue_querier=='qstata':
#already gives status as queued, running, etc.
None
elif self.queue_querier=='squeue':
self.job_states=dict(CG = 'exiting',
TO = 'timeout',
CA = 'failed',
F = 'failed',
NF = 'node_fail',
PD = 'waiting',
R = 'running',
S = 'suspended',
CD = 'complete',
RD = 'held',
BF = 'failed',
CF = 'configuring',
DL = 'deadline',
OOM= 'out_of_memory',
PR = 'preempted',
RF = 'requeue_fed',
RH = 'requeue_hold',
RQ = 'requeued',
RS = 'resizing',
RV = 'revoked',
SI = 'signaling',
SE = 'special_exit',
SO = 'stage_out',
ST = 'stopped',
)
elif self.queue_querier=='sacct':
self.job_states=dict(CANCELLED = 'failed', #long form
COMPLETED = 'complete',
COMPLETING = 'exiting',
CONFIGURING = 'waiting',
FAILED = 'failed',
PREMEEMPTED = 'failed',
PENDING = 'waiting',
NODE_FAIL = 'failed',
RESIZING = 'resizing',
RUNNING = 'running',
SUSPENDED = 'suspended',
TIMEOUT = 'failed',
CA = 'failed', #short form
CD = 'complete',
CG = 'exiting',
CF = 'waiting',
F = 'failed',
PR = 'failed',
PD = 'waiting',
NF = 'failed',
RS = 'resizing',
R = 'running',
S = 'suspended',
TO = 'failed'
)
elif self.queue_querier=='llq':
self.job_states=dict(I = 'idle',
NQ = 'not_queued',
H = 'user_hold',
S = 'system_hold',
HS = 'user_system_hold',
D = 'deferred',
R = 'running',
P = 'pending',
ST = 'starting',
C = 'complete',
CA = 'canceled',
E = 'preempted',
EP = 'preempt_pending',
MP = 'resume_pending',
)
elif self.queue_querier=='bjobs':
self.job_states=dict(PEND = 'pending',
RUN = 'running',
DONE = 'complete',
EXIT = 'failed',
PSUSP = 'suspended',
USUSP = 'suspended',
SSUSP = 'suspended',
)
elif self.queue_querier=='test_query':
None
else:
self.error('ability to query queue with '+self.queue_querier+' has not yet been implemented')
#end if
#end def __init__
# test needed
[docs]
def interactive_representation(self,cores):
return InteractiveCluster(self,cores)
#end def interactive_representation
# test needed
[docs]
def requeue_job(self,job):
if isinstance(job,Job):
jid = job.internal_id
pid = job.system_id
if pid is None:
self.error('job {0} does not have a process id issued by the scheduler'.format(jid))
#end if
self.process_job(job)
self.jobs[jid] = job
if not nexus_core.dynamic:
self.running.add(jid)
process = obj(job=job)
self.processes[pid] = process
else:
# If a workstation job is requeued
# then it is waking from interruption
# and should be resubmitted from the top
job.status = job.states.running
job.status = job.states.waiting
self.waiting.add(jid)
else:
self.error('requeue_job received non-Job instance '+job.__class__.__name__)
#end if
#end def requeue_job
[docs]
def process_job(self,job):
if job.fake_job:
return
#end if
self.pre_process_job(job)
job.subfile = job.name+'.'+self.sub_launcher+'.in'
no_cores = job.cores is None
no_nodes = job.nodes is None
if no_cores and no_nodes:
self.error('job did not specify cores or nodes\nAt least one must be provided')
elif no_cores:
job.cores = self.cores_per_node*job.nodes
elif no_nodes:
job.nodes = int(np.ceil(float(job.cores)/self.cores_per_node))
if abs(job.nodes-1-float(job.cores)/self.cores_per_node)<1e-6:
job.nodes-=1
#end if
else:
job.cores = min(job.cores,job.nodes*self.cores_per_node)
#end if
if job.processes_per_node is not None:
job.processes = job.nodes*job.processes_per_node
else:
job.processes = max(1,int(float(job.cores)/job.threads))
#end if
job.tot_cores = job.nodes*self.cores_per_node
job.procs = job.nodes*self.procs_per_node
if np.mod(job.processes,job.nodes)!=0:
job.processes_per_node = None
else:
job.processes_per_node = job.processes//job.nodes
#end if
if np.mod(job.processes,job.procs)!=0:
job.processes_per_proc = None
else:
job.processes_per_proc = job.processes//job.procs
#end if
if job.ppn is None:
job.ppn = self.cores_per_node
#end if
if job.account is None:
if self.account is not None:
job.account = self.account
elif self.requires_account:
self.error('account not specified for job on '+self.name)
#end if
#end if
self.post_process_job(job)
job.set_environment(OMP_NUM_THREADS=job.threads)
self.process_job_options(job)
#end def process_job
[docs]
def process_job_options(self,job):
launcher = self.app_launcher
if launcher=='mpirun':
job.run_options.add(np='-np '+str(job.processes))
elif launcher=='mpiexec':
job.run_options.add(n='-n '+str(job.processes))
elif launcher=='aprun':
if 'n' in self.aprun_options:
job.run_options.add(n='-n '+str(job.processes))
#end if
if 'd' in self.aprun_options and job.threads>1:
job.run_options.add(d='-d '+str(job.threads))
#end if
if 'N' in self.aprun_options and job.processes_per_node is not None:
job.run_options.add(N='-N '+str(job.processes_per_node))
#end if
if 'S' in self.aprun_options and job.processes_per_proc is not None:
job.run_options.add(S='-S '+str(job.processes_per_proc))
#end if
elif launcher=='runjob':
#bypass setup_environment
if job.env is not None:
envs='--envs'
for name,value in job.env.items():
envs+=' {0}={1}'.format(name,value)
#end for
job.env = None
elif 'envs' in job.run_options:
envs = job.run_options.envs
else:
self.error('failed to set env options for runjob')
#end if
job.run_options.add(
np = '--np '+str(job.processes),
p = '-p '+str(job.processes_per_node),
xlocargs = '$LOCARGS',
verbose = '--verbose=INFO',
envs = envs
)
elif launcher=='srun': # Amos contribution from Ryan McAvoy
None
elif launcher=='ibrun': # Lonestar contribution from Paul Young
job.run_options.add(
np = '-n '+str(job.processes),
p = '-o '+str(0),
)
elif launcher=='jsrun': # Summit
None # Summit class takes care of this in post_process_job
elif launcher=='lrun': # Lassen
None # Lassen class takes care of this in post_process_job
else:
self.error(launcher+' is not yet implemented as an application launcher')
#end if
#end def process_job_options
[docs]
def pre_process_job(self,job):
None
#end def pre_process_job
[docs]
def post_process_job(self,job):
None
#end def post_process_job
[docs]
def query_queue(self,out=None):
self.system_queue.clear()
if self.query_with_username and self.user is None:
self.error('querying queue on machine "{}" requires user name\nplease provide username via the "user" keyword in settings'.format(self.name))
#end if
if self.queue_querier=='qstat':
if out is None:
out,err,rc = execute('qstat -a')
#end if
lines = out.splitlines()
for line in lines:
tokens=line.split()
if len(tokens)>0:
if '.' in tokens[0]:
spid = tokens[0].split('.')[0]
else:
spid = tokens[0]
#endif
if spid.isdigit() and len(tokens)==11:
pid = int(spid)
jid,uname,queue,jname,sessid,nodes,tasks,mem,rtime,status,etime = tokens
if status in self.job_states:
self.system_queue[pid] = self.job_states[status]
else:
self.error('job state '+status+' is unrecognized')
#end if
#end if
#end if
#end for
elif self.queue_querier=='qstata':
if out is None:
out,err,rc = execute('qstat')
#end if
lines = out.splitlines()
for line in lines:
tokens=line.split()
if len(tokens)>0:
if '.' in tokens[0]:
spid = tokens[0].split('.')[0]
else:
spid = tokens[0]
#endif
if spid.isdigit() and len(tokens)==6:
pid = int(spid)
jid,uname,wtime,nodes,status,loc = tokens
self.system_queue[pid] = status
#end if
#end if
#end for
elif self.queue_querier=='squeue': # contributed by Ryan McAvoy
if out is None:
if isinstance(self.user,bool) and self.user==False:
extra = ''
elif self.user is not None:
extra = ' -u {}'.format(self.user)
else:
extra = ' --user=$USER'
#end if
out,err,rc = execute('squeue'+extra)
#end if
lines = out.splitlines()
for line in lines:
tokens=line.split()
if len(tokens)>0:
if '.' in tokens[0]:
spid = tokens[0].split('.')[0]
else:
spid = tokens[0]
#endif
if spid.isdigit():
pid = int(spid)
status = None
jid,loc,name,uname,status,wtime,nodes,reason = tokens[:8]
if status is not None:
if status in self.job_states:
self.system_queue[pid] = self.job_states[status]
else:
self.error('job state '+status+' is unrecognized')
#end if
#end if
#end if
#end if
#end for
elif self.queue_querier=='sacct': # contributed by Ryan McAvoy
if out is None:
out,err,rc = execute('sacct')
#end if
lines = out.splitlines()
for line in lines:
tokens=line.split()
if len(tokens)>0:
if '.' in tokens[0]:
spid = tokens[0].split('.')[0]
else:
spid = tokens[0]
#endif
if spid.isdigit() and len(tokens)==6: #if account is empty, only 6 tokens.
pid = int(spid)
jid,name,loc,cores,status,exit_code = tokens
status = status.split('+')[0] ## get rid of '+' in the end
if status in self.job_states:
self.system_queue[pid] = self.job_states[status]
else:
self.error('job state '+status+' is unrecognized')
#end if
elif spid.isdigit() and len(tokens)==7:
pid = int(spid)
jid,name,loc,uname,cores,status,exit_code = tokens
status = status.split('+')[0] ## get rid of '+' in the end
if status in self.job_states:
self.system_queue[pid] = self.job_states[status]
else:
self.error('job state '+status+' is unrecognized')
#end if
#end if
#end if
#end for
elif self.queue_querier=='llq':
if out is None:
out,err,rc = execute('sacct')
#end if
lines = out.splitlines()
for line in lines:
tokens=line.split()
if len(tokens)>0:
if '.' in tokens[0]:
spid = tokens[0].split('.')[1]
else:
spid = tokens[0]
#endif
if spid.isdigit() and (len(tokens)==7 or len(tokens)==8):
pid = int(spid)
if len(tokens)==7:
jid,owner,subdate,subtime,status,pri,class_ = tokens
elif len(tokens)==8:
jid,owner,subdate,subtime,status,pri,class_,running_on = tokens
#end if
if status in self.job_states:
self.system_queue[pid] = self.job_states[status]
else:
self.error('job state '+status+' is unrecognized')
#end if
#end if
#end if
#end for
elif self.queue_querier=='bjobs':
if out is None:
out,err,rc = execute('bjobs')
#end if
lines = out.splitlines()
for line in lines:
tokens=line.split()
if len(tokens)>0:
spid = tokens[0]
if spid.isdigit() and len(tokens)==8:
pid = int(spid)
jid,uname,status,slots,queue,start,finish,jname = tokens
if status in self.job_states:
self.system_queue[pid] = self.job_states[status]
else:
self.error('job state '+status+' is unrecognized')
#end if
#end if
#end if
#end for
elif self.queue_querier=='test_query': # for testing
# pretend that all jobs have finished
for pid in self.processes.keys():
self.system_queue[pid] = 'complete'
#end for
else:
self.error('ability to query queue with '+self.queue_querier+' has not yet been implemented')
#end if
done = []
for pid,process in self.processes.items():
if pid not in self.system_queue or self.system_queue[pid]=='complete' or nexus_core.generate_only:
job = process.job
job.status = job.states.finished
job.finished = True
iid = job.internal_id
self.running.remove(iid)
self.finished.add(iid)
done.append(pid)
#end if
#end for
for pid in done:
del self.processes[pid]
#end for
return self.system_queue
#end def query_queue
[docs]
def submit_jobs(self):
nprocesses_running = len(self.processes)
queue_slots_available = self.queue_size-nprocesses_running
remove = []
for iid in self.waiting:
if queue_slots_available>0:
remove.append(iid)
self.running.add(iid)
job = self.jobs[iid]
self.submit_job(job)
queue_slots_available -= 1
else:
break
#end if
#end for
for iid in remove:
self.waiting.remove(iid)
#end for
#end def submit_jobs
[docs]
def submit_job(self,job):
pad = self.enter(job.directory,msg=job.internal_id)
if job.subfile is None:
self.error('submission file not specified for job')
elif not os.path.exists(job.subfile):
self.error('job submission file was not written prior to submission\n submission file: '+os.path.join(job.directory,job.subfile))
#end if
command = self.sub_command(job)
if nexus_core.generate_only:
self.log(pad+'Would have executed: '+command)
job.status = job.states.running
process = obj()
process.job = job
self.processes[job.internal_id] = process
else:
self.log(pad+'Executing: '+command)
job.status = job.states.running
process = obj()
process.job = job
out,err,rc = execute(command)
output=out+'\n'+err
pid = self.read_process_id(output)
if pid is None:
self.error('process id could not be determined from submission output\n output:\n'+output)
else:
self.log(pad+' pid: {0}'.format(pid))
#end if
#pid = 'fakepid_'+str(job.internal_id)
job.system_id = pid
self.processes[pid] = process
#end if
self.leave()
#end def submit_job
[docs]
def sub_command(self,job):
return self.sub_launcher+job.sub_options.write()+' '+job.subfile
#end def sub_command
[docs]
def remove_job(self,job):
if self.job_remover=='qdel':
command = 'qdel '+str(job.system_id)
elif self.job_remover=='scancel':
command = 'scancel '+str(job.system_id)
else:
self.error('ability to remove job using '+self.job_remover+' has not yet been implemented')
#endif
os.system(command)
#end def remove_job
[docs]
def setup_environment(self,job):
env = ''
if job.env is not None:
for name,val in job.env.items():
env +='export {0}={1}\n'.format(name,val)
#end for
#end if
return env
#end def setup_environment
[docs]
def write_job(self,job,file=False):
job.subfile = job.name+'.'+self.sub_launcher+'.in'
if job.template is None:
env = self.setup_environment(job)
command = job.run_command(self.app_launcher,serial=job.serial)
c = self.write_job_header(job)+'\n'
if len(job.presub)>0:
c+=job.presub+'\n'
#end if
c+=env
c+=command+'\n'
if len(job.postsub)>0:
c+=job.postsub+'\n'
#end if
else:
c = job.template
#end if
if file:
filepath = os.path.join(job.directory,job.subfile)
fobj = open(filepath,'w')
fobj.write(c)
fobj.close()
if self.executable_subfile:
os.system('chmod +x '+filepath)
#end if
#end if
return c
#end def write_job
#end def write_job_header
[docs]
@staticmethod
def walltime_to_seconds(walltime_str):
"""
Convert walltime string to total seconds
Handles formats: 'dd:hh:mm:ss', 'hh:mm:ss', 'mm:ss', 'seconds'
"""
try:
parts = walltime_str.split(':')
seconds = 0
if len(parts) == 4: # dd:hh:mm:ss
seconds += int(parts[0]) * 24 * 3600 # days
seconds += int(parts[1]) * 3600 # hours
seconds += int(parts[2]) * 60 # minutes
seconds += int(parts[3]) # seconds
elif len(parts) == 3: # hh:mm:ss
seconds += int(parts[0]) * 3600 # hours
seconds += int(parts[1]) * 60 # minutes
seconds += int(parts[2]) # seconds
elif len(parts) == 2: # mm:ss
seconds += int(parts[0]) * 60 # minutes
seconds += int(parts[1]) # seconds
elif len(parts) == 1: # seconds only
seconds += int(parts[0])
else:
raise ValueError(f"Invalid walltime format: {walltime_str}, accepted formats: 'dd:hh:mm:ss', 'hh:mm:ss', 'mm:ss', 'seconds'")
return seconds
except (ValueError, IndexError) as e:
raise ValueError(f"Failed to parse walltime '{walltime_str}': {str(e)}")
#end try
#end def walltime_to_seconds
[docs]
@ staticmethod
def seconds_to_walltime(seconds):
"""
Convert total seconds to walltime string
Handles formats: 'dd:hh:mm:ss', 'hh:mm:ss', 'mm:ss', 'seconds'
"""
days = seconds // 86400
hours = (seconds % 86400) // 3600
minutes = (seconds % 3600) // 60
seconds = seconds % 60
return f"{days}:{hours:02d}:{minutes:02d}:{seconds:02d}"
#end def seconds_to_walltime
[docs]
def validate_queue_config(self, job):
"""Validate job against queue configuration constraints
Returns
-------
bool
``True`` if valid, raises an error with detailed message if invalid
Examples
--------
Queue config format
>>> queue_configs = {
... 'default': 'queue_name',
... 'queue_name': {
... 'min_nodes' : minimum number of nodes,
... 'max_nodes' : maximum number of nodes,
... 'max_walltime' : maximum walltime in hours,
... 'cores_per_node': cores per node,
... 'ram_per_node' : RAM per node in GB,
... 'constraints' : {
... 'knl': {
... 'max_nodes': knl specific max nodes,
... 'max_time': knl specific max time,
... ...
... },
... 'haswell': {...},
... 'gpu': {...},
... 'cpu': {...}
... }
... }
... }
"""
if self.queue_configs is None:
return True
#end if
# Use only warnings here, as smaller computers may not have a queue
if job.queue is None:
if 'default' in self.queue_configs:
job.queue = self.queue_configs['default']
self.warn('No default queue is specified. Using default queue {}'.format(job.queue))
else:
# No queue or default queue is specified
self.warn('No queue or default queue is specified.')
return True
#end if
#end if
# Check if queue exists
if job.queue not in self.queue_configs:
# Queue is defined but config is not available
self.warn('Queue "{}" is not available. Available queues: {}'.format(
job.queue, list(self.queue_configs.keys())))
return False
else:
# Queue is defined and config is available
config = self.queue_configs[job.queue]
#end if
errors = []
# Get constraint-specific config if applicable
constraint_config = None
if hasattr(job, 'constraint') and job.constraint is not None:
if 'constraints' in config and job.constraint in config['constraints']:
constraint_config = config['constraints'][job.constraint]
else:
valid_constraints = list(config.get('constraints', {}).keys())
if valid_constraints:
errors.append('Invalid constraint "{}". Valid constraints for queue {}: {}'.format(
job.constraint, job.queue, valid_constraints))
#end if
#end if
#end if
# Use constraint specific config if available, otherwise use base config
active_config = constraint_config if constraint_config is not None else config
# Validate nodes
if 'min_nodes' in active_config and job.nodes < active_config['min_nodes']:
errors.append('Number of nodes ({}) is below minimum ({}) for queue {}'.format(
job.nodes, active_config['min_nodes'], job.queue))
#end if
if 'max_nodes' in active_config and job.nodes > active_config['max_nodes']:
errors.append('Number of nodes ({}) exceeds maximum ({}) for queue {}'.format(
job.nodes, active_config['max_nodes'], job.queue))
#end if
# Validate cores per node
if 'cores_per_node' in active_config and hasattr(job, 'processes_per_node'):
if job.processes_per_node > active_config['cores_per_node']:
errors.append('Processes per node ({}) exceeds cores per node ({}) for queue {}'.format(
job.processes_per_node, active_config['cores_per_node'], job.queue))
#end if
#end if
# Validate memory
if 'ram_per_node' in active_config and hasattr(job, 'ram_per_node'):
if job.ram_per_node > active_config['ram_per_node']:
errors.append('RAM per node ({} GB) exceeds maximum ({} GB) for queue {}'.format(
job.ram_per_node, active_config['ram_per_node'], job.queue))
#end if
#end if
# Validate walltime
if 'max_time' in active_config:
job.total_hours = job.days*24 + job.hours + job.minutes/60.0 + job.seconds/3600.0
if job.total_hours > active_config['max_time']:
errors.append('Walltime ({} hours) exceeds maximum ({} hours) for queue {}'.format(
job.total_hours, active_config['max_time'], job.queue))
#end if
#end if
# Validate GPU requirements if specified
if 'gpus_per_node' in active_config and hasattr(job, 'gpus_per_node'):
if job.gpus_per_node > active_config['gpus_per_node']:
errors.append('GPUs per node ({}) exceeds maximum ({}) for queue {}'.format(
job.gpus_per_node, active_config['gpus_per_node'], job.queue))
#end if
#end if
# Report all validation errors
if errors:
self.error('Queue validation failed for queue {}:\n {}'.format(
job.queue, '\n '.join(errors)))
return False
#end if
return True
#end def validate_queue_config
[docs]
def read_process_id(self,output):
pid = None
lines = output.splitlines()
if self.sub_launcher=='llsubmit': # specialization for load leveler (SuperMUC)
for line in lines:
if 'llsubmit: The job' in line and '"' in line:
spid = line.split('"')[1].split('.')[1].strip()
if spid.isdigit():
pid = int(spid)
break
#end if
#end if
#end for
else: # most other machines follow the pattern below
for line in lines:
ls = line.strip()
if ls.isdigit():
pid = int(ls)
break
elif '.' in line:
spid = line.split('.')[0]
if spid.isdigit():
pid = int(spid)
break
#end if
elif ' ' in line: # specialized for Amos?
spid = line.split(' ')[-1]
if spid.isdigit():
pid = int(spid)
break
#end if
#end if
#end for
#end if
return pid
#end def read_process_id
#end class Supercomputer
# Load local class for local cluster's setting from ~/.nexus/local.py
# The following is an example of this file (see machines.py for more examples):
'''
from machines import Supercomputer
class Clustername(Supercomputer):
name = 'clustername'
requires_account = False
batch_capable = True
def write_job_header(self,job):
if job.queue is None:
job.queue='batch'
#end if
c= '#!/bin/bash\n'
c+='#PBS -l nodes={0}:ppn={1}\n'.format(job.nodes,job.ppn)
c+='#PBS -l walltime='+job.pbs_walltime()+'\n'
c+='#PBS -N '+job.name +'\n'
c+='#PBS -o '+job.outfile+'\n'
c+='#PBS -e '+job.errfile+'\n'
c+='#PBS -V\n'
c+='#PBS -q '+job.queue+'\n'
c+=' \n '
return c
#end def write_job_header
#end class Clustername
# nodes sockets cores ram qslots qlaunch qsubmit qstatus qdelete
Clustername( 4, 1, 16, 24, 4, 'mpirun', 'qsub', 'qstat', 'qdel')
'''
try:
our_load_source('*',os.path.expanduser('~/.nexus/local_machines.py'))
except IOError:
pass
except:
raise
#end try
#Decommissioned
[docs]
class Kraken(Supercomputer):
name = 'kraken'
requires_account = True
#end def write_job_header
#end class Kraken
#Decommissioned
[docs]
class Jaguar(Supercomputer):
name = 'jaguar'
requires_account = True
#end def write_job_header
#end class Jaguar
#Unknown
[docs]
class Golub(Supercomputer):
name = 'golub'
#end def write_job_header
#end class Golub
# Decommissioned
[docs]
class OIC5(Supercomputer):
name = 'oic5'
batch_capable = True
#end def write_job_header
[docs]
def read_process_id(self,output):
pid = None
lines = output.splitlines()
for line in lines:
if 'oic.ornl.gov' in line:
spid = line.split('.')[0]
if spid.isdigit():
pid = int(spid)
#end if
#end if
#end for
return pid
#end def read_process_id
#end class OIC5
[docs]
class NerscMachine(Supercomputer):
batch_capable = True
#end def write_job_header
#end class NerscMachine
#Decommissioned
[docs]
class Cori(NerscMachine):
name = 'cori'
[docs]
def pre_process_job(self,job):
if job.queue is None:
job.queue = 'regular'
#end if
if job.constraint is None:
job.constraint = 'knl'
#end if
# account for dual nature of Cori
if 'knl' in job.constraint:
self.nodes = 9688
self.procs_per_node = 1
self.cores_per_node = 68
self.ram_per_node = 96
elif 'haswell' in job.constraint:
self.nodes = 2388
self.procs_per_node = 2
self.cores_per_node = 32
self.ram_per_node = 128
elif 'amd' in job.constraint:
self.nodes = 20
self.procs_per_node = 2
self.cores_per_node = 32
self.ram_per_node = 2048
else:
self.error('SLURM input "constraint" must contain either "knl", "haswell", or "amd" on Cori\nyou provided: {0}'.format(job.constraint))
#end if
if job.core_spec is not None:
self.cores_per_node -= job.core_spec
#end if
#end def pre_process_job
#end def write_job_header
#end class Cori
# Active
[docs]
class Perlmutter(NerscMachine):
name = 'perlmutter'
[docs]
def pre_process_job(self,job):
# Set default queue and node type
if job.queue is None:
job.queue = 'regular'
#end if
if job.constraint is None:
job.constraint = 'cpu'
#end if
# Account for dual nature of Perlmutter
if 'cpu' in job.constraint:
self.nodes = 3072
self.procs_per_node = 2
self.cores_per_node = 128
self.ram_per_node = 512
elif 'gpu' in job.constraint:
self.nodes = 1536
self.procs_per_node = 1
self.cores_per_node = 64
self.ram_per_node = 256
self.gpus_per_node = 4
else:
self.error('SLURM input "constraint" must contain either "cpu" or "gpu" on Perlmutter\nyou provided: {0}'.format(job.constraint))
#end if
#end def pre_process_job
#end def write_job_header
#end class Perlmutter
# Active
[docs]
class BlueWatersXK(Supercomputer):
name = 'bluewaters_xk'
requires_account = False
batch_capable = True
#end def write_job_header
#end class BlueWatersXK
# Active
[docs]
class BlueWatersXE(Supercomputer):
name = 'bluewaters_xe'
requires_account = False
batch_capable = True
#end def write_job_header
#end class BlueWatersXE
#Decommissioned
[docs]
class Titan(Supercomputer):
name = 'titan'
requires_account = True
batch_capable = True
#end def write_job_header
#end class Titan
# Active
[docs]
class EOS(Supercomputer):
name = 'eos'
requires_account = True
batch_capable = True
[docs]
def post_process_job(self,job):
if job.threads>1:
if job.threads<=8:
job.run_options.add(ss='-ss')
#end if
job.run_options.add(cc='-cc numa_node')
#end if
#end def post_process_job
#end def write_job_header
#end class EOS
[docs]
class ALCF_Machine(Supercomputer):
requires_account = True
batch_capable = True
executable_subfile = True
prefixed_output = True
outfile_extension = '.output'
errfile_extension = '.error'
base_partition = None
[docs]
def post_process_job(self,job):
job.sub_options.add(
env = '--env BG_SHAREDMEMSIZE=32',
mode = '--mode script'
)
#if job.processes<job.nodes: # seems like a good idea, but breaks idempotency
# job.processes_per_node=1
##end if
if job.nodes<self.base_partition:
self.warn('!!! ATTENTION !!!\n number of nodes on {0} should not be less than {1}\n you requested: {2}'.format(self.name,self.base_partition,job.nodes))
else:
partition = np.log(float(job.nodes)/self.base_partition)/np.log(2.)
if abs(partition-int(partition))>1e-6:
self.warn('!!! ATTENTION !!!\n number of nodes on {0} must be {1} times a power of two\n you requested: {2}\n nearby valid node count: {3}'.format(self.name,self.base_partition,job.nodes,self.base_partition*2**int(np.round(partition))))
#end if
#end if
valid_ppn = (1,2,4,8,16,32,64)
if job.processes_per_node is None:
self.warn('job may not run properly\nplease specify processes_per_node in each job to be launched with runjob on {0}'.format(self.name))
elif job.processes_per_node not in valid_ppn:
self.warn('job may not run properly\nprocesses_per_node is not a valid value for {0}\nprocesses_per_node provided: {1}\nvalid options are: {2}'.format(self.name,job.processes_per_node,valid_ppn))
#end if
#end def post_process_job
#end def write_job_header
#end class ALCF_Machine
# Decommissioned
[docs]
class Vesta(ALCF_Machine):
name = 'vesta'
base_partition = 32
#end class Vesta
# Decommissioned
[docs]
class Cetus(ALCF_Machine):
name = 'cetus'
base_partition = 128
#end class Cetus
# Decommissioned
[docs]
class Mira(ALCF_Machine):
name = 'mira'
base_partition = 512
#end class Mira
# Active
[docs]
class Cooley(Supercomputer):
name = 'cooley'
requires_account = True
batch_capable = True
executable_subfile = True
prefixed_output = True
outfile_extension = '.output'
errfile_extension = '.error'
[docs]
def post_process_job(self,job):
#if job.processes_per_node is None and job.threads!=1:
# self.error('threads must be 1,2,3,4,6, or 12 on Cooley\nyou provided: {0}'.format(job.threads))
##end if
#job.run_options.add(
#f = '-f $COBALT_NODEFILE',
#ppn = '-ppn {0}'.format(job.processes_per_node),
#)
return
#end def post_process_job
#end def write_job_header
#end class Cooley
# Active
[docs]
class Theta(Supercomputer):
name = 'theta'
requires_account = True
batch_capable = True
executable_subfile = True
prefixed_output = True
outfile_extension = '.output'
errfile_extension = '.error'
[docs]
def post_process_job(self,job):
if job.hyperthreads is None:
job.hyperthreads = 1
#end if
job.run_options.add(
N = '-N {0}'.format(job.processes_per_node),
cc = '-cc depth',
d = '-d {0}'.format(job.threads),
j = '-j {0}'.format(job.hyperthreads),
e = '-e OMP_NUM_THREADS={0}'.format(job.threads),
)
#end def post_process_job
#end def write_job_header
#end class Theta
# Active
[docs]
class Lonestar(Supercomputer): # Lonestar contribution from Paul Young
name = 'lonestar' # will be converted to lowercase anyway
requires_account = False
batch_capable = True
#end def write_job_header
[docs]
def read_process_id(self,output):
pid = None
lines = output.splitlines()
for line in lines:
if 'Your job' in line:
spid = line.split(' ')[2]
if spid.isdigit():
pid = int(spid)
#end if
#end if
#end for
return pid
#end def read_process_id
#end class Lonestar
[docs]
class ICMP_Machine(Supercomputer): # ICMP and Amos contributions from Ryan McAvoy
batch_capable = True
executable_subfile = True
prefixed_output = True
outfile_extension = '.output'
errfile_extension = '.error'
#end def write_job_header
#end class ICMP_Machine
# Unknown
[docs]
class Komodo(ICMP_Machine):
name = 'komodo'
#end class Komodo
# Unknown
[docs]
class Matisse(ICMP_Machine):
name = 'matisse'
#end class Matisse
# Active
[docs]
class Amos(Supercomputer):
name = 'amos'
#requires_account = True
batch_capable = True
executable_subfile = True
prefixed_output = True
outfile_extension = '.output'
errfile_extension = '.error'
#end def write_job_header
#end class Amos
[docs]
class SnlMachine(Supercomputer):
requires_account = True
batch_capable = True
gpu_machine = False
reservation_required = False
prefixed_output = True
outfile_extension = '.output'
errfile_extension = '.error'
#end def write_job_header
#end class SnlMachine
#Unknown
[docs]
class Eclipse(SnlMachine):
name = 'eclipse'
#end class Eclipse
#Unknown
[docs]
class Attaway(SnlMachine):
name = 'attaway'
#end class Attaway
#Unknown
[docs]
class Manzano(SnlMachine):
name = 'manzano'
#end class Manzano
#Unknown
[docs]
class Ghost(SnlMachine):
name = 'ghost'
#end class Ghost
#Unknown
[docs]
class Amber(SnlMachine):
name = 'amber'
#end class Amber
#Unknown
[docs]
class Solo(SnlMachine):
name = 'solo'
#end class Solo
#Unknown
[docs]
class Flight(SnlMachine):
name = 'flight'
reservation_required = True
#end class Flight
[docs]
class Hops(SnlMachine):
name = 'hops'
gpu_machine = True
gpus_per_node = 4
#end class Hops
# Active
# machines at LRZ https://www.lrz.de/english/
[docs]
class SuperMUC(Supercomputer):
name = 'supermuc'
requires_account = False
batch_capable = True
query_with_username = False
#end def write_job_header
#end class SuperMUC
# Active
[docs]
class SuperMUC_NG(Supercomputer):
name = 'supermucng'
requires_account = True
batch_capable = True
query_with_username = True
#end def write_job_header
#end class SuperMUC_NG
# Decommissioned
[docs]
class Stampede2(Supercomputer):
name = 'stampede2'
requires_account = True
batch_capable = True
#executable_subfile = True
prefixed_output = True
outfile_extension = '.output'
errfile_extension = '.error'
#end def write_job_header
#end class Stampede2
# Decommissioned
# CADES at ORNL
[docs]
class CadesMoab(Supercomputer):
name = 'cades_moab'
requires_account = True
batch_capable = True
[docs]
def post_process_job(self,job):
ppn = job.processes_per_node
if job.threads>1 and ppn is not None and ppn>1:
processes_per_socket = int(np.floor(job.processes_per_node/2))
job.run_options.add(npersocket='--npersocket {0}'.format(processes_per_socket))
#end if
#end def post_process_job
#end def write_job_header
#end class CadesMoab
# Active
[docs]
class CadesSlurm(Supercomputer):
name = 'cades'
requires_account = True
batch_capable = True
#end def write_job_header
#end class CadesSlurm
# Active
# Inti at ORNL
[docs]
class Inti(Supercomputer):
name = 'inti'
requires_account = False
batch_capable = True
#end def write_job_header
#end class Inti
# Active
# Baseline at ORNL https://docs.cades.olcf.ornl.gov/baseline_user_guide/baseline_user_guide.html
[docs]
class Baseline(Supercomputer):
name = 'baseline'
requires_account = True
batch_capable = True
queue_configs={
'default': 'batch_cnms',
'batch': {
'max_nodes': 138,
'max_walltime': '24:00:00',
},
'batch_low_memory': {
'max_nodes': 68,
'max_walltime': '24:00:00',
},
'batch_high_memory': {
'max_nodes': 70,
'max_walltime': '24:00:00',
},
'batch_ccsi': {
'max_nodes': 20,
'max_walltime': '24:00:00',
},
'batch_cnms': {
'max_nodes': 20,
'max_walltime': '24:00:00',
},
'gpu_acmhs': {
'max_nodes': 1,
'max_walltime': '24:00:00',
}
}
#end def write_job_header
#end class Baseline
[docs]
class Frontier(Supercomputer):
name = 'frontier'
requires_account = True
batch_capable = True
queue_configs = {
'default': 'batch',
'batch': {
'max_nodes': 9664,
'max_walltime': '12:00:00',
},
'extended': {
'max_nodes': 1894,
'max_walltime': '24:00:00',
},
'debug': {
'max_nodes': 1,
'max_walltime': '01:00:00',
}
}
[docs]
def pre_process_job(self,job):
# Set default queue and node type
if job.queue is None:
job.queue = 'batch'
#end if
if job.constraint is None:
job.constraint = 'cpu'
#end if
# Account for dual nature of Frontier
if 'cpu' in job.constraint:
self.cores_per_node = 56
elif 'gpu' in job.constraint:
self.cores_per_node = 56
self.gpus_per_node = 4
else:
self.error('SLURM input "constraint" must contain either "cpu" or "gpu" on Frontier\nyou provided: {0}'.format(job.constraint))
#end if
#end def pre_process_job
[docs]
def post_process_job(self, job):
if 'cpu' in job.constraint:
job.run_options.add(
cpu_bind='--cpu-bind=threads',
threads_per_core='--threads-per-core={0}'.format(job.threads)
)
elif 'gpu' in job.constraint:
gpus_per_task = int(np.floor(float(self.gpus_per_node)/job.processes_per_node))
job.run_options.add(
gpu_bind='--gpu-bind=closest',
gpus_per_task='--gpus-per-task={0}'.format(gpus_per_task)
)
#end if
job.run_options.add(
N='-N {}'.format(job.nodes),
n='-n {}'.format(job.processes),
c='-c {}'.format(job.threads),
)
#end class Frontier
# Active
# BESMS is at ORNL
[docs]
class Besms(Supercomputer):
name = 'besms'
requires_account = True
batch_capable = True
# Using sinfo to get the queue configs
queue_configs={
'default': 't92',
't92': {
'max_nodes': 10,
'max_walltime': '672:00:00',
},
't92-burst': {
'max_nodes': 25,
'max_walltime': '672:00:00',
},
'burst': {
'max_nodes': 166,
'max_walltime': '48:00:00',
},
}
#end def write_job_header
#end class Baseline
# Decommissioned
# Summit at ORNL
[docs]
class Summit(Supercomputer):
name = 'summit'
requires_account = True
batch_capable = True
[docs]
def post_process_job(self,job):
# add the options only if the user has not supplied options
if len(job.run_options)==0:
opt = obj(
launch_dist = '-d packed',
bind = '-b rs',
)
if job.gpus is None:
job.gpus = 6 # gpus to use per node
#end if
if job.alloc_flags is None:
job.alloc_flags = 'smt1'
#end if
if job.gpus==0:
if job.processes%2==0:
resource_sets_per_node = 2
else:
resource_sets_per_node = 1
#end if
nrs = job.nodes*resource_sets_per_node
pprs = job.processes_per_node//resource_sets_per_node
gpurs = 0
else:
ppn = job.processes_per_node
if ppn is None:
self.warn('job may not run properly on Summit\nat least one mpi process should be present for each node\nplease check the generated bsub file for correctness')
ppn = 0
#end if
if ppn%job.gpus!=0:
self.warn('job may not run properly on Summit\nprocesses per node should divide evenly into number of gpus requested\nprocesses per node requested: {0}\ngpus per node requested: {1}\nplease check the generated bsub file for correctness'.format(job.processes_per_node,job.gpus))
#end if
resource_sets_per_node = job.gpus
nrs = job.nodes*resource_sets_per_node
pprs = ppn//resource_sets_per_node
gpurs = 1
#end if
opt.set(
resource_sets= '-n {0}'.format(nrs),
rs_per_node = '-r {0}'.format(resource_sets_per_node),
tasks_per_rs = '-a {0}'.format(pprs),
cpus_per_rs = '-c {0}'.format(pprs*job.threads),
gpus_per_rs = '-g {0}'.format(gpurs),
)
job.run_options.add(**opt)
#end if
#end def post_process_job
#end def write_job_header
[docs]
def read_process_id(self,output):
pid = None
tokens = output.split()
for t in tokens:
if t.startswith('<'):
spid = t.strip('<>').strip()
if spid.isdigit():
pid = int(spid)
break
#end if
#end if
#end for
return pid
#end def read_process_id
#end class Summit
# Unknown
## Added 28/11/2019 by A Zen
[docs]
class Rhea(Supercomputer):
name = 'rhea'
requires_account = True
batch_capable = True
#executable_subfile = True
prefixed_output = True
outfile_extension = '.output'
errfile_extension = '.error'
[docs]
def post_process_job(self,job):
job.run_options.add(
N='-N {}'.format(job.nodes),
n='-n {}'.format(job.processes),
)
if job.threads>1:
job.run_options.add(
c = '-c {}'.format(job.threads),
)
if 'cpu_bind' not in job.run_options:
if job.processes_per_node==self.cores_per_node:
cpu_bind = '--cpu-bind=threads'
else:
cpu_bind = '--cpu-bind=cores'
#end if
job.run_options.add(
cpu_bind = cpu_bind
)
#end if
#end if
#end def post_process_job
#end def write_job_header
#end class Rhea
# Active
# Leonardo BOOSTER at CINECA
## Added 24/11/2025 by A Zen
[docs]
class Leonardo(Supercomputer):
name = 'leonardo'
requires_account = True
batch_capable = True
prefixed_output = True
outfile_extension = '.out'
errfile_extension = '.err'
# QOS on Booster (boost_usr_prod)
# https://docs.hpc.cineca.it/hpc/leonardo.html#file-systems-and-data-managment
# parallel partition: boost_usr_prod
# GPUs: up to 4 gpus per node
booster_qos = {
'normal': {
'max_nodes' : 64,
'min_nodes' : 1,
'max_hours' : 24.0,
},
'boost_qos_dbg': {
'max_nodes' : 4,
'min_nodes' : 1,
'max_hours' : 0.5, # 30 minuti
},
'boost_qos_bprod': {
'min_nodes' : 65,
'max_nodes' : 256,
'max_hours' : 24.0,
},
'boost_qos_lprod': {
'min_nodes' : 1,
'max_nodes' : 8,
'max_hours' : 96.0, # 4 giorni
},
}
# serial partition: lrd_all_serial
# No GPUs, Hyperthreading x 2, Budget Free, 30800 MB RAM
serial_partition = 'lrd_all_serial'
serial_max_cores = 4 # 4 core max
serial_max_hours = 4.0 # 4 ore
[docs]
def post_process_job(self, job):
"""
Adjust srun options based on job properties (nodes, processes, threads).
This is called after the Job object is created and before writing the script.
"""
# Base srun options: total nodes and total MPI processes
job.run_options.add(
N='-N {}'.format(job.nodes),
n='-n {}'.format(job.processes),
)
# If OpenMP threads are requested, set -c and cpu binding
if job.threads > 1:
job.run_options.add(
c='-c {}'.format(job.threads),
)
# Only add cpu_bind if not already specified by the user
if 'cpu_bind' not in job.run_options:
if job.processes_per_node == self.cores_per_node:
# If all cores are used per node, bind by threads
cpu_bind = '--cpu-bind=threads'
else:
# Otherwise, bind by cores (more standard)
cpu_bind = '--cpu-bind=cores'
# end if
job.run_options.add(cpu_bind=cpu_bind)
# end if
# end if
# end def post_process_job
# end def write_job_header
# end class Leonardo
# Active
# Andes at ORNL
## Added 19/03/2021 by A Zen
[docs]
class Andes(Supercomputer):
name = 'andes'
requires_account = True
batch_capable = True
#executable_subfile = True
prefixed_output = True
outfile_extension = '.output'
errfile_extension = '.error'
[docs]
def post_process_job(self,job):
if job.threads>1:
job.run_options.add(
c = '-c {}'.format(job.threads),
)
if 'cpu_bind' not in job.run_options:
if job.processes_per_node==self.cores_per_node:
cpu_bind = '--cpu-bind=threads'
else:
cpu_bind = '--cpu-bind=cores'
#end if
job.run_options.add(
cpu_bind = cpu_bind
)
#end if
#end if
job.run_options.add(
N='-N {}'.format(job.nodes),
n='-n {}'.format(job.processes),
)
#end def post_process_job
[docs]
def write_job_header(self,job):
if job.queue is None:
job.queue='batch'
#end if
base_partition = None
max_partition = 384
if job.nodes <= 16:
max_time = 48
elif job.nodes <= 64:
max_time = 36
else:
max_time = 3
job.total_hours = job.days*24 + job.hours + job.minutes/60.0 + job.seconds/3600.0
if job.total_hours > max_time: # warn if job will take more than 96 hrs.
self.warn('!!! ATTENTION !!!\n the maximum runtime on {0} should not be more than {1}\n you requested: {2}'.format(job.queue,max_time,job.total_hours))
job.hours = max_time
job.minutes =0
job.seconds =0
#end if
c='#!/bin/bash\n'
c+='#SBATCH --job-name '+str(job.name)+'\n'
c+='#SBATCH --account='+str(job.account)+'\n'
c+='#SBATCH -N '+str(job.nodes)+'\n'
c+='#SBATCH -t {0}:{1}:{2}\n'.format(str(job.hours+24*job.days).zfill(2),str(job.minutes).zfill(2),str(job.seconds).zfill(2))
c+='#SBATCH -o {0}\n'.format(job.outfile)
c+='#SBATCH -e {0}\n'.format(job.errfile)
if job.email is not None:
c+='#SBATCH --mail-user {}\n'.format(job.email)
c+='#SBATCH --mail-type ALL\n'
#c+='#SBATCH --mail-type FAIL\n'
#end if
c+='\n'
c+='cd $SLURM_SUBMIT_DIR\n'
c+='\n'
c+='echo JobID : $SLURM_JOBID \n'
c+='echo Number of nodes requested: $SLURM_JOB_NUM_NODES \n'
c+='echo List of nodes assigned to the job: $SLURM_NODELIST \n'
c+='\n'
return c
#end def write_job_header
#end class Andes
# Active
## Added 05/04/2022 by A Zen
[docs]
class Archer2(Supercomputer):
# https://docs.archer2.ac.uk/user-guide/hardware/
name = 'archer2'
requires_account = True
batch_capable = True
#executable_subfile = True
prefixed_output = True
outfile_extension = '.output'
errfile_extension = '.error'
[docs]
def post_process_job(self,job):
job.run_options.add(
distribution='--distribution=block:block',
hint='--hint=nomultithread',
N='-N {}'.format(job.nodes),
n='-n {}'.format(job.processes),
)
if job.threads>1:
job.run_options.add(
c = '-c {}'.format(job.threads),
)
# if 'cpu_bind' not in job.run_options:
# if job.processes_per_node==self.cores_per_node:
# cpu_bind = '--cpu-bind=threads'
# else:
# cpu_bind = '--cpu-bind=cores'
# #end if
# job.run_options.add(
# cpu_bind = cpu_bind
# )
#end if
#end if
#end def post_process_job
#end def write_job_header
#end class Archer2
# Unknown
[docs]
class Tomcat3(Supercomputer):
name = 'tomcat3'
requires_account = False
batch_capable = True
redirect_output = True
#end def write_job_header
#end class Tomcat3
# Active
# Polaris at ANL
[docs]
class Polaris(Supercomputer):
name = 'polaris'
requires_account = True
batch_capable = True
special_bundling = True
[docs]
def post_process_job(self,job):
if len(job.run_options)==0:
opt = obj(
ppn = '--ppn {}'.format(job.processes_per_node),
depth = '--depth={}'.format(job.threads),
cpubind = '--cpu-bind depth',
threads = '--env OMP_NUM_THREADS={}'.format(job.threads),
)
job.run_options.add(**opt)
#end if
#end def post_process_job
#end def write_job_header
[docs]
def specialized_bundle_commands(self,job,launcher,serial):
c = ''
j0 = job.bundled_jobs[0]
c+='split --lines={} --numeric-suffixes=1 --suffix-length=3 $PBS_NODEFILE local_hostfile.\n'.format(j0.nodes)
c+='\n'
lhfiles = ['local_hostfile.'+str(n+1).zfill(3) for n in range(len(job.bundled_jobs))]
for j,lh in zip(job.bundled_jobs,lhfiles):
c+='cp {} {}\n'.format(lh,j.abs_subdir)
#end for
for j,lh in zip(job.bundled_jobs,lhfiles):
j.run_options.add(hostfile='--hostfile '+lh)
c+='\ncd '+j.abs_subdir+'\n'
c+=j.run_command(launcher,redirect=True,serial=serial)+'\n'
#end for
c+='\nwait\n'
return c
#end def specialized_bundle_commands
#end class Polaris
# Active
# Aurora at ANL
[docs]
class Aurora(Supercomputer):
name = 'aurora'
requires_account = True
batch_capable = True
special_bundling = True
[docs]
def pre_process_job(self,job):
# Set default queue and node type
if job.queue is None:
job.queue = 'prod'
#end if
if job.filesystems is None:
job.filesystems = 'flare'
#end if
if job.constraint is None:
job.constraint = 'cpu'
#end if
# Account for dual nature of Perlmutter
if 'cpu' in job.constraint:
self.procs_per_node = 2
self.cores_per_node = 102
self.ram_per_node = 1024
elif 'gpu' in job.constraint:
self.procs_per_node = 1
self.cores_per_node = 102
self.ram_per_node = 768
self.gpus_per_node = 6
else:
self.error('CPU or GPU constraint must be specified for Aurora')
#end if
#end def pre_process_job
[docs]
def post_process_job(self,job):
if len(job.run_options)==0:
if 'cpu' in job.constraint:
threads = '--env OMP_NUM_THREADS={} --env OMP_PLACES=cores'.format(job.threads)
cpubind = '--cpu-bind depth'
elif 'gpu' in job.constraint:
threads = '--env OMP_NUM_THREADS={}'.format(job.threads)
cpubind = '--cpu-bind=list'
ind = 1
for _ in range(job.processes_per_node):
cpubind += ':{}-{}'.format(ind, ind + job.threads - 1)
if ind + 2*(job.threads - 1) > 53 and ind < 53:
ind = 53
else:
ind += job.threads
# end for
else:
self.error('CPU or GPU constraint must be specified for Aurora')
#end if
opt = obj(
ppn = '--ppn {}'.format(job.processes_per_node),
depth = '--depth={}'.format(job.threads),
cpubind = cpubind,
threads = threads,
# affinity= '/soft/tools/mpi_wrapper_utils/gpu_tile_compact.sh',
)
job.run_options.add(**opt)
#end if
#end def post_process_job
#end def write_job_header
[docs]
def specialized_bundle_commands(self,job,launcher,serial):
c = ''
j0 = job.bundled_jobs[0]
c+='split --lines={} --numeric-suffixes=1 --suffix-length=3 $PBS_NODEFILE local_hostfile.\n'.format(j0.nodes)
c+='\n'
lhfiles = ['local_hostfile.'+str(n+1).zfill(3) for n in range(len(job.bundled_jobs))]
for j,lh in zip(job.bundled_jobs,lhfiles):
c+='cp {} {}\n'.format(lh,j.abs_subdir)
#end for
for j,lh in zip(job.bundled_jobs,lhfiles):
j.run_options.add(hostfile='--hostfile '+lh)
c+='\ncd '+j.abs_subdir+'\n'
c+=j.run_command(launcher,redirect=True,serial=serial)+'\n'
#end for
c+='\nwait\n'
return c
#end def specialized_bundle_commands
#end class Aurora
# Active
# Improv at ANL (LCRC)
[docs]
class Improv(Supercomputer):
name = 'improv'
requires_account = True
batch_capable = True
[docs]
def post_process_job(self,job):
if len(job.run_options)==0:
opt = obj(
mapby = '--map-by ppr:{}:package'.format(job.processes_per_proc),
bindto = '--bind-to socket',
)
job.run_options.add(**opt)
#end if
if job.threads>1 and job.env is None:
job.set_environment(
OMP_PLACES = 'cores',
OMP_PROC_BIND = 'close',
)
#end if
#end def post_process_job
#end def write_job_header
#end class Improv
# Active
# Kagayaki at JAIST
## Added 05/04/2023 by Tom Ichibha
[docs]
class Kagayaki(Supercomputer):
name = 'kagayaki'
requires_account = False
batch_capable = True
special_bundling = False
[docs]
def process_job_options(self,job):
# job.run_options.add(nodefile='-machinefile $PBS_NODEFILE', np='-np '+str(job.processes))
opt = obj(
nodefile='-machinefile $PBS_NODEFILE',
omp='-x OMP_NUM_THREADS',
np='-np {}'.format(job.processes),
)
job.run_options.add(**opt)
#end def write_job_header
#end class Kagayaki
# Active
# Kestrel at NREL
[docs]
class Kestrel(Supercomputer):
name = 'kestrel'
requires_account = True
batch_capable = True
#end def write_job_header
#end class Kestrel
# Active
# Lassen at LLNL
[docs]
class Lassen(Supercomputer):
name = 'lassen'
requires_account = True
batch_capable = True
[docs]
def post_process_job(self,job):
# add the options only if the user has not supplied options
if len(job.run_options)==0:
opt = obj(
nodes = '-N {}'.format(job.nodes),
tasks = '-T {}'.format(job.processes_per_node),
)
if job.threads>1:
opt.threads = '--threads={}'.format(job.threads)
#end if
if job.gpus is None:
job.gpus = 4# gpus to use per node
#end if
#if job.alloc_flags is None:
# job.alloc_flags = 'smt1'
##end if
if job.gpus==0:
None
else:
opt.mgpu = '-M "-gpu"'
#end if
job.run_options.add(**opt)
#end if
#end def post_process_job
#end def write_job_header
[docs]
def read_process_id(self,output):
pid = None
tokens = output.split()
for t in tokens:
if t.startswith('<'):
spid = t.strip('<>').strip()
if spid.isdigit():
pid = int(spid)
break
#end if
#end if
#end for
return pid
#end def read_process_id
#end class Lassen
# Ruby at LLNL
[docs]
class Ruby(Supercomputer):
name = 'ruby'
requires_account = True
batch_capable = True
#end def write_job_header
#end class Ruby
#Known machines
# workstations
for cores in range(1,128+1):
Workstation('ws'+str(cores),cores,'mpirun'),
#end for
# supercomputers and clusters
# nodes sockets cores ram qslots qlaunch qsubmit qstatus qdelete
Leonardo( 3456, 1, 32, 512, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Jaguar( 18688, 2, 8, 32, 100, 'aprun', 'qsub', 'qstat', 'qdel')
Kraken( 9408, 2, 6, 16, 100, 'aprun', 'qsub', 'qstat', 'qdel')
Golub( 512, 2, 6, 32, 1000, 'mpirun', 'qsub', 'qstat', 'qdel')
OIC5( 28, 2, 16, 128, 1000, 'mpirun', 'qsub', 'qstat', 'qdel')
Cori( 9688, 1, 68, 96, 100, 'srun', 'sbatch', 'squeue', 'scancel')
BlueWatersXK( 3072, 1, 16, 32, 100, 'aprun', 'qsub', 'qstat', 'qdel')
BlueWatersXE(22640, 2, 16, 64, 100, 'aprun', 'qsub', 'qstat', 'qdel')
Titan( 18688, 1, 16, 32, 100, 'aprun', 'qsub', 'qstat', 'qdel')
EOS( 744, 2, 8, 64, 1000, 'aprun', 'qsub', 'qstat', 'qdel')
Vesta( 2048, 1, 16, 16, 10, 'runjob', 'qsub', 'qstata', 'qdel')
Cetus( 1024, 1, 16, 16, 10, 'runjob', 'qsub', 'qstata', 'qdel')
Mira( 49152, 1, 16, 16, 10, 'runjob', 'qsub', 'qstata', 'qdel')
Cooley( 126, 2, 6, 384, 10, 'mpirun', 'qsub', 'qstata', 'qdel')
Theta( 4392, 1, 64, 192, 1000, 'aprun', 'qsub', 'qstata', 'qdel')
Lonestar( 22656, 2, 6, 12, 128, 'ibrun', 'qsub', 'qstat', 'qdel')
Matisse( 20, 2, 8, 64, 2, 'mpirun', 'sbatch', 'sacct', 'scancel')
Komodo( 24, 2, 6, 48, 2, 'mpirun', 'sbatch', 'sacct', 'scancel')
Amos( 5120, 1, 16, 16, 128, 'srun', 'sbatch', 'sacct', 'scancel')
Eclipse( 1488, 2, 18, 128, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Attaway( 1488, 2, 18, 192, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Manzano( 1488, 2, 24, 192, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Ghost( 740, 2, 18, 128, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Amber( 1496, 2, 56, 256, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Solo( 374, 2, 18, 128, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Flight( 744, 2, 56, 256, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Hops( 64, 2, 56, 256, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
SuperMUC( 512, 1, 28, 256, 8,'mpiexec', 'llsubmit', 'llq','llcancel')
Stampede2( 4200, 1, 68, 96, 50, 'ibrun', 'sbatch', 'squeue', 'scancel')
CadesMoab( 156, 2, 18, 128, 100, 'mpirun', 'qsub', 'qstat', 'qdel')
CadesSlurm( 156, 2, 18, 128, 100, 'mpirun', 'sbatch', 'squeue', 'scancel')
Summit( 4608, 2, 21, 512, 100, 'jsrun', 'bsub', 'bjobs', 'bkill')
Rhea( 512, 2, 8, 128, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Andes( 704, 2, 16, 256, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Tomcat3( 8, 1, 64, 192, 1000, 'mpirun', 'sbatch', 'sacct', 'scancel')
SuperMUC_NG( 6336, 1, 48, 96, 1000,'mpiexec', 'sbatch', 'sacct', 'scancel')
Archer2( 5860, 2, 64, 512, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Polaris( 560, 1, 32, 512, 8,'mpiexec', 'qsub', 'qstat', 'qdel')
Kagayaki( 240, 2, 64, 512, 20, 'mpirun', 'qsub', 'qstat', 'qdel')
Perlmutter( 3072, 2, 128, 512, 5000, 'srun', 'sbatch', 'squeue', 'scancel')
Improv( 825, 2, 64, 256, 100, 'mpirun', 'qsub', 'qstat', 'qdel')
Lassen( 756, 2, 21, 512, 100, 'lrun', 'bsub', 'bjobs', 'bkill')
Ruby( 1480, 2, 28, 192, 100, 'srun', 'sbatch', 'squeue', 'scancel')
Kestrel( 2144, 2, 52, 256, 100, 'srun', 'sbatch', 'squeue', 'scancel')
Inti( 13, 2, 64, 256, 100, 'srun', 'sbatch', 'squeue', 'scancel')
Baseline( 128, 2, 64, 512, 100, 'srun', 'sbatch', 'squeue', 'scancel')
Besms( 166, 1, 96, 768, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Frontier( 9856, 4, 14, 64, 1000, 'srun', 'sbatch', 'squeue', 'scancel')
Aurora( 10624, 2, 102, 512, 1000,'mpiexec', 'qsub', 'qstat', 'qdel')
#machine accessor functions
get_machine_name = Machine.get_hostname
get_machine = Machine.get
#rename Job with lowercase
job=Job
