testcase.py

import asyncio
import multiprocessing
import threading
import time
import requests
import aiohttp
import psutil
from termcolor import colored
from tabulate import tabulate

print("""
.____                          __      __         __         .__     
|    |   _____    ____    ____/  \    /  \_____ _/  |_  ____ |  |__  
|    |   \__  \  /    \  / ___\   \/\/   /\__  \\   __\/ ___\|  |  \ 
|    |___ / __ \|   |  \/ /_/  >        /  / __ \|  | \  \___|   Y  |
|_______ (____  /___|  /\___  / \__/\  /  (____  /__|  \___  >___|  /
        \/    \/     \//_____/       \/        \/          \/     \/  @system00-security

=> SpeedTest of Asyncio, Threading, Multiprocessing
""")




def print_colored(text, color):
    print(colored(text, color))

def cpu_usage():
    return psutil.cpu_percent(interval=1)

def normal_run():
    start_time = time.time()
    loop_times = 10
    request_url = "https://google.com"
    results = []

    print_colored("Test 1: Looping", 'cyan')
    start = time.time()
    for i in range(loop_times):
        pass
    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Looping", elapsed_time, cpu_utilization])

    print_colored("Test 2: Sending Requests", 'cyan')
    start = time.time()
    for i in range(loop_times):
        r = requests.get(request_url)
    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Sending Requests", elapsed_time, cpu_utilization])

    return results

async def async_run():
    start_time = time.time()
    loop_times = 10
    request_url = "https://google.com"
    results = []

    print_colored("Test 1: Looping", 'cyan')
    start = time.time()
    for i in range(loop_times):
        pass
    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Looping", elapsed_time, cpu_utilization])

    print_colored("Test 2: Sending Requests", 'cyan')
    start = time.time()
    async with aiohttp.ClientSession() as session:
        for i in range(loop_times):
            async with session.get(request_url) as response:
                pass
    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Sending Requests", elapsed_time, cpu_utilization])

    return results

async def async_run_with_thread():
    start_time = time.time()
    loop_times = 10
    request_url = "https://google.com"
    results = []

    print_colored("Test 1: Looping", 'cyan')
    start = time.time()
    for i in range(loop_times):
        pass
    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Looping", elapsed_time, cpu_utilization])

    print_colored("Test 2: Sending Requests", 'cyan')
    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = []
        for i in range(loop_times):
            tasks.append(asyncio.create_task(session.get(request_url)))
        await asyncio.gather(*tasks)
    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Sending Requests", elapsed_time, cpu_utilization])

    return results

async def async_run_with_process():
    start_time = time.time()
    loop_times = 10
    request_url = "https://google.com"
    results = []

    print_colored("Test 1: Looping", 'cyan')
    start = time.time()
    for i in range(loop_times):
        pass
    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Looping", elapsed_time, cpu_utilization])

    print_colored("Test 2: Sending Requests", 'cyan')
    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = []
        for i in range(loop_times):
            tasks.append(asyncio.create_task(session.get(request_url)))
        await asyncio.gather(*tasks)
    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Sending Requests", elapsed_time, cpu_utilization])
    

    return results

def only_multiprocessing():
    start_time = time.time()
    loop_times = 10
    request_url = "https://google.com"
    results = []

    print_colored("Test 1: Looping", 'cyan')
    start = time.time()
    for i in range(loop_times):
        pass
    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Looping", elapsed_time, cpu_utilization])

    print_colored("Test 2: Sending Requests", 'cyan')
    start = time.time()
    with multiprocessing.Pool(processes=loop_times) as pool:
        pool.map(requests.get, [request_url for i in range(loop_times)])
    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Sending Requests", elapsed_time, cpu_utilization])

    return results

def only_threading():
    start_time = time.time()
    loop_times = 10
    request_url = "https://google.com"
    results = []

    print_colored("Test 1: Looping", 'cyan')
    start = time.time()
    for i in range(loop_times):
        pass
    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Looping", elapsed_time, cpu_utilization])

    print_colored("Test 2: Sending Requests", 'cyan')
    start = time.time()
    threads = []
    for i in range(loop_times):
        t = threading.Thread(target=lambda: requests.get(request_url))
        threads.append(t)
        t.start()

    for t in threads:
        t.join()

    end = time.time()
    elapsed_time = end - start
    cpu_utilization = cpu_usage()
    results.append(["Sending Requests", elapsed_time, cpu_utilization])

    return results


def ExplainBackendWorkProcessofMultiprocessing():
    print("----------------------------------------------------")
    print_colored("Explain Backend Work Process of Multiprocessing", 'green')
    print("----------------------------------------------------")
    print("1. Create a Pool of Processes")
    print("2. Create a Queue")
    print("3. Create a Process for each task")
    print("4. Put the task into the Queue")
    print("5. Start the Process")
    print("6. Get the result from the Queue")
    print("7. Terminate the Processes")
    print("----------------------------------------------------")

def ExplainBackendWorkProcessofThreading():
    print("----------------------------------------------------")
    print_colored("Explain Backend Work Process of Threading", 'green')
    print("----------------------------------------------------")
    print("1. Create a Queue")
    print("2. Create a Thread for each task")
    print("3. Put the task into the Queue")
    print("4. Start the Thread")
    print("5. Get the result from the Queue")
    print("6. Terminate the Threads")
    print("----------------------------------------------------")

def ExplainBackendWorkProcessofAsyncio():
    print("----------------------------------------------------")
    print_colored("Explain Backend Work Process of Asyncio", 'green')
    print("----------------------------------------------------")
    print("1. Create a Queue")
    print("2. Create a Coroutine for each task")
    print("3. Put the task into the Queue")
    print("4. Start the Coroutine")
    print("5. Get the result from the Queue")
    print("6. Terminate the Coroutines")
    print("----------------------------------------------------")



def main():
    print_colored("> Normal Run", 'green')
    print(tabulate(normal_run(), headers=["Test Name", "Elapsed Time", "CPU Utilization"]))
    print("----------------------------------------------------")
    print_colored("> Async Run", 'green')
    ExplainBackendWorkProcessofAsyncio()
    print(tabulate(asyncio.run(async_run()), headers=["Test Name", "Elapsed Time", "CPU Utilization"]))
    print("----------------------------------------------------")
    print_colored("> Only Multiprocessing", 'green')
    ExplainBackendWorkProcessofMultiprocessing()
    print(tabulate(only_multiprocessing(), headers=["Test Name", "Elapsed Time", "CPU Utilization"]))
    print("----------------------------------------------------")
    print_colored("> Only Threading", 'green')
    ExplainBackendWorkProcessofThreading()
    print(tabulate(only_threading(), headers=["Test Name", "Elapsed Time", "CPU Utilization"]))
    print("----------------------------------------------------")
    print_colored("> Async Run with Thread", 'green')
    print(tabulate(asyncio.run(async_run_with_thread()), headers=["Test Name", "Elapsed Time", "CPU Utilization"]))
    print("----------------------------------------------------")
    print_colored("> Async Run with Multiprocessing", 'green')
    print(tabulate(asyncio.run(async_run_with_process()), headers=["Test Name", "Elapsed Time", "CPU Utilization"]))

if __name__ == "__main__":
    main()