Zappy Load Tester

load_tester.py

#!/usr/bin/python3
import sys, socket
from multiprocessing.dummy import Pool as ThreadPool
import multiprocessing
import time

def starter(sockets):
    for sock in sockets:
        print("Broadcasting")
        sock.send("Broadcast tutut\n".encode())
        msg = sock.recv(1024)
        # print(msg)

def init(l):
    global lock
    lock = l

def     start(port, nb_team, user_by_team):
    print("Starting on 127.0.0.1:{} on {} teams with {} users by team".format(port, nb_team, user_by_team))
    team_sockets = []
    for i in range(1, nb_team + 1):
        print ("Creating sockets for team{}".format(i))
        sockets = []
        for _ in range(0, user_by_team - 1):
            sockets.append(socket.socket(socket.AF_INET, socket.SOCK_STREAM))
            # time.sleep(0.2)
        for sock in sockets:
            sock.connect(("127.0.0.1", port))
            sock.send("team{}".format(i).encode())
            msg = sock.recv(1024)
            print(msg)
        team_sockets.append(sockets)
    l = multiprocessing.Lock()
    for i in range(0, 20):
        pool = ThreadPool(8, initializer=init, initargs=(l,))
        pool.map(starter, team_sockets)
        pool.close()

if __name__ == "__main__":
    items = map(int, sys.argv[1:])
    start(*items)

👍