Asyncio Load Testing

Python is such a fun and powerful language. Sure, it can be messy. But it makes up for it with speed and power. Below is a handy example of the power of asynchronous call functionality that can be employed with asyncio.

"Asyncio, oh asyncio, a twisty thread, where coroutines dance and futures are bred. We "await" with bated breath, while the event loop spins, juggling tasks like digital, caffeinated twins. But beware the blocking call, a villain in disguise, it'll freeze your loop solid, and bring tears to your eyes. Debugging's a riddle, a mystical, strange quest, where "pending" futures mock you, and put your skills to the test. Yet, when it all clicks, and the code flows so free, you'll feel like a maestro, conducting a symphony."

Well that was fun. Anyway, here's a simple script that leverages asyncio to call Google:

loader.py

#!/usr/bin/python
import argparse, subprocess, time, asyncio, requests

default_rps = 1
default_duration = 5

async def do():
  #Call google's homepage, disable HTTPS Cert verification, because who cares?
  requests.get('https://google.com', verify=False)
  return

async def do_loop(rps, duration):
  start_time = time.time()
  total_requests = 0
  actual_rps = rps * 1
  times_to_run = round(actual_rps * duration)

  for i in range(times_to_run):
    time.sleep(1/actual_rps)
    
    asyncio.ensure_future(do())

    total_requests += 1
    view_rps(start_time, total_requests)
  
  end_time = time.time()
  elapsed_time = end_time - start_time
  average_rps = total_requests / elapsed_time
  print("\n")
  print("_________________ RESULTS _________________")
  print("Elapsed Time:", round(elapsed_time, 1))
  print("Total Requests:", total_requests)
  print("Average RPS:", round(average_rps, 1))
  print("_________________ RESULTS _________________")
  
def view_rps(start_time, total_requests):
  seconds_ran = (time.time() - start_time) + 1
  current_rps = total_requests / seconds_ran
  print("Current RPS:", round(current_rps, 1))

# This is a basic script that clones all repo's in a group.
if __name__ == "__main__":

  # Parse arguments passed into this command.
  parser = argparse.ArgumentParser()
  parser.add_argument("-r", help="Rate Per Second.", type=int)
  parser.add_argument("-d", help="Duration in Seconds.", type=int)
  args = parser.parse_args()

  # Determine if a caching version argument was passed.
  if args.r == None:
    args.r = default_rps

  # Determine if a caching version argument was passed.
  if args.d == None:
    args.d = default_duration

  print("Starting Process...")
  print("Estimated RPS: %s" % args.r)
  print("Estimated Duration: %ss" % args.d)

  loop = asyncio.new_event_loop()
  loop.run_until_complete(do_loop(args.r, args.d))

Run a load test with 10 Requests per second, for 5 seconds, generating 50 requests:

python3 loader.py -r 10 -d 5

And we can see from the above that we averaged 9.6 RPS over 5.2 seconds. Amazing! Now, point the script at your service and try turning up the numbers.