ExternalVendorCode/trick
1
0
Fork 0
mirror of https://github.com/nasa/trick.git synced 2024-12-21 14:07:51 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity
8198ce8a7f
trick/trick_source/sim_services/MasterSlave/Master.cpp

657 lines
25 KiB
C++
Raw Normal View History

Initial commit of everything.
2015-02-26 15:02:31 +00:00
/*
PURPOSE: (Master for master/slave syncrhonization)
*/
#include <iostream>
#include <sstream>
#include <string>
#include <pwd.h>
#include <stdlib.h>
Remove duplicate stand-alone libraries I changed trick comm so that there is only one version built, not a slightly different version if you are in or out of a sim. While I was in the trick comm code, I removed all of the unnecessary system header file inclusions out of tc.h. I modified each of the source files to include the system files it requires. Some sim_services files were including tc.h and also had to be edited to add headers. I removed the stand-alone makefiles out of all of the trick_utils directories. Finally I modified the master makefile to exclude the trick_util directories comm, math, and units from being archived into libtrick.a. Each of those directories will create their own library and will be included when linking a Trick sim. refs #71
2015-06-24 20:58:17 +00:00
#include <unistd.h>
Initial commit of everything.
2015-02-26 15:02:31 +00:00
#include <sys/types.h>
#include <sys/wait.h>
Standardize directory names Reorganized. Created a new top level include directory that will hold all of Trick's header files. Moved all of the Trick headers to this directory. Created a libexec directory that holds all of the executables that users don't need to execute directly. Changed all of the executables remaining in bin to start with "trick-". In the sim_services directories changed all source files to find the Trick headers in their new location. Since all of the include files are gone in sim_services, removed the src directories as well, moving all of the source files up a level. Moved the makefiles, docs, man, and other architecture independent files into a top level share directory. Renamed lib_${TRICK_HOST_CPU} to lib64 or lib depending on the platform we're currently on. refs #63
2015-06-01 15:28:29 +00:00
#include "trick/Master.hh"
#include "trick/master_proto.h"
#include "trick/MSSocket.hh"
#include "trick/MSSharedMem.hh"
#include "trick/exec_proto.h"
#include "trick/sim_mode.h"
#include "trick/command_line_protos.h"
#include "trick/unix_commands.h"
#include "trick/message_proto.h"
#include "trick/message_type.h"
#include "trick/env_proto.h"
#include "trick/CheckPointRestart_c_intf.hh" // for checkpoint_get_output_file
Initial commit of everything.
2015-02-26 15:02:31 +00:00
Trick::Master * the_ms_master ;
Trick::SlaveInfo::SlaveInfo() {
enabled = true ;
activated = false ;
slave_type = "undefined";
/* default the slaves to start with ssh */
remote_shell = TRICK_SSH ;
/* default the master not to quit if a slave dies */
sync_error_terminate = false ;
/* default the slaves to start on the localhost */
machine_name = "localhost" ;
sync_wait_limit = 0.0 ;
reconnect_wait_limit = 0.0 ;
reconnect_count = 0;
chkpnt_dump_auto = true ;
chkpnt_load_auto = true ;
chkpnt_binary = false ;
}
int Trick::SlaveInfo::set_connection_type(Trick::MSConnect * in_connection) {
connection = in_connection ;
return(0) ;
}
int Trick::SlaveInfo::start() {
int arg_i;
char *display;
std::stringstream startup_command ;
std::stringstream temp_stream ;
struct passwd *passp;
int argc ;
char ** argv ;
std::string slave = "undefined";
/** @par Detailed Design */
argc = command_line_args_get_argc() ;
argv = command_line_args_get_argv() ;
/** @li If the slave connection type is not specified return an error */
if ( connection == NULL ) {
message_publish(MSG_ERROR , "Slave connection type not specified.\n") ;
return(-1) ;
}
/** @li Start remote startup command with remote shell to use and remote machine name*/
switch ( remote_shell ) {
case TRICK_RSH:
startup_command << unix_rsh << " " << remote_shell_args ;
break;
case TRICK_USER_REMOTE_SH:
if ( user_remote_shell.empty() ) {
message_publish(MSG_WARNING , "TRICK_USER_REMOTE_SH specified for Slave startup, but no shell given.\nDefaulting to %s.\n", unix_ssh) ;
user_remote_shell = unix_ssh ;
}
startup_command << user_remote_shell << " " << remote_shell_args ;
break;
case TRICK_SSH:
default:
startup_command << unix_ssh << " " << remote_shell_args ;
break;
}
startup_command << " " << machine_name ;
/** @li Add the remote display. If a remote display has not been specified, use the environment's */
if ( machine_display.empty() ) {
if ((display = (char *) env_get_var("DISPLAY")) == NULL) {
message_publish(MSG_ERROR, "Cannot get environment variable $DISPLAY for Slave.\n") ;
} else {
machine_display = display ;
}
}
/** @li cd to the simulation path in the remote startup command */
if ( sim_path.empty() ) {
message_publish(MSG_ERROR, "Slave startup sim_path is empty.\n") ;
return(-1) ;
}
/** @li Set the DISPLAY environment variable in the remote startup command.
Use setenv for *csh or export for all other shells */
passp = getpwuid(getuid());
if ( ! machine_display.empty() ) {
if (strstr(passp->pw_shell, "csh")) {
startup_command << " 'setenv DISPLAY " << machine_display ;
} else {
startup_command << " 'export DISPLAY=" << machine_display ;
}
startup_command << " ; cd " << sim_path << " ; " ;
} else {
startup_command << " 'cd " << sim_path << " ; " ;
}
if (strstr(passp->pw_shell, "csh")) {
startup_command << " setenv TRICK_HOST_CPU `gte TRICK_HOST_CPU` ; " ;
} else {
startup_command << " export TRICK_HOST_CPU=`gte TRICK_HOST_CPU` ; " ;
}
/** @li start the simulation in the remote startup command */
if ( S_main_name.empty() ) {
S_main_name = "./S_main_${TRICK_HOST_CPU}.exe" ;
}
startup_command << S_main_name ;
/** @li add the user provided run input file if provided */
if ( ! run_input_file.empty() ) {
startup_command << " " << run_input_file ;
/** @li check to see if master is running with dmtcp slave */
if (run_input_file.find("dmtcp") != std::string::npos)
slave_type = "dmtcp";
}
/** @li Add the connection specific arguments to the startup command */
startup_command << " " << connection->add_sim_args( slave_type ) ;
/* @li Add the additional user arguments to the remote command */
if ( ! other_args.empty() ) {
startup_command << " -u " << other_args ;
}
for (arg_i = 2; arg_i < argc ; arg_i++) {
startup_command << " " << argv[arg_i] ;
}
/* @li start the remote command in the background */
startup_command << "' &" ;
message_publish(MSG_INFO, startup_command.str().c_str()) ;
/* @li Execute the startup command */
if (system(startup_command.str().c_str())) {
perror("Slave_start");
return (-2) ;
}
/* @li Wait for the slave to connect to the master */
connection->accept() ;
/* @li Set the synchronization wait limit */
connection->set_sync_wait_limit(sync_wait_limit) ;
/* Set slave to activated */
activated = true;
return (0) ;
}
int Trick::SlaveInfo::read_slave_status() {
MS_SIM_COMMAND slave_command ;
MS_SIM_COMMAND exec_command ;
/** @par Detailed Design: */
/** @li If the slave is an active synchronization partner (activated == true) */
if (activated == true) {
/** @li read the current slave exec_command */
slave_command = connection->read_command() ;
//printf("DEBUG master read %d command from slave\n", slave_command);fflush(stdout);
exec_command = (MS_SIM_COMMAND)exec_get_exec_command() ;
// fixup: is it possible we won't get slave's Exit command over socket when it terminates?, set it here if that happens
if (dynamic_cast<MSSocket*>(connection)) {
if ((slave_command == MS_ErrorCmd) && (reconnect_wait_limit > 0.0) && (reconnect_count == 0)) {
slave_command = MS_ExitCmd;
}
}
/** @li If the master is not currently exiting, change modes if the slave is freezing/exiting or has an error */
if ( exec_command != MS_ExitCmd ) {
switch ( slave_command ) {
case (MS_ErrorCmd):
/** @li if the user has set a reconnect_wait_limit, continue on if we are still under that limit, otherwise
if the current slave mode cannot be read, exit the master if sync_error_terminate == true,
otherwise set the activated flag to false */
if ( (reconnect_count * exec_get_freeze_frame()) < reconnect_wait_limit) {
reconnect_count++;
} else if (sync_error_terminate == true) {
message_publish(MSG_ERROR, "Master lost sync with slave, so master is terminating.\n") ;
exec_terminate_with_return(-1, __FILE__, __LINE__ , "Master lost sync with slave.") ;
}
else {
message_publish(MSG_ERROR, "Master lost sync with slave, so slave is being deactivated.\n") ;
activated = false ;
return(0) ;
}
break ;
case (MS_ExitCmd):
/** @li if the current slave mode is exiting, exit the master if sync_error_terminate == true.
otherwise wait for slave to reconnect. when wait limit is 0, set the activated flag to false */
if (sync_error_terminate == true){
message_publish(MSG_WARNING, "sync_error_terminate is true: Slave is exiting, so master is terminating.\n") ;
exec_terminate_with_return(-1, __FILE__, __LINE__ , "Slave is exiting, so is the master.") ;
}
else {
message_publish(MSG_WARNING, "Slave is exiting.\n") ;
// if reconnect_wait_limit is set, master waits for slave to reconnect (e.g. dmtcp restarting)
if (reconnect_wait_limit > 0.0) {
message_publish(MSG_WARNING, "Master will wait %f seconds for slave to reconnect.\n", reconnect_wait_limit) ;
// make reads (shared mem connection) return quickly so we don't overrun waiting for reconnect
// TODO: for socket connection we will overrun in the accept call (see restart_dmtcp_slave)
connection->set_sync_wait_limit(exec_get_freeze_frame());
if (chkpnt_binary) {
restart_dmtcp_slave(); // restart the slave dmtcp executable
}
}
else {
message_publish(MSG_WARNING, "reconnect_wait_limit: 0.0 - Master will stop communicating with slave.\n") ;
activated = false ;
}
return(0) ;
}
break ;
case (MS_ChkpntLoadBinCmd):
// slave has received our load command and is now sending us his dmtcp port and checkpoint file name
dmtcp_port = connection->read_port() ;
connection->read_name(chkpnt_name, sizeof(chkpnt_name)); // dir/filename
message_publish(MSG_WARNING , "Master received DMTCP Port and Checkpoint Filename from slave.\n");
connection->write_command((MS_SIM_COMMAND)exec_get_exec_command()) ; // send this as an ack so slove can shut down
break ;
case (MS_FreezeCmd):
/** @li if the current slave is freezing, freeze the master too */
message_publish(MSG_INFO, "Slave is freezing.\n") ;
exec_set_exec_command(FreezeCmd) ;
reconnect_count = 0;
break ;
case (MS_ReconnectCmd):
// set the sync wait limit back to its default
connection->set_sync_wait_limit(sync_wait_limit);
message_publish(MSG_INFO, "Master has reconnected to slave.\n") ;
reconnect_count = 0;
break ;
default:
break ;
}
}
}
return(0) ;
}
int Trick::SlaveInfo::write_master_status() {
/** @par Detailed Design: */
/** @li If the slave is an active synchronization partner (activated == true) */
/** @li and we are not currently waiting for slave to reconnect, */
if (( activated == true ) && (reconnect_count == 0)) {
/** @li write the current time according to the master to the slave */
connection->write_time(exec_get_time_tics()) ;
/** @li write the current exec_command according to the master to the slave */
connection->write_command((MS_SIM_COMMAND)exec_get_exec_command()) ;
}
if ((MS_SIM_COMMAND)exec_get_exec_command() == MS_ChkpntLoadBinCmd) {
// dmtcp slave will exit, so stop writing status to slave until it reconnects
// reconnect_count prevents us from writing status to slave, & is incremented every freeze cycle until we have reconnected
reconnect_count = 1;
}
return(0) ;
}
int Trick::SlaveInfo::write_master_chkpnt_name(std::string full_path_name) {
/** @par Detailed Design: */
/** @li If the slave is an active synchronization partner (activated == true) */
if ( activated == true ) {
/** @li write the checkpoint dir/filename to the slave */
if (full_path_name.length() > sizeof(chkpnt_name)-1) {
message_publish(MSG_ERROR, "Master could not send checkpoint name to slave because name too long (max = %d).\n",
sizeof(chkpnt_name)) ;
chkpnt_name[0] = MS_ERROR_NAME; // send error character
} else {
strcpy(chkpnt_name, full_path_name.c_str());
}
connection->write_name(chkpnt_name, sizeof(chkpnt_name)) ;
}
return(0) ;
}
Trick::Master::Master() {
enabled = false ;
the_ms_master = this ;
num_slaves = 0 ;
}
int Trick::Master::enable() {
enabled = true ;
return(0);
}
int Trick::Master::disable() {
enabled = false ;
return(0);
}
Trick::SlaveInfo * Trick::Master::add_slave(Trick::SlaveInfo * new_slave) {
/** @par Detailed Design: */
/** @li if the incoming parameter is NULL or not specified allocate space for a new slave */
if (new_slave == NULL ) {
new_slave = new Trick::SlaveInfo ;
}
/** @li push the slave onto the vector of slaves */
slaves.push_back(new_slave) ;
num_slaves++ ;
/** @li return the address of the new slave */
return(new_slave) ;
}
int Trick::Master::process_sim_args() {
return(0) ;
}
int Trick::Master::init() {
unsigned int ii ;
/** @li Call Trick::SlaveInfo::start() for each slave */
if ( enabled ) {
for ( ii = 0 ; ii < slaves.size() ; ii++ ) {
slaves[ii]->start() ;
/** @li Write the master's software frame to each slave */
slaves[ii]->connection->write_time(exec_get_software_frame_tics()) ;
/** @li Write the master's time tic value to each slave */
slaves[ii]->connection->write_time(exec_get_time_tic_value()) ;
/** @li Write the master's sync_wait_limit to each slave */
slaves[ii]->connection->write_time((long long)(slaves[ii]->sync_wait_limit * exec_get_time_tic_value())) ;
/** @li Cast the freeze command and write it to the slave */
slaves[ii]->connection->write_command((MS_SIM_COMMAND)exec_get_freeze_command()) ;
/** @li Write a flag word containing the checkpoint pre_init, post_init, and end flags to the slave */
slaves[ii]->connection->write_time((long long) ((get_checkpoint_pre_init() << 2) +
(get_checkpoint_post_init() << 1) +
(get_checkpoint_end())) );
}
// Freezes are only allowed on frame boundaries when Master/Slave is enabled.
exec_set_freeze_on_frame_boundary(true) ;
}
return(0) ;
}
/**
@details
-# Read the status of all slaves
*/
int Trick::Master::end_of_frame_status_from_slave() {
unsigned int ii ;
if ( enabled ) {
for ( ii = 0 ; ii < slaves.size() ; ii++ ) {
slaves[ii]->read_slave_status() ;
}
}
return(0) ;
}
/**
@details
-# Write the master status to all slaves
*/
int Trick::Master::end_of_frame_status_to_slave() {
unsigned int ii ;
if ( enabled ) {
for ( ii = 0 ; ii < slaves.size() ; ii++ ) {
slaves[ii]->write_master_status() ;
}
}
return(0) ;
}
int Trick::Master::freeze_init() {
unsigned int ii ;
Trick::SlaveInfo * curr_slave ;
if ( enabled ) {
/** @par Detailed Design: */
/** @li Set the connection sync_wait_limit to infinite as we enter freeze */
for ( ii = 0 ; ii < slaves.size() ; ii++ ) {
curr_slave = slaves[ii] ;
if ( curr_slave->activated == true ) {
curr_slave->connection->set_sync_wait_limit(-1.0) ;
}
}
}
return(0) ;
}
int Trick::Master::freeze() {
/** @par Detailed Design: */
/** @li Sync with slave(s) */
end_of_frame_status_from_slave() ;
end_of_frame_status_to_slave() ;
return(0) ;
}
int Trick::Master::checkpoint() {
/** @par Detailed Design: */
/** @li If chkpnt_dump_auto, tell slave to dump a checkpoint */
unsigned int ii ;
// do not tell slave to dump if this is a pre_init, post_init, or end checkpoint
// those are handled with flags sent to slave in init()
if ((exec_get_mode() == Initialization) || (exec_get_mode() == ExitMode)) {
return(0);
}
if (enabled) {
// Use 2 loops to read all slave status before writing any status out.
for ( ii = 0 ; ii < slaves.size() ; ii++ ) {
slaves[ii]->read_slave_status() ;
}
SIM_COMMAND save_command = exec_get_exec_command() ;
std::string full_path_name = checkpoint_get_output_file();
for ( ii = 0 ; ii < slaves.size() ; ii++ ) {
if (slaves[ii]->chkpnt_dump_auto) {
if (slaves[ii]->chkpnt_binary) {
if (slaves[ii]->slave_type == "dmtcp") {
exec_set_exec_command((SIM_COMMAND)MS_ChkpntDumpBinCmd) ;
slaves[ii]->write_master_status() ;
slaves[ii]->write_master_chkpnt_name(full_path_name) ;
exec_set_exec_command(save_command) ;
} else {
message_publish(MSG_ERROR, "Slave is not running under dmtcp control so it cannot dump binary checkpoint.\n") ;
slaves[ii]->write_master_status() ;
}
} else { // ascii
exec_set_exec_command((SIM_COMMAND)MS_ChkpntDumpAsciiCmd) ;
slaves[ii]->write_master_status() ;
slaves[ii]->write_master_chkpnt_name(full_path_name) ;
exec_set_exec_command(save_command) ;
}
} else { // no auto dump
slaves[ii]->write_master_status() ;
}
}
}
return(0) ;
}
int Trick::Master::preload_checkpoint() {
/** @par Detailed Design: */
/** @li If chkpnt_load_auto, tell slave to load a checkpoint */
unsigned int ii ;
if (enabled) {
// Use 2 loops to read all slave status before writing any status out.
for ( ii = 0 ; ii < slaves.size() ; ii++ ) {
slaves[ii]->read_slave_status() ;
}
SIM_COMMAND save_command = exec_get_exec_command() ;
std::string full_path_name = checkpoint_get_load_file();
for ( ii = 0 ; ii < slaves.size() ; ii++ ) {
if (slaves[ii]->chkpnt_load_auto) {
if (slaves[ii]->chkpnt_binary) {
if (slaves[ii]->slave_type == "dmtcp") {
exec_set_exec_command((SIM_COMMAND)MS_ChkpntLoadBinCmd) ;
slaves[ii]->write_master_status() ;
slaves[ii]->write_master_chkpnt_name(full_path_name) ;
exec_set_exec_command(save_command) ;
} else {
message_publish(MSG_ERROR, "Slave is not running under dmtcp control so it cannot load binary checkpoint.\n") ;
slaves[ii]->write_master_status() ;
}
} else { // ascii
exec_set_exec_command((SIM_COMMAND)MS_ChkpntLoadAsciiCmd) ;
slaves[ii]->write_master_status() ;
slaves[ii]->write_master_chkpnt_name(full_path_name) ;
exec_set_exec_command(save_command) ;
}
} else { // no auto load
slaves[ii]->write_master_status() ;
}
}
}
return(0) ;
}
int Trick::Master::unfreeze() {
unsigned int ii ;
Trick::SlaveInfo * curr_slave ;
if ( enabled ) {
for ( ii = 0 ; ii < slaves.size() ; ii++ ) {
/** @par Detailed Design: */
/** @li Set the connection sync_wait_limit to the default for each slave */
curr_slave = slaves[ii] ;
if ( curr_slave->activated == true ) {
curr_slave->connection->set_sync_wait_limit(curr_slave->sync_wait_limit) ;
}
}
}
return(0) ;
}
int Trick::Master::shutdown() {
/** @par Detailed Design: */
/** @li Sync with slave(s) */
if ( enabled ) {
end_of_frame_status_from_slave() ;
end_of_frame_status_to_slave() ;
}
return(0) ;
}
int Trick::SlaveInfo::restart_dmtcp_slave() {
#ifdef _DMTCP
FILE *fp;
char *dmtcp_path, line[256];
std::string config_file;
std::string dmtcp_command;
std::stringstream dmtcp_port_str;
pid_t pid, dmtcp_pid;
/** @par Detailed Design: */
if ( enabled ) {
if (slave_type != "dmtcp") {
message_publish(MSG_ERROR, "Cannot auto-start slave because it was not running under dmtcp control.\n") ;
return(0);
}
/** @li If chkpnt_load_auto is specified, restart the slave by executing the user-supplied chkpnt_name... */
if (chkpnt_load_auto) {
if (chkpnt_name[0] == MS_ERROR_NAME) {
message_publish(MSG_WARNING, "Cannot auto-start slave because master did not receive chkpnt_name from slave.\n");
} else {
/** @li First kill slave's dmtcp_coordinator because sometimes it does not quit like it's supposed to. */
if (dmtcp_port > 0) { // slave sends 0 if it can't get the port num from the environment
/** @li Get dmtcp path from trick's configure output file (dmtcp is only supported in linux). */
config_file = std::string(getenv("TRICK_HOME")) + "/config_Linux.mk";
if ((fp = fopen(config_file.c_str() , "r")) != NULL ) {
while (fgets(line, sizeof(line), fp) != NULL) {
if (strncmp(line, "DMTCP", 5)==0) {
dmtcp_path = strchr(line, '/');
dmtcp_path[strlen(dmtcp_path)-1] = '\0'; // remove newline character
break;
}
}
}
/** @li Issue a dmtcp_command to kill the dmtcp_coordinator. */
fprintf(stderr, "Master attempting to kill slave's dmtcp_coordinator port= %ld"
" (it may not exist, that's ok)\n", dmtcp_port);
//dmtcp_command.str(""); // reset our command string
dmtcp_command = dmtcp_path + std::string("/bin/dmtcp_command");
if (access(dmtcp_command.c_str(), F_OK) != 0) {
fprintf(stderr, "\nCould not find %s in order to kill the dmtcp_coordinator.\n",
dmtcp_command.c_str());
} else {
//dmtcp_command << " --quiet -p " << dmtcp_port << " q";
message_publish(MSG_WARNING, "Restarting DMTCP coordinator\n");
if((dmtcp_pid = fork()) == 0) {
setsid();
dmtcp_port_str << dmtcp_port;
int execReturn = execl(dmtcp_command.c_str(), "dmtcp_command", "--quiet", "-p", dmtcp_port_str.str().c_str(), "q", NULL);
_Exit(0);
} else {
int f_status = 0;
if(dmtcp_pid > 0) {
waitpid(dmtcp_pid, &f_status, 0);
} else {
message_publish(MSG_ERROR, "Unable to send DMTCP restart command\n");
}
}
//system(dmtcp_command.str().c_str());
}
} // end if dmtcp_port > 0
/** @li Finally invoke the slave's dmtcp checkpoint script. */
message_publish(MSG_WARNING, "Auto-starting slave: %s.\n", chkpnt_name);
if ((pid = fork()) == 0) {
setsid();
std::istringstream sChkpnt(chkpnt_name);
std::string fileName;
while (std::getline(sChkpnt, fileName, '/'));
//fprintf(stderr, "------> Starting: %s\n", fileName.c_str());
int execReturn = execl(chkpnt_name, fileName.c_str(), NULL);
_Exit(0);
}
}
} // end chkpnt_auto
/** @li If our connection is a socket, disconnect the socket and call accept again */
if (dynamic_cast<MSSocket*>(connection)) {
connection->disconnect();
//TODO: this will block until slave restarts, possibly causing overruns in freeze mode
connection->accept();
}
reconnect_count = 0; // start writing status to slave again
}
#endif
return(0) ;
}
/**
* @relates Trick::Master
* C binded function to toggle the master/slave synchronization flag to on.
* @return always 0
*/
extern "C" int ms_master_enable() {
the_ms_master->enable() ;
return(0) ;
}
/**
* @relates Trick::Master
* C binded function to toggle the master/slave synchronization flag to off.
* @return always 0
*/
extern "C" int ms_master_disable() {
the_ms_master->disable() ;
return(0) ;
}
Reference in New Issue Copy Permalink