""" CCP4JobController.py: CCP4 GUI Project Copyright (C) 2010 University of York This library is free software: you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License version 3, modified in accordance with the provisions of the license to address the requirements of UK law. You should have received a copy of the modified GNU Lesser General Public License along with this library. If not, copies may be downloaded from http://www.ccp4.ac.uk/ccp4license.php This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details. """ """ Liz Potterton May 2010 - Non-graphical controller for running scripts Liz Potterton May 2011 - Rewrite to use CCP4DbApi to RDMS database and separate jobs as processes """ from PyQt4 import QtCore import CCP4Utils,CCP4Container,CCP4JobServer from CCP4ErrorHandling import * import re,os,copy,sys if sys.platform.startswith("win"): import ctypes class WinProcInfo(ctypes.Structure): _fields_ = [ ('hProcess', ctypes.c_void_p), ('hThread', ctypes.c_void_p), ('dwProcessID', ctypes.c_ulong), ('dwThreadID', ctypes.c_ulong), ] LPWinProcInfo = ctypes.POINTER(WinProcInfo) def WinQPidToPid(p): """ Converts the sip.voidptr from QProcess.pid on Windows to a real pid. """ lp = ctypes.cast(int(p), LPWinProcInfo) return lp.contents.dwProcessID class CServerSetup(CCP4Container.CContainer): insts = None def __init__(self,source=None): CCP4Container.CContainer.__init__(self,name='SERVER_SETUP') CServerSetup.insts = self self.__dict__['source'] = None import CCP4Modules defFile = CCP4Modules.TASKMANAGER().searchDefFile('serverSetup') self.loadContentsFromXml(defFile) self.load(source=source) def load(self,source=None): prefFile,source = self.preferencesFile(source=source) if os.path.exists(prefFile): # Beware if params file has >1 serverGroup we need to add the 2+ groups to the # the container as they are not in the def file import CCP4File, CCP4Annotation fObj = CCP4File.CI2XmlDataFile(prefFile) for sGEle in fObj.getBodyEtree(): if self.get(str(sGEle.tag)) is None: self.setContents( { str(sGEle.tag) : { 'class' :CCP4Annotation.CServerGroup }} ) self.loadDataFromXml(prefFile) self.__dict__['source'] = source def writeAccess(self,source): dir = os.path.split(self.preferencesFile(source)[0])[0] return os.access(dir , os.W_OK | os.X_OK) def preferencesFile(self,source=None): import CCP4Utils,os if source is None or source == 'user': filename = str(os.path.join(CCP4Utils.getDotDirectory(),'configs','serverSetup.params.xml')) if os.path.exists(filename) or source == 'user': return filename,'user' filename = str(os.path.join(CCP4Utils.getCCP4I2Dir(),'local_setup','serverSetup.params.xml')) return filename,'installation' def save(self,source=None): prefFile,source = self.preferencesFile(source=source) if os.path.exists(prefFile): import shutil shutil.copyfile(prefFile,prefFile+'.bak') #print 'CServerSetup.save',prefFile self.saveDataToXml(fileName=prefFile) self.__dict__['source'] = source try: self.emit(QtCore.SIGNAL('serverSetupSaved')) except: pass class CJobController(QtCore.QObject,CCP4JobServer.CJobServer): insts = None MAXTHREADS = 3 CHECKINTERVAL = 1000 SERVERCHECK = 10 REPORTCHECK = 20 # Beware using QProcess does handle the environment customisation USE_QPROCESS = True SERVERSENABLED = None ERROR_CODES = { 101 : { 'description' : 'Control file not found' }, 102 : { 'description' :'Error attempting to start job' }, 103 : { 'description' :'No spare job capacity' }, 104 : { 'description' :'Job queue does not provide next job id' }, 105 : { 'description' :'No plugin name in control file' }, 106 : { 'description' :'Could not find plugin' }, 111 : { 'description' :'Request to watch job that is not running' }, 112 : { 'description' :'Failed killing job' }, 113 : { 'description' :'No job process found - job probably finished' }, 114 : { 'description' :'Failed trying to kill sub-processes of the task process' }, 115 : { 'description' : 'Error in handling remote job' } } def __init__(self,parent=None,db=None): CCP4JobServer.CJobServer.__init__(self) import CCP4Modules if parent is None: parent = CCP4Modules.QTAPPLICATION() if db is None: db = CCP4Modules.PROJECTSMANAGER().db() QtCore.QObject.__init__(self,parent) if CJobController.insts is None: CJobController.insts = self self.db = db # Update the serverParams list if job deleted # When serverParams are moved to db then this should be unnecessary self.connect(self.db,QtCore.SIGNAL('jobDeleted'),self.handleJobDeleted) self.connect(self,QtCore.SIGNAL('failedOpenConnection'),self.handleFail) self.connect(self,QtCore.SIGNAL('failedRemoteCommand'),self.handleFail) self.configFile = None self._watchedJobs = {} self._errorReport = CErrorReport() self._diagnostic = False self._dbFile = None self.blockExit = False def setDbFile(self,dbFile): self._dbFile = dbFile def handleJobDeleted(self,args): self.deleteServerParams(args['jobId']) #print 'CJobController.handleJobDeleted done',self.serverParams.keys() def serversEnabled(self): #FIXME Need to test that param file content is actually usable!! return True if CJobController.SERVERSENABLED is None: import CCP4Utils,os filename = str(os.path.join(CCP4Utils.getDotDirectory(),'configs','serverSetup.params.xml')) if os.path.exists(filename): CJobController.SERVERSENABLED = True filename = str(os.path.join(CCP4Utils.getCCP4I2Dir(),'local_setup','serverSetup.params.xml')) CJobController.SERVERSENABLED = os.path.exists(filename) return CJobController.SERVERSENABLED def resetServersEnabled(self): if CJobController.SERVERSENABLED is None: self.serversEnabled() else: previous = copy.deepcopy(CJobController.SERVERSENABLED) current = self.serversEnabled() if current != previous: self.emit('serversEnabledReset',current) def setDiagnostic(self,mode): self._diagnostic = mode def setConfigFile(self,fileName): # Set the config file to be passed to the sub-process # This is most mechanism to specify alternative config file that # specifies database used for testing purposes self.configFile = fileName def startTimer(self): self.timer = QtCore.QTimer(self) self.connect(self.timer, QtCore.SIGNAL("timeout()"), self.doChecks) self.timer.start(CJobController.CHECKINTERVAL) def pollRemoteReportUpdate(self): for jobId,sP in self._serverParams.items(): if sP.pollReport>0: return True return False def pollRemoteFinish(self): for jobId,sP in self._serverParams.items(): if sP.pollFinish>0: return True return False def pollRemoteFinishCount(self): modes = [ 0,0,0,0,0 ] for jobId,sP in self._serverParams.items(): if sP.pollFinish>=0: modes[sP.pollFinish] += 1 if self._diagnostic: print 'pollRemoteFinishCount',modes return modes def doChecks(self): self.blockExit = True self.checkJobQueue() if self.pollRemoteFinish(): # Perhaps only poll the server every 2 (or more) calls to doCheck() self.finishPollCount = getattr(self,'finishPollCount',0) + 1 if self.finishPollCount >= CJobController.SERVERCHECK: self.finishPollCount = 0 modes = self.pollRemoteFinishCount() #print 'poll remote jobs modes',modes if modes[2]>0: self.pollQsubStat() if modes[1]>0: self.pollForFinishedFlagFile() if modes[3]+modes[4]>0: for jobId,sP in self._serverParams.items(): if sP.pollFinish in [3,4]: try: status = self.customHandler(jobId).pollForFinishedJob(jobId) if status>0: self.handleFinishedServerJob(jobId,status=status) except: pass if self.pollRemoteReportUpdate(): self.reportPollCount = getattr(self,'reportPollCount',0) + 1 #print 'reportPollCount',self.reportPollCount if self.reportPollCount >= CJobController.REPORTCHECK: self.reportPollCount = 0 self.updateReports() """ This is potential cause of database access error in the running script??? if len(CCP4ProjectViewer.FILEWATCHER().jobsByUpdateInterval)>0: self.triggerReportCount = getattr(self,'triggerReportCount',0) + 1 if self.triggerReportCount > 5: self.triggerReportCount = 0 CCP4ProjectViewer.FILEWATCHER().triggerJobsByUpdateInterval() """ self.blockExit = False #self.watchLogFiles() def checkJobQueue(self): # getJobsByStatus defaults to getting queued jobs queuedJobs = self.db.getJobsByStatus() #print 'JOBCONTROLLER.checkJobQueue',queuedJobs if len(queuedJobs)>0: jobId = queuedJobs[0][0] if self.serverParams(jobId) is None: sp = self.runTask(jobId=jobId) else: self.runOnServer(jobId=jobId) def handleFinishedServerJob(self,jobId,status=None): import functools #print 'handleFinishedServerJob',jobId sP = self.serverParams(jobId) if sP is None: return self.setServerParam(jobId,'pollFinish',0) self.setServerParam(jobId,'pollReport',0) if sP.mechanism in [ 'qsub_shared', 'qsub_local', 'ssh_shared'] : status = self.loadRemoteRun(jobId,xmlDbFile=sP.dbXml) elif sP.mechanism != 'custom': if sP.mechanism != 'test': jobDir = os.path.join(sP.projectDirectory,'CCP4_JOBS','job_'+sP.jobNumber) self.transportFiles(jobId,[[sP.local_finished_tarball,sP.remote_finished_tarball], [os.path.join(jobDir,'stdout.txt'),sP.remotePath+'stdout.txt'], [os.path.join(jobDir,'stderr.txt'),sP.remotePath+'stderr.txt'] ],'get',failSignal=False,finishHandler=self.handleFinishedServerJob1) return else: import shutil try: shutil.copyfile(sP.remote_finished_tarball,sP.local_finished_tarball) except Exception as e: print 'ERROR: Copying tarball for local test:',sP.remote_finished_tarball,'to',sP.local_finished_tarball print e else: if self._diagnostic: print 'Copied tarball for local test:',sP.remote_finished_tarball,'to',sP.local_finished_tarball status = self.loadRemoteRun(jobId,sP.local_finished_tarball) else: try: self.customHandler(jobId).handleFinishedJob(jobId,status=status) except Exception as e: self.handleServerFail(jobId=jobId,exception=e) else: status = self.loadRemoteRun(jobId,sP.local_finished_tarball) if status: self.deleteJobBalls(jobId) self.deleteServerParams(jobId) def handleFinishedServerJob1(self,jobId): #print 'handleFinishedServerJob1',jobId sP = self.serverParams(jobId) status = self.loadRemoteRun(jobId,sP.local_finished_tarball) if status: self.deleteJobBalls(jobId) self.deleteServerParams(jobId) def deleteJobBalls(self,jobId,dbXml=False): sP = self.serverParams(jobId) if dbXml: try: os.remove(str(sP.dbXml)) except: print 'ERROR deleting ',sP.dbXml else: try: os.remove(str(sP.local_finished_tarball)) except: print 'ERROR deleting ',sP.local_finished_tarball try: os.remove(str(sP.local_tarball)) except: print 'ERROR deleting ',sP.local_tarball def watchLogFiles(self): # Initialise the runningJobs list is necessary ''' This is for watching all jobs - probably wrong approach if self.runningJobs is None: runningJobs = self.db.getJobsByStatus(status=CCP4DbApi.JOB_STATUS_RUNNING) self.runningJobs = {} for jobId in runningJobs: logFile = self.db._makeJobFileName(jobId=jobId,mode='LOG') size = os.path.getsize(logFile) self.runningJobs[jobId] = { 'logFile' : logFile, 'size' : size } ''' for jobId,details in self._watchedJobs.items(): size = self.getFileSize(details['logFile']) #print 'JOBCONTROLLER.watchLogFiles',jobId,details['logFile'],size if size is not None and \ (details['size'] is None or size > details['size']): #print 'watchLogFiles logUpdated',jobId self._watchedJobs[jobId]['size'] = size self.emit(QtCore.SIGNAL('logFileUpdated'),jobId) def getFileSize(self,fileName): import os try: sz = os.path.getsize(fileName) size = int(sz.strip('L')) except: size = None return size def watchJob(self,jobId=None,mode=True): if mode: if not self._watchedJobs.has_key(jobId): status = self.db.getJobInfo(jobId=jobId,mode='status') if not status in ['Queued','Running']: raise CException(self.__class__,111,'Job id'+str(jobId)) import CCP4Modules logFile = CCP4Modules.PROJECTSMANAGER().makeFileName(jobId=jobId,mode='LOG') size = self.getFileSize(logFile) self._watchedJobs[jobId] = { 'logFile' : logFile, 'size' : size } else: if self._watchedJobs.has_key(jobId): del self._watchedJobs[jobId] #print 'JOBCONTROLLER.watchJob',self._watchedJobs def getArgList(self,fileName,ccp4Dir=None,ccp4i2Dir=None): import sys if ccp4Dir is not None: # This is only used for 'remote' jobs which will only be run on Linux # so can assume '/' separator if ccp4i2Dir is None: ccp4i2Dir = ccp4Dir + '/share/ccp4i2' runTask = ccp4i2Dir + '/bin/runTask.py' exe = ccp4Dir + '/bin/ccp4-python' else: # running locally ccp4i2Dir=CCP4Utils.getCCP4I2Dir() runTask = os.path.join(ccp4i2Dir,'bin','runTask.py') exe = sys.executable argList = [exe,runTask,fileName] if self.configFile is not None: argList.extend(['-config',self.configFile]) if self._dbFile is not None: argList.extend(['-db',self._dbFile]) return argList def runTask(self,jobId=None,wait=None): import subprocess,os,functools,sys #print 'CJobController.runTask pythonExecutable',CCP4Utils.pythonExecutable() controlFile = self.db._makeJobFileName(jobId=jobId,mode='JOB_INPUT') argList = self.getArgList(controlFile) if self._diagnostic: print 'JOBCONTROLLER starting job:',argList callDict = {} path,name = os.path.split(controlFile) if not self.USE_QPROCESS: callDict['stdout'] = open(os.path.join(path,'stdout.txt'),'w') callDict['stderr'] = open(os.path.join(path,'stderr.txt'),'w') #MN run tasks with modified environment carrying : #-port of HTTP Server #-projectid and #-projectname import os, CCP4Modules,CCP4HTTPServerThread try: httpServerPort = CCP4Modules.HTTPSERVER().port except: httpServerPort = CCP4HTTPServerThread.DEFAULT_PORT my_env = os.environ.copy() my_env['CCP4I2_HTTP_PORT'] = str(httpServerPort) if jobId is not None: db = CCP4Modules.PROJECTSMANAGER().db() jobInfo = db.getJobInfo(jobId=jobId,mode=['projectid','projectname','jobnumber']) #print '\*\* jobInfo' #print jobInfo my_env['CCP4I2_PROJECT_ID'] = jobInfo['projectid'] my_env['CCP4I2_PROJECT_NAME'] = jobInfo['projectname'] callDict['env'] = my_env if not self.USE_QPROCESS: try: sp = subprocess.Popen(argList,**callDict) except Exception as e: # Failed to start job -- sp = None self._errorReport.append(self.__class__,102,'Control file: '+str(controlFile)) return sp else: qArgList = QtCore.QStringList() for item in argList[1:]: qArgList.append(item) p = QtCore.QProcess(self) p.setObjectName(str(jobId)) stdoutFile = os.path.join(path,'stdout.txt') stderrFile = os.path.join(path,'stderr.txt') if stdoutFile is not None: if os.path.exists(stdoutFile): CCP4Utils.backupFile(stdoutFile,delete=True) p.setStandardOutputFile(stdoutFile) if stderrFile is not None: if os.path.exists(stderrFile): CCP4Utils.backupFile(stderrFile,delete=True) p.setStandardErrorFile(stderrFile) #MN Changed here to apply environment edits to inform Coot or other tasks of how to talk to the http server processEnvironment = QtCore.QProcessEnvironment.systemEnvironment() for editItem in [('CCP4I2_HTTP_PORT',str(httpServerPort)),('CCP4I2_PROJECT_ID',jobInfo['projectid']),('CCP4I2_PROJECT_NAME',jobInfo['projectname'])]: processEnvironment.insert(editItem[0],editItem[1]) # Fudge for MRBUMP task on OS X, because of rosetta requiring cctbx to set DYLD_LIBRARY_PATH. dbtn = CCP4Modules.PROJECTSMANAGER().db() jobInfoTN = dbtn.getJobInfo(jobId=jobId,mode=['taskname']) if jobInfoTN == "mrbump_basic" and sys.platform == "darwin": if os.environ.has_key('CCP4'): print "Copying $CCP4/lib to DYLD_LIBRARY_PATH" if os.environ.has_key('DYLD_LIBRARY_PATH'): processEnvironment.insert('DYLD_LIBRARY_PATH',os.path.join(os.environ['CCP4'],'lib')+os.pathsep+os.environ['DYLD_LIBRARY_PATH']) else: processEnvironment.insert('DYLD_LIBRARY_PATH',os.path.join(os.environ['CCP4'],'lib')) p.setProcessEnvironment(processEnvironment) #MN end edit self.connect(p,QtCore.SIGNAL('finished(int,QProcess::ExitStatus)'),functools.partial(self.handleFinish,jobId)) p.start(argList[0],qArgList) if wait is not None: p.waitForFinished(wait) import CCP4DbApi self.db.updateJobStatus(jobId=jobId,status=CCP4DbApi.JOB_STATUS_RUNNING) if sys.platform.startswith("win"): self.db.updateJob(jobId,'processId',WinQPidToPid(p.pid())) else: self.db.updateJob(jobId,'processId',int(p.pid())) #print 'runTask processId',p.pid(),type(p.pid()) return p def saveSh(self,jobId,argList,local_sh,pidfile=None): #pyi2 = '$CCP4/share/ccp4i2/bin/pyi2' pyi2 = '$CCP4/bin/ccp4-python' command = pyi2+" "+' '.join(argList[1:]) #print 'saveSh',command tx = '''#!/bin/sh CCP4='''+self.getServerParam(jobId,'ccp4Dir')+''' export CCP4 . ${CCP4}/bin/ccp4.setup-sh echo "Done CCP4 setup" ''' + command + '\n' #print 'saveSh',pidfile if pidfile is not None: tx = tx[0:-1] + \ ''' & pid=$! /bin/cat < ''' + pidfile + ''' $pid EOM echo "PID=$pid" ''' CCP4Utils.saveFile(local_sh, tx ) #print 'saved local sh file',local_sh def qsubOptionsFile(self,jobId=None): # Look for qsub option file in installation, user .CCP4I2/configs # job directory could also be searched import CCP4Utils for filename in [ os.path.join(CCP4Utils.getDotDirectory(),'configs','qsub_options'), os.path.join(CCP4Utils.getCCP4I2Dir(),'local_setup','qsub_options') ]: if os.path.exists(filename): return filename return None def runOnServer(self,jobId=None): '''Run ssh or qsub with or without shared file system always use a temporary db on remote machine to avoid sqlite/NFS issues''' import functools, tempfile, socket, sys import CCP4DbApi import CCP4Modules self.db.updateJobStatus(jobId=jobId,status=CCP4DbApi.JOB_STATUS_REMOTE) sP = self.serverParams(jobId) #print 'runOnServer',sP.machine,sP.mechanism,sP.username controlFile = self.db._makeJobFileName(jobId=jobId,mode='JOB_INPUT') local_sh = os.path.join(os.path.split(controlFile)[0],'remote.sh') if sP.mechanism in [ 'ssh_shared', 'qsub_local', 'qsub_shared' ]: ccp4i2Dir = os.environ.get('CCP4I2_REMOTE','$CCP4/share/ccp4i2') #print 'runOnServer ccp4i2Dir',ccp4i2Dir #argList = self.getArgList(controlFile,ccp4Dir='$CCP4',ccp4i2Dir='$CCP4/share/ccp4i2-devel') argList = self.getArgList(controlFile,ccp4Dir='$CCP4',ccp4i2Dir=ccp4i2Dir) # argList add the -dbxml arg for running on shared file system argList.extend ( [ '-dbxml' , sP.dbXml ] ) # Add redirect stderr/stdout argList.extend ( [ '>',os.path.join(os.path.split(controlFile)[0],'stdout.txt'), '2>', os.path.join(os.path.split(controlFile)[0],'stderr.txt')]) if self._diagnostic: print 'JOBCONTROLLER starting remote job:',argList self.saveSh(jobId,argList=argList,local_sh=local_sh,pidfile=self.pidFile(jobId)) sP.local_report_file = os.path.join( sP.projectDirectory,'CCP4_JOBS','job_'+sP.jobNumber,'program.xml') sP.remoteSh = local_sh elif sP.mechanism != 'custom': # Not share disks - need to create and transport jobball # Use same remote dir and base of filename for all files # Beware remotePath setup assuming server is linux self.setServerParam(jobId,'remotePath',re.sub(r'\$USER',sP.username,sP.tempDir)+'/'+sP.projectName +'_'+sP.jobNumber+'_'+str(jobId)[0:4]+'_') #sP.local_tarball = os.path.join( sP.projectDirectory,'CCP4_TMP','job_'+sP.jobNumber+'_setup.ccp4db.zip') remote_tarball = sP.remotePath+'setup.ccp4db.zip' #sP.local_finished_tarball=os.path.join( sP.projectDirectory,'CCP4_TMP','job_'+sP.jobNumber+'_finished.ccp4db.zip') #sP.remote_finished_tarball=sP.remotePath+'finished.ccp4db.zip' #sP.remote_report_file = sP.remotePath+'work/project/CCP4_JOBS/job_'+sP.jobNumber+'/program.xml' #print 'remote_report_file',sP.remote_report_file #sP.local_report_file = os.path.join( sP.projectDirectory,'CCP4_JOBS','job_'+sP.jobNumber,'program.xml') ccp4i2Dir = os.environ.get('CCP4I2_REMOTE','$CCP4/share/ccp4i2') #print 'runOnServer ccp4i2Dir',ccp4i2Dir argList = self.getArgList(remote_tarball,ccp4Dir='$CCP4',ccp4i2Dir=ccp4i2Dir) argList.extend ( [ '>',sP.remotePath+'stdout.txt', '2>', sP.remotePath+'stderr.txt' ]) if self._diagnostic: print 'JOBCONTROLLER remote command:',argList self.saveSh(jobId,argList=argList,local_sh=local_sh,pidfile=self.pidFile(jobId)) sP.remoteSh = sP.remotePath+'remote.sh' if sP.mechanism not in ['test']: try: self.transportFiles(jobId,[[local_sh,sP.remoteSh],[str(sP.local_tarball),remote_tarball]],'put',self.runOnServer2) except CException as e: self.handleServerFail(jobId=jobId,exception=e) return else: return else: # copy tarball and remote.sh to local tmp for local test import shutil for localFile,remoteFile in [[local_sh,sP.remoteSh],[str(sP.local_tarball),remote_tarball]]: try: shutil.copyfile(localFile,remoteFile) except Exception as e: print 'ERROR copying file for local test',localFile,'to',remoteFile else: print 'Copy for local test:',localFile,'to',remoteFile else: sP.local_tarball = os.path.join( sP.projectDirectory,'CCP4_TMP','job_'+sP.jobNumber+'_setup.ccp4db.zip') sP.local_report_file = os.path.join( sP.projectDirectory,'CCP4_JOBS','job_'+sP.jobNumber,'program.xml') sP.local_finished_tarball=os.path.join( sP.projectDirectory,'CCP4_TMP','job_'+sP.jobNumber+'_finished.ccp4db.zip') try: self.customHandler(jobId).setup(jobId) except Exception as e: self.handleServerFail(jobId=jobId,exception=e) self.runOnServer2(jobId) def runOnServer2(self,jobId): sP = self.serverParams(jobId) if sP is None: return self._transportFilesFinished(jobId) import CCP4DbApi # Use paramiko to get ssh connection - beware failing and not resetting job status if sP.mechanism in [ 'ssh_shared', 'ssh' ]: try: client = self.runInSSH(jobId,sP.remoteSh) except CException as e: self.handleServerFail(jobId=jobId,exception=e) else: self.db.updateJobStatus(jobId=jobId,status=CCP4DbApi.JOB_STATUS_REMOTE) elif sP.mechanism in [ 'test' ]: try: self.runLocalTest(jobId,sP.remoteSh) except CException as e: self.handleServerFail(jobId=jobId,exception=e) else: self.db.updateJobStatus(jobId=jobId,status=CCP4DbApi.JOB_STATUS_REMOTE) elif sP.mechanism in [ 'qsub_local','qsub_shared','qsub_remote' ]: # Fire off qsub optionsFile = self.qsubOptionsFile() try: self.runInQsub(jobId,sP.remoteSh,optionsFile=optionsFile,mechanism=sP.mechanism) #print 'after runInQsub serverParams',sP except Exception as e: self.handleServerFail(jobId=jobId,exception=e) else: self.db.updateJobStatus(jobId=jobId,status=CCP4DbApi.JOB_STATUS_REMOTE) elif sP.mechanism in [ 'custom' ]: try: self.customHandler(jobId).openConnection(jobId) except Exception as e: print 'openConnection fail',e self.handleServerFail(jobId=jobId,exception=e) else: try: self.customHandler(jobId).submit(jobId) except Exception as e: self.handleServerFail(jobId=jobId,exception=e) else: print 'runOnServer2 setServerParam pollReport' self.setServerParam(jobId,'pollFinish',3) self.setServerParam(jobId,'pollReport',True) self.db.updateJobStatus(jobId=jobId,status=CCP4DbApi.JOB_STATUS_REMOTE) def handleServerFail(self,jobId,exception=None): self.handleFail(jobId,exception) def handleFail(self,jobId=None,exception=None): import copy import CCP4DbApi if jobId is None: return self.db.updateJobStatus(jobId=jobId,status=CCP4DbApi.JOB_STATUS_PENDING) sP = self.serverParams(jobId) projectId = copy.deepcopy(sP.projectId) self.deleteServerParams(jobId) if not isinstance(exception,CErrorReport): err = CErrorReport(self.__class__,115,details = str(exception)) else: err = exception print 'Remote run fail:',str(exception) import traceback traceback.print_exc() self.emit(QtCore.SIGNAL('serverJobFailed'),jobId,projectId,err) def handleFinish(self,jobId,status,exitStatus): #print 'CJobController.handleFinish',jobId,status,exitStatus if (exitStatus == QtCore.QProcess.CrashExit or status>0) and jobId is not None: import CCP4DbApi self.db.updateJobStatus(jobId,CCP4DbApi.JOB_STATUS_FAILED) #import CCP4DbUtils #CCP4DbUtils.makeJobBackup(jobId=jobId,db=self.db) def killJobProcess(self,jobId): err = CErrorReport() if self.serverParams(jobId) is not None: self.killRemoteJob(jobId) self.deleteServerParams(jobId) return pid = self.db.getJobInfo(jobId,'processId') #print 'CJobController.killJobProcess',jobId,pid if pid is not None: import signal,sys try: self.killChildProcesses(pid) except Exception as e: err.append(self.__class__,114,details=str(e)) if sys.platform.startswith('win'): try: """ os.kill(pid,signal.CTRL_C_EVENT) """ os.popen("taskkill /pid "+str(pid)+" /F") except Exception as e: err.append(self.__class__112,details=str(e)) else: # SJM 23/09/2014 - SIGQUIT generates coredump which on OS X causes # unwanted stack trace window to appear. Is there # a reason to use QUIT instead of INT (Ctrl-C)? # os.kill(process,signal.SIGQUIT) try: os.kill(pid,signal.SIGINT) except Exception as e: err.append(self.__class__,112,details=str(e)) import CCP4DbApi #print 'to updateJobStatus',jobId,type(jobId) self.db.updateJobStatus(jobId,CCP4DbApi.JOB_STATUS_FAILED) else: err.append(self.__class__,113) if err.maxSeverity()>SEVERITY_WARNING: print 'Error killing job process',err.report() return err def killChildProcesses(self,parent_pid): import signal,sys,psutil # BEWARE this is picking up some other psutil!!! # See https://github.com/giampaolo/psutil # http://stackoverflow.com/questions/3332043/obtaining-pid-of-child-process if sys.platform == 'win32': sig = signal.CTRL_C_EVENT else: sig = signal.SIGQUIT try: proc = psutil.Process(parent_pid) except: print 'killChildProcesses no child process found',parent_pid return #print 'killChildProcesses proc',proc,type(proc),dir(proc) child_proc = proc.children(recursive=True) print 'killChildProcesses child_proc',child_proc for proc in child_proc: try: if sys.platform.startswith('win'): os.popen("taskkill /pid "+str(proc.pid)+" /F") else: os.kill(proc.pid, sig) except Exception as e: print 'Failed to kill process',proc print e def Exit(self): self.timer.stop() #sys.__stdout__.write('CJobController blockExit '+str(self.blockExit)+'\n');sys.__stdout__.flush() CCP4JobServer.CJobServer.Exit(self) def loadRemoteRun(self,jobId=None,compressedFile=None,xmlDbFile=None): import CCP4DbApi,CCP4Modules #print 'loadRemoteRun',jobId,compressedFile,xmlDbFile # Extract database xmlfile if compressedFile is not None: #try: xmlDbFile = CCP4Modules.PROJECTSMANAGER().extractDatabaseXml(compressedFile) ''' except CException as e: e.warningMessage('Reload exported job','Failed extracting database XML file from compressed file') return False except Exception as e: print 'Failed unpacking compressed file',str(e) ''' else: xmlDbFile = os.path.join(os.path.split(xmlDbFile)[0],'DATABASE_final.db.xml') if not os.path.exists(xmlDbFile): try: CCP4Modules.PROJECTSMANAGER().updateJobStatus(jobId,CCP4DbApi.JOB_STATUS_FAILED) except: print 'Failed updating remote job status for:',jobId,'Probably deleted job' pass return False try: projectId = CCP4Modules.PROJECTSMANAGER().db().getJobInfo(jobId=jobId,mode='projectid') except Exception as e: projectId = None print 'Error in loadRemoteRun',jobId,compressedFile print e return # Read dbxml file into a CDbXml and check that it is for this project import CCP4DbApi dbImport = CCP4DbApi.CDbXml(db=CCP4Modules.PROJECTSMANAGER().db(),xmlFile=xmlDbFile) importProjectInfo = dbImport.loadProjectInfo() if projectId is not None and dbImport.projectId != projectId: return dbImport.createTempTables() dbImport.loadTempTable() # If loading jobs to an existing project flag up jobs in temp tables that are already in db #dbImport._diagnostic = True dbImport.setExclInTempTables() # Extract job files from the compressed file if compressedFile is not None: projectDir = CCP4Modules.PROJECTSMANAGER().db().getProjectInfo(projectId=projectId,mode='projectdirectory') import CCP4Export # Despit the name this is not a separate thread! importThread = CCP4Export.ImportProjectThread(self,projectDir=projectDir,compressedFile=compressedFile) importThread.extractJobs(importThread.compressedFile,importThread.projectDir,dbImport=dbImport) # Update the database dbImport.cleanupTempTables() stats = dbImport.importStats() #print 'loadRemoteRun stats',stats if compressedFile is None: dbImport.importTempTables(includeCode=0) else: dbImport.importTempTables() dbImport.removeTempTables() #print 'loadRemoteRun DONE' dbImport.db.emitSignal('projectReset',{'projectId':dbImport.projectId}) status = dbImport.db.getJobInfo(jobId,'status') dbImport.db.emitSignal('jobFinished',{'jobId':jobId,'projectId':dbImport.projectId,'status':status}) def updateReports(self): import CCP4ProjectViewer currentlyOpenJobs = CCP4ProjectViewer.currentlyOpenJobs() #print 'updateReports',self.getJobsToPollReport(),'currentlyOpenJobs',currentlyOpenJobs for jobId in self.getJobsToPollReport(): if jobId in currentlyOpenJobs: sP = self.serverParams(jobId) #print 'Polling for report update',jobId try: if sP.mechanism in [ 'ssh_shared', 'qsub_local', 'qsub_shared' ]: if not os.path.exists(sP.local_report_file): #print 'updateReports program.xml not there' pass else: # QFileSystemWatcher does not see changes made by another machine so must watch the # local program.xml here size = os.stat(sP.local_report_file).st_size #print 'updateReports program.xml size',size if not hasattr(sP,'local_report_size') or size>sP.local_report_size: sP.local_report_size = size self.emit(QtCore.SIGNAL('remoteJobUpdated'),jobId,sP.local_report_file) elif sP.mechanism in [ 'custom' ]: self.customHandler(jobId).transportFiles(jobId,copyList= [[sP.local_report_file,sP.remote_report_file]] , mode='get') self.emit(QtCore.SIGNAL('remoteJobUpdated'),jobId,sP.local_report_file) else: #print 'CJobController.updateReports recover',jobId try: self.transportFiles(jobId,copyList= [[sP.local_report_file,sP.remote_report_file]] , mode='get', failSignal=False, diagnostic=False) except Exception as e: print 'Failed copy program.xml' pass else: #print 'CJobController.updateReports emiting jobFileChanged',sP.local_report_file #self.emit(QtCore.SIGNAL('remoteJobUpdated'),jobId,sP.local_report_file) pass except: pass def listLocalProcesses(self,containsList = []): import psutil import CCP4Utils def contains(exe,containsList): if exe is None: return False for item in containsList: if exe.count(item): return True return False pInfoDict = {} me = CCP4Utils.getUserId() for proc in psutil.process_iter(): try: pinfo = proc.as_dict(attrs=['pid', 'name', 'username','exe','create_time']) except psutil.NoSuchProcess: pass else: if pinfo['username'] == me and (len(containsList)==0 or contains(pinfo['exe'],containsList)): try: pinfo['parent'] = proc.parent().as_dict(attrs=['pid'])['pid'] except: pinfo['parent'] = -1 pinfo['children'] = [] for p in proc.children(recursive=True): try: pinfo['children'].append(p.pid) except: pass #print(pinfo) pInfoDict[pinfo['pid']] = pinfo return pInfoDict #=========================================================================================================== import unittest class testController(unittest.TestCase): def setUp(self): self.controller = CJobController() def testRunFreerflag(self): import os import CCP4Container controlFile = os.path.join(CCP4Utils.getCCP4I2Dir(),'wrappers','freerflag','test_data','test1.data.xml') stdoutFile = os.path.join(CCP4Utils.getTestTmpDir(),'CJobController_test.stdout') c = CCP4Container.CContainer() c.loadDataFromXml(controlFile) output = str(c.outputData.HKLOUT) print 'Output file:',output,'Stdout:',stdoutFile if os.path.exists(output): os.remove(output) self.controller.runTask(fileName=controlFile,wait=3000) self.assertTrue(os.path.exists(output),'No output file created') def TESTSUITE(): suite = unittest.TestLoader().loadTestsFromTestCase(testController) return suite def testModule(): suite = TESTSUITE() unittest.TextTestRunner(verbosity=2).run(suite)