serv.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471
  1. import os,sys,logging
  2. import signal, time
  3. from xmlrpc.server import SimpleXMLRPCServer, SimpleXMLRPCRequestHandler
  4. import threading
  5. import queue
  6. import socket
  7. import io
  8. try:
  9. import sqlite3
  10. except ImportError:
  11. from pysqlite2 import dbapi2 as sqlite3
  12. import bb.server.xmlrpc
  13. import prserv
  14. import prserv.db
  15. import errno
  16. logger = logging.getLogger("BitBake.PRserv")
  17. if sys.hexversion < 0x020600F0:
  18. print("Sorry, python 2.6 or later is required.")
  19. sys.exit(1)
  20. class Handler(SimpleXMLRPCRequestHandler):
  21. def _dispatch(self,method,params):
  22. try:
  23. value=self.server.funcs[method](*params)
  24. except:
  25. import traceback
  26. traceback.print_exc()
  27. raise
  28. return value
  29. PIDPREFIX = "/tmp/PRServer_%s_%s.pid"
  30. singleton = None
  31. class PRServer(SimpleXMLRPCServer):
  32. def __init__(self, dbfile, logfile, interface, daemon=True):
  33. ''' constructor '''
  34. try:
  35. SimpleXMLRPCServer.__init__(self, interface,
  36. logRequests=False, allow_none=True)
  37. except socket.error:
  38. ip=socket.gethostbyname(interface[0])
  39. port=interface[1]
  40. msg="PR Server unable to bind to %s:%s\n" % (ip, port)
  41. sys.stderr.write(msg)
  42. raise PRServiceConfigError
  43. self.dbfile=dbfile
  44. self.daemon=daemon
  45. self.logfile=logfile
  46. self.working_thread=None
  47. self.host, self.port = self.socket.getsockname()
  48. self.pidfile=PIDPREFIX % (self.host, self.port)
  49. self.register_function(self.getPR, "getPR")
  50. self.register_function(self.quit, "quit")
  51. self.register_function(self.ping, "ping")
  52. self.register_function(self.export, "export")
  53. self.register_function(self.dump_db, "dump_db")
  54. self.register_function(self.importone, "importone")
  55. self.register_introspection_functions()
  56. self.requestqueue = queue.Queue()
  57. self.handlerthread = threading.Thread(target = self.process_request_thread)
  58. self.handlerthread.daemon = False
  59. def process_request_thread(self):
  60. """Same as in BaseServer but as a thread.
  61. In addition, exception handling is done here.
  62. """
  63. iter_count = 1
  64. # 60 iterations between syncs or sync if dirty every ~30 seconds
  65. iterations_between_sync = 60
  66. bb.utils.set_process_name("PRServ Handler")
  67. while not self.quit:
  68. try:
  69. (request, client_address) = self.requestqueue.get(True, 30)
  70. except queue.Empty:
  71. self.table.sync_if_dirty()
  72. continue
  73. try:
  74. self.finish_request(request, client_address)
  75. self.shutdown_request(request)
  76. iter_count = (iter_count + 1) % iterations_between_sync
  77. if iter_count == 0:
  78. self.table.sync_if_dirty()
  79. except:
  80. self.handle_error(request, client_address)
  81. self.shutdown_request(request)
  82. self.table.sync()
  83. self.table.sync_if_dirty()
  84. def sigint_handler(self, signum, stack):
  85. if self.table:
  86. self.table.sync()
  87. def sigterm_handler(self, signum, stack):
  88. if self.table:
  89. self.table.sync()
  90. self.quit=True
  91. def process_request(self, request, client_address):
  92. self.requestqueue.put((request, client_address))
  93. def export(self, version=None, pkgarch=None, checksum=None, colinfo=True):
  94. try:
  95. return self.table.export(version, pkgarch, checksum, colinfo)
  96. except sqlite3.Error as exc:
  97. logger.error(str(exc))
  98. return None
  99. def dump_db(self):
  100. """
  101. Returns a script (string) that reconstructs the state of the
  102. entire database at the time this function is called. The script
  103. language is defined by the backing database engine, which is a
  104. function of server configuration.
  105. Returns None if the database engine does not support dumping to
  106. script or if some other error is encountered in processing.
  107. """
  108. buff = io.StringIO()
  109. try:
  110. self.table.sync()
  111. self.table.dump_db(buff)
  112. return buff.getvalue()
  113. except Exception as exc:
  114. logger.error(str(exc))
  115. return None
  116. finally:
  117. buff.close()
  118. def importone(self, version, pkgarch, checksum, value):
  119. return self.table.importone(version, pkgarch, checksum, value)
  120. def ping(self):
  121. return not self.quit
  122. def getinfo(self):
  123. return (self.host, self.port)
  124. def getPR(self, version, pkgarch, checksum):
  125. try:
  126. return self.table.getValue(version, pkgarch, checksum)
  127. except prserv.NotFoundError:
  128. logger.error("can not find value for (%s, %s)",version, checksum)
  129. return None
  130. except sqlite3.Error as exc:
  131. logger.error(str(exc))
  132. return None
  133. def quit(self):
  134. self.quit=True
  135. return
  136. def work_forever(self,):
  137. self.quit = False
  138. self.timeout = 0.5
  139. bb.utils.set_process_name("PRServ")
  140. # DB connection must be created after all forks
  141. self.db = prserv.db.PRData(self.dbfile)
  142. self.table = self.db["PRMAIN"]
  143. logger.info("Started PRServer with DBfile: %s, IP: %s, PORT: %s, PID: %s" %
  144. (self.dbfile, self.host, self.port, str(os.getpid())))
  145. self.handlerthread.start()
  146. while not self.quit:
  147. self.handle_request()
  148. self.handlerthread.join()
  149. self.db.disconnect()
  150. logger.info("PRServer: stopping...")
  151. self.server_close()
  152. return
  153. def start(self):
  154. if self.daemon:
  155. pid = self.daemonize()
  156. else:
  157. pid = self.fork()
  158. # Ensure both the parent sees this and the child from the work_forever log entry above
  159. logger.info("Started PRServer with DBfile: %s, IP: %s, PORT: %s, PID: %s" %
  160. (self.dbfile, self.host, self.port, str(pid)))
  161. def delpid(self):
  162. os.remove(self.pidfile)
  163. def daemonize(self):
  164. """
  165. See Advanced Programming in the UNIX, Sec 13.3
  166. """
  167. try:
  168. pid = os.fork()
  169. if pid > 0:
  170. os.waitpid(pid, 0)
  171. #parent return instead of exit to give control
  172. return pid
  173. except OSError as e:
  174. raise Exception("%s [%d]" % (e.strerror, e.errno))
  175. os.setsid()
  176. """
  177. fork again to make sure the daemon is not session leader,
  178. which prevents it from acquiring controlling terminal
  179. """
  180. try:
  181. pid = os.fork()
  182. if pid > 0: #parent
  183. os._exit(0)
  184. except OSError as e:
  185. raise Exception("%s [%d]" % (e.strerror, e.errno))
  186. self.cleanup_handles()
  187. os._exit(0)
  188. def fork(self):
  189. try:
  190. pid = os.fork()
  191. if pid > 0:
  192. return pid
  193. except OSError as e:
  194. raise Exception("%s [%d]" % (e.strerror, e.errno))
  195. bb.utils.signal_on_parent_exit("SIGTERM")
  196. self.cleanup_handles()
  197. os._exit(0)
  198. def cleanup_handles(self):
  199. signal.signal(signal.SIGINT, self.sigint_handler)
  200. signal.signal(signal.SIGTERM, self.sigterm_handler)
  201. os.chdir("/")
  202. sys.stdout.flush()
  203. sys.stderr.flush()
  204. si = open('/dev/null', 'r')
  205. so = open(self.logfile, 'a+')
  206. se = so
  207. os.dup2(si.fileno(),sys.stdin.fileno())
  208. os.dup2(so.fileno(),sys.stdout.fileno())
  209. os.dup2(se.fileno(),sys.stderr.fileno())
  210. # Clear out all log handlers prior to the fork() to avoid calling
  211. # event handlers not part of the PRserver
  212. for logger_iter in logging.Logger.manager.loggerDict.keys():
  213. logging.getLogger(logger_iter).handlers = []
  214. # Ensure logging makes it to the logfile
  215. streamhandler = logging.StreamHandler()
  216. streamhandler.setLevel(logging.DEBUG)
  217. formatter = bb.msg.BBLogFormatter("%(levelname)s: %(message)s")
  218. streamhandler.setFormatter(formatter)
  219. logger.addHandler(streamhandler)
  220. # write pidfile
  221. pid = str(os.getpid())
  222. pf = open(self.pidfile, 'w')
  223. pf.write("%s\n" % pid)
  224. pf.close()
  225. self.work_forever()
  226. self.delpid()
  227. class PRServSingleton(object):
  228. def __init__(self, dbfile, logfile, interface):
  229. self.dbfile = dbfile
  230. self.logfile = logfile
  231. self.interface = interface
  232. self.host = None
  233. self.port = None
  234. def start(self):
  235. self.prserv = PRServer(self.dbfile, self.logfile, self.interface, daemon=False)
  236. self.prserv.start()
  237. self.host, self.port = self.prserv.getinfo()
  238. def getinfo(self):
  239. return (self.host, self.port)
  240. class PRServerConnection(object):
  241. def __init__(self, host, port):
  242. if is_local_special(host, port):
  243. host, port = singleton.getinfo()
  244. self.host = host
  245. self.port = port
  246. self.connection, self.transport = bb.server.xmlrpc._create_server(self.host, self.port)
  247. def terminate(self):
  248. try:
  249. logger.info("Terminating PRServer...")
  250. self.connection.quit()
  251. except Exception as exc:
  252. sys.stderr.write("%s\n" % str(exc))
  253. def getPR(self, version, pkgarch, checksum):
  254. return self.connection.getPR(version, pkgarch, checksum)
  255. def ping(self):
  256. return self.connection.ping()
  257. def export(self,version=None, pkgarch=None, checksum=None, colinfo=True):
  258. return self.connection.export(version, pkgarch, checksum, colinfo)
  259. def dump_db(self):
  260. return self.connection.dump_db()
  261. def importone(self, version, pkgarch, checksum, value):
  262. return self.connection.importone(version, pkgarch, checksum, value)
  263. def getinfo(self):
  264. return self.host, self.port
  265. def start_daemon(dbfile, host, port, logfile):
  266. ip = socket.gethostbyname(host)
  267. pidfile = PIDPREFIX % (ip, port)
  268. try:
  269. pf = open(pidfile,'r')
  270. pid = int(pf.readline().strip())
  271. pf.close()
  272. except IOError:
  273. pid = None
  274. if pid:
  275. sys.stderr.write("pidfile %s already exist. Daemon already running?\n"
  276. % pidfile)
  277. return 1
  278. server = PRServer(os.path.abspath(dbfile), os.path.abspath(logfile), (ip,port))
  279. server.start()
  280. # Sometimes, the port (i.e. localhost:0) indicated by the user does not match with
  281. # the one the server actually is listening, so at least warn the user about it
  282. _,rport = server.getinfo()
  283. if port != rport:
  284. sys.stdout.write("Server is listening at port %s instead of %s\n"
  285. % (rport,port))
  286. return 0
  287. def stop_daemon(host, port):
  288. import glob
  289. ip = socket.gethostbyname(host)
  290. pidfile = PIDPREFIX % (ip, port)
  291. try:
  292. pf = open(pidfile,'r')
  293. pid = int(pf.readline().strip())
  294. pf.close()
  295. except IOError:
  296. pid = None
  297. if not pid:
  298. # when server starts at port=0 (i.e. localhost:0), server actually takes another port,
  299. # so at least advise the user which ports the corresponding server is listening
  300. ports = []
  301. portstr = ""
  302. for pf in glob.glob(PIDPREFIX % (ip,'*')):
  303. bn = os.path.basename(pf)
  304. root, _ = os.path.splitext(bn)
  305. ports.append(root.split('_')[-1])
  306. if len(ports):
  307. portstr = "Wrong port? Other ports listening at %s: %s" % (host, ' '.join(ports))
  308. sys.stderr.write("pidfile %s does not exist. Daemon not running? %s\n"
  309. % (pidfile,portstr))
  310. return 1
  311. try:
  312. PRServerConnection(ip, port).terminate()
  313. except:
  314. logger.critical("Stop PRService %s:%d failed" % (host,port))
  315. try:
  316. if pid:
  317. wait_timeout = 0
  318. print("Waiting for pr-server to exit.")
  319. while is_running(pid) and wait_timeout < 50:
  320. time.sleep(0.1)
  321. wait_timeout += 1
  322. if is_running(pid):
  323. print("Sending SIGTERM to pr-server.")
  324. os.kill(pid,signal.SIGTERM)
  325. time.sleep(0.1)
  326. if os.path.exists(pidfile):
  327. os.remove(pidfile)
  328. except OSError as e:
  329. err = str(e)
  330. if err.find("No such process") <= 0:
  331. raise e
  332. return 0
  333. def is_running(pid):
  334. try:
  335. os.kill(pid, 0)
  336. except OSError as err:
  337. if err.errno == errno.ESRCH:
  338. return False
  339. return True
  340. def is_local_special(host, port):
  341. if host.strip().upper() == 'localhost'.upper() and (not port):
  342. return True
  343. else:
  344. return False
  345. class PRServiceConfigError(Exception):
  346. pass
  347. def auto_start(d):
  348. global singleton
  349. host_params = list(filter(None, (d.getVar('PRSERV_HOST', True) or '').split(':')))
  350. if not host_params:
  351. return None
  352. if len(host_params) != 2:
  353. logger.critical('\n'.join(['PRSERV_HOST: incorrect format',
  354. 'Usage: PRSERV_HOST = "<hostname>:<port>"']))
  355. raise PRServiceConfigError
  356. if is_local_special(host_params[0], int(host_params[1])) and not singleton:
  357. import bb.utils
  358. cachedir = (d.getVar("PERSISTENT_DIR", True) or d.getVar("CACHE", True))
  359. if not cachedir:
  360. logger.critical("Please set the 'PERSISTENT_DIR' or 'CACHE' variable")
  361. raise PRServiceConfigError
  362. bb.utils.mkdirhier(cachedir)
  363. dbfile = os.path.join(cachedir, "prserv.sqlite3")
  364. logfile = os.path.join(cachedir, "prserv.log")
  365. singleton = PRServSingleton(os.path.abspath(dbfile), os.path.abspath(logfile), ("localhost",0))
  366. singleton.start()
  367. if singleton:
  368. host, port = singleton.getinfo()
  369. else:
  370. host = host_params[0]
  371. port = int(host_params[1])
  372. try:
  373. connection = PRServerConnection(host,port)
  374. connection.ping()
  375. realhost, realport = connection.getinfo()
  376. return str(realhost) + ":" + str(realport)
  377. except Exception:
  378. logger.critical("PRservice %s:%d not available" % (host, port))
  379. raise PRServiceConfigError
  380. def auto_shutdown(d=None):
  381. global singleton
  382. if singleton:
  383. host, port = singleton.getinfo()
  384. try:
  385. PRServerConnection(host, port).terminate()
  386. except:
  387. logger.critical("Stop PRService %s:%d failed" % (host,port))
  388. singleton = None
  389. def ping(host, port):
  390. conn=PRServerConnection(host, port)
  391. return conn.ping()