PRSONA/prsona/src/orchestratorMain.cpp

#include <iostream>
#include <random>
#include <fstream>
#include <sstream>
#include <string>
#include <cstring>
#include <cstdlib>
#include <vector>
#include <chrono>
#include <thread>

#include "print_helpers.hpp"
#include "networking.hpp"

using namespace std;

const chrono::seconds ONE_SECOND(1); 
const chrono::seconds FIVE_SECONDS(5); 

int clients_websocket_data_handler(
    struct mg_connection *conn,
    int bits,
    char *data,
    size_t data_len,
    void *user_data)
{
    if ((bits & 0xf) == MG_WEBSOCKET_OPCODE_CONNECTION_CLOSE || (bits & 0xf) == MG_WEBSOCKET_OPCODE_DATACOMPLETE)
        return false;

    if ((bits & 0xf) == MG_WEBSOCKET_OPCODE_BINARY)
    {
        struct synchronization_tool *sync = (struct synchronization_tool *) user_data;

        unique_lock<mutex> lck(sync->mtx);
        
        if (data_len == sizeof(sync->val))
            sync->val = *((size_t *) data);

        return false;
    }

    std::cerr << "Unknown response when trying to get update lock." << std::endl;
    return false;    
}

int start_remote_actor(string target, bool server, string id, string output)
{
    stringstream buffer;
    string command;

    // buffer << "ssh tmgurtle@" << target << " \"screen \'~/prsona/prsona/scripts/startup.sh "
    //     << (server ? "server " : "client ") << id << " t " << output << "\'\"" ;

    buffer << "bin/" << (server ? "server " : "client ") << id << " T " << output << " &";
    command = buffer.str();

    return system(command.c_str());
}

void wait_for_servers_ready(string dealer, int dealerPort)
{
    bool flag = false;
    while (!flag)
    {
        this_thread::sleep_for(ONE_SECOND);

        stringstream sysString;
        string data;

        struct mg_connection *conn =
            mg_connect_client(
                dealer.c_str(), 
                dealerPort,
                USE_SSL, 
                NULL,
                0);

        if (!conn)
        {
            cerr << "couldn't make connection while waiting for servers ready" << endl;
            continue;
        }

        sysString << "GET " << EPOCH_READY_URI << " HTTP/1.1\r\n";
        sysString << "Host: " << dealer << ":" << dealerPort << "\r\n\r\n";
        data = sysString.str();

        mg_write(conn, data.c_str(), data.length());
        mg_get_response(conn, NULL, 0, 250);
        const struct mg_response_info *info = mg_get_response_info(conn);

        if (info->status_code == 200)
            flag = true;

        mg_close_connection(conn);
    }
}

void wait_for_clients_ready(string dealer, int dealerPort, size_t numClients)
{
    struct synchronization_tool numClientsSync;

    bool flag = false;
    while (!flag)
    {
        this_thread::sleep_for(ONE_SECOND);

        stringstream sysString;
        string data;

        struct mg_connection *conn =
            mg_connect_websocket_client(
                dealer.c_str(),
                dealerPort,
                USE_SSL,
                NULL,
                0,
                NUM_CLIENTS_URI,
                "null",
                clients_websocket_data_handler,
                synchro_websocket_close_handler,
                &numClientsSync);

        if (!conn)
        {
            cerr << "couldn't make connection while waiting for clients ready" << endl;
            continue;
        }

        unique_lock<mutex> lck(numClientsSync.mtx);
        numClientsSync.val = 0;
        numClientsSync.val2 = 0;

        mg_websocket_client_write(
            conn,
            MG_WEBSOCKET_OPCODE_DATACOMPLETE,
            "",
            0);

        while (!numClientsSync.val2)
            numClientsSync.cv.wait(lck);

        mg_close_connection(conn);

        if (numClientsSync.val == numClients)
            flag = true;
    }
}

void trigger_epoch(string dealer, int dealerPort)
{
    wait_for_servers_ready(dealer, dealerPort);

    bool flag = false;
    while (!flag)
    {
        stringstream sysString;
        string data;

        struct mg_connection *conn =
            mg_connect_client(
                dealer.c_str(), 
                dealerPort,
                USE_SSL, 
                NULL,
                0);

        if (!conn)
            continue;

        sysString << "GET " << TRIGGER_EPOCH_URI << " HTTP/1.1\r\n";
        sysString << "Host: " << dealer << "\r\n\r\n";
        data = sysString.str();

        mg_write(conn, data.c_str(), data.length());
        mg_close_connection(conn);

        flag = true;
    }

    wait_for_servers_ready(dealer, dealerPort);
}

void trigger_vote(string target, int port)
{
    bool flag = false;
    while (!flag)
    {
        stringstream sysString;
        string data;

        struct mg_connection *conn =
            mg_connect_client(
                target.c_str(), 
                port,
                USE_SSL, 
                NULL,
                0);

        if (!conn)
            continue;

        sysString << "GET " << TRIGGER_VOTE_URI << " HTTP/1.1\r\n";
        sysString << "Host: " << target << "\r\n\r\n";
        data = sysString.str();

        mg_write(conn, data.c_str(), data.length());
        mg_close_connection(conn);

        flag = true;
    }
}

void trigger_reputation_proof(string target, int targetPort, string verifier, int verifierPort)
{
    bool flag = false;
    while (!flag)
    {
        stringstream sysString;
        string data;

        struct mg_connection *conn =
            mg_connect_client(
                target.c_str(), 
                targetPort,
                USE_SSL, 
                NULL,
                0);

        if (!conn)
            continue;

        sysString << "GET " << TRIGGER_REP_URI << "?" << verifier << ":" << verifierPort << " HTTP/1.1\r\n";
        sysString << "Host: " << target << "\r\n\r\n";
        data = sysString.str();

        mg_write(conn, data.c_str(), data.length());
        mg_close_connection(conn);

        flag = true;
    }

    this_thread::sleep_for(ONE_SECOND);
}

void execute_experiment(string dealer, int dealerPort)
{
    size_t line = 1;

    char buffer[128];
    ifstream commands("cfg/commands.cfg");

    while (!commands.eof())
    {
        commands.getline(buffer, 128);
        if (strlen(buffer) == 0)
        {
            line++;
            continue;
        }

        cout << "Command " << line << ": " << string(buffer) << endl;

        switch(buffer[0])
        {
            case 'V':
                char *voter, *voterPort;
                voter = strtok(buffer + 1, " :");
                voterPort = strtok(NULL, " :");

                trigger_vote(string(voter), atoi(voterPort));
                break;

            case 'R':
                char *target, *targetPortStr, *verifier, *verifierPortStr;
                target = strtok(buffer + 1, " :");
                targetPortStr = strtok(NULL, " :");
                verifier = strtok(NULL, " :");
                verifierPortStr = strtok(NULL, " :");

                trigger_reputation_proof(
                    string(target),
                    atoi(targetPortStr),
                    string(verifier),
                    atoi(verifierPortStr));
                break;

            case 'E':
                trigger_epoch(dealer, dealerPort);
                break;

            default:
                break;
        }

        line++;
    }

    wait_for_servers_ready(dealer, dealerPort);
}

int main(int argc, char* argv[])
{
    string seedStr = "seed";

    seed_seq seed(seedStr.begin(), seedStr.end());
    default_random_engine rng(seed);

    string experimentOutput = random_string(rng, 8);

#if USE_SSL
    mg_init_library(0);
#else
    mg_init_library(MG_FEATURES_SSL);
#endif

    vector<string> serverIPs, clientIPs;
    vector<int> serverPorts, clientPorts;
    string dealerIP;
    int dealerPort = 0;

    char buffer[46], *helper;
    ifstream serverConfig("cfg/serverIPs.cfg");
    while (!serverConfig.eof())
    {
        serverConfig.getline(buffer, 46);
        if (strlen(buffer) > 0)
        {
            helper = buffer;
            if (strchr(helper, ':'))
            {
                helper = strtok(helper, ":");
                serverIPs.push_back(string(helper));
                helper = strtok(NULL, ":");
                serverPorts.push_back(atoi(helper));
            }
            else
            {
                serverIPs.push_back(string(helper));
                serverPorts.push_back(atoi(PRSONA_PORT_STR));
            }
        }
    }

    ifstream clientConfig("cfg/clientIPs.cfg");
    while (!clientConfig.eof())
    {
        clientConfig.getline(buffer, 46);
        if (strlen(buffer) > 0)
        {
            helper = buffer;
            if (strchr(helper, ':'))
            {
                helper = strtok(helper, ":");
                clientIPs.push_back(string(helper));
                helper = strtok(NULL, ":");
                clientPorts.push_back(atoi(helper));
            }
            else
            {
                clientIPs.push_back(string(helper));
                clientPorts.push_back(atoi(PRSONA_PORT_STR));
            }
        }
    }

    ifstream dealerConfig("cfg/dealerIP.cfg");
    while (!dealerConfig.eof())
    {
        dealerConfig.getline(buffer, 46);
        if (strlen(buffer) > 0)
        {
            helper = buffer;
            if (strchr(helper, ':'))
            {
                helper = strtok(helper, ":");
                dealerIP = helper;
                helper = strtok(NULL, ":");
                dealerPort = atoi(helper);
            }
            else
            {
                dealerIP = helper;
                dealerPort = atoi(PRSONA_PORT_STR);
            }
        }
    }

    size_t numServers = serverIPs.size();
    size_t numClients = clientIPs.size();

    cout << "[ORC] This experiment is running with output code: " << experimentOutput << endl;
    cout << "[ORC] Starting BGN dealer server." << endl;

    vector<thread> serverStartup, clientStartup;
    serverStartup.push_back(thread(start_remote_actor, dealerIP, true, "d", experimentOutput));

    this_thread::sleep_for(ONE_SECOND);

    cout << "[ORC] Starting other servers." << endl;

    for (size_t i = 0; i < numServers; i++)
    {
        if (serverIPs[i] == dealerIP && serverPorts[i] == dealerPort)
            continue;

        serverStartup.push_back(thread(start_remote_actor, serverIPs[i], true, "s" + to_string(i), experimentOutput));
    }

    cout << "[ORC] Waiting for confirmation that servers are ready to continue." << endl;

    for (size_t i = 0; i < numServers; i++)
        serverStartup[i].join();

    wait_for_servers_ready(dealerIP, dealerPort);

    cout << "[ORC] Starting clients." << endl;

    for (size_t i = 0; i < numClients; i++)
    {
        clientStartup.push_back(thread(start_remote_actor, clientIPs[i], false, "c" + to_string(i), experimentOutput));
        this_thread::sleep_for(ONE_SECOND);
    }

    cout << "[ORC] Waiting for confirmation that servers have all clients logged." << endl;    

    for (size_t i = 0; i < numClients; i++)
        clientStartup[i].join();

    wait_for_clients_ready(dealerIP, dealerPort, numClients);

    cout << "[ORC] Beginning experiment." << endl;

    execute_experiment(dealerIP, dealerPort);

    cout << "[ORC] Finishing experiment." << endl;
    cout << "[ORC] Sending shutdown commands to clients." << endl;

    for (size_t i = 0; i < clientIPs.size(); i++)
    {
        stringstream sysString;
        string data;

        struct mg_connection *conn =
            mg_connect_client(
                clientIPs[i].c_str(), 
                clientPorts[i],
                USE_SSL, 
                NULL,
                0);

        sysString << "GET " << EXIT_URI << " HTTP/1.1\r\n";
        sysString << "Host: " << clientIPs[i] << "\r\n\r\n";
        data = sysString.str();

        mg_write(conn, data.c_str(), data.length());
        mg_close_connection(conn);
    }

    cout << "[ORC] Sending shutdown commands to servers." << endl;

    for (size_t i = 0; i < serverIPs.size(); i++)
    {
        stringstream sysString;
        string data;

        struct mg_connection *conn =
            mg_connect_client(
                serverIPs[i].c_str(), 
                serverPorts[i],
                USE_SSL, 
                NULL,
                0);

        sysString << "GET " << EXIT_URI << " HTTP/1.1\r\n";
        sysString << "Host: " << serverIPs[i] << ":" << serverPorts[i] << "\r\n\r\n";
        data = sysString.str();

        mg_write(conn, data.c_str(), data.length());
        mg_close_connection(conn);
    }

    return 0;
}