M1-nw-project/neighbours.cpp

/***************************************************************************
 * By Théophile Bastian, 2017
 * M1 Network course project at ENS Cachan, Juliusz Chroboczek.
 * License: WTFPL v2 <http://www.wtfpl.net/>
 **************************************************************************/

#include "neighbours.h"
#include "nw_constants.h"

#include <cstdlib>

using namespace std;

Neighbours::Neighbours(Protocol* proto, DataStore* dataStore) :
	proto(proto), dataStore(dataStore), lastPeerPeek(0), lastSentNR(0)
{}

Neighbours::~Neighbours() {
	for(auto fl : dataFlooder) {
		delete fl.second;
	}
}

void Neighbours::fullCheck() {
	for(auto nei : neiType) {
		switch(nei.second) {
			case NEI_SYM:
				if(time(NULL) - lastRecv[nei.first] > csts::TIMEOUT_SYM_RECV
						|| (time(NULL) - lastIHU[nei.first]
						   > csts::TIMEOUT_SYM_IHU))
				{
					changeNeiType(nei.first, NEI_POTENTIAL);
				}
				break;
			case NEI_UNIDIR:
				if(time(NULL) - lastRecv[nei.first] > csts::TIMEOUT_UNIDIR)
					changeNeiType(nei.first, NEI_POTENTIAL);
				break;
			default:
				break;
		}
	}
}

void Neighbours::fullUpdate() {
	fullCheck();

	for(auto nei : unidirNei)
		updateSendPackets(nei);
	for(auto nei : symNei)
		updateSendPackets(nei);

	if(potentialNei.size() > 0
			&& symNei.size() < csts::SYM_COUNT_BEFORE_PEEK
			&& time(NULL) - lastPeerPeek >= csts::TIME_PEER_PEEK)
	{
		auto it = randPeer(&potentialNei);
		proto->sendEmpty(it->id);
		lastPckSent[it->id] = time(NULL);
		lastPeerPeek = time(NULL);
	}


	if(symNei.size() > 0 && potentialNei.size() < 5
			&& time(NULL) - lastSentNR >= csts::TIME_SEND_NR)
	{
		auto it = randPeer(&symNei);
		proto->sendNReq(it->id);
		lastPckSent[it->id] = time(NULL);
		lastSentNR = time(NULL);
	}

	for(auto flooder=dataFlooder.begin(); flooder != dataFlooder.end(); )
	{
		if(flooder->second->done()) {
			delete flooder->second;
			flooder = dataFlooder.erase(flooder);
		}
		else {
			flooder->second->update();
			++flooder;
		}
	}

	for(u64 datId : dataStore->toFlood()) {
		dataFlooder.insert({datId,
				new Flooder(
					(*dataStore)[datId], datId, dataStore->getSeqno(datId),
					proto, symNei)});
		dataStore->setFlooded(datId);
	}
}

void Neighbours::addPotentialNei(const Neighbour& nei) {
	if(neiType.find(nei.id) != neiType.end())
		return; // We already know him
	fprintf(stderr, "[INFO] Adding potential neighbour %lX\n", nei.id);
	potentialNei.push_back(nei);
	lastRecv.insert({nei.id, 0});
	lastIHU.insert({nei.id, 0});
	neiType.insert({nei.id, NEI_POTENTIAL});
	proto->addIdAddr(nei.addr, nei.id);
}

void Neighbours::receivedFrom(u64 id, const SockAddr& addr) {
	if(neiType.find(id) == neiType.end())
		addPotentialNei(Neighbour(id, addr));

	NeiType typ = neiType[id];
	lastRecv[id] = time(NULL);

	if(typ == NEI_POTENTIAL)
		changeNeiType(id, NEI_UNIDIR);
}

void Neighbours::hadIHU(u64 id, const SockAddr& addr) {
	if(neiType.find(id) == neiType.end()) {
		fprintf(stderr, "[WARNING] %lX heard us — yet we did not send "
				"anything?\n", id);
		addPotentialNei(Neighbour(id, addr));
	}

	NeiType typ = neiType[id];
	lastRecv[id] = time(NULL);
	lastIHU[id] = time(NULL);
	if(typ == NEI_POTENTIAL || typ == NEI_UNIDIR)
		changeNeiType(id, NEI_SYM);
}

void Neighbours::getNeighbours(vector<Neighbour>& out, u64 except, int count) {
	vector<list<Neighbour>::iterator > permutation;
	for(auto it = symNei.begin(); it != symNei.end(); ++it)
		permutation.push_back(it);
	for(size_t i=0; i < permutation.size(); i++) {
		int swapWith = rand() % permutation.size();
		auto tmp = permutation[i];
		permutation[i] = permutation[swapWith];
		permutation[swapWith] = tmp;
	}

	for(int cur=0, pos=0; cur < count && (size_t)pos < permutation.size();
			pos++) {
		if(permutation[pos]->id != except) {
			out.push_back(*permutation[pos]);
			cur++;
		}
	}
}

list<Neighbour>* Neighbours::listOfType(NeiType typ) {
	switch(typ) {
		case NEI_POTENTIAL:
			return &potentialNei;
		case NEI_UNIDIR:
			return &unidirNei;
		case NEI_SYM:
			return &symNei;
		default:
			break;
	}
	throw WrongNeiType();
}

void Neighbours::gotIHave(u64 from, u64 datId, u32 seqno) {
	if(dataFlooder.find(datId) != dataFlooder.end()) {
		dataFlooder.at(datId)->gotIHave(from, seqno);
	}
}

void Neighbours::dump() {
	for(auto nei : potentialNei)
		dumpNei(nei, NEI_POTENTIAL);
	for(auto nei : unidirNei)
		dumpNei(nei, NEI_UNIDIR);
	for(auto nei : symNei)
		dumpNei(nei, NEI_SYM);
}

void Neighbours::dumpNei(const Neighbour& nei, NeiType type) {
	printf(">> NEIGHBOUR ");
	switch(type) {
		case NEI_POTENTIAL:
			printf("potential ");
			break;
		case NEI_UNIDIR:
			printf("unidir ");
			break;
		case NEI_SYM:
			printf("sym ");
			break;
		default:
			break;
	}
	printf("%lX lastPck=%ld lastIHU=%ld stateMatch=%d\n",
			nei.id, time(NULL) - lastRecv[nei.id],
			time(NULL) - lastIHU[nei.id],
			type == neiType[nei.id]);
}

void Neighbours::changeNeiType(u64 id, NeiType nType) {
	NeiType cType = neiType[id];
	if(cType == nType)
		return;
	list<Neighbour> *fromList = listOfType(cType),
		*toList = listOfType(nType);
	neiType[id] = nType;

	bool wasSpliced=false;
	for(auto it=fromList->begin(); it != fromList->end(); ++it) {
		if(it->id == id) {
			toList->push_back(*it); // splice() doesn't work?!
			fromList->erase(it);
			wasSpliced=true;
			break;
		}
	}
	if(!wasSpliced) {
		fprintf(stderr, "[ERROR] Node %lX wasn't found (type change)\n", id);
		return;
	}

	if(nType == NEI_SYM) {
		floodTo(id);
	}
}

void Neighbours::updateSendPackets(const Neighbour& nei) {
	if(!sendIHU(nei.id))
		sendEmpty(nei.id);
}

bool Neighbours::sendEmpty(u64 id) {
	if(time(NULL) - lastPckSent[id] >= csts::TIME_RESEND_EMPTY) {
		lastPckSent[id] = time(NULL);
		proto->sendEmpty(id);
		return true;
	}
	return false;
}

bool Neighbours::sendIHU(u64 id) {
	if(time(NULL) - lastIHUSent[id] >= csts::TIME_RESEND_IHU) {
		lastIHUSent[id] = time(NULL);
		lastPckSent[id] = time(NULL);
		proto->sendIHU(id);
		return true;
	}
	return false;
}

list<Neighbour>::iterator Neighbours::randPeer(list<Neighbour>* list) {
	int nPeerId = rand() % list->size();
	auto it = list->begin();
	for(int at=0; at < nPeerId; ++at, ++it);
	return it;
}

void Neighbours::floodTo(u64 peer) {
	vector<u64> datIds;
	dataStore->ids(datIds);
	for(u64 datId : datIds) {
		auto curFlooder = dataFlooder.find(datId);
		if(curFlooder != dataFlooder.end() && !curFlooder->second->done()) {
			curFlooder->second->addPeer(peer);
		}
		else {
			dataFlooder.insert({datId, new Flooder(
						(*dataStore)[datId], datId,
						dataStore->getSeqno(datId),	proto, peer)});
		}
	}
}