由于打开 Paramiko ssh 连接，Python 进程挂起

Question

我正在使用 Paramiko 在测试运行期间监视远程计算机上的日志。

监视器发生在一个守护线程中，它几乎是这样做的：

        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        transport = ssh.get_transport()
        channel = transport.open_session()
        channel.exec_command('sudo tail -f ' + self.logfile)

        last_partial = ''
        while not self.stopped.isSet():
            try:
                if None == select or None == channel:
                    break
                rl, wl, xl = select.select([channel], [], [],  1.0)
                if None == rl:
                    break
                if len(rl) > 0:
                    # Must be stdout, how can I check?
                    line = channel.recv(1024)
                else:
                    time.sleep(1.0)
                    continue

            except:
                break
            if line:
               #handle saving the line... lines are 'merged' so that one log is made from all the sources
        ssh.close()

我遇到了阻塞读取的问题，所以我开始以这种方式做事，而且大部分时间它都能很好地工作。我认为当网络速度慢时我会遇到问题。

有时我会在运行结束时看到此错误（在设置了上面的 self.stopped 之后）。在设置停止并加入所有监视器线程后，我尝试过睡眠，但仍然会发生挂起。

Exception in thread Thread-9 (most likely raised during interpreter shutdown):
Traceback (most recent call last):
  File "/usr/lib64/python2.6/threading.py", line 532, in __bootstrap_inner
  File "/usr/lib/python2.6/site-packages/paramiko/transport.py", line 1470, in run
<type 'exceptions.AttributeError'>: 'NoneType' object has no attribute 'error'

在 Paramiko 的 transport.py 中，我认为这是错误所在。在下面查找#

                       self._channel_handler_table[ptype](chan, m)
                    elif chanid in self.channels_seen:
                        self._log(DEBUG, 'Ignoring message for dead channel %d' % chanid)
                    else:
                        self._log(ERROR, 'Channel request for unknown channel %d' % chanid)
                        self.active = False
                        self.packetizer.close()
                elif (self.auth_handler is not None) and (ptype in self.auth_handler._handler_table):
                    self.auth_handler._handler_table[ptype](self.auth_handler, m)
                else:
                    self._log(WARNING, 'Oops, unhandled type %d' % ptype)
                    msg = Message()
                    msg.add_byte(cMSG_UNIMPLEMENTED)
                    msg.add_int(m.seqno)
                    self._send_message(msg)
        except SSHException as e:
            self._log(ERROR, 'Exception: ' + str(e))
            self._log(ERROR, util.tb_strings())   #<<<<<<<<<<<<<<<<<<<<<<<<<<< line 1470
            self.saved_exception = e
        except EOFError as e:
            self._log(DEBUG, 'EOF in transport thread')
            #self._log(DEBUG, util.tb_strings())
            self.saved_exception = e
        except socket.error as e:
            if type(e.args) is tuple:
                if e.args:
                    emsg = '%s (%d)' % (e.args[1], e.args[0])
                else:  # empty tuple, e.g. socket.timeout
                    emsg = str(e) or repr(e)
            else:
                emsg = e.args
            self._log(ERROR, 'Socket exception: ' + emsg)
            self.saved_exception = e
        except Exception as e:
            self._log(ERROR, 'Unknown exception: ' + str(e))
            self._log(ERROR, util.tb_strings())

当运行卡住时，我可以运行>>>>> sudo lsof -i -n | egrep '\' 看到确实有卡住的 ssh 连接（无限期卡住）。我的主要测试进程是 PID 15010。

sshd       6478          root    3u  IPv4   46405      0t0  TCP *:ssh (LISTEN)
sshd       6478          root    4u  IPv6   46407      0t0  TCP *:ssh (LISTEN)
sshd      14559          root    3r  IPv4 3287615      0t0  TCP 172.16.0.171:ssh-    >10.42.80.100:59913 (ESTABLISHED)
sshd      14563         cmead    3u  IPv4 3287615      0t0  TCP 172.16.0.171:ssh->10.42.80.100:59913 (ESTABLISHED)
python    15010          root   12u  IPv4 3291525      0t0  TCP 172.16.0.171:43227->172.16.0.142:ssh (ESTABLISHED)
python    15010          root   15u  IPv4 3291542      0t0  TCP 172.16.0.171:41928->172.16.0.227:ssh (ESTABLISHED)
python    15010          root   16u  IPv4 3291784      0t0  TCP 172.16.0.171:57682->172.16.0.48:ssh (ESTABLISHED)
python    15010          root   17u  IPv4 3291779      0t0  TCP 172.16.0.171:43246->172.16.0.142:ssh (ESTABLISHED)
python    15010          root   20u  IPv4 3291789      0t0  TCP 172.16.0.171:41949->172.16.0.227:ssh (ESTABLISHED)
python    15010          root   65u  IPv4 3292014      0t0  TCP 172.16.0.171:51886->172.16.0.226:ssh (ESTABLISHED)
sshd      15106          root    3r  IPv4 3292962      0t0  TCP 172.16.0.171:ssh->10.42.80.100:60540 (ESTABLISHED)
sshd      15110         cmead    3u  IPv4 3292962      0t0  TCP 172.16.0.171:ssh->10.42.80.100:60540 (ESTABLISHED)

所以，我真的只是希望我的进程不要挂起。哦，如果 Paramiko 需要更新超过 2.6.6 的 Python，我不想更新它，因为我在 centos 上并且从我所读到的超过 2.6.6 的内容可能会“复杂”。

感谢您的任何想法。

---

评论 shavenwarthog 对于 cmets 来说太长了：

您好，感谢您的回答。我有几个简单的问题。 1）如果我需要在未知时间停止线程怎么办？换句话说，tail -f blah.log 线程可能会运行 3 分钟，我想在这 3 分钟内检查累积数据 10 次？ 2）我猜是一样的，当我用一些实际的远程机器尝试这个时它不会退出（因为 tail -f 永远不会退出）。我已经忘记了这一点，但我认为非阻塞读取是为了解决这个问题。你认为other thread you commented on 加上这个足以完成这项工作吗？基本上使用我的非阻塞读取来收集每个运行线程的本地数据。然后我只需要在主线程想要来自每个跑步者的数据时锁定，这似乎会将我的一个锁分配给 10 个锁，这会有所帮助。那有意义吗？

Answer 1

以下代码在多个主机上运行命令。当每个命令都有一些数据等待时，它会被打印到屏幕上。

整体形式改编自Alex Martelli's code。此版本具有更多日志记录，包括显示每个连接主机的人类可读版本。

原始代码是为运行然后退出的命令编写的。当数据可用时，我将其更改为增量打印数据。以前，第一个获取锁的线程会阻塞read()，所有线程都会饿死。新的解决方案绕过了这一点。

编辑，一些注释：

为了稍后停止程序，我们遇到了一个相当棘手的情况。线程是不可中断的——我们不能只为sys.exit() 程序设置一个信号处理程序。更新后的代码设置为在 3 秒后安全退出，方法是使用 while 循环到每个线程 join()。对于真正的代码，如果父级退出，那么线程也应该正确。仔细注意代码中的两个 WARNING，因为信号/退出/线程的交互相当松散。

代码处理数据，因为它来了——现在数据只是打印到控制台。它不使用非阻塞读取，因为 1) 非阻塞代码要复杂得多，以及 2) 原始程序没有处理父线程中的子线程数据。对于线程，子线程更容易做所有事情，写入文件、数据库或服务。对于更复杂的事情，请使用multiprocessing，这更容易，并且有很好的设施可以完成大量工作并在它们死亡时重新启动它们。该库还允许您跨多个 CPU 分配负载，这是线程不允许的。

玩得开心！

编辑#2

请注意，在不使用 threading 或 multiprocessing 的情况下运行多个进程是可能的，并且可能是首选。 TLDR：使用Popen 和select() 循环来处理批量输出。请参阅 Pastebin 中的示例代码：run multiple commands without subprocess/multiprocessing

来源

# adapted from https://*.com/questions/3485428/creating-multiple-ssh-connections-at-a-time-using-paramiko

import signal, sys, threading
import paramiko

CMD = 'tail -f /var/log/syslog'

def signal_cleanup(_signum, _frame):
    print '\nCLEANUP\n'
    sys.exit(0)

def workon(host):

    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(host)
    _stdin, stdout, _stderr = ssh.exec_command(CMD)

    for line in stdout:
        print threading.current_thread().name, line,

def main():
    hosts = ['localhost', 'localhost']

    # exit after a few seconds (see WARNINGs)
    signal.signal(signal.SIGALRM, signal_cleanup)
    signal.alarm(3)

    threads = [
        threading.Thread(
            target=workon, 
            args=(host,),
            name='host #{}'.format(num+1)
            )
        for num,host in enumerate(hosts)
        ]


    print 'starting'
    for t in threads:
        # WARNING: daemon=True allows program to exit when main proc
        # does; otherwise we'll wait until all threads complete.
        t.daemon = True    
        t.start()

    print 'joining'
    for t in threads:
        # WARNING: t.join() is uninterruptible; this while loop allows
        # signals
        # see: http://snakesthatbite.blogspot.com/2010/09/cpython-threading-interrupting.html
        while t.is_alive():
            t.join(timeout=0.1)

    print 'done!'

if __name__=='__main__':
    main()

输出

starting
joining
host #2 Jun 27 16:28:25 palabras kernel: [158950.369443] ideapad_laptop: Unknown event: 1
host #2 Jun 27 16:29:12 palabras kernel: [158997.098833] ideapad_laptop: Unknown event: 1
host #1 Jun 27 16:28:25 palabras kernel: [158950.369443] ideapad_laptop: Unknown event: 1
host #1 Jun 27 16:29:12 palabras kernel: [158997.098833] ideapad_laptop: Unknown event: 1
host #1 Jun 27 16:29:36 palabras kernel: [159020.809748] ideapad_laptop: Unknown event: 1