Index: sm/trunk-jpl/src/m/classes/clusters/stallo.m =================================================================== --- /issm/trunk-jpl/src/m/classes/clusters/stallo.m (revision 26178) +++ (revision ) @@ -1,216 +1,0 @@ -% Stallo cluster class definition -% This is a SLURM queue -% The priorities are given to: -% - Large jobs -% - Short jobs -% - small number of job per user -% -% There are some 20cpu nodes and 16cpu nodes, with 32GB (a few with 128GB) mem per node, you can ask for part of a node if you need more memory.(1 node, 2 CPUS and 10GB per cpu for example) - -% Usage: -% cluster=stallo(); -% cluster=stallo('np',3); -% cluster=stallo('np',3,'login','username'); - -%node task and cpu must be defined, ask andreas for his stallo.py? - -classdef stallo - properties (SetAccess=public) - % {{{ - name = 'stallo'; - login = ''; - numnodes = 2; - cpuspernode = 10; - mem = 1.5; - queue = 'normal'; - time = 2*60; - codepath = ''; - executionpath = ''; - interactive = 0; - port = []; - accountname = ''; - profiling = 0; - bbftp = 0; - % }}} - end - methods - function cluster=stallo(varargin) % {{{ - - %initialize cluster using default settings if provided - if (exist('stallo_settings')==2), stallo_settings; end - - %use provided options to change fields - cluster=AssignObjectFields(pairoptions(varargin{:}),cluster); - end - %}}} - function disp(cluster) % {{{ - % display the object - disp(sprintf('class ''%s'' object ''%s'' = ',class(cluster),inputname(1))); - disp(sprintf(' name: %s',cluster.name)); - disp(sprintf(' login: %s',cluster.login)); - disp(sprintf(' accountname: %s',cluster.accountname)); - disp(sprintf(' numnodes: %i',cluster.numnodes)); - disp(sprintf(' cpuspernode: %i, cpu per nodes',cluster.cpuspernode)); - disp(sprintf(' queue: %s, name of the queue (normal (D), short,singlenode,multinode,devel)',cluster.queue)); - disp(sprintf(' codepath: %s',cluster.codepath)); - disp(sprintf(' executionpath: %s',cluster.executionpath)); - disp(sprintf(' interactive: %i',cluster.interactive)); - disp(sprintf(' time: %i, walltime requested in minutes',cluster.time)); - disp(sprintf(' memory: %i, memory per CPU',cluster.mem)); - disp(sprintf(' profiling: %i, enable profiling if 1 default is 0',cluster.profiling)); - end - %}}} - function md = checkconsistency(cluster,md,solution,analyses) % {{{ - - available_queues={'short','normal','singlenode','multinode','devel'}; - queue_requirements_time=[60 2*24*60 28*24*60 28*24*60 4*60]; - queue_requirements_np=[2048 2048 20 2048 2048]; - - QueueRequirements(available_queues,queue_requirements_time,queue_requirements_np,cluster.queue,cluster.np,1) - - %Miscelaneous - if isempty(cluster.login), md = checkmessage(md,'login empty'); end - if isempty(cluster.accountname), md = checkmessage(md,'accountname empty'); end - if isempty(cluster.codepath), md = checkmessage(md,'codepath empty'); end - if isempty(cluster.executionpath), md = checkmessage(md,'executionpath empty'); end - - end - %}}} - function numprocs=np(cluster) % {{{ - %compute number of processors - numprocs=cluster.numnodes*cluster.cpuspernode; - end - %}}} - function BuildQueueScript(cluster,dirname,modelname,solution,io_gather,isvalgrind,isgprof,isdakota,isoceancoupling) % {{{ - - if(isvalgrind), disp('valgrind not supported by cluster, ignoring...'); end - if(isgprof), disp('gprof not supported by cluster, ignoring...'); end - - executable='issm.exe'; - if isdakota, - version=IssmConfig('_DAKOTA_VERSION_'); version=str2num(version(1:3)); - if (version>=6), - executable='issm_dakota.exe'; - end - end - if isoceancoupling, - executable='issm_ocean.exe'; - end - - days=floor(cluster.time/(60*24)); - hours=floor((cluster.time-(days*60*24))/60); - minutes=floor(mod((cluster.time-(days*60*24)-(hours*60)),60)); - - %write queuing script - fid=fopen([modelname '.queue'],'w'); - fprintf(fid,'#!/bin/bash -l\n'); - fprintf(fid,'#SBATCH --job-name=%s \n',modelname); - fprintf(fid,'#SBATCH --qos=%s \n',cluster.queue); - fprintf(fid,'#SBATCH --nodes=%i \n',cluster.numnodes); - fprintf(fid,'#SBATCH --ntasks-per-node=%i \n',cluster.cpuspernode); - fprintf(fid,'#SBATCH --time=%02i-%02i:%02i:00 \n',days,hours,minutes); - fprintf(fid,'#SBATCH --mem-per-cpu=%iMB\n',1000*cluster.mem); % mem in MB - if (mod(cluster.np,16)+mod(cluster.np,20))==0, - fprintf(fid,'#SBATCH --ntask=%i\n',cluster.np); - end - fprintf(fid,'#SBATCH --account=%s\n',cluster.accountname); - fprintf(fid,'#SBATCH -o %s.outlog \n',[cluster.executionpath '/' dirname '/' modelname]); - fprintf(fid,'#SBATCH -e %s.errlog \n\n',[cluster.executionpath '/' dirname '/' modelname]); - - fprintf(fid,'export ISSM_DIR="%s/../"\n',cluster.codepath);%FIXME - fprintf(fid,'module purge\n'); - fprintf(fid,'module load CMake/3.8.0-GCCcore-6.3.0\n'); - fprintf(fid,'module load Automake/1.15.1-GCCcore-6.3.0\n'); - fprintf(fid,'module load libtool/2.4.6\n'); - fprintf(fid,'module load OpenSSL/1.1.0e-intel-2017a\n'); - fprintf(fid,'module load PETSc/3.7.5-intel-2017a-downloaded-deps\n'); - - fprintf(fid,'cd %s/%s/\n\n',cluster.executionpath,dirname); - if cluster.profiling==1, - fprintf(fid,'module load perf-report\n'); - fprintf(fid,'perf-report mpirun -np %i %s/%s %s %s/%s %s\n',cluster.np,cluster.codepath,executable,solution,cluster.executionpath,dirname,modelname); - else - fprintf(fid,'mpirun -np %i %s/%s %s %s/%s %s\n',cluster.np,cluster.codepath,executable,solution,cluster.executionpath,dirname,modelname); - end - if ~io_gather, %concatenate the output files: - fprintf(fid,'cat %s.outbin.* > %s.outbin',modelname,modelname); - end - fclose(fid); - - %in interactive mode, create a run file, and errlog and outlog file - if cluster.interactive, - fid=fopen([modelname '.run'],'w'); - fprintf(fid,'mpiexec -np %i %s/issm.exe %s %s %s\n',cluster.np,cluster.codepath,solution,[cluster.executionpath '/' dirname],modelname); - if ~io_gather, %concatenate the output files: - fprintf(fid,'cat %s.outbin.* > %s.outbin',modelname,modelname); - end - fclose(fid); - fid=fopen([modelname '.errlog'],'w'); - fclose(fid); - fid=fopen([modelname '.outlog'],'w'); - fclose(fid); - end - end - %}}} - function UploadQueueJob(cluster,modelname,dirname,filelist)% {{{ - - %compress the files into one zip. - compressstring=['tar -zcf ' dirname '.tar.gz ']; - for i=1:numel(filelist), - compressstring = [compressstring ' ' filelist{i}]; - end - if cluster.interactive, - compressstring = [compressstring ' ' modelname '.run ' modelname '.errlog ' modelname '.outlog ']; - end - system(compressstring); - disp('uploading input file and queueing script'); - if cluster.interactive==10, - directory=[pwd() '/run/']; - elseif cluster.interactive, - directory=[cluster.executionpath '/Interactive' num2str(cluster.interactive)]; - else - directory=cluster.executionpath; - end - - if ~cluster.bbftp, - issmscpout(cluster.name,directory,cluster.login,cluster.port,{[dirname '.tar.gz']}); - else - issmbbftpout(cluster.name,directory,cluster.login,cluster.port,cluster.numstreams,{[dirname '.tar.gz']}); - - end - end %}}} - function LaunchQueueJob(cluster,modelname,dirname,filelist,restart,batch)% {{{ - - %lauch command, to be executed via ssh - if ~cluster.interactive, - if ~isempty(restart) - launchcommand=['cd ' cluster.executionpath ' && cd ' dirname ' && sbatch ' modelname '.queue ']; - else - launchcommand=['cd ' cluster.executionpath ' && rm -rf ./' dirname ' && mkdir ' dirname ... - ' && cd ' dirname ' && mv ../' dirname '.tar.gz ./ && tar -zxf ' dirname '.tar.gz && sbatch ' modelname '.queue ']; - end - else - if ~isempty(restart) - launchcommand=['cd ' cluster.executionpath '/Interactive' num2str(cluster.interactive)]; - else - if cluster.interactive==10, - launchcommand=['cd ' pwd() '/run && tar -zxf ' dirname '.tar.gz']; - else - launchcommand=['cd ' cluster.executionpath '/Interactive' num2str(cluster.interactive) ' && tar -zxf ' dirname '.tar.gz']; - end - end - end - - disp('launching solution sequence on remote cluster'); - issmssh(cluster.name,cluster.login,cluster.port,launchcommand); - end - %}}} - function Download(cluster,dirname,filelist)% {{{ - - %copy files from cluster to current directory - directory=[cluster.executionpath '/' dirname '/']; - issmscpin(cluster.name,cluster.login,cluster.port,directory,filelist); - - end %}}} - end -end Index: sm/trunk-jpl/src/m/classes/clusters/stallo.py =================================================================== --- /issm/trunk-jpl/src/m/classes/clusters/stallo.py (revision 26178) +++ (revision ) @@ -1,179 +1,0 @@ -import subprocess -import numpy as np -from fielddisplay import fielddisplay -from pairoptions import pairoptions -from issmssh import issmssh -from issmscpin import issmscpin -from issmscpout import issmscpout -from QueueRequirements import QueueRequirements -from IssmConfig import IssmConfig -import datetime -try: - from stallo_settings import stallo_settings -except ImportError: - print('You need stallo_settings.py to proceed, check presence and sys.path') - - -class stallo(object): - """ - Stallo cluster class definition - This is a SLURM queue - The priorities are given to: - - Large jobs - - Short jobs - - small number of job per user - - There are some 20cpu nodes and 16cpu nodes, with 32GB (a few with 128GB) mem per node, you can ask for part of a node if you need more memory.(1 node, 2 CPUS and 10GB per cpu for example) - - - Usage: - cluster = stallo() - """ - - def __init__(self, *args): # {{{ - self.name = 'stallo' - self.login = '' - self.numnodes = 2 - self.cpuspernode = 20 - self.mem = 1.6 - self.queue = 'normal' - self.time = 2 * 60 - self.codepath = '' - self.executionpath = '' - self.interactive = 0 - self.port = [] - self.accountname = '' - self.profiling = 0 - self.exclusive = False - #use provided options to change fields - options = pairoptions(*args) - #initialize cluster using user settings if provided - self = stallo_settings(self) - #OK get other fields - self = options.AssignObjectFields(self) - self.np = self.numnodes * self.cpuspernode - # }}} - - def __repr__(self): # {{{ - # display the object - s = "class vilje object:" - s = "%s\n%s" % (s, fielddisplay(self, 'name', 'name of the cluster')) - s = "%s\n%s" % (s, fielddisplay(self, 'login', 'login')) - s = "%s\n%s" % (s, fielddisplay(self, 'numnodes', 'number of nodes')) - s = "%s\n%s" % (s, fielddisplay(self, 'cpuspernode', 'number of CPUs per nodes')) - s = "%s\n%s" % (s, fielddisplay(self, 'mem', 'memory per CPU')) - s = "%s\n%s" % (s, fielddisplay(self, 'queue', 'name of the queue (normal (D), short, singlenode, multinode, devel)')) - s = "%s\n%s" % (s, fielddisplay(self, 'time', 'walltime requested in minutes')) - s = "%s\n%s" % (s, fielddisplay(self, 'codepath', 'code path on the cluster')) - s = "%s\n%s" % (s, fielddisplay(self, 'executionpath', 'execution path on the cluster')) - s = "%s\n%s" % (s, fielddisplay(self, 'interactive', '')) - s = "%s\n%s" % (s, fielddisplay(self, 'accountname', 'your cluster account')) - s = "%s\n%s" % (s, fielddisplay(self, 'profiling', 'enable profiling if 1 default is 0')) - return s - # }}} - - def checkconsistency(self, md, solution, analyses): # {{{ - #Queue dictionarry gives queue name as key and max walltime and cpus as var - queuedict = {'short': [2 * 24 * 60, 2048], - 'normal': [5 * 24 * 60, 2048], - 'singlenode': [28 * 24 * 60, 20], - 'multinode': [28 * 24 * 60, 2048], - 'devel': [4 * 60, 2048]} - QueueRequirements(queuedict, self.queue, self.np, self.time) - - #Miscelaneous - if not self.login: - md = md.checkmessage('login empty') - if not self.codepath: - md = md.checkmessage('codepath empty') - if not self.executionpath: - md = md.checkmessage('executionpath empty') - if self.interactive == 1: - md = md.checkmessage('interactive mode not implemented') - return self - # }}} - - def BuildQueueScript(self, dirname, modelname, solution, io_gather, isvalgrind, isgprof, isdakota, isoceancoupling): # {{{ - - executable = 'issm.exe' - if isdakota: - version = IssmConfig('_DAKOTA_VERSION_')[0:2] - version = float(version) - if version >= 6: - executable = 'issm_dakota.exe' - if isoceancoupling: - executable = 'issm_ocean.exe' - #write queuing script - shortname = modelname[0:min(12, len(modelname))] - timeobj = datetime.timedelta(minutes=self.time) - m, s = divmod(timeobj.total_seconds(), 60) - h, m = divmod(m, 60) - d, h = divmod(h, 24) - timestring = "%02d-%02d:%02d:%02d" % (d, h, m, s) - print('timestring') - fid = open(modelname + '.queue', 'w') - fid.write('#!/bin/bash -l\n') - fid.write('#SBATCH --job-name=%s \n' % shortname) - if self.queue in ['devel', 'short']: - fid.write('#SBATCH --qos=%s \n' % self.queue) - else: - fid.write('#SBATCH --partition=%s \n' % self.queue) - if self.time < 2 * 24 * 60: - fid.write('#SBATCH --qos=short \n') - fid.write('#SBATCH --nodes=%i \n' % self.numnodes) - fid.write('#SBATCH --ntasks-per-node=%i \n' % self.cpuspernode) - fid.write('#SBATCH --time={}\n'.format(timestring)) #walltime is minutes - fid.write('#SBATCH --mem-per-cpu={}MB\n'.format(int(1000 * self.mem))) # mem is in MB - if (np.mod(self.np, 16) + np.mod(self.np, 20)) == 0: - fid.write(' #SBATCH --ntask=%i\n' % self.np) - fid.write('#SBATCH --account=%s\n' % self.accountname) - fid.write('#SBATCH --output %s/%s/%s.outlog \n' % (self.executionpath, dirname, modelname)) - fid.write('#SBATCH --error %s/%s/%s.errlog \n\n' % (self.executionpath, dirname, modelname)) - if self.exclusive: - fid.write('#SBATCH --exclusive \n') - - fid.write('export ISSM_DIR="%s/../"\n' % self.codepath) - fid.write('module purge\n') - fid.write('module load CMake/3.8.0-GCCcore-6.3.0\n') - fid.write('module load Automake/1.15.1-GCCcore-6.3.0\n') - fid.write('module load libtool/2.4.6\n') - fid.write('module load OpenSSL/1.1.0e-intel-2017a\n') - fid.write('module load PETSc/3.7.5-intel-2017a-downloaded-deps\n') - - fid.write('cd %s/%s/ \n\n' % (self.executionpath, dirname)) - if self.profiling == 1: - fid.write('module load perf-report\n') - fid.write('perf-report mpirun -np %i %s/%s %s %s/%s %s\n' % (self.np, self.codepath, executable, str(solution), self.executionpath, dirname, modelname)) - else: - fid.write('mpirun -np %i %s/%s %s %s/%s %s\n' % (self.np, self.codepath, executable, str(solution), self.executionpath, dirname, modelname)) - fid.close() - - # }}} - def UploadQueueJob(self, modelname, dirname, filelist): # {{{ - #compress the files into one zip. - compressstring = 'tar -zcf %s.tar.gz ' % dirname - for file in filelist: - compressstring += ' %s' % file - subprocess.call(compressstring, shell=True) - - print('uploading input file and queueing script') - issmscpout(self.name, self.executionpath, self.login, self.port, [dirname + '.tar.gz']) - - # }}} - - def LaunchQueueJob(self, modelname, dirname, filelist, restart, batch): # {{{ - - print('launching solution sequence on remote cluster') - if restart: - launchcommand = 'cd %s && cd %s && sbatch %s.queue' % (self.executionpath, dirname, modelname) - else: - launchcommand = 'cd %s && rm -rf ./%s && mkdir %s && cd %s && mv ../%s.tar.gz ./ && tar -zxf %s.tar.gz && sbatch %s.queue' % (self.executionpath, dirname, dirname, dirname, dirname, dirname, modelname) - issmssh(self.name, self.login, self.port, launchcommand) - - # }}} - - def Download(self, dirname, filelist): # {{{ - #copy files from cluster to current directory - directory = '%s/%s/' % (self.executionpath, dirname) - issmscpin(self.name, self.login, self.port, directory, filelist) - # }}}