commit 8174f6de960a0584d134340e55ae970567a63c8c Author: Ano-sys Date: Sun Jan 26 21:55:49 2025 +0100 Initial push diff --git a/CLion_DEBUGGING.pdf b/CLion_DEBUGGING.pdf new file mode 100644 index 0000000..5963ada Binary files /dev/null and b/CLion_DEBUGGING.pdf differ diff --git a/VSCode_DEBUGGING.zip b/VSCode_DEBUGGING.zip new file mode 100644 index 0000000..c7fb05d Binary files /dev/null and b/VSCode_DEBUGGING.zip differ diff --git a/praxis3.pdf b/praxis3.pdf new file mode 100644 index 0000000..1ab8758 Binary files /dev/null and b/praxis3.pdf differ diff --git a/praxis3.skeleton.tar.gz b/praxis3.skeleton.tar.gz new file mode 100644 index 0000000..550cabe Binary files /dev/null and b/praxis3.skeleton.tar.gz differ diff --git a/praxis3/requirements.txt b/praxis3/requirements.txt new file mode 100644 index 0000000..31695e0 --- /dev/null +++ b/praxis3/requirements.txt @@ -0,0 +1,6 @@ +numpy>=1.20.3 +pandas +requests +zmq +pytest +pytest-timeout diff --git a/praxis3/test/check_submission.sh b/praxis3/test/check_submission.sh new file mode 100755 index 0000000..6869f44 --- /dev/null +++ b/praxis3/test/check_submission.sh @@ -0,0 +1,43 @@ +#!/bin/bash + +set -e + +if [[ ! $# -eq 1 ]]; then + echo "Missing argument." + echo "Usage: '$0 praxis{0,1,2,3}'" + exit 1 +fi + +TEST_SCRIPT="$(dirname $0)/test_$1.py" + +if [[ ! -e ${TEST_SCRIPT} ]]; then + echo "Requested test file ${TEST_SCRIPT} does not exist!" + exit 1 +fi + +echo "Packaging submission..." +PACKAGING_DIR=$(mktemp -d) +cmake -B${PACKAGING_DIR} +cmake --build ${PACKAGING_DIR} -t package_source + +echo "Extracting submission..." +SOURCE_DIR=$(mktemp -d) +(cd ${SOURCE_DIR} && find ${PACKAGING_DIR} -name '*.tar.gz' | head -n1 | xargs tar -xzf) +EXTRACTED_SOURCE_DIR=$(find ${SOURCE_DIR} -maxdepth 1 -mindepth 1 -type d | head -n1) + +echo "Building submission..." +BUILD_DIR=$(mktemp -d) +cmake -S"${EXTRACTED_SOURCE_DIR}" -B${BUILD_DIR} +TARGET="webserver" +EXECUTABLE="--executable ${BUILD_DIR}/webserver" +if [[ "$1" == "praxis0" ]]; then + TARGET="hello_world" + EXECUTABLE="--executable ${BUILD_DIR}/hello_world" +elif [[ "$1" == "praxis3" ]]; then + TARGET="zmq_distributor zmq_worker" + EXECUTABLE="--dist_exec ${BUILD_DIR}/zmq_distributor --work_exec ${BUILD_DIR}/zmq_worker" +fi +cmake --build ${BUILD_DIR} -t ${TARGET} + +echo "Executing tests..." +python3 -m pytest -o cache_dir=${BUILD_DIR} ${TEST_SCRIPT} ${EXECUTABLE} diff --git a/praxis3/test/conftest.py b/praxis3/test/conftest.py new file mode 100644 index 0000000..55e96f8 --- /dev/null +++ b/praxis3/test/conftest.py @@ -0,0 +1,7 @@ +import pytest + + +def pytest_addoption(parser): + parser.addoption('--dist_exec', action='store', default='build/zmq_distributor') + parser.addoption('--work_exec', action='store', default='build/zmq_worker') + parser.addoption('--debug_test', action='store_true', default=False) diff --git a/praxis3/test/test_praxis3.py b/praxis3/test/test_praxis3.py new file mode 100644 index 0000000..bf3ed06 --- /dev/null +++ b/praxis3/test/test_praxis3.py @@ -0,0 +1,515 @@ +""" +Tests for RN Praxis 3 +""" + +import multiprocessing +from sys import stderr + +import numpy as np +import pytest +import subprocess +import sys +import time +import util +import zmq + +zmq.Context() + +test_args = None +distributor_valgrind_test_passed = False +debug_tests = False + + +@pytest.fixture +def program_args(request): + d_exec = request.config.option.dist_exec + w_exec = request.config.option.work_exec + global debug_tests + debug_tests = request.config.option.debug_test + # cache = request.config.option.cache_texts + # cache_dir = request.config.option.cache_dir + return {"distributor": d_exec, "worker": w_exec} + + +@pytest.mark.timeout(15) +def test_workers_reachable(program_args): + global test_args + test_args = util.init_tests(program_args) + + base_port = test_args["base_port"] + + for num_workers in [1, 2, 4, 8]: + workers = np.arange(base_port, base_port + num_workers).tolist() + + # kill any zmq procs currently running + util.kill_zmq_distributor_and_worker() + + # launch worker here + port_list = [str(x) for x in workers] + worker_procs = util.start_threaded_workers(test_args["worker"], port_list) + + workers_reached = 0 + + test_message = "map\0" + rip_message = "rip\0" + context = zmq.Context.instance() + for w in workers: + socket = context.socket(zmq.REQ) + socket.setsockopt(zmq.LINGER, 0) + socket.setsockopt(zmq.RCVTIMEO, 500) + socket.connect("tcp://127.0.0.1:" + str(w)) + socket.send(bytes(test_message, "ascii")) + + message = None + try: + message = socket.recv() + except zmq.ZMQError: + print("Failed to receive reply from worker " + str(w)) + socket.close() + continue + + reply = message[:-1].decode("ascii") + if reply == "": + workers_reached += 1 + + # rip worker + socket.send(bytes(rip_message, "ascii")) + try: + socket.recv() + except zmq.ZMQError: + print("Failed to receive rip from worker " + str(w)) + + socket.close() + + util.join_workers(worker_procs) + + assert workers_reached == num_workers, f"Failed to reach all workers for {num_workers} worker threads! Expected {num_workers} but reached {workers_reached}." + + +@pytest.mark.timeout(15) +def test_distributor_message_size(): + word_list = util.get_part_of_word_list(test_args["word_list"], 250) + + simple_delimiters = test_args["simple_delimiters"] + + max_msg_size = 1500 + max_string_length = 50 * max_msg_size + + test_string = util.generate_text_from_word_list(word_list, simple_delimiters, max_string_length) + + filename = test_args["filename_simple"] + f = open(filename, "w") + f.write(test_string) + f.close() + + port = str(test_args["base_port"]) + + # kill any zmq procs currently running + util.kill_zmq_distributor_and_worker() + + proc = util.start_distributor([test_args["distributor"], filename, port]) + + context = zmq.Context.instance() + socket = context.socket(zmq.REP) + socket.bind("tcp://*:" + port) + + map_response = "test1" * (int(max_msg_size/5)-1) + "\0" + + message_lengths_received = [] + + while True: + msg = socket.recv() + message_lengths_received.append(len(msg)) + req_type = msg[0:3].decode("ascii") + + if req_type == "rip": + socket.send(b"rip\0") + break + elif req_type == "map": + socket.send(map_response.encode("ascii")) + elif req_type == "red": + socket.send(b"test1\0") + socket.close() + proc.wait() + + for l in message_lengths_received: + assert l <= max_msg_size, f"Received message size exceeds max message size of {max_msg_size} bytes." + + +@pytest.mark.timeout(60) +def test_simple_text(): + filename_simple = test_args["filename_simple"] + + f = open(filename_simple, "r") + simple_text = f.read() + f.close() + + base_port = test_args["base_port"] + + for num_workers in [1, 2, 4, 8]: + workers = np.arange(base_port, base_port + num_workers).tolist() + port_list = [str(x) for x in workers] + + # kill any zmq procs currently running + util.kill_zmq_distributor_and_worker() + + # launch worker and distributor + worker_procs = util.start_threaded_workers(test_args["worker"], port_list) + proc_distributor = util.start_distributor([test_args["distributor"], filename_simple] + + port_list) + + util.join_workers(worker_procs) + + distributor_output, distributor_err = proc_distributor.communicate() + correct_word_count = util.count_words(simple_text) + + if debug_tests: + util.create_test_debug_output("test_simple_text", num_workers, correct_word_count, distributor_output) + + assert distributor_output == correct_word_count, f"{num_workers} workers failed simple text test." + + +@pytest.mark.timeout(60) +def test_complex_text(): + word_list = util.get_part_of_word_list(test_args["word_list"], 2500) + all_delimiters = test_args["all_delimiters"] + max_string_length = 2.5e5 + + test_string = util.generate_text_from_word_list(word_list, all_delimiters, max_string_length) + + filename = test_args["filename_complex"] + + f = open(filename, "w") + f.write(test_string) + f.close() + + base_port = test_args["base_port"] + + for num_workers in [2, 4, 8]: + workers = np.arange(base_port, base_port + num_workers).tolist() + port_list = [str(x) for x in workers] + + # kill any zmq procs currently running + util.kill_zmq_distributor_and_worker() + + # launch worker and distributor + worker_procs = util.start_threaded_workers(test_args["worker"], port_list) + proc_distributor = util.start_distributor([test_args["distributor"], filename] + + port_list) + + util.join_workers(worker_procs) + + distributor_output, distributor_err = proc_distributor.communicate() + correct_word_count = util.count_words(test_string) + + if debug_tests: + util.create_test_debug_output("test_complex_text", num_workers, correct_word_count, distributor_output) + + assert distributor_output == correct_word_count, f"{num_workers} workers failed complex text test." + + +@pytest.mark.timeout(90) +def test_book_1(): + filename = test_args["filename_book_1"] + base_port = test_args["base_port"] + book_text = test_args["books"][0] + + file_out = open(filename, "wb") + file_out.write(book_text) + file_out.close() + + for num_workers in [2, 4, 8]: + workers = np.arange(base_port, base_port + num_workers).tolist() + port_list = [str(x) for x in workers] + + # kill any zmq procs currently running + util.kill_zmq_distributor_and_worker() + + # launch worker and distributor + worker_procs = util.start_threaded_workers(test_args["worker"], port_list) + proc_distributor = util.start_distributor([test_args["distributor"], filename] + + port_list) + + util.join_workers(worker_procs) + + distributor_output, distributor_err = proc_distributor.communicate() + + book_text_str = book_text.decode("ascii", errors="ignore") + correct_word_count = util.count_words(book_text_str) + + if debug_tests: + util.create_test_debug_output("test_book_1", num_workers, correct_word_count, distributor_output) + + assert distributor_output == correct_word_count, f"{num_workers} workers failed book 1 test." + + +@pytest.mark.timeout(90) +def test_book_2(): + filename = test_args["filename_book_2"] + base_port = test_args["base_port"] + book_text = test_args["books"][1] + + file_out = open(filename, "wb") + file_out.write(book_text) + file_out.close() + + for num_workers in [2, 4, 8]: + workers = np.arange(base_port, base_port + num_workers).tolist() + port_list = [str(x) for x in workers] + + # kill any zmq procs currently running + util.kill_zmq_distributor_and_worker() + + # launch worker and distributor + worker_procs = util.start_threaded_workers(test_args["worker"], port_list) + proc_distributor = util.start_distributor([test_args["distributor"], filename] + + port_list) + + util.join_workers(worker_procs) + + distributor_output, distributor_err = proc_distributor.communicate() + + book_text_str = book_text.decode("ascii", errors="ignore") + correct_word_count = util.count_words(book_text_str) + + if debug_tests: + util.create_test_debug_output("test_book_2", num_workers, correct_word_count, distributor_output) + + assert distributor_output == correct_word_count, f"{num_workers} workers failed book 2 test." + + +@pytest.mark.timeout(30) +def test_interoperability(): + base_port = test_args["base_port"] + + # test workers first + for num_workers in [2, 4, 8]: + + workers = np.arange(base_port, base_port + num_workers).tolist() + port_list = [str(x) for x in workers] + + # kill any zmq procs currently running + util.kill_zmq_distributor_and_worker() + + worker_procs = util.start_threaded_workers(test_args["worker"], port_list) + + context = zmq.Context.instance() + + map_message = "mapInteroperability test. Test uses python distributor.\0" + map_message_expected_return = "interoperability1test11uses1python1distributor1\0" + reduce_message_expected_return = "interoperability1test2uses1python1distributor1\0" + rip_message = "rip\0" + + worker_responses = dict() + + for port in port_list: + curr_responses = [] + + socket = context.socket(zmq.REQ) + socket.connect("tcp://127.0.0.1:" + port) + + socket.send(bytes(map_message, "ascii")) + message = socket.recv() + message = message.decode("ascii") + curr_responses.append(message) + # assert message == map_message_expected_return + + socket.send(bytes("red" + map_message_expected_return, "ascii")) + message = socket.recv() + message = message.decode("ascii") + curr_responses.append(message) + # assert message == reduce_message_expected_return + + socket.send(bytes(rip_message, "ascii")) + message = socket.recv() + message = message.decode("ascii") + curr_responses.append(message) + # assert message == rip_message + + socket.close() + + worker_responses[port] = curr_responses + + util.join_workers(worker_procs) + + for res in worker_responses.values(): + assert res[0] == map_message_expected_return + assert res[1] == reduce_message_expected_return + assert res[2] == rip_message + + # test distributor second + port_list = [str(base_port)] + + # kill any zmq procs currently running + util.kill_zmq_distributor_and_worker() + + context = zmq.Context.instance() + socket = context.socket(zmq.REP) + socket.bind("tcp://*:" + str(base_port)) + + map_message = "mapInteroperability test. Test uses python distributor.\0" + map_message_expected_return = "interoperability1test11uses1python1distributor1\0" + reduce_message_expected_return = "interoperability1test2uses1python1distributor1\0" + rip_message = "rip\0" + + distributor_messages = [] + + out_filename = test_args["filename_interop"] + f = open(out_filename, "w") + f.write(map_message[3:-1]) + f.close() + proc_distributor = util.start_distributor([test_args["distributor"], out_filename] + port_list) + + message = socket.recv() + message = message.decode("ascii") + # print(message) + # assert message == map_message + distributor_messages.append(message) + socket.send(bytes(map_message_expected_return, "ascii")) + + message = socket.recv() + message = message.decode("ascii") + # print(message) + # assert message == "red" + map_message_expected_return + distributor_messages.append(message) + socket.send(bytes(reduce_message_expected_return, "ascii")) + + message = socket.recv() + message = message.decode("ascii") + # print(message) + # assert message == rip_message + distributor_messages.append(message) + socket.send(bytes(rip_message, "ascii")) + + socket.close() + + distributor_output, distributor_err = proc_distributor.communicate() + + assert distributor_messages[0] == map_message + assert distributor_messages[1] == "red" + map_message_expected_return + assert distributor_messages[2] == rip_message + + assert distributor_output == util.count_words(map_message[3:]) + + +@pytest.mark.timeout(60) +def test_load_distribution(): + base_port = test_args["base_port"] + + for num_workers in [2, 4, 8]: + workers = np.arange(base_port, base_port + num_workers).tolist() + port_list = [str(x) for x in workers] + + # kill any zmq procs currently running + util.kill_zmq_distributor_and_worker() + + stds = [] + + # run test 3 times, because sometimes bad scheduling may affect load distribution + for i in range(3): + manager = multiprocessing.Manager() + return_dict = manager.dict() + worker_threads = [] + for p in port_list: + proc = multiprocessing.Process(target=util.run_worker_load_distribution, args=(p, return_dict)) + worker_threads.append(proc) + proc.start() + + proc_distributor = util.start_distributor([test_args["distributor"], test_args["filename_complex"]] + port_list) + + proc_distributor.wait() + for t in worker_threads: + t.join() + + requests_per_worker = return_dict.values() + curr_std = np.std(requests_per_worker) + stds.append(curr_std) + # stop because std is already satisfying condition, no need for extra runs + if curr_std <= 2: + break + + std = np.min(stds) + assert std <= 2 + + +@pytest.mark.timeout(120) +def test_memory_leaks(): + global test_args + + valgrind_output_file = "valgrind_output" + valgrind_command_base = ["valgrind", "--tool=memcheck", "--xml=yes", "--xml-file=" + valgrind_output_file, + "--gen-suppressions=all", "--leak-check=full", "--leak-resolution=med", "--track-origins=yes", + "--trace-children=yes", "--vgdb=no", "--fair-sched=yes"] + + valgrind_command_distributor = valgrind_command_base.copy() + valgrind_command_distributor[3] = "--xml-file=" + valgrind_output_file + "_distributor.xml" + + test_args["filename_valgrind_distributor"] = valgrind_command_distributor[3].split("=")[1] + + valgrind_commands_worker = [] + valgrind_worker_out_files = [] + for i in range(8): + curr_command = valgrind_command_base.copy() + curr_filename = valgrind_output_file + "_worker_" + str(i) + ".xml" + test_args["filename_valgrind_worker" + str(i)] = curr_filename + valgrind_worker_out_files.append(curr_filename) + curr_command[3] = "--xml-file=" + curr_filename + curr_command.append(test_args["worker"]) + valgrind_commands_worker.append(curr_command) + + word_list = util.get_part_of_word_list(test_args["word_list"], 500) + simple_delimiters = test_args["simple_delimiters"] + text_length = 100e3 + + test_string = util.generate_text_from_word_list(word_list, simple_delimiters, text_length) + + valgrind_test_filename = test_args["filename_valgrind"] + f = open(valgrind_test_filename, "w") + f.write(test_string) + f.close() + + base_port = test_args["base_port"] + + worker_results = [] + distributor_results = [] + + for num_workers in [2, 4, 8]: + + workers = np.arange(base_port, base_port + num_workers).tolist() + port_list = [str(x) for x in workers] + + # kill any zmq procs currently running + util.kill_zmq_distributor_and_worker() + + # launch worker and distributor + proc_workers = util.start_threaded_workers(valgrind_commands_worker[:num_workers], port_list) + proc_distributor = util.start_distributor(valgrind_command_distributor + [test_args["distributor"], valgrind_test_filename] + + port_list) + + proc_distributor.wait() + util.join_workers(proc_workers) + + for i in range(num_workers): + worker_results.append(util.check_valgrind_output_no_errors(valgrind_worker_out_files[i])) + distributor_results.append(util.check_valgrind_output_no_errors(valgrind_command_distributor[3].split("=")[1])) + + assert int(np.mean(worker_results)) == 1, f"Worker failed to pass valgrind test!" + + global distributor_valgrind_test_passed + distributor_valgrind_test_passed = int(np.mean(distributor_results)) == 1 + + +@pytest.mark.timeout(3) +def test_valgrind_second_point_and_cleanup(): + time.sleep(1) + # kill any zmq procs currently running, including memcheck this time + util.kill_zmq_distributor_and_worker(kill_memcheck=True) + + context = zmq.Context.instance() + context.destroy() + + if not debug_tests: + remove_valgrind_xml_files = True + util.cleanup_files(test_args, remove_valgrind_xml_files) + + assert distributor_valgrind_test_passed, f"Distributor failed to pass valgrind test!" diff --git a/praxis3/test/util.py b/praxis3/test/util.py new file mode 100644 index 0000000..e2e7d6a --- /dev/null +++ b/praxis3/test/util.py @@ -0,0 +1,301 @@ +import multiprocessing +import os +import random +import re +import signal +import string +from io import StringIO +import subprocess +import sys +import time +import xml.etree.ElementTree as ET + +from collections import Counter +from typing import List, Union + +import numpy as np +import pandas as pd +import requests +import shutil +import zmq + + +def init_tests(args): + random.seed() + + distributor_exec = args["distributor"] + worker_exec = args["worker"] + # cache = args["cache"] + # cache_dir = args["cache_dir"] + + simple_delimiters = [", ", ". ", " "] + all_delimiters = [x + " " for x in string.punctuation] + + filename_simple = "test_simple_text.txt" + filename_complex = "test_complex_text.txt" + + base_port = 6001 + base_port = check_ports_free(base_port) + + word_list_url = "https://raw.githubusercontent.com/dwyl/english-words/master/words_alpha.txt" + word_list = None + for i in range(4): + try: + url_content = requests.get(word_list_url, timeout=5) + word_list = url_content.text.splitlines() + break + except requests.exceptions.Timeout: + continue + if word_list is None: + sys.exit("Failed to get word list. Can not run tests without it. Quitting...") + + book_base_url = "https://www.gutenberg.org/cache/epub/" + book_options = ["1513/pg1513.txt", "2701/pg2701.txt", "1342/pg1342.txt", "84/pg84.txt", "145/pg145.txt"] + book_1 = random.choice(book_options) + book_options.remove(book_1) + book_2 = random.choice(book_options) + book_text_1 = get_book_text(book_base_url + book_1) + book_list = [book_text_1] + book_text_2 = get_book_text(book_base_url + book_2) + book_list.append(book_text_2) + + filename_book_1 = "book_1.txt" + filename_book_2 = "book_2.txt" + + filename_interop = "interop_test.txt" + + filename_valgrind = "valgrind_test.txt" + + + test_args = {"distributor": distributor_exec, + "worker": worker_exec, + # "cache": cache, + # "cache_dir": cache_dir, + "simple_delimiters": simple_delimiters, + "all_delimiters": all_delimiters, + "filename_simple": filename_simple, + "filename_complex": filename_complex, + "base_port": base_port, + "word_list": word_list, + "books": book_list, + "filename_book_1": filename_book_1, + "filename_book_2": filename_book_2, + "filename_interop": filename_interop, + "filename_valgrind": filename_valgrind, + } + + return test_args + + +# checks if the base port and all following 8 ports are actually free for testing +# if not, check the next 10 +def check_ports_free(base_port): + if not shutil.which("netstat") or sys.platform != "linux": + print("netstat not found. " + "To ensure all used ports during testing are free, please install net-tools with: sudo apt install net-tools", + file=sys.stderr) + return base_port + + proc = subprocess.Popen(["netstat -tuna | tail -n +2 | tr -s [:blank:] ','"], stdout=subprocess.PIPE, shell=True) + netstat_output, netstat_err = proc.communicate() + + netstat_output = netstat_output.decode("ascii") + df = pd.read_csv(StringIO(netstat_output)) + df = df["Local"] + df = df.apply(lambda x: x.split(":")[-1]) + + ports = np.arange(base_port, base_port + 8).tolist() + port_list = [str(x) for x in ports] + + if sum((df == i).any() for i in port_list) > 0: + base_port += 10 + base_port = check_ports_free(base_port) + return base_port + + +def kill_zmq_distributor_and_worker(kill_memcheck=False): + try: + distributor_procs = subprocess.check_output(["pidof", "zmq_distributor"]).decode("utf-8").split() + distributor_procs = [int(x) for x in distributor_procs] + except subprocess.CalledProcessError: + distributor_procs = [] + try: + worker_procs = subprocess.check_output(["pidof", "zmq_worker"]).decode("utf-8").split() + worker_procs = [int(x) for x in worker_procs] + except subprocess.CalledProcessError: + worker_procs = [] + if kill_memcheck: + # pgrep required, because pidof only returns exact matches + try: + memcheck_procs = subprocess.check_output(["pgrep", "memcheck"]).decode("utf-8").split() + memcheck_procs = [int(x) for x in memcheck_procs] + except subprocess.CalledProcessError: + memcheck_procs = [] + else: + memcheck_procs = [] + procs = distributor_procs + worker_procs + memcheck_procs + for p in procs: + try: + os.kill(p, signal.SIGKILL) + except ProcessLookupError: + continue + + +def get_part_of_word_list(word_list, n_words): + word_list_part = random.sample(word_list, n_words) + return word_list_part + + +def random_capitalization(words): + for i in range(len(words)): + words[i] = ''.join(map(random.choice, zip(words[i].lower(), words[i].upper()))) + + +def generate_text_from_word_list(w_list, delim, text_size): + word_lengths = [len(x) for x in w_list] + median_length = np.median(word_lengths) + words_to_pick = int(text_size / median_length) + words = random.choices(w_list, k=words_to_pick) + random_capitalization(words) + + i = random.randint(25,100) + while i < len(words): + words.insert(i, "\n") + i += random.randint(25,100) + + test_string = "".join(x + random.choice(delim) for x in words) + + return test_string + +def count_words(input_string): + input_string = input_string.lower() + + words = re.findall("[a-z]+", input_string) + + words = Counter(words) + words = sorted(words.items(), key=lambda a: (-a[1], a[0])) + + output_string = "word,frequency\n" + + for w in words: + output_string += w[0] + "," + str(w[1]) + "\n" + + return output_string + +def get_book_text(url): + book_text = None + for i in range(4): + try: + url_content = requests.get(url, timeout=5) + book_text = url_content.text.encode(encoding="ascii", errors="ignore") + break + except requests.exceptions.Timeout: + continue + if book_text is None: + sys.exit("Failed to get book text. Can not run tests without it. Quitting...") + return book_text + + +def create_test_debug_output(test_name, num_w, expected_output, dist_output): + debug_base_path = "debug/" + if not os.path.isdir(debug_base_path): + os.mkdir(debug_base_path) + out_name = debug_base_path + test_name + "_n-workers=" + str(num_w) + "_" + + f = open(out_name + "expected.txt", "w") + f.write(expected_output) + f.close() + + f = open(out_name + "distributor_output.txt", "w") + f.write(dist_output) + f.close() + + +def run_worker(w_exec : Union[str, List[str]], port : str): + proc_worker = None + if isinstance(w_exec, str): + proc_worker = subprocess.Popen([w_exec, port]) + elif isinstance(w_exec, list): + w_exec.append(port) + proc_worker = subprocess.Popen(w_exec) + proc_worker.wait() + + +def join_workers(workers: list): + for w in workers: + if isinstance(w, subprocess.Popen): + w.wait() + elif isinstance(w, multiprocessing.Process): + w.join() + + +def start_threaded_workers(worker_exec : Union[str, List[List[str]]], port_list : list, use_multiprocessing = True): + # use_multiprocessing = False + if use_multiprocessing: + worker_procs = [] + if isinstance(worker_exec, str): + for p in port_list: + proc = multiprocessing.Process(target=run_worker, args=(worker_exec, p)) + worker_procs.append(proc) + proc.start() + elif isinstance(worker_exec, list): + for p, exe in zip(port_list, worker_exec): + proc = multiprocessing.Process(target=run_worker, args=(exe, p)) + worker_procs.append(proc) + proc.start() + return worker_procs + else: + proc_workers = None + if isinstance(worker_exec, str): + proc_workers = subprocess.Popen([worker_exec] + port_list) + elif isinstance(worker_exec, list): + proc_workers = subprocess.Popen(worker_exec[0] + port_list) + return [proc_workers] + + +def start_distributor(dist_args : List[str]): + return subprocess.Popen(dist_args, stdout=subprocess.PIPE, encoding="ascii") + + +def run_worker_load_distribution(port, return_dict): + return_message = "test1worker1load1distribution1\0" + + context = zmq.Context.instance() + socket = context.socket(zmq.REP) + socket.bind("tcp://*:" + str(port)) + + received_requests = 0 + + while True: + message = socket.recv() + message = message.decode("ascii") + + msg_type = message[:3] + if msg_type == "map" or msg_type == "red": + received_requests += 1 + socket.send(bytes(return_message, encoding="ascii")) + elif msg_type == "rip": + socket.send(bytes("rip\0", encoding="ascii")) + break + else: + print("Unknown message type!") + break + socket.close() + + return_dict[port] = received_requests + + +def check_valgrind_output_no_errors(filename): + if not os.path.isfile(filename): + return True + tree = ET.parse(filename) + errors = tree.findall("error") + return len(errors) == 0 + + +def cleanup_files(test_args, remove_valgrind_files=True): + for k, v in test_args.items(): + if not remove_valgrind_files and "valgrind" in k and ".xml" in v: + continue + if "filename" in k and os.path.isfile(v): + os.remove(v) diff --git a/praxis3/tox.ini b/praxis3/tox.ini new file mode 100644 index 0000000..90e333e --- /dev/null +++ b/praxis3/tox.ini @@ -0,0 +1,10 @@ +[tox] +env_list = py{38,39,310,311} +skipsdist = true + +[testenv] +description = run unit tests +deps = + pytest +commands = + pytest