// **********************************************************************
//
// Copyright (c) 2003-2016 ZeroC, Inc. All rights reserved.
//
// This copy of Ice is licensed to you under the terms described in the
// ICE_LICENSE file included in this distribution.
//
// **********************************************************************

#include <IceStorm/NodeI.h>
#include <IceStorm/Observers.h>
#include <IceStorm/TraceLevels.h>

using namespace IceStorm;
using namespace IceStormElection;
using namespace std;

namespace
{

class CheckTask : public IceUtil::TimerTask
{
    const NodeIPtr _node;

public:

    CheckTask(const NodeIPtr& node) : _node(node) { }
    virtual void runTimerTask()
    {
        _node->check();
    }
};

class MergeTask : public IceUtil::TimerTask
{
    const NodeIPtr _node;
    const set<int> _s;

public:

    MergeTask(const NodeIPtr& node, const set<int>& s) : _node(node), _s(s) { }
    virtual void runTimerTask()
    {
        _node->merge(_s);
    }
};

class MergeContinueTask : public IceUtil::TimerTask
{
    const NodeIPtr _node;

public:

    MergeContinueTask(const NodeIPtr& node) : _node(node) { }
    virtual void runTimerTask()
    {
        _node->mergeContinue();
    }
};

class TimeoutTask: public IceUtil::TimerTask
{
    const NodeIPtr _node;

public:

    TimeoutTask(const NodeIPtr& node) : _node(node) { }
    virtual void runTimerTask()
    {
        _node->timeout();
    }
};

}

namespace
{

LogUpdate emptyLU = {0, 0};

}

GroupNodeInfo::GroupNodeInfo(int i) :
    id(i), llu(emptyLU)
{
}

GroupNodeInfo::GroupNodeInfo(int i, LogUpdate l, const Ice::ObjectPrx& o) :
    id(i), llu(l), observer(o)
{
}

bool
GroupNodeInfo::operator<(const GroupNodeInfo& rhs) const
{
    return id < rhs.id;
}

bool
GroupNodeInfo::operator==(const GroupNodeInfo& rhs) const
{
    return id == rhs.id;
}

//
// COMPILER FIX: Clang using libc++ requires to define operator=
//
#if defined(__clang__) && defined(_LIBCPP_VERSION)
GroupNodeInfo&
GroupNodeInfo::operator=(const GroupNodeInfo& other)
    
{
    const_cast<int&>(this->id) = other.id;
    const_cast<LogUpdate&>(this->llu) = other.llu;
    const_cast<Ice::ObjectPrx&>(this->observer) = other.observer;
    return *this;
}
#endif

Replica::~Replica()
{
    //cout << "~Replica" << endl;
}

namespace
{
static IceUtil::Time
getTimeout(const string& key, int def, const Ice::PropertiesPtr& properties, const TraceLevelsPtr& traceLevels)
{
    int t = properties->getPropertyAsIntWithDefault(key, def);
    if(t < 0)
    {
        Ice::Warning out(traceLevels->logger);
        out << traceLevels->electionCat << ": " << key << " < 0; Adjusted to 1";
        t = 1;
    }
    return IceUtil::Time::seconds(t);
}

static string
toString(const set<int>& s)
{
    ostringstream os;
    os << "(";
    for(set<int>::const_iterator p = s.begin(); p != s.end(); ++p)
    {
        if(p != s.begin())
        {
            os << ",";
        }
        os << *p;
    }
    os << ")";
    return os.str();
}

}

NodeI::NodeI(const InstancePtr& instance,
             const ReplicaPtr& replica,
             const Ice::ObjectPrx& replicaProxy,
             int id, const map<int, NodePrx>& nodes) :
    _timer(instance->timer()),
    _traceLevels(instance->traceLevels()),
    _observers(instance->observers()),
    _replica(replica),
    _replicaProxy(replicaProxy),
    _id(id),
    _nodes(nodes),
    _state(NodeStateInactive),
    _updateCounter(0),
    _max(0),
    _generation(-1),
    _destroy(false)
{
    map<int, NodePrx> oneway;
    for(map<int, NodePrx>::const_iterator p = _nodes.begin(); p != _nodes.end(); ++p)
    {
        oneway[p->first] = NodePrx::uncheckedCast(p->second->ice_oneway());
    }
    const_cast<map<int, NodePrx>& >(_nodesOneway) = oneway;

    Ice::PropertiesPtr properties = instance->communicator()->getProperties();
    const_cast<IceUtil::Time&>(_masterTimeout) = getTimeout(
        instance->serviceName() + ".Election.MasterTimeout", 10, properties, _traceLevels);
    const_cast<IceUtil::Time&>(_electionTimeout) = getTimeout(
        instance->serviceName() + ".Election.ElectionTimeout", 10, properties, _traceLevels);
    const_cast<IceUtil::Time&>(_mergeTimeout) = getTimeout(
        instance->serviceName() + ".Election.ResponseTimeout", 10, properties, _traceLevels);
}

NodeI::~NodeI()
{
    //cout << "~NodeI" << endl;
}

void
NodeI::start()
{
    // As an optimization we want the initial election to occur as
    // soon as possible.
    //
    // However, if we have the node trigger the election immediately
    // upon startup then we'll have a clash with lower priority nodes
    // starting an election denying a higher priority node the
    // opportunity to start the election that results in it becoming
    // the leader. Of course, things will eventually reach a stable
    // state but it will take longer.
    //
    // As such as we schedule the initial election check inversely
    // proportional to our priority.
    //
    // By setting _checkTask first we stop recovery() from setting it
    // to the regular election interval.
    //
    
    //
    // We use this lock to ensure that recovery is called before CheckTask
    // is scheduled, even if timeout is 0
    //
    Lock sync(*this);
    
    _checkTask = new CheckTask(this);
    _timer->schedule(_checkTask, IceUtil::Time::seconds((_nodes.size() - _id) * 2));
    recovery();
}

void
NodeI::check()
{
    {
        Lock sync(*this);
        if(_destroy)
        {
            return;
        }
        assert(!_mergeTask);

        if(_state == NodeStateElection || _state == NodeStateReorganization || _coord != _id)
        {
            assert(_checkTask);
            _timer->schedule(_checkTask, _electionTimeout);
            return;
        }
        
        // Next get the set of nodes that were detected as unreachable
        // from the replica and remove them from our slave list.
        vector<int> dead;
        _observers->getReapedSlaves(dead);
        if(!dead.empty())
        {
            for(vector<int>::const_iterator p = dead.begin(); p != dead.end(); ++p)
            {
                set<GroupNodeInfo>::iterator q = _up.find(GroupNodeInfo(*p));
                if(q != _up.end())
                {
                    if(_traceLevels->election > 0)
                    {
                        Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
                        out << "node " << _id << ": reaping slave " << *p;
                    }
                    _up.erase(q);
                }
            }
                
            // If we no longer have the majority of the nodes under our
            // care then we need to stop our replica.
            if(_up.size() < _nodes.size()/2)
            {
                if(_traceLevels->election > 0)
                {
                    Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
                    out << "node " << _id << ": stopping replica";
                }
                // Clear _checkTask -- recovery() will reset the
                // timer.
                assert(_checkTask);
                _checkTask = 0;
                recovery();
                return;
            }
        }
    }

    // See if other groups exist for possible merge.
    set<int> tmpset;
    int max = -1;
    for(map<int, NodePrx>::const_iterator p = _nodes.begin(); p != _nodes.end(); ++p)
    {
        if(p->first == _id)
        {
            continue;
        }
        try
        {
            if(p->second->areYouCoordinator())
            {
                if(p->first > max)
                {
                    max = p->first;
                }
                tmpset.insert(p->first);
            }
        }
        catch(const Ice::Exception& ex)
        {
            if(_traceLevels->election > 0)
            {
                Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
                out << "node " << _id << ": call on node " << p->first << " failed: " << ex;
            }
        }
    }

    Lock sync(*this);

    // If the node state has changed while the mutex has been released
    // then bail. We don't schedule a re-check since we're either
    // destroyed in which case we're going to terminate or the end of
    // the election/reorg will re-schedule the check.
    if(_destroy || _state == NodeStateElection || _state == NodeStateReorganization || _coord != _id)
    {
        _checkTask = 0;
        return;
    }

    // If we didn't find any coordinators then we're done. Reschedule
    // the next check and terminate.
    if(tmpset.empty())
    {
        assert(_checkTask);
        _timer->schedule(_checkTask, _electionTimeout);
        return;
    }

    // _checkTask == 0 means that the check isn't scheduled.
    _checkTask = 0;

    if(_traceLevels->election > 0)
    {
        Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
        out << "node " << _id << ": highest priority node count: " << max;
    }

    IceUtil::Time delay = IceUtil::Time::seconds(0);
    if(_id < max)
    {
        // Reschedule timer proportial to p-i.
        delay = _mergeTimeout + _mergeTimeout * (max - _id);
        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": scheduling merge in " << delay.toDuration()
                << " seconds";
        }
    }

    assert(!_mergeTask);
    _mergeTask = new MergeTask(this, tmpset);
    _timer->schedule(_mergeTask, delay);
}

// Called if the node has not heard from the coordinator in some time.
void
NodeI::timeout()
{
    int myCoord;
    string myGroup;
    {
        Lock sync(*this);
        // If we're destroyed or we are our own coordinator then we're
        // done.
        if(_destroy || _coord == _id)
        {
            return;
        }
        myCoord = _coord;
        myGroup = _group;
    }

    bool failed = false;
    try
    {
        map<int, NodePrx>::const_iterator p = _nodes.find(myCoord);
        assert(p != _nodes.end());
        if(!p->second->areYouThere(myGroup, _id))
        {
            if(_traceLevels->election > 0)
            {
                Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
                out << "node " << _id << ": lost connection to coordinator " << myCoord
                    << ": areYouThere returned false";
            }
            failed = true;
        }
    }
    catch(const Ice::Exception& ex)
    {
        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": lost connection to coordinator " << myCoord << ": " << ex;
        }
        failed = true;
    }
    if(failed)
    {
        recovery();
    }
}

void
NodeI::merge(const set<int>& coordinatorSet)
{
    set<int> invited;
    string gp;
    {
        Lock sync(*this);
        _mergeTask = 0;

        // If the node is currently in an election, or reorganizing
        // then we're done.
        if(_state == NodeStateElection || _state == NodeStateReorganization)
        {
            return;
        }

        // This state change prevents this node from accepting
        // invitations while the merge is executing.
        setState(NodeStateElection);

        // No more replica changes are permitted.
        while(!_destroy && _updateCounter > 0)
        {
            wait();
        }
        if(_destroy)
        {
            return;
        }

        ostringstream os;
        os << _id << ":" << IceUtil::generateUUID();
        _group = os.str();
        gp = _group;

        _invitesAccepted.clear();
        _invitesIssued.clear();

        // Construct a set of node ids to invite. This is the union of
        // _up and set of coordinators gathered in the check stage.
        invited = coordinatorSet;
        for(set<GroupNodeInfo>::const_iterator p = _up.begin(); p != _up.end(); ++p)
        {
            invited.insert(p->id);
        }

        _coord = _id;
        _up.clear();

        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": inviting " << toString(invited) << " to group " << _group;
        }
    }

    set<int>::iterator p = invited.begin();
    while(p != invited.end())
    {
        try
        {
            if(_traceLevels->election > 0)
            {
                Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
                out << "node " << _id << ": inviting node " << *p << " to group " << gp;
            }
            map<int, NodePrx>::const_iterator node = _nodesOneway.find(*p);
            assert(node != _nodesOneway.end());
            node->second->invitation(_id, gp);
            ++p;
        }
        catch(const Ice::Exception&)
        {
            invited.erase(p++);
        }
    }

    // Now we wait for responses to our invitation.
    {
        Lock sync(*this);
        if(_destroy)
        {
            return;
        }

        // Add each of the invited nodes in the invites issed set.
        _invitesIssued.insert(invited.begin(), invited.end());

        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": invites pending: " << toString(_invitesIssued);
        }

        // Schedule the mergeContinueTask.
        assert(_mergeContinueTask == 0);
        _mergeContinueTask = new MergeContinueTask(this);
            
        // At this point we may have already accepted all of the
        // invitations, if so then we want to schedule the
        // mergeContinue immediately.
        IceUtil::Time timeout = _mergeTimeout;
        if(_up.size() == _nodes.size()-1 || _invitesIssued == _invitesAccepted)
        {
            timeout = IceUtil::Time::seconds(0);
        }
        _timer->schedule(_mergeContinueTask, timeout);
    }
}

void
NodeI::mergeContinue()
{
    string gp;
    set<GroupNodeInfo> tmpSet;
    {
        Lock sync(*this);
        if(_destroy)
        {
            return;
        }

        // Copy variables for thread safety.
        gp = _group;
        tmpSet = _up;

        assert(_mergeContinueTask);
        _mergeContinueTask = 0;

        // The node is now reorganizing.
        assert(_state == NodeStateElection);
        setState(NodeStateReorganization);

        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": coordinator for " << (tmpSet.size() +1) << " nodes (including myself)";
        }
        
        // Now we need to decide whether we can start serving content. If
        // we're on initial startup then we need all nodes to participate
        // in the election. If we're running a subsequent election then we
        // need a majority of the nodes to be active in order to start
        // running.
        unsigned int ingroup = static_cast<unsigned int>(tmpSet.size());
        if((_max != _nodes.size() && ingroup != _nodes.size() -1) || ingroup < _nodes.size()/2)
        {
            if(_traceLevels->election > 0)
            {
                Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
                out << "node " << _id << ": not enough nodes " << (ingroup+1) << "/" << _nodes.size()
                     << " for replication to commence";
                if(_max != _nodes.size())
                {
                    out << " (require full participation for startup)";
                }
            }
            recovery();
            return;
        }
    }

    // Find out who has the highest available set of database
    // updates.
    int maxid = -1;
    LogUpdate maxllu = { -1, 0 };
    for(set<GroupNodeInfo>::const_iterator p = tmpSet.begin(); p != tmpSet.end(); ++p)
    {
        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node id=" << p->id << " llu=" << p->llu.generation << "/" << p->llu.iteration;
        }
        if(p->llu.generation > maxllu.generation ||
           (p->llu.generation == maxllu.generation && p->llu.iteration > maxllu.iteration))
        {
            maxid = p->id;
            maxllu = p->llu;
        }
    }

    LogUpdate myLlu = _replica->getLastLogUpdate();
    if(_traceLevels->election > 0)
    {
        Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
        out << "node id=" << _id << " llu=" << myLlu.generation << "/" << myLlu.iteration;
    }

    // If its not us then we have to get the latest database data from
    // the replica with the latest set.
    //if(maxllu > _replica->getLastLogUpdate())
    if(maxllu > myLlu)
    {
        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": syncing database state with node " << maxid;
        }
        try
        {
            map<int, NodePrx>::const_iterator node = _nodes.find(maxid);
            assert(node != _nodes.end());
            _replica->sync(node->second->sync());
        }
        catch(const Ice::Exception& ex)
        {
            if(_traceLevels->election > 0)
            {
                Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
                out << "node " << _id << ": syncing database state with node "
                     << maxid << " failed: " << ex;
            }
            recovery();
            return;
        }
    }
    else
    {
        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": I have the latest database state.";
        }
    }

    // At this point we've ensured that we have the latest database
    // state, as such we can set our _max flag.
    unsigned int max = static_cast<unsigned int>(tmpSet.size()) + 1;
    {
        Lock sync(*this);
        if(max > _max)
        {
            _max = max;
        }
        max = _max;
    }

    // Prepare the LogUpdate for this generation.
    maxllu.generation++;
    maxllu.iteration = 0;

    try
    {
        // Tell the replica that it is now the master with the given
        // set of slaves and llu generation.
        _replica->initMaster(tmpSet, maxllu);
    }
    catch(const Ice::Exception& ex)
    {
        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": initMaster failed: " << ex;
        }
        recovery();
        return;
    }

    // Tell each node to go.
    for(set<GroupNodeInfo>::const_iterator p = tmpSet.begin(); p != tmpSet.end(); ++p)
    {
        try
        {
            map<int, NodePrx>::const_iterator node = _nodes.find(p->id);
            assert(node != _nodes.end());
            node->second->ready(_id, gp, _replicaProxy, max, maxllu.generation);
        }
        catch(const Ice::Exception& ex)
        {
            if(_traceLevels->election > 0)
            {
                Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
                out << "node " << _id << ": error calling ready on " << p->id << " ex: " << ex;
            }
            recovery();
            return;
        }
    }

    {
        Lock sync(*this);
        if(_destroy)
        {
            return;
        }
        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": reporting for duty in group " << _group << " as coordinator. ";
            out << "replication commencing with " << _up.size()+1 << "/" << _nodes.size()
                << " nodes with llu generation: " << maxllu.generation;
        }
        setState(NodeStateNormal);
        _coordinatorProxy = 0;

        _generation = maxllu.generation;

        assert(!_checkTask);
        _checkTask = new CheckTask(this);
        _timer->schedule(_checkTask, _electionTimeout);
    }
}

void
NodeI::invitation(int j, const string& gn, const Ice::Current&)
{
    if(_traceLevels->election > 0)
    {
        Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
        out << "node " << _id << ": invitation from " << j << " to group " << gn;
    }

    // Verify that j exists in our node set.
    if(_nodes.find(j) == _nodes.end())
    {
        Ice::Warning warn(_traceLevels->logger);
        warn << _traceLevels->electionCat << ": ignoring invitation from unknown node " << j;
        return;
    }

    int tmpCoord = -1;
    int max = -1;
    set<GroupNodeInfo> tmpSet;
    {
        Lock sync(*this);
        if(_destroy)
        {
            return;
        }
        // If we're in the election or reorg state a merge has already
        // started, so ignore the invitation.
        if(_state == NodeStateElection || _state == NodeStateReorganization)
        {
            if(_traceLevels->election > 0)
            {
                Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
                out << "node " << _id << ": invitation ignored";
            }
            return;
        }

        //
        // Upon receipt of an invitation we cancel any pending merge
        // task.
        //
        if(_mergeTask)
        {
            // If the timer doesn't cancel it means that the timer has
            // fired and the merge is currently in-progress in which
            // case we should reject the invitation.
            if(!_timer->cancel(_mergeTask))
            {
                // The merge task is cleared in the merge. This
                // ensures two invitations cannot cause a race with
                // the merge.
                //_mergeTask = 0;
                return;
            }
            _mergeTask = 0;
        }

        // We're now joining with another group. If we are active we
        // must stop serving as a master or slave.
        setState(NodeStateElection);
        while(!_destroy && _updateCounter > 0)
        {
            wait();
        }
        if(_destroy)
        {
            return;
        }

        tmpCoord = _coord;
        tmpSet = _up;

        _coord = j;
        _group = gn;
        max = _max;
    }

    Ice::IntSeq forwardedInvites;
    if(tmpCoord == _id) // Forward invitation to my old members.
    {
        for(set<GroupNodeInfo>::const_iterator p = tmpSet.begin(); p != tmpSet.end(); ++p)
        {
            try
            {
                map<int, NodePrx>::const_iterator node = _nodesOneway.find(p->id);
                assert(node != _nodesOneway.end());
                node->second->invitation(j, gn);
                forwardedInvites.push_back(p->id);
            }
            catch(const Ice::Exception&)
            {
            }
        }
    }

    // Set the state and timer before calling accept. This ensures
    // that if ready is called directly after accept is called then
    // everything is fine. Setting the state *after* calling accept
    // can cause a race.
    {
        Lock sync(*this);
        if(_destroy)
        {
            return;
        }
        assert(_state == NodeStateElection);
        setState(NodeStateReorganization);
        if(!_timeoutTask)
        {
            _timeoutTask = new TimeoutTask(this);
            _timer->scheduleRepeated(_timeoutTask, _masterTimeout);
        }
    }

    try
    {
        map<int, NodePrx>::const_iterator node = _nodesOneway.find(j);
        assert(node != _nodesOneway.end());
        node->second->accept(_id, gn, forwardedInvites, _replica->getObserver(), _replica->getLastLogUpdate(), max);
    }
    catch(const Ice::Exception&)
    {
        recovery();
        return;
    }
}

void
NodeI::ready(int j, const string& gn, const Ice::ObjectPrx& coordinator, int max, Ice::Long generation,
             const Ice::Current&)
{
    Lock sync(*this);
    if(!_destroy && _state == NodeStateReorganization && _group == gn)
    {
        // The coordinator must be j (this was set in the invitation).
        if(_coord != j)
        {
            Ice::Warning warn(_traceLevels->logger);
            warn << _traceLevels->electionCat << ": ignoring ready call from replica node " << j
                 << " (real coordinator is " << _coord << ")";
            return;
        }

        // Here we've already validated j in the invite call
        // (otherwise _group != gn).
        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": reporting for duty in group " << gn << " with coordinator " << j;
        }

        if(static_cast<unsigned int>(max) > _max)
        {
            _max = max;
        }
        _generation = generation;

        // Activate the replica here since the replica is now ready
        // for duty.
        setState(NodeStateNormal);
        _coordinatorProxy = coordinator;

        if(!_checkTask)
        {
            _checkTask = new CheckTask(this);
            _timer->schedule(_checkTask, _electionTimeout);
        }
    }
}

void
NodeI::accept(int j, const string& gn, const Ice::IntSeq& forwardedInvites, const Ice::ObjectPrx& observer,
              const LogUpdate& llu, int max, const Ice::Current&)
{
    // Verify that j exists in our node set.
    if(_nodes.find(j) == _nodes.end())
    {
        Ice::Warning warn(_traceLevels->logger);
        warn << _traceLevels->electionCat << ": ignoring accept from unknown node " << j;
        return;
    }

    Lock sync(*this);
    if(!_destroy && _state == NodeStateElection && _group == gn && _coord == _id)
    {
        _up.insert(GroupNodeInfo(j, llu, observer));

        if(static_cast<unsigned int>(max) > _max)
        {
            _max = max;
        }

        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": accept " << j << " forward invites (";
            for(Ice::IntSeq::const_iterator p = forwardedInvites.begin(); p != forwardedInvites.end(); ++p)
            {
                if(p != forwardedInvites.begin())
                {
                    out << ",";
                }
                out << *p;
            }
            out << ") with llu "
                << llu.generation << "/" << llu.iteration << " into group " << gn
                << " group size " << (_up.size() + 1);
        }

        // Add each of the forwarded invites to the list of issued
        // invitations. This doesn't use set_union since
        // forwardedInvites may not be sorted.
        _invitesIssued.insert(forwardedInvites.begin(), forwardedInvites.end());
        // We've accepted the invitation from node j.
        _invitesAccepted.insert(j);

        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": invites pending: " << toString(_invitesIssued)
                << " invites accepted: " << toString(_invitesAccepted);
        }

        // If invitations have been accepted from all nodes and the
        // merge task has already been scheduled then reschedule the
        // merge continue immediately. Otherwise, we let the existing
        // merge() schedule continue.
        if((_up.size() == _nodes.size()-1 || _invitesIssued == _invitesAccepted) &&
           _mergeContinueTask && _timer->cancel(_mergeContinueTask))
        {
            _timer->schedule(_mergeContinueTask, IceUtil::Time::seconds(0));
        }
    }
}

bool
NodeI::areYouCoordinator(const Ice::Current&) const
{
    Lock sync(*this);
    return _state != NodeStateElection && _state != NodeStateReorganization && _coord == _id;
}

bool
NodeI::areYouThere(const string& gn, int j, const Ice::Current&) const
{
    Lock sync(*this);
    return _group == gn && _coord == _id && _up.find(GroupNodeInfo(j)) != _up.end();
}

Ice::ObjectPrx
NodeI::sync(const Ice::Current&) const
{
    return _replica->getSync();
}

NodeInfoSeq
NodeI::nodes(const Ice::Current&) const
{
    NodeInfoSeq seq;
    for(map<int, NodePrx>::const_iterator q = _nodes.begin(); q != _nodes.end(); ++q)
    {
        NodeInfo ni;
        ni.id = q->first;
        ni.n = q->second;
        seq.push_back(ni);
    }

    return seq;
}

QueryInfo
NodeI::query(const Ice::Current&) const
{
    Lock sync(*this);
    QueryInfo info;
    info.id = _id;
    info.coord = _coord;
    info.group = _group;
    info.replica = _replicaProxy;
    info.state = _state;
    info.max = _max;

    for(set<GroupNodeInfo>::const_iterator p = _up.begin(); p != _up.end(); ++p)
    {
        GroupInfo gi;
        gi.id = p->id;
        gi.llu = p->llu;
        info.up.push_back(gi);
    }

    return info;
}

void
NodeI::recovery(Ice::Long generation)
{
    Lock sync(*this);

    // Ignore the recovery if the node has already advanced a
    // generation.
    if(generation != -1 && generation != _generation)
    {
        return;
    }

    setState(NodeStateInactive);
    while(!_destroy && _updateCounter > 0)
    {
        wait();
    }
    if(_destroy)
    {
        return;
    }

    ostringstream os;
    os << _id << ":" << IceUtil::generateUUID();
    _group = os.str();

    _generation = -1;
    _coord = _id;
    _up.clear();
    
    if(_traceLevels->election > 0)
    {
        Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
        out << "node " << _id << ": creating new self-coordinated group " << _group;
    }

    // Reset the timer states.
    if(_mergeTask)
    {
        _timer->cancel(_mergeTask);
        _mergeTask = 0;
    }
    if(_timeoutTask)
    {
        _timer->cancel(_timeoutTask);
        _timeoutTask = 0;
    }
    if(!_checkTask)
    {
        _checkTask = new CheckTask(this);
        _timer->schedule(_checkTask, _electionTimeout);
    }
}

void
NodeI::destroy()
{
    Lock sync(*this);
    assert(!_destroy);

    while(_updateCounter > 0)
    {
        wait();
    }
    _destroy = true;
    notifyAll();

    // Cancel the timers.
    if(_checkTask)
    {
        _timer->cancel(_checkTask);
        _checkTask = 0;
    }

    if(_timeoutTask)
    {
        _timer->cancel(_timeoutTask);
        _timeoutTask = 0;
    }

    if(_mergeTask)
    {
        _timer->cancel(_mergeTask);
        _mergeTask = 0;
    }
}

// A node should only receive an observer init call if the node is
// reorganizing and its not the coordinator.
void
NodeI::checkObserverInit(Ice::Long /*generation*/)
{
    Lock sync(*this);
    if(_state != NodeStateReorganization)
    {
        throw ObserverInconsistencyException("init cannot block when state != NodeStateReorganization");
    }
    if(_coord == _id)
    {
        throw ObserverInconsistencyException("init called on coordinator");
    }
}

// Notify the node that we're about to start an update.
Ice::ObjectPrx
NodeI::startUpdate(Ice::Long& generation, const char* file, int line)
{
    bool majority = _observers->check();

    Lock sync(*this);
    
    // If we've actively replicating & lost the majority of our replicas then recover.
    if(!_coordinatorProxy && !_destroy && _state == NodeStateNormal && !majority)
    {
        recovery();
    }

    while(!_destroy && _state != NodeStateNormal)
    {
        wait();
    }
    if(_destroy)
    {
        throw Ice::UnknownException(file, line);
    }
    if(!_coordinatorProxy)
    {
        ++_updateCounter;
    }
    generation = _generation;
    return _coordinatorProxy;
}

bool
NodeI::updateMaster(const char* /*file*/, int /*line*/)
{
    bool majority = _observers->check();

    Lock sync(*this);

    // If the node is destroyed, or is not a coordinator then we're
    // done.
    if(_destroy || _coordinatorProxy)
    {
        return false;
    }
    
    // If we've lost the majority of our replicas then recover.
    if(_state == NodeStateNormal && !majority)
    {
        recovery();
    }

    // If we're not replicating then we're done.
    if(_state != NodeStateNormal)
    {
        return false;
    }

    // Otherwise adjust the update counter, and return success.
    ++_updateCounter;
    return true;
}

Ice::ObjectPrx
NodeI::startCachedRead(Ice::Long& generation, const char* file, int line)
{
    Lock sync(*this);
    while(!_destroy && _state != NodeStateNormal)
    {
        wait();
    }
    if(_destroy)
    {
        throw Ice::UnknownException(file, line);
    }
    generation = _generation;
    ++_updateCounter;
    return _coordinatorProxy;
}

void
NodeI::startObserverUpdate(Ice::Long generation, const char* file, int line)
{
    Lock sync(*this);
    if(_destroy)
    {
        throw Ice::UnknownException(file, line);
    }
    if(_state != NodeStateNormal)
    {
        throw ObserverInconsistencyException("update called on inactive node");
    }
    if(!_coordinatorProxy)
    {
        throw ObserverInconsistencyException("update called on the master");
    }
    if(generation != _generation)
    {
        throw ObserverInconsistencyException("invalid generation");
    }
    ++_updateCounter;
}

void
NodeI::finishUpdate()
{
    Lock sync(*this);
    assert(!_destroy);
    --_updateCounter;
    assert(_updateCounter >= 0);
    if(_updateCounter == 0)
    {
        notifyAll();
    }
}

namespace
{
static string
stateToString(NodeState s)
{
    switch(s)
    {
    case NodeStateInactive:
        return "inactive";
    case NodeStateElection:
        return "election";
    case NodeStateReorganization:
        return "reorganization";
    case NodeStateNormal:
        return "normal";
    }
    return "unknown";
}
}

void
NodeI::setState(NodeState s)
{
    if(s != _state)
    {
        if(_traceLevels->election > 0)
        {
            Ice::Trace out(_traceLevels->logger, _traceLevels->electionCat);
            out << "node " << _id << ": transition from " << stateToString(_state) << " to "
                << stateToString(s);
        }
        _state = s;
        if(_state == NodeStateNormal)
        {
            notifyAll();
        }
    }
}