process.py 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268
  1. #
  2. # BitBake Process based server.
  3. #
  4. # Copyright (C) 2010 Bob Foerster <robert@erafx.com>
  5. #
  6. # This program is free software; you can redistribute it and/or modify
  7. # it under the terms of the GNU General Public License version 2 as
  8. # published by the Free Software Foundation.
  9. #
  10. # This program is distributed in the hope that it will be useful,
  11. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. # GNU General Public License for more details.
  14. #
  15. # You should have received a copy of the GNU General Public License along
  16. # with this program; if not, write to the Free Software Foundation, Inc.,
  17. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18. """
  19. This module implements a multiprocessing.Process based server for bitbake.
  20. """
  21. import bb
  22. import bb.event
  23. import itertools
  24. import logging
  25. import multiprocessing
  26. import os
  27. import signal
  28. import sys
  29. import time
  30. import select
  31. from queue import Empty
  32. from multiprocessing import Event, Process, util, Queue, Pipe, queues, Manager
  33. from . import BitBakeBaseServer, BitBakeBaseServerConnection, BaseImplServer
  34. logger = logging.getLogger('BitBake')
  35. class ServerCommunicator():
  36. def __init__(self, connection, event_handle, server):
  37. self.connection = connection
  38. self.event_handle = event_handle
  39. self.server = server
  40. def runCommand(self, command):
  41. # @todo try/except
  42. self.connection.send(command)
  43. if not self.server.is_alive():
  44. raise SystemExit
  45. while True:
  46. # don't let the user ctrl-c while we're waiting for a response
  47. try:
  48. for idx in range(0,4): # 0, 1, 2, 3
  49. if self.connection.poll(5):
  50. return self.connection.recv()
  51. else:
  52. bb.warn("Timeout while attempting to communicate with bitbake server")
  53. bb.fatal("Gave up; Too many tries: timeout while attempting to communicate with bitbake server")
  54. except KeyboardInterrupt:
  55. pass
  56. def getEventHandle(self):
  57. return self.event_handle.value
  58. class EventAdapter():
  59. """
  60. Adapter to wrap our event queue since the caller (bb.event) expects to
  61. call a send() method, but our actual queue only has put()
  62. """
  63. def __init__(self, queue):
  64. self.queue = queue
  65. def send(self, event):
  66. try:
  67. self.queue.put(event)
  68. except Exception as err:
  69. print("EventAdapter puked: %s" % str(err))
  70. class ProcessServer(Process, BaseImplServer):
  71. profile_filename = "profile.log"
  72. profile_processed_filename = "profile.log.processed"
  73. def __init__(self, command_channel, event_queue, featurelist):
  74. BaseImplServer.__init__(self)
  75. Process.__init__(self)
  76. self.command_channel = command_channel
  77. self.event_queue = event_queue
  78. self.event = EventAdapter(event_queue)
  79. self.featurelist = featurelist
  80. self.quit = False
  81. self.quitin, self.quitout = Pipe()
  82. self.event_handle = multiprocessing.Value("i")
  83. def run(self):
  84. for event in bb.event.ui_queue:
  85. self.event_queue.put(event)
  86. self.event_handle.value = bb.event.register_UIHhandler(self, True)
  87. bb.cooker.server_main(self.cooker, self.main)
  88. def main(self):
  89. # Ignore SIGINT within the server, as all SIGINT handling is done by
  90. # the UI and communicated to us
  91. self.quitin.close()
  92. signal.signal(signal.SIGINT, signal.SIG_IGN)
  93. bb.utils.set_process_name("Cooker")
  94. while not self.quit:
  95. try:
  96. if self.command_channel.poll():
  97. command = self.command_channel.recv()
  98. self.runCommand(command)
  99. if self.quitout.poll():
  100. self.quitout.recv()
  101. self.quit = True
  102. try:
  103. self.runCommand(["stateForceShutdown"])
  104. except:
  105. pass
  106. self.idle_commands(.1, [self.command_channel, self.quitout])
  107. except Exception:
  108. logger.exception('Running command %s', command)
  109. self.event_queue.close()
  110. bb.event.unregister_UIHhandler(self.event_handle.value)
  111. self.command_channel.close()
  112. self.cooker.shutdown(True)
  113. self.quitout.close()
  114. def idle_commands(self, delay, fds=None):
  115. nextsleep = delay
  116. if not fds:
  117. fds = []
  118. for function, data in list(self._idlefuns.items()):
  119. try:
  120. retval = function(self, data, False)
  121. if retval is False:
  122. del self._idlefuns[function]
  123. nextsleep = None
  124. elif retval is True:
  125. nextsleep = None
  126. elif isinstance(retval, float) and nextsleep:
  127. if (retval < nextsleep):
  128. nextsleep = retval
  129. elif nextsleep is None:
  130. continue
  131. else:
  132. fds = fds + retval
  133. except SystemExit:
  134. raise
  135. except Exception as exc:
  136. if not isinstance(exc, bb.BBHandledException):
  137. logger.exception('Running idle function')
  138. del self._idlefuns[function]
  139. self.quit = True
  140. if nextsleep is not None:
  141. select.select(fds,[],[],nextsleep)
  142. def runCommand(self, command):
  143. """
  144. Run a cooker command on the server
  145. """
  146. self.command_channel.send(self.cooker.command.runCommand(command))
  147. def stop(self):
  148. self.quitin.send("quit")
  149. self.quitin.close()
  150. class BitBakeProcessServerConnection(BitBakeBaseServerConnection):
  151. def __init__(self, serverImpl, ui_channel, event_queue):
  152. self.procserver = serverImpl
  153. self.ui_channel = ui_channel
  154. self.event_queue = event_queue
  155. self.connection = ServerCommunicator(self.ui_channel, self.procserver.event_handle, self.procserver)
  156. self.events = self.event_queue
  157. self.terminated = False
  158. def sigterm_terminate(self):
  159. bb.error("UI received SIGTERM")
  160. self.terminate()
  161. def terminate(self):
  162. if self.terminated:
  163. return
  164. self.terminated = True
  165. def flushevents():
  166. while True:
  167. try:
  168. event = self.event_queue.get(block=False)
  169. except (Empty, IOError):
  170. break
  171. if isinstance(event, logging.LogRecord):
  172. logger.handle(event)
  173. signal.signal(signal.SIGINT, signal.SIG_IGN)
  174. self.procserver.stop()
  175. while self.procserver.is_alive():
  176. flushevents()
  177. self.procserver.join(0.1)
  178. self.ui_channel.close()
  179. self.event_queue.close()
  180. self.event_queue.setexit()
  181. # Wrap Queue to provide API which isn't server implementation specific
  182. class ProcessEventQueue(multiprocessing.queues.Queue):
  183. def __init__(self, maxsize):
  184. multiprocessing.queues.Queue.__init__(self, maxsize, ctx=multiprocessing.get_context())
  185. self.exit = False
  186. bb.utils.set_process_name("ProcessEQueue")
  187. def setexit(self):
  188. self.exit = True
  189. def waitEvent(self, timeout):
  190. if self.exit:
  191. return self.getEvent()
  192. try:
  193. if not self.server.is_alive():
  194. return self.getEvent()
  195. return self.get(True, timeout)
  196. except Empty:
  197. return None
  198. def getEvent(self):
  199. try:
  200. if not self.server.is_alive():
  201. self.setexit()
  202. return self.get(False)
  203. except Empty:
  204. if self.exit:
  205. sys.exit(1)
  206. return None
  207. class BitBakeServer(BitBakeBaseServer):
  208. def initServer(self, single_use=True):
  209. # establish communication channels. We use bidirectional pipes for
  210. # ui <--> server command/response pairs
  211. # and a queue for server -> ui event notifications
  212. #
  213. self.ui_channel, self.server_channel = Pipe()
  214. self.event_queue = ProcessEventQueue(0)
  215. self.serverImpl = ProcessServer(self.server_channel, self.event_queue, None)
  216. self.event_queue.server = self.serverImpl
  217. def detach(self):
  218. self.serverImpl.start()
  219. return
  220. def establishConnection(self, featureset):
  221. self.connection = BitBakeProcessServerConnection(self.serverImpl, self.ui_channel, self.event_queue)
  222. _, error = self.connection.connection.runCommand(["setFeatures", featureset])
  223. if error:
  224. logger.error("Unable to set the cooker to the correct featureset: %s" % error)
  225. raise BaseException(error)
  226. signal.signal(signal.SIGTERM, lambda i, s: self.connection.sigterm_terminate())
  227. return self.connection