Changeset 26435

Timestamp:

09/13/21 13:56:40 (4 years ago)

Author:

jdquinn

Message:

CHG: Updates to SMCE cluster class to properly download results; waitonlock now supports ID file; minor formatting

Location:

issm/trunk-jpl/src/m

Files:

• classes/clusters/smce_eis.py (modified) (10 diffs)
• classes/model.py (modified) (1 diff)
• solve/waitonlock.py (modified) (2 diffs)

issm/trunk-jpl/src/m/classes/clusters/smce_eis.py

@@ +1 @@
+import os
+import shutil
 import subprocess
 
@@ -8 +10 @@
 from helpers import *
 from IssmConfig import IssmConfig
-from issmscpin import issmscpin
-from issmscpout import issmscpout
 from issmssh import issmssh
 from MatlabFuncs import *
 from pairoptions import pairoptions
-from QueueRequirements import QueueRequirements
 
 class smce_eis(object):
@@ -25 +24 @@
 
     def __init__(self, *args):  # {{{
-        self.name = oshostname()
-        self.login = ''
+        self.name = '52.54.75.45'
+        self.login = 'jdquinn1'
+        self.idfile = '~/.ssh/smce-aws-eis-pilot'
         self.modules = []
-        self.numnodes = 20
-        self.cpuspernode = 8
+        self.numnodes = 8
+        self.cpuspernode = 1
         self.port = 0
-        self.queue = 'general'
         self.time = 12 * 60 * 60
-        self.processor = 'west'
+        self.processor = 'skylake'
         self.srcpath = '/shared/issm/issm/trunk-jpl'
         self.codepath = '/shared/issm/issm/trunk-jpl/bin'
-        self.executionpath = ''
-        self.grouplist = ''
+        self.executionpath = '/home/jdquinn1/issm-exec'
         self.interactive = 0
-        self.bbftp = 0
-        self.numstreams = 8
+        self.numstreams = 1
         self.hyperthreading = 0
         self.email = ''
@@ -62 +59 @@
         s += '    name: {}\n'.format(self.name)
         s += '    login: {}\n'.format(self.login)
+        s += '    idfile: {}\n'.format(self.idfile)
         s += '    modules: {}\n'.format(strjoin(self.modules, ', '))
         s += '    numnodes: {}\n'.format(self.numnodes)
@@ -67 +65 @@
         s += '    np: {}\n'.format(self.nprocs())
         s += '    port: {}\n'.format(self.port)
-        s += '    queue: {}\n'.format(self.queue)
         s += '    time: {}\n'.format(self.time)
         s += '    processor: {}\n'.format(self.processor)
@@ -73 +70 @@
         s += '    codepath: {}\n'.format(self.codepath)
         s += '    executionpath: {}\n'.format(self.executionpath)
-        s += '    grouplist: {}\n'.format(self.grouplist)
         s += '    interactive: {}\n'.format(self.interactive)
-        s += '    bbftp: {}\n'.format(self.bbftp)
         s += '    numstreams: {}\n'.format(self.numstreams)
         s += '    hyperthreading: {}\n'.format(self.hyperthreading)
@@ -86 +81 @@
 
     def checkconsistency(self, md, solution, analyses):  # {{{
-        queuedict = {'long': [24 * 60 * 60, 560],
-                     'allnccs': [12 * 60 * 60, 6000],
-                     'debug': [1 * 60 * 60, 532]}
-        QueueRequirements(queuedict, self.queue, self.nprocs(), self.time)
-
         # Now, check cluster.cpuspernode according to processor type
-        if self.processor == 'sand':
-            if self.cpuspernode > 16 or self.cpuspernode < 1:
-                md = md.checkmessage('cpuspernode should be between 1 and 16 for \'sand\' processors in hyperthreading mode')
-        elif self.processor == 'hasw':
-            if self.cpuspernode > 28 or self.cpuspernode < 1:
-                md = md.checkmessage('cpuspernode should be between 1 and 28 for \'hasw\' processors in hyperthreading mode')
+        if self.processor == 'skylake':
+            if self.cpuspernode > 14 or self.cpuspernode < 1:
+                md = md.checkmessage('cpuspernode should be between 1 and 14 for \'skyw\' processors in hyperthreading mode')
         else:
-            md = md.checkmessage('unknown processor type, should be \'sand\' or \'hasw\'')
+            md = md.checkmessage('unknown processor type, should be \'skylake\'')
 
         # Miscellaneous
         if not self.login:
             md = md.checkmessage('login empty')
+        if self.port:
+            md = md.checkmessage('port must be set to 0 as we do not have an SSH tunnel')
         if not self.codepath:
             md = md.checkmessage('codepath empty')
@@ -129 +118 @@
 
         fid.write('#!/bin/bash\n')
+        fid.write('#SBATCH --partition=hpc-spot \n')
         fid.write('#SBATCH -J {} \n'.format(modelname))
-        fid.write('#SBATCH --qos={} \n'.format(self.queue))
         fid.write('#SBATCH -o {}.outlog \n'.format(modelname))
         fid.write('#SBATCH -e {}.errlog \n'.format(modelname))
-        fid.write('#SBATCH -n {} \n'.format(self.nprocs()))
-        fid.write('#SBATCH -N {} \n'.format(self.numnodes))
+        fid.write('#SBATCH --nodes={} \n'.format(self.numnodes))
+        fid.write('#SBATCH --ntasks-per-node={} \n'.format(self.cpuspernode))
+        fid.write('#SBATCH --cpus-per-task={} \n'.format(self.numstreams))
         fid.write('#SBATCH -t {:02d}:{:02d}:00 \n'.format(int(floor(self.time / 3600)), int(floor(self.time % 3600) / 60)))
-        fid.write('#SBATCH -A {} \n\n'.format(self.grouplist))
         if (self.email.find('@')>-1):
             fid.write('#SBATCH --mail-user={} \n'.format(self.email))
-            fid.write('#SBATCH --mail-type=end \n\n')
-        # fid.write('. /usr/share/modules/init/bash\n\n')
+            fid.write('#SBATCH --mail-type=BEGIN,END,FAIL \n\n')
         # for i in range(len(self.modules)):
         #     fid.write('module load {}\n'.format(self.modules[i]))
         fid.write('export MPI_GROUP_MAX=64\n\n')
         fid.write('export MPI_UNBUFFERED_STDIO=true\n\n')
+        fid.write('export PATH="$PATH:/opt/slurm/bin"\n') # TODO: Figure out how to add this to PATH by sourcing environment script
         fid.write('export PATH="$PATH:."\n\n')
         fid.write('export ISSM_DIR="{}"\n'.format(self.srcpath)) # FIXME
+        fid.write('export HYDRA_HOST_FILE={}/{}/'.format(self.executionpath, dirname))
+        fid.write('${USER}-hydranodes-sj:${SLURM_JOB_ID}\n')
+        #fid.write('source /usr/share/modules/init/bash\n')
+        #fid.write('source /efs/spack-justin/share/spack/setup-env.sh\n')
+        #fid.write('module load gcc-9.3.0-gcc-9.3.0-ug2hqa3\n')
+        fid.write('touch ${HYDRA_HOST_FILE}\n')
+        fid.write('/efs/mrilee/snodelist/build/snodelist -m -f "%h%[:]c" >> ${HYDRA_HOST_FILE}\n')
         fid.write('source $ISSM_DIR/etc/environment.sh\n') # FIXME
         fid.write('cd {}/{}/\n\n'.format(self.executionpath, dirname))
@@ -187 +183 @@
             directory = self.executionpath
 
-        if self.bbftp:
-            issmbbftpout(self.name, directory, self.login, self.port, self.numstreams, '{}.tar.gz'.format(dirname))
-        else:
-            issmscpout(self.name, directory, self.login, self.port, ['{}.tar.gz'.format(dirname)])
+        # NOTE: Replacement for issmscpout(self.name, directory, self.login, self.port, ['{}.tar.gz'.format(dirname)])
+        copystring = 'cp {}.tar.gz /efs/issm-tmp/'.format(dirname, dirname)
+        subprocess.call(copystring, shell=True)
     # }}}
 
@@ -201 +196 @@
         else:
             if not isempty(restart):
-                launchcommand = 'cd {} && cd {} && sbatch {}.queue'.format(self.executionpath, dirname, modelname)
+                launchcommand = 'cd {} && cd {} && /opt/slurm/bin/sbatch {}.queue'.format(self.executionpath, dirname, modelname)
             else:
-                launchcommand = 'cd {} && rm -rf ./{} && mkdir {} && cd {} && mv ../{}.tar.gz ./ && tar -zxf {}.tar.gz && sbatch {}.queue'.format(self.executionpath, dirname, dirname, dirname, dirname, dirname, modelname)
+                launchcommand = 'cd {} && rm -rf ./{} && mkdir {} && cd {} && cp /efs/issm-tmp/{}.tar.gz ./ && tar -zxf {}.tar.gz && /opt/slurm/bin/sbatch {}.queue'.format(self.executionpath, dirname, dirname, dirname, dirname, dirname, modelname)
 
         print('launching solution sequence on remote cluster')
-        issmssh(self.name, self.login, self.port, launchcommand)
+
+        # NOTE: Replacement for issmssh(self.name, self.login, self.port, launchcommand)
+        subprocess.call('ssh -l {} -i {} {} "{}"'.format(self.login, self.idfile, self.name, launchcommand), shell=True)
     # }}}
 
     def Download(self, dirname, filelist):  # {{{
         # Copy files from cluster to current directory
-        if self.interactive:
-            directory = '{}/Interactive{}'.format(self.executionpath, self.interactive)
-        else:
-            directory = '{}/{}/'.format(self.executionpath, dirname)
-
-        if self.bbftp:
-            issmbbftpin(self.name, self.login, self.port, self.numstreams, directory, filelist)
-        else:
-            issmscpin(self.name, self.login, self.port, directory, filelist)
-    # }}}
+    
+        # NOTE: Replacement for issmscpin(self.name, self.login, self.port, directory, filelist)
+        directory = '{}/{}/'.format(self.executionpath, dirname)
+        fileliststr = "'{" + ','.join([str(x) for x in filelist]) + "}'"
+        downloadcommand = 'scp -T -i {} {}@{}:{} {}/.'.format(self.idfile, self.login, self.name, os.path.join(directory, fileliststr), os.getcwd())
+        subprocess.call(downloadcommand, shell=True) 
+    # }}}

issm/trunk-jpl/src/m/classes/model.py

@@ -169 +169 @@
         s = "%s\n%s" % (s, "%19s: %-22s -- %s" % ("steadystate", "[%s %s]" % ("1x1", obj.steadystate.__class__.__name__), "parameters for steadystate solution"))
         s = "%s\n%s" % (s, "%19s: %-22s -- %s" % ("transient", "[%s %s]" % ("1x1", obj.transient.__class__.__name__), "parameters for transient solution"))
-        s = "%s\n%s" % (s, "%19s: %-22s -- %s" % ("levelset", "[%s %s]" % ("1x1", obj.levelset.__class__.__name__), "parameters for moving boundaries (level - set method)"))
+        s = "%s\n%s" % (s, "%19s: %-22s -- %s" % ("levelset", "[%s %s]" % ("1x1", obj.levelset.__class__.__name__), "parameters for moving boundaries (level-set method)"))
         s = "%s\n%s" % (s, "%19s: %-22s -- %s" % ("calving", "[%s %s]" % ("1x1", obj.calving.__class__.__name__), "parameters for calving"))
         s = "%s\n%s" % (s, "%19s: %-22s -- %s" % ("frontalforcings", "[%s %s]" % ("1x1", obj.frontalforcings.__class__.__name__), "parameters for frontalforcings"))

issm/trunk-jpl/src/m/solve/waitonlock.py

@@ -53 +53 @@
     # Prepare command if the job is not running on the local machine
     if not strcmpi(oshostname(), cluster.name):
-        login = cluster.login
-        port = 0
-        if isprop(cluster, 'port'):
-            port = cluster.port
-        if port:
-            command = 'ssh -l {} -p {} localhost "[ -f {} ] && [ -f {} ]" 2>/dev/null'.format(login, port, lockfilename, logfilename)
-        elif cluster.name == 'cloud':
+        if cluster.name == 'cloud':
             command = '[ -f {} ] && [ -f {} ] 2>/dev/null'.format(lockfilename, logfilename)
             command = '{} sshmaster {} --user {} \'{}\''.format(starcluster(), cluster.name, cluster.login, command)
         else:
-            command = 'ssh -l {} {} "[ -f {} ] && [ -f {} ]" 2>/dev/null'.format(login, cluster.name, lockfilename, logfilename)
+            command = 'ssh -l {}'.format(cluster.login)
+            if isprop(cluster, 'idfile') and cluster.idfile != '':
+                command += ' -i {}'.format(cluster.idfile)
+            port = 0
+            if isprop(cluster, 'port'):
+                port = cluster.port
+            if port: # Check if port is non-zero
+                command += ' -p {} localhost'.format(port)
+            else:
+                command += ' {}'.format(cluster.name)
+            command += ' "[ -f {} ] && [ -f {} ]" 2>/dev/null'.format(lockfilename, logfilename)
 
     while not ispresent and elapsedtime < timelimit:
@@ -81 +85 @@
             # UPDATE: Works in testing under Debian Linux system. Leaving comment for now so that it is easier to backtrace this issue if someone else encounters it.
             #
-            if errs != '':
-                raise Exception('waitonlock: check for existence of files failed: {}'.format(errs))
+            #if errs != '':
+            #    raise Exception('waitonlock: check for existence of files failed: {}'.format(errs))
             ispresent = not subproc.returncode
             if ispresent: