Make async jid's unique per host

The jid will now also contain the PID of the async_wrapper process,
and can each unique jid from each host is tracked rather than just
relying on one global jid per task.

Fixes #5582

lib/ansible/runner/poller.py

@@ -30,18 +30,21 @@ class AsyncPoller(object):
         self.hosts_to_poll = []
         self.completed = False
 
-        # Get job id and which hosts to poll again in the future
-        jid = None
+        # flag to determine if at least one host was contacted
+        self.active = False
         # True to work with & below
         skipped = True
         for (host, res) in results['contacted'].iteritems():
             if res.get('started', False):
                 self.hosts_to_poll.append(host)
                 jid = res.get('ansible_job_id', None)
+                self.runner.setup_cache[host]['ansible_job_id'] = jid
+                self.active = True
             else:
                 skipped = skipped & res.get('skipped', False)
                 self.results['contacted'][host] = res
         for (host, res) in results['dark'].iteritems():
+            self.runner.setup_cache[host]['ansible_job_id'] = ''
             self.results['dark'][host] = res
 
         if not skipped:
@@ -49,14 +52,13 @@ class AsyncPoller(object):
                 raise errors.AnsibleError("unexpected error: unable to determine jid")
             if len(self.hosts_to_poll)==0:
                 raise errors.AnsibleError("unexpected error: no hosts to poll")
-        self.jid = jid
 
     def poll(self):
         """ Poll the job status.
 
             Returns the changes in this iteration."""
         self.runner.module_name = 'async_status'
-        self.runner.module_args = "jid=%s" % self.jid
+        self.runner.module_args = "jid={{ansible_job_id}}"
         self.runner.pattern = "*"
         self.runner.background = 0
         self.runner.complex_args = None
@@ -75,13 +77,14 @@ class AsyncPoller(object):
                 self.results['contacted'][host] = res
                 poll_results['contacted'][host] = res
                 if res.get('failed', False) or res.get('rc', 0) != 0:
-                    self.runner.callbacks.on_async_failed(host, res, self.jid)
+                    self.runner.callbacks.on_async_failed(host, res, self.runner.setup_cache[host]['ansible_job_id'])
                 else:
-                    self.runner.callbacks.on_async_ok(host, res, self.jid)
+                    self.runner.callbacks.on_async_ok(host, res, self.runner.setup_cache[host]['ansible_job_id'])
         for (host, res) in results['dark'].iteritems():
             self.results['dark'][host] = res
             poll_results['dark'][host] = res
-            self.runner.callbacks.on_async_failed(host, res, self.jid)
+            if host in self.hosts_to_poll:
+                self.runner.callbacks.on_async_failed(host, res, self.runner.setup_cache[host].get('ansible_job_id','XX'))
 
         self.hosts_to_poll = hosts
         if len(hosts)==0:
@@ -92,7 +95,7 @@ class AsyncPoller(object):
     def wait(self, seconds, poll_interval):
         """ Wait a certain time for job completion, check status every poll_interval. """
         # jid is None when all hosts were skipped
-        if self.jid is None:
+        if not self.active:
             return self.results
 
         clock = seconds - poll_interval
@@ -103,7 +106,7 @@ class AsyncPoller(object):
 
             for (host, res) in poll_results['polled'].iteritems():
                 if res.get('started'):
-                    self.runner.callbacks.on_async_poll(host, res, self.jid, clock)
+                    self.runner.callbacks.on_async_poll(host, res, self.runner.setup_cache[host]['ansible_job_id'], clock)
 
             clock = clock - poll_interval
 

library/internal/async_wrapper

@@ -72,7 +72,7 @@ if len(sys.argv) < 3:
     })
     sys.exit(1)
 
-jid = sys.argv[1]
+jid = "%s.%d" % (sys.argv[1], os.getpid())
 time_limit = sys.argv[2]
 wrapped_module = sys.argv[3]
 argsfile = sys.argv[4]