qmemmand.py 9.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298
  1. # pylint: skip-file
  2. #
  3. # The Qubes OS Project, http://www.qubes-os.org
  4. #
  5. # Copyright (C) 2010 Rafal Wojtczuk <rafal@invisiblethingslab.com>
  6. #
  7. # This program is free software; you can redistribute it and/or
  8. # modify it under the terms of the GNU General Public License
  9. # as published by the Free Software Foundation; either version 2
  10. # of the License, or (at your option) any later version.
  11. #
  12. # This program is distributed in the hope that it will be useful,
  13. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  15. # GNU General Public License for more details.
  16. #
  17. # You should have received a copy of the GNU General Public License
  18. # along with this program; if not, write to the Free Software
  19. # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
  20. #
  21. #
  22. import configparser
  23. import socketserver
  24. import logging
  25. import logging.handlers
  26. import os
  27. import socket
  28. import sys
  29. import threading
  30. import xen.lowlevel.xs
  31. import qubes.qmemman
  32. import qubes.qmemman.algo
  33. import qubes.utils
  34. SOCK_PATH = '/var/run/qubes/qmemman.sock'
  35. LOG_PATH = '/var/log/qubes/qmemman.log'
  36. system_state = qubes.qmemman.SystemState()
  37. global_lock = threading.Lock()
  38. # If XS_Watcher will
  39. # handle meminfo event before @introduceDomain, it will use
  40. # incomplete domain list for that and may redistribute memory
  41. # allocated to some VM, but not yet used (see #1389).
  42. # To fix that, system_state should be updated (refresh domain
  43. # list) before processing other changes, every time some process requested
  44. # memory for a new VM, before releasing the lock. Then XS_Watcher will check
  45. # this flag before processing other event.
  46. force_refresh_domain_list = False
  47. def only_in_first_list(l1, l2):
  48. ret = []
  49. for i in l1:
  50. if not i in l2:
  51. ret.append(i)
  52. return ret
  53. def get_domain_meminfo_key(domain_id):
  54. return '/local/domain/'+domain_id+'/memory/meminfo'
  55. class WatchType(object):
  56. def __init__(self, fn, param):
  57. self.fn = fn
  58. self.param = param
  59. class XS_Watcher(object):
  60. def __init__(self):
  61. self.log = logging.getLogger('qmemman.daemon.xswatcher')
  62. self.log.debug('XS_Watcher()')
  63. self.handle = xen.lowlevel.xs.xs()
  64. self.handle.watch('@introduceDomain', WatchType(
  65. XS_Watcher.domain_list_changed, False))
  66. self.handle.watch('@releaseDomain', WatchType(
  67. XS_Watcher.domain_list_changed, False))
  68. self.watch_token_dict = {}
  69. def domain_list_changed(self, refresh_only=False):
  70. """
  71. Check if any domain was created/destroyed. If it was, update
  72. appropriate list. Then redistribute memory.
  73. :param refresh_only If True, only refresh domain list, do not
  74. redistribute memory. In this mode, caller must already hold
  75. global_lock.
  76. """
  77. self.log.debug('domain_list_changed(only_refresh={!r})'.format(
  78. refresh_only))
  79. got_lock = False
  80. if not refresh_only:
  81. self.log.debug('acquiring global_lock')
  82. global_lock.acquire()
  83. got_lock = True
  84. self.log.debug('global_lock acquired')
  85. try:
  86. curr = self.handle.ls('', '/local/domain')
  87. if curr is None:
  88. return
  89. # check if domain is really there, it may happen that some empty
  90. # directories are left in xenstore
  91. curr = list(filter(
  92. lambda x:
  93. self.handle.read('',
  94. '/local/domain/{}/domid'.format(x)
  95. ) is not None,
  96. curr
  97. ))
  98. self.log.debug('curr={!r}'.format(curr))
  99. for i in only_in_first_list(curr, self.watch_token_dict.keys()):
  100. # new domain has been created
  101. watch = WatchType(XS_Watcher.meminfo_changed, i)
  102. self.watch_token_dict[i] = watch
  103. self.handle.watch(get_domain_meminfo_key(i), watch)
  104. system_state.add_domain(i)
  105. for i in only_in_first_list(self.watch_token_dict.keys(), curr):
  106. # domain destroyed
  107. self.handle.unwatch(get_domain_meminfo_key(i), self.watch_token_dict[i])
  108. self.watch_token_dict.pop(i)
  109. system_state.del_domain(i)
  110. finally:
  111. if got_lock:
  112. global_lock.release()
  113. self.log.debug('global_lock released')
  114. if not refresh_only:
  115. system_state.do_balance()
  116. def meminfo_changed(self, domain_id):
  117. self.log.debug('meminfo_changed(domain_id={!r})'.format(domain_id))
  118. untrusted_meminfo_key = self.handle.read(
  119. '', get_domain_meminfo_key(domain_id))
  120. if untrusted_meminfo_key == None or untrusted_meminfo_key == '':
  121. return
  122. self.log.debug('acquiring global_lock')
  123. global_lock.acquire()
  124. self.log.debug('global_lock acquired')
  125. if force_refresh_domain_list:
  126. self.domain_list_changed(refresh_only=True)
  127. system_state.refresh_meminfo(domain_id, untrusted_meminfo_key)
  128. global_lock.release()
  129. self.log.debug('global_lock released')
  130. def watch_loop(self):
  131. self.log.debug('watch_loop()')
  132. while True:
  133. result = self.handle.read_watch()
  134. self.log.debug('watch_loop result={!r}'.format(result))
  135. token = result[1]
  136. token.fn(self, token.param)
  137. class QMemmanReqHandler(socketserver.BaseRequestHandler):
  138. """
  139. The RequestHandler class for our server.
  140. It is instantiated once per connection to the server, and must
  141. override the handle() method to implement communication to the
  142. client.
  143. """
  144. def handle(self):
  145. self.log = logging.getLogger('qmemman.daemon.reqhandler')
  146. got_lock = False
  147. try:
  148. # self.request is the TCP socket connected to the client
  149. while True:
  150. self.data = self.request.recv(1024).strip()
  151. self.log.debug('data={!r}'.format(self.data))
  152. if len(self.data) == 0:
  153. self.log.info('EOF')
  154. if got_lock:
  155. global force_refresh_domain_list
  156. force_refresh_domain_list = True
  157. return
  158. # XXX something is wrong here: return without release?
  159. if got_lock:
  160. self.log.warning('Second request over qmemman.sock?')
  161. return
  162. self.log.debug('acquiring global_lock')
  163. global_lock.acquire()
  164. self.log.debug('global_lock acquired')
  165. got_lock = True
  166. if system_state.do_balloon(int(self.data.decode('ascii'))):
  167. resp = b"OK\n"
  168. else:
  169. resp = b"FAIL\n"
  170. self.log.debug('resp={!r}'.format(resp))
  171. self.request.send(resp)
  172. except BaseException as e:
  173. self.log.exception(
  174. "exception while handling request: {!r}".format(e))
  175. finally:
  176. if got_lock:
  177. global_lock.release()
  178. self.log.debug('global_lock released')
  179. parser = qubes.tools.QubesArgumentParser(want_app=False)
  180. parser.add_argument('--config', '-c', metavar='FILE',
  181. action='store', default='/etc/qubes/qmemman.conf',
  182. help='qmemman config file')
  183. parser.add_argument('--foreground',
  184. action='store_true', default=False,
  185. help='do not close stdio')
  186. def main():
  187. args = parser.parse_args()
  188. # setup logging
  189. ha_syslog = logging.handlers.SysLogHandler('/dev/log')
  190. ha_syslog.setFormatter(
  191. logging.Formatter('%(name)s[%(process)d]: %(message)s'))
  192. logging.root.addHandler(ha_syslog)
  193. # leave log for backwards compatibility
  194. ha_file = logging.FileHandler(LOG_PATH)
  195. ha_file.setFormatter(
  196. logging.Formatter('%(asctime)s %(name)s[%(process)d]: %(message)s'))
  197. logging.root.addHandler(ha_file)
  198. if args.foreground:
  199. ha_stderr = logging.StreamHandler(sys.stderr)
  200. ha_file.setFormatter(
  201. logging.Formatter('%(asctime)s %(name)s[%(process)d]: %(message)s'))
  202. logging.root.addHandler(ha_stderr)
  203. else:
  204. # close io
  205. sys.stdout.close()
  206. sys.stderr.close()
  207. sys.stdin.close()
  208. logging.root.setLevel(parser.get_loglevel_from_verbosity(args))
  209. log = logging.getLogger('qmemman.daemon')
  210. config = configparser.SafeConfigParser({
  211. 'vm-min-mem': str(qubes.qmemman.algo.MIN_PREFMEM),
  212. 'dom0-mem-boost': str(qubes.qmemman.algo.DOM0_MEM_BOOST),
  213. 'cache-margin-factor': str(qubes.qmemman.algo.CACHE_FACTOR)
  214. })
  215. config.read(args.config)
  216. if config.has_section('global'):
  217. qubes.qmemman.algo.MIN_PREFMEM = \
  218. qubes.utils.parse_size(config.get('global', 'vm-min-mem'))
  219. qubes.qmemman.algo.DOM0_MEM_BOOST = \
  220. qubes.utils.parse_size(config.get('global', 'dom0-mem-boost'))
  221. qubes.qmemman.algo.CACHE_FACTOR = \
  222. config.getfloat('global', 'cache-margin-factor')
  223. log.info('MIN_PREFMEM={algo.MIN_PREFMEM}'
  224. ' DOM0_MEM_BOOST={algo.DOM0_MEM_BOOST}'
  225. ' CACHE_FACTOR={algo.CACHE_FACTOR}'.format(
  226. algo=qubes.qmemman.algo))
  227. try:
  228. os.unlink(SOCK_PATH)
  229. except:
  230. pass
  231. log.debug('instantiating server')
  232. os.umask(0)
  233. server = socketserver.UnixStreamServer(SOCK_PATH, QMemmanReqHandler)
  234. os.umask(0o077)
  235. # notify systemd
  236. nofity_socket = os.getenv('NOTIFY_SOCKET')
  237. if nofity_socket:
  238. log.debug('notifying systemd')
  239. s = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
  240. if nofity_socket.startswith('@'):
  241. nofity_socket = '\0%s' % nofity_socket[1:]
  242. s.connect(nofity_socket)
  243. s.sendall(b"READY=1")
  244. s.close()
  245. threading.Thread(target=server.serve_forever).start()
  246. XS_Watcher().watch_loop()