#ifndef _PRAGMA_COPYRIGHT_
#define _PRAGMA_COPYRIGHT_
#pragma comment(copyright, "%Z% %I% %W% %D% %T%\0")
#endif /* _PRAGMA_COPYRIGHT_ */
/****************************************************************************

* Copyright (c) 2008, 2010 IBM Corporation.
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0s
* which accompanies this distribution, and is available at
* http://www.eclipse.org/legal/epl-v10.html

 Classes: ErrorDetector

 Description: The functions of error detector include:
     a. detect connection broken events from peer processes.
     b. detect heartbeat packets from peer processes
     c. propagate failure data to peer processes
     d. propagate recovery information to peer processes
     e. establish new connection dynamically
     f. accept error injection data for testing purposes
     g. delegate eror information to error handling thread (EHT)
   
 Author: Nicole Nie

 History:
   Date     Who ID    Description
   -------- --- ---   -----------
   04/28/09 nieyy       Initial code (F156654)

****************************************************************************/

#include "errdetector.hpp"
#include <assert.h>
#include <stdlib.h>

#include "log.hpp"
#include "exception.hpp"
#include "socket.hpp"
#include "packer.hpp"
#include "stream.hpp"

#include "ctrlblock.hpp"
#include "routinglist.hpp"
#include "statemachine.hpp"
#include "message.hpp"
#include "queue.hpp"
#include "parent.hpp"
#include "errevent.hpp"
#include "topology.hpp"
#include "initializer.hpp"

ErrorDetector::ErrorDetector(int hndl) 
    : Processor(hndl)
{
    name = "ErrorDetector";

    inQueue = NULL;
    outQueue = NULL;

    parent = NULL;
    uncleList = NULL;
    uncleList2 = NULL;

    needForward = false;
}

ErrorDetector::~ErrorDetector()
{
    if (parent) {
        delete parent;
    }
    if (uncleList) {
        delete uncleList;
    }
    if (uncleList2) {
        delete uncleList2;
    }
}

Message * ErrorDetector::read()
{
    assert(inQueue);

    Message *msg = NULL;

    msg = inQueue->consume();
    
    return msg;
}

void ErrorDetector::process(Message * msg)
{
    assert(msg);

    needForward = false;
    switch(msg->getType()) {
        case Message::UNCLE:
        case Message::UNCLE_LIST:
        case Message::PARENT:
            processParentInfo(msg);
            break;
        case Message::ERROR_EVENT:
            processErrorEvent(msg);
            break;
        case Message::SHUTDOWN:
        case Message::KILLNODE:
            processErrorInjection(msg);
            break;
        case Message::GROUP_MERGE:
            // transfer this message to relay processor
            msg->setRefCount(msg->getRefCount() + 1);
            gCtrlBlock->getRouterInQueue()->produce(msg);
            break;
        default:
            log_error("Processor %s: received unknown command", name.c_str());
            break;
    }
}

void ErrorDetector::write(Message * msg)
{
    if (needForward) {
        if (outQueue) {
            msg->setRefCount(msg->getRefCount() + 1);
        }
        if (gCtrlBlock->getMonitorInQueue()) {
            msg->setRefCount(msg->getRefCount() + 1);
            gCtrlBlock->getMonitorInQueue()->produce(msg);
        }
        if (outQueue) {
            outQueue->produce(msg);
        }
    }
    inQueue->remove();
}

void ErrorDetector::seize()
{
    gStateMachine->parse(StateMachine::FATAL_EXCEPTION);
}

void ErrorDetector::clean()
{
    // no action
}

bool ErrorDetector::isActive()
{
    return gCtrlBlock->isEnabled() || (inQueue->getSize() > 0);
}

void ErrorDetector::processParentInfo(Message * msg)
{
    needForward = false;
    if (msg->getType() == Message::UNCLE) {
        // accumulate uncle information of my grandsons
        Parent *p = new Parent();
        p->unpackMsg(*msg);

        if (NULL == uncleList2) {
            uncleList2 = new ParentList();
            uncleList2->setLevel(gCtrlBlock->getTopology()->getLevel());
        }
        uncleList2->add(p);

        if (uncleList2->isAllGathered()) {
            // pack message and bcast it
            Message *unclesmsg = uncleList2->packMsg();
            gCtrlBlock->getRouterInQueue()->produce(unclesmsg);
        }
    } else if (msg->getType() == Message::UNCLE_LIST) {
        ParentList *tmpList = new ParentList();
        tmpList->unpackMsg(*msg);

        bool ignore = true;
        if (gCtrlBlock->getMyRole() == CtrlBlock::BACK_END) {
            ignore = false;
        } else if (gCtrlBlock->getTopology()->getLevel() > tmpList->getLevel()+1) {
            ignore = false;
        }

        if (ignore) {
            // if it is from my parent, just bcast it and ignore it
            msg->setRefCount(msg->getRefCount() + 1);
            gCtrlBlock->getRouterInQueue()->produce(msg);
            delete tmpList;
        } else {
            // else it is from my grandparent, it's what I need
            if (uncleList) {
                delete uncleList;
                uncleList = NULL;
            }
            uncleList = tmpList;
        }
    } else {
        if (parent) {
            delete parent;
            parent = NULL;
        }
        parent = new Parent();
        parent->unpackMsg(*msg);
    }
}

void ErrorDetector::processErrorEvent(Message * msg)
{
    ErrorEvent event;
    event.unpackMsg(*msg);

    Message *notifymsg = NULL;

    needForward = true;
    switch(event.getErrCode()) {
        case SCI_ERR_PARENT_BROKEN:
            // if I am an agent, broadcast this message to all my children
            if (gCtrlBlock->getMyRole() == CtrlBlock::AGENT) {
                msg->setRefCount(msg->getRefCount() + 1);
                gCtrlBlock->getRouterInQueue()->produce(msg);
            }
            // and if this event is generated by myself, try to recover the connection
            if (event.getNodeID() == gCtrlBlock->getMyHandle()) {
                recover();
            }
            break;
        case SCI_ERR_CHILD_BROKEN:
            // notify router processor to drop groups related to this child, use 'id' field to
            // store the successor id information
            notifymsg = new Message();
            notifymsg->build(SCI_FILTER_NULL, SCI_GROUP_ALL, 0, NULL, NULL, Message::GROUP_DROP, 
                event.getNodeID());
            gCtrlBlock->getRouterInQueue()->produce(notifymsg);
            
            // and if I am an agent, upload this message to my parent
            if (gCtrlBlock->getMyRole() == CtrlBlock::AGENT) {
                msg->setRefCount(msg->getRefCount() + 1);
                gCtrlBlock->getFilterOutQueue()->produce(msg);
            }
            break;
        case SCI_ERR_RECOVERED:
        case SCI_ERR_RECOVER_FAILED:
            if (gCtrlBlock->getMyRole() == CtrlBlock::AGENT) {
                if (msg->getID() == gCtrlBlock->getMyHandle()) {
                    // from my parent, bcast to my children
                    msg->setRefCount(msg->getRefCount() + 1);
                    gCtrlBlock->getRouterInQueue()->produce(msg);
                } else {
                    // from my child, upload to my parent
                    msg->setRefCount(msg->getRefCount() + 1);
                    gCtrlBlock->getFilterOutQueue()->produce(msg);
                }
            }
            break;
        default:
            break;
    }
}

void ErrorDetector::processErrorInjection(Message * msg)
{
    Packer packer(msg->getContentBuf());

    int nodeId = packer.unpackInt();
    if (nodeId == gCtrlBlock->getMyHandle()) {
        if (msg->getType() == Message::SHUTDOWN) {
            // if i am not a front end, shutdown the connection between me and my parent
            if (gCtrlBlock->getMyRole() != CtrlBlock::FRONT_END) {
                gCtrlBlock->getParentStream()->stop();
            }
        } else {
            // force this node to exit
            ::_exit(1);
        }
    } else {
        // if i am not a back end, bcast messages to all my children
        if (gCtrlBlock->getMyRole() != CtrlBlock::BACK_END) {
            msg->setRefCount(msg->getRefCount() + 1);
            gCtrlBlock->getRouterInQueue()->produce(msg);
        }
    }
}

void ErrorDetector::recover()
{
    Stream *stream = NULL;
    int pid;

    if (parent) {
        // first try my parent
        stream = parent->connect();
        pid = parent->getNodeID();
    }
    if ((!stream) && uncleList) {
        // if can't connect to parent, then try my uncles
        stream = uncleList->select(&pid);
    }

    // generate an error message and put it into error message queue
    ErrorEvent event;
    event.setNodeID(gCtrlBlock->getMyHandle());
    event.setBENum(gRoutingList->numOfBE(SCI_GROUP_ALL));
    
    if (stream) {
        event.setErrCode(SCI_ERR_RECOVERED);

        // reset the corresponding processor
        if (gCtrlBlock->getMyRole() == CtrlBlock::BACK_END) {
            gInitializer->recoverBE(stream);
        } else {
            gInitializer->recoverAgent(stream);
        }

        gCtrlBlock->registerStream(stream);
        gCtrlBlock->setParentStream(stream);
        gStateMachine->parse(StateMachine::RECOVER_OK);
        gRoutingList->updateParentId(pid);
    } else {
        event.setErrCode(SCI_ERR_RECOVER_FAILED);
        gStateMachine->parse(StateMachine::RECOVER_FAILED);
    }

    Message *msg = event.packMsg();
    if (gCtrlBlock->getMyRole() == CtrlBlock::AGENT) {
        msg->setRefCount(msg->getRefCount() + 1);
        // bcast this message to all my successors
        gCtrlBlock->getRouterInQueue()->produce(msg);
        // upload this message to my parent
        gCtrlBlock->getFilterOutQueue()->produce(msg);
    } else if (gCtrlBlock->getMyRole() == CtrlBlock::BACK_END) {
        // forward this message to Error Handling Thread (EHT) if have
        if (outQueue) {
            msg->setRefCount(msg->getRefCount() + 1);
            outQueue->produce(msg);
        }
        
        // upload this message to my parent
        gCtrlBlock->getRouterInQueue()->produce(msg);
    } else {
        assert(!"Should never be here");
    }
}

void ErrorDetector::setInQueue(MessageQueue * queue)
{
    inQueue = queue;
}

void ErrorDetector::setOutQueue(MessageQueue * queue)
{
    outQueue = queue;
}