//
// Copyright (C) 2007 Institut fuer Telematik, Universitaet Karlsruhe (TH)
//
// This program is free software; you can redistribute it and/or
// modify it under the terms of the GNU General Public License
// as published by the Free Software Foundation; either version 2
// of the License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
//

/**
 * @file MyDHT.cc
 * @author Gregoire Menuel, Ingmar Baumgart, Dilum Bandara
 * @version 0.1.1
 * @date 07/31/2010
 */

#include <IPAddressResolver.h>
#include "MyDHT.h"
#include <RpcMacros.h>
#include <BaseRpc.h>
#include <GlobalStatistics.h>

Define_Module(MyDHT);

using namespace std;


MyDHT::MyDHT()
{
    dataStorage = NULL;
    lookupStorage = NULL;
}


MyDHT::~MyDHT()
{
    PendingRpcs::iterator it;
    for (it = pendingRpcs.begin(); it != pendingRpcs.end(); it++) 
        delete(it->second.putCallMsg); //Cancel remaining PUT
    pendingRpcs.clear();

    if (dataStorage != NULL)
        dataStorage->clear();
    if (lookupStorage != NULL)
        lookupStorage->clear();

	cancelAndDelete(cacheTimer);
}


void MyDHT::initializeApp(int stage)
{
    if (stage != MIN_STAGE_APP)
        return;

    dataStorage = check_and_cast<MyDHTDataStorage*>
                      (getParentModule()->getSubmodule("myDhtDataStorage"));
    lookupStorage = check_and_cast<MyDHTLookupStorage*>
                      (getParentModule()->getSubmodule("myDhtLookupStorage"));

    numReplica = par("numReplica");
    numGetRequests = par("numGetRequests");
    ratioIdentical = par("ratioIdentical");
    cacheRefreshTime = par("cacheRefreshTime");
	cacheOthersKeys = par("cacheOthersKeys");
	useMsgGroupId = par("useMsgGroupId");
	maxCacheSize = par("maxCacheSize");
    cachingThreshold = par("cachingThreshold");
    alpha = par("alpha");
	
	//Make sure no of replicas are lower than what Overlay will handle as siblings
    if ((int)numReplica > overlay->getMaxNumSiblings()) 
	{
        opp_error("MyDHT::initialize(): numReplica bigger than what this "
                  "overlay can handle (%d)", overlay->getMaxNumSiblings());
    }

    normalMessages = 0;
    numBytesNormal = 0;
    numQueriesAnswered = 0;
    numQueriesForwarded = 0;
    numCacheRequests = 0;
    numCacheResponses = 0;
	cacheSize = 0;

    WATCH(normalMessages);
    WATCH(numBytesNormal);
    WATCH_MAP(pendingRpcs);
	
	bindToPort(par("appPort"));	//Bind to application layer port

    // initiate cache reevaluate timer message
    cacheTimer = new cMessage("cacheTimer");
    if (cacheRefreshTime > 0)
        scheduleAt(simTime() + cacheRefreshTime, cacheTimer);
}


void MyDHT::handleTimerEvent(cMessage* msg)
{
	if(msg == cacheTimer)
	{
        scheduleAt(simTime() + cacheRefreshTime, msg);	//Reschedule
		reevaluateCache();
	}
	else
	{
		//When DHT entry timer expires
	    MyDHTTtlTimer* msg_timer = dynamic_cast<MyDHTTtlTimer*> (msg);
    	if (msg_timer) 
		{
        	EV << "[MyDHT::handleTimerEvent()]\n"
	           << "    received timer ttl, key: "
    	       << msg_timer->getKey().toString(16)
        	   << "\n (overlay->getThisNode().getKey() = "
	           << overlay->getThisNode().getKey().toString(16) << ")" << endl;

			//Remove expired entry from dataStorage
			//TODO this need to be modified to remove only the instance of key that expired
        	dataStorage->removeData(msg_timer->getKey(), msg_timer->getKind(),
            	                    msg_timer->getId());
	    }
	}
}


bool MyDHT::handleRpcCall(BaseCallMessage* msg)
{
    RPC_SWITCH_START(msg)
		// RPC_DELEGATE( <messageName>[Call|Response], <methodToCall> )
        RPC_DELEGATE(MyDHTPut, handlePutRequest);		// RPCs between nodes
    	RPC_DELEGATE(DHTputCAPI, handlePutCAPIRequest);	// Internal RPCs
        RPC_DELEGATE(DHTgetCAPI, handleGetCAPIRequest);	// Internal RPCs
        RPC_DELEGATE(DHTdump, handleDumpDhtRequest);
    RPC_SWITCH_END( )

    return RPC_HANDLED;
}


void MyDHT::handleRpcResponse(BaseResponseMessage* msg, cPolymorphic* context,
                            int rpcId, simtime_t rtt)
{
    RPC_SWITCH_START(msg)
        RPC_ON_RESPONSE(MyDHTPut)	//Response from another node for PUT
		{
	        handlePutResponse(_MyDHTPutResponse, rpcId);
    	    EV << "[MyDHT::handleRpcResponse()]\n"
        	   << "    DHT Put RPC Response received: id=" << rpcId
	           << " msg=" << *_MyDHTPutResponse << " rtt=" << rtt << endl;
	        break;
    	}
    RPC_SWITCH_END()
}


void MyDHT::handleRpcTimeout(BaseCallMessage* msg, const TransportAddress& dest,
                           cPolymorphic* context, int rpcId,
                           const OverlayKey& destKey)
{
    RPC_SWITCH_START(msg)
	    RPC_ON_CALL(MyDHTPut)
		{
	        EV << "[MyDHT::handleRpcResponse()]\n" << " DHTPut Timeout" << endl;

	        PendingRpcs::iterator it = pendingRpcs.find(rpcId);
    	    if (it == pendingRpcs.end()) // unknown request
        	    return;

	        it->second.numFailed++;
			//If more than 50% failure, inform failure
    	    if (it->second.numFailed / (double)it->second.numSent >= 0.5) 
			{
        	    DHTputCAPIResponse* capiPutRespMsg = new DHTputCAPIResponse();
	            capiPutRespMsg->setIsSuccess(false);
        	    sendRpcResponse(it->second.putCallMsg, capiPutRespMsg);
            	pendingRpcs.erase(rpcId);
        	}
        	break;
    	}
	    RPC_SWITCH_END( )
}


void MyDHT::handlePutRequest(MyDHTPutCall* dhtMsg)
{
	//Am I suppose to keep this key?
    bool err;
    bool isSibling = overlay->isSiblingFor(overlay->getThisNode(),
                  dhtMsg->getKey(), 1, &err);
    if (err)
        isSibling = true;

    // remove data item from local data storage & add again
	// TODO Comment following if multiple copies for same key to be added
	//dataStorage->removeData(dhtMsg->getKey(), dhtMsg->getKind(), dhtMsg->getId());

    if (dhtMsg->getValue().size() > 0) 
	{
		if(dhtMsg->getTtl() > 0)        // add ttl timer
		{
	        MyDHTTtlTimer *timerMsg = new MyDHTTtlTimer("ttl_timer");
    	    timerMsg->setKey(dhtMsg->getKey());
        	timerMsg->setKind(dhtMsg->getKind());
	        timerMsg->setId(dhtMsg->getId());
    	    scheduleAt(simTime() + dhtMsg->getTtl(), timerMsg);
        	// storage data item in local data storage
	        dataStorage->addData(dhtMsg->getKey(), dhtMsg->getKind(), dhtMsg->getId(), 
								 dhtMsg->getValue(), timerMsg, dhtMsg->getIsModifiable(), 
								 dhtMsg->getSrcNode(), isSibling);
		}
		else
		{	// storage data item in local data storage
        	dataStorage->addData(dhtMsg->getKey(), dhtMsg->getKind(), dhtMsg->getId(), 
								 dhtMsg->getValue(), NULL, dhtMsg->getIsModifiable(), 
								 dhtMsg->getSrcNode(), isSibling);
		}
    }

    // send Response to caller
    MyDHTPutResponse* responseMsg = new MyDHTPutResponse();
    responseMsg->setSuccess(true);
    responseMsg->setBitLength(PUTRESPONSE_L(responseMsg));
    RECORD_STATS(normalMessages++; numBytesNormal += responseMsg->getByteLength());
    sendRpcResponse(dhtMsg, responseMsg);
}


void MyDHT::handlePutCAPIRequest(DHTputCAPICall* capiPutMsg)
{
    if ((capiPutMsg->getId() == 0) && //If no ID
                (capiPutMsg->getValue().size() > 0)) 
	{
        // pick a random id before replication of the data item
        // id 0 is kept for delete requests (i.e. a put with empty value)
        capiPutMsg->setId(intuniform(1, 2147483647));
    }

	//For each sibling send a seperate PUT message
    MyDHTPutCall* dhtMsg = new MyDHTPutCall();
    dhtMsg->setKey(capiPutMsg->getKey());
    dhtMsg->setKind(capiPutMsg->getKind());
    dhtMsg->setId(capiPutMsg->getId());
    dhtMsg->setValue(capiPutMsg->getValue());
	dhtMsg->setTtl(capiPutMsg->getTtl());
    dhtMsg->setIsModifiable(capiPutMsg->getIsModifiable());
    dhtMsg->setMaintenance(false);
    dhtMsg->setBitLength(PUTCALL_L(dhtMsg));
    RECORD_STATS(normalMessages++; numBytesNormal += dhtMsg->getByteLength());
    sendRouteRpcCall(OVERLAY_COMP, capiPutMsg->getKey(),
                         dhtMsg, NULL, DEFAULT_ROUTING, -1,
                         0, capiPutMsg->getNonce());

    PendingRpcsEntry entry;	//Keep track
    entry.putCallMsg = capiPutMsg;
    entry.state = PUT_SENT;
    entry.numResponses = 0;
    entry.numFailed = 0;
    entry.numSent = 1;
    pendingRpcs.insert(make_pair(capiPutMsg->getNonce(), entry));
}


void MyDHT::handleGetCAPIRequest(DHTgetCAPICall* capiGetMsg)
{
	MyDHTGetMessage* getMsg = new MyDHTGetMessage();
	getMsg->setKey(capiGetMsg->getKey());
	getMsg->setKind(capiGetMsg->getKind());
	getMsg->setId(capiGetMsg->getId());
	if(useMsgGroupId)
		getMsg->setGroupId(groupId);			//Optional depending how it's used
	else
		getMsg->setGroupId(-1);
	getMsg->setIsHash(false);
	getMsg->setCache(false);					//Not a cache request
	getMsg->setCRequestersArraySize(0);
	getMsg->setBitLength(GETMSG_L(getMsg));
	getMsg->setNonce(capiGetMsg->getNonce());
	RECORD_STATS(normalMessages++;
				numBytesNormal += getMsg->getByteLength());
	callRoute(capiGetMsg->getKey(), getMsg);	//Send msg

    PendingRpcsEntry entry;	//Keep track
    entry.getCallMsg = capiGetMsg;
    entry.state = GET_VALUE_SENT;
    pendingRpcs.insert(make_pair(capiGetMsg->getNonce(), entry));
}


void MyDHT::handleDumpDhtRequest(DHTdumpCall* call)
{
    DHTdumpResponse* response = new DHTdumpResponse();
    DhtDumpVector* dumpVector = dataStorage->dumpDht();

    response->setRecordArraySize(dumpVector->size());
    for (uint32_t i = 0; i < dumpVector->size(); i++)
        response->setRecord(i, (*dumpVector)[i]);

    delete dumpVector;
    sendRpcResponse(call, response);
}


void MyDHT::handlePutResponse(MyDHTPutResponse* dhtMsg, int rpcId)
{
    PendingRpcs::iterator it = pendingRpcs.find(rpcId);
    if (it == pendingRpcs.end()) // unknown request
        return;

    if (dhtMsg->getSuccess())
	{
        it->second.numResponses++;
		//RECORD_STATS(globalStatistics->recordOutVector("DHT: PUT Hops", dhtMsg->getHopCount()));
		//RECORD_STATS(globalStatistics->addStdDev("DHT: PUT Hops", dhtMsg->getHopCount()));
		//RECORD_STATS(globalStatistics->recordOutVector("DHT: PUT Latency", SIMTIME_DBL(delay)));
		//RECORD_STATS(globalStatistics->addStdDev("DHT: PUT Latency", SIMTIME_DBL(delay)));
	}
	else
        it->second.numFailed++;

    //If more than required no of valid responses, indicate success
    //Otherwise wait till time out or required no of entries are stored
    if (it->second.numResponses / (double)it->second.numSent > 0.5) 
	{
        DHTputCAPIResponse* capiPutRespMsg = new DHTputCAPIResponse();
        capiPutRespMsg->setIsSuccess(true);
        sendRpcResponse(it->second.putCallMsg, capiPutRespMsg);
        pendingRpcs.erase(rpcId);
    }
}


void MyDHT::update(const NodeHandle& node, bool joined)
{
    OverlayKey key;
    bool err = false;
    DhtDataEntry entry;
    std::map<OverlayKey, DhtDataEntry>::iterator it;

    EV << "[MyDHT::update() @ " << overlay->getThisNode().getAddress()
       << " (" << overlay->getThisNode().getKey().toString(16) << ")]\n"
       << "    Update called()" << endl;

	//No secure maintenance
    for (it = dataStorage->begin(); it != dataStorage->end(); it++) 
	{
        key = it->first;
        entry = it->second;
        if (joined) 		//If new node
		{
			//Am I responsible for key && is the node a sibling for key
            if (entry.responsible && (overlay->isSiblingFor(node, key, numReplica, &err)
                    || err)) // hack for Chord, if we've got a new predecessor
			{
                if (err) 
				{
                    EV << "[MyDHT::update()]\n" << " Unable to know if key: " << key
                       << " is in range of node: " << node << endl;
                    // For Chord: we've got a new predecessor
                    // TODO: only send records, if we are not responsible any more
                    // TODO: check all protocols to change routing table first, then call update

                    //if (overlay->isSiblingFor(overlay->getThisNode(), key, 1, &err))
                    //    continue;
                }
                sendMaintenancePutCall(node, key, entry);	//Send entry
            }
        }
        //TODO: move this to the inner block above?
        entry.responsible = overlay->isSiblingFor(overlay->getThisNode(),
                                                  key, 1, &err);
		// TODO Remove this if you are not responsible
		// Doesn't matter if keys are store after INIT phase of all nodes
    }
}


void MyDHT::deliver(OverlayKey& key, cMessage* msg)
{
	MyDHTGetMessage *dhtMsg = dynamic_cast<MyDHTGetMessage*>(msg);
	if(dhtMsg != NULL)
	{
	    if (dhtMsg->getKey().isUnspecified())
        	throw cRuntimeError("MyDHT::handleGetRequest: Unspecified key!");

    	numQueriesAnswered++;
		DhtDumpVector* dataVect = dataStorage->dumpDht(dhtMsg->getKey(),
                                                   dhtMsg->getKind(),
                                                   dhtMsg->getId());
	    // send reply back
    	MyDHTGetResMessage* responseMsg = new MyDHTGetResMessage();
	    responseMsg->setKey(dhtMsg->getKey());
	    responseMsg->setIsHash(dhtMsg->getIsHash());
	    responseMsg->setNonce(dhtMsg->getNonce());
	    responseMsg->setRtt(simTime() - dhtMsg->getCreationTime());	//Track RTT

		OverlayCtrlInfo* overlayCtrlInfo =
        		check_and_cast<OverlayCtrlInfo*>(msg->removeControlInfo());
	    responseMsg->setHopCount(overlayCtrlInfo->getHopCount());	//Track hops

    	if (dataVect->size() == 0)	//No matching entries 
		{
        	responseMsg->setHashValue(BinaryValue::UNSPECIFIED_VALUE);
	        responseMsg->setResultArraySize(0);
    	} 
		else	//Has matching entires 
		{
        	if (dhtMsg->getIsHash()) //Hash needed
			{
	            // TODO: verify this
    	        BinaryValue resultValues;
        	    for (uint32_t i = 0; i < dataVect->size(); i++)
            	    resultValues += (*dataVect)[i].getValue();

	            CSHA1 sha1;
    	        BinaryValue hashValue(20);
        	    sha1.Reset();
            	sha1.Update((uint8_t*) (&(*resultValues.begin())),
                	        resultValues.size());
	            sha1.Final();
    	        sha1.GetHash((unsigned char*)&hashValue[0]);

        	    responseMsg->setHashValue(hashValue);
        	} 
			else 	//value needed
			{
	            responseMsg->setResultArraySize(dataVect->size());
    	        for (uint32_t i = 0; i < dataVect->size(); i++)
        	        responseMsg->setResult(i, (*dataVect)[i]);
        	}
    	}
    	delete dataVect;

	    responseMsg->setBitLength(GETRESPONSE_L(responseMsg));
    	RECORD_STATS(normalMessages++; numBytesNormal += responseMsg->getByteLength());

		//Application port no = overlay port no + 1
		NodeHandle* destNode = new NodeHandle(overlayCtrlInfo->getSrcNode().getKey(),
									overlayCtrlInfo->getSrcNode().getAddress(), 
									overlayCtrlInfo->getSrcNode().getPort() + 1);
		sendMessageToUDP(*destNode, responseMsg->dup());
		//Send to cache requesters, if also present
		for(unsigned int i = 0; i < dhtMsg->getCRequestersArraySize(); i++)
		{
    		RECORD_STATS(normalMessages++; numBytesNormal += responseMsg->getByteLength());
			numCacheResponses++;
			NodeHandle* destNode2 = new NodeHandle(dhtMsg->getCRequesters(i).getKey(),
													dhtMsg->getCRequesters(i).getAddress(), 
													dhtMsg->getCRequesters(i).getPort() + 1);
			sendMessageToUDP(*destNode2, responseMsg->dup());
		}
		delete responseMsg;
		delete dhtMsg;
	}
}


void MyDHT::handleUDPMessage(cMessage* msg)
{
	MyDHTGetResMessage *dhtResMsg = dynamic_cast<MyDHTGetResMessage*>(msg);
	if(dhtResMsg != NULL)
	{
	    PendingRpcs::iterator it = pendingRpcs.find(dhtResMsg->getNonce());
	    if (it == pendingRpcs.end()) // unknown request
		{
	        throw cRuntimeError("MyDHT::forward: Unspecified RPC!");
			delete dhtResMsg;
    	    return;
		}

		RECORD_STATS(globalStatistics->addStdDev("DHT: GET Hops", dhtResMsg->getHopCount()));
		RECORD_STATS(globalStatistics->addStdDev("DHT: GET Latency", SIMTIME_DBL(dhtResMsg->getRtt())));

	    if (it->second.state == GET_VALUE_SENT) 	//Reply for GET request
		{
			RECORD_STATS(globalStatistics->recordOutVector("DHT: GET App Hops", dhtResMsg->getHopCount()));
			RECORD_STATS(globalStatistics->recordOutVector("DHT: GET App Latency", SIMTIME_DBL(dhtResMsg->getRtt())));
			RECORD_STATS(globalStatistics->recordOutVector("DHT: GET App GroupID", groupId));

    	    DHTgetCAPIResponse* capiGetRespMsg = new DHTgetCAPIResponse();

	        if (dhtResMsg->getResultArraySize() > 0) 
			{        		 
				capiGetRespMsg->setResultArraySize(dhtResMsg->getResultArraySize());
		        for (uint i = 0; i < dhtResMsg->getResultArraySize(); i++) //Copy results
    		        capiGetRespMsg->setResult(i, dhtResMsg->getResult(i));
        		capiGetRespMsg->setIsSuccess(true);
			
				//Passive caching
			/*
				CacheDataEntry cacheEntry;
				cacheEntry.key = dhtResMsg->getKey();
    			cacheEntry.value =  dhtResMsg->getResult(0).getValue();
				cacheEntry.kind = 1;
				cacheEntry.id = 1;
				cacheEntry.is_modifiable = dhtResMsg->getResult(0).getIs_modifiable();
				cacheEntry.sourceNode = dhtResMsg->getResult(0).getOwnerNode();

				cacheMap.push_front(cacheEntry);
				//If too much to cache
				if(cacheMap.size() > maxCacheSize)
					cacheMap.pop_back();
			*/
			}
			else	//No results
			{	
                DhtDumpEntry result;
                result.setKey(dhtResMsg->getKey());
                result.setValue(BinaryValue::UNSPECIFIED_VALUE);
				capiGetRespMsg->setResultArraySize(1);
				capiGetRespMsg->setResult(0, result);
				capiGetRespMsg->setIsSuccess(false);
			}
			sendRpcResponse(it->second.getCallMsg, capiGetRespMsg);
			pendingRpcs.erase(dhtResMsg->getNonce());
    	}
		else if(it->second.state == GET_VALUE_CACHE_SENT)	//Reply for a cache request
		{
			RECORD_STATS(globalStatistics->recordOutVector("DHT: GET Cache Hops", dhtResMsg->getHopCount()));
			RECORD_STATS(globalStatistics->recordOutVector("DHT: GET Cache Latency", SIMTIME_DBL(dhtResMsg->getRtt())));
			RECORD_STATS(globalStatistics->recordOutVector("DHT: GET Cache GroupID", groupId));

			for (uint i = 0; i < dhtResMsg->getResultArraySize(); i++)
			{
				if(dhtResMsg->getResult(i).getTtl() > 0)
				{
	        		// add ttl timer
			        MyDHTTtlTimer *timerMsg = new MyDHTTtlTimer("ttl_timer");
        			timerMsg->setKey(dhtResMsg->getKey());
		    	    timerMsg->setKind(1);
        			timerMsg->setId(1);
			        scheduleAt(simTime() + dhtResMsg->getResult(i).getTtl(), timerMsg);

					//Mark as cache entry & not responsible for key
		        	dataStorage->addData(dhtResMsg->getKey(), dhtResMsg->getResult(i).getKind(), 
									dhtResMsg->getResult(i).getId(), dhtResMsg->getResult(i).getValue(), 
									timerMsg, dhtResMsg->getResult(i).getIs_modifiable(), 
									dhtResMsg->getResult(i).getOwnerNode(), false);
				}
				else
				{
		        	dataStorage->addData(dhtResMsg->getKey(), dhtResMsg->getResult(i).getKind(), 
									dhtResMsg->getResult(i).getId(), dhtResMsg->getResult(i).getValue(), 
									NULL, dhtResMsg->getResult(i).getIs_modifiable(), 
									dhtResMsg->getResult(i).getOwnerNode(), false);
				}
				//Cache request is no longer pending
				lookupStorage->setPending(dhtResMsg->getKey(), false);
				cacheSize++;	//Increment cache size
			}
		}
		delete dhtResMsg;
	}
}


void MyDHT::forward(OverlayKey* key, cPacket** msg, NodeHandle* nextHopNode)
{
	if(maxCacheSize == 0)	//No caching
	{
    	numQueriesForwarded++;
		return;
	}

	MyDHTGetMessage* dhtMsg = dynamic_cast<MyDHTGetMessage*>(*msg);	//Extract message
	if(dhtMsg != NULL)
	{
    	if (dhtMsg->getKey().isUnspecified())
	        throw cRuntimeError("MyDHT::forward: Unspecified key!");
    	if ((dhtMsg->getGroupId() != groupId) && !cacheOthersKeys)	//Are we in same group?
			return;

	    OverlayCtrlInfo* overlayCtrlInfo =
    	    check_and_cast<OverlayCtrlInfo*>(dhtMsg->getControlInfo());

		//Repond based on passive cache
	/*
		for(unsigned int i = 0; i < cacheMap.size(); i++)
		{
			if (cacheMap[i].key == dhtMsg->getKey())
			{
		    	MyDHTGetResMessage* responseMsg = new MyDHTGetResMessage(); //send reply back
		    	responseMsg->setKey(dhtMsg->getKey());
	    		responseMsg->setIsHash(dhtMsg->getIsHash());
		    	responseMsg->setNonce(dhtMsg->getNonce());
	    		responseMsg->setRtt(simTime() - dhtMsg->getCreationTime());
	    		responseMsg->setHopCount(overlayCtrlInfo->getHopCount());
	            responseMsg->setResultArraySize(1);

				DhtDumpEntry entry;
				entry.setKey(cacheMap[i].key);
            	entry.setKind(cacheMap[i].kind);
            	entry.setId(cacheMap[i].id);
            	entry.setValue(cacheMap[i].value);
            	entry.setOwnerNode(cacheMap[i].sourceNode);
            	entry.setIs_modifiable(cacheMap[i].is_modifiable);
            	entry.setResponsible(false);

        	    responseMsg->setResult(0, entry);

	    		responseMsg->setBitLength(GETRESPONSE_L(responseMsg));
				//Send to original caller
    			RECORD_STATS(normalMessages++; numBytesNormal += responseMsg->getByteLength());

				//Application port no = overlay port no + 1
				NodeHandle* destNode = new NodeHandle(overlayCtrlInfo->getSrcNode().getKey(),
									overlayCtrlInfo->getSrcNode().getAddress(), 
									overlayCtrlInfo->getSrcNode().getPort() + 1);
				sendMessageToUDP(*destNode, responseMsg);

				*nextHopNode = NodeHandle::UNSPECIFIED_NODE;	//Drop message
				*msg = NULL;
				delete dhtMsg;
    			numQueriesAnswered++;
				return;
			}
		}
    	numQueriesForwarded++;
	}	
	*/

		DhtDumpVector* dataVect = dataStorage->dumpDht(dhtMsg->getKey(),
                                                   dhtMsg->getKind(),
                                                   dhtMsg->getId());

    	if (dataVect->size() == 0) //No matching entries
		{
			//Add as a failed lookup & do nothing to message, should not be a cache request
			if(!dhtMsg->getCache())
			{
	        	bool result = lookupStorage->addLookup(dhtMsg->getKey(), overlayCtrlInfo->getHopCount(), 
														false, alpha, removeLookupThreshold);
				//If a key appeared for 2nd time, it may need to be cached. Should not be periodic caching
				if(result && (cacheRefreshTime <= 0) && (overlay->getThisNode() != overlayCtrlInfo->getSrcNode()))	
				{
					ProfitMap profitMap;	//Sort by profit, max profit at end
					CacheEntry tmpEntry;

					//Get entries from Lookup table	
				    DhtLookupVector* dumpVector = lookupStorage->dumpLookup();
				    std::vector<std::pair<OverlayKey, DhtLookupEntry> >::iterator iter1;

				    for (iter1 = dumpVector->begin(); iter1 != dumpVector->end(); iter1++)
					{
						// TODO Profit function goes here
						double profit = iter1->second.demand;
						tmpEntry.key = iter1->first;
						tmpEntry.success = iter1->second.success;	//Is query successfully answered
						tmpEntry.pending = iter1->second.pending;	//Is cache request is pending
						profitMap.insert(make_pair(profit, tmpEntry));
					}

					//Find the demand for current key with respect to others
					map<double, CacheEntry>::reverse_iterator iter2;
					int count = 0;
					int newKeys = 0;

					//Go in reverse order
					for(iter2 = profitMap.rbegin(); iter2 != profitMap.rend(); iter2++)
					{
						if (count < maxCacheSize)		//Can't cache more than my capacity
						{
							if(iter2->second.key == dhtMsg->getKey())	//Is this the key?
							{
								//get key if not already cached, no cache request is pending, & large profit
								if(!iter2->second.success && !iter2->second.pending && (iter2->first > cachingThreshold))	
								{	
									// Append me as a cache requester to dhtMsg
									int tmpArraySize = dhtMsg->getCRequestersArraySize();
									dhtMsg->setCRequestersArraySize(tmpArraySize + 1);
        	        				dhtMsg->setCRequesters(tmpArraySize, overlay->getThisNode());
									lookupStorage->setPending(iter2->second.key, true);
    								PendingRpcsEntry entry;	//Keep track
								    entry.state = GET_VALUE_CACHE_SENT;
								    pendingRpcs.insert(make_pair(dhtMsg->getNonce(), entry));
    								numCacheRequests++;
									newKeys++;
								}
								break;
							}
							count++;
						}
						else
							break;
					}

					if((newKeys + cacheSize) > maxCacheSize)	//Not enough cache space
					{
						//Delete cache entries giving priority to ones in lookup table
						int noDeleted = dataStorage->deleteCacheWithPriority(&profitMap, 
										(newKeys + cacheSize - maxCacheSize));
						cacheSize -= noDeleted;
					}

    				delete dumpVector;
				}
			}
    		numQueriesForwarded++;
		}
		else	//Matching entries. Reply with what I know and drop message
		{
			//Add as a successful lookup, if not a request to cache
			if(!dhtMsg->getCache())
	        	lookupStorage->addLookup(dhtMsg->getKey(), overlayCtrlInfo->getHopCount(), true, 
											alpha, removeLookupThreshold);
    		numQueriesAnswered++;

		    MyDHTGetResMessage* responseMsg = new MyDHTGetResMessage(); //send reply back
		    responseMsg->setKey(dhtMsg->getKey());
	    	responseMsg->setIsHash(dhtMsg->getIsHash());
		    responseMsg->setNonce(dhtMsg->getNonce());
	    	responseMsg->setRtt(simTime() - dhtMsg->getCreationTime());
	    	responseMsg->setHopCount(overlayCtrlInfo->getHopCount());

        	if (dhtMsg->getIsHash()) //Hash needed
			{
	            // TODO: verify this
    	        BinaryValue resultValues;
        	    for (uint32_t i = 0; i < dataVect->size(); i++)
            	    resultValues += (*dataVect)[i].getValue();

	            CSHA1 sha1;
    	        BinaryValue hashValue(20);
        	    sha1.Reset();
            	sha1.Update((uint8_t*) (&(*resultValues.begin())),
                	        resultValues.size());
	            sha1.Final();
    	        sha1.GetHash((unsigned char*)&hashValue[0]);

        	    responseMsg->setHashValue(hashValue);
        	} 
			else 	//value needed
			{
	            responseMsg->setResultArraySize(dataVect->size());
    	        for (uint32_t i = 0; i < dataVect->size(); i++)
        	        responseMsg->setResult(i, (*dataVect)[i]);
        	}

	    	responseMsg->setBitLength(GETRESPONSE_L(responseMsg));

			//Send to original caller
    		RECORD_STATS(normalMessages++; numBytesNormal += responseMsg->getByteLength());

			//Application port no = overlay port no + 1
			NodeHandle* destNode = new NodeHandle(overlayCtrlInfo->getSrcNode().getKey(),
									overlayCtrlInfo->getSrcNode().getAddress(), 
									overlayCtrlInfo->getSrcNode().getPort() + 1);
			sendMessageToUDP(*destNode, responseMsg->dup());

			//Send to cache requesters, if also present
			for(unsigned int i = 0; i < dhtMsg->getCRequestersArraySize(); i++)
			{
    			RECORD_STATS(normalMessages++; numBytesNormal += responseMsg->getByteLength());
				numCacheResponses++;
				NodeHandle* destNode2 = new NodeHandle(dhtMsg->getCRequesters(i).getKey(),
													dhtMsg->getCRequesters(i).getAddress(), 
													dhtMsg->getCRequesters(i).getPort() + 1);
				sendMessageToUDP(*destNode2, responseMsg->dup());
			}

			delete responseMsg;
			*nextHopNode = NodeHandle::UNSPECIFIED_NODE;	//Drop message
			*msg = NULL;
			delete dhtMsg;
		}
    	delete dataVect;
	}

}


void MyDHT::sendMaintenancePutCall(const TransportAddress& node,
									const OverlayKey& key,                                 
									const DhtDataEntry& entry) 
{
	MyDHTPutCall* dhtMsg = new MyDHTPutCall();

    dhtMsg->setKey(key);
    dhtMsg->setKind(entry.kind);
    dhtMsg->setId(entry.id);
    dhtMsg->setValue(entry.value);

	if(entry.ttlMessage != NULL)	//Remaining TTL
	{
	    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);
}



void MyDHT::reevaluateCache()
{
	ProfitMap profitMap;	//Sort by profit, max profit at end
	CacheEntry tmpEntry;

	//Get entries from Lookup table	
    DhtLookupVector* dumpVector = lookupStorage->dumpLookup();
    std::vector<std::pair<OverlayKey, DhtLookupEntry> >::iterator iter1;

    for (iter1 = dumpVector->begin(); iter1 != dumpVector->end(); iter1++)
	{
		// TODO Profit function goes here
		//int profit = iter1->second.noLookups;
		double profit = iter1->second.demand;
		//int profit = iter1->second.totalHops;
		//int profit = iter1->second.noLookups * iter1->second.totalHops;
		tmpEntry.key = iter1->first;
		tmpEntry.success = iter1->second.success;	//Is query successfully answered
		tmpEntry.pending = iter1->second.pending;	//Is cache request is pending
		profitMap.insert(make_pair(profit, tmpEntry));
	}

	//Find keys with maximum profit
	//map<int, CacheEntry>::reverse_iterator iter2;
	map<double, CacheEntry>::reverse_iterator iter2;
	int count = 0;
	int newKeys = 0;

	//Go in reverse order
	for(iter2 = profitMap.rbegin(); iter2 != profitMap.rend(); iter2++)
	{
		if (count < maxCacheSize)		//Can't cache more than my capacity
		{
			//get key if not already cached, no cache request is pending, & large profit
			if(!iter2->second.success && !iter2->second.pending && (iter2->first > 0.1))	
			//if(!iter2->second.success)
			{	
				//Don't ask if already in Cache
				DhtDataEntry* entry;
				entry = dataStorage->getDataEntry(iter2->second.key, 1, 1);
				if(entry == NULL)
				{
					lookupStorage->setPending(iter2->second.key, true);
					newKeys++;
					getCacheKey(iter2->second.key);
				}
			}
			count++;
		}
		else
			break;
	}

	if((newKeys + cacheSize) > maxCacheSize)	//Not enough cache space
	{
		//Delete cache entries giving priority to ones in lookup table
		int noDeleted = dataStorage->deleteCacheWithPriority(&profitMap, 
										(newKeys + cacheSize - maxCacheSize));
		cacheSize -= noDeleted;
	}

	lookupStorage->clear();	//Cleanup stat for last round
    delete dumpVector;
}


void MyDHT::getCacheKey(OverlayKey key)
{
	MyDHTGetMessage* getMsg = new MyDHTGetMessage();
	getMsg->setKey(key);
	getMsg->setKind(1);
	getMsg->setId(1);
	if(useMsgGroupId)
		getMsg->setGroupId(groupId);
	else
		getMsg->setGroupId(-1);
	getMsg->setIsHash(false);
	getMsg->setCache(true);
	getMsg->setBitLength(GETMSG_L(getMsg));
	uint32_t nonce = intuniform(1, 2147483647);	//generate nonce
	getMsg->setNonce(nonce);
	RECORD_STATS(normalMessages++;
				numBytesNormal += getMsg->getByteLength());
	callRoute(key, getMsg);	//Send msg

    PendingRpcsEntry entry;	//Keep track
    entry.state = GET_VALUE_CACHE_SENT;
    pendingRpcs.insert(make_pair(nonce, entry));
}

void MyDHT::finishApp()
{
/*
    DhtLookupVector* dumpVector2 = lookupStorage->dumpLookup();
    std::vector<std::pair<OverlayKey, DhtLookupEntry> >::iterator iter;
    for (iter = dumpVector2->begin(); iter != dumpVector2->end(); iter++)
	{
		cout << "Key " << iter->first << " noLookups " 
			<< iter->second.noLookups << " totalHops" 
			<< iter->second.totalHops << " success " 
			<< iter->second.success << endl;
	}
    delete dumpVector2;
*/
    simtime_t time = globalStatistics->calcMeasuredLifetime(creationTime);
    if (time >= GlobalStatistics::MIN_MEASURED) 
	{
        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);
		RECORD_STATS(globalStatistics->recordOutVector(
			"DHT: Queries Answered", numQueriesAnswered));
		RECORD_STATS(globalStatistics->recordOutVector(
			"DHT: Queries Forwarded", numQueriesForwarded));
		RECORD_STATS(globalStatistics->recordOutVector(
			"DHT: Cache Requests", numCacheRequests));
		RECORD_STATS(globalStatistics->recordOutVector(
			"DHT: Cache Responses", numCacheResponses));
		DhtDumpVector* dataVect = dataStorage->dumpDht();
		RECORD_STATS(globalStatistics->recordOutVector(
					"DHT: Keys per node", dataVect->size()));
    }
}


int MyDHT::resultValuesBitLength(MyDHTGetResMessage* msg) 
{
    int bitSize = 0;
    for (uint i = 0; i < msg->getResultArraySize(); i++) 
        bitSize += msg->getResult(i).getValue().size();

    return bitSize;
}

std::ostream& operator<<(std::ostream& os, const MyDHT::PendingRpcsEntry& entry)
{
    if (entry.getCallMsg) 
        os << "GET";
    else if (entry.putCallMsg) 
        os << "PUT";

    os << " state: " << entry.state
       << " numSent: " << entry.numSent
       << " numResponses: " << entry.numResponses
       << " numFailed: " << entry.numFailed
       << " numAvailableReplica: " << entry.numAvailableReplica;

    if (entry.replica.size() > 0)
        os << " replicaSize: " << entry.replica.size();

    if (entry.hashVector != NULL) 
        os << " hashVectorSize: " << entry.hashVector->size();

    if (entry.hashes.size() > 0) 
	{
        os << " hashes:";
        std::map<BinaryValue, NodeVector>::const_iterator it;

        int i = 0;
        for (it = entry.hashes.begin(); it != entry.hashes.end(); it++, i++) 
            os << " hash" << i << ":" << it->second.size();
    }

    return os;
}

