123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481 |
- #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;
- chrono::seconds oneSecond(1);
- chrono::seconds fiveSeconds(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);
- stringstream buffer;
- buffer << data;
- BinarySizeT numClients;
- buffer >> numClients;
- std::cout << "raw data from websocket on number clients: " << numClients.val() << std::endl;
- sync->val = numClients.val();
- 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/bin/"
- // << (server ? "startServer.sh " : "startClient.sh ") << 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(oneSecond);
- 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(fiveSeconds);
- 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)
- {
- this_thread::sleep_for(oneSecond);
- 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;
- }
- }
- 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, " :");
- 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++;
- }
- }
- 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(oneSecond);
- 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));
- 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;
- }
|