Performance improvements for HostVars and some bugfixes

lib/ansible/executor/process/result.py

@@ -58,7 +58,7 @@ class ResultProcess(multiprocessing.Process):
 
     def _send_result(self, result):
         debug(u"sending result: %s" % ([text_type(x) for x in result],))
-        self._final_q.put(result, block=False)
+        self._final_q.put(result)
         debug("done sending result")
 
     def _read_worker_result(self):
@@ -73,7 +73,7 @@ class ResultProcess(multiprocessing.Process):
             try:
                 if not rslt_q.empty():
                     debug("worker %d has data to read" % self._cur_worker)
-                    result = rslt_q.get(block=False)
+                    result = rslt_q.get()
                     debug("got a result from worker %d: %s" % (self._cur_worker, result))
                     break
             except queue.Empty:
@@ -101,7 +101,7 @@ class ResultProcess(multiprocessing.Process):
             try:
                 result = self._read_worker_result()
                 if result is None:
-                    time.sleep(0.01)
+                    time.sleep(0.0001)
                     continue
 
                 clean_copy = strip_internal_keys(result._result)

lib/ansible/executor/process/worker.py

@@ -59,11 +59,13 @@ class WorkerProcess(multiprocessing.Process):
     for reading later.
     '''
 
-    def __init__(self, tqm, main_q, rslt_q, loader):
+    def __init__(self, tqm, main_q, rslt_q, hostvars_manager, loader):
 
+        super(WorkerProcess, self).__init__()
         # takes a task queue manager as the sole param:
         self._main_q   = main_q
         self._rslt_q   = rslt_q
+        self._hostvars = hostvars_manager
         self._loader   = loader
 
         # dupe stdin, if we have one
@@ -82,8 +84,6 @@ class WorkerProcess(multiprocessing.Process):
             # couldn't get stdin's fileno, so we just carry on
             pass
 
-        super(WorkerProcess, self).__init__()
-
     def run(self):
         '''
         Called when the process is started, and loops indefinitely
@@ -100,14 +100,15 @@ class WorkerProcess(multiprocessing.Process):
         while True:
             task = None
             try:
-                debug("waiting for a message...")
-                (host, task, basedir, zip_vars, hostvars, compressed_vars, play_context, shared_loader_obj) = self._main_q.get()
+                #debug("waiting for work")
+                (host, task, basedir, zip_vars, compressed_vars, play_context, shared_loader_obj) = self._main_q.get(block=False)
 
                 if compressed_vars:
                     job_vars = json.loads(zlib.decompress(zip_vars))
                 else:
                     job_vars = zip_vars
-                job_vars['hostvars'] = hostvars
+
+                job_vars['hostvars'] = self._hostvars.hostvars()
 
                 debug("there's work to be done! got a task/handler to work on: %s" % task)
 
@@ -142,7 +143,7 @@ class WorkerProcess(multiprocessing.Process):
                 debug("done sending task result")
 
             except queue.Empty:
-                pass
+                time.sleep(0.0001)
             except AnsibleConnectionFailure:
                 try:
                     if task:

lib/ansible/executor/task_queue_manager.py

@@ -19,6 +19,7 @@
 from __future__ import (absolute_import, division, print_function)
 __metaclass__ = type
 
+from multiprocessing.managers import SyncManager, DictProxy
 import multiprocessing
 import os
 import tempfile
@@ -32,6 +33,7 @@ from ansible.executor.stats import AggregateStats
 from ansible.playbook.play_context import PlayContext
 from ansible.plugins import callback_loader, strategy_loader, module_loader
 from ansible.template import Templar
+from ansible.vars.hostvars import HostVars
 
 try:
     from __main__ import display
@@ -98,7 +100,7 @@ class TaskQueueManager:
             main_q = multiprocessing.Queue()
             rslt_q = multiprocessing.Queue()
 
-            prc = WorkerProcess(self, main_q, rslt_q, self._loader)
+            prc = WorkerProcess(self, main_q, rslt_q, self._hostvars_manager, self._loader)
             prc.start()
 
             self._workers.append((prc, main_q, rslt_q))
@@ -173,11 +175,6 @@ class TaskQueueManager:
         are done with the current task).
         '''
 
-        # Fork # of forks, # of hosts or serial, whichever is lowest
-        contenders =  [self._options.forks, play.serial, len(self._inventory.get_hosts(play.hosts))]
-        contenders =  [ v for v in contenders if v is not None and v > 0 ]
-        self._initialize_processes(min(contenders))
-
         if not self._callbacks_loaded:
             self.load_callbacks()
 
@@ -187,6 +184,34 @@ class TaskQueueManager:
         new_play = play.copy()
         new_play.post_validate(templar)
 
+        class HostVarsManager(SyncManager):
+            pass
+
+        hostvars = HostVars(
+            play=new_play,
+            inventory=self._inventory,
+            variable_manager=self._variable_manager,
+            loader=self._loader,
+        )
+
+        HostVarsManager.register(
+            'hostvars',
+            callable=lambda: hostvars,
+            # FIXME: this is the list of exposed methods to the DictProxy object, plus our
+            #        one special one (set_variable_manager). There's probably a better way
+            #        to do this with a proper BaseProxy/DictProxy derivative
+            exposed=('set_variable_manager', '__contains__', '__delitem__', '__getitem__',
+                     '__len__', '__setitem__', 'clear', 'copy', 'get', 'has_key', 'items',
+                     'keys', 'pop', 'popitem', 'setdefault', 'update', 'values'),
+        )
+        self._hostvars_manager = HostVarsManager()
+        self._hostvars_manager.start()
+
+        # Fork # of forks, # of hosts or serial, whichever is lowest
+        contenders =  [self._options.forks, play.serial, len(self._inventory.get_hosts(new_play.hosts))]
+        contenders =  [ v for v in contenders if v is not None and v > 0 ]
+        self._initialize_processes(min(contenders))
+
         play_context = PlayContext(new_play, self._options, self.passwords, self._connection_lockfile.fileno())
         for callback_plugin in self._callback_plugins:
             if hasattr(callback_plugin, 'set_play_context'):
@@ -221,6 +246,7 @@ class TaskQueueManager:
         # and run the play using the strategy and cleanup on way out
         play_return = strategy.run(iterator, play_context)
         self._cleanup_processes()
+        self._hostvars_manager.shutdown()
         return play_return
 
     def cleanup(self):