transport: Wait for thread termination before closing the socket

Make sure the Thread.run() method has terminated before closing the
socket. Currently, the socket is closed through Packetizer.close(),
which happens too early. Move the socket.close() into Transport.close()
and after the Thread.join() call.

While at it, modify the stop_thread() method and use it in
Transport.close() to avoid code duplication. Use join() with a timeout
to make it possible to terminate the main thread with KeyboardInterrupt.
Also, remove the now obsolete socket.close() from Transport.atfork().

This fixes a potential infinite loop if paramiko.SSHClient is connected
through a paramiko.Channel instead of a regular socket (tunneling).

Details:

Using a debug patch to dump the current stack of the thread every
couple of seconds while trying to close it, I've seen the following
over and over again:

Thread could not be stopped, still running.
Current traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 524, in __bootstrap
    self.__bootstrap_inner()
  File "/usr/lib/python2.7/threading.py", line 551, in __bootstrap_inner
    self.run()
  File ".../paramiko/transport.py", line 1564, in run
    self._channel_handler_table[ptype](chan, m)
  File ".../paramiko/channel.py", line 1102, in _handle_close
    self.transport._send_user_message(m)
  File ".../paramiko/transport.py", line 1418, in _send_user_message
    self._send_message(data)
  File ".../paramiko/transport.py", line 1398, in _send_message
    self.packetizer.send_message(data)
  File ".../paramiko/packet.py", line 319, in send_message
    self.write_all(out)
  File ".../paramiko/packet.py", line 248, in write_all
    n = self.__socket.send(out)
  File ".../paramiko/channel.py", line 732, in send
    self.lock.release()

The thread was running Packetizer.write_all() in an endless loop:

while len(out) > 0:
    ...
    n = Channel.send(out) # n == 0 because channel got closed
    ...
    out = out[n:]         # essentially out = out

Signed-off-by: Frank Arnold <farnold@amazon.com>

paramiko/packet.py

@@ -156,7 +156,6 @@ class Packetizer (object):
 
     def close(self):
         self.__closed = True
-        self.__socket.close()
 
     def set_hexdump(self, hexdump):
         self.__dump_packets = hexdump

paramiko/transport.py

@@ -400,7 +400,6 @@ class Transport (threading.Thread):
 
         @since: 1.5.3
         """
-        self.sock.close()
         self.close()
 
     def get_security_options(self):
@@ -614,11 +613,10 @@ class Transport (threading.Thread):
         """
         if not self.active:
             return
-        self.active = False
+        self.stop_thread()
-        self.packetizer.close()
-        self.join()
         for chan in self._channels.values():
             chan._unlink()
+        self.sock.close()
 
     def get_remote_server_key(self):
         """
@@ -1391,6 +1389,8 @@ class Transport (threading.Thread):
     def stop_thread(self):
         self.active = False
         self.packetizer.close()
+        while self.is_alive():
+            self.join(10)
 
 
     ###  internals...