Minor changes

4 files changed, 722 insertions, 0 deletions

diff --git a/simulations/src/Inputs.cc b/simulations/src/Inputs.cc
new file mode 100644
index 0000000..8c3ac9c
--- /dev/null
+++ b/simulations/src/Inputs.cc
@@ -0,0 +1,184 @@
+#include "Inputs.hpp"
+
+#include <algorithm>
+#include <iostream>
+#include <fstream>
+
+Inputs::Inputs(std::string node_name){
+    // Here we do all the boring stuff
+    FILE* input_file = fopen(INPUTS_FILE, "rb");
+    char input_file_buffer[JSON_BUFFER_SIZE];
+    rapidjson::FileReadStream is(input_file, input_file_buffer, sizeof(input_file_buffer));
+    d.ParseStream(is);
+    fclose(input_file);
+    
+    // Init all variables
+    is_sender=d["nodes"][node_name.c_str()]["is_sender"].GetBool();
+    use_hint=d["nodes"][node_name.c_str()]["use_hint"].GetBool();
+    data_size=d["nodes"][node_name.c_str()]["data_size"].GetInt();
+    extended=d["extended"].GetBool();
+    seed=d["seed"].GetInt();
+    hint_size=d["hint_size"].GetInt();
+    n_nodes=d["nodes"].MemberCount();
+    latency=d["latency"].GetDouble();
+
+    // Instantiate wake_ts
+    for(auto& v:d["nodes"][node_name.c_str()]["wake_ts"].GetArray()){
+        wake_ts.push_back(v.GetDouble());
+    }
+
+    // Instantiate wake_duration
+    for(auto& v:d["nodes"][node_name.c_str()]["wake_duration"].GetArray()){
+        wake_duration.push_back(v.GetDouble());
+    }
+    
+    // Identity check
+    if(wake_ts.size()<1){
+        std::cerr << "Invalid node configuration: wake_ts.size() == 0" <<std::endl;
+        exit(1);
+    }
+    if(wake_ts.size()!=wake_duration.size()){
+        std::cerr << "Invalid node configuration: wake_ts.size() != wake_duration.size()" <<std::endl;
+        exit(1);
+    }
+    if(!std::is_sorted(wake_ts.begin(),wake_ts.end())){
+        std::cerr << "Invalid node configuration: wake_ts is not sorted" <<std::endl;
+        exit(1);
+    }
+
+    // Ensure events are merged
+    MergeEvents();
+}
+
+void Inputs::MergeEvents(){
+    for(int i=0;i<(wake_ts.size()-1);i++){
+        double cur_ts=wake_ts[i];
+        double next_ts=wake_ts[i+1];
+        double cur_duration=wake_duration[i];
+        double next_duration=wake_duration[i+1];
+        // If we should merge then
+        if((cur_ts+cur_duration)>=next_ts){
+            // Create variable for convenience
+            double start=cur_ts;
+            double end=std::max(cur_ts+cur_duration,next_ts+next_duration);
+            wake_duration[i]=end-start;
+            // Now remove next event
+            wake_ts.erase(wake_ts.begin() + i + 1);
+            wake_duration.erase(wake_duration.begin() + i +1);
+            // This is not optimal. Yet it is simple :D
+            MergeEvents();
+        }
+    }
+}
+
+double Inputs::GetNextTS(){
+    // Ensure that the caller is smart
+    if(wake_duration.size()<2){
+        std::cerr << "You are trying to access to the next timestamp but it does not exists" <<std::endl;
+        exit(1);
+    }
+    return wake_ts[1];
+}
+
+double Inputs::GetNextDuration(){
+    // Ensure that the caller is smart
+    if(wake_duration.size()<2){
+        std::cerr << "You are trying to access to the next duration but it does not exists" <<std::endl;
+        exit(1);
+    }
+    return wake_duration[1];
+}
+
+void Inputs::GotoNextEvent(){
+    wake_ts.erase(wake_ts.begin());
+    wake_duration.erase(wake_duration.begin());
+}
+
+void Inputs::DumpEvents(){
+    std::cout << "Timestamps:     ";
+    for(auto a:wake_ts){
+        std::cout << std::setw(5) << a << " ";
+    }
+    std::cout << std::endl;
+    std::cout << "Wake Durations: ";
+    for(auto a:wake_duration){
+        std::cout << std::setw(5) << a << " ";
+    }
+    std::cout << std::endl;
+}   
+
+void Inputs::AddEvent(double ts, double duration){
+    // First handle timestamp
+    int pos=0;
+    for(auto it = std::begin(wake_ts); it != std::end(wake_ts); ++it) {
+        if(*it>=ts){
+            wake_ts.insert(it,ts);
+            break;
+        }
+        pos++;
+    }
+
+    // Ensure that ts and duration should not go to the end
+    if(pos==wake_ts.size()){
+        wake_ts.push_back(ts);
+        wake_duration.push_back(duration);
+    }
+    else {
+        // Handle durations here
+        int pos2=0;
+        for(auto it = std::begin(wake_duration); it != std::end(wake_duration); ++it) {
+            if(pos==pos2){
+                wake_duration.insert(it,duration);
+                break;
+            }
+            else if (it+1==std::end(wake_duration)) {
+                wake_duration.push_back(duration);
+            }
+            pos2++;
+        }
+    }
+    // Don't forget
+    MergeEvents();
+}
+
+void Inputs::GeneratePlatform(std::string p){
+    
+    // The boring stuff
+    FILE* input_file = fopen(INPUTS_FILE, "rb");
+    char input_file_buffer[JSON_BUFFER_SIZE];
+    rapidjson::FileReadStream is(input_file, input_file_buffer, sizeof(input_file_buffer));
+    rapidjson::Document d;
+    d.ParseStream(is);
+    fclose(input_file);
+    
+    // Write platform file
+    std::ofstream pf;
+    pf.open (p);
+    pf << "<?xml version='1.0'?>\n";
+    pf << "<!DOCTYPE platform SYSTEM \"http://simgrid.gforge.inria.fr/simgrid/simgrid.dtd\">\n";
+    pf << "<platform version=\"4.1\">\n    <AS id=\"AS0\" routing=\"Full\">\n";
+    pf << "    <link id=\"link\" bandwidth=\""<<d["bitrate"].GetString()<<"\" latency=\"0ms\" sharing_policy=\"SHARED\"></link>\n";
+    for (Value::ConstMemberIterator itr = d["nodes"].MemberBegin(); itr != d["nodes"].MemberEnd(); ++itr)
+    {
+        std::string name=itr->name.GetString();
+        double power_off=d["nodes"][itr->name.GetString()]["power_off"].GetDouble();
+        double power_on=d["nodes"][itr->name.GetString()]["power_on"].GetDouble();
+        double power_rx=power_on+d["nodes"][itr->name.GetString()]["power_rx"].GetDouble();
+        double power_tx=power_on+d["nodes"][itr->name.GetString()]["power_tx"].GetDouble();
+
+        // Create node
+        pf << "    <host id=\""<<name<<"\" speed=\"100.0f,100.0f,100.0f,100.0f\" pstate=\"0\">\n";
+        pf << "        <prop id=\"wattage_per_state\" value=\""<< power_off<<":"<<power_off<<", "<< power_on<<":"<<power_on<<", "<<power_rx<<":"<<power_rx<<", "<<power_tx<<":"<<power_tx<<"\" />\n";
+        pf << "        <prop id=\"wattage_off\" value=\"0\" />\n    </host>\n";
+    }
+    for (Value::ConstMemberIterator src = d["nodes"].MemberBegin(); src != d["nodes"].MemberEnd(); ++src)
+    {
+        for (Value::ConstMemberIterator dst = d["nodes"].MemberBegin(); dst != d["nodes"].MemberEnd(); ++dst)
+        {
+            if(src->name.GetString() != dst->name.GetString())
+                pf << "    <route src=\""<<src->name.GetString()<<"\" dst=\""<<dst->name.GetString()<<"\" symmetrical=\"no\"><link_ctn id=\"link\"/></route>\n";
+        }
+    }
+    pf << "    </AS>\n</platform>\n";
+    pf.close();
+}
\ No newline at end of file
diff --git a/simulations/src/Inputs.hpp b/simulations/src/Inputs.hpp
new file mode 100644
index 0000000..487686f
--- /dev/null
+++ b/simulations/src/Inputs.hpp
@@ -0,0 +1,84 @@
+#include <rapidjson/document.h>
+#include <rapidjson/filereadstream.h>
+
+#include <cstdio>
+#include <string>
+#include <vector>
+#include <iomanip>
+
+#define INPUTS_FILE "inputs.json"
+/// @brief Pay attention to this strange number, you could tear your hairs out
+#define JSON_BUFFER_SIZE 65536
+
+using namespace rapidjson;
+
+class Inputs {
+    /// @brief RapidJSON
+    Document d;
+    /// @brief Current node associated with the Inputs
+    std::string node_name;
+    /// @brief Timestamps (at which time the nodes should wake up)
+    std::vector<double> wake_ts;
+    /// @brief Wake up time durations
+    std::vector<double> wake_duration;
+    /**
+     * Recursively merge overlapping events
+     */ 
+    void MergeEvents();
+public:
+    /**
+     * Load node_name configuration
+     */
+    Inputs(std::string node_name);
+    /**
+     * Generate a SimGrid platform file from the json configuration
+     */
+    static void GeneratePlatform(std::string p);
+    /**
+     * Is there any event that remains in the queue ?
+     */
+    bool ShouldContinue(){return wake_ts.size()!=0;}
+    /**
+     * Is there another event to process ?
+     */
+    bool HasNext(){return wake_ts.size()>1 ;}
+    /**
+     * Get current event timestamp
+     */
+    double GetTS(){return wake_ts.front();}
+    /**
+     * Get current event duration
+     */
+    double GetDuration(){return wake_duration.front();}
+    /**
+     * Get next event timestamp
+     */
+    double GetNextTS();
+    /**
+     * Get next event duration
+     */
+    double GetNextDuration();
+    /**
+     * Time travel machine (note that this function is following the second principle
+     * of thermodynamics)
+     */
+    void GotoNextEvent();
+    /**
+     * Allows to add a *FUTURE* event and merge overlapping events
+     */
+    void AddEvent(double ts, double duration);
+    /**
+     * This is the timeline
+     */
+    void DumpEvents();
+
+    /// @brief These are public attributes, please take care they are fragile
+    bool is_sender;
+    bool use_hint;
+    bool extended;
+    int data_size;
+    int hint_size;
+    int seed;
+    int n_nodes;
+    double latency;
+};
\ No newline at end of file
diff --git a/simulations/src/scenarios.cc b/simulations/src/scenarios.cc
new file mode 100644
index 0000000..90a7849
--- /dev/null
+++ b/simulations/src/scenarios.cc
@@ -0,0 +1,100 @@
+#include <cstdlib>
+#include <cstring>
+#include <rapidjson/document.h>
+#include <rapidjson/filewritestream.h>
+#include <rapidjson/prettywriter.h>
+
+#include <cstdio>
+#include <iostream>
+#include <time.h>
+#include <sstream>
+
+
+#define RAND(min,max) (rand()%((max)-(min)+1)+(min))
+
+using namespace std;
+using namespace rapidjson;
+
+int main(int argc, char **argv){
+    // Setup seed
+    if(argc!=16){
+        cerr << "Usage: " << argv[0] <<
+            " <seed> <simtime> <wakeupevery> <wakeupfor> <n_nodes>" <<
+            " <extended> <hint> <poff> <pon> <prx> <ptx> <datasize> <bitrate> <hintsize> <latency>" <<
+            endl;
+        exit(1);
+    }
+
+    // Init parameters
+    int seed=atoi(argv[1]);
+    double simtime=stod(argv[2]);
+    unsigned int wakeupevery=atoi(argv[3]);
+    unsigned int wakeupfor=stoi(argv[4]);
+    unsigned int n_nodes=atoi(argv[5]);
+    bool extended=!strcmp("true",argv[6]);
+    bool hint=!strcmp("true",argv[7]);
+    double poff=stod(argv[8]);
+    double pon=stod(argv[9]);
+    double prx=stod(argv[10]);
+    double ptx=stod(argv[11]);
+    unsigned int datasize=atoi(argv[12]);
+    string bitrate(argv[13]);
+    unsigned int hintsize=atoi(argv[14]);
+    double latency=stod(argv[15]);
+
+
+    // Setup seed
+    srand(seed);
+
+    // Create document
+    Document d;
+    d.SetObject();
+    d.AddMember("seed",Value().SetInt(seed),d.GetAllocator());
+    Value bitrateValue;
+    bitrateValue.SetString(bitrate.c_str(),bitrate.size(),d.GetAllocator());
+    d.AddMember("bitrate",bitrateValue,d.GetAllocator());
+    d.AddMember("latency",latency,d.GetAllocator());
+    d.AddMember("extended",extended,d.GetAllocator());
+    d.AddMember("hint_size",hintsize,d.GetAllocator());
+
+    // Create nodes
+    Value nodes(kObjectType);
+    for(int i=0;i<n_nodes;i++){
+        Value node(kObjectType);
+        node.SetObject();
+        node.AddMember("use_hint",hint,d.GetAllocator());
+        node.AddMember("power_off",poff,d.GetAllocator());
+        node.AddMember("power_on",pon,d.GetAllocator());
+        node.AddMember("power_rx",prx,d.GetAllocator());
+        node.AddMember("power_tx",ptx,d.GetAllocator());
+        node.AddMember("is_sender",i==0,d.GetAllocator());
+        node.AddMember("data_size",datasize,d.GetAllocator());
+
+        // Setup ts and durations
+        Value ts(kArrayType);
+        Value duration(kArrayType);
+        for(unsigned int i=0;i<simtime;i+=wakeupevery){
+            ts.PushBack(Value().SetDouble(RAND(i,i+wakeupevery-wakeupfor)),d.GetAllocator());
+            duration.PushBack(Value().SetDouble(wakeupfor),d.GetAllocator());
+        }
+        node.AddMember("wake_ts",ts,d.GetAllocator());
+        node.AddMember("wake_duration",duration,d.GetAllocator());
+
+
+        // Add node to nodes
+        std::ostringstream ss;
+        ss<< "on" <<i;
+        Value key(ss.str().c_str(), d.GetAllocator());
+        nodes.AddMember(key,node,d.GetAllocator());
+    }
+    d.AddMember("nodes",nodes,d.GetAllocator());
+
+
+    // Write to stdout
+    StringBuffer buffer;
+    PrettyWriter<StringBuffer> writer(buffer);
+    d.Accept(writer);
+    cout << buffer.GetString();
+
+    return 0;
+}
\ No newline at end of file
diff --git a/simulations/src/simulator.cc b/simulations/src/simulator.cc
new file mode 100644
index 0000000..0215b32
--- /dev/null
+++ b/simulations/src/simulator.cc
@@ -0,0 +1,354 @@
+#include <simgrid/s4u.hpp>
+#include <simgrid/s4u/Mailbox.hpp>
+#include <simgrid/s4u/Host.hpp>
+#include <simgrid/plugins/energy.h>
+#include <xbt/log.h>
+
+#include <string>
+#include <sstream>
+
+#include "Inputs.hpp"
+#include "simgrid/s4u/Actor.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);
+#define CLOCK (simgrid::s4u::Engine::get_clock())
+#define CNAME (selfName.c_str())
+#define FOR(t) (t<uptime?t:uptime)
+#define ADD_EVENT(HINT) \
+    { \
+        XBT_INFO("%s add a new hint at %f for a duration of %f",CNAME,HINT->hint,HINT->duration); \
+        i.AddEvent(HINT->hint, HINT->duration); \
+    }
+#define TRACK_UPTIME(instruction) \
+    { \
+        instruction; \
+        uptime=upuntil-CLOCK; \
+        uptime=uptime > 0 ? uptime : 0; \
+    }
+/// @brief Note that we need to simulate latency our self since we need to send instantaneous messages
+#define SEND(instruction) \
+    { \
+        TRACK_UPTIME(simgrid::s4u::this_actor::sleep_for(i.latency > uptime ? uptime : i.latency)); \
+        instruction; \
+    }
+
+#define FORWARD_HINT(TRY_FORWARD_DURING) \
+    { \
+        if(hint_forward!=NULL && CLOCK<hint_forward->hint){ \
+            hint_forward->HisForward=true; \
+            hint_forward->DedicatedMailbox="hint_forward"; \
+            try { \
+                XBT_INFO("%s try to forward a hint",CNAME); \
+                TRACK_UPTIME(m->put(hint_forward,0,TRY_FORWARD_DURING)); \
+                simgrid::s4u::Mailbox *m_ded= simgrid::s4u::Mailbox::by_name(hint_forward->DedicatedMailbox); \
+                MODE_TX(); \
+                SEND(m_ded->put(hint_forward,0,uptime)); \
+                XBT_INFO("%s forward a hint successfully",CNAME); \
+            } \
+            catch(...){ \
+                XBT_INFO("%s fail to forward a hint",CNAME); \
+                MODE_ON(); \
+                uptime=upuntil-CLOCK; \
+                TRACK_UPTIME(simgrid::s4u::this_actor::sleep_for(FOR(TRY_FORWARD_DURING))); \
+            } \
+        } \
+    }
+
+
+/// @brief Required by SimGrid
+XBT_LOG_NEW_DEFAULT_CATEGORY(simulator, "[DAO] Loosely Coupled DSS");
+
+/// @brief For convenience sake
+typedef unsigned int u32;
+
+/**
+ * Data that will be exchange between the nodes
+ */
+class Payload{
+public:
+    Payload():hint(0),duration(0),HasHint(false),HisForward(false),HasData(false),DataSize(0),Abort(false){}
+    Payload(Payload &p):hint(p.hint),duration(p.duration),HasHint(p.HasHint),DedicatedMailbox(p.DedicatedMailbox),HisForward(p.HisForward),HasData(p.HasData),DataSize(p.DataSize),Abort(p.Abort){}
+    double hint; // The timestamp that should be used by the receiver
+    double duration; // The duration that should be used by the receiver
+    bool HasHint;
+    bool HasData; // This way observer could check if they want to receive data (maybe they already received data)
+    bool HisForward;
+    bool Abort; // Allow the receiver to abort a communication (if they already received the data for example) and unlock the sender
+    u32 DataSize;
+    std::string DedicatedMailbox; // Dedicated mailbox used by the sender/receiver
+};
+
+/// @brief Observation node code
+static void obs_node(std::vector<std::string> args);
+
+/**
+ * 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);
+
+    // Headline
+    XBT_INFO("-------------------------------------------------");
+    XBT_INFO("Sarting loosely coupled data dissemination experiments");
+    XBT_INFO("-------------------------------------------------");
+
+    // Init all nodes actors
+    u32 nON=simgrid::s4u::Engine::get_instance()->get_host_count();
+    for(u32 i=0;i<nON;i++){
+        std::vector<std::string> args; // No args
+        std::ostringstream ss;
+        ss<< "on" <<i;
+        simgrid::s4u::Actor::create("ON", simgrid::s4u::Host::by_name(ss.str()), obs_node, args);
+    }
+
+    // Launch the simulation
+    engine.run();
+    XBT_INFO("Simulation took %fs", simgrid::s4u::Engine::get_clock());
+    XBT_INFO("The simulated platform file is available in \"%s\"",PLATFORM_FILE);
+    return (0);
+}
+
+/**
+ * This is the brain behind each node
+ */
+static void obs_node(std::vector<std::string> args) {
+    // 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); // Load node input parameters from the json file 
+    simgrid::s4u::Mailbox *m = simgrid::s4u::Mailbox::by_name("medium");
+    XBT_INFO("Deploying observation node %s",CNAME);
+    
+    // Starting node
+    u32 nWakeUp=0;
+    u32 nDataRcv=0;
+    u32 nSendFail=0;
+    u32 nRcvFail=0;
+    u32 nSend=0;
+    u32 hint_added=0;
+    double totalUptime=0;
+    Payload *hint_forward=NULL; // Contains the hint to forward
+    bool is_sender=i.is_sender; // This variable might change if all receiver have received the data
+    bool isObserver=false;
+    double timeDataRcv=-1;
+    while(i.ShouldContinue()){
+        // Start by sleeping
+        XBT_INFO("%s is sleeping",CNAME);
+        MODE_OFF();
+        simgrid::s4u::this_actor::sleep_until(i.GetTS());
+        MODE_ON();
+        XBT_INFO("%s wakes up",CNAME);
+
+        // Doing wake up stuff
+        double uptime=i.GetDuration(); // Store the remaining wake up duration (updated during the node uptime)
+        double upsince=CLOCK; // Store the time at which the node woke up
+        double upuntil=i.GetTS()+i.GetDuration(); // Store the time at which the node should sleep
+        bool forward_mode=false; // Turned on and off every x seconds by the receiver (to switch between forward hint mode and receiving data mode)
+        bool forward_only=false; // When observer receive a hint it switch to forward only up to the next wake up time
+        bool sendhint_mode=false; // Turned on and off every x seconds by the sender (to switch between send hint and send data)
+        while(CLOCK < upuntil)
+        { 
+            // ---------- SENDER ----------
+            if(is_sender){
+                // Send hint if send hint mode is enable
+                if(i.use_hint && sendhint_mode && i.HasNext()){
+                    Payload *p=new Payload();
+                    p->DedicatedMailbox="hintmailbox"+selfName; // Use a dedicated mailbox
+                    p->HasHint=true;
+                    p->duration=i.GetNextDuration();
+                    p->hint=i.GetNextTS();
+                    p->DataSize=i.hint_size;
+                    try {
+                        TRACK_UPTIME(m->put(p,0,FOR(0.3))); // Init connection with a receiver
+                        simgrid::s4u::Mailbox *m_ded= simgrid::s4u::Mailbox::by_name(p->DedicatedMailbox);
+                        MODE_TX();
+                        SEND(m_ded->put(p,p->DataSize,uptime)); // Send the actual hint
+                        MODE_ON();
+                        XBT_INFO("%s sent a hint successfully",CNAME);
+                    }
+                    catch(...){}
+                }
+                // Send data if send hint mode is disable
+                else{
+                    Payload *p=new Payload();
+                    p->DedicatedMailbox="datamailbox"+selfName;
+                    p->HasData=true;
+                    p->HasHint=false;
+                    p->DataSize=i.data_size;
+                    // Add hint to the data if possible
+                    if(i.use_hint && i.HasNext()){
+                        p->HasHint=true;
+                        p->duration=i.GetNextDuration();
+                        p->hint=i.GetNextTS();
+                        p->DataSize+=i.hint_size; // Don't forget!!
+                    }
+                    // Send the data
+                    try {
+                        TRACK_UPTIME(m->put(p,0,FOR(1)));
+                        simgrid::s4u::Mailbox *m_ded= simgrid::s4u::Mailbox::by_name(p->DedicatedMailbox);
+                        Payload *ack=m_ded->get<Payload>();
+                        if(!ack->Abort){
+                            MODE_TX();
+                            XBT_INFO("%s try a send",CNAME);
+                            if(i.extended){
+                                SEND(m_ded->put(p,p->DataSize));
+                            }
+                            else{
+                                SEND(m_ded->put(p,p->DataSize,uptime));
+                            }
+                            XBT_INFO("%s sent data successfully",CNAME);
+                            nSend++;
+                            is_sender=(nSend<(i.n_nodes-1)); // Stop sending if all nodes received
+                            isObserver=!is_sender; // Switch to observer mode if all nodes received the data
+                        }
+                        else {
+                            simgrid::s4u::this_actor::sleep_for(FOR(1));
+                        }
+                    }
+                    catch(...){}
+                }
+                sendhint_mode=!sendhint_mode; // Switch back and forth between sending hint and data
+                MODE_ON();
+            }
+            // ---------- RECEIVER ----------
+            else if(!isObserver){
+                // Forward hint mode
+                if(forward_mode){
+                    if(i.use_hint && hint_forward!=NULL && CLOCK < hint_forward->hint){
+                        try {
+                        FORWARD_HINT(FOR(0.3)); // Try forward for 0.3 seconds then switch to received mode
+                        }
+                        catch(...){}
+                    }
+                }
+                else { // Receiving mode
+                    Payload *p; // Received data
+                    try {
+                        // Get the instantaneous message
+                        do {
+                            TRACK_UPTIME(p=m->get<Payload>(FOR(1)));
+                            if(p->HisForward){
+                                if(hint_forward==NULL || (hint_forward !=NULL && p->hint>hint_forward->hint)){
+                                    simgrid::s4u::Mailbox *m_ded=simgrid::s4u::Mailbox::by_name(p->DedicatedMailbox);
+                                    MODE_RX();
+                                    TRACK_UPTIME(p=m_ded->get<Payload>(uptime));
+                                    MODE_ON();
+                                    XBT_INFO("%s received a forwarded hint successfully",CNAME);
+                                    if(CLOCK < p->hint){
+                                        ADD_EVENT(p);
+                                        hint_forward=new Payload(*p);
+                                        hint_added++;
+                                    }
+                                }
+                            }
+                        } while(p->HisForward);
+                        simgrid::s4u::Mailbox *m_ded= simgrid::s4u::Mailbox::by_name(p->DedicatedMailbox);
+                        // Start receiving data
+                        MODE_RX();
+                        if(p->HasHint && !p->HasData){
+                            TRACK_UPTIME(p=m_ded->get<Payload>(uptime));
+                            XBT_INFO("%s received a hint successfully",CNAME);
+                            hint_forward=new Payload(*p); // Enable hint forwarding
+                            if(CLOCK < p->hint){
+                                ADD_EVENT(p);
+                                hint_forward=new Payload(*p);
+                                hint_added++;
+                            }
+                        }
+                        else {
+                            // Inform the sender that we do not want to abort
+                            Payload *ack=new Payload();
+                            ack->Abort=false;
+                            m_ded->put(ack,0); // Instantaneous msg
+   
+                            if(i.extended){
+                                p=m_ded->get<Payload>(); // Fetch data until sended
+                            }
+                            else{
+                                TRACK_UPTIME(p=m_ded->get<Payload>(uptime)); // Fetch data until sended or uptime expire
+                            }
+                            // If we reach here, data has been received successfully
+                            XBT_INFO("%s received data successfully",CNAME);
+                            timeDataRcv=CLOCK;
+                            if(p->HasHint){
+                                XBT_INFO("%s received a hint along with data successfully",CNAME);
+                                hint_forward=new Payload(*p); // Enable hint forwarding
+                            }
+                            nDataRcv++;
+                            isObserver=true;
+                            is_sender=false;
+                        }
+
+                    }catch(...){
+                        XBT_INFO("%s could not receive any data",CNAME);
+                        nRcvFail++;
+                    }
+                }
+                forward_mode=!forward_mode; // Toggle mode (go back and forth between receiving and forwarding)
+            }
+            // ---------- OBSERVER ----------
+            else {
+                XBT_INFO("%s is observing his environment...",CNAME);
+                MODE_ON();
+                // If use hint we should listen for the sender
+                if(i.use_hint){
+                    if((forward_mode|forward_only) && hint_forward!=NULL && CLOCK < hint_forward->hint){
+                        FORWARD_HINT(FOR(1));
+                    }
+                    else {
+                        Payload *p;
+                        try {
+                            do {
+                                TRACK_UPTIME(p=m->get<Payload>(FOR(1)));
+                            } while(p->HisForward); // Ignore forwarded hint
+                            simgrid::s4u::Mailbox *m_ded= simgrid::s4u::Mailbox::by_name(p->DedicatedMailbox);
+                            // Start receiving hint from sender
+                            if(p->HasData){
+                                Payload *ack=new Payload();
+                                ack->Abort=true;
+                                m_ded->put(ack,0);
+                                simgrid::s4u::this_actor::sleep_for(FOR(1));
+                            }
+                            else if(p->HasHint){
+                                MODE_RX();
+                                TRACK_UPTIME(p=m_ded->get<Payload>(uptime));
+                                XBT_INFO("%s received a hint successfully",CNAME);
+                                hint_forward=new Payload(*p); // Enable hint forwarding
+                                forward_only=true;
+                            }
+                            else {
+                                simgrid::s4u::this_actor::sleep_for(FOR(1));
+                            }
+                           }
+                        catch(...){
+                        }
+                        
+                    }
+                    forward_mode=!forward_mode;
+                }
+                else {
+                    simgrid::s4u::this_actor::sleep_until(upuntil);
+                }
+            }
+            uptime=upuntil-CLOCK; // Note that uptime can be < 0 in extended mode
+            uptime=uptime > 0 ? uptime : 0; // Just in case
+        }
+        // Load next event
+        i.GotoNextEvent();
+        nWakeUp++; // Increase the number of wake up
+        totalUptime+=CLOCK-upsince; // Synchronize total uptime
+    }
+    // Done
+    MODE_OFF()
+    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|hint_added:%d|timeDataRcv:%f)",CNAME,CNAME,i.is_sender,nSend,nWakeUp,nDataRcv,nSendFail,nRcvFail,totalUptime,i.seed,hint_added,timeDataRcv);
+}
\ No newline at end of file