#include #include #include #include #include #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& relevantIPs, const std::vector& 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 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 :` 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 : :` 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); }