Concurrency

Difference Between Asynchronous and Multi-Threading Programming

• Asynchronous programming allows only one part of a program to run at a specific time.

• Consider three functions in a Python program: fn1(), fn2(), and fn3().

• In asynchronous programming, if fn1() is not actively executing (e.g., it’s asleep, waiting, or has completed its task), it won’t block the entire program.

• Instead, the program optimizes CPU time by allowing other functions (e.g., fn2()) to execute while fn1() is inactive.

• Only when fn2() finishes or sleeps, the third function, fn3(), starts executing.

• This concept of asynchronous programming ensures that one task is performed at a time, and other tasks can proceed independently.

• In contrast, in multi-threading or multi-processing, all three functions run concurrently without waiting for each other to finish.

• With asynchronous programming, specific functions are designated as asynchronous using the async keyword, and the asyncio Python library helps manage this asynchronous behavior.

Asyncio

• asyncio is a library to write concurrent code using the async/await syntax

• asyncio is often a perfect fit for IO-bound and high-level structured network code

• If code is CPU-bound consider using the multiprocessing library instead

import asyncio
import aiohttp

async def fetch(session, url, headers=None, params=None):
    """
    Fetch data from a single API endpoint asynchronously.
    """
    try:
        async with session.get(url, headers=headers, params=params) as response:
            response.raise_for_status()
            return await response.json()
    except aiohttp.ClientError as e:
        print(f"Request to {url} failed: {e}")
        return None

async def fetch_all(urls, headers=None, params=None):
    """
    Fetch data from multiple API endpoints concurrently.
    """
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url, headers, params) for url in urls]
        return await asyncio.gather(*tasks)

# 🔧 Example usage
if __name__ == "__main__":
    urls = [
        "https://api.github.com/repos/python/cpython",
        "https://api.github.com/repos/django/django",
        "https://api.github.com/repos/pallets/flask"
    ]

    results = asyncio.run(fetch_all(urls))
    for i, result in enumerate(results):
        if result:
            print(f"{urls[i]} ➜ {result.get('description')}")

Concurrent.futures

import concurrent.futures
def collect_vlan(host):
  --------some code here---------
def main():
  with concurrent.futures.ProcessPoolExecutor(max_workers=5) as e:
    tasks = {e.submit(collect_vlan,host): host for host in getinfo.tb.devices}
    for task in concurrent.futures.as_completed(tasks):
      h =tasks[task]
      try:
        result=task.result()
        print(result)
      except Exception as exc:
         print('%r generated an exception: %s' % (h, exc))

from netmiko import ConnectHandler
from concurrent.futures import ThreadPoolExecutor

def ping_from_device(device_params, ip_list):
    results = {}
    try:
        with ConnectHandler(**device_params) as conn:
            for ip in ip_list:
                output = conn.send_command(f"ping {ip}")
                results[ip] = output
    except Exception as e:
        results["error"] = str(e)
    return {device_params["host"]: results}

def ping_all_devices(devices, ip_list):
    with ThreadPoolExecutor(max_workers=len(devices)) as executor:
        futures = [executor.submit(ping_from_device, device, ip_list) for device in devices]
        return [f.result() for f in futures]

# 🔧 Example usage
devices = [
    {"device_type": "cisco_ios", "host": "192.168.1.1", "username": "admin", "password": "pass"},
    {"device_type": "cisco_ios", "host": "192.168.2.1", "username": "admin", "password": "pass"},
]

ip_targets = ["8.8.8.8", "1.1.1.1"]
results = ping_all_devices(devices, ip_targets)

for device_result in results:
    for device, pings in device_result.items():
        print(f"\n🔍 {device}:")
        for ip, output in pings.items():
            print(f"Ping to {ip}:\n{output}\n")

Reference

• How to make 2500 HTTP requests in 2 seconds with Async & Await: https://www.youtube.com/watch?v=Ii7x4mpIhIs

• Python Parallel Processing

• Python Asyncio Course

• Concurrency in Python - Pool of Processes

• asyncio in Python