<commit_before>
#include "SerializedInstance.h"
#include "TimeWarpSimulationManager.h"
#include "ThreadedTimeWarpSimulationManager.h"
#include "CommunicationManagerImplementationBase.h"
#include "TerminateToken.h"
#include "StartMessage.h"
#include "InitializationMessage.h"
#include "CirculateInitializationMessage.h"
#include <sstream>

using std::istringstream;
using std::ostringstream;
using std::cerr;
using std::endl;

CommunicationManagerImplementationBase::CommunicationManagerImplementationBase(
    PhysicalCommunicationLayer* physicalLayer,
    TimeWarpSimulationManager* simMgr) :
    CommunicationManager(physicalLayer, simMgr),
    mySimulationManager(simMgr) {
}

void
CommunicationManagerImplementationBase::initializeCommunicationManager() {}

void
CommunicationManagerImplementationBase::sendMessage(KernelMessage* message,
                                                    unsigned int dest) {
    message->setIncarnationNumber(numCatastrophicRollbacks);

    if (!recoveringFromCkpt) {
        myPhysicalCommunicationLayer->physicalSend(message->serialize(),
                                                   dest);
    } else {
        // When recovering from rollback, only send the restore messages.
        if (message->getDataType() == "RestoreCkptMessage") {
            myPhysicalCommunicationLayer->physicalSend(message->serialize(),
                                                       dest);
        }
    }
    delete message;
}


void
CommunicationManagerImplementationBase::routeMessage(KernelMessage* msg) {
    ASSERT(msg != NULL);
    string messageType = msg->getDataType();

    if (recoveringFromCkpt) {
        if (msg->getIncarnationNumber() < numCatastrophicRollbacks) {
            delete msg;
            return;
        }
    }

    // It's not a CHECK-IDLE MESSAGE.
    // look-up and call the appropriate receiver for this message
    //map::find returns an iterator pair, (Key, Value), the second selects the value
    CommunicatingEntity* receiver = (*(listOfReceivers.find(messageType))).second;

    if (receiver == 0) {
        cerr << "CommunicationManagerImplementationBase::routeMessage - " <<
             "don't know how to handle a message of type " << messageType << "." << endl;
        abort();
    } else {
        receiver->receiveKernelMessage(msg);
    }
}

SerializedInstance*
CommunicationManagerImplementationBase::retrieveMessageFromPhysicalLayer() {
    return myPhysicalCommunicationLayer->physicalProbeRecv();
}

unsigned int
CommunicationManagerImplementationBase::checkPhysicalLayerForMessages(int maxNum) {
    int numberOfMessagesReceived = 0;
    bool noMessagesLeft = false;

    while (numberOfMessagesReceived != maxNum && noMessagesLeft == false) {
        SerializedInstance* msg = retrieveMessageFromPhysicalLayer();
        if (msg == NULL) {
            // there are no messages at this time
            noMessagesLeft = true;<commit_msg>OFC recovery token missing problem solved.<commit_after>
#include "SerializedInstance.h"
#include "TimeWarpSimulationManager.h"
#include "ThreadedTimeWarpSimulationManager.h"
#include "CommunicationManagerImplementationBase.h"
#include "TerminateToken.h"
#include "StartMessage.h"
#include "InitializationMessage.h"
#include "CirculateInitializationMessage.h"
#include <sstream>

using std::istringstream;
using std::ostringstream;
using std::cerr;
using std::endl;

CommunicationManagerImplementationBase::CommunicationManagerImplementationBase(
    PhysicalCommunicationLayer* physicalLayer,
    TimeWarpSimulationManager* simMgr) :
    CommunicationManager(physicalLayer, simMgr),
    mySimulationManager(simMgr) {
}

void
CommunicationManagerImplementationBase::initializeCommunicationManager() {}

void
CommunicationManagerImplementationBase::sendMessage(KernelMessage* message,
                                                    unsigned int dest) {
    message->setIncarnationNumber(numCatastrophicRollbacks);

    if (!recoveringFromCkpt) {
        myPhysicalCommunicationLayer->physicalSend(message->serialize(),
                                                   dest);
    } else {
        // When recovering from rollback, only send the restore messages.
        if (message->getDataType() == "RestoreCkptMessage") {
            myPhysicalCommunicationLayer->physicalSend(message->serialize(),
                                                       dest);
        }
    }
    delete message;
}


void
CommunicationManagerImplementationBase::routeMessage(KernelMessage* msg) {
    ASSERT(msg != NULL);
    string messageType = msg->getDataType();

    if (recoveringFromCkpt) {
		if(messageType != "RestoreCkptMessage"){
        	if (msg->getIncarnationNumber() < numCatastrophicRollbacks) {
            	delete msg;
            	return;
        	}
		}
    }

    // It's not a CHECK-IDLE MESSAGE.
    // look-up and call the appropriate receiver for this message
    //map::find returns an iterator pair, (Key, Value), the second selects the value
    CommunicatingEntity* receiver = (*(listOfReceivers.find(messageType))).second;

    if (receiver == 0) {
        cerr << "CommunicationManagerImplementationBase::routeMessage - " <<
             "don't know how to handle a message of type " << messageType << "." << endl;
        abort();
    } else {
        receiver->receiveKernelMessage(msg);
    }
}

SerializedInstance*
CommunicationManagerImplementationBase::retrieveMessageFromPhysicalLayer() {
    return myPhysicalCommunicationLayer->physicalProbeRecv();
}

unsigned int
CommunicationManagerImplementationBase::checkPhysicalLayerForMessages(int maxNum) {
    int numberOfMessagesReceived = 0;
    bool noMessagesLeft = false;

    while (numberOfMessagesReceived != maxNum && noMessagesLeft == false) {
        SerializedInstance* msg = retrieveMessageFromPhysicalLayer();
        if (msg == NULL) {
            // there are no messages at this time
            noMessagesLeft = true;
{'fraction_non_alphanumeric': 0.09346581340518693, 'fraction_numerical': 0.0006736274840013472, 'mean_word_length': 3.3067440174039158, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6532396', 'n_tokens_mistral': 1592, 'n_tokens_neox': 1515, 'n_words': 332}