Fix minion failover after disconnect

salt/minion.py:
- In the `__master_disconnected` case, it originally set
`self.opts['master']` to the result of `self.eval_master`. This
makes no sense since `self.eval_master` returns a future. Now we
yield on `self.eval_master` and store both return values.
- Just like in `Minion.destroy`, we clear `self.pub_channel.on_recv`.
Also, if `self.pub_channel` has a `close`, it is invoked. This
resolves a circular reference issue with the TCP transport. The
`SaltMessageClient` contains function references to the
`connect_callback` and `disconnect_callback` functions in
`AsyncTCPPubChannel`. This circular reference prevents the `del` from
deallocating the `AsyncTCPPubChannel` object.

salt/transport/tcp.py:
- In `SaltMessageClient.close`, reset the function references
`self.connect_callback` and `self.disconnect_callback`. This allows a
subsequent `del` on `AsyncTCPPubChannel` to actually deallocate the
object.

Signed-off-by: Sergey Kizunov <sergey.kizunov@ni.com>

salt/minion.py

@@ -1593,18 +1593,25 @@ class Minion(MinionBase):
                 if self.opts['master_type'] == 'failover':
                     log.info('Trying to tune in to next master from master-list')
 
+                    if hasattr(self, 'pub_channel'):
+                        self.pub_channel.on_recv(None)
+                        if hasattr(self.pub_channel, 'close'):
+                            self.pub_channel.close()
+                        del self.pub_channel
+
                     # if eval_master finds a new master for us, self.connected
                     # will be True again on successful master authentication
-                    self.opts['master'] = self.eval_master(opts=self.opts,
-                                                           failed=True)
+                    master, self.pub_channel = yield self.eval_master(
+                                                        opts=self.opts,
+                                                        failed=True)
                     if self.connected:
+                        self.opts['master'] = master
+
                         # re-init the subsystems to work with the new master
                         log.info('Re-initialising subsystems for new '
                                  'master {0}'.format(self.opts['master']))
-                        del self.pub_channel
-                        self._connect_master_future = self.connect_master()
-                        self.block_until_connected()  # TODO: remove
                         self.functions, self.returners, self.function_errors = self._load_modules()
+                        self.pub_channel.on_recv(self._handle_payload)
                         self._fire_master_minion_start()
                         log.info('Minion is ready to receive requests!')
 
@@ -1813,6 +1820,8 @@ class Minion(MinionBase):
         self._running = False
         if hasattr(self, 'pub_channel'):
             self.pub_channel.on_recv(None)
+            if hasattr(self.pub_channel, 'close'):
+                self.pub_channel.close()
             del self.pub_channel
         if hasattr(self, 'periodic_callbacks'):
             for cb in six.itervalues(self.periodic_callbacks):

salt/transport/tcp.py

@@ -426,6 +426,10 @@ class SaltMessageClient(object):
                 # 'StreamClosedError' when the stream is closed.
                 self._read_until_future.exc_info()
         self._tcp_client.close()
+        # Clear callback references to allow the object that they belong to
+        # to be deleted.
+        self.connect_callback = None
+        self.disconnect_callback = None
 
     def __del__(self):
         self.close()