Fixes to work with pyzmq with --enable-drafts

1. Let pyzmq to correctly cleanup stream by itself

2. Don't close IOLoop that closes managed IO objects until full async
object cleanup that produces closing streams and attempts to operate to
strutures deinitialized by the closed IOLoop.

salt/transport/zeromq.py

@@ -359,9 +359,7 @@ class AsyncZeroMQPubChannel(salt.transport.mixins.auth.AESPubClientMixin, salt.t
             self._monitor.stop()
             self._monitor = None
         if hasattr(self, '_stream'):
-            # TODO: Optionally call stream.close() on newer pyzmq? Its broken on some
-            self._stream.io_loop.remove_handler(self._stream.socket)
-            self._stream.socket.close(0)
+            self._stream.close(0)
         elif hasattr(self, '_socket'):
             self._socket.close(0)
         if hasattr(self, 'context') and self.context.closed is False:
@@ -913,14 +911,9 @@ class AsyncReqMessageClient(object):
     # TODO: timeout all in-flight sessions, or error
     def destroy(self):
         if hasattr(self, 'stream') and self.stream is not None:
-            # TODO: Optionally call stream.close() on newer pyzmq? It is broken on some.
-            if self.stream.socket:
-                self.stream.socket.close()
-            self.stream.io_loop.remove_handler(self.stream.socket)
-            # set this to None, more hacks for messed up pyzmq
-            self.stream.socket = None
+            self.stream.close()
+            self.socket = None
             self.stream = None
-            self.socket.close()
         if self.context.closed is False:
             self.context.term()
 

salt/utils/async.py

@@ -94,10 +94,10 @@ class SyncWrapper(object):
                 # their associated io_loop is closed to allow for proper
                 # cleanup.
                 self.async.close()
-            self.io_loop.close()
             # Other things should be deallocated after the io_loop closes.
             # See Issue #26889.
             del self.async
+            self.io_loop.close()
             del self.io_loop
         elif hasattr(self, 'io_loop'):
             self.io_loop.close()