leader.cc

clustering for ns-2 simulation

			send_ACTION(leData.buffer_action[round].first, leData.buffer_action[round].second, round);
			
			break;
            
        case LEADER_LEADER:
			
			if (_DEBUG_)
				printf("%d timeout LEADER from %d\n", myAddress, from);

            send_LEADER(leData.leader, 
                        leData.tree_size, 
                        leData.parent, 
                        leData.level, 
                        true,
						from);

            break;
    }
}

void
LEADER_Agent::timeoutConfirm(NodeAddress from, int round)
{
	/*
    struct InfoMessage im;
    for (BufferInfoMessages::iterator i = leData.stored_messages.info.begin(); i != leData.stored_messages.info.end(); i++) {
        if (i->first == round) {
            im = i->second;
             if (_DEBUG_)
				 printf("%d TIMEOUT (CONFIRM) from %d (round = %d) (%f)\n", myAddress, from, round, Scheduler::instance().clock());
            send_INFO(im.fragment.ID,
                      im.new_fragment.ID,
                      im.new_fragment.size,
                      im.parent,
                      im.round,
                      from);
        }
    }
	 */
}

//
// Verifica se il messaggio di INFO e'stato gia'mandato per un certo round.
//
bool 
LEADER_Agent::isInfoInBuffer(int round)
{
    for (BufferInfoMessages::iterator i = leData.stored_messages.info.begin(); i != leData.stored_messages.info.end(); i++)
        if (i->first == round)
            return true;
    
    return false;
}

//
// Verifica se il messaggio di FEEDBACK e'stato gia'mandato per un certo round.
//
bool 
LEADER_Agent::isFeedbackInBuffer(int round)
{
    for (BufferFeedbackMessages::iterator i = leData.stored_messages.feedback.begin(); i != leData.stored_messages.feedback.end(); i++)
        if (i->first == round)
            return true;
    
    return false;
}

void
LEADER_Agent::lastTimeoutConfirm(NodeAddress from, int round)
{
    if (_DEBUG_)
        printf("%d LAST TIMEOUT (CONFIRM) from %d [round = %d](%f)\n", myAddress, from, round, Scheduler::instance().clock());
    
}

void 
LEADER_Agent::lastTimeout(LeaderMessageType type, NodeAddress from, int round)
{
    switch (type) {
        case LEADER_INFO : 
			if (_DEBUG_)
				printf("%d LAST TIMEOUT (INFO) from %d (round = %d) (%f)\n", myAddress, from, leData.round, Scheduler::instance().clock());
            break;
		case LEADER_FEEDBACK:
			if (_DEBUG_)
				printf("%d LAST TIMEOUT (FEEDBACK) from %d (round = %d) (%f)\n", myAddress, from, leData.round, Scheduler::instance().clock());            
			break;
		case LEADER_ACTION:
			if (_DEBUG_)
				printf("%d LAST TIMEOUT (ACTION) from %d (round = %d) (%f)\n", myAddress, from, leData.round, Scheduler::instance().clock());            
			break;
        case LEADER_LEADER : 
			if (_DEBUG_)
				printf("%d LAST TIMEOUT (LEADER) from %d (%f)\n", myAddress, from, Scheduler::instance().clock());
            break;
    }
}

//
// Ricezione del messaggio di INFO (anche da buffer).
//
void
LEADER_Agent::recvInfoMessage(NodeAddress from, struct InfoMessage info)
{        
	//
	// Se il messaggio è passato, viene ignorato.
	//
	if (info.round < leData.round)
		return;

    //
    // Se il messaggio è futuro viene bufferizzato.
    //
    if (info.round > leData.round) {
        leData.buffer_info.push_back(pair<NodeAddress, struct InfoMessage>(from, info));
        return;
    }
    
    //
    // Se il messaggio e'il secondo ad arrivare da un nodo allora lo si scarta.
    //
    if (leData.info_messages.find(from) == leData.info_messages.end())
        return;
    
    //
    // Riceve effettivamente il messaggio
    //
    receive_INFO(from, info);
}

//
// Ricezione del messaggio di INFO.
//
void
LEADER_Agent::recvFeedbackMessage(NodeAddress from, struct FeedbackMessage feedback)
{
    //
    // Se il messaggio è un messaggio futuro si bufferizza.
    //
    if (feedback.round > leData.round) {
        leData.buffer_feedback.push_back(pair<NodeAddress, struct FeedbackMessage>(from, feedback));
        return;
    }
    
    //
    // Se il messaggio e'il secondo ad arrivare da un nodo allora lo si scarta.
    //
    if (leData.feedback_messages.find(from) == leData.feedback_messages.end())
        return;
    
    //
    // Riceve il messaggio (round corrente).
    //
    receive_FEEDBACK(from, feedback);
}

bool 
LEADER_Agent::belongToAnotherFragment(NodeAddress node, struct InfoMessage info)
{
    // Se non ho ancora inviato un messaggio di INFO per questo round il mio 
    // "fragment.ID" rappresenta il mio frammento corrente,
    // altrimenti il mio vecchio frammento e' memorizzato in "old_fragment_identity".
    //if (!isInfoInBuffer(info.round)) {
        if (info.fragment.ID == leData.fragment.ID)
            return false;
        // Il messaggio portava un ID che non apparteneva al mio frammento.
        // Cio'nonostante il messaggio puo'indicare un cambiamento di frammento.
        // Effettuo gli stessi controlli sulla variabile info.new_fragment.ID
        return (info.new_fragment.ID != leData.fragment.ID);
	/*
    }
    else {
        if (info.fragment.ID == leData.old_fragment_identity)
            return false;
        // Il messaggio portava un ID che non apparteneva al mio frammento.
        // Cio'nonostante il messaggio puo'indicare un cambiamento di frammento.
        // Effettuo gli stessi controlli sulla variabile info.new_fragment.ID
        return (info.new_fragment.ID != leData.old_fragment_identity);
    }
	*/
}

//
// Gestione dello scadere dei timeout.
//
void
LeaderTimer::handle(Event* e) 
{
    //
    // Verifica che l'evento sia un timeout di INFO.
    //
	for (map<int, TimeoutMap>::iterator rr = info_timeouts.begin(); rr != info_timeouts.end(); rr++) {
		// rr->first e' il round'
		// rr->second e'la mappa dei timeout.
		for (TimeoutMap::iterator i = rr->second.begin(); i != rr->second.end(); i++) {
			if ((i->second).timeout == e) {
				(i->second).num--;
				if ((i->second).num <= 0)
					agent->lastTimeout(LEADER_INFO, i->first, rr->first);
				else {
					Scheduler::instance().schedule(this, e, LEADER_Agent::timeout_info + Random::uniform(LEADER_Agent::jitter_timeout_info));
					agent->timeout(LEADER_INFO, i->first, rr->first);
				}
				return;
			}
		}
    }    
        
    //
    // Verify if is a FEEDBACK timeout
    //
    for (TimeoutFromMap::iterator i = feedback_timeouts.begin(); i != feedback_timeouts.end(); i++) {
        if (i->second.timeout == e) {
            i->second.num--;
            if (i->second.num <= 0)
                agent->lastTimeout(LEADER_FEEDBACK, i->second.from, i->first);
            else {
                Scheduler::instance().schedule(this, e, LEADER_Agent::timeout_feedback + Random::uniform(LEADER_Agent::jitter_timeout_feedback));
                agent->timeout(LEADER_FEEDBACK, i->second.from, i->first);
            }
            return;
    	}
    }
        
	//
	// Messaggi ACTION.
	//
	for (map<int, struct TimeoutFrom>::iterator i = action_timeouts.begin(); i != action_timeouts.end(); i++) {
		if (i->second.timeout == e) {
            i->second.num--;
            if (i->second.num <= 0)
                agent->lastTimeout(LEADER_ACTION, i->second.from, i->first);
            else {
                Scheduler::instance().schedule(this, e, LEADER_Agent::timeout_action + Random::uniform(LEADER_Agent::jitter_timeout_action));
                agent->timeout(LEADER_ACTION, i->second.from, i->first);
            }
            return;
		}
	}

    //
    // Messaggi LEADER.
    //
    for (TimeoutMap::iterator i = leader_timeouts.begin(); i != leader_timeouts.end(); i++) {
        if (i->second.timeout == e) {
            i->second.num--;
            if (i->second.num <= 0)
                agent->lastTimeout(LEADER_LEADER, i->first, 0);
            else {
                Scheduler::instance().schedule(this, e, LEADER_Agent::timeout_leader + Random::uniform(LEADER_Agent::jitter_timeout_leader));
                agent->timeout(LEADER_LEADER, i->first, 0);
            }
            return;
        }
    }    
    
}

//
// Lancia i timeout per i messaggi di INFO del round.
//
void 
LeaderTimer::launchInfoTimeouts(NodeList & info_neighbors, int round)
{
    for (NodeList::iterator i = info_neighbors.begin(); i != info_neighbors.end(); i++) {
        // Inizializza solo quelli che non sono già stati annullati
        struct Timeout t;
        t.timeout = new Event();
        t.num = LEADER_Agent::max_timeout_info;
        info_timeouts[round][*i] = t;
        Scheduler::instance().schedule(this, t.timeout, LEADER_Agent::timeout_info + Random::uniform(LEADER_Agent::jitter_timeout_info));
    }
}

//
// Lancia i timeout per i messaggi di FEEDBACK del round.
//
void 
LeaderTimer::launchFeedbackTimeouts(NodeAddress parent, int round)
{
    struct TimeoutFrom t;
    t.timeout = new Event();
    t.num = LEADER_Agent::max_timeout_feedback;
    t.from = parent;
    feedback_timeouts[round] = t;
    Scheduler::instance().schedule(this, t.timeout, LEADER_Agent::timeout_feedback + Random::uniform(LEADER_Agent::jitter_timeout_feedback));
}

//
// Lancia il timeout per la ricezione del messaggio ACTION per il round.
//
void 
LeaderTimer::launchActionTimeouts(NodeAddress to, int round)
{
	struct TimeoutFrom tf;
	tf.from = to;
	tf.timeout = new Event();
	tf.num = LEADER_Agent::max_timeout_action;
	action_timeouts[round] = tf;
	Scheduler::instance().schedule(this, tf.timeout, LEADER_Agent::timeout_action + Random::uniform(LEADER_Agent::jitter_timeout_action));
}

//
// Lancia i timeout per i messaggi di LEADER del round.
//
void 
LeaderTimer::launchLeaderTimeouts(NodeList & leader_neighbors)
{
    for (NodeList::iterator i = leader_neighbors.begin(); i != leader_neighbors.end(); i++) {
        struct Timeout t;
        t.timeout = new Event();
        t.num = LEADER_Agent::max_timeout_leader;
        leader_timeouts[*i] = t;
        Scheduler::instance().schedule(this, t.timeout, LEADER_Agent::timeout_leader + Random::uniform(LEADER_Agent::jitter_timeout_leader));
    }
}

//
// Riceve un messaggio di INFO
//
void 
LeaderTimer::cancelInfo(NodeAddress node, int round)
{
    TimeoutMap::iterator i;
    for (i = info_timeouts[round].begin(); i != info_timeouts[round].end(); i++) {
        if (i->first == node) {
            Scheduler::instance().cancel((i->second).timeout);
            break;
        }
    }
    if (i != info_timeouts[round].end())
        info_timeouts[round].erase(i);
}

void 
LeaderTimer::cancelAction(NodeAddress node, int round)
{
	map<int, struct TimeoutFrom>::iterator i;
    for (i = action_timeouts.begin(); i != action_timeouts.end(); i++)
        if (((i->first) == round) && ((i->second.from) == node))
			break;
	if (i != action_timeouts.end())
		action_timeouts.erase(i);
}

//
// Riceve un messaggio di INFO
//
void 
LeaderTimer::cancelFeedback(NodeAddress node, int round)
{
    TimeoutFromMap::iterator i;
    for (i = feedback_timeouts.begin(); i != feedback_timeouts.end(); i++) {
        if (((i->first) == round) && ((i->second.from) == node)) {
            Scheduler::instance().cancel(i->second.timeout);
            break;
        }
    }
    if (i != feedback_timeouts.end()) {
        feedback_timeouts.erase(i);
    }
}

//
// Riceve un messaggio di LEADER
//
void 
LeaderTimer::cancelLeader(NodeAddress node)
{
    TimeoutMap::iterator i;
    for (i = leader_timeouts.begin(); i != leader_timeouts.end(); i++) {
        if (i->first == node) {
            Scheduler::instance().cancel(i->second.timeout);
            break;
        }
    }
    if (i != leader_timeouts.end())
        leader_timeouts.erase(i);
}

// Verifica che un pacchetto sia di tipo DATA.
bool 
LEADER_Agent::isDataPacket(Packet * p)
{
    struct hdr_leader *leaderh = HDR_LEADER(p);
	return (leaderh->msg_type == LEADER_DATA);
}
		
// Preparazione all'invio di un messaggio al layer inferiore
void 
LEADER_Agent::prepareSendDataDown(Packet * p)
{
    struct hdr_leader *leaderh = HDR_LEADER(p);

	// Marca il nodo come dato.
	leaderh->msg_type = LEADER_DATA;
}

/**
* Query if algorithm support a particular protocol:
* that method is probably called before a getData
* method.
*/
bool 
LEADER_Agent::supportProtocol(string protocol)
{
	if (protocol == "leader")
		return true;
	return false;
}

/**
* That method is used to get data from an algorithm
* that was check to implements a particular protocol
*/
void * 
LEADER_Agent::getData(string data)
{
	if (data == "leader") {
		return &(leData.leader);
	}
	else if (data == "tree_size") {
		return &(leData.tree_size);
	}
	else if (data == "parent") {
		return &(leData.parent);
	}
	else if (data == "childs") {
		return &(leData.childs);
	}
	else if (data == "level") {
		return &(leData.level);
	}
	else if (data == "levels") {
		return &(leData.levels);
	}
	return NULL;
}

void
LEADER_Agent::beginLeaderPropagation()
{
	// Se il nodo non conosce nessun frammento adiacente massimale, allora 
	// vuol dire che e'l'unico: viene eletto LEADER.
	
	// Genitore
	leData.parent = -1;
	// Numero di nodi della topologia.
	leData.tree_size = leData.fragment.size;
	// Livello nell'albero
	leData.level = 0;
	// Leader.
	leData.leader = myAddress;
	// Childs
	leData.childs.clear();
	
	init_LEADER_cycle();
	
	send_LEADER (myAddress, leData.tree_size, -1, 0);
	
	// Imposta i timeout verso i nodi che devono inviare un messaggio LEADER.
	timer->launchLeaderTimeouts(leData.leader_messages);
}