PRSONA/prsona/src/orchestratorMain.cpp

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

#include "networking.hpp"

using namespace std;

chrono::seconds oneSecond(1); 

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)
        return false;

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

        unique_lock<mutex> lck(synch->mtx);
        synch->val = atoi(data);

        return false;
    }

    if ((bits & 0xf) == MG_WEBSOCKET_OPCODE_DATACOMPLETE)
        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 output)
{
    stringstream buffer;
    string command;

    buffer << "ssh tmgurtle@" << target << " \"screen \'~/prsona/prsona/bin/"
        << (server ? "startServer.sh " : "startClient.sh ") << output << "\'\"" ;
    command = buffer.str();

    return system(command.c_str());
}

void wait_for_servers_ready(string dealer)
{
    bool flag = false;
    while (!flag)
    {
        this_thread::sleep_for(oneSecond);

        stringstream sysString;
        string data;
        char buffer[255];

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

        if (!conn)
            continue;

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

        mg_write(conn, data.c_str(), data.length());
        size_t readLen = mg_read(conn, (uint8_t*) buffer, 254);
        buffer[readLen] = 0;

        if (strstr(buffer, "200"))
            flag = true;

        mg_close_connection(conn);
    }
}

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

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

        stringstream sysString;
        string data;

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

        if (!conn)
            continue;

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

        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)
{
    this_thread::sleep_for(oneSecond);

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

        struct mg_connection *conn =
            mg_connect_client(
                dealer.c_str(), 
                PRSONA_PORT,
                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);
}

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

        struct mg_connection *conn =
            mg_connect_client(
                target.c_str(), 
                PRSONA_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, string verifier)
{
    bool flag = false;
    while (!flag)
    {
        stringstream sysString;
        string data;

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

        if (!conn)
            continue;

        sysString << "GET " << TRIGGER_REP_URI << verifier << " 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 execute_experiment(string dealer)
{
    size_t line = 1;

    char buffer[128];
    ifstream commands("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':
                trigger_vote(string(buffer + 2));
                break;

            case 'R':
                char *target, *verifier;
                target = strtok(buffer, " ");
                verifier = strtok(NULL, " ");
                trigger_reputation_proof(string(target), string(verifier));
                break;

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

            default:
                break;
        }

        line++;
    }
}

int main(int argc, char* argv[])
{
    string experimentOutput = random_string(8);

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

    vector<string> serverIPs, clientIPs;
    string dealerIP;

    char buffer[40];
    ifstream serverConfig("serverIPs.cfg");
    while (!serverConfig.eof())
    {
        serverConfig.getline(buffer, 40);
        if (strlen(buffer) > 0)
            serverIPs.push_back(string(buffer));
    }

    ifstream clientConfig("clientIPs.cfg");
    while (!clientConfig.eof())
    {
        clientConfig.getline(buffer, 40);
        if (strlen(buffer) > 0)
            clientIPs.push_back(string(buffer));
    }

    ifstream dealerConfig("dealerIP.cfg");
    while (!dealerConfig.eof())
    {
        dealerConfig.getline(buffer, 40);
        if (strlen(buffer) > 0)
            dealerIP = buffer;
    }

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

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

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

    this_thread::sleep_for(oneSecond);

    cout << "Starting other servers." << endl;

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

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

    cout << "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);

    cout << "Starting clients." << endl;

    for (size_t i = 0; i < numClients; i++)
        clientStartup.push_back(thread(start_remote_actor, clientIPs[i], false, experimentOutput));

    cout << "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, numClients);

    cout << "Beginning experiment." << endl;

    execute_experiment(dealerIP);

    cout << "Finishing experiment." << endl;
    cout << "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(), 
                PRSONA_PORT,
                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 << "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(), 
                PRSONA_PORT,
                USE_SSL, 
                NULL,
                0);

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

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

    return 0;
}