#!/usr/bin/env python # runShell.py - start a shell over XCPU3# Copyright 2010 Pravin Shinde# def showUsage () : print """USAGE: runShell.py [-l hare_location] [-9 py9p_location] [-h brasil_host] [-p brasil_port] [-n numnodes] [-o os_type] [-a architecture_type] [-s shell_name] ARGUMENTS: -l: hare_location default value ["/bgsys/argonne-utils/profiles/plan9/LATEST/bin/"] -9: py9p_loation (specify only if py9p is installed separately) -D: debug mode -h: host running brasil daemon (defaults to current host) -p: port running brasil daemon (default: 5670) -n: number of cpu nodes to provision (mandetory) -o: OS type (default value "any") (supported values: Linux, Plan9, MacOSX, Nt) -a: arch type (default value "any") (supported values: 386, arm, power, spim) -s: shell name name of the shell to run (NOTE: this option *MUST* be provided) ENVIRONMENT VARIABLES: (can be used to override defaults in place of cmdline) HARE_LOCATION - Place where hare executables are installed BRASIL_HOST - host running brasil daemon BRASIL_PORT - port running brasil daemon EXAMPLES runShell.py -n 4 -s bash runSell.py -l /path/to/hare/binaries/ -n 8 -o Linux -a 386 -s bash runJob.py -9 /path/to/py9p/ -n 3 -o Plan9 -s rc NOTES To close the session, you should send command that will terminate the shell, For example, in case of bash, user should type "exit" and then to terminate the input, press "CTRL+D" which will close the input file.""" import socketimport sysimport osimport getoptimport getpassimport codeimport readlineimport atexitimport threadingimport thread # Find the location of py9pDEFAULTPY9P = "/bgsys/argonne-utils/profiles/plan9/LATEST/bin/"import getoptpy9pPath = DEFAULTPY9P + "/py9p/"pathSource = "Default Path [" + DEFAULTPY9P + "] " if 'HARE_PATH' in os.environ: py9pPath = os.environ['HARE_PATH'] + "/py9p/" pathSource = "Environment Variable HARE_PATH [" + os.environ['HARE_PATH'] + "] " try: opts, args = getopt.getopt (sys.argv[1:],"Dl:9:h:p:n:o:a:s:") for o,a in opts: if o == "-l" : py9pPath = a + "/py9p/" pathSource = "cmdline arg -l [" + a + "] " if o == "-9" : py9pPath = a pathSource = "cmdline arg -9 [" + a + "] "except getopt.GetoptError: pass if not os.path.exists(py9pPath+"/py9p/py9p.py"): print "ERROR: Could not find py9p in the location [" \ + py9pPath + "] specified by : " + pathSource print """ You can do following to solve this problem. 1. use -l to specify the path to brasil binary executables. 2. use -9 to specify the path to py9p if it is not inside brasil binaries. 3. Set environment variable HARE_PATH pointing to brasil binary executables 4. Copy the brasil binary executables in default location : """ + DEFAULTPY9P showUsage () sys.exit() #print "Using py9p location [" + py9pPath +"]"sys.path.append (py9pPath + '/py9p')try: import py9pexcept : print "ERROR: Could not import py9p" print "Please provide path to py9p or hare installation " \ + "from environment variable or command line parameter" sys.exit() remoteURL = ""#baseURL = "/cmd2/local/"#baseURL = "/task/remote/"baseURL = "/csrv/local/"#baseURL = "/local/" def catAsync (obj, *args): debug = False if debug : print "thread started" while 1: if debug : print "thread reading" buf = obj.read(512) #buf = obj.readline() if debug : print "thread read " + buf if len(buf) <= 0: break sys.stdout.write (buf) if debug : print "thread dead" class XCPU3Client: sessionID = None DEBUG = False bufsize = 512 def __init__(self, srv, port, authmode='none', user='', passwd=None, authsrv=None, chatty=0, key=None): self.sessionID = None self.bufSize = 512 self.DEBUG = False self.extraLink = self.get9pClient(srv, port, authmode, user, passwd, authsrv, chatty, key) self.inputLink = self.get9pClient(srv, port, authmode, user, passwd, authsrv, chatty, key) self.ctlLink = self.get9pClient(srv, port, authmode, user, passwd, authsrv, chatty, key) self.outputLink = self.get9pClient(srv, port, authmode, user, passwd, authsrv, chatty, key) self.debugLink = self.get9pClient(srv, port, authmode, user, passwd, authsrv, chatty, key) def dPrint (self, msg): if (self.DEBUG): print msg def enableDebug (self) : name = "/csrv/local/ctl" print "enabling debugging" if self.debugLink.open (name, py9p.ORDWR) is None: raise Exception ("XCPU3: could not open " + name) self.debugLink.write ("debug 1"); self.debugLink.close (); def disableDebug (self) : name = "/csrv/local/ctl" print "disabling debugging" if self.debugLink.open (name, py9p.ORDWR) is None: raise Exception ("XCPU3: could not open " + name) self.debugLink.write ("debug 0"); self.debugLink.close (); def get9pClient (self, srv, port, authmode='none', user='', passwd=None, authsrv=None, chatty=0, key=None) : privkey = None sock = socket.socket(socket.AF_INET) try: sock.connect((srv, port),) except socket.error,e: print "%s: %s" % (srv, e.args[1]) raise return py9p.Client(py9p.Sock(sock, 0, chatty), authmode, user, passwd, authsrv, chatty, key=privkey) def cat(self, name, obj=None, out=None): if out is None: out = sys.stdout if obj is None: obj = self.extraLink if obj.open(name) is None: raise Exception("XCPU3: Could not open " + name) while 1: buf = obj.read(self.bufSize) if len(buf) <= 0: break out.write(buf) obj.close() def topStat (self, out=None): name = baseURL + "status" self.cat(name, None, out) def getSessionStatus (self, out = None) : if self.sessionID is None : print "Error: Session is not started yet" name = baseURL + str(self.sessionID) + "/status" self.cat (name, self.extraLink, out) def startSession (self): if self.sessionID is not None : print "Session already running" return self.sessionID name = baseURL + "clone" if self.ctlLink.open (name, py9p.ORDWR) is None : raise Exception("XCPU3: Could not open " + name) buf = self.ctlLink.read(self.bufSize) self.sessionID = int (buf) return self.sessionID def requestReservation (self, res) : if self.sessionID is None : self.startSession() if res is None : return param = "res " + res self.ctlLink.write(param) def requestExecution (self, cmd) : if self.sessionID is None : self.startSession() param = "exec " + cmd self.ctlLink.write(param) name = baseURL + str(self.sessionID) + "/stdio" if self.outputLink.open (name, py9p.OREAD ) is None : raise Exception("XCPU3: Could not open " + name) if self.inputLink.open(name, py9p.OWRITE) is None: raise Exception("XCPU3: Could not open stdio file " + name) def sendInput (self, input): if self.sessionID is None : raise Exception("XCPU3: Session not created") if input is None : return if input == '-' : inf = sys.stdin self.dPrint ("Using stdin for input") else : inf = open(input, "r", 0) sz = self.bufSize while 1: # buf = inf.read(sz) buf = inf.readline() if len(buf) <= 0: break self.dPrint ("sending line [" + buf + "]") self.inputLink.write(buf) self.dPrint ("input sent") inf.close() def getOutputAsync (self) : if self.sessionID is None : raise Exception("XCPU3: Session not created") try: threading.Thread(target=catAsync, args=(self.outputLink, 1)).start() except Exception, errtxt: print errtxt raise Exception return self.dPrint ("Thread started ...") def getOutput (self) : if self.sessionID is None : raise Exception("XCPU3: Session not created") while 1: buf = self.outputLink.read(self.bufSize) if len(buf) <= 0: break sys.stdout.write(buf) def endSession (self) : if self.sessionID is None : self.dPrint ( "Session already close") return self.inputLink.close() self.ctlLink.close() self.outputLink.close() self.sessionID = None def runJob (self, cmd, res = None, input = None ): self.dPrint ("Requesting reservation..") self.requestReservation(res) self.dPrint ( "Requesting execution..") self.requestExecution(cmd) self.dPrint ( "getting output") self.getOutputAsync() self.dPrint ( "sending input") self.sendInput ("-")# self.getOutput() if self.DEBUG: self.dPrint ( "checking session status") self.getSessionStatus () self.dPrint ( "Closing session") self.endSession () self.dPrint ( "Done..") def main (): # Filling default values to all the variables which are not provided with brasilHost = "localhost" brasilPort = 5670 nflag = None osType = None archType = None debug = False shellType = None # Overriding the values from environment variables if 'BRASIL_HOST' in os.environ: brasilHost = os.environ['BRASIL_HOST'] if 'BRASIL_PORT' in os.environ: brasilPort = int (os.environ['BRASIL_PORT']) try: opts, args = getopt.getopt (sys.argv[1:],"Dl:9:h:p:n:o:a:s:") for o,a in opts: if o == "-h" : brasilHost = a elif o == "-p" : brasilPort = int (a) elif o == "-n" : nflag = a elif o == "-o" : osType = a elif o == "-a" : archType = a elif o == "-s" : shellType = a elif o == "-D" : debug = True elif ( o in ( "-9", "-l" ) ) : a = 1 + 1 else : print "ERROR: Wrong option provided " + o showUsage () sys.exit () argLen = len (args) if nflag == None : print "ERROR: Number of CPU's needed is not provided" showUsage () sys.exit () if (int(nflag) < 0 ) : print "ERROR: Number of CPU's provided are invalid" showUsage () sys.exit () resReq = nflag if osType not in (None, "any", "Any", "ANY", "*" ) : resReq = resReq + " " + osType if archType not in (None, "any", "Any", "ANY", "*" ) : resReq = resReq + " " + archType if shellType == None : print "ERROR: shell name not provided" showUsage () sys.exit () except getopt.GetoptError: print "exception" showUsage () sys.exit() # acutal code responsible for working mycpu = XCPU3Client (brasilHost, brasilPort) mycpu.DEBUG = debug if mycpu.DEBUG : mycpu.enableDebug() print "Top statistics is" mycpu.topStat() mycpu.runJob (shellType, resReq ) if mycpu.DEBUG : mycpu.disableDebug() if __name__ == "__main__" : main ()
#!/usr/bin/env python
# runShell.py - start a shell over XCPU3
# Copyright 2010 Pravin Shinde
#
def showUsage () :
print """
USAGE: runShell.py [-l hare_location] [-9 py9p_location]
[-h brasil_host] [-p brasil_port]
[-n numnodes] [-o os_type] [-a architecture_type]
[-s shell_name]
ARGUMENTS:
-l: hare_location default value ["/bgsys/argonne-utils/profiles/plan9/LATEST/bin/"]
-9: py9p_loation (specify only if py9p is installed separately)
-D: debug mode
-h: host running brasil daemon (defaults to current host)
-p: port running brasil daemon (default: 5670)
-n: number of cpu nodes to provision (mandetory)
-o: OS type (default value "any") (supported values: Linux, Plan9, MacOSX, Nt)
-a: arch type (default value "any") (supported values: 386, arm, power, spim)
-s: shell name name of the shell to run (NOTE: this option *MUST* be provided)
ENVIRONMENT VARIABLES: (can be used to override defaults in place of cmdline)
HARE_LOCATION - Place where hare executables are installed
BRASIL_HOST - host running brasil daemon
BRASIL_PORT - port running brasil daemon
EXAMPLES
runShell.py -n 4 -s bash
runSell.py -l /path/to/hare/binaries/ -n 8 -o Linux -a 386 -s bash
runJob.py -9 /path/to/py9p/ -n 3 -o Plan9 -s rc
NOTES
To close the session, you should send command that will terminate the shell,
For example, in case of bash, user should type "exit" and then to terminate
the input, press "CTRL+D" which will close the input file.
"""
import socket
import sys
import os
import getopt
import getpass
import code
import readline
import atexit
import threading
import thread
# Find the location of py9p
DEFAULTPY9P = "/bgsys/argonne-utils/profiles/plan9/LATEST/bin/"
import getopt
py9pPath = DEFAULTPY9P + "/py9p/"
pathSource = "Default Path [" + DEFAULTPY9P + "] "
if 'HARE_PATH' in os.environ:
py9pPath = os.environ['HARE_PATH'] + "/py9p/"
pathSource = "Environment Variable HARE_PATH [" + os.environ['HARE_PATH'] + "] "
try:
opts, args = getopt.getopt (sys.argv[1:],"Dl:9:h:p:n:o:a:s:")
for o,a in opts:
if o == "-l" :
py9pPath = a + "/py9p/"
pathSource = "cmdline arg -l [" + a + "] "
if o == "-9" :
py9pPath = a
pathSource = "cmdline arg -9 [" + a + "] "
except getopt.GetoptError:
pass
if not os.path.exists(py9pPath+"/py9p/py9p.py"):
print "ERROR: Could not find py9p in the location [" \
+ py9pPath + "] specified by : " + pathSource
print """
You can do following to solve this problem.
1. use -l to specify the path to brasil binary executables.
2. use -9 to specify the path to py9p if it is not inside brasil binaries.
3. Set environment variable HARE_PATH pointing to brasil binary executables
4. Copy the brasil binary executables in default location : """ + DEFAULTPY9P
showUsage ()
sys.exit()
#print "Using py9p location [" + py9pPath +"]"
sys.path.append (py9pPath + '/py9p')
try:
import py9p
except :
print "ERROR: Could not import py9p"
print "Please provide path to py9p or hare installation " \
+ "from environment variable or command line parameter"
sys.exit()
remoteURL = ""
#baseURL = "/cmd2/local/"
#baseURL = "/task/remote/"
baseURL = "/csrv/local/"
#baseURL = "/local/"
def catAsync (obj, *args):
debug = False
if debug : print "thread started"
while 1:
if debug : print "thread reading"
buf = obj.read(512)
#buf = obj.readline()
if debug : print "thread read " + buf
if len(buf) <= 0:
break
sys.stdout.write (buf)
if debug : print "thread dead"
class XCPU3Client:
sessionID = None
DEBUG = False
bufsize = 512
def __init__(self, srv, port, authmode='none', user='', passwd=None, authsrv=None, chatty=0, key=None):
self.sessionID = None
self.bufSize = 512
self.DEBUG = False
self.extraLink = self.get9pClient(srv, port, authmode, user, passwd, authsrv, chatty, key)
self.inputLink = self.get9pClient(srv, port, authmode, user, passwd, authsrv, chatty, key)
self.ctlLink = self.get9pClient(srv, port, authmode, user, passwd, authsrv, chatty, key)
self.outputLink = self.get9pClient(srv, port, authmode, user, passwd, authsrv, chatty, key)
self.debugLink = self.get9pClient(srv, port, authmode, user, passwd, authsrv, chatty, key)
def dPrint (self, msg):
if (self.DEBUG):
print msg
def enableDebug (self) :
name = "/csrv/local/ctl"
print "enabling debugging"
if self.debugLink.open (name, py9p.ORDWR) is None:
raise Exception ("XCPU3: could not open " + name)
self.debugLink.write ("debug 1");
self.debugLink.close ();
def disableDebug (self) :
name = "/csrv/local/ctl"
print "disabling debugging"
if self.debugLink.open (name, py9p.ORDWR) is None:
raise Exception ("XCPU3: could not open " + name)
self.debugLink.write ("debug 0");
self.debugLink.close ();
def get9pClient (self, srv, port, authmode='none', user='', passwd=None, authsrv=None, chatty=0, key=None) :
privkey = None
sock = socket.socket(socket.AF_INET)
try:
sock.connect((srv, port),)
except socket.error,e:
print "%s: %s" % (srv, e.args[1])
raise
return py9p.Client(py9p.Sock(sock, 0, chatty), authmode, user, passwd, authsrv, chatty, key=privkey)
def cat(self, name, obj=None, out=None):
if out is None:
out = sys.stdout
if obj is None:
obj = self.extraLink
if obj.open(name) is None:
raise Exception("XCPU3: Could not open " + name)
while 1:
buf = obj.read(self.bufSize)
if len(buf) <= 0:
break
out.write(buf)
obj.close()
def topStat (self, out=None):
name = baseURL + "status"
self.cat(name, None, out)
def getSessionStatus (self, out = None) :
if self.sessionID is None :
print "Error: Session is not started yet"
name = baseURL + str(self.sessionID) + "/status"
self.cat (name, self.extraLink, out)
def startSession (self):
if self.sessionID is not None :
print "Session already running"
return self.sessionID
name = baseURL + "clone"
if self.ctlLink.open (name, py9p.ORDWR) is None :
raise Exception("XCPU3: Could not open " + name)
buf = self.ctlLink.read(self.bufSize)
self.sessionID = int (buf)
return self.sessionID
def requestReservation (self, res) :
if self.sessionID is None :
self.startSession()
if res is None :
return
param = "res " + res
self.ctlLink.write(param)
def requestExecution (self, cmd) :
if self.sessionID is None :
self.startSession()
param = "exec " + cmd
self.ctlLink.write(param)
name = baseURL + str(self.sessionID) + "/stdio"
if self.outputLink.open (name, py9p.OREAD ) is None :
raise Exception("XCPU3: Could not open " + name)
if self.inputLink.open(name, py9p.OWRITE) is None:
raise Exception("XCPU3: Could not open stdio file " + name)
def sendInput (self, input):
if self.sessionID is None :
raise Exception("XCPU3: Session not created")
if input is None :
return
if input == '-' :
inf = sys.stdin
self.dPrint ("Using stdin for input")
else :
inf = open(input, "r", 0)
sz = self.bufSize
while 1:
# buf = inf.read(sz)
buf = inf.readline()
if len(buf) <= 0:
break
self.dPrint ("sending line [" + buf + "]")
self.inputLink.write(buf)
self.dPrint ("input sent")
inf.close()
def getOutputAsync (self) :
if self.sessionID is None :
raise Exception("XCPU3: Session not created")
try:
threading.Thread(target=catAsync, args=(self.outputLink, 1)).start()
except Exception, errtxt:
print errtxt
raise Exception
return
self.dPrint ("Thread started ...")
def getOutput (self) :
if self.sessionID is None :
raise Exception("XCPU3: Session not created")
while 1:
buf = self.outputLink.read(self.bufSize)
if len(buf) <= 0:
break
sys.stdout.write(buf)
def endSession (self) :
if self.sessionID is None :
self.dPrint ( "Session already close")
return
self.inputLink.close()
self.ctlLink.close()
self.outputLink.close()
self.sessionID = None
def runJob (self, cmd, res = None, input = None ):
self.dPrint ("Requesting reservation..")
self.requestReservation(res)
self.dPrint ( "Requesting execution..")
self.requestExecution(cmd)
self.dPrint ( "getting output")
self.getOutputAsync()
self.dPrint ( "sending input")
self.sendInput ("-")
# self.getOutput()
if self.DEBUG:
self.dPrint ( "checking session status")
self.getSessionStatus ()
self.dPrint ( "Closing session")
self.endSession ()
self.dPrint ( "Done..")
def main ():
# Filling default values to all the variables which are not provided with
brasilHost = "localhost"
brasilPort = 5670
nflag = None
osType = None
archType = None
debug = False
shellType = None
# Overriding the values from environment variables
if 'BRASIL_HOST' in os.environ:
brasilHost = os.environ['BRASIL_HOST']
if 'BRASIL_PORT' in os.environ:
brasilPort = int (os.environ['BRASIL_PORT'])
try:
opts, args = getopt.getopt (sys.argv[1:],"Dl:9:h:p:n:o:a:s:")
for o,a in opts:
if o == "-h" :
brasilHost = a
elif o == "-p" :
brasilPort = int (a)
elif o == "-n" :
nflag = a
elif o == "-o" :
osType = a
elif o == "-a" :
archType = a
elif o == "-s" :
shellType = a
elif o == "-D" :
debug = True
elif ( o in ( "-9", "-l" ) ) :
a = 1 + 1
else :
print "ERROR: Wrong option provided " + o
showUsage ()
sys.exit ()
argLen = len (args)
if nflag == None :
print "ERROR: Number of CPU's needed is not provided"
showUsage ()
sys.exit ()
if (int(nflag) < 0 ) :
print "ERROR: Number of CPU's provided are invalid"
showUsage ()
sys.exit ()
resReq = nflag
if osType not in (None, "any", "Any", "ANY", "*" ) :
resReq = resReq + " " + osType
if archType not in (None, "any", "Any", "ANY", "*" ) :
resReq = resReq + " " + archType
if shellType == None :
print "ERROR: shell name not provided"
showUsage ()
sys.exit ()
except getopt.GetoptError:
print "exception"
showUsage ()
sys.exit()
# acutal code responsible for working
mycpu = XCPU3Client (brasilHost, brasilPort)
mycpu.DEBUG = debug
if mycpu.DEBUG :
mycpu.enableDebug()
print "Top statistics is"
mycpu.topStat()
mycpu.runJob (shellType, resReq )
if mycpu.DEBUG :
mycpu.disableDebug()
if __name__ == "__main__" :
main ()