A Distributed Hash Table (DHT) for KBR protocols. More...

#include <CBR-DHT.h>

Inheritance diagram for CBRDHT:

Classes
struct	GetMapEntry
struct	PutMapEntry

Public Member Functions
	CBRDHT ()
virtual	~CBRDHT ()
Public Member Functions inherited from BaseApp
	BaseApp ()
virtual	~BaseApp ()
	virtual destructor
Public Member Functions inherited from BaseRpc
	BaseRpc ()
const NodeHandle &	getThisNode ()
	Returns the NodeHandle of this node.
simtime_t	getUdpTimeout ()
Public Member Functions inherited from RpcListener
virtual	~RpcListener ()
	destructor
Public Member Functions inherited from BaseTcpSupport
virtual void	socketDataArrived (int connId, void yourPtr, cPacket msg, bool urgent)
virtual void	socketEstablished (int connId, void *yourPtr)
virtual void	socketPeerClosed (int connId, void *yourPtr)
virtual void	socketFailure (int connId, void *yourPtr, int code)
virtual void	socketStatusArrived (int connId, void yourPtr, TCPStatusInfo status)

Protected Types
typedef std::vector< NodeHandle >	ReplicaVector

Protected Member Functions
void	initializeApp (int stage)
	initializes derived class-attributes
void	finishApp ()
	collects statistical data of derived app
void	handleTimerEvent (cMessage *msg)
bool	handleRpcCall (BaseCallMessage *msg)
	Processes Remote-Procedure-Call invocation messages.
void	handleRpcResponse (BaseResponseMessage msg, cPolymorphic context, int rpcId, simtime_t rtt)
	This method is called if an RPC response has been received.
void	handleRpcTimeout (BaseCallMessage msg, const TransportAddress &dest, cPolymorphic context, int rpcId, const OverlayKey &destKey)
	This method is called if an RPC timeout has been reached.
void	handleUpperMessage (cMessage *msg)
	handleUpperMessage gets called of handleMessage(cMessage* msg) if msg arrivedOn from_upperTier (currently msg gets deleted in this function)
void	handlePutRequest (DHTPutCall *dhtMsg)
void	handleGetRequest (CBRDHTGetCall *dhtMsg)
void	handlePutResponse (DHTPutResponse *dhtMsg, int rpcId)
void	handleGetResponse (CBRDHTGetResponse *dhtMsg, int rpcId)
void	handlePutCAPIRequest (DHTputCAPICall *capiPutMsg)
void	handleGetCAPIRequest (DHTgetCAPICall *capiGetMsg, int teamnum=0)
void	handleDumpDhtRequest (DHTdumpCall *call)
void	update (const NodeHandle &node, bool joined)
	Common API function: informs application about neighbors and own nodeID.
void	handleLookupResponse (LookupResponse *lookupMsg)
int	resultValuesBitLength (DHTGetResponse *msg)
Protected Member Functions inherited from BaseApp
int	numInitStages () const
	method to set InitStage
void	initialize (int stage)
	initializes base class-attributes
void	handleMessage (cMessage *msg)
	checks for message type and calls corresponding method
virtual void	receiveChangeNotification (int category, const cPolymorphic *details)
	callback-method for events at the NotificationBoard
virtual void	handleTransportAddressChangedNotification ()
	This method gets call if the node has a new TransportAddress (IP address) because he changed his access network.
virtual void	handleNodeLeaveNotification ()
	This method gets call **.gracefulLeaveDelay seconds before it is killed.
virtual void	handleNodeGracefulLeaveNotification ()
	This method gets call **.gracefulLeaveDelay seconds before it is killed if this node is among the gracefulLeaveProbability nodes.
void	finish ()
	collects statistical data
void	callRoute (const OverlayKey &key, cPacket *msg, const TransportAddress &hint=TransportAddress::UNSPECIFIED_NODE, RoutingType routingType=DEFAULT_ROUTING)
	Common API function: calls route-method in overlay.
void	callRoute (const OverlayKey &key, cPacket *msg, const std::vector< TransportAddress > &sourceRoute, RoutingType routingType=DEFAULT_ROUTING)
virtual void	deliver (OverlayKey &key, cMessage *msg)
	Common API function: handles delivered messages from overlay.
virtual void	forward (OverlayKey key, cPacket msg, NodeHandle nextHopNode)
	Common API function: handles messages from overlay to be forwarded.
NodeVector *	callLocalLookup (const OverlayKey &key, int num, bool safe)
	Common API function: produces a list of nodes that can be used as next hops towards key.
NodeVector *	callNeighborSet (int num)
	Common API function: produces a list of neighbor nodes.
bool	isSiblingFor (const NodeHandle &node, const OverlayKey &key, int numSiblings, bool *err)
	Query if a node is among the siblings for a given key.
virtual void	handleLowerMessage (cMessage *msg)
	processes self-messages
virtual void	handleUDPMessage (cMessage *msg)
	method to handle messages that come directly from the UDP gate
virtual void	handleReadyMessage (CompReadyMessage *msg)
	method to handle ready messages from the overlay
virtual void	bindToPort (int port)
	Tells UDP we want to get all packets arriving on the given port.
virtual void	sendMessageToUDP (const TransportAddress &destAddr, cPacket *msg, simtime_t delay=SIMTIME_ZERO)
	Sends a packet over UDP.
virtual void	handleTraceMessage (cMessage *msg)
	handleTraceMessage gets called of handleMessage(cMessage* msg) if a message arrives at trace_in.
void	sendMessageToLowerTier (cPacket *msg)
	sends non-commonAPI message to the lower tier
bool	internalHandleRpcCall (BaseCallMessage *msg)
	Handles internal rpc requests.
void	internalHandleRpcResponse (BaseResponseMessage msg, cPolymorphic context, int rpcId, simtime_t rtt)
	Handles rpc responses internal in base classes
void	internalSendRouteRpc (BaseRpcMessage *message, const OverlayKey &destKey, const std::vector< TransportAddress > &sourceRoute, RoutingType routingType)
virtual CompType	getThisCompType ()
	Return the component type of this module.
void	sendReadyMessage (bool ready=true, const OverlayKey &nodeId=OverlayKey::UNSPECIFIED_KEY)
Protected Member Functions inherited from BaseRpc
virtual void	internalHandleRpcTimeout (BaseCallMessage msg, const TransportAddress &dest, cPolymorphic context, int rpcId, const OverlayKey &destKey)
	Handles rpc timeouts internal in base classes
void	initRpcs ()
	Initializes Remote-Procedure state.
void	finishRpcs ()
	Deinitializes Remote-Procedure state.
virtual void	internalHandleRpcMessage (BaseRpcMessage *msg)
	Handles incoming rpc messages and delegates them to the corresponding listeners or handlers.
uint32_t	sendRouteRpcCall (CompType destComp, const TransportAddress &dest, const OverlayKey &destKey, BaseCallMessage msg, cPolymorphic context=NULL, RoutingType routingType=DEFAULT_ROUTING, simtime_t timeout=-1, int retries=0, int rpcId=-1, RpcListener *rpcListener=NULL)
	Routes a Remote-Procedure-Call message to an OverlayKey.
uint32_t	sendRouteRpcCall (CompType destComp, const OverlayKey &destKey, BaseCallMessage msg, cPolymorphic context=NULL, RoutingType routingType=DEFAULT_ROUTING, simtime_t timeout=-1, int retries=0, int rpcId=-1, RpcListener *rpcListener=NULL)
	Routes a Remote-Procedure-Call message to an OverlayKey.
uint32_t	sendRouteRpcCall (CompType destComp, const TransportAddress &dest, BaseCallMessage msg, cPolymorphic context=NULL, RoutingType routingType=DEFAULT_ROUTING, simtime_t timeout=-1, int retries=0, int rpcId=-1, RpcListener *rpcListener=NULL)
	Sends a Remote-Procedure-Call message using the overlay's UDP port This replaces ROUTE_DIRECT calls!
uint32_t	sendUdpRpcCall (const TransportAddress &dest, BaseCallMessage msg, cPolymorphic context=NULL, simtime_t timeout=-1, int retries=0, int rpcId=-1, RpcListener *rpcListener=NULL)
	Sends a Remote-Procedure-Call message to the underlay
uint32_t	sendInternalRpcCall (CompType destComp, BaseCallMessage msg, cPolymorphic context=NULL, simtime_t timeout=-1, int retries=0, int rpcId=-1, RpcListener *rpcListener=NULL)
	Sends an internal Remote-Procedure-Call between two tiers
void	cancelRpcMessage (uint32_t nonce)
	Cancels a Remote-Procedure-Call.
void	cancelAllRpcs ()
	Cancels all RPCs.
void	sendRpcResponse (TransportType transportType, CompType destComp, const TransportAddress &dest, const OverlayKey &destKey, BaseCallMessage call, BaseResponseMessage response)
	Send Remote-Procedure response message and deletes call message.
void	sendRpcResponse (BaseCallMessage call, BaseResponseMessage response)
	Send Remote-Procedure response message to UDP and deletes call message.
int	pingNode (const TransportAddress &dest, simtime_t timeout=-1, int retries=0, cPolymorphic context=NULL, const char caption="PING", RpcListener *rpcListener=NULL, int rpcId=-1, TransportType transportType=INVALID_TRANSPORT)
	ping a node by its TransportAddress
virtual void	pingResponse (PingResponse pingResponse, cPolymorphic context, int rpcId, simtime_t rtt)
virtual void	pingTimeout (PingCall pingCall, const TransportAddress &dest, cPolymorphic context, int rpcId)
bool	internalHandleMessage (cMessage *msg)
Protected Member Functions inherited from RpcListener
virtual void	handleRpcResponse (BaseResponseMessage *msg, const RpcState &rpcState, simtime_t rtt)
	This method is called if an RPC response has been received.
virtual void	handleRpcTimeout (const RpcState &rpcState)
	This method is called if an RPC timeout has been reached.
Protected Member Functions inherited from BaseTcpSupport
void	handleTCPMessage (cMessage *msg)
	Member function to handle incoming TCP messages.
void	bindAndListenTcp (int port)
	Member function to bind service to the specified port and listen afterwards.
bool	isAlreadyConnected (TransportAddress address)
	Member function to check if the service is already connected.
void	establishTcpConnection (TransportAddress address)
	Member function to establish a connection to the specified node.
void	sendTcpData (cPacket *msg, TransportAddress address)
	Member function to send TCP data to the specified node.
virtual void	handleConnectionEvent (EvCode code, TransportAddress address)
	Member function to handle passive connection events.
virtual void	handleDataReceived (TransportAddress address, cPacket *msg, bool urgent)
	Member function to handle incoming data.
virtual void	handleIncomingConnection (TransportAddress address)
	Member function to handle newly opened connections.
void	closeTcpConnection (TransportAddress address)
	Member function to close an established connection.
void	setTcpOut (cGate *gate)
	Member function to set local gate towards the TCP module during init phase.
cGate *	getTcpOut ()
	Member function to get local gate towards the TCP module.

Protected Attributes
int	numReplica
uint8_t	numReplicaTeams
double	maintenanceMessages
double	normalMessages
double	numBytesMaintenance
double	numBytesNormal
double	lastGetCall
std::map< unsigned int, BaseCallMessage * >	rpcIdMap
	List of the Rpc Ids of the messages sent following the reception of an rpc request (the second member)
std::map< int, GetMapEntry >	getMap
std::map< int, PutMapEntry >	putMap
DHTDataStorage *	dataStorage
	pointer to the dht data storage
CoordBasedRouting *	coordBasedRouting
NeighborCache *	neighborCache
Protected Attributes inherited from BaseApp
UnderlayConfigurator *	underlayConfigurator
	pointer to UnderlayConfigurator in this node
GlobalNodeList *	globalNodeList
	pointer to GlobalNodeList in this node
GlobalStatistics *	globalStatistics
	pointer to GlobalStatistics module in this node
NotificationBoard *	notificationBoard
	pointer to NotificationBoard in this node
bool	debugOutput
	debug output yes/no?
int	numOverlaySent
	number of sent packets to overlay
int	bytesOverlaySent
	number of sent bytes to overlay
int	numOverlayReceived
	number of received packets from overlay
int	bytesOverlayReceived
	number of received bytes from overlay
int	numUdpSent
	number of sent packets to UDP
int	bytesUdpSent
	number of sent bytes to UDP
int	numUdpReceived
	number of received packets from UDP
int	bytesUdpReceived
	number of received bytes from UDP
simtime_t	creationTime
	simTime when the App has been created
Protected Attributes inherited from BaseRpc
NodeHandle	thisNode
	NodeHandle to this node.
BaseOverlay *	overlay
bool	debugOutput
	debug output ?
GlobalStatistics *	globalStatistics
	pointer to GlobalStatistics module in this node
CompType	thisCompType
NeighborCache *	neighborCache
	pointer to the neighbor cache
CryptoModule *	cryptoModule
	pointer to CryptoModule
int	numPingSent
int	bytesPingSent
int	numPingResponseSent
int	bytesPingResponseSent

Additional Inherited Members
Public Types inherited from BaseTcpSupport
enum	EvCode { NO_EST_CONNECTION, PEER_CLOSED, PEER_TIMEDOUT, PEER_REFUSED, CONNECTION_RESET, CONNECTION_SUCC_ClOSED }

Detailed Description

A Distributed Hash Table (DHT) for KBR protocols.

A Distributed Hash Table (DHT) for KBR protocols

Definition at line 48 of file CBR-DHT.h.

Member Typedef Documentation

typedef std::vector<NodeHandle> CBRDHT::ReplicaVector

protected

Definition at line 55 of file CBR-DHT.h.

Constructor & Destructor Documentation

CBRDHT::CBRDHT ( )

Definition at line 38 of file CBR-DHT.cc.

{
    dataStorage = NULL;
}

CBRDHT::~CBRDHT ( )

virtual

Definition at line 43 of file CBR-DHT.cc.

{
    std::map<unsigned int, BaseCallMessage*>::iterator it;
    for (it = rpcIdMap.begin(); it != rpcIdMap.end(); it++) {
        delete it->second;
        it->second = NULL;
    }
    std::map<int, GetMapEntry>::iterator it2;
    for (it2 = getMap.begin(); it2 != getMap.end(); it2++) {
        //cancelAndDelete(it2->second.callMsg);
        delete it2->second.callMsg;
        it2->second.callMsg = NULL;
    }
    std::map<int, PutMapEntry>::iterator it3;
    for (it3 = putMap.begin(); it3 != putMap.end(); it3++) {
        //if (it3->second.callMsg != NULL) {
        //    cancelAndDelete(it3->second.callMsg);
        //}
        delete it3->second.callMsg;
        it3->second.callMsg = NULL;
    }
    rpcIdMap.clear();
    getMap.clear();
    putMap.clear();
    if (dataStorage != NULL) {
        dataStorage->clear();
    }
}

Member Function Documentation

void CBRDHT::finishApp ( )

protectedvirtual

collects statistical data of derived app

Reimplemented from BaseApp.

Definition at line 914 of file CBR-DHT.cc.

{
    simtime_t time = globalStatistics->calcMeasuredLifetime(creationTime);
    if (time != 0) {
        // std::cout << dataStorage->getSize() << " " << overlay->getThisNode().getKey().toString(16) << std::endl;
        globalStatistics->addStdDev("DHT: Sent Maintenance Messages/s",
                                    maintenanceMessages / time);
        globalStatistics->addStdDev("DHT: Sent Normal Messages/s",
                                    normalMessages / time);
        globalStatistics->addStdDev("DHT: Sent Maintenance Bytes/s",
                                    numBytesMaintenance / time);
        globalStatistics->addStdDev("DHT: Sent Normal Bytes/s",
                                    numBytesNormal / time);
    }
}

void CBRDHT::handleDumpDhtRequest ( DHTdumpCall * call )

protected

Definition at line 552 of file CBR-DHT.cc.

{
    DHTdumpResponse* response = new DHTdumpResponse();
    DhtDumpVector* dumpVector = dataStorage->dumpDht();
    response->setRecordArraySize(dumpVector->size());
    for (uint i = 0; i < dumpVector->size(); i++) {
        response->setRecord(i, (*dumpVector)[i]);
    }
    delete dumpVector;
    sendRpcResponse(call, response);
}

void CBRDHT::handleGetCAPIRequest	(	DHTgetCAPICall *	capiGetMsg,
		int	teamnum = `0`
	)

protected

Definition at line 422 of file CBR-DHT.cc.

                                                                         {
    // Extended multi team version, default: teamnum = 0
        if (teamnum >= numReplicaTeams)
                return;
        OverlayKey originalKey = capiGetMsg->getKey();
        std::vector<OverlayKey> possibleKeys;
        assert(!originalKey.isUnspecified());
        possibleKeys.push_back(originalKey);
        for (int i = 1; i < numReplicaTeams; i++) {
                // multiple SHA1 hashing of original key
                OverlayKey keyHash = originalKey;
                for (int j = 0; j < i; j++) {
                        keyHash = OverlayKey::sha1(BinaryValue(keyHash.toString(16).c_str()));
                }
                assert(!keyHash.isUnspecified());
                possibleKeys.push_back(keyHash);
        }
    // Order possible keys by euclidian distance to this node
    std::vector<OverlayKey> orderedKeys;
    OverlayKey compareKey = overlay->getThisNode().getKey();
    while (possibleKeys.size() > 0) {
        OverlayKey bestKey = possibleKeys[0];
        int bestpos = 0;
        // TODO: i = 1?
        for (uint i = 0; i < possibleKeys.size(); i++) {
            //std::cout << neighborCache->getOwnEuclidianDistanceToKey(possibleKeys[i]) << std::endl;
            if (coordBasedRouting
                    ->getEuclidianDistanceByKeyAndCoords(possibleKeys[i],
                                                         ((const Nps&)neighborCache->getNcsAccess()).getOwnCoordinates(), //TODO
                                                         overlay->getBitsPerDigit()) <
                coordBasedRouting
                    ->getEuclidianDistanceByKeyAndCoords(bestKey,
                                                         ((const Nps&)neighborCache->getNcsAccess()).getOwnCoordinates(), //TODO
                                                         overlay->getBitsPerDigit())) {
                bestKey = possibleKeys[i];
                bestpos = i;
            }
        }
        //std::cout << neighborCache->getOwnEuclidianDistanceToKey(bestKey) << "\n" << std::endl;
        orderedKeys.push_back(bestKey);
        possibleKeys.erase(possibleKeys.begin()+bestpos);
    }
    /*
    std::cout << "NodeID: " << overlay->getThisNode().getKey().toString(16) << std::endl;
    std::cout << "Original Key: " << originalKey.toString(16) << std::endl;
    for (int i = 0; i < orderedKeys.size(); i++) {
        std::cout << "Sorted Key " << i << ": " << orderedKeys[i].toString(16) << " (" << overlay->getOwnEuclidianDistanceToKey(orderedKeys[i]) << ")" << std::endl;
    }
    */
    OverlayKey searchKey = orderedKeys[teamnum];
#define DIRECT_ROUTE_GET
#ifndef DIRECT_ROUTE_GET
    LookupCall* replicaMsg = new LookupCall();
    replicaMsg->setKey(searchKey);
    replicaMsg->setNumSiblings(floor(numReplica / numReplicaTeams));
    int nonce = sendInternalRpcCall(OVERLAY_COMP, replicaMsg);
    rpcIdMap.insert(make_pair(nonce, capiGetMsg));
    lastGetCall = SIMTIME_DBL(simTime());
#else
    GetMapEntry mapEntry;
    mapEntry.numSent = 0;
    // Multi team version: Already mapEntry from earlier team?
    std::map<int, GetMapEntry>::iterator it2 =
        getMap.find(capiGetMsg->getNonce());
    if (it2 != getMap.end()) {
        mapEntry = it2->second;
    } else {
        mapEntry.teamNumber = 0;
    }
    mapEntry.numAvailableReplica = 1;//lookupMsg->getSiblingsArraySize();
    mapEntry.numResponses = 0;
    mapEntry.callMsg = capiGetMsg;
    mapEntry.hashVector = NULL;
    mapEntry.replica.clear();
    for (unsigned int i = 0; i < 1/*lookupMsg->getSiblingsArraySize()*/; i++) {
        // Simplified GET Request: Just one real request, rest is for fallback
        if (i == 0) {
            CBRDHTGetCall* dhtMsg = new CBRDHTGetCall();
            dhtMsg->setOriginalKey(capiGetMsg->getKey());
            dhtMsg->setKey(searchKey);//lookupMsg->getKey());
            dhtMsg->setIsHash(false);
            dhtMsg->setKind(capiGetMsg->getKind());
            dhtMsg->setId(capiGetMsg->getId());
            dhtMsg->setBitLength(GETCALL_L(dhtMsg));
            RECORD_STATS(normalMessages++;
            numBytesNormal += dhtMsg->getByteLength());
            /*int nonce = */sendRouteRpcCall(TIER1_COMP, searchKey, dhtMsg,
                                         NULL, DEFAULT_ROUTING, -1, 0,
                                         capiGetMsg->getNonce());
            //rpcIdMap.insert(make_pair(nonce, capiGetMsg));
            //sendRouteRpcCall(TIER1_COMP, lookupMsg->getSiblings(i), dhtMsg,
            //                 NULL, DEFAULT_ROUTING, -1, 0,
            //                 capiGetMsg->getNonce());
            mapEntry.numSent++;
        } else {
            //We don't send, we just store the remaining keys as fallback
            //mapEntry.replica.push_back(lookupMsg->getSiblings(i));
        }
    }
    /*
                std::cout << "New replica: " <<  std::endl;
                for (int i = 0; i < mapEntry.replica.size(); i++) {
                    std::cout << mapEntry.replica[i] << std::endl;
                }
                std::cout << "*************************" << std::endl;
     */
    if (it2 != getMap.end())
        getMap.erase(it2);
    getMap.insert(make_pair(capiGetMsg->getNonce(), mapEntry));
#endif
}

void CBRDHT::handleGetRequest ( CBRDHTGetCall * dhtMsg )

protected

Definition at line 336 of file CBR-DHT.cc.

{
    std::string tempString = "GET_REQUEST received: "
            + std::string(dhtMsg->getKey().toString(16));
    getParentModule()->getParentModule()->bubble(tempString.c_str());
    BinaryValue storedValue;
    DhtDataEntry* dataEntry = dataStorage->getDataEntry(dhtMsg->getKey(), 1, 1);
    if (dataEntry) {
        storedValue = dataStorage->getDataEntry(dhtMsg->getKey(), 1, 1)->value;
    } else {
        storedValue = BinaryValue::UNSPECIFIED_VALUE;
    }
    // send back
    CBRDHTGetResponse* responseMsg = new CBRDHTGetResponse();
    responseMsg->setKey(dhtMsg->getKey());
    responseMsg->setOriginalKey(dhtMsg->getOriginalKey());
    responseMsg->setIsHash(false);
    if (storedValue.isUnspecified()) {
        //responseMsg->setValue(BinaryValue::UNSPECIFIED_VALUE);
        DhtDumpEntry result;
        result.setKey(dhtMsg->getKey());
        result.setValue(BinaryValue::UNSPECIFIED_VALUE);
        responseMsg->setResultArraySize(1);
        responseMsg->setResult(0, result);
    } else {
        //responseMsg->setValue(storedValue);
        DhtDumpEntry result;
        result.setKey(dhtMsg->getKey());
        result.setValue(storedValue);
        responseMsg->setResultArraySize(1);
        responseMsg->setResult(0, result);
    }
    rpcIdMap.insert(make_pair(dhtMsg->getNonce(), (BaseCallMessage*)NULL));
    responseMsg->setBitLength(GETRESPONSE_L(responseMsg));
    RECORD_STATS(normalMessages++;
                 numBytesNormal += responseMsg->getByteLength());
    sendRpcResponse(dhtMsg, responseMsg);
}

void CBRDHT::handleGetResponse	(	CBRDHTGetResponse *	dhtMsg,
		int	rpcId
	)

protected

Definition at line 591 of file CBR-DHT.cc.

{
    std::map<int, GetMapEntry>::iterator it2 =
            getMap.find(rpcId);
    //unknown request
    if (it2 == getMap.end()) {
        std::cout << "- 1 -" << std::endl;
        return;
    }
    if (!dhtMsg->getIsHash()) {
        //std::cout << "[" << overlay->getThisNode().getIp() << "] " << "Received an answer! Sending up key " << dhtMsg->getKey().toString(16) << "(orig: " << dhtMsg->getOriginalKey().toString(16) << ") -- value " << dhtMsg->getHashValue() << std::endl;
        //std::cout << "Replica left: " << it2->second.replica.size() << std::endl;
        if (dhtMsg->getHashValue().size() > 0 || dhtMsg->getResultArraySize() > 0) {
            // Successful Lookup, received a value
            DHTgetCAPIResponse* capiGetRespMsg = new DHTgetCAPIResponse();
            //capiGetRespMsg->setKey(dhtMsg->getOriginalKey());
            //capiGetRespMsg->setValue(dhtMsg->getHashValue());
            DhtDumpEntry result;
            result.setKey(dhtMsg->getKey());
            result.setValue(dhtMsg->getResult(0).getValue());//getHashValue());
            capiGetRespMsg->setResultArraySize(1);
            capiGetRespMsg->setResult(0, result);
            //std::cout << "[" << overlay->getThisNode().getIp() << "] " << "SUCCESSFUL LOOKUP! Sending up key " << dhtMsg->getKey().toString(16) << "(orig: " << dhtMsg->getOriginalKey().toString(16) << ") -- value " << dhtMsg->getHashValue() << std::endl;
            capiGetRespMsg->setIsSuccess(true);
            sendRpcResponse(it2->second.callMsg, capiGetRespMsg);
            getMap.erase(rpcId);
            return;
        } else if (it2->second.replica.size() > 0) {
            // Received empty value, try fallback replica
            NodeHandle fallbackReplica = it2->second.replica.back();
            std::cout << "[" << overlay->getThisNode().getIp() << "] " << "Empty value received. Asking replica now ("<< it2->second.replica.size()<<" left)!" << std::endl;
            CBRDHTGetCall* dhtRecall = new CBRDHTGetCall();
            dhtRecall->setOriginalKey(dhtMsg->getOriginalKey());
            dhtRecall->setKey(dhtMsg->getKey());
            dhtRecall->setIsHash(false);
            dhtRecall->setBitLength(GETCALL_L(dhtRecall));
            RECORD_STATS(normalMessages++;
            numBytesNormal += dhtRecall->getByteLength());
            sendRouteRpcCall(TIER1_COMP, fallbackReplica, dhtRecall,
                             NULL, DEFAULT_ROUTING, -1, 0,
                             it2->second.callMsg->getNonce());
            it2->second.numSent++;
            it2->second.replica.pop_back();
            return;
        } else if (it2->second.teamNumber < (numReplicaTeams - 1)) {
            // No more fallback replica in this team, try next one
            std::cout << "it2->second.teamNumber (" << it2->second.teamNumber << ") < (numReplicaTeams - 1) (" << (numReplicaTeams - 1) << ")" << std::endl;
            std::cout << "[" << overlay->getThisNode().getIp() << "] " << "No more fallback replica in this team "<< it2->second.teamNumber<<". Trying next one ("<< it2->second.teamNumber+1 <<  ")..." << std::endl;
            it2->second.teamNumber++;
            handleGetCAPIRequest(it2->second.callMsg, it2->second.teamNumber);
            return;
        } else {
            // No more replica, no more teams, send success == false to Tier 2 :(
            std::cout << "[" << overlay->getThisNode().getIp() << "] " << "No more fallback replica. Lookup failed. :(" << std::endl;
            DHTgetCAPIResponse* capiGetRespMsg = new DHTgetCAPIResponse();
            //capiGetRespMsg->setKey(dhtMsg->getOriginalKey());
            //capiGetRespMsg->setValue(BinaryValue::UNSPECIFIED_VALUE);
            DhtDumpEntry result;
            result.setKey(dhtMsg->getKey());
            result.setValue(BinaryValue::UNSPECIFIED_VALUE);
            capiGetRespMsg->setResultArraySize(1);
            capiGetRespMsg->setResult(0, result);
            capiGetRespMsg->setIsSuccess(false);
            sendRpcResponse(it2->second.callMsg, capiGetRespMsg);
            getMap.erase(rpcId);
        }
    }
}

void CBRDHT::handleLookupResponse ( LookupResponse * lookupMsg )

protected

Definition at line 740 of file CBR-DHT.cc.

{
        std::map<unsigned int, BaseCallMessage*>::iterator it =
            rpcIdMap.find(lookupMsg->getNonce());
    if (it == rpcIdMap.end() || it->second == NULL)
        return;
    if (dynamic_cast<DHTputCAPICall*> (it->second)) {
        #if 0
        cout << "DHT::handleLookupResponse(): PUT "
             << lookupMsg->getKey() << " ("
             << overlay->getThisNode().getKey() << ")" << endl;
        for (unsigned int i = 0; i < lookupMsg->getSiblingsArraySize(); i++) {
            cout << i << ": " << lookupMsg->getSiblings(i) << endl;
        }
#endif
        DHTputCAPICall* capiPutMsg = dynamic_cast<DHTputCAPICall*> (it->second);
        rpcIdMap.erase(lookupMsg->getNonce());
        if ((lookupMsg->getIsValid() == false)
                || (lookupMsg->getSiblingsArraySize() == 0)) {
            EV << "[DHT::handleLookupResponse()]\n"
               << "    Unable to get replica list : invalid lookup"
               << endl;
            DHTputCAPIResponse* capiPutRespMsg = new DHTputCAPIResponse();
            capiPutRespMsg->setIsSuccess(false);
            sendRpcResponse(capiPutMsg, capiPutRespMsg);
            return;
        }
        for (unsigned int i = 0; i < lookupMsg->getSiblingsArraySize(); i++) {
            DHTPutCall* dhtMsg = new DHTPutCall();
            dhtMsg->setKey(capiPutMsg->getKey());
            dhtMsg->setValue(capiPutMsg->getValue());
            dhtMsg->setKind(capiPutMsg->getKind());
            dhtMsg->setId(capiPutMsg->getId());
            dhtMsg->setTtl(capiPutMsg->getTtl());
            dhtMsg->setIsModifiable(capiPutMsg->getIsModifiable());
            dhtMsg->setMaintenance(false);
            dhtMsg->setBitLength(PUTCALL_L(dhtMsg));
            RECORD_STATS(normalMessages++;
                         numBytesNormal += dhtMsg->getByteLength());
            sendRouteRpcCall(TIER1_COMP, lookupMsg->getSiblings(i),
                             dhtMsg, NULL, DEFAULT_ROUTING, -1,
                             0, capiPutMsg->getNonce());
        }
        PutMapEntry mapEntry;
        mapEntry.callMsg = capiPutMsg;
        mapEntry.numResponses = 0;
        mapEntry.numFailed = 0;
        mapEntry.numSent = lookupMsg->getSiblingsArraySize();
        putMap.insert(make_pair(capiPutMsg->getNonce(), mapEntry));
    }
    else if (dynamic_cast<DHTgetCAPICall*>(it->second)) {
#if 0
        cout << "DHT::handleLookupResponse(): GET "
             << lookupMsg->getKey() << " ("
             << overlay->getThisNode().getKey() << ")" << endl;
        for (unsigned int i = 0; i < lookupMsg->getSiblingsArraySize(); i++) {
            cout << i << ": " << lookupMsg->getSiblings(i) << endl;
        }
#endif
        DHTgetCAPICall* capiGetMsg = dynamic_cast<DHTgetCAPICall*>(it->second);
        rpcIdMap.erase(lookupMsg->getNonce());
        // Invalid lookup
        if ((lookupMsg->getIsValid() == false)
                || (lookupMsg->getSiblingsArraySize() == 0)) {
            EV << "[DHT::handleLookupResponse()]\n"
               << "    Unable to get replica list : invalid lookup"
               << endl;
            DHTgetCAPIResponse* capiGetRespMsg = new DHTgetCAPIResponse();
            //capiGetRespMsg->setKey(lookupMsg->getKey());
            //capiGetRespMsg->setValue(BinaryValue::UNSPECIFIED_VALUE);
            DhtDumpEntry result;
            result.setKey(lookupMsg->getKey());
            result.setValue(BinaryValue::UNSPECIFIED_VALUE);
            capiGetRespMsg->setResultArraySize(1);
            capiGetRespMsg->setResult(0, result);
            capiGetRespMsg->setIsSuccess(false);
            sendRpcResponse(capiGetMsg, capiGetRespMsg);
            return;
        }
        // Valid lookup
        GetMapEntry mapEntry;
        mapEntry.numSent = 0;
        // Multi team version: Already mapEntry from earlier team?
        std::map<int, GetMapEntry>::iterator it2 =
            getMap.find(capiGetMsg->getNonce());
        if (it2 != getMap.end()) {
            mapEntry = it2->second;
        } else {
            mapEntry.teamNumber = 0;
        }
        mapEntry.numAvailableReplica = lookupMsg->getSiblingsArraySize();
        mapEntry.numResponses = 0;
        mapEntry.callMsg = capiGetMsg;
        mapEntry.hashVector = NULL;
        mapEntry.replica.clear();
        for (unsigned int i = 0; i < lookupMsg->getSiblingsArraySize(); i++) {
            // Simplified GET Request: Just one real request, rest is for fallback
            if (i == 0) {
                CBRDHTGetCall* dhtMsg = new CBRDHTGetCall();
                dhtMsg->setOriginalKey(capiGetMsg->getKey());
                dhtMsg->setKey(lookupMsg->getKey());
                dhtMsg->setIsHash(false);
                dhtMsg->setKind(capiGetMsg->getKind());
                dhtMsg->setId(capiGetMsg->getId());
                dhtMsg->setBitLength(GETCALL_L(dhtMsg));
                RECORD_STATS(normalMessages++;
                numBytesNormal += dhtMsg->getByteLength());
                sendRouteRpcCall(TIER1_COMP, lookupMsg->getSiblings(i), dhtMsg,
                                 NULL, DEFAULT_ROUTING, -1, 0,
                                 capiGetMsg->getNonce());
                mapEntry.numSent++;
            } else {
                //We don't send, we just store the remaining keys as fallback
                mapEntry.replica.push_back(lookupMsg->getSiblings(i));
            }
        }
        /*
            std::cout << "New replica: " <<  std::endl;
            for (int i = 0; i < mapEntry.replica.size(); i++) {
                std::cout << mapEntry.replica[i] << std::endl;
            }
            std::cout << "*************************" << std::endl;
         */
        if (it2 != getMap.end())
            getMap.erase(it2);
        getMap.insert(make_pair(capiGetMsg->getNonce(), mapEntry));
    } else if (dynamic_cast<DHTPutCall*>(it->second)) {
        DHTPutCall* putMsg = dynamic_cast<DHTPutCall*>(it->second);
        rpcIdMap.erase(lookupMsg->getNonce());
        if ((lookupMsg->getIsValid() == false)
            || (lookupMsg->getSiblingsArraySize() == 0)) {
            EV << "[DHT::handleLookupResponse()]\n"
               << "    Unable to get replica list : invalid lookup"
               << endl;
            delete putMsg;
            return;
        }
        for( unsigned int i = 0; i < lookupMsg->getSiblingsArraySize(); i++ ) {
            RECORD_STATS(maintenanceMessages++;
                         numBytesMaintenance += putMsg->getByteLength());
            sendRouteRpcCall(TIER1_COMP, lookupMsg->getSiblings(i),
                             new DHTPutCall(*putMsg));
        }
        delete putMsg;
    }
}

void CBRDHT::handlePutCAPIRequest ( DHTputCAPICall * capiPutMsg )

protected

Definition at line 380 of file CBR-DHT.cc.

{
    // provide copies of this message for other teams
    for (int i = 1; i < numReplicaTeams; i++) {
        DHTPutCall* teamCopyPutMsg = new DHTPutCall; //TODO memleak
        // transfer attributes of original DHTputCAPICall to DHTPutCall for teams
        teamCopyPutMsg->setValue(capiPutMsg->getValue());
        teamCopyPutMsg->setTtl(capiPutMsg->getTtl());
        teamCopyPutMsg->setIsModifiable(capiPutMsg->getIsModifiable());
        teamCopyPutMsg->setKind(capiPutMsg->getKind());
        teamCopyPutMsg->setId(capiPutMsg->getId());
        // control info needs to be copied by value
        OverlayCtrlInfo controlInfo = *(check_and_cast<OverlayCtrlInfo*>(capiPutMsg->getControlInfo()));
        OverlayCtrlInfo* controlInfoCopy = new OverlayCtrlInfo;
        *controlInfoCopy = controlInfo;
        teamCopyPutMsg->setControlInfo(controlInfoCopy);
        // multiple SHA1 hashing of original key
        OverlayKey destKey = capiPutMsg->getKey();
        for (int j = 0; j < i; j++) {
            destKey = OverlayKey::sha1(BinaryValue(destKey.toString(16).c_str()));
        }
        teamCopyPutMsg->setKey(destKey);
        // rest is analog to handlePutCAPIRequest, but for DHTPutCall
        LookupCall* replicaMsg = new LookupCall();
        replicaMsg->setKey(teamCopyPutMsg->getKey());
        replicaMsg->setNumSiblings(floor(numReplica / numReplicaTeams));
        int nonce = sendInternalRpcCall(OVERLAY_COMP, replicaMsg);
        rpcIdMap.insert(make_pair(nonce, teamCopyPutMsg));
    }
    //asks the replica list
    LookupCall* replicaMsg = new LookupCall();
    replicaMsg->setKey(capiPutMsg->getKey());
    replicaMsg->setNumSiblings(floor(numReplica / numReplicaTeams));
    int nonce = sendInternalRpcCall(OVERLAY_COMP, replicaMsg);
    rpcIdMap.insert(make_pair(nonce, capiPutMsg));
}

void CBRDHT::handlePutRequest ( DHTPutCall * dhtMsg )

protected

Definition at line 279 of file CBR-DHT.cc.

{
    std::string tempString = "PUT_REQUEST received: "
            + std::string(dhtMsg->getKey().toString(16));
    getParentModule()->getParentModule()->bubble(tempString.c_str());
    if (!(dataStorage->isModifiable(dhtMsg->getKey(), dhtMsg->getKind(),
                                    dhtMsg->getId()))) {
        //check if the put request came from the right node
        NodeHandle sourceNode = dataStorage->getSourceNode(dhtMsg->getKey(),
                                                           dhtMsg->getKind(), dhtMsg->getId());
        if (((!sourceNode.isUnspecified())
                && (!dhtMsg->getSrcNode().isUnspecified()) && (sourceNode
                != dhtMsg->getSrcNode())) || ((dhtMsg->getMaintenance())
                && (dhtMsg->getOwnerNode() == sourceNode))) {
            // TODO: set owner
            DHTPutResponse* responseMsg = new DHTPutResponse();
            responseMsg->setSuccess(false);
            responseMsg->setBitLength(PUTRESPONSE_L(responseMsg));
            RECORD_STATS(normalMessages++;
                         numBytesNormal += responseMsg->getByteLength());
            sendRpcResponse(dhtMsg, responseMsg);
            return;
        }
    }
    // remove data item from local data storage
    //cancelAndDelete(dataStorage->getTtlMessage(dhtMsg->getKey()));
    //dataStorage->removeData(dhtMsg->getKey());
    dataStorage->removeData(dhtMsg->getKey(), dhtMsg->getKind(),
                                dhtMsg->getId());
    if (dhtMsg->getValue().size() > 0) {
        // add ttl timer
        DHTTtlTimer *timerMsg = new DHTTtlTimer("ttl_timer");
        timerMsg->setKey(dhtMsg->getKey());
        scheduleAt(simTime() + dhtMsg->getTtl(), timerMsg);
        // storage data item in local data storage
        bool err;
        dataStorage->addData(dhtMsg->getKey(), dhtMsg->getKind(),
                             dhtMsg->getId(), dhtMsg->getValue(), timerMsg,
                             dhtMsg->getIsModifiable(), dhtMsg->getSrcNode(),
                             overlay->isSiblingFor(overlay->getThisNode(),
                                                   dhtMsg->getKey(),
                                                   1, &err));
    }
    // send back
    DHTPutResponse* responseMsg = new DHTPutResponse();
    responseMsg->setSuccess(true);
    responseMsg->setBitLength(PUTRESPONSE_L(responseMsg));
    RECORD_STATS(normalMessages++; numBytesNormal += responseMsg->getByteLength());
    sendRpcResponse(dhtMsg, responseMsg);
}

void CBRDHT::handlePutResponse	(	DHTPutResponse *	dhtMsg,
		int	rpcId
	)

protected

Definition at line 568 of file CBR-DHT.cc.

{
    std::map<int, PutMapEntry>::iterator it2 =
            putMap.find(rpcId);
    if (it2 == putMap.end()) //unknown request
        return;
    if (dhtMsg->getSuccess()) {
        it2->second.numResponses++;
    } else {
        it2->second.numFailed++;
    }
    if (it2->second.numResponses / (double)it2->second.numSent > 0.5) {
        DHTputCAPIResponse* capiPutRespMsg = new DHTputCAPIResponse();
        capiPutRespMsg->setIsSuccess(true);
        sendRpcResponse(it2->second.callMsg, capiPutRespMsg);
        it2->second.callMsg = NULL;
        putMap.erase(rpcId);
    }
}

bool CBRDHT::handleRpcCall ( BaseCallMessage * msg )

protectedvirtual

Processes Remote-Procedure-Call invocation messages.

This method should be overloaded when the overlay provides RPC functionality.

Returns: true, if rpc has been handled

Reimplemented from BaseRpc.

Definition at line 148 of file CBR-DHT.cc.

{
    // delegate messages
    RPC_SWITCH_START( msg )
        // RPC_DELEGATE( <messageName>[Call|Response], <methodToCall> )
        RPC_DELEGATE( DHTPut, handlePutRequest );
        RPC_DELEGATE( CBRDHTGet, handleGetRequest );
        RPC_DELEGATE( DHTputCAPI, handlePutCAPIRequest ); //requests coming from an upper tier
        RPC_DELEGATE( DHTgetCAPI, handleGetCAPIRequest );
        RPC_DELEGATE( DHTdump, handleDumpDhtRequest );
    RPC_SWITCH_END( )
    return RPC_HANDLED;
}

void CBRDHT::handleRpcResponse	(	BaseResponseMessage *	msg,
		cPolymorphic *	context,
		int	rpcId,
		simtime_t	rtt
	)

protectedvirtual

This method is called if an RPC response has been received.

Parameters

msg	The response message.
context	Pointer to an optional state object. The object has to be handled/deleted by the handleRpcResponse() code
rpcId	The RPC id.
rtt	The Round-Trip-Time of this RPC

Reimplemented from RpcListener.

Definition at line 163 of file CBR-DHT.cc.

{
    RPC_SWITCH_START(msg)
        RPC_ON_RESPONSE(DHTPut){
        handlePutResponse(_DHTPutResponse, rpcId);
        EV << "[DHT::handleRpcResponse()]\n"
           << "    DHT Put RPC Response received: id=" << rpcId
           << " msg=" << *_DHTPutResponse << " rtt=" << rtt
           << endl;
        break;
    }
    RPC_ON_RESPONSE(CBRDHTGet) {
        handleGetResponse(_CBRDHTGetResponse, rpcId);
        EV << "[DHT::handleRpcResponse()]\n"
           << "    DHT Get RPC Response received: id=" << rpcId
           << " msg=" << *_CBRDHTGetResponse << " rtt=" << rtt
           << endl;
        break;
    }
    RPC_ON_RESPONSE(Lookup) {
        handleLookupResponse(_LookupResponse);
        EV << "[DHT::handleRpcResponse()]\n"
           << "    Replica Set RPC Response received: id=" << rpcId
           << " msg=" << *_LookupResponse << " rtt=" << rtt
           << endl;
        break;
    }
    RPC_SWITCH_END()
}

void CBRDHT::handleRpcTimeout	(	BaseCallMessage *	msg,
		const TransportAddress &	dest,
		cPolymorphic *	context,
		int	rpcId,
		const OverlayKey &	destKey
	)

protectedvirtual

This method is called if an RPC timeout has been reached.

Parameters

msg	The original RPC message.
dest	The destination node
context	Pointer to an optional state object. The object has to be handled/deleted by the handleRpcResponse() code
rpcId	The RPC id.
destKey	the destination OverlayKey

Reimplemented from RpcListener.

Definition at line 194 of file CBR-DHT.cc.

{
    RPC_SWITCH_START(msg)
    RPC_ON_CALL(DHTPut){
        EV << "[DHT::handleRpcResponse()]\n"
           << "    DHTPut Timeout"
           << endl;
        std::map<int, PutMapEntry>::iterator it2 =
                putMap.find(rpcId);
        if (it2 == putMap.end()) //unknown request
            return;
        it2->second.numFailed++;
        if (it2->second.numFailed / (double)it2->second.numSent >= 0.5) {
            DHTputCAPIResponse* capiPutRespMsg = new DHTputCAPIResponse();
            capiPutRespMsg->setIsSuccess(false);
            sendRpcResponse(it2->second.callMsg, capiPutRespMsg);
            it2->second.callMsg = NULL;
            putMap.erase(rpcId);
        }
        break;
    }
    RPC_ON_CALL(CBRDHTGet) {
        EV << "[DHT::handleRpcResponse()]\n"
           << "    DHTGet Timeout"
           << endl;
        std::map<int, GetMapEntry>::iterator it2 =
            getMap.find(rpcId);
        if (it2 == getMap.end()) //unknown request
            return;
        if (it2->second.replica.size() > 0) {
            // Received empty value, try fallback replica
            NodeHandle fallbackReplica = it2->second.replica.back();
            CBRDHTGetCall* dhtRecall = new CBRDHTGetCall();
            dhtRecall->setOriginalKey(_CBRDHTGetCall->getOriginalKey());
            dhtRecall->setKey(_CBRDHTGetCall->getKey());
            dhtRecall->setIsHash(false);
            dhtRecall->setBitLength(GETCALL_L(dhtRecall));
            RECORD_STATS(normalMessages++;
            numBytesNormal += dhtRecall->getByteLength());
            sendRouteRpcCall(TIER1_COMP, fallbackReplica, dhtRecall,
                             NULL, DEFAULT_ROUTING, -1, 0,
                             it2->second.callMsg->getNonce());
            it2->second.numSent++;
            it2->second.replica.pop_back();
            return;
        } else if (it2->second.teamNumber < (numReplicaTeams - 1)) {
            // No more fallback replica in this team, try next one
            it2->second.teamNumber++;
            handleGetCAPIRequest(it2->second.callMsg, it2->second.teamNumber);
            return;
        } else {
            // No more replica, no more teams, send success == false to Tier 2 :(
            DHTgetCAPIResponse* capiGetRespMsg = new DHTgetCAPIResponse();
            //capiGetRespMsg->setKey(_CBRDHTGetCall->getOriginalKey());
            //capiGetRespMsg->setValue(BinaryValue::UNSPECIFIED_VALUE);
            DhtDumpEntry result;
            result.setKey(_CBRDHTGetCall->getKey());
            result.setValue(BinaryValue::UNSPECIFIED_VALUE);
            capiGetRespMsg->setResultArraySize(1);
            capiGetRespMsg->setResult(0, result);
            capiGetRespMsg->setIsSuccess(false);
            sendRpcResponse(it2->second.callMsg, capiGetRespMsg);
            getMap.erase(rpcId);
        }
        break;
    }
    RPC_SWITCH_END( )
}

void CBRDHT::handleTimerEvent ( cMessage * msg )

protectedvirtual

Reimplemented from BaseRpc.

Definition at line 114 of file CBR-DHT.cc.

{
    DHTTtlTimer* msg_timer = dynamic_cast<DHTTtlTimer*> (msg);
    if (msg_timer) {
        EV << "[DHT::handleTimerEvent()]\n"
           << "    received timer ttl, key: "
           << msg_timer->getKey().toString(16)
           << "\n (overlay->getThisNode().key = "
           << overlay->getThisNode().getKey().toString(16) << ")"
           << endl;
        dataStorage->removeData(msg_timer->getKey(), msg_timer->getKind(),
                                msg_timer->getId());
        //delete msg_timer;
    }
    /*DHTTtlTimer* msg_timer;
    if (msg->isName("ttl_timer")) {
        msg_timer = check_and_cast<DHTTtlTimer*> (msg);
        EV << "[DHT::handleTimerEvent()]\n"
           << "    received timer ttl, key: "
           << msg_timer->getKey().toString(16)
           << "\n (overlay->getThisNode().key = "
           << overlay->getThisNode().getKey().toString(16) << ")"
           << endl;
        dataStorage->removeData(msg_timer->getKey(), msg_timer->getKind(),
                                msg_timer->getId());
        delete msg_timer;
    }*/
}

void CBRDHT::handleUpperMessage ( cMessage * msg )

protectedvirtual

handleUpperMessage gets called of handleMessage(cMessage* msg) if msg arrivedOn from_upperTier (currently msg gets deleted in this function)

Parameters

msg	the message to handle

Reimplemented from BaseApp.

Definition at line 272 of file CBR-DHT.cc.

{
    error("DHT::handleUpperMessage(): Received message with unknown type!");
    delete msg;
}

void CBRDHT::initializeApp ( int stage )

protectedvirtual

initializes derived class-attributes

Parameters

stage the init stage

Reimplemented from BaseApp.

Definition at line 78 of file CBR-DHT.cc.

{
        if (stage != MIN_STAGE_APP)
        return;
    dataStorage = check_and_cast<DHTDataStorage*>
                      (getParentModule()->getSubmodule("dhtDataStorage"));
    coordBasedRouting = CoordBasedRoutingAccess().get();
    neighborCache = (NeighborCache*)getParentModule()
        ->getParentModule()->getSubmodule("neighborCache");
    numReplica = par("numReplica");
    numReplicaTeams = par("numReplicaTeams");
    if (numReplica > numReplicaTeams * overlay->getMaxNumSiblings()) {
        throw cRuntimeError("DHT::initialize(): numReplica bigger than what this "
                "overlay can handle (%d)", numReplicaTeams*overlay->getMaxNumSiblings());
    }
    if (numReplica < numReplicaTeams) {
        throw cRuntimeError("DHT::initialize(): numReplica (%d) smaller than numReplicaTeam (%d)",
                            numReplica, numReplicaTeams);
    }
    maintenanceMessages = 0;
    normalMessages = 0;
    numBytesMaintenance = 0;
    numBytesNormal = 0;
    WATCH(maintenanceMessages);
    WATCH(normalMessages);
    WATCH(numBytesNormal);
    WATCH(numBytesMaintenance);
    WATCH_MAP(rpcIdMap);
}

int CBRDHT::resultValuesBitLength ( DHTGetResponse * msg )

protected

Definition at line 931 of file CBR-DHT.cc.

                                                     {
    int bitSize = 0;
    for (uint i = 0; i < msg->getResultArraySize(); i++) {
        bitSize += msg->getResult(i).getValue().size();
    }
    return bitSize;
}

void CBRDHT::update	(	const NodeHandle &	node,
		bool	joined
	)

protectedvirtual

Common API function: informs application about neighbors and own nodeID.

Parameters

node	new or lost neighbor
joined	new or lost?

Reimplemented from BaseApp.

Definition at line 671 of file CBR-DHT.cc.

{
    OverlayKey key;
    DHTPutCall* dhtMsg;
    bool err = false;
    //DHTData entry;
    DhtDataEntry entry;
    //std::map<OverlayKey, DHTData>::iterator it = dataStorage->begin();
    DhtDataMap::iterator it = dataStorage->begin();
    for (unsigned int i = 0; i < dataStorage->getSize(); i++) {
        key = it->first;
        entry = it->second;
        if (joined) {
            if (entry.responsible && (overlay->isSiblingFor(node, key,
                                                            numReplica, &err)
                    || err)) { // hack for Chord, if we've got a new predecessor
                dhtMsg = new DHTPutCall();
                dhtMsg->setKey(key);
                dhtMsg->setValue(entry.value);
                dhtMsg->setKind(entry.kind);
                dhtMsg->setId(entry.id);
                //dhtMsg->setTtl((int) (entry.ttlMessage->arrivalTime()
                //        - simTime()));
                dhtMsg->setTtl((int)SIMTIME_DBL(entry.ttlMessage->getArrivalTime()
                                                - simTime()));
                dhtMsg->setIsModifiable(entry.is_modifiable);
                dhtMsg->setMaintenance(true);
                dhtMsg->setBitLength(PUTCALL_L(dhtMsg));
                RECORD_STATS(maintenanceMessages++;
                        numBytesMaintenance += dhtMsg->getByteLength());
                sendRouteRpcCall(TIER1_COMP, node, dhtMsg);
            }
            if (err) {
                EV << "[DHT::update()]\n"
                   << "    Unable to know if key: " << key
                   << " is in range of node: " << node
                   << endl;
            }
        } else {
#if 0
            //the update concerns a node who has left
            //replicate
            LookupCall* replicaMsg = new LookupCall();
            replicaMsg->setKey(key);
            replicaMsg->setNumSiblings(numReplica);
            int nonce = sendInternalRpcCall(OVERLAY_COMP,
                                            replicaMsg);
            dhtMsg = new DHTPutCall();
            dhtMsg->setKey(key);
            dhtMsg->setValue(entry.value);
            dhtMsg->setTtl((int)(entry.ttlMessage->arrivalTime()
                    - simulation.simTime()));
            dhtMsg->setIsModifiable(entry.is_modifiable);
            dhtMsg->setMaintenance(true);
            dhtMsg->setLength(PUTCALL_L(dhtMsg));
            rpcIdMap.insert(make_pair(nonce, dhtMsg));
#endif
        }
        entry.responsible = overlay->isSiblingFor(overlay->getThisNode(),
                                                  key, 1, &err);
        it++;
    }
}

Member Data Documentation

CoordBasedRouting* CBRDHT::coordBasedRouting

protected

Definition at line 116 of file CBR-DHT.h.

DHTDataStorage* CBRDHT::dataStorage

protected

pointer to the dht data storage

Definition at line 115 of file CBR-DHT.h.

std::map<int, GetMapEntry> CBRDHT::getMap

protected

Definition at line 111 of file CBR-DHT.h.

double CBRDHT::lastGetCall

protected

Definition at line 109 of file CBR-DHT.h.

double CBRDHT::maintenanceMessages

protected

Definition at line 105 of file CBR-DHT.h.

NeighborCache* CBRDHT::neighborCache

protected

Definition at line 117 of file CBR-DHT.h.

double CBRDHT::normalMessages

protected

Definition at line 106 of file CBR-DHT.h.

double CBRDHT::numBytesMaintenance

protected

Definition at line 107 of file CBR-DHT.h.

double CBRDHT::numBytesNormal

protected

Definition at line 108 of file CBR-DHT.h.

int CBRDHT::numReplica

protected

Definition at line 102 of file CBR-DHT.h.

uint8_t CBRDHT::numReplicaTeams

protected

Definition at line 103 of file CBR-DHT.h.

std::map<int, PutMapEntry> CBRDHT::putMap

protected

Definition at line 112 of file CBR-DHT.h.

std::map<unsigned int, BaseCallMessage*> CBRDHT::rpcIdMap

protected

List of the Rpc Ids of the messages sent following the reception of an rpc request (the second member)

Definition at line 110 of file CBR-DHT.h.

The documentation for this class was generated from the following files:

Classes

Public Member Functions

Protected Types

Protected Member Functions

Protected Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation