process_aggregated_data_timers_baseline_2midd_2vms.py

"""
     ASL project - fall 2017

        author: Jovan Nikolic

        Processes aggregated logs generated by middleware
"""
import numpy as np
import csv

agg_path_base = "aggregated_data/baseline_2midd_2vms/timers/"
plot_path_base = "plots/baseline_2midd_2vms/timers/"
name_base = "timer_aggregated_data_"
client_threads_basename = "clientThreads_"
worker_threads_basename = "_workerThreads_"
counters_basename = "counter_"
timers_basename = "timers_"

number_of_middlewares = 2
virtual_clients_pt = [1, 5, 8, 15, 22, 28, 32, 42, 52, 64]
worker_threads = [8, 16, 32, 64]
command_types = ["_S1-G0"]
repetitions = 3

memtier_vms = 2
memtier_instances_per_vm = 2
memtier_threads_per_inst = 1


def read_csv(client_thread, worker_thread, command_type):
    response_time = {}
    net_thread_processing_time = {}
    waiting_in_queue_time = {}
    worker_preprocessing_time = {}
    server_service_time = {}
    worker_postprocessing_time = {}
    queue_size = {}
    beginning_of_time = {}
    for mw in range(number_of_middlewares):
        current_mw = mw + 1

        final_agg_path = agg_path_base + "timer_aggregated_data_" + \
                         "mw_" + str(current_mw) +\
                         client_threads_basename + str(client_thread) + \
                         worker_threads_basename + str(worker_thread) + \
                         command_type + ".csv"

        response_time[mw] = {}
        net_thread_processing_time[mw] = {}
        waiting_in_queue_time[mw] = {}
        worker_preprocessing_time[mw] = {}
        server_service_time[mw] = {}
        worker_postprocessing_time[mw] = {}
        queue_size[mw] = {}
        beginning_of_time[mw] = {}

        for rep in range(repetitions):
            a = []
            response_time[mw][rep] = a
            b = []
            net_thread_processing_time[mw][rep] = b
            c = []
            waiting_in_queue_time[mw][rep] = c
            d = []
            worker_preprocessing_time[mw][rep] = d
            e = []
            server_service_time[mw][rep] = e
            f = []
            worker_postprocessing_time[mw][rep] = f
            g = []
            queue_size[mw][rep] = g

        with open(final_agg_path, 'r') as file:
            timer_data = file.readlines()
            timer_data = [x.strip() for x in timer_data]
            for k, line in enumerate(timer_data):
                if k == 0:
                    continue
                parsed_line = line.split(',')
                [x.strip() for x in parsed_line]

                column = 1

                if k == 1:
                    for rep in range(repetitions):
                        beginning_of_time[mw][rep] = float(parsed_line[column])
                        column += 1
                else:
                    column = repetitions + 1

                for rep in range(repetitions):
                    response_time[mw][rep].append(float(parsed_line[column]))
                    column += 1
                for rep in range(repetitions):
                    net_thread_processing_time[mw][rep].append(float(parsed_line[column]))
                    column += 1
                for rep in range(repetitions):
                    waiting_in_queue_time[mw][rep].append(float(parsed_line[column]))
                    column += 1
                for rep in range(repetitions):
                    worker_preprocessing_time[mw][rep].append(float(parsed_line[column]))
                    column += 1
                for rep in range(repetitions):
                    server_service_time[mw][rep].append(float(parsed_line[column]))
                    column += 1
                for rep in range(repetitions):
                    worker_postprocessing_time[mw][rep].append(float(parsed_line[column]))
                    column += 1
                for rep in range(repetitions):
                    queue_size[mw][rep].append(float(parsed_line[column]))
                    column += 1

    cut_left = 10
    cut_right = 80

    # if worker_thread == 8:
    #     cut_left = 5
    #     cut_right = 80

    response_time_pr = {}
    net_thread_processing_time_pr = {}
    waiting_in_queue_time_pr = {}
    worker_preprocessing_time_pr = {}
    server_service_time_pr = {}
    worker_postprocessing_time_pr = {}
    queue_size_pr = {}

    for rep in range(repetitions):
        full_list = []
        for mw in range(number_of_middlewares):
            print("MW = " + str(mw) + ", rep = " + str(rep) + ", len = " )
            full_list = np.concatenate([np.asarray(full_list), response_time[mw][rep]])
        full_list_sorted = sorted(full_list)
        mean_val = np.mean(np.asarray(full_list_sorted)[cut_left:cut_right])
        response_time_pr[rep] = mean_val

        full_list = []
        for mw in range(number_of_middlewares):
            full_list = np.concatenate([np.asarray(full_list), net_thread_processing_time[mw][rep]])
        full_list_sorted = sorted(full_list)
        mean_val = np.mean(np.asarray(full_list_sorted)[cut_left:cut_right])
        net_thread_processing_time_pr[rep] = mean_val

        full_list = []
        for mw in range(number_of_middlewares):
            full_list = np.concatenate([np.asarray(full_list), waiting_in_queue_time[mw][rep]])
        full_list_sorted = sorted(full_list)
        mean_val = np.mean(np.asarray(full_list_sorted)[cut_left:cut_right])
        waiting_in_queue_time_pr[rep] = mean_val

        full_list = []
        for mw in range(number_of_middlewares):
            full_list = np.concatenate([np.asarray(full_list), worker_preprocessing_time[mw][rep]])
        full_list_sorted = sorted(full_list)
        mean_val = np.mean(np.asarray(full_list_sorted)[cut_left:cut_right])
        worker_preprocessing_time_pr[rep] = mean_val

        full_list = []
        for mw in range(number_of_middlewares):
            full_list = np.concatenate([np.asarray(full_list), server_service_time[mw][rep]])
        full_list_sorted = sorted(full_list)
        mean_val = np.mean(np.asarray(full_list_sorted)[cut_left:cut_right])
        server_service_time_pr[rep] = mean_val

        full_list = []
        for mw in range(number_of_middlewares):
            full_list = np.concatenate([np.asarray(full_list), worker_postprocessing_time[mw][rep]])
        full_list_sorted = sorted(full_list)
        mean_val = np.mean(np.asarray(full_list_sorted)[cut_left:cut_right])
        worker_postprocessing_time_pr[rep] = mean_val

        full_list = []
        for mw in range(number_of_middlewares):
            full_list = np.concatenate([np.asarray(full_list), queue_size[mw][rep]])
        full_list_sorted = sorted(full_list)
        mean_val = np.mean(np.asarray(full_list_sorted)[cut_left:cut_right])
        queue_size_pr[rep] = mean_val

    # now we calculate the mean between repetitions and standard deviation between them

    array = []
    for rep in range(repetitions):
        array.append(response_time_pr[rep])
    response_time_mean = np.mean(array)
    response_time_std = np.std(array)

    array = []
    for rep in range(repetitions):
        array.append(net_thread_processing_time_pr[rep])
    net_thread_processing_time_mean = np.mean(array)
    net_thread_processing_time_std = np.std(array)

    array = []
    for rep in range(repetitions):
        array.append(waiting_in_queue_time_pr[rep])
    waiting_in_queue_time_mean = np.mean(array)
    waiting_in_queue_time_std = np.std(array)

    array = []
    for rep in range(repetitions):
        array.append(worker_preprocessing_time_pr[rep])
    worker_preprocessing_time_mean = np.mean(array)
    worker_preprocessing_time_std = np.std(array)

    array = []
    for rep in range(repetitions):
        array.append(server_service_time_pr[rep])
    server_service_time_mean = np.mean(array)
    server_service_time_std = np.std(array)

    array = []
    for rep in range(repetitions):
        array.append(worker_postprocessing_time_pr[rep])
    worker_postprocessing_time_mean = np.mean(array)
    worker_postprocessing_time_std = np.std(array)

    array = []
    for rep in range(repetitions):
        array.append(queue_size_pr[rep])
    queue_size_mean = np.mean(array)
    queue_size_std = np.std(array)

    return [response_time_mean, response_time_std], [net_thread_processing_time_mean, net_thread_processing_time_std], [waiting_in_queue_time_mean, waiting_in_queue_time_std], [worker_preprocessing_time_mean, worker_preprocessing_time_std], [server_service_time_mean, server_service_time_std], [worker_postprocessing_time_mean, worker_postprocessing_time_std], [queue_size_mean, queue_size_std], beginning_of_time
    # return [response_time_mean, response_time_std], [net_thread_processing_time_mean, net_thread_processing_time_std], [
    #     waiting_in_queue_time_mean, waiting_in_queue_time_std], [worker_preprocessing_time_mean,
    #                                                              worker_preprocessing_time_std], [
    #            server_service_time_mean, server_service_time_std], [worker_postprocessing_time_mean,
    #                                                                 worker_postprocessing_time_std], [queue_size_mean,
    #                                                                                                   queue_size_std], [
    #            beginning_of_time[0]]


def print_csv(path, header, full_data):
    print("Header length is: " + str(len(header)))
    print("Number of rows is: " + str(len(full_data)))
    print("NUmber of columns is: " + str(len(full_data[0])))

    with open(path, 'w') as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=header)
        writer.writeheader()

        for row in range(len(full_data)):
            one_row = {}
            for i in range(len(header)):
                one_row[header[i]] = full_data[row][i]
            writer.writerow(one_row)
        csv_file.close()


def print_csv_1(path, header, full_data):

    with open(path, 'w') as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=header)
        writer.writeheader()

        for virtual_client in virtual_clients_pt:
            for worker_thread in worker_threads:
                one_row = {}
                rep = 0
                mw = 0
                for i in range(len(header)):
                    if i == 0:
                        one_row[header[i]] = virtual_client
                    elif i == 1:
                        one_row[header[i]] = worker_thread
                    else:
                        one_row[header[i]] = full_data[virtual_client][worker_thread][mw][rep]
                        rep += 1
                        if rep >= 3:
                            rep = 0
                            mw += 1
                writer.writerow(one_row)
        csv_file.close()


def print_all_times(suffix, header, response_time, net_thread_processing_time,
                    wait_in_queue_time, worker_preprocessing_time, server_service_time,
                    worker_postprocessing_time):
    data_8 = []
    data_16 = []
    data_32 = []
    data_64 = []

    for row in range(len(response_time)):
        row_8 = [response_time[row][0], net_thread_processing_time[row][1], wait_in_queue_time[row][1],
                 worker_preprocessing_time[row][1], server_service_time[row][1], worker_postprocessing_time[row][1],
                 response_time[row][1], response_time[row][2]]
        data_8.append(row_8)

        row_16 = [response_time[row][0], net_thread_processing_time[row][3], wait_in_queue_time[row][3],
                  worker_preprocessing_time[row][3], server_service_time[row][3], worker_postprocessing_time[row][3],
                  response_time[row][3], response_time[row][4]]
        data_16.append(row_16)

        row_32 = [response_time[row][0], net_thread_processing_time[row][5], wait_in_queue_time[row][5],
                  worker_preprocessing_time[row][5], server_service_time[row][5], worker_postprocessing_time[row][5],
                  response_time[row][5], response_time[row][6]]
        data_32.append(row_32)

        row_64 = [response_time[row][0], net_thread_processing_time[row][7], wait_in_queue_time[row][7],
                  worker_preprocessing_time[row][7], server_service_time[row][7], worker_postprocessing_time[row][7],
                  response_time[row][7], response_time[row][8]]
        data_64.append(row_64)

    path = plot_path_base + "all_times_8workers_" + suffix + ".csv"
    with open(path, 'w') as the_file:
        the_file.write(header)
        for row in range(len(data_8)):
            row_data = str(int(data_8[row][0]))
            for k in range(1, len(data_8[row])):
                row_data += "\t" + str(data_8[row][k])
            row_data += "\n"
            the_file.write(row_data)
        the_file.close()

    path = plot_path_base + "all_times_16workers_" + suffix + ".csv"
    with open(path, 'w') as the_file:
        the_file.write(header)
        for row in range(len(data_16)):
            row_data = str(int(data_16[row][0]))
            for k in range(1, len(data_16[row])):
                row_data += "\t" + str(data_16[row][k])
            row_data += "\n"
            the_file.write(row_data)
        the_file.close()

    path = plot_path_base + "all_times_32workers_" + suffix + ".csv"
    with open(path, 'w') as the_file:
        the_file.write(header)
        for row in range(len(data_32)):
            row_data = str(int(data_32[row][0]))
            for k in range(1, len(data_32[row])):
                row_data += "\t" + str(data_32[row][k])
            row_data += "\n"
            the_file.write(row_data)
        the_file.close()

    path = plot_path_base + "all_times_64workers_" + suffix + ".csv"
    with open(path, 'w') as the_file:
        the_file.write(header)
        for row in range(len(data_64)):
            row_data = str(int(data_64[row][0]))
            for k in range(1, len(data_64[row])):
                row_data += "\t" + str(data_64[row][k])
            row_data += "\n"
            the_file.write(row_data)
        the_file.close()


def main():
    for z, command_type in enumerate(command_types):
        if z == 0:
            suffix = "write-only"
        else:
            suffix = "read-only"
        response_time = []
        throughput = []
        net_thread_processing_time = []
        waiting_in_queue_time = []
        worker_preprocessing_time = []
        server_service_time = []
        worker_postprocessing_time = []
        queue_size = []
        beginning_of_time = {}

        for client_thread in virtual_clients_pt:
            current_client = memtier_vms * memtier_instances_per_vm * memtier_threads_per_inst * client_thread
            response_time_row = [current_client]
            throughput_row = [current_client]
            net_thread_processing_time_row = [current_client]
            waiting_in_queue_time_row = [current_client]
            worker_preprocessing_time_row = [current_client]
            server_service_time_row = [current_client]
            worker_postprocessing_time_row = [current_client]
            queue_size_row = [current_client]
            beginning_of_time_row = {}
            for worker_thread in worker_threads:
                a, b, c, d, e, f, g, h = read_csv(client_thread, worker_thread, command_type)
                response_time_row = np.concatenate([np.asarray(response_time_row), a])

                thr = current_client*1000 / a[0]
                throughput_row = np.concatenate([np.asarray(throughput_row), [thr, 0]])

                net_thread_processing_time_row = np.concatenate([np.asarray(net_thread_processing_time_row), b])
                waiting_in_queue_time_row = np.concatenate([np.asarray(waiting_in_queue_time_row), c])
                worker_preprocessing_time_row = np.concatenate([np.asarray(worker_preprocessing_time_row), d])
                server_service_time_row = np.concatenate([np.asarray(server_service_time_row), e])
                worker_postprocessing_time_row = np.concatenate([np.asarray(worker_postprocessing_time_row), f])
                queue_size_row = np.concatenate([np.asarray(queue_size_row), g])
                beginning_of_time_row[worker_thread] = h

            response_time.append(response_time_row)
            throughput.append(throughput_row)
            net_thread_processing_time.append(net_thread_processing_time_row)
            waiting_in_queue_time.append(waiting_in_queue_time_row)
            worker_preprocessing_time.append(worker_preprocessing_time_row)
            server_service_time.append(server_service_time_row)
            worker_postprocessing_time.append(worker_postprocessing_time_row)
            queue_size.append(queue_size_row)
            beginning_of_time[client_thread] = beginning_of_time_row

        header = ["#Clients", "Mean Response Time - 8 workers [ms]", "Std Response Time - 8 workers",
                  "Mean Response Time - 16 workers [ms]", "Std Response Time - 16 workers",
                  "Mean Response Time - 32 workers [ms]", "Std Response Time - 32 workers",
                  "Mean Response Time - 64 workers [ms]", "Std Response Time - 64 workers"]
        path = plot_path_base + "response_time_" + suffix + ".csv"
        print_csv(path, header, response_time)

        header = ["#Clients", "Mean NetThread Processing Time - 8 workers [ms]",
                  "Std NetThread Processing Time - 8 workers",
                  "Mean NetThread Processing - 16 workers [ms]", "Std NetThread Processing Time - 16 workers",
                  "Mean NetThread Processing - 32 workers [ms]", "Std NetThread Processing Time - 32 workers",
                  "Mean NetThread Processing - 64 workers [ms]", "Std NetThread Processing Time - 64 workers"]
        path = plot_path_base + "netthread_processing_time_" + suffix + ".csv"
        print_csv(path, header, net_thread_processing_time)

        header = ["#Clients", "Mean Wait-In-Queue Time - 8 workers [ms]", "Std Wait-In-Queue Time - 8 workers",
                  "Mean Wait-In-Queue Time - 16 workers [ms]", "Std Wait-In-Queue Time - 16 workers",
                  "Mean Wait-In-Queue Time - 32 workers [ms]", "Std Wait-In-Queue Time - 32 workers",
                  "Mean Wait-In-Queue Time - 64 workers [ms]", "Std Wait-In-Queue Time - 64 workers"]
        path = plot_path_base + "wait_in_queue_time_" + suffix + ".csv"
        print_csv(path, header, waiting_in_queue_time)

        header = ["#Clients", "Mean Pre-Processing Time - 8 workers [ms]", "Std Pre-Processing Time - 8 workers",
                  "Mean Pre-Processing Time - 16 workers [ms]", "Std Pre-Processing Time - 16 workers",
                  "Mean Pre-Processing Time - 32 workers [ms]", "Std Pre-Processing Time - 32 workers",
                  "Mean Pre-Processing Time - 64 workers [ms]", "Std Pre-Processing Time - 64 workers"]
        path = plot_path_base + "worker_preprocessing_time_" + suffix + ".csv"
        print_csv(path, header, worker_preprocessing_time)

        header = ["#Clients", "Mean Server Service Time - 8 workers [ms]", "Std Server Service Time - 8 workers",
                  "Mean Server Service Time - 16 workers [ms]", "Std Server Service Time - 16 workers",
                  "Mean Server Service Time - 32 workers [ms]", "Std Server Service Time - 32 workers",
                  "Mean Server Service Time - 64 workers [ms]", "Std Server Service Time - 64 workers"]
        path = plot_path_base + "server_service_time_" + suffix + ".csv"
        print_csv(path, header, server_service_time)

        header = ["#Clients", "Mean Post-Processing Time - 8 workers [ms]", "Std Post-Processing Time - 8 workers",
                  "Mean Post-Processing Time - 16 workers [ms]", "Std Post-Processing Time - 16 workers",
                  "Mean Post-Processing Time - 32 workers [ms]", "Std Post-Processing Time - 32 workers",
                  "Mean Post-Processing Time - 64 workers [ms]", "Std Post-Processing Time - 64 workers"]
        path = plot_path_base + "worker_postprocessing_time_" + suffix + ".csv"
        print_csv(path, header, worker_postprocessing_time)

        header = ["#Clients", "Mean Queue Size - 8 workers", "Std Queue Size - 8 workers",
                  "Mean Queue Size - 16 workers", "Std Queue Size - 16 workers",
                  "Mean Queue Size - 32 workers", "Std Queue Size - 32 workers",
                  "Mean Queue Size - 64 workers", "Std Queue Size - 64 workers"]
        path = plot_path_base + "queue_size_" + suffix + ".csv"
        print_csv(path, header, queue_size)

        header = ["#Clients", "Mean Throughput [req/s] - 8 workers", "Std Throughput - 8 workers",
                  "Mean Throughput [req/s] - 16 workers", "Std Throughput - 16 workers",
                  "Mean Throughput [req/s] - 32 workers", "Std Throughput - 32 workers",
                  "Mean Throughput [req/s] - 64 workers", "Std Throughput - 64 workers"]
        path = plot_path_base + "throughput_from_resptime_" + suffix + ".csv"
        print_csv(path, header, throughput)

        header = ["#Virtual Clients per Thread", "Number of Worker Threads", "MW1: Timestamp - rep1",
                  "MW1: Timestamp - rep2", "MW1: Timestamp - rep3", "MW2: Timestamp - rep1",
                  "MW2: Timestamp - rep2", "MW2: Timestamp - rep3"]
        # header = ["#Virtual Clients per Thread", "Number of Worker Threads", "MW1: Timestamp - rep1",
        #           "MW2: Timestamp - rep1"]
        path = plot_path_base + "beginning_of_time_" + suffix + ".csv"
        print_csv_1(path, header, beginning_of_time)

        header = "#Client   MNT WIQ PrP SST PsP RspT    RespT_std\n"
        print_all_times(suffix, header, response_time, net_thread_processing_time, waiting_in_queue_time,
                        worker_preprocessing_time, server_service_time, worker_postprocessing_time)


if __name__ == "__main__":
    main()