address filehandle/event leak in async run_job invocations

extending on the idea used in saltstack#32145, when _check_pub_data is called it it will
create jid subscriptions, regardless of whether anyone will ever come back to
retrieve them; in the case of local_async calls noone ever does.
In addition to the above, we use the listen kwarg provided by c59a5ad to
know whether we need to subscribe to events in addition to ensuring the ioloop
is listening before a call is made.
This should fix saltstack#40245, saltstack#20639, saltstack#36374

salt/client/__init__.py

@@ -248,7 +248,7 @@ def gather_job_info(self, jid, tgt, tgt_type, **kwargs):
 
         return pub_data
 
-    def _check_pub_data(self, pub_data):
+    def _check_pub_data(self, pub_data, listen=True):
         '''
         Common checks on the pub_data data structure returned from running pub
         '''
@@ -281,7 +281,13 @@ def _check_pub_data(self, pub_data):
                 print('No minions matched the target. '
                       'No command was sent, no jid was assigned.')
                 return {}
-        else:
+
+        # don't install event subscription listeners when the request is async
+        # and doesn't care. this is important as it will create event leaks otherwise
+        if not listen:
+            return pub_data
+
+        if self.opts.get('order_masters'):
             self.event.subscribe('syndic/.*/{0}'.format(pub_data['jid']), 'regex')
 
         self.event.subscribe('salt/job/{0}'.format(pub_data['jid']))
@@ -336,7 +342,7 @@ def run_job(
             # Convert to generic client error and pass along message
             raise SaltClientError(general_exception)
 
-        return self._check_pub_data(pub_data)
+        return self._check_pub_data(pub_data, listen=listen)
 
     def gather_minions(self, tgt, expr_form):
         _res = salt.utils.minions.CkMinions(self.opts).check_minions(tgt, tgt_type=expr_form)
@@ -393,7 +399,7 @@ def run_job_async(
             # Convert to generic client error and pass along message
             raise SaltClientError(general_exception)
 
-        raise tornado.gen.Return(self._check_pub_data(pub_data))
+        raise tornado.gen.Return(self._check_pub_data(pub_data, listen=listen))
 
     def cmd_async(
             self,
@@ -425,6 +431,7 @@ def cmd_async(
                                 tgt_type,
                                 ret,
                                 jid=jid,
+                                listen=False,
                                 **kwargs)
         try:
             return pub_data['jid']