[FIXED] Asyncio big list of Task with sequential combine run_in_executor and standard Coroutine in each

Issue

I need to handle list of 2500 ip-addresses from csv file. So I need to create_task from coroutine 2500 times. Inside every coroutine firstly I need to fast-check access of IP:PORT via python module "socket" and it is a synchronous function want to be in loop.run_in_executor(). Secondly if IP:PORT is opened I need to connect to this socket via asyncssh.connect() for doing some bash commands and this is standart asyncio coroutine. Then I need to collect results of this bash commands to another csv file.

Additionaly there is an issue in Linux: system can not open more than 1024 connections at same time. I think it may be solved by making list of lists[1000] with asyncio.sleep(1) between or something like that.

I expected my tasks will be executed by 1000 in 1 second but it only 20 in 1 sec. Why?

Little working code snippet with comments here:

#!/usr/bin/env python3
import asyncio
import csv
import time
from pathlib import Path
import asyncssh
import socket
from concurrent.futures import ThreadPoolExecutor as Executor

PARALLEL_SESSIONS_COUNT = 1000
LEASES_ALL = Path("ip_list.csv")
PORT = 22
TIMEOUT = 1
USER = "testuser1"
PASSWORD = "123"

def is_open(ip,port,timeout):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(timeout)
    try:
        s.connect((ip, int(port)))
        s.shutdown(socket.SHUT_RDWR)
        return {"result": True, "error": "NoErr"}
    except Exception as ex:
        return {"result": False, "error": str(ex)}
    finally:
        s.close()

def get_leases_list():
    # Minimal csv content:
    # header must contain "IPAddress"
    # every other line is concrete ip-address.
    result = []

    with open(LEASES_ALL, newline="") as csvfile_1:
        reader_1 = csv.DictReader(csvfile_1)
        result = list(reader_1)

    return result

def split_list(some_list, sublist_count):
    result = []

    while len(some_list) > sublist_count:
        result.append(some_list[:sublist_count])
        some_list = some_list[sublist_count:]
    result.append(some_list)

    return result

async def do_single_host(one_lease_dict): # Function for each Task

    # Firstly
    IP = one_lease_dict["IPAddress"]
    loop = asyncio.get_event_loop()
    socket_check = await loop.run_in_executor(None, is_open, IP, PORT, TIMEOUT)
    print(socket_check, IP)
    # Secondly
    if socket_check["result"] == True:
        async with asyncssh.connect(host=IP, port=PORT, username=USER, password=PASSWORD, known_hosts=None) as conn:
            result = await conn.run("uname -r", check=True)
            print(result.stdout, end="") # Just print without write in file at this point.

def aio_root():
    leases_list = get_leases_list()

    list_of_lists = split_list(leases_list, PARALLEL_SESSIONS_COUNT)

    r = []
    loop = asyncio.get_event_loop()
    for i in list_of_lists:
        for j in i:
            task = loop.create_task(do_single_host(j))
            r.append(task)
        group = asyncio.wait(r)
        loop.run_until_complete(group) # At this line execute only by 20 in 1sec. Can't understand why :(
    loop.close()

def main():
    aio_root()

if __name__ == '__main__':
    main()

Solution

loop.run_in_exectutor signature:

awaitable loop.run_in_executor(executor, func, *args)¶

The default ThreadPoolExecutor is used if executor is None.

ThreadPoolExecutor document:

Changed in version 3.5: If max_workers is None or not given, it will default to the number of processors on the machine, multiplied by 5, assuming that ThreadPoolExecutor is often used to overlap I/O instead of CPU work and the number of workers should be higher than the number of workers for ProcessPoolExecutor.

Changed in version 3.8: Default value of max_workers is changed to min(32, os.cpu_count() + 4). This default value preserves at least 5 workers for I/O bound tasks. It utilizes at most 32 CPU cores for CPU bound tasks which release the GIL. And it avoids using very large resources implicitly on many-core machines.

Answered By - kyungmin