refs #532 let jobmgr kill jobs
parent
6740704919
commit
d4e21dae13
|
@ -1,6 +1,7 @@
|
||||||
import threading
|
import threading
|
||||||
import subprocess
|
import subprocess
|
||||||
import hashlib
|
import hashlib
|
||||||
|
import time
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from opengnsys import operations
|
from opengnsys import operations
|
||||||
from opengnsys.log import logger
|
from opengnsys.log import logger
|
||||||
|
@ -34,14 +35,28 @@ class JobMgr():
|
||||||
|
|
||||||
def prepare_jobs(self):
|
def prepare_jobs(self):
|
||||||
## can't return self.jobs because the Popen object at self.jobs[id]['p'] is not serializable. So, need to create a new dict to return
|
## can't return self.jobs because the Popen object at self.jobs[id]['p'] is not serializable. So, need to create a new dict to return
|
||||||
st = {}
|
st = []
|
||||||
for jobid in self.jobs:
|
for jobid in self.jobs:
|
||||||
j = self.jobs[jobid]
|
j = self.jobs[jobid]
|
||||||
st[jobid] = dict ((k, j[k]) for k in ['pid', 'starttime', 'script', 'client', 'status', 'stdout', 'stderr'])
|
entry = dict ((k, j[k]) for k in ['pid', 'starttime', 'script', 'client', 'status', 'stdout', 'stderr'])
|
||||||
|
entry['jobid'] = jobid
|
||||||
if j['p'].poll() is not None: ## process finished
|
if j['p'].poll() is not None: ## process finished
|
||||||
st[jobid]['rc'] = j['p'].returncode
|
entry['rc'] = j['p'].returncode
|
||||||
st[jobid]['status'] = 'finished'
|
entry['status'] = 'finished'
|
||||||
|
st.append (entry)
|
||||||
return st
|
return st
|
||||||
|
|
||||||
#def kill_job(self, jobid):
|
def terminate_job(self, jobid):
|
||||||
# self.jobs[jobid]['p'].kill()
|
if jobid not in self.jobs: return {}
|
||||||
|
p = self.jobs[jobid]['p']
|
||||||
|
p.terminate()
|
||||||
|
time.sleep (1)
|
||||||
|
if p.poll() is not None:
|
||||||
|
return { 'terminated': True }
|
||||||
|
|
||||||
|
p.kill()
|
||||||
|
time.sleep (1)
|
||||||
|
if p.poll() is not None:
|
||||||
|
return { 'killed': True }
|
||||||
|
|
||||||
|
return { 'killed': False }
|
||||||
|
|
|
@ -47,13 +47,17 @@ class OpenGnSysWorker(ClientWorker):
|
||||||
def onDeactivation(self):
|
def onDeactivation(self):
|
||||||
logger.debug('Deactivate invoked')
|
logger.debug('Deactivate invoked')
|
||||||
|
|
||||||
# Processes script execution
|
|
||||||
def process_script(self, json_params):
|
def process_script(self, json_params):
|
||||||
script = json_params['code']
|
script = json_params['code']
|
||||||
logger.debug('Processing message: script({})'.format(script))
|
logger.debug('Processing message: script({})'.format(script))
|
||||||
self.jobmgr.launch_job (script, True)
|
self.jobmgr.launch_job (script, True)
|
||||||
#self.sendServerMessage('script', {'op', 'launched'})
|
#self.sendServerMessage('script', {'op', 'launched'})
|
||||||
|
|
||||||
|
def process_terminatescript(self, json_params):
|
||||||
|
jobid = json_params['jobid']
|
||||||
|
logger.debug('Processing terminatescript request, jobid "{}"'.format (jobid))
|
||||||
|
self.jobmgr.terminate_job (jobid)
|
||||||
|
|
||||||
def process_preparescripts(self, json_params):
|
def process_preparescripts(self, json_params):
|
||||||
logger.debug('Processing preparescripts request')
|
logger.debug('Processing preparescripts request')
|
||||||
st = self.jobmgr.prepare_jobs()
|
st = self.jobmgr.prepare_jobs()
|
||||||
|
|
|
@ -345,6 +345,17 @@ class OpenGnSysWorker(ServerWorker):
|
||||||
#return {'op': 'launched', 'jobid': jobid} ## TODO obtain jobid generated at the client (can it be done?)
|
#return {'op': 'launched', 'jobid': jobid} ## TODO obtain jobid generated at the client (can it be done?)
|
||||||
return {'op': 'launched'}
|
return {'op': 'launched'}
|
||||||
|
|
||||||
|
@execution_level('full')
|
||||||
|
@check_secret
|
||||||
|
def process_terminatescript(self, path, get_params, post_params, server):
|
||||||
|
jobid = post_params.get('jobid', None)
|
||||||
|
logger.debug('Processing terminate_script request, jobid "{}"'.format (jobid))
|
||||||
|
if jobid is None:
|
||||||
|
return {}
|
||||||
|
self.sendClientMessage('terminatescript', {'jobid': jobid})
|
||||||
|
self.jobmgr.terminate_job (jobid)
|
||||||
|
return {}
|
||||||
|
|
||||||
@execution_level('full')
|
@execution_level('full')
|
||||||
@check_secret
|
@check_secret
|
||||||
def process_preparescripts(self, path, get_params, post_params, server):
|
def process_preparescripts(self, path, get_params, post_params, server):
|
||||||
|
@ -358,7 +369,7 @@ class OpenGnSysWorker(ServerWorker):
|
||||||
logger.debug('Processing preparescripts message from client')
|
logger.debug('Processing preparescripts message from client')
|
||||||
for p in params:
|
for p in params:
|
||||||
#logger.debug ('p "{}"'.format(p))
|
#logger.debug ('p "{}"'.format(p))
|
||||||
self.st[p] = params[p]
|
self.st.append (p)
|
||||||
|
|
||||||
@execution_level('full')
|
@execution_level('full')
|
||||||
@check_secret
|
@check_secret
|
||||||
|
|
Loading…
Reference in New Issue