Public Access
move to a more standard python packaging structure
This commit is contained in:
@@ -0,0 +1,296 @@
|
||||
import os
|
||||
import os.path
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import re
|
||||
import glob
|
||||
from . import ExperimentWriter
|
||||
from . import ResultReader
|
||||
import threading
|
||||
import types
|
||||
import signal
|
||||
|
||||
if sys.platform=="win32":
|
||||
import winreg
|
||||
|
||||
__doc__ = """
|
||||
This class handles the backend driver
|
||||
"""
|
||||
|
||||
class BackendDriver(threading.Thread):
|
||||
|
||||
def __init__(self, executable, spool, clear_jobs=False, clear_results=False):
|
||||
threading.Thread.__init__(self, name="Backend Driver")
|
||||
self.core_pid = None
|
||||
self.core_input = None
|
||||
self.core_output = None
|
||||
self.statefilename = None
|
||||
|
||||
self.executable=str(executable)
|
||||
self.spool_dir=spool
|
||||
self.experiment_pattern="job.%09d"
|
||||
self.result_pattern=self.experiment_pattern+".result"
|
||||
|
||||
if not os.path.isfile(self.executable):
|
||||
raise AssertionError("could not find backend %s "%self.executable)
|
||||
if not os.access(self.executable,os.X_OK):
|
||||
raise AssertionError("insufficient rights for backend %s execution"%self.executable)
|
||||
if not os.path.isdir(self.spool_dir):
|
||||
try:
|
||||
os.makedirs(os.path.abspath(self.spool_dir))
|
||||
except OSError as e:
|
||||
print(e)
|
||||
raise AssertionError("could not create backend's spool directory %s "%self.spool_dir)
|
||||
|
||||
# remove stale state filenames
|
||||
if sys.platform.startswith("linux") or sys.platform.startswith("darwin"):
|
||||
old_state_files=glob.glob(os.path.join(self.spool_dir,"*.state"))
|
||||
statelinepattern=re.compile("<state name=\"([^\"]+)\" pid=\"([^\"]+)\" starttime=\"([^\"]+)\">")
|
||||
for statefilename in old_state_files:
|
||||
statefile=open(statefilename,"r")
|
||||
statelines=statefile.readlines()
|
||||
statefile.close
|
||||
del statefile
|
||||
core_pid=None
|
||||
for l in statelines:
|
||||
matched=statelinepattern.match(l)
|
||||
if matched:
|
||||
core_pid=int(matched.group(2))
|
||||
break
|
||||
if core_pid is not None:
|
||||
if os.path.isdir("/proc/%d"%core_pid):
|
||||
raise AssertionError("found backend with pid %d (state file %s) in same spool dir"%(core_pid,statefilename))
|
||||
else:
|
||||
print("removing stale backend state file", statefilename)
|
||||
os.remove(statefilename)
|
||||
else:
|
||||
print("todo: take care of existing backend state files")
|
||||
|
||||
self.result_reader = ResultReader.BlockingResultReader(self.spool_dir,
|
||||
no=0,
|
||||
result_pattern=self.result_pattern,
|
||||
clear_jobs=clear_jobs,
|
||||
clear_results=clear_results)
|
||||
self.experiment_writer = ExperimentWriter.ExperimentWriterWithCleanup(self.spool_dir,
|
||||
no=0,
|
||||
job_pattern=self.experiment_pattern,
|
||||
inform_last_job=self.result_reader)
|
||||
|
||||
self.quit_flag=threading.Event()
|
||||
self.raised_exception=None
|
||||
|
||||
def run(self):
|
||||
# take care of older logfiles
|
||||
self.core_output_filename=os.path.join(self.spool_dir,"logdata")
|
||||
if os.path.isfile(self.core_output_filename):
|
||||
i=0
|
||||
max_logs=100
|
||||
while os.path.isfile(self.core_output_filename+".%02d"%i):
|
||||
i+=1
|
||||
while (i>=max_logs):
|
||||
i-=1
|
||||
os.remove(self.core_output_filename+".%02d"%i)
|
||||
for j in range(i):
|
||||
os.rename(self.core_output_filename+".%02d"%(i-j-1),self.core_output_filename+".%02d"%(i-j))
|
||||
os.rename(self.core_output_filename, self.core_output_filename+".%02d"%0)
|
||||
# create logfile
|
||||
self.core_output=open(self.core_output_filename,"w")
|
||||
|
||||
# again look out for existing state files
|
||||
state_files=glob.glob(os.path.join(self.spool_dir,"*.state"))
|
||||
if state_files:
|
||||
self.raised_exception="found other state file(s) in spool directory: "+",".join(state_files)
|
||||
self.quit_flag.set()
|
||||
return
|
||||
|
||||
# start backend
|
||||
if sys.platform.startswith("linux") or sys.platform.startswith("darwin"):
|
||||
self.core_input=subprocess.Popen([self.executable, "--spool", self.spool_dir],
|
||||
stdout=self.core_output,
|
||||
stderr=self.core_output)
|
||||
|
||||
if sys.platform=="win32":
|
||||
cygwin_root_key=winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, "SOFTWARE\\Cygnus Solutions\\Cygwin\\mounts v2\\/")
|
||||
cygwin_path=winreg.QueryValueEx(cygwin_root_key,"native")[0]
|
||||
os.environ["PATH"]+=";"+os.path.join(cygwin_path,"bin")+";"+os.path.join(cygwin_path,"lib")
|
||||
self.core_input=subprocess.Popen("\"" + self.executable + "\"" + " --spool "+self.spool_dir,
|
||||
stdout=self.core_output,
|
||||
stderr=self.core_output)
|
||||
|
||||
# wait till state file shows up
|
||||
timeout=10
|
||||
# to do: how should I know core's state name????!!!!!
|
||||
self.statefilename=None
|
||||
state_files=glob.glob(os.path.join(self.spool_dir,"*.state"))
|
||||
while len(state_files)==0:
|
||||
if timeout<0 or self.core_input is None or self.core_input.poll() is not None or self.quit_flag.isSet():
|
||||
# look into core log file and include contents
|
||||
print(timeout, self.core_input, self.quit_flag.isSet())
|
||||
if self.core_input is not None:
|
||||
print(self.core_input.poll())
|
||||
log_message=''
|
||||
self.core_input=None
|
||||
if os.path.isfile(self.core_output_filename):
|
||||
# to do include log data
|
||||
log_message='\n'+''.join(open(self.core_output_filename,"r", errors='replace').readlines()[:10])
|
||||
if not log_message:
|
||||
log_message=" no error message from core"
|
||||
self.core_output.close()
|
||||
self.raised_exception="no state file appeared or backend died away:"+log_message
|
||||
print(self.raised_exception)
|
||||
self.quit_flag.set()
|
||||
return
|
||||
time.sleep(0.05)
|
||||
timeout-=0.05
|
||||
state_files=glob.glob(os.path.join(self.spool_dir,"*.state"))
|
||||
|
||||
# save the one
|
||||
if len(state_files)>1:
|
||||
print("did find more than one state file, taking first one!")
|
||||
self.statefilename=state_files[0]
|
||||
|
||||
# read state file
|
||||
statefile=open(self.statefilename,"r")
|
||||
statelines=statefile.readlines()
|
||||
statefile=None
|
||||
statelinepattern=re.compile("<state name=\"([^\"]+)\" pid=\"([^\"]+)\" starttime=\"([^\"]+)\">")
|
||||
self.core_pid=-1
|
||||
for l in statelines:
|
||||
matched=statelinepattern.match(l)
|
||||
if matched:
|
||||
self.core_pid=int(matched.group(2))
|
||||
break
|
||||
|
||||
# wait on flag and look after backend
|
||||
while not self.quit_flag.isSet() and self.is_busy():
|
||||
self.quit_flag.wait(0.1)
|
||||
|
||||
if self.quit_flag.isSet():
|
||||
self.stop_queue()
|
||||
while self.is_busy():
|
||||
time.sleep(0.1)
|
||||
|
||||
if not self.is_busy():
|
||||
if self.core_input is not None:
|
||||
backend_result=self.core_input.poll()
|
||||
wait_loop_counter=0
|
||||
while backend_result is None:
|
||||
# waiting in tenth of a second
|
||||
time.sleep(0.1)
|
||||
wait_loop_counter+=1
|
||||
backend_result=self.core_input.poll()
|
||||
if backend_result is not None: break
|
||||
if wait_loop_counter==10:
|
||||
print("sending termination signal to backend process")
|
||||
self.send_signal("SIGTERM")
|
||||
elif wait_loop_counter==20:
|
||||
print("sending kill signal to backend process")
|
||||
self.send_signal("SIGKILL")
|
||||
elif wait_loop_counter>30:
|
||||
print("no longer waiting for backend shutdown")
|
||||
break
|
||||
|
||||
if backend_result is None:
|
||||
print("backend dit not end properly, please stop it manually")
|
||||
elif backend_result>0:
|
||||
print("backend returned ", backend_result)
|
||||
elif backend_result<0:
|
||||
sig_name=[x for x in dir(signal) if x.startswith("SIG") and \
|
||||
x[3]!="_" and \
|
||||
(type(signal.__dict__[x])is int) and \
|
||||
signal.__dict__[x]==-backend_result]
|
||||
if sig_name:
|
||||
print("backend was terminated by signal ",sig_name[0])
|
||||
else:
|
||||
print("backend was terminated by signal no",-backend_result)
|
||||
self.core_input = None
|
||||
self.core_pid = None
|
||||
|
||||
# the experiment handler should stop
|
||||
if self.experiment_writer is not None:
|
||||
# self.experiment_writer.
|
||||
self.experiment_writer=None
|
||||
|
||||
# tell result reader, game is over...
|
||||
#self.result_reader.stop_no=self.experiment_writer.no
|
||||
if self.result_reader is not None:
|
||||
self.result_reader.poll_time=-1
|
||||
self.result_reader=None
|
||||
|
||||
def clear_job(self,no):
|
||||
jobfilename=os.path.join(self.spool_dir,"job.%09d")
|
||||
resultfilename=os.path.join(self.spool_dir,"job.%09d.result")
|
||||
if os.path.isfile(jobfilename):
|
||||
os.remove(jobfilename)
|
||||
if os.path.isfile(resultfilename):
|
||||
os.remove(resultfilename)
|
||||
|
||||
def get_messages(self):
|
||||
# return pending messages
|
||||
if self.core_output.tell()==os.path.getsize(self.core_output_filename):
|
||||
return None
|
||||
return self.core_output.read()
|
||||
|
||||
def restart_queue(self):
|
||||
self.send_signal("SIGUSR1")
|
||||
|
||||
def stop_queue(self):
|
||||
self.send_signal("SIGQUIT")
|
||||
# assumes success
|
||||
#self.core_pid=None
|
||||
#self.core_input=None
|
||||
|
||||
def abort(self):
|
||||
# abort execution
|
||||
self.send_signal("SIGTERM")
|
||||
# assumes success
|
||||
#self.core_pid=None
|
||||
#self.core_input=None
|
||||
|
||||
def send_signal(self, sig):
|
||||
if self.core_pid is None:
|
||||
print("BackendDriver.send_signal is called with core_pid=None")
|
||||
return
|
||||
try:
|
||||
if sys.platform[:5]=="linux":
|
||||
os.kill(self.core_pid,signal.__dict__[sig])
|
||||
if sys.platform[:7]=="win32":
|
||||
# reg_handle=_winreg.ConnectRegistry(None,_winreg.HKEY_LOCAL_MACHINE)
|
||||
cygwin_root_key=winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, "SOFTWARE\\Cygnus Solutions\\Cygwin\\mounts v2\\/")
|
||||
cygwin_path=winreg.QueryValueEx(cygwin_root_key,"native")[0]
|
||||
kill_command=os.path.join(cygwin_path,"bin","kill.exe")
|
||||
os.popen("%s -%s %d"%(kill_command,sig,self.core_pid))
|
||||
except OSError as e:
|
||||
print("could not send signal %s to core: %s"%(sig, str(e)))
|
||||
|
||||
|
||||
def is_busy(self):
|
||||
"Checks for state file"
|
||||
return self.statefilename is not None and os.path.isfile(self.statefilename) and \
|
||||
self.core_input is not None and self.core_input.poll() is None
|
||||
|
||||
#file_list = glob.glob(os.path.join(self.spool_dir, self.core_state_file))
|
||||
#if len(file_list) != 0:
|
||||
# return True
|
||||
#else:
|
||||
# return False
|
||||
|
||||
|
||||
def get_exp_writer(self):
|
||||
return self.experiment_writer
|
||||
|
||||
def get_res_reader(self):
|
||||
return self.result_reader
|
||||
|
||||
def __del__(self):
|
||||
# stop core and wait for it
|
||||
if self.core_pid is not None:
|
||||
try:
|
||||
self.abort()
|
||||
except OSError:
|
||||
pass
|
||||
self.core_input=None
|
||||
if self.core_output:
|
||||
self.core_output.close()
|
||||
self.core_output=None
|
||||
Reference in New Issue
Block a user