experiment.py

#!/usr/bin/env python

import csv
import multiprocessing as mp
import os
import subprocess
import sys

import pandas as pd

# POOL_SIZE is the parallelism level. There should be the same amount of cores or more on your CPU.
POOL_SIZE = 4
# There must be the same amount of network namespaces with each an nginx running as NETWORK_NAMESPACES.
# The amount of network namespaces should be at least POOL_SIZE. The amount is specified by setup.sh's first argument.
NETWORK_NAMESPACES = 4
# Ensure that TIMERS * MEASUREMENTS_PER_TIMER is at least 200 to get a good results.
# TIMERS should be a multiple of POOL_SIZE. Best is to have TIMERS = POOL_SIZE.
TIMERS = 4
MEASUREMENTS_PER_TIMER = 50


PROTOCOLS_TO_BENCH = ["quic", "tlstcp"]
DEFAULT_SCENARIO_TO_BENCH = "testscenarios/scenario_static.csv"
ALGORITHMS_CSV_FILE = "algorithms.csv"
MAIN_DIR = "results"
S_TIMER = "./s_timer"
QUIC_S_TIMER = "./quic_s_timer"
LD_LIBRARY_PATH = "../tmp/.local/openssl/lib64"

SRV_NS = "srv_ns"
CLI_NS = "cli_ns"
SRV_VE = "srv_ve"
CLI_VE = "cli_ve"


def main():
    shared_condition = mp.Condition()
    shared_array = mp.Array("b", [False] * NETWORK_NAMESPACES)
    timer_pool = mp.Pool(
        processes=POOL_SIZE,
        initializer=init_worker,
        initargs=(
            shared_condition,
            shared_array,
        ),
    )
    scenariofiles = parse_scenariofiles_to_bench()
    algorithms_to_bench_dict = read_algorithms()

    for scenariofile in scenariofiles:
        for protocol in PROTOCOLS_TO_BENCH:
            scenarios = pd.read_csv(scenariofile)
            testscenario_name = scenarios.columns[0]
            make_dirs(testscenario_name, protocol, algorithms_to_bench_dict)
            # get_emulated_rtt(scenarios)

            for _, parameters in scenarios.iterrows():
                # set network parameters of scenario
                set_network_parameters(parameters)

                for algorithm_class, algorithms in algorithms_to_bench_dict.items():
                    for kem_alg in algorithms:
                        path_to_results_csv_file = os.path.join(
                            MAIN_DIR,
                            testscenario_name,
                            protocol,
                            algorithm_class,
                            f"{kem_alg}.csv",
                        )
                        print(path_to_results_csv_file)
                        with open(
                            path_to_results_csv_file,
                            "a",
                        ) as out:
                            result = run_timers(timer_pool, protocol, kem_alg)
                            csv.writer(out).writerow(result)

    timer_pool.close()
    timer_pool.join()


# This function declares the global variables namespace_condition and acquired_network_namespaces so they can be used by the pool subprocesses.
def init_worker(condition, shared_array):
    global namespace_condition
    global acquired_network_namespaces
    namespace_condition = condition
    acquired_network_namespaces = shared_array


def parse_scenariofiles_to_bench():
    scenariofiles = []
    # Check for handed scenarios
    if len(sys.argv) > 1:
        for i in range(1, len(sys.argv)):
            print(sys.argv[i])
            scenariofiles.append(sys.argv[i])
    else:
        scenariofiles.append(DEFAULT_SCENARIO_TO_BENCH)
    return scenariofiles


def read_algorithms():
    with open(ALGORITHMS_CSV_FILE, "r") as csv_file:
        return {row[0]: row[1:] for row in csv.reader(csv_file)}


def make_dirs(testscenario_name, protocol, algorithms_to_bench_dict):
    for algorithm_class in algorithms_to_bench_dict.keys():
        os.makedirs(
            os.path.join(MAIN_DIR, testscenario_name, protocol, algorithm_class),
            exist_ok=True,
        )


def set_network_parameters(parameters):
    for i in range(1, NETWORK_NAMESPACES + 1):
        change_qdisc(
            f"{SRV_NS}_{i}",
            SRV_VE,
            parameters["srv_rate"],
            parameters["srv_delay"],
            parameters["srv_jitter"],
            parameters["srv_pkt_loss"],
            parameters["srv_duplicate"],
            parameters["srv_corrupt"],
            parameters["srv_reorder"],
        )
        change_qdisc(
            f"{CLI_NS}_{i}",
            CLI_VE,
            parameters["cli_rate"],
            parameters["cli_delay"],
            parameters["cli_jitter"],
            parameters["cli_pkt_loss"],
            parameters["cli_duplicate"],
            parameters["cli_corrupt"],
            parameters["cli_reorder"],
        )


# TODO maybe add slot configuration for WLAN emulation
def change_qdisc(ns, dev, rate, delay, jitter, pkt_loss, duplicate, corrupt, reorder):
    command = [
        "ip",
        "netns",
        "exec",
        ns,
        "tc",
        "qdisc",
        "change",
        "dev",
        dev,
        "root",
        "netem",
        # limit 1000 is the default value for the number of packets that can be queued
        "limit",
        "1000",
        "rate",
        f"{rate}mbit",
        "delay",
        f"{delay}ms",
        f"{jitter}ms",
        "loss",
        f"{pkt_loss}%",
        "duplicate",
        f"{duplicate}%",
        "corrupt",
        f"{corrupt}%",
        "reorder",
        f"{reorder}%",
    ]

    print(" > " + " ".join(command))
    run_subprocess(command)


def run_timers(timer_pool, protocol, kem_alg):
    results_nested = timer_pool.starmap(
        time_handshake, [(protocol, kem_alg, MEASUREMENTS_PER_TIMER)] * TIMERS
    )
    return [item for sublist in results_nested for item in sublist]


# do TLS handshake (s_timer.c)
def time_handshake(protocol, kem_alg, measurements) -> list[float]:
    def aquire_network_namespace():
        with namespace_condition:
            while True:
                for i in range(1, len(acquired_network_namespaces) + 1):
                    if not acquired_network_namespaces[i - 1]:
                        acquired_network_namespaces[i - 1] = True
                        return i
                # make this process sleep until another wakes him up
                namespace_condition.wait()

    def release_network_namespace(i):
        with namespace_condition:
            acquired_network_namespaces[i - 1] = False
            # wake another process that is sleeping
            namespace_condition.notify()

    network_namespace = aquire_network_namespace()
    program = QUIC_S_TIMER if protocol == "quic" else S_TIMER
    command = [
        "ip",
        "netns",
        "exec",
        f"{CLI_NS}_{network_namespace}",
        "env",
        f"LD_LIBRARY_PATH={LD_LIBRARY_PATH}",
        program,
        kem_alg,
        str(measurements),
    ]
    result = run_subprocess(command)
    release_network_namespace(network_namespace)
    return [float(i) for i in result.strip().split(",")]


def run_subprocess(command, working_dir=".", expected_returncode=0) -> str:
    result = subprocess.run(
        command,
        stdout=subprocess.PIPE,  # puts stdout in result.stdout
        stderr=subprocess.PIPE,
        cwd=working_dir,
    )
    if result.stderr:
        print(result.stderr)
    assert result.returncode == expected_returncode
    return result.stdout.decode("utf-8")


# TODO think about what to do with this rtt calculation, maybe just delete
# maybe put it below the setting of the network parameters and save to a different file
# To get actual (emulated) RTT
def get_emulated_rtt(scenarios):
    change_qdisc(
        SRV_NS,
        SRV_VE,
        scenarios.iloc[0]["srv_rate"],
        scenarios.iloc[0]["srv_delay"],
        0,
        0,
        0,
        0,
        0,
    )
    change_qdisc(
        CLI_NS,
        CLI_VE,
        scenarios.iloc[0]["cli_rate"],
        scenarios.iloc[0]["cli_delay"],
        0,
        0,
        0,
        0,
        0,
    )
    rtt_str = get_rtt_ms()
    print(
        f"Emulated average RTT: {rtt_str} ms",
    )


def get_rtt_ms():
    command = ["ip", "netns", "exec", "cli_ns", "ping", "10.0.0.1", "-c", "10"]

    print(" > " + " ".join(command))
    result = run_subprocess(command)

    # last line is "rtt min/avg/max/mdev = 5.978/6.107/6.277/0.093 ms"
    result_fmt = result.splitlines()[-1].split("/")
    return result_fmt[4]


main()