mirror of
https://git.yoctoproject.org/poky
synced 2026-02-08 18:02:12 +01:00
4602408c69
This changes the way bitbake server works quite radically. Now, the server is always a process based server with the option of starting an XMLRPC listener on a specific inferface/port. Behind the scenes this is done with a "bitbake.sock" file alongside the bitbake.lock file. If we can obtain the lock, we know we need to start a server. The server always listens on the socket and UIs can then connect to this. UIs connect by sending a set of three file descriptors over the domain socket, one for sending commands, one for receiving command results and the other for receiving events. These changes meant we can throw away all the horrid server abstraction code, the plugable transport option to bitbake and the code becomes much more readable and debuggable. It also likely removes a ton of ways you could hang the UI/cooker in weird ways due to all the race conditions that existed with previous processes. Changes: * The foreground option for bitbake-server was dropped. Just tail the log if you really want this, the codepaths were complicated enough without adding one for this. * BBSERVER="autodetect" was dropped. The server will autostart and autoconnect in process mode. You have to specify an xmlrpc server address since that can't be autodetected. I can't see a use case for autodetect now. * The transport/servetype option to bitbake was dropped. * A BB_SERVER_TIMEOUT variable is added which allows the server to stay resident for a period of time after the last client disconnects before unloading. This is used if the -T/--idle-timeout option is not passed to bitbake. This change is invasive and may well introduce new issues however I believe the codebase is in a much better position for further development and debugging. (Bitbake rev: 72a3dbe13a23588e24c0baca6d58c35cdeba3f63) Signed-off-by: Richard Purdie <richard.purdie@linuxfoundation.org>
480 lines
15 KiB
Python
480 lines
15 KiB
Python
import os,sys,logging
|
|
import signal, time
|
|
from xmlrpc.server import SimpleXMLRPCServer, SimpleXMLRPCRequestHandler
|
|
import threading
|
|
import queue
|
|
import socket
|
|
import io
|
|
import sqlite3
|
|
import bb.server.xmlrpcclient
|
|
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 '''
|
|
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.dump_db, "dump_db")
|
|
self.register_function(self.importone, "importone")
|
|
self.register_introspection_functions()
|
|
|
|
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
|
|
|
|
bb.utils.set_process_name("PRServ Handler")
|
|
|
|
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):
|
|
if self.table:
|
|
self.table.sync()
|
|
|
|
def sigterm_handler(self, signum, stack):
|
|
if self.table:
|
|
self.table.sync()
|
|
self.quit=True
|
|
|
|
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 dump_db(self):
|
|
"""
|
|
Returns a script (string) that reconstructs the state of the
|
|
entire database at the time this function is called. The script
|
|
language is defined by the backing database engine, which is a
|
|
function of server configuration.
|
|
Returns None if the database engine does not support dumping to
|
|
script or if some other error is encountered in processing.
|
|
"""
|
|
buff = io.StringIO()
|
|
try:
|
|
self.table.sync()
|
|
self.table.dump_db(buff)
|
|
return buff.getvalue()
|
|
except Exception as exc:
|
|
logger.error(str(exc))
|
|
return None
|
|
finally:
|
|
buff.close()
|
|
|
|
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
|
|
|
|
bb.utils.set_process_name("PRServ")
|
|
|
|
# DB connection must be created after all forks
|
|
self.db = prserv.db.PRData(self.dbfile)
|
|
self.table = self.db["PRMAIN"]
|
|
|
|
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.db.disconnect()
|
|
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.chdir("/")
|
|
|
|
sys.stdout.flush()
|
|
sys.stderr.flush()
|
|
|
|
# We could be called from a python thread with io.StringIO as
|
|
# stdout/stderr or it could be 'real' unix fd forking where we need
|
|
# to physically close the fds to prevent the program launching us from
|
|
# potentially hanging on a pipe. Handle both cases.
|
|
si = open('/dev/null', 'r')
|
|
try:
|
|
os.dup2(si.fileno(),sys.stdin.fileno())
|
|
except (AttributeError, io.UnsupportedOperation):
|
|
sys.stdin = si
|
|
so = open(self.logfile, 'a+')
|
|
try:
|
|
os.dup2(so.fileno(),sys.stdout.fileno())
|
|
except (AttributeError, io.UnsupportedOperation):
|
|
sys.stdout = so
|
|
try:
|
|
os.dup2(so.fileno(),sys.stderr.fileno())
|
|
except (AttributeError, io.UnsupportedOperation):
|
|
sys.stderr = so
|
|
|
|
# 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 = open(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.xmlrpcclient._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 dump_db(self):
|
|
return self.connection.dump_db()
|
|
|
|
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):
|
|
ip = socket.gethostbyname(host)
|
|
pidfile = PIDPREFIX % (ip, port)
|
|
try:
|
|
pf = open(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), (ip,port))
|
|
server.start()
|
|
|
|
# Sometimes, the port (i.e. localhost:0) indicated by the user does not match with
|
|
# the one the server actually is listening, so at least warn the user about it
|
|
_,rport = server.getinfo()
|
|
if port != rport:
|
|
sys.stdout.write("Server is listening at port %s instead of %s\n"
|
|
% (rport,port))
|
|
return 0
|
|
|
|
def stop_daemon(host, port):
|
|
import glob
|
|
ip = socket.gethostbyname(host)
|
|
pidfile = PIDPREFIX % (ip, port)
|
|
try:
|
|
pf = open(pidfile,'r')
|
|
pid = int(pf.readline().strip())
|
|
pf.close()
|
|
except IOError:
|
|
pid = None
|
|
|
|
if not pid:
|
|
# when server starts at port=0 (i.e. localhost:0), server actually takes another port,
|
|
# so at least advise the user which ports the corresponding server is listening
|
|
ports = []
|
|
portstr = ""
|
|
for pf in glob.glob(PIDPREFIX % (ip,'*')):
|
|
bn = os.path.basename(pf)
|
|
root, _ = os.path.splitext(bn)
|
|
ports.append(root.split('_')[-1])
|
|
if len(ports):
|
|
portstr = "Wrong port? Other ports listening at %s: %s" % (host, ' '.join(ports))
|
|
|
|
sys.stderr.write("pidfile %s does not exist. Daemon not running? %s\n"
|
|
% (pidfile,portstr))
|
|
return 1
|
|
|
|
try:
|
|
PRServerConnection(ip, 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 = list(filter(None, (d.getVar('PRSERV_HOST') 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") or d.getVar("CACHE"))
|
|
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()
|