MP2Node.cpp

/**********************************
 * FILE NAME: MP2Node.cpp
 *
 * DESCRIPTION: MP2Node class definition
 **********************************/
#include "MP2Node.h"
static int debug = 0;
/*Custom log redirector*/
#define mylog(fmt,...)  if(debug) log->LOG(&(memberNode->addr),fmt,__VA_ARGS__);


/* Definitons of the custom message wrapper */
string MyMessage::stripMyHeader(string message){
	/*Strip the custom header from the message and return the rest */
	int pos = message.find('@');
	return message.substr(pos+1);
}
MyMessage::MyMessage(string message):Message(MyMessage::stripMyHeader(message)){
	int  header = stoi(message.substr(0,message.find('@')));
	msgType = static_cast<MyMessageType>(header);
}
MyMessage::MyMessage(MyMessage::MyMessageType mt, string normalMsg):Message(normalMsg),msgType(mt){
}
MyMessage::MyMessage(MyMessage::MyMessageType mt, Message normalMsg):Message(normalMsg),msgType(mt){
}
string MyMessage::toString(){
	return to_string(msgType) + '@' + Message::toString();
}


/**
 * constructor
 */
MP2Node::MP2Node(Member *memberNode, Params *par, EmulNet * emulNet, Log * log, Address * address) {
	this->memberNode = memberNode;
	this->par = par;
	this->emulNet = emulNet;
	this->log = log;
	ht = new HashTable();
	this->memberNode->addr = *address;
	this->inited = false;
}

/**
 * Destructor
 */
MP2Node::~MP2Node() {
	delete ht;
	delete memberNode;
}

/**
 * FUNCTION NAME: updateRing
 *
 * DESCRIPTION: This function does the following:
 * 				1) Gets the current membership list from the Membership Protocol (MP1Node)
 * 				   The membership list is returned as a vector of Nodes. See Node class in Node.h
 * 				2) Constructs the ring based on the membership list
 * 				3) Calls the Stabilization Protocol
 */
void MP2Node::updateRing() {
	/*
	 * Implement this. Parts of it are already implemented
	 */
	vector<Node> curMemList;
	bool change = false;
	/*
	 *  Step 1. Get the current membership list from Membership Protocol / MP1
	 */
	curMemList = getMembershipList();

	/*
	 * Step 2: Construct the ring
	 */
	// Sort the list based on the hashCode
	sort(curMemList.begin(), curMemList.end());
	/* right now create the ring as a copy of the sorted member list */
	ring = curMemList;
	/*Check the status of replicas relative to your position in the ring */
	/*
	 * Step 3: Run the stabilization protocol IF REQUIRED
	 */
	stabilizationProtocol();
}

/**
 * FUNCTION NAME: getMemberhipList
 *
 * DESCRIPTION: This function goes through the membership list from the Membership protocol/MP1 and
 * 				i) generates the hash code for each member
 * 				ii) populates the ring member in MP2Node class
 * 				It returns a vector of Nodes. Each element in the vector contain the following fields:
 * 				a) Address of the node
 * 				b) Hash code obtained by consistent hashing of the Address
 */
vector<Node> MP2Node::getMembershipList() {
	unsigned int i;
	vector<Node> curMemList;
	for ( i = 0 ; i < this->memberNode->memberList.size(); i++ ) {
		Address addressOfThisMember;
		int id = this->memberNode->memberList.at(i).getid();
		short port = this->memberNode->memberList.at(i).getport();
		memcpy(&addressOfThisMember.addr[0], &id, sizeof(int));
		memcpy(&addressOfThisMember.addr[4], &port, sizeof(short));
		curMemList.emplace_back(Node(addressOfThisMember));
	}
	return curMemList;
}

/**
 * FUNCTION NAME: hashFunction
 *
 * DESCRIPTION: This functions hashes the key and returns the position on the ring
 * 				HASH FUNCTION USED FOR CONSISTENT HASHING
 *
 * RETURNS:
 * size_t position on the ring
 */
size_t MP2Node::hashFunction(string key) {
	std::hash<string> hashFunc;
	size_t ret = hashFunc(key);
	return ret%RING_SIZE;
}

/**
 * FUNCTION NAME: clientCreate
 *
 * DESCRIPTION: client side CREATE API
 * 				The function does the following:
 * 				1) Constructs the message
 * 				2) Finds the replicas of this key
 * 				3) Sends a message to the replica
 */
void MP2Node::clientCreate(string key, string value) {
	/*
	 * Implement this
	 */
	g_transID++;
	vector<Node> recipients = findNodes(key);
	assert(recipients.size()==NUM_KEY_REPLICAS);
	Address* sendaddr = &(this->memberNode->addr);
	for (int i=0;i<NUM_KEY_REPLICAS;++i){
		MyMessage createMsg(MyMessage::QUERY,Message(g_transID,this->memberNode->addr,MessageType::CREATE,key,value,static_cast<ReplicaType>(i)));
		unicastMessage(createMsg,recipients[i].nodeAddress);
	}
	/*Store the transaction ID in your list*/
	transaction tr(g_transID,par->getcurrtime(),QUORUM_COUNT,MessageType::CREATE,key,value);
	translog.push_front(tr);
}

/**
 * FUNCTION NAME: clientRead
 *
 * DESCRIPTION: client side READ API
 * 				The function does the following:
 * 				1) Constructs the message
 * 				2) Finds the replicas of this key
 * 				3) Sends a message to the replica
 */
void MP2Node::clientRead(string key){
	/*
	 * Implement this
	 */
	g_transID++;
	MyMessage createMsg(MyMessage::QUERY,Message(g_transID,this->memberNode->addr,MessageType::READ,key));
	/*Store the transaction ID in your list*/
	transaction tr(g_transID,par->getcurrtime(),QUORUM_COUNT,MessageType::READ,key,"");
	translog.push_front(tr);
	vector<Node> recipients = findNodes(key);
	multicastMessage(createMsg,recipients);
}

/**
 * FUNCTION NAME: clientUpdate
 *
 * DESCRIPTION: client side UPDATE API
 * 				The function does the following:
 * 				1) Constructs the message
 * 				2) Finds the replicas of this key
 * 				3) Sends a message to the replica
 */
void MP2Node::clientUpdate(string key, string value){
	/*
	 * Implement this
	 */
	g_transID++;
	vector<Node> recipients = findNodes(key);
	assert(recipients.size()==NUM_KEY_REPLICAS);
	Address* sendaddr = &(this->memberNode->addr);
	for (int i=0;i<NUM_KEY_REPLICAS;++i){
		MyMessage createMsg(MyMessage::QUERY,Message(g_transID,this->memberNode->addr,MessageType::UPDATE,key,value,static_cast<ReplicaType>(i)));
		unicastMessage(createMsg,recipients[i].nodeAddress);
	}
	/*Store the transaction ID in your list*/
	transaction tr(g_transID,par->getcurrtime(),QUORUM_COUNT,MessageType::UPDATE,key,value);
	translog.push_front(tr);
}

/**
 * FUNCTION NAME: clientDelete
 *
 * DESCRIPTION: client side DELETE API
 * 				The function does the following:
 * 				1) Constructs the message
 * 				2) Finds the replicas of this key
 * 				3) Sends a message to the replica
 */
void MP2Node::clientDelete(string key){
	/*
	 * Implement this
	 */
	if (key=="invalidKey")
		mylog("Got and invalid key client %s","key");
	g_transID++;
	MyMessage createMsg(MyMessage::QUERY,Message(g_transID,this->memberNode->addr,MessageType::DELETE,key));
	/*Store the transaction ID in your list*/
	transaction tr(g_transID,par->getcurrtime(),QUORUM_COUNT,MessageType::DELETE,key,"");
	translog.push_front(tr);
	vector<Node> recipients = findNodes(key);
	multicastMessage(createMsg,recipients);
}

/**
 * FUNCTION NAME: createKeyValue
 *
 * DESCRIPTION: Server side CREATE API
 * 			   	The function does the following:
 * 			   	1) Inserts key value into the local hash table
 * 			   	2) Return true or false based on success or failure
 */
bool MP2Node::createKeyValue(string key, string value, ReplicaType replica) {
	/*
	 * Implement this
	 */
	// Insert key, value, replicaType into the hash table
	Entry e(value,par->getcurrtime(),replica);
	keymap.insert(pair<string,Entry>(key,e));
	return true;

}

/**
 * FUNCTION NAME: readKey
 *
 * DESCRIPTION: Server side READ API
 * 			    This function does the following:
 * 			    1) Read key from local hash table
 * 			    2) Return value
 */
string MP2Node::readKey(string key) {
	/*
	 * Implement this
	 */
	// Read key from local hash table and return value
	KeyMap::iterator it;
	if ((it=keymap.find(key))!=keymap.end())
		return it->second.convertToString();
	else return "";
}

/**
 * FUNCTION NAME: updateKeyValue
 *
 * DESCRIPTION: Server side UPDATE API
 * 				This function does the following:
 * 				1) Update the key to the new value in the local hash table
 * 				2) Return true or false based on success or failure
 */
bool MP2Node::updateKeyValue(string key, string value, ReplicaType replica) {
	/*
	 * Implement this
	 */
	// Update key in local hash table and return true or false
	Entry e(value,par->getcurrtime(),replica);
	KeyMap::iterator it;
	if ((it=keymap.find(key))!=keymap.end()){
		keymap.insert(pair<string,Entry>(key,e));
		return true;
	}else return false;
}

/**
 * FUNCTION NAME: deleteKey
 *
 * DESCRIPTION: Server side DELETE API
 * 				This function does the following:
 * 				1) Delete the key from the local hash table
 * 				2) Return true or false based on success or failure
 */
bool MP2Node::deletekey(string key) {
	/*
	 * Implement this
	 */
	// Delete the key from the local hash table
	if(keymap.erase(key)) return true;
	else return false;
}

/**
 * FUNCTION NAME: checkMessages
 *
 * DESCRIPTION: This function is the message handler of this node.
 * 				This function does the following:
 * 				1) Pops messages from the queue
 * 				2) Handles the messages according to message types
 */
void MP2Node::checkMessages() {
	/*
	 * Implement this. Parts of it are already implemented
	 */
	char * data;
	int size;

	/*
	 * Declare your local variables here
	 */

	// dequeue all messages and handle them
	while ( !memberNode->mp2q.empty() ) {
		/*
		 * Pop a message from the queue
		 */
		data = (char *)memberNode->mp2q.front().elt;
		size = memberNode->mp2q.front().size;
		memberNode->mp2q.pop();

		string message(data, data + size);
		mylog("got the message :%s",data);

		/*
		 * Handle the message types here
		 */
		MyMessage msg(message);
		dispatchMessages(msg);

	}

	/*
	 * This function should also ensure all READ and UPDATE operation
	 * get QUORUM replies
	 */
}

/**
 * FUNCTION NAME: findNodes
 *
 * DESCRIPTION: Find the replicas of the given keyfunction
 * 				This function is responsible for finding the replicas of a key
 */
vector<Node> MP2Node::findNodes(string key) {
	size_t pos = hashFunction(key);
	vector<Node> addr_vec;
	if (ring.size() >= 3) {
		// if pos <= min || pos > max, the leader is the min
		if (pos <= ring.at(0).getHashCode() || pos > ring.at(ring.size()-1).getHashCode()) {
			addr_vec.emplace_back(ring.at(0));
			addr_vec.emplace_back(ring.at(1));
			addr_vec.emplace_back(ring.at(2));
		}
		else {
			// go through the ring until pos <= node
			for (int i=1; i<ring.size(); i++){
				Node addr = ring.at(i);
				if (pos <= addr.getHashCode()) {
					addr_vec.emplace_back(addr);
					addr_vec.emplace_back(ring.at((i+1)%ring.size()));
					addr_vec.emplace_back(ring.at((i+2)%ring.size()));
					break;
				}
			}
		}
	}
	return addr_vec;
}

/* Called everytime in the recvLoop to decrement transaction timers */
void MP2Node::updateTransactionLog(){

	list<transaction>::iterator it=translog.begin();
	while(it!=translog.end())
		if((par->getcurrtime()-it->local_ts)>RESPONSE_WAIT_TIME) {
			MessageType mtype = it->trans_type;
			int transid = it->gtransID;
			mylog("Transaction %d timeout",transid);
			switch(mtype){
			case MessageType::CREATE: log->logCreateFail(&memberNode->addr,true,transid,it->key,it->latest_val.second);break;
			case MessageType::UPDATE: log->logUpdateFail(&memberNode->addr,true,transid,it->key,it->latest_val.second);break;
			case MessageType::READ: log->logReadFail(&memberNode->addr,true,transid,it->key);break;
			case MessageType::DELETE: log->logDeleteFail(&memberNode->addr,true,transid,it->key);break;
			}
			translog.erase(it++);
		}else it++;

}

/**
 * FUNCTION NAME: recvLoop
 *
 * DESCRIPTION: Receive messages from EmulNet and push into the queue (mp2q)
 */
bool MP2Node::recvLoop() {
	if ( memberNode->bFailed ) {
		return false;
	}
	else {
		/*Update the timeout counters on the transaction log*/
		updateTransactionLog();
		return emulNet->ENrecv(&(memberNode->addr), this->enqueueWrapper, NULL, 1, &(memberNode->mp2q));
	}
}

/**
 * FUNCTION NAME: enqueueWrapper
 *
 * DESCRIPTION: Enqueue the message from Emulnet into the queue of MP2Node
 */
int MP2Node::enqueueWrapper(void *env, char *buff, int size) {
	Queue q;
	return q.enqueue((queue<q_elt> *)env, (void *)buff, size);
}
/* Handles the message type and sends it to the appropriate handler */
void MP2Node::dispatchMessages(MyMessage message){
	if(message.msgType==MyMessage::QUERY){
		switch(message.type){
			/* server side message types */
		case MessageType::CREATE: processKeyCreate(message);break;
		case MessageType::UPDATE: processKeyUpdate(message);break;
		case MessageType::DELETE: processKeyDelete(message);break;
		case MessageType::READ: processKeyRead(message);break;

			/* client side message types */
			/* reply message to read requests*/
		case MessageType::READREPLY: processReadReply(message);break;
			/* generic REPLY message (sent/recived in response to CREATE , READ , DELETE requests */
		case MessageType::REPLY: processReply(message);break;
		}
	}else if(message.msgType==MyMessage::REPUPDATE)
		processReplicaUpdate(message);
	else{
		/*message type is unkown, do nothign*/
		mylog("Dropping corrupted packet %s",message.toString().c_str());
		return;
	}
}
/*Stabilization process message handlers */
void MP2Node::processReplicate(Node toNode, ReplicaType repType){
	//push out these keys to the node as a replicate message
	KeyMap::iterator it;
	for(it=keymap.begin();it!=keymap.end();++it){
		if(it->second.replica==repType){
			MyMessage keyupdate(MyMessage::REPUPDATE,Message(-1,(memberNode->addr),MessageType::CREATE,it->first,it->second.value,ReplicaType::TERTIARY));
			unicastMessage(keyupdate,toNode.nodeAddress);
		}
	}
}
void MP2Node::processReplicaUpdate(Message message){
	/*Add the key to the map withe desired key type buffer*/
	createKeyValue(message.key,message.value,message.replica);
}
/* server side message handlers */
void MP2Node::processKeyCreate(Message message){
	MyMessage reply(MyMessage::QUERY,Message(message.transID,(this->memberNode->addr),MessageType::REPLY,false));
	if( createKeyValue(message.key,message.value,message.replica)){
		reply.success=true;
		mylog("create server node with replica %d for key %s",message.replica,message.key.c_str());
		log->logCreateSuccess(&memberNode->addr,false,message.transID,message.key,message.value);
	}else{
		reply.success=false;
		log->logCreateFail(&memberNode->addr,false,message.transID,message.key,message.value);
	}
	unicastMessage(reply,message.fromAddr);
}
void MP2Node::processKeyUpdate(Message message){
	MyMessage reply(MyMessage::QUERY,Message(message.transID,(this->memberNode->addr),MessageType::REPLY,false));
	if( updateKeyValue(message.key,message.value,message.replica)){
		reply.success=true;
		mylog("update server node with replica %d for key %s",message.replica,message.key.c_str());
		log->logUpdateSuccess(&memberNode->addr,false,message.transID,message.key,message.value);
	}else{
		reply.success=false;
		log->logUpdateFail(&memberNode->addr,false,message.transID,message.key,message.value);
	}
	unicastMessage(reply,message.fromAddr);
}
void MP2Node::processKeyDelete(Message message){
	MyMessage reply(MyMessage::QUERY,Message(message.transID,(this->memberNode->addr),MessageType::REPLY,false));
	if( deletekey(message.key)){
		reply.success=true;
		mylog("delete server node with replica %d for key %s",message.replica,message.key.c_str());
		log->logDeleteSuccess(&memberNode->addr,false,message.transID,message.key);
	}
	else{
		reply.success=false;
		log->logDeleteFail(&memberNode->addr,false,message.transID,message.key);
	}
	unicastMessage(reply,message.fromAddr);

}
/*The Key read message format does not have a separate flag for success*/
void MP2Node::processKeyRead(Message message){
	string keyval = readKey(message.key);
	if(!keyval.empty()){
		mylog("read server node with replica %d for key %s",message.replica,message.key.c_str());
		log->logReadSuccess(&memberNode->addr,false,message.transID,message.key,keyval);
	}else{
		log->logReadFail(&memberNode->addr,false,message.transID,message.key);
	}
	MyMessage reply(MyMessage::QUERY,Message(message.transID,(this->memberNode->addr),keyval));
	unicastMessage(reply,message.fromAddr);
}

/* client side message handlers */
/* Process the reply recieved in response to a read request*/
void MP2Node::processReadReply(Message message){
	/* over here, match the reply with the transaction id , and keep
	count of the replies received . if quorum is reached then log the reply,
	else just decrement the quorum counter.
	There may be a conflict in the returned values , some may be older.
	Ideally the reply with the latest global timestamp should be returned
	*/
	string value = message.value;
	//If the node did not have the key there's nothing you can do about it
	if(value.empty()) return;
	//split the value to extract the actual value , timestamp and the replica type
	string delim = ":";
	vector<string> tuple;
	int start = 0;
	int pos = 0;
	while((pos=value.find(delim,start))!=string::npos){
		string token = value.substr(start,pos-start);
		tuple.push_back(token);
		start = pos+1;
	}
	tuple.push_back(value.substr(start));
	assert(tuple.size()==3);
	string keyval = tuple[0];
	int timestamp = stoi(tuple[1]);
	ReplicaType repType = static_cast<ReplicaType>(stoi(tuple[2]));
	int transid = message.transID;
	list<transaction>::iterator it;
	for (it = translog.begin(); it!=translog.end();++it)
		if(it->gtransID==transid)
			break;
		else mylog("unmatched transaction transid %d",it->gtransID);

	//now we have for the key, its value and the timestamp
	if(it==translog.end()){
		//The reply has come in too late and the transaction has been dropped
		//from the log. ignore the reply.
		mylog("dropping reply for transid: %d",transid);
	}else if(--(it->quorum_count)==0){
		//quorum replies received
		//LOG success with the latest val;
		mylog("Received reply for op %d for transid: %d,%d replies remaining",it->trans_type,it->gtransID,it->quorum_count);
		log->logReadSuccess(&memberNode->addr,true,message.transID,it->key,it->latest_val.second);
		//delete from translog
		translog.erase(it);
	}else{
		//LOG reply
		mylog("Received reply for op %d for transid: %d,%d replies remaining",it->trans_type,it->gtransID,it->quorum_count);
		if(timestamp>=it->latest_val.first){
			it->latest_val = pair<int,string>(timestamp,keyval);
			mylog("Changing latest val for transid :%d and key: %s",it->gtransID,it->key.c_str());
		}
	}

}
/* reply received in response to create, read and update queries */
void MP2Node::processReply(Message message){
	int transid = message.transID;
	list<transaction>::iterator it;
	for (it = translog.begin(); it!=translog.end();++it)
		if(it->gtransID==transid)
			break;
	if(it==translog.end()) {
		//The reply has come in too late and the transaction has been dropped
		//from the log. ignore the reply.
		mylog("dropping reply for transid: %d",transid);
	}else if(!message.success){
		//no luck!
	}else if(--(it->quorum_count)==0){
		//quorum replies received
		//LOG success with the latest val and for type trans_type;
		mylog("Received reply for op %d for transid: %d,%d replies remaining",it->trans_type,it->gtransID,it->quorum_count);
		switch(it->trans_type){
		case MessageType::CREATE: log->logCreateSuccess(&memberNode->addr,true,message.transID,it->key,it->latest_val.second);break;
		case MessageType::UPDATE: log->logUpdateSuccess(&memberNode->addr,true,message.transID,it->key,it->latest_val.second);break;
		case MessageType::DELETE: log->logDeleteSuccess(&memberNode->addr,true,message.transID,it->key);break;
		}
		//delete from translog
		translog.erase(it);
	}else{
		mylog("Received reply for op %d for transid: %d,%d replies remaining",it->trans_type,it->gtransID,it->quorum_count);
		//LOG reply
		//safety, as the reply must come after the request
	}
}


/* Send a multicast message about a CRUD operation to all nodes */
void MP2Node::multicastMessage(MyMessage message,vector<Node>& recipients){

	Address* sendaddr = &(this->memberNode->addr);
	string strrep = message.toString();
	char * msgstr = (char*)strrep.c_str();
	size_t msglen = strlen(msgstr);
	mylog("Sending the client request : %s of length %d",msgstr,msglen);
	for (size_t i=0;i<recipients.size();++i)
		this->emulNet->ENsend(sendaddr,&(recipients[i].nodeAddress),msgstr,msglen);
}
/* Send a unicast message */
void MP2Node::unicastMessage(MyMessage message,Address& toaddr){
	Address* sendaddr = &(this->memberNode->addr);
	string strrep = message.toString();
	char * msgstr = (char*)strrep.c_str();
	size_t msglen = strlen(msgstr);
	mylog("Sending the client request : %s  of length %d",msgstr,msglen);
	this->emulNet->ENsend(sendaddr,&toaddr,msgstr,msglen);
}
/**
 * FUNCTION NAME: stabilizationProtocol
 *
 * DESCRIPTION: This runs the stabilization protocol in case of Node joins and leaves
 * 				It ensures that there always 3 copies of all keys in the DHT at all times
 * 				The function does the following:
 *				1) Ensures that there are three "CORRECT" replicas of all the keys in spite of failures and joins
 *				Note:- "CORRECT" replicas implies that every key is replicated in its two neighboring nodes in the ring
 */
void MP2Node::stabilizationProtocol() {
	/*
	 * Implement this
	 */
	int i=0;
	for (i=0;i<ring.size();++i){
		if((ring[i].nodeAddress==this->memberNode->addr))
			break;
	}

	/* Initialize the list of neighbours*/
	int p_2 = ((i-2)<0?i-2+ring.size():i-2)%ring.size();
	int p_1 = ((i-1)<0?i-1+ring.size():i-2)%ring.size() ;
	int n_1 = (i+1)%ring.size();
	int n_2 = (i+2)%ring.size();
	if(!inited){
		/* first time invocation, just update the neighbours, you dont want to detect failure when just joining */
		mylog("Created the member tables initially at ring pos %s",ring[i].nodeAddress.getAddress().c_str());
		haveReplicasOf.push_back(ring[p_2]);
		haveReplicasOf.push_back(ring[p_1]);
		hasMyReplicas.push_back(ring[n_1]);
		hasMyReplicas.push_back(ring[n_2]);
		inited = true;
	}else{
		/*Just another invocation of the event update*/
		mylog("Created the neighbours, position values are %s,%s,<%s>,%s,%s",
				ring[p_2].nodeAddress.getAddress().c_str(),
				ring[p_1].nodeAddress.getAddress().c_str(),
				ring[i].nodeAddress.getAddress().c_str(),
				ring[n_1].nodeAddress.getAddress().c_str(),
				ring[n_2].nodeAddress.getAddress().c_str());
		Node n1 = ring[n_1]; //next node in the ring
		Node n2 = ring[n_2]; //next to next node in the ring
		Node n3 = ring[p_1]; //previous node in the ring
		if(!(n1.nodeAddress==hasMyReplicas[0].nodeAddress)){
			//The next node in the ring failed, send primary to tertiary of next to next node n2
			processReplicate(n2,ReplicaType::PRIMARY);
		}else if(!(n2.nodeAddress==hasMyReplicas[1].nodeAddress)){
			//The next to next node in the ring failed, send primary to tertiary of next to next node n2
			processReplicate(n2,ReplicaType::PRIMARY);
		}else if(!(n3.nodeAddress==haveReplicasOf[0].nodeAddress)){
			//The node previous to this node failed, send secondary to tertiary of the next to next node n2
			processReplicate(n2,ReplicaType::SECONDARY);
		}else{
			//not bothered about processing a change
		}
		//update the has-have tables
		haveReplicasOf.clear();
		hasMyReplicas.clear();
		haveReplicasOf.push_back(ring[p_2]);
		haveReplicasOf.push_back(ring[p_1]);
		hasMyReplicas.push_back(ring[n_1]);
		hasMyReplicas.push_back(ring[n_2]);
	}

}