PRSONA/prsona/src/orchestratorMain.cpp

/**
 * orchestratorMain.cpp
 *      - compiles to bin/orchestrator
 *      - initiates a set of servers and clients, then commands them, for PRSONA experiments
 *
 * Stan Gurtler
 */

#include <iostream>
#include <fstream>
#include <thread>
#include <csignal>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>

#include "networkOrchestrator.hpp"

using namespace std;

void cleanup(int signum)
{
    while (waitpid(-1, NULL, WNOHANG) > 0) {}
}

/**
 * This program (bin/orchestrator) expects to be called as follows:
 * `bin/orchestrator <output> <servers_are_malicious>`
 *
 * <output> - a string that will name the files in which outputs for this run of
 *      the experiment will be written (that is, timings and traffic data)
 * <lambda> - a positive integer that determines the absolute soundness parameter
 *      for batched proofs
 * <servers_are_malicious> - a bool (given as T/t or F/f)
 *      which is true when servers are in malicious security
 *      and false when they are in HBC security
 */
int main(int argc, char* argv[])
{
    /*
     * PRELIMINARY SETUP CODE
     */

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

    string output = "default";
    if (argc > 1)
        output = argv[1];

    // Default to not proof batching if not specified
    size_t lambda = 0;
    if (argc > 2)
        lambda = atoi(argv[2]);

    // Default to malicious security if not specified
    bool maliciousServers = true;
    if (argc > 3)
        maliciousServers = argv[3][0] == 't' || argv[3][0] == 'T';

    // This seed is used to pick which users vote during which epochs
    // and which users make reputation proofs to which other users
    string seedStr = output;    
    seedStr += "-orchestrator";
    seed_seq seed(seedStr.begin(), seedStr.end());
    default_random_engine rng(seed);    

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

    std::map<std::string,std::string> targeter;
    targeter["129.97.119.208"] = "tick0";
    targeter["129.97.119.209"] = "tick1";
    targeter["129.97.119.215"] = "tock";
    targeter["127.0.0.1"] = "self";

    string configDir = "cfg/" + output;

    // Read in from config files the server locations
    load_multiple_instances_config(serverIPs, serverPorts, (configDir + "/serverIPs.cfg").c_str());

    // And now the client locations
    load_multiple_instances_config(clientIPs, clientPorts, (configDir + "/clientIPs.cfg").c_str());

    // And finally the dealer location
    load_single_instance_config(dealerIP, dealerPortStr, dealerPort, (configDir + "/dealerIP.cfg").c_str());

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

    signal(SIGCHLD, cleanup);

    /*
     * ORCHESTRATOR SETUP CODE
     */

    cout << "[ORC] This experiment is running with output code: " << output << endl;
    cout << "[ORC] This experiment is running with " << (maliciousServers ? "MALICIOUS" : "HBC") << " servers." << endl;
    cout << "[ORC] Lambda is " << lambda << "." << endl;
    cout << endl;
    
    cout << "[ORC] Starting BGN dealer server." << endl;

    vector<thread> serverStartup, clientStartup, clientReady;
    serverStartup.push_back(thread(start_remote_actor, targeter[dealerIP], true, "d", output, lambda, maliciousServers));
    this_thread::sleep_for(TWO_SECONDS);

    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, targeter[serverIPs[i]], true, "s" + to_string(i), output, lambda, maliciousServers));
    }

    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, targeter[clientIPs[i]], false, "c" + to_string(i), output, lambda, maliciousServers));
        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_created(dealerIP, dealerPort, numClients);

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

    for (size_t i = 0; i < numClients; i++)
        clientReady.push_back(thread(wait_for_client_ready, clientIPs[i], clientPorts[i]));

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

    /*
     * MAIN ORCHESTRATOR LOOP CODE
     */

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

    execute_experiment(rng, dealerIP, dealerPort, serverIPs, serverPorts, clientIPs, clientPorts, (configDir + "/commands.cfg").c_str());

    /*
     * SHUTDOWN CODE
     */

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

    shut_down_remote_actors(clientIPs, clientPorts);

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

    shut_down_remote_actors(serverIPs, serverPorts);

    return 0;
}