mirror of
https://git.yoctoproject.org/poky
synced 2026-02-07 09:16:36 +01:00
85256c8fa9
Worryingly, if you SIGKILL the bitbake cooker, an autostarted PR server will remain behind. It turns out there are a few things we should do: * The PR service doesn't need to daemonize when started from cooker, it just complicated the process lifecycle. Add a fork() method to handle this and use the non-daemon mode for the singleton. * Reset the sigterm and sigint handlers. Bitbake cooker installs its own which we inherit meaning PR server was ignoring SIGTERM. Installing our own handlers which include a sync makes most sense here. Since we're in the code, make it sync the database on SIGINT. * Use the new bb.utils.signal_on_parent_exit() call so that we get a SIGTERM when the parent (usually cooker) exits and we can shutdown too. Alternatives would be having an open pipe or polling os.getppid() for changes but this seems more effective. (Bitbake rev: 05d31fa1f56bd3d3d363a16a421d9ba7541d4293) Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>
419 lines
13 KiB
Python
419 lines
13 KiB
Python
import os,sys,logging
|
|
import signal, time
|
|
from SimpleXMLRPCServer import SimpleXMLRPCServer, SimpleXMLRPCRequestHandler
|
|
import threading
|
|
import Queue
|
|
|
|
try:
|
|
import sqlite3
|
|
except ImportError:
|
|
from pysqlite2 import dbapi2 as sqlite3
|
|
|
|
import bb.server.xmlrpc
|
|
import prserv
|
|
import prserv.db
|
|
import errno
|
|
|
|
logger = logging.getLogger("BitBake.PRserv")
|
|
|
|
if sys.hexversion < 0x020600F0:
|
|
print("Sorry, python 2.6 or later is required.")
|
|
sys.exit(1)
|
|
|
|
class Handler(SimpleXMLRPCRequestHandler):
|
|
def _dispatch(self,method,params):
|
|
try:
|
|
value=self.server.funcs[method](*params)
|
|
except:
|
|
import traceback
|
|
traceback.print_exc()
|
|
raise
|
|
return value
|
|
|
|
PIDPREFIX = "/tmp/PRServer_%s_%s.pid"
|
|
singleton = None
|
|
|
|
|
|
class PRServer(SimpleXMLRPCServer):
|
|
def __init__(self, dbfile, logfile, interface, daemon=True):
|
|
''' constructor '''
|
|
import socket
|
|
try:
|
|
SimpleXMLRPCServer.__init__(self, interface,
|
|
logRequests=False, allow_none=True)
|
|
except socket.error:
|
|
ip=socket.gethostbyname(interface[0])
|
|
port=interface[1]
|
|
msg="PR Server unable to bind to %s:%s\n" % (ip, port)
|
|
sys.stderr.write(msg)
|
|
raise PRServiceConfigError
|
|
|
|
self.dbfile=dbfile
|
|
self.daemon=daemon
|
|
self.logfile=logfile
|
|
self.working_thread=None
|
|
self.host, self.port = self.socket.getsockname()
|
|
self.pidfile=PIDPREFIX % (self.host, self.port)
|
|
|
|
self.register_function(self.getPR, "getPR")
|
|
self.register_function(self.quit, "quit")
|
|
self.register_function(self.ping, "ping")
|
|
self.register_function(self.export, "export")
|
|
self.register_function(self.importone, "importone")
|
|
self.register_introspection_functions()
|
|
|
|
self.db = prserv.db.PRData(self.dbfile)
|
|
self.table = self.db["PRMAIN"]
|
|
|
|
self.requestqueue = Queue.Queue()
|
|
self.handlerthread = threading.Thread(target = self.process_request_thread)
|
|
self.handlerthread.daemon = False
|
|
|
|
def process_request_thread(self):
|
|
"""Same as in BaseServer but as a thread.
|
|
|
|
In addition, exception handling is done here.
|
|
|
|
"""
|
|
iter_count = 1
|
|
# 60 iterations between syncs or sync if dirty every ~30 seconds
|
|
iterations_between_sync = 60
|
|
|
|
while not self.quit:
|
|
try:
|
|
(request, client_address) = self.requestqueue.get(True, 30)
|
|
except Queue.Empty:
|
|
self.table.sync_if_dirty()
|
|
continue
|
|
try:
|
|
self.finish_request(request, client_address)
|
|
self.shutdown_request(request)
|
|
iter_count = (iter_count + 1) % iterations_between_sync
|
|
if iter_count == 0:
|
|
self.table.sync_if_dirty()
|
|
except:
|
|
self.handle_error(request, client_address)
|
|
self.shutdown_request(request)
|
|
self.table.sync()
|
|
self.table.sync_if_dirty()
|
|
|
|
def sigint_handler(self, signum, stack):
|
|
self.table.sync()
|
|
|
|
def sigterm_handler(self, signum, stack):
|
|
self.table.sync()
|
|
raise SystemExit
|
|
|
|
def process_request(self, request, client_address):
|
|
self.requestqueue.put((request, client_address))
|
|
|
|
def export(self, version=None, pkgarch=None, checksum=None, colinfo=True):
|
|
try:
|
|
return self.table.export(version, pkgarch, checksum, colinfo)
|
|
except sqlite3.Error as exc:
|
|
logger.error(str(exc))
|
|
return None
|
|
|
|
def importone(self, version, pkgarch, checksum, value):
|
|
return self.table.importone(version, pkgarch, checksum, value)
|
|
|
|
def ping(self):
|
|
return not self.quit
|
|
|
|
def getinfo(self):
|
|
return (self.host, self.port)
|
|
|
|
def getPR(self, version, pkgarch, checksum):
|
|
try:
|
|
return self.table.getValue(version, pkgarch, checksum)
|
|
except prserv.NotFoundError:
|
|
logger.error("can not find value for (%s, %s)",version, checksum)
|
|
return None
|
|
except sqlite3.Error as exc:
|
|
logger.error(str(exc))
|
|
return None
|
|
|
|
def quit(self):
|
|
self.quit=True
|
|
return
|
|
|
|
def work_forever(self,):
|
|
self.quit = False
|
|
self.timeout = 0.5
|
|
|
|
logger.info("Started PRServer with DBfile: %s, IP: %s, PORT: %s, PID: %s" %
|
|
(self.dbfile, self.host, self.port, str(os.getpid())))
|
|
|
|
self.handlerthread.start()
|
|
while not self.quit:
|
|
self.handle_request()
|
|
self.handlerthread.join()
|
|
self.table.sync()
|
|
logger.info("PRServer: stopping...")
|
|
self.server_close()
|
|
return
|
|
|
|
def start(self):
|
|
if self.daemon:
|
|
pid = self.daemonize()
|
|
else:
|
|
pid = self.fork()
|
|
|
|
# Ensure both the parent sees this and the child from the work_forever log entry above
|
|
logger.info("Started PRServer with DBfile: %s, IP: %s, PORT: %s, PID: %s" %
|
|
(self.dbfile, self.host, self.port, str(pid)))
|
|
|
|
def delpid(self):
|
|
os.remove(self.pidfile)
|
|
|
|
def daemonize(self):
|
|
"""
|
|
See Advanced Programming in the UNIX, Sec 13.3
|
|
"""
|
|
try:
|
|
pid = os.fork()
|
|
if pid > 0:
|
|
os.waitpid(pid, 0)
|
|
#parent return instead of exit to give control
|
|
return pid
|
|
except OSError as e:
|
|
raise Exception("%s [%d]" % (e.strerror, e.errno))
|
|
|
|
os.setsid()
|
|
"""
|
|
fork again to make sure the daemon is not session leader,
|
|
which prevents it from acquiring controlling terminal
|
|
"""
|
|
try:
|
|
pid = os.fork()
|
|
if pid > 0: #parent
|
|
os._exit(0)
|
|
except OSError as e:
|
|
raise Exception("%s [%d]" % (e.strerror, e.errno))
|
|
|
|
self.cleanup_handles()
|
|
os._exit(0)
|
|
|
|
def fork(self):
|
|
try:
|
|
pid = os.fork()
|
|
if pid > 0:
|
|
return pid
|
|
except OSError as e:
|
|
raise Exception("%s [%d]" % (e.strerror, e.errno))
|
|
|
|
bb.utils.signal_on_parent_exit("SIGTERM")
|
|
self.cleanup_handles()
|
|
os._exit(0)
|
|
|
|
def cleanup_handles(self):
|
|
signal.signal(signal.SIGINT, self.sigint_handler)
|
|
signal.signal(signal.SIGTERM, self.sigterm_handler)
|
|
os.umask(0)
|
|
os.chdir("/")
|
|
|
|
sys.stdout.flush()
|
|
sys.stderr.flush()
|
|
si = file('/dev/null', 'r')
|
|
so = file(self.logfile, 'a+')
|
|
se = so
|
|
os.dup2(si.fileno(),sys.stdin.fileno())
|
|
os.dup2(so.fileno(),sys.stdout.fileno())
|
|
os.dup2(se.fileno(),sys.stderr.fileno())
|
|
|
|
# Clear out all log handlers prior to the fork() to avoid calling
|
|
# event handlers not part of the PRserver
|
|
for logger_iter in logging.Logger.manager.loggerDict.keys():
|
|
logging.getLogger(logger_iter).handlers = []
|
|
|
|
# Ensure logging makes it to the logfile
|
|
streamhandler = logging.StreamHandler()
|
|
streamhandler.setLevel(logging.DEBUG)
|
|
formatter = bb.msg.BBLogFormatter("%(levelname)s: %(message)s")
|
|
streamhandler.setFormatter(formatter)
|
|
logger.addHandler(streamhandler)
|
|
|
|
# write pidfile
|
|
pid = str(os.getpid())
|
|
pf = file(self.pidfile, 'w')
|
|
pf.write("%s\n" % pid)
|
|
pf.close()
|
|
|
|
self.work_forever()
|
|
self.delpid()
|
|
|
|
class PRServSingleton(object):
|
|
def __init__(self, dbfile, logfile, interface):
|
|
self.dbfile = dbfile
|
|
self.logfile = logfile
|
|
self.interface = interface
|
|
self.host = None
|
|
self.port = None
|
|
|
|
def start(self):
|
|
self.prserv = PRServer(self.dbfile, self.logfile, self.interface, daemon=False)
|
|
self.prserv.start()
|
|
self.host, self.port = self.prserv.getinfo()
|
|
|
|
def getinfo(self):
|
|
return (self.host, self.port)
|
|
|
|
class PRServerConnection(object):
|
|
def __init__(self, host, port):
|
|
if is_local_special(host, port):
|
|
host, port = singleton.getinfo()
|
|
self.host = host
|
|
self.port = port
|
|
self.connection, self.transport = bb.server.xmlrpc._create_server(self.host, self.port)
|
|
|
|
def terminate(self):
|
|
try:
|
|
logger.info("Terminating PRServer...")
|
|
self.connection.quit()
|
|
except Exception as exc:
|
|
sys.stderr.write("%s\n" % str(exc))
|
|
|
|
def getPR(self, version, pkgarch, checksum):
|
|
return self.connection.getPR(version, pkgarch, checksum)
|
|
|
|
def ping(self):
|
|
return self.connection.ping()
|
|
|
|
def export(self,version=None, pkgarch=None, checksum=None, colinfo=True):
|
|
return self.connection.export(version, pkgarch, checksum, colinfo)
|
|
|
|
def importone(self, version, pkgarch, checksum, value):
|
|
return self.connection.importone(version, pkgarch, checksum, value)
|
|
|
|
def getinfo(self):
|
|
return self.host, self.port
|
|
|
|
def start_daemon(dbfile, host, port, logfile):
|
|
pidfile = PIDPREFIX % (host, port)
|
|
try:
|
|
pf = file(pidfile,'r')
|
|
pid = int(pf.readline().strip())
|
|
pf.close()
|
|
except IOError:
|
|
pid = None
|
|
|
|
if pid:
|
|
sys.stderr.write("pidfile %s already exist. Daemon already running?\n"
|
|
% pidfile)
|
|
return 1
|
|
|
|
server = PRServer(os.path.abspath(dbfile), os.path.abspath(logfile), (host,port))
|
|
server.start()
|
|
return 0
|
|
|
|
def stop_daemon(host, port):
|
|
pidfile = PIDPREFIX % (host, port)
|
|
try:
|
|
pf = file(pidfile,'r')
|
|
pid = int(pf.readline().strip())
|
|
pf.close()
|
|
except IOError:
|
|
pid = None
|
|
|
|
if not pid:
|
|
sys.stderr.write("pidfile %s does not exist. Daemon not running?\n"
|
|
% pidfile)
|
|
|
|
try:
|
|
PRServerConnection(host, port).terminate()
|
|
except:
|
|
logger.critical("Stop PRService %s:%d failed" % (host,port))
|
|
|
|
try:
|
|
if pid:
|
|
wait_timeout = 0
|
|
print("Waiting for pr-server to exit.")
|
|
while is_running(pid) and wait_timeout < 50:
|
|
time.sleep(0.1)
|
|
wait_timeout += 1
|
|
|
|
if is_running(pid):
|
|
print("Sending SIGTERM to pr-server.")
|
|
os.kill(pid,signal.SIGTERM)
|
|
time.sleep(0.1)
|
|
|
|
if os.path.exists(pidfile):
|
|
os.remove(pidfile)
|
|
|
|
except OSError as e:
|
|
err = str(e)
|
|
if err.find("No such process") <= 0:
|
|
raise e
|
|
|
|
return 0
|
|
|
|
def is_running(pid):
|
|
try:
|
|
os.kill(pid, 0)
|
|
except OSError as err:
|
|
if err.errno == errno.ESRCH:
|
|
return False
|
|
return True
|
|
|
|
def is_local_special(host, port):
|
|
if host.strip().upper() == 'localhost'.upper() and (not port):
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
class PRServiceConfigError(Exception):
|
|
pass
|
|
|
|
def auto_start(d):
|
|
global singleton
|
|
|
|
host_params = filter(None, (d.getVar('PRSERV_HOST', True) or '').split(':'))
|
|
if not host_params:
|
|
return None
|
|
|
|
if len(host_params) != 2:
|
|
logger.critical('\n'.join(['PRSERV_HOST: incorrect format',
|
|
'Usage: PRSERV_HOST = "<hostname>:<port>"']))
|
|
raise PRServiceConfigError
|
|
|
|
if is_local_special(host_params[0], int(host_params[1])) and not singleton:
|
|
import bb.utils
|
|
cachedir = (d.getVar("PERSISTENT_DIR", True) or d.getVar("CACHE", True))
|
|
if not cachedir:
|
|
logger.critical("Please set the 'PERSISTENT_DIR' or 'CACHE' variable")
|
|
raise PRServiceConfigError
|
|
bb.utils.mkdirhier(cachedir)
|
|
dbfile = os.path.join(cachedir, "prserv.sqlite3")
|
|
logfile = os.path.join(cachedir, "prserv.log")
|
|
singleton = PRServSingleton(os.path.abspath(dbfile), os.path.abspath(logfile), ("localhost",0))
|
|
singleton.start()
|
|
if singleton:
|
|
host, port = singleton.getinfo()
|
|
else:
|
|
host = host_params[0]
|
|
port = int(host_params[1])
|
|
|
|
try:
|
|
connection = PRServerConnection(host,port)
|
|
connection.ping()
|
|
realhost, realport = connection.getinfo()
|
|
return str(realhost) + ":" + str(realport)
|
|
|
|
except Exception:
|
|
logger.critical("PRservice %s:%d not available" % (host, port))
|
|
raise PRServiceConfigError
|
|
|
|
def auto_shutdown(d=None):
|
|
global singleton
|
|
if singleton:
|
|
host, port = singleton.getinfo()
|
|
try:
|
|
PRServerConnection(host, port).terminate()
|
|
except:
|
|
logger.critical("Stop PRService %s:%d failed" % (host,port))
|
|
singleton = None
|
|
|
|
def ping(host, port):
|
|
conn=PRServerConnection(host, port)
|
|
return conn.ping()
|