многопроцессорный модуль с парамико

Question

Я пытаюсь использовать модуль pym paramiko (1.7.7.1) для параллельного выполнения команд и / или файлов xfer для группы удаленных серверов. Одна задача выглядит так:

jobs = []   
for obj in appObjs:
    if obj.stop_app:
        p = multiprocessing.Process(target=exec_cmd, args=(obj, obj.stop_cmd))
        jobs.append(p)
        print "Starting job %s" % (p)
        p.start()

"obj" содержит, среди прочего, paramiko SSHClient, transport и SFTPClient. Список appObjs содержит примерно 25 таких объектов и, следовательно, 25 подключений к 25 различным серверам.

Я получаю следующую ошибку с transport.py paramiko в обратном следе

raise AssertionError("PID check failed. RNG must be re-initialized after fork(). 
Hint:   Try Random.atfork()")

Я пропатчил /usr/lib/python2.6/site-packages/paramiko/transport.py на основе поста в https://github.com/newsapps/beeswithmachineguns/issues/17, но, похоже, это не помогло. Я убедился, что использовался файл transport.py по указанному выше пути. Список рассылки paramiko, похоже, исчез.

Похоже ли это на проблему в paramiko или я неправильно понимаю / неправильно использую многопроцессорный модуль? Кто-нибудь захочет предложить практический обходной путь? Большое спасибо,

Answer 1

ОБНОВЛЕНИЕ: Как отмечает @ento, разветвленный пакет ssh был объединен обратно в paramiko , поэтому приведенный ниже ответ теперь не имеет значения, и теперь вы должны снова использовать Paramiko.

Это известная проблема в Paramiko, которая была исправлена ​​в развилке Paramiko (остановлена ​​в версии 1.7.7.1), которая теперь известна как пакет ssh на pypi (который приводит к версии 1.7. 11 на момент написания статьи).

Очевидно, что были проблемы с получением некоторых важных исправлений в основной ветке Paramiko, и сопровождающий не отвечал, поэтому @ bitprophet , сопровождающий Fabric, разветвлял Paramiko под новым именем пакета ssh package на pypi . Вы можете увидеть конкретную проблему, о которой вы упомянули, - , обсуждаемую здесь , и является одной из причин, по которой он решил ее раскошелиться; Вы можете прочитать кровавые подробности , если вы действительно хотите.

Answer 2

Как отмечается в одном комментарии внутри проблемы Paramiko , ошибки ГСЧ можно избежать, открыв один отдельный дескриптор ssh на процесс, тогда paramiko больше не будет жаловаться.Этот пример сценария демонстрирует это (я использую пул вместо процессов):

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import ssh
from multiprocessing import Pool
import getpass

hostnames = [HOST1, HOST2]
user = USERNAME
pw = getpass.getpass("Enter ssh password:")

def processFunc(hostname):
    handle = ssh.SSHClient()
    handle.set_missing_host_key_policy(ssh.AutoAddPolicy())
    handle.connect(hostname, username=user, password=pw)
    print("child")
    stdin, stdout, stderr = handle.exec_command("ls -l /var/log; sleep 5")
    cmdOutput = ""
    while True:
        try:
            cmdOutput += stdout.next()
        except StopIteration:
            break
    print("Got output from host %s:%s" % (hostname, cmdOutput))
    handle.close()

pool = Pool(len(hostnames))
pool.map(processFunc, hostnames, 1)
pool.close()
pool.join()

## If you want to compare speed:
# for hostname in hostnames:
#     processFunc(hostname)