aiohboxes.py

#!/usr/bin/env python3.5

import asyncssh
import asyncio
import sys

"""
Testing for 200 hosts:

5 => 40s
50 => 10.6s
100 => 10.2s
200 = 9.1s

Testing for 525 hosts with 200 concurrency: 13.7s (with key on disk, not using agent)

"""


queue = []

ASYNCWORKERS = 200

async def worker(id):

    while True:
        try:
            host, cmd = queue.pop()
        except IndexError:
            break
        print ("Worker %s: host %s" % (id, host))
        with (await asyncssh.connect(host, username='root', known_hosts=None)) as conn:
            stdin, stdout, stderr = await conn.open_session(cmd)
            output = await stdout.read()
#            print(output, end='')

            status = stdout.channel.get_exit_status()
            if status:
                print('Program exited with status %d' % status, file=sys.stderr)
            else:
                pass
                #print('Program exited successfully')

if __name__ == '__main__':

    cmd = 'hostname'
    hosts = open('/tmp/hosts').readlines()
    for h in hosts:
        queue.append((h.strip(), cmd))

    tasks = [worker(id) for id in range(ASYNCWORKERS)]
    # tasks = [asyncio.ensure_future(worker(redis, id)) for id in range(ASYNCWORKERS)]
    future = asyncio.wait(tasks)
    asyncio.get_event_loop().run_until_complete(future)