PRSONA/prsona/src/networkOrchestrator.cpp

#include <iostream>
#include <fstream>
#include <sstream>
#include <cstdlib>
#include <thread>

#include "networkOrchestrator.hpp"

/***************************************************
 *********                                 *********
 *********  orchestrator public functions  *********
 *********                                 *********
 ***************************************************/

/*
 * START UP AND SHUT DOWN INSTANCES
 */

int start_remote_actor(
    const std::string& target,
    bool server,
    const std::string& id,
    const std::string& output,
    size_t lambda,
    bool maliciousServers)
{
    std::stringstream buffer;
    std::string command;

    if (target != "tick0" && !target.empty())
        buffer << "ssh -n " << target << " \"~/prsona/prsona/scripts/startup.sh " << (server ? "server " : "client ") << id << " " << output << " " << lambda << (maliciousServers ? " T\" &" : " F\" &");
    else
        buffer << "bin/" << (server ? "server " : "client ") << id << " " << output << " " << lambda << (maliciousServers ? " T &" : " F &");
    command = buffer.str();

    return system(command.c_str());
}

void shut_down_remote_actors(
    const std::vector<std::string>& relevantIPs,
    const std::vector<int>& relevantPorts)
{
    for (size_t i = 0; i < relevantIPs.size(); i++)
    {
        // Shut downs are triggered by a GET request to the correct location
        std::stringstream sysString;
        std::string data;
        sysString << "GET " << EXIT_URI << " HTTP/1.1\r\n";
        sysString << "Host: " << relevantIPs[i] << ":" << relevantPorts[i] << "\r\n\r\n";
        data = sysString.str();

        struct mg_connection *conn = NULL;

        // Connect to the instance
        while (!conn)
        {
            conn = mg_connect_client(relevantIPs[i].c_str(), relevantPorts[i], USE_SSL, NULL, 0);

            if (!conn)
                std::cerr << "Couldn't connect to instance at " << relevantIPs[i] << ":" << relevantPorts[i] << " for shut down." << std::endl;
        }

        // Make correct GET request
        mg_write(conn, data.c_str(), data.length());

        // Close connection
        mg_close_connection(conn);
    }
}

/*
 * SYNCHRONIZATION
 */

void wait_for_servers_ready(
    std::string dealer,
    int dealerPort)
{
    // Requesting information about servers being ready is done via a GET request
    std::stringstream sysString;
    std::string data;
    sysString << "GET " << EPOCH_READY_URI << " HTTP/1.1\r\n";
    sysString << "Host: " << dealer << ":" << dealerPort << "\r\n\r\n";
    data = sysString.str();

    bool ready = false;
    while (!ready)
    {
        struct mg_connection *conn = NULL;

        // Connect to the dealer
        while (!conn)
        {
            std::this_thread::sleep_for(ONE_SECOND);

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

            if (!conn)
                std::cerr << "Couldn't make connection while waiting for servers to be ready." << std::endl;
        }

        // Make the correct GET request
        mg_write(conn, data.c_str(), data.length());

        // Wait for a response
        mg_get_response(conn, NULL, 0, 250);
        const struct mg_response_info *info = mg_get_response_info(conn);

        // Close connection
        mg_close_connection(conn);

        // If the dealer says it's ready, then we can move on
        if (info->status_code == 200)
            ready = true;
    }
}

void wait_for_clients_ready(
    std::string dealer,
    int dealerPort,
    size_t numClients)
{
    bool ready = false;
    while (!ready)
    {
        struct synchronization_tool sync;
        struct mg_connection *conn = NULL;

        // Connect to the dealer
        std::unique_lock<std::mutex> lck(sync.mtx);
        sync.val = 0;
        sync.val2 = 0;
        while (!conn)
        {
            std::this_thread::sleep_for(ONE_SECOND);

            conn = mg_connect_websocket_client(dealer.c_str(), dealerPort, USE_SSL, NULL, 0, REQUEST_NUM_CLIENTS_URI, "null", clients_websocket_data_handler, synchro_websocket_close_handler, &sync);

            if (!conn)
                std::cerr << "Couldn't make connection while waiting for clients to be ready." << std::endl;
        }

        // Tell the dealer we're ready for its response
        mg_websocket_client_write(conn, MG_WEBSOCKET_OPCODE_DATACOMPLETE, "", 0);

        // Wait for that response
        while (!sync.val2)
            sync.cv.wait(lck);

        // Close connection
        mg_close_connection(conn);

        // If the dealer says it's ready, then we can move on
        if (sync.val == numClients)
            ready = true;
    }
}

/*
 * RUN EXPERIMENT
 */

void execute_experiment(
    std::string dealer,
    int dealerPort,
    const char *filename)
{
    size_t line = 1;

    // Iterate across each line in the command file, which contains one command per line
    char buffer[128];
    std::ifstream commands(filename);
    while (!commands.eof())
    {
        commands.getline(buffer, 128);
        if (strlen(buffer) == 0)
        {
            line++;
            continue;
        }

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

        // The first character of each command tells us which it is
        switch(buffer[0])
        {
            // Vote triggers come in form `V <voterIP>:<voterPort>`
            case 'V':
                char *voter, *voterPort;
                voter = strtok(buffer + 1, " :");
                voterPort = strtok(NULL, " :");

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

            // Reputation proof triggers come in form `R <proverIP>:<proverPort> <verifierIP>:<verifierPort>`
            case 'R':
                char *target, *targetPortStr, *verifier, *verifierPortStr;
                target = strtok(buffer + 1, " :");
                targetPortStr = strtok(NULL, " :");
                verifier = strtok(NULL, " :");
                verifierPortStr = strtok(NULL, " :");

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

            // Epoch change triggers come in form `E`
            case 'E':
                trigger_epoch_change(dealer, dealerPort);
                break;

            default:
                break;
        }

        line++;
    }

    // Don't let ourselves shut down servers and clients until we're sure they're not in the middle of anything else
    wait_for_servers_ready(dealer, dealerPort);
}

/****************************************************
 *********                                  *********
 *********  orchestrator private functions  *********
 *********                                  *********
 ****************************************************/

/*
 * TRIGGER EXPERIMENT EVENTS
 */

void trigger_epoch_change(
    std::string dealer,
    int dealerPort)
{
    // Give other updates a chance to resolve
    wait_for_servers_ready(dealer, dealerPort);

    // Epoch changes are triggered via GET request to the correct location
    std::stringstream sysString;
    std::string data;
    sysString << "GET " << TRIGGER_EPOCH_URI << " HTTP/1.1\r\n";
    sysString << "Host: " << dealer << ":" << dealerPort << "\r\n\r\n";
    data = sysString.str();

    struct mg_connection *conn = NULL;

    // Connect to the dealer
    while (!conn)
    {
        conn = mg_connect_client(dealer.c_str(), dealerPort, USE_SSL, NULL, 0);

        if (!conn)
            std::cerr << "Couldn't connect to dealer to trigger epoch change." << std::endl;
    }

    // Make the relevant GET request
    mg_write(conn, data.c_str(), data.length());

    // Close connection
    mg_close_connection(conn);

    // Don't bother giving new commands until this one has resolved
    wait_for_servers_ready(dealer, dealerPort);
}

void trigger_vote(
    std::string target,
    int targetPort)
{
    // New votes are triggered via GET request to the correct location
    std::stringstream sysString;
    std::string data;
    sysString << "GET " << TRIGGER_VOTE_URI << " HTTP/1.1\r\n";
    sysString << "Host: " << target << ":" << targetPort << "\r\n\r\n";
    data = sysString.str();

    struct mg_connection *conn = NULL;

    // Connect to the client
    while (!conn)
    {
        conn = mg_connect_client(target.c_str(), targetPort, USE_SSL, NULL, 0);

        if (!conn)
            std::cerr << "Couldn't connect to client at " << target << ":" << targetPort << " to trigger new vote." << std::endl;
    }

    // Make the relevant GET request
    mg_write(conn, data.c_str(), data.length());

    // Close connection
    mg_close_connection(conn);
}

void trigger_reputation_proof(
    std::string target,
    int targetPort,
    std::string verifier,
    int verifierPort)
{
    // Reputation proofs are triggered via GET request to the correct location (with a parameter for the intended verifier)
    std::stringstream sysString;
    std::string data;
    sysString << "GET " << TRIGGER_REP_URI << "?" << verifier << ":" << verifierPort << " HTTP/1.1\r\n";
    sysString << "Host: " << target << "\r\n\r\n";
    data = sysString.str();

    struct mg_connection *conn = NULL;

    // Connect to the client
    while (!conn)
    {
        conn = mg_connect_client(target.c_str(), targetPort, USE_SSL, NULL, 0);

        if (!conn)
            std::cerr << "Couldn't connect to client at " << target << ":" << targetPort << " to trigger reputation proof." << std::endl;
    }

    // Make the relevant GET request
    mg_write(conn, data.c_str(), data.length());

    // Close connection
    mg_close_connection(conn);

    // Give this command a small amount of time to resolve
    std::this_thread::sleep_for(ONE_SECOND);
}