loosely-coupled-dss/src/simulator.cc

220 lines
8.5 KiB
C++
Raw Normal View History

2021-05-06 17:46:34 +02:00
#include <simgrid/s4u.hpp>
#include <simgrid/s4u/Mailbox.hpp>
2021-05-06 17:46:34 +02:00
#include <simgrid/s4u/Host.hpp>
#include <simgrid/plugins/energy.h>
#include <xbt/log.h>
#include <string>
#include <sstream>
2021-05-06 17:46:34 +02:00
#include "Inputs.hpp"
#define PLATFORM_FILE "platform.xml"
#define MODE_OFF() simgrid::s4u::this_actor::get_host()->set_pstate(0);
#define MODE_ON() simgrid::s4u::this_actor::get_host()->set_pstate(1);
#define MODE_RX() simgrid::s4u::this_actor::get_host()->set_pstate(2);
#define MODE_TX() simgrid::s4u::this_actor::get_host()->set_pstate(3);
2021-05-06 16:25:14 +02:00
/// @brief Required by SimGrid
XBT_LOG_NEW_DEFAULT_CATEGORY(simulator, "[DAO]");
2021-05-07 09:07:20 +02:00
/// @brief For convenience sake
typedef unsigned int u32;
2021-05-06 16:25:14 +02:00
/**
* Data that will be exchange between the nodes
*/
2021-05-06 11:57:50 +02:00
class Payload{
2021-05-06 16:25:14 +02:00
public:
2021-05-11 08:53:11 +02:00
Payload():hint(0),duration(0),containsHint(false),size(0){}
double hint;
2021-05-06 16:14:57 +02:00
double duration;
2021-05-11 08:53:11 +02:00
int size;
2021-05-06 11:57:50 +02:00
bool containsHint;
std::string node;
2021-05-06 11:57:50 +02:00
};
2021-05-06 16:25:14 +02:00
/// @brief Observation node code
static void obs_node(std::vector<std::string> args);
2021-05-06 16:25:14 +02:00
/**
* No arguments are require (cf inputs.json)
*/
int main(int argc, char **argv) {
// Build engine
sg_host_energy_plugin_init();
simgrid::s4u::Engine engine(&argc, argv);
Inputs::GeneratePlatform(PLATFORM_FILE);
engine.load_platform(PLATFORM_FILE);
2021-05-06 16:25:14 +02:00
// Headline
XBT_INFO("-------------------------------------------------");
XBT_INFO("Sarting loosely coupled data dissemination experiments");
XBT_INFO("-------------------------------------------------");
2021-05-06 16:25:14 +02:00
// Init all nodes actors
2021-05-06 11:57:50 +02:00
u32 nON=simgrid::s4u::Engine::get_instance()->get_host_count();
for(u32 i=0;i<nON;i++){
std::vector<std::string> args;
std::ostringstream ss;
ss<< "on" <<i;
simgrid::s4u::Actor::create("ON", simgrid::s4u::Host::by_name(ss.str()), obs_node, args);
}
2021-05-07 09:07:20 +02:00
// Launch the simulation
engine.run();
XBT_INFO("Simulation took %fs", simgrid::s4u::Engine::get_clock());
2021-05-06 16:25:14 +02:00
XBT_INFO("The simulated platform file is available in \"%s\"",PLATFORM_FILE);
return (0);
}
2021-05-06 16:25:14 +02:00
/**
* This is the brain behind each node
*/
static void obs_node(std::vector<std::string> args) {
2021-05-06 16:25:14 +02:00
// Init various variables
std::string selfName = simgrid::s4u::this_actor::get_host()->get_name();
simgrid::s4u::this_actor::get_host()->turn_on();
Inputs i(selfName);
simgrid::s4u::Mailbox *m = simgrid::s4u::Mailbox::by_name("medium");
XBT_INFO("Deploying observation node %s",selfName.c_str());
2021-05-06 17:46:34 +02:00
2021-05-06 16:25:14 +02:00
// Init convenient variables
bool isSender=i.is_sender;
2021-05-06 11:57:50 +02:00
bool useHint=i.use_hint;
bool isObserver=false;
u32 data_size=i.data_size;
// Starting node
2021-05-11 17:17:49 +02:00
bool hintReceived=false;
Payload *hint;
2021-05-06 16:14:57 +02:00
u32 nWakeUp=0;
u32 nDataRcv=0;
u32 nSendFail=0;
u32 nRcvFail=0;
2021-05-09 11:07:16 +02:00
u32 nSend=0;
2021-05-10 16:03:46 +02:00
double totalUptime=0;
2021-05-06 16:14:57 +02:00
while(i.ShouldContinue()){
XBT_INFO("%s is spleeping",selfName.c_str());
MODE_OFF();
2021-05-06 16:14:57 +02:00
simgrid::s4u::this_actor::sleep_until(i.GetTS());
MODE_ON();
XBT_INFO("%s wakes up",selfName.c_str());
2021-05-06 16:14:57 +02:00
// Doing wake up stuff
2021-05-09 11:07:16 +02:00
double uptime=i.GetDuration();
2021-05-10 16:03:46 +02:00
double startUptime=simgrid::s4u::Engine::get_clock();
2021-05-09 11:07:16 +02:00
while(uptime>0.00001){ // Ensure not infinite loop even if it should not happend (we loose accuracy here but just in case)
try {
if(isSender){ // If I am a sender
2021-05-11 17:17:49 +02:00
if(useHint&&i.HasNext()){
Payload *phint=new Payload();
phint->containsHint=true;
phint->hint=i.GetNextTS();
phint->duration=i.GetNextDuration();
phint->size=i.hint_size;
// Send data and update uptime
double hint_duration=simgrid::s4u::Engine::get_clock();
m->put(phint,phint->size,uptime);
hint_duration=simgrid::s4u::Engine::get_clock()-hint_duration;
hint_duration=hint_duration>0 ? hint_duration : 0; // Ensure valid hint_duration
uptime-=hint_duration;
uptime=uptime>0 ? uptime:0; // Ensure valid uptime
}
2021-05-09 11:07:16 +02:00
Payload *p=new Payload();
p->node=selfName;
2021-05-11 08:53:11 +02:00
p->size=data_size;
2021-05-09 11:07:16 +02:00
if(i.extended){
// We use a trick here
// First we send an instantaneous message (size=0) with the usual timeout
// to check whether there is a receiver!
// If there is one, we are sure that a put in the "medium"+selfName
// will not lead to a deadlock (cf anchor:5623) and we are using a exclusive
// channel (to avoid other receivers to get the message)
m->put(p,0,uptime);
simgrid::s4u::Mailbox *m_ext= simgrid::s4u::Mailbox::by_name("medium"+selfName);
MODE_TX();
2021-05-11 08:53:11 +02:00
m_ext->put(p,p->size);
2021-05-09 11:07:16 +02:00
}
else{
MODE_TX();
2021-05-11 08:53:11 +02:00
m->put(p,p->size,uptime);
2021-05-09 11:07:16 +02:00
}
nSend++;
isSender=(nSend<(i.n_nodes-1));
isObserver=!isSender;
XBT_INFO("%s sent data successfully",selfName.c_str());
2021-05-11 17:17:49 +02:00
2021-05-06 11:57:50 +02:00
}
2021-05-09 11:07:16 +02:00
else if (!isObserver){
Payload* p;
if(i.extended){
// anchor:5623 We can see here that
// we first receive the instantaneous message
// and then we use a mailbox specific to the sender (to have an exclusive channel)
p=m->get<Payload>(uptime);
2021-05-11 17:17:49 +02:00
if(p->containsHint){
hintReceived=true;
hint=p;
XBT_INFO("%s received and hint of %f",selfName.c_str(),p->hint);
}
else {
simgrid::s4u::Mailbox *m_ext_sender = simgrid::s4u::Mailbox::by_name("medium"+p->node);
MODE_RX();
p=m_ext_sender->get<Payload>();
}
2021-05-09 11:07:16 +02:00
}
else{
MODE_RX();
p=m->get<Payload>(uptime);
2021-05-11 17:17:49 +02:00
nDataRcv++; // New data received
if(p->containsHint){
hintReceived=true;
hint=p;
XBT_INFO("%s received and hint of %f",selfName.c_str(),p->hint);
}
else{
isObserver=true; // Now we received the data we switch to observer
XBT_INFO("%s received data successfully",selfName.c_str());
}
2021-05-09 11:07:16 +02:00
}
}
2021-05-09 11:07:16 +02:00
else {
XBT_INFO("%s is observing his environment...",selfName.c_str());
MODE_ON();
simgrid::s4u::this_actor::sleep_for(uptime);
}
}
2021-05-09 11:07:16 +02:00
catch (...)
{
if(isSender){
XBT_INFO("%s could not send any data",selfName.c_str());
nSendFail++;
}
else{
XBT_INFO("%s could not receive any data",selfName.c_str());
2021-05-11 17:17:49 +02:00
if(hintReceived){
hintReceived=false;
i.AddEvent(hint->hint, hint->duration); // Schedule a new wake up time
}
2021-05-09 11:07:16 +02:00
nRcvFail++;
}
}
2021-05-09 11:07:16 +02:00
uptime=(i.GetTS()+i.GetDuration())-simgrid::s4u::Engine::get_clock();
}
2021-05-06 16:25:14 +02:00
// Load next event
2021-05-06 16:14:57 +02:00
i.GotoNextEvent();
2021-05-06 16:25:14 +02:00
nWakeUp++; // Increase the number of wake up
2021-05-10 16:03:46 +02:00
totalUptime+=simgrid::s4u::Engine::get_clock()-startUptime;
2021-05-11 17:17:49 +02:00
if(hintReceived){
hintReceived=false;
i.AddEvent(hint->hint, hint->duration); // Schedule a new wake up time
}
}
// Done
MODE_OFF()
2021-05-11 08:53:11 +02:00
XBT_INFO("Observation node %s finished [LOG2PARSE](node:%s|isSender:%d|nSend:%d|nWakeUp:%d|nDataRcv:%d|nSendFail:%d|nRcvFail:%d|totalUptime:%f|seed:%d)",selfName.c_str(),selfName.c_str(),i.is_sender,nSend,nWakeUp,nDataRcv,nSendFail,nRcvFail,totalUptime,i.seed);
}