[FIXED] SSH and Ping to hosts concurrently with Python asyncio?

Issue

I'm trying to SSH/Ping to hosts concurrently, but I don't see any result so for, probably my implementation isn't correct. This is what I have so far. Any idea appreciated.

import paramiko
import time
import asyncio
import subprocess

async def sshTest(ipaddress,deviceUsername,devicePassword,sshPort): #finalDict
    try:
            print("Performing SSH Connection to the device")

            client = paramiko.SSHClient()
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

            client.connect(ipaddress, username=deviceUsername, password=devicePassword, port=sshPort, look_for_keys=False, allow_agent=False)
            print("Channel established")     

    except Exception as e:
        print(e)       

async def pingf(ip):
    p1 = subprocess.Popen(['ping', '-c','5', ip], stdout=subprocess.PIPE)
    output = p1.communicate()[0]
    print(output)

async def main():
    taskA = loop.create_task(sshTest('192.168.255.68','admin','admin','22'))
    taskB = loop.create_task(sshTest('192.168.254.108','admin','admin','22'))
    taskC = loop.create_task(sshTest('192.168.249.134','admin','admin','22'))
    taskD = loop.create_task(sshTest('192.168.254.108','admin','admin','22'))
    task1 = loop.create_task(pingf('192.168.255.68'))
    task2 = loop.create_task(pingf('192.168.254.108'))
    task3 = loop.create_task(pingf('192.168.249.134'))
    task4 = loop.create_task(pingf('192.168.254.108'))
    await asyncio.wait([taskA,taskB,taskC,taskD,task1,task2,task3,task4]) 
    
if __name__ == "__main__":
    start = time.time()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    end = time.time()
    print("The time of execution of above program is :", end-start)
    

Solution

Asyncio is a form of cooperative multitasking. This means that in order for tasks to run concurrently, a task must explicitly yield control back to the scheduler, which in Python means "your tasks need to await on something".

Neither of your tasks ever calls await, so they're not going to run concurrently. What you have right now is going to run serially.

If you want to run ssh connections concurrently, you're going to have to either:

• Replace paramiko with something like AsyncSSH, which is written to work with asyncio, or
• Use threading or multiprocessing to parallelize your tasks, rather than using asyncio.

Additionally, if you're working with asyncio, anything that involves running an external command (such as your pingf task) is going to need to use asyncio's run_in_executor method.

---

For the example you've shown here, I would suggest instead using the concurrent.futures module. Your code might end up looking something like this (I've modified the code to run in my test environment and given the sshTest task something to do beyond simply connecting):

import concurrent.futures

import paramiko
import asyncio
import subprocess


def sshTest(ipaddress, deviceUsername, devicePassword, sshPort):  # finalDict
    try:
        print("Performing SSH Connection to the device")

        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        client.connect(
            ipaddress,
            username=deviceUsername,
            password=devicePassword,
            port=sshPort,
            look_for_keys=True,
            allow_agent=True,
        )
        stdin, stdout, stderr = client.exec_command("sh -c 'sleep 2; uptime'")
        output = stdout.read()
        return output
    except Exception:
        return "failed to connect"


def pingf(ip):
    output = subprocess.check_output(["ping", "-c", "5", ip])
    return output


def main():
    futures = []
    with concurrent.futures.ThreadPoolExecutor() as pool:
        futures.append(pool.submit(sshTest, "localhost", "root", "admin", "2200"))
        futures.append(pool.submit(sshTest, "localhost", "root", "admin", "2201"))
        futures.append(pool.submit(sshTest, "localhost", "root", "admin", "2202"))
        futures.append(pool.submit(pingf, "192.168.1.1"))
        futures.append(pool.submit(pingf, "192.168.1.5"))
        futures.append(pool.submit(pingf, "192.168.1.254"))

    for future in concurrent.futures.as_completed(futures):
        print("return value from task:", future.result())


if __name__ == "__main__":
    main()

Answered By - larsks