123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471 |
- import os,sys,logging
- import signal, time
- from xmlrpc.server import SimpleXMLRPCServer, SimpleXMLRPCRequestHandler
- import threading
- import queue
- import socket
- import io
- 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 '''
- 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()
- si = open('/dev/null', 'r')
- so = open(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 = 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.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 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', 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()
|