由于打开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,因为我在cmos和我读过2.6.6之后可能会"复杂".

谢谢你的任何想法.

---

对shavenwarthog的评论,评论太长了:

嗨,谢谢你的回答.我有几个简单的问题.1)如果我需要在未知时间停止线程怎么办？换句话说,tail -f blah.log线程将运行大约3分钟,我想在这三分钟内检查累计数据10次？2)有点相同,我想,当我尝试使用一些实际的远程机器时,它不会退出(因为tail -f永远不会退出).我忘记了这一点,但我认为非阻塞读取是为了解决这个问题.你认为你评论的另一个线程加上这个线程是否足以使这个工作？基本上使用我的非阻塞读取来收集每个运行程序线程的本地数据.然后我只需要在主线程需要来自每个跑步者的数据时锁定,这似乎会分配我的一个锁来说10个锁,这将有所帮助.那有意义吗？

Answer 1

以下代码在多个主机上运行命令.当每个命令等待一些数据时,它将被打印到屏幕上.

整体形式改编自Alex Martelli的代码.此版本具有更多日志记录,包括显示每个连接主机的人类可读版本.

原始代码是为运行然后退出的命令编写的.当它可用时,我将其更改为逐步打印数据.以前,抓住锁的第一个线程会阻塞read(),并且所有线程都会饿死.新的解决方案绕过了这个问题.

编辑,一些说明:

为了在以后停止该程序,我们遇到了一个相当棘手的情况.线程是不间断的 - 我们不能只为程序设置信号处理sys.exit()程序.通过对join()每个线程使用while循环,更新后的代码设置为在3秒后安全退出.对于实际代码,如果父进程退出,则线程也应该正确.请仔细注意代码中的两个警告,因为信号/退出/线程交互非常简单.

代码处理数据 - 现在数据只是打印到控制台.它不使用非阻塞读取,因为1)非阻塞代码要复杂得多,2)原始程序不处理父级中子线程的数据.对于线程,可以更轻松地执行子线程的所有操作,该线程会写入文件,数据库或服务.对于任何更复杂的东西,使用multiprocessing更容易,并有很好的设施,可以做很多工作,如果他们死了就重新启动它们.该库还允许您跨多个CPU分配负载,这是线程不允许的.

玩得开心!

编辑#2

请注意,在不使用threadingnor的情况下运行多个进程是可能的,也可能是首选multiprocessing.TLDR:使用Popen和select()循环来处理批量输出.请参阅Pastebin中的示例代码:运行多个命令而不进行子进程/多处理

资源

# adapted from https://stackoverflow.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