/*
    * Copyright (c) 1998-2002 The Jgroup Team.
    *
    * This program is free software; you can redistribute it and/or modify
    * it under the terms of the GNU Lesser General Public License version 2 as
    * published by the Free Software Foundation.
    *
    * This program is distributed in the hope that it will be useful,
    * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   * GNU Lesser General Public License for more details.
   *
   * You should have received a copy of the GNU Lesser General Public License
   * along with this program; if not, write to the Free Software
   * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
   *
   */
  
  package jgroup.relacs.daemon;
  
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  
  import jgroup.core.EndPoint;
  import jgroup.core.MemberId;
  import jgroup.core.View;
  import jgroup.core.multicast.ChainIdentifier;
  import jgroup.relacs.events.DeliveryAck;
  import jgroup.relacs.events.InstallAck;
  import jgroup.relacs.events.JoinRequest;
  import jgroup.relacs.events.LeaveRequest;
  import jgroup.relacs.events.MemberLeftEvent;
  import jgroup.relacs.events.MulticastRequest;
  import jgroup.relacs.events.PrepareAck;
  import jgroup.relacs.events.PrepareEvent;
  import jgroup.relacs.mss.HostTable;
  import jgroup.relacs.mss.Mss;
  import jgroup.relacs.mss.MssConstants;
  import jgroup.relacs.mss.MssHost;
  import jgroup.relacs.mss.MssUser;
  import jgroup.relacs.types.LocalId;
  import jgroup.relacs.types.MemberIdImpl;
  import jgroup.relacs.types.ViewId;
  import jgroup.relacs.types.ViewImpl;
  import jgroup.util.Queue;
  import jgroup.util.Util;
  
  import org.apache.log4j.Logger;
  import org.apache.log4j.MDC;
  
  
  /**
   * The <code>Group</code> class
   *
   * @author  Alberto Montresor
   * @since   Jgroup 0.1
   */
  final class Group
    implements Tag, MssConstants
  {
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Logger
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Obtain logger for this class */
    public static final Logger log = Logger.getLogger(Group.class);
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Fields
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    // References to other data structures of this layer
    
    /** Group table */
    private final GroupTable grouptable;
    
    /** Table containing the hosts handled by this daemon */
    private final HostTable hosttable;
  
    /** Mss used to send messages */
    private final Mss mss;
    
    // General fields
    
    /** Group identifier */
    private int gid;
    
    /** Group status (idle, synch, estimate) */
    private int status;
    
   /** Current reachable set */
   private EndPoint[] rset;
   
   /** Information related to this host */
   private final HostData me; 
   
   /** My identifier (32 bit IP address + port number) */
   private final EndPoint myaddress;
   
   /** Number of views installed so far */
   private int ninstalled;
   
   /** Number of views coordinated so far */
   private int ncoordinated;
   
   /** Number of suspect events */
   private int nsuspects;
 
   /** Complete list of hosts interested in the group */
   private final Map<EndPoint,HostData> thosts;
   
   /** Complete list of local members interested in the group */
   private final Map<LocalId,MemberData> tmembers;
 
   /** Number of members contained in the system, excluding the leaving members */
   private int nmembers;
   
   
   // Current view information
   
   /** Complete view identifier */
   private long cvid;
   
   /** View identifier */
   private long pvid;
   
   /** View composition */
   private ViewDescription viewDesc;
 
   
   // Agreement phase data
   
   /** Messages whose sending is postponed */
   private Queue delayedSent; 
   
   /** Messages whose delivery is postponed */
   private Queue delayedRcvd;
   
   /** Hosts in the current estimate */
   private Estimate estimate;
   
   /** Current proposal */
   private MsgProp proposal;
   
   /** True if message buffer is consistent */
   private boolean msgsStable;
   
   /** 
    * True if the installation of the view has been completed 
    * by every local member included in the view. When false, 
    * the local host cannot participate in the agreement protocol
    * by sending proposal, as the current view has not been
    * installed by one of the local members. Once true, it will
    * stay true until the next view installation.
    */
   private boolean installationComplete;
 
   /** Send buffer */
   private final SendBuffer sbuffer;
 
   
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Constructors
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Creates a group identified by the group identifier <code> gid</code>.
    *
    *  @param gid             group identifier
    *  @param vid             inital view identifier (both complete and partial)
    *  @param hd              this host additional info
    *  @param md              joining member additional info
    */
   Group(GroupTable grouptable, Mss mss, int gid, View view, HostData hd, MemberData md)
   {
     // References to other data structures
     this.grouptable = grouptable;
     this.mss = mss;
     this.hosttable = mss.getHostTable();
 
     // General fields
     this.gid = gid;
     // Initial status of group
     this.status = S_IDLE;
     MDC.put("group", toString());
     myaddress = hd.getEndPoint();
     rset = new EndPoint[] { myaddress };
     me = hd;
     ncoordinated = 0;
     ninstalled = 0;
     nsuspects = 0;
     thosts = new HashMap<EndPoint,HostData>();
     tmembers = new HashMap<LocalId,MemberData>();
     thosts.put(myaddress, hd);
     tmembers.put(md.getMemberId().getLocalId(), md);
     nmembers = 1;
 
     // Initialization of the first view
     cvid = view.getVid();
     pvid = cvid;
     viewDesc = new ViewDescription(hd, md);
     sbuffer = new SendBuffer();
     sbuffer.setView(view, 1);
 
     // Initialization of agreement phase info
     delayedSent = new Queue();
     delayedRcvd = new Queue();
     proposal    = new MsgProp();
     msgsStable  = true;
     installationComplete  = false;
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Group management
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Return the identifier of this group. 
    */
   int getGid()
   {
     return gid;
   }
   
   
   /**
    *  Returns a collection containing the hosts of the group
    */
   Collection<HostData> getHosts()
   {
     return thosts.values();
   }
   
   /**
    *  Returns a collection containing the local members of the group
    */
   Collection<MemberData> getMembers()
   {
     return tmembers.values();
   }
 
   /**
    *  Returns true if this group contains no local members;
    *  otherwise false is returned.
    */
   boolean isEmpty()
   {
     return tmembers.isEmpty();
   }
 
   /**
    * Returns a string representation of this Group object.
    */
   public String toString()
   {
     return toString(false);
   }
 
   /**
    * Returns a string representation of this Group object.
    */
   public String toString(boolean full)
   {
     StringBuilder buf = new StringBuilder("[Group: ");
     buf.append(gid);
     buf.append(", ");
     switch (status) {
       case S_IDLE :
         buf.append("IDLE");
         break;
       case S_SYNCH :
         buf.append("SYNCH");
         break;
       case S_ESTIMATE :
         buf.append("ESTIMATE");
         break;
       default :
         buf.append("INVALID STATUS");
         break;
     }
     if (full) {
       buf.append(", ");
       buf.append(estimate);
     }
     buf.append("]");
     return buf.toString();
   }
 
   /**
    * 
    */
   HostData getHost(EndPoint endpoint)
   {
     return (HostData) thosts.get(endpoint);
   }
   
   /**
    *  Add the host <CODE>src</CODE> to the group. 
    *  
    *  @param src the host to be added
    *  @param hosts the hosts known to src, that we need to add as well
    *  @return true if <CODE>src</CODE> was added to the group
    */
   boolean join(MssHost src, MssHost[] hosts)
   {
     if (log.isDebugEnabled())
       log.debug("Group.join: START");
     
     EndPoint endpoint = src.getEndPoint();
     HostData sender = (HostData) thosts.get(endpoint);
     if (sender == null) {
       if (log.isDebugEnabled())
         log.debug("Group.join: new host");
 
       /* Add the host in src and the hosts in hosts to the group */   
       thosts.put(endpoint, new HostData(src));
       grouptable.addEndpointToGroup(endpoint, this);
       for (int i=0; i < hosts.length; i++) {
         HostData scan = (HostData) thosts.get(hosts[i].getEndPoint());
         if (scan == null) {
           thosts.put(hosts[i].getEndPoint(), new HostData(hosts[i]));
           grouptable.addEndpointToGroup(hosts[i].getEndPoint(), this);
         }
       }
 
       /* We have added src to the group */
       return true;
       
     } else if (sender.isLeaving()) {
       
       /* The sender is not leaving any more */
       sender.setLeaving(false);
       if (status == S_IDLE && Util.in(rset, endpoint))
         synchronizationPhaseInit();
     }
     
     /* We have not added anything */
     return false;
   }
 
   
   /**
    *  Remove the host described in src from the group.
    */
   void leave(EndPoint src)
   {
     HostData sender = thosts.get(src);
     if (sender != null) {
       sender.setLeaving(true);
       switch (status) {
         /* IDLE PHASE */
       case S_IDLE:
         synchronizationPhaseInit();
         if (log.isDebugEnabled())
           log.debug("Leaving in S_IDLE: " );
         break;
         /* SYNCHRONIZATION PHASE */
       case S_SYNCH:
         estimate.remove(src);
         if (log.isDebugEnabled())
           log.debug("Leaving in S_SYNCH: EST= " + estimate);
         checkSynchronization();
         break;
         /* ESTIMATE EXCHANGE PHASE */
       case S_ESTIMATE:
         estimate.remove(src);
         if (log.isDebugEnabled())
           log.debug("Leaving in S_ESTIMATE: EST= " + estimate);
         sendEstimate();
         break;
       }
     }
   }
 
   /**
    *  Handles the crash of the local member identified by the given
    *  local identifier. This method is invoked by <CODE>Daemon</CODE>
    *  on all groups included in the group table. It is up to the
    *  <CODE>Group</CODE> objects to react to the crash.
    *  
    *  @param id the identifier of the crashed member
    */
   void handleLocalSuspect(MemberId id)
   {
     MemberData md = (MemberData) tmembers.get(id.getLocalId());
     if (md != null) {
       /* The member belongs to this group */
       if (log.isDebugEnabled())
         log.debug("Group:handleLocalSuspect: " + id);
       if (!md.isLeaving() && !md.isCrashed()) {
         md.setCrashed();
         nmembers--;
         runAgreementProtocol();
       }
     }
   }
   
   /**
    *  Invoked by activateGroups
    */
   void handleRemoteSuspect(int nsuspects, EndPoint[] trset, EndPoint[] nrset, EndPoint[] nuset)
   {
     if (nsuspects > this.nsuspects) {
       if (log.isDebugEnabled())
         log.debug("nsuspects=" + nsuspects);
       this.nsuspects = nsuspects;
 
       /*
        * Construct symmetrical reachability information and send it to
        * the newly reachable members (nrset) that were previously
        * unreachable.
        */
       try {
         MsgSymm tosend = MsgSymm.marshal(gid, me.getVersion(),
           HostData.extract(thosts, nrset), rset, nrset.length);
         mss.msend(R_SYMM, tosend, nrset);
       } catch (IOException e) {
         log.error("Unable to marshal symmetrical reachability message", e);
         return;
       }
 
       /* Set the reachability set */
       rset = trset;
 
       switch (status) {
     
       case S_IDLE:
         synchronizationPhaseInit();
         break;
     
       case S_SYNCH:
         estimate.remove(nuset);
         if (log.isDebugEnabled())
           log.debug(estimate);
         checkSynchronization();
         break;
     
       case S_ESTIMATE:
         estimate.remove(nuset);
         if (log.isDebugEnabled())
           log.debug(estimate);
         sendEstimate();
         break;
       }
     }
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Interactions with members
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   void handleLocalDlvrAck(DeliveryAck ack)
   {
     if (log.isDebugEnabled()) {
       log.debug("host.pos=" + ack.getHostIndex() + ", mem.pos=" + ack.getMemberIndex());
     }
     LocalId ackSender = ack.getAckSender().getLocalId();
     MemberData md = (MemberData) tmembers.get(ackSender);
     md.removeAck(ack.getMessageId());
 
     log.assertLog(viewDesc.size() > ack.getHostIndex(),
       "Host index of DeliveryAck is greater than the current view size ("
       + viewDesc.size() + "): " + ack + "; msgStable: " + msgsStable);
 
     /*
      *  Send the RESULT message (ack + info) to all hosts; in the
      *  previous versions of the protocol, result info were sent using
      *  the mss (by msending the message also to the sender).  The new
      *  version takes care of locally delivering the message.
      */
     MsgResult result = new MsgResult(gid, pvid, ack.getMessageId(),
       ack.getSender(), viewDesc.getDelivered(ack.getHostIndex()), 
       ack.getHostIndex(), ack.getViewIndex(), ack.getResult());
     try {
       EndPoint[] receivers = viewDesc.getEndPoints();
       MsgResult tosend = result.marshal(receivers.length);
       sbuffer.insertMsgResult(result);
       if (log.isDebugEnabled())
         log.debug("Sending R_RESULT: " + tosend);
       mss.msend(R_RESULT, tosend, receivers);
     } catch (IOException e) {
       log.error("Unable to marshal result message", e);
       return;
     }
 
     boolean stable = viewDesc.localDeliveryAck(ack);
     if (log.isDebugEnabled()) {
       log.debug("Handling deliveryACK (stable=" + stable + ")");
       log.debug(viewDesc);
     }
     if (stable) {
       msgsStable = true;
       if (status == S_ESTIMATE) {
         try {
           int[] delivered = viewDesc.getDelivered(estimate.getHosts());
           proposal.encodeMsgsProp(delivered);
           sendProposal();
         } catch (IOException e) {
           log.error("Unable to encode messages proposal", e);
           return;
         }
       }
     }
   }
 
 
   /**
    *  Handles a multicast request generated by a member.
    * 
    *  @param request the multicast request
    */
   void handleMulticastRequest(MulticastRequest request)
   {
     if (log.isDebugEnabled())
       log.debug(request);
 
     // Message to send
     MsgMcast msg = null;
     try {
       msg = new MsgMcast(request);
     } catch (IOException e) {
       log.error("Cannot marshal multicast message", e);
       return;
     }
 
     if (status == S_IDLE) {
       acceptMessage(msg, false);
     } else if (!installationComplete) {
       if (log.isDebugEnabled())
         log.debug("View installation for local members not complete");
       acceptMessage(msg, true);
     } else {
       if (log.isDebugEnabled())
         log.debug(viewDesc);
       ChainIdentifier chainId = request.getChainId();
       if (chainId != null) {
         EndPoint orgSender = chainId.getOriginalSender().getEndPoint();
         HostData hd = (HostData) thosts.get(orgSender);
         assert hd != null : "Unknown chain sender";
         assert hd.hasValidViewIndex() : "Uninitialized chain sender";
         if (log.isDebugEnabled())
           log.debug("Received chained message: " + chainId);
         if (!viewDesc.isNew(hd.getViewIndex(), chainId.getMsgId())) {
           if (log.isDebugEnabled())
             log.debug("Accepting chained message: " + chainId);
           /*
            * Since this multicast is chained with another multicast
            * being processed by the Dispatcher thread, and since we
            * are not in the IDLE state, this message should be
            * considered 'late' and thus R_FORWARD will be used to
            * pass this chained message.  This forwarded message will
            * be processed in the handleMsgDlvrd() method at the receiver
            * side.
            */
           acceptMessage(msg, true);
         } else {
           assert false : "Chain message cannot be new";
         }
       } else {
         delayedSent.insert(msg);
         if (log.isDebugEnabled())
           log.debug("Message (" + msg.mid + ") put in the delayedSent queue: " + delayedSent);
       }
     }
   }
 
 
   void handleLocalPrepareAck(PrepareAck event)
   {
     if (log.isDebugEnabled())
       log.debug(event);
     // Get sender of the PrepareAck event
     MemberData md = (MemberData) tmembers.get(event.getMemberId().getLocalId());
     if (!md.isPrepared() && !md.isCrashed()) {
       md.setPrepared();
       if (log.isDebugEnabled())
         log.debug(md + " is prepared");
     }
   }
 
 
   void handleLocalViewAck(InstallAck upcall)
   {
     installationComplete = viewDesc.localViewAck(upcall);
     if (log.isDebugEnabled()) {
       log.debug("Local view is " + (installationComplete ? "stable" : "not stable"));
     }
     if (status == S_ESTIMATE)
       sendProposal();
   }
 
 
   void handleLocalJoin(JoinRequest event, MssUser mssuser)
   {
     if (log.isDebugEnabled())                    
       log.debug(event);
 
     MemberId id = event.getMemberId();
     MemberData md = new MemberData(id, event.getDispatcher());
     md.schedulePingTimer(mssuser);
     if (tmembers.put(id.getLocalId(), md) == null) {
       nmembers++;
       runAgreementProtocol();
     }
   }
 
 
   /**
    *  This method handle a leave request coming from a member.
    */
   void handleLeaveRequest(LeaveRequest event)
   {
     if (log.isDebugEnabled())
       log.debug(event);
 
     // Sender member and additional data
     MemberId id = event.getMemberId();
     MemberData md = (MemberData) tmembers.get(id.getLocalId());
     md.cancelPingTimer();
     /* 
      *  Note: I don't expect that a crashed member can send a leave
      *  request ;-) But members request and crash notifications may
      *  end up in the Mss queue in unpredictable order, so it is
      *  safer to perform the check in this way
      */
     if (!md.isLeaving() && !md.isCrashed()) {
       md.setLeaving();
       if (log.isDebugEnabled())
         log.debug(md + " is leaving");
       nmembers--;
       runAgreementProtocol();
     }
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Interactions with remote hosts (group-related messages)
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Handles a mcast invocation received from a remote host; if the host
    *  is in Estimate Exchange Phase, the message is buffered; otherwise,
    *  if the view of the message corresponds to the current one, the
    *  message is locally delivered and an ack message is m-sent.
    *
    *  @param     msg     the m-received message.
    *  @param     src     the sender of the message.
    */
   void handleMsgMcast(MsgMcast msg, EndPoint src)
   {
     if (log.isDebugEnabled()) {
       log.debug("Handle mcast msg (" + msg.mid + ") received from " + src);
     }
 
     if (status == S_ESTIMATE) {
       delayedRcvd.insert(msg);
       if (log.isDebugEnabled()) {
         log.debug("Message buffered in receive queue: " + delayedRcvd);
         log.debug(viewDesc);
       }
     } else if (pvid == msg.vid) {
       boolean newCrashed = viewDesc.notifyMessage(gid, msg);
       if (log.isDebugEnabled())
         log.debug("Multicast message being delivered (stable=false)");
       msgsStable = false;
       /*
        * If the current view size is equal to one, it contains just the
        * local host; there is no need to send this message to it
        */
       if (viewDesc.size() > 1) {
         try {
           MsgAck tosend = MsgAck.marshal(gid, pvid, viewDesc.getLocalIndex(),
             viewDesc.getAcks(), viewDesc.size());
           mss.msend(R_ACK, tosend, viewDesc.getEndPoints());
         } catch (IOException e) {
           log.error("Cannot marshal ACK message", e);
         }
       }
       /*
        *  If the message has not been delivered because of a new
        *  crashed hosts, start a new agreement protocol (if needed),
        *  or complete the member list and send a proposal, otherwise.
        */
       if (newCrashed) {
         if (log.isDebugEnabled()) {
           log.debug("Detected new crashed host; running agreement protocol");
         }
         runAgreementProtocol();
       }
     } else {
       log.warn("Received a multicast message with incorrect view id: "
         + msg.vid + " (" + pvid + ")");
     }
   }
 
   /**
    *
    *
    *  @param     msg     the m-received message.
    *  @param     src     the sender of the message.
    */
   void handleMsgAck(MsgAck msg, EndPoint src)
   {
     if (log.isDebugEnabled())
       log.debug("handleMsgAck: " + msg+ " from src = " + src);
 
     /* Set the ack */
     if (pvid == msg.vid) {
       viewDesc.remoteMessageAck(msg);
     } else {
       log.warn("Received an ack message with incorrect view id: " + msg.vid + " (" + pvid + ")");
     }
   }
 
 
   /**
    *  First of all, the method checks whether (i) the group is known;
    *  (ii) the view associated to the message is equal to the current
    *  one; (iii) the messge is unknown; (iv) the forwarding sender is
    *  still in the estimate. If so, the message is delivered and an ack
    *  message is sent. The message is forwarded again to the remaining
    *  members.
    *
    *  @param     msg     the m-received message.
    *  @param     src     the sender of the message.
    */
   void handleMsgDlvrd(MsgMcast msg, EndPoint src)
   {
     if (log.isDebugEnabled())
       log.debug("handleMsgDlvrd: " + msg + " from src = " + src);
 
     if ( pvid == msg.vid && estimate.contains(src) && viewDesc.isNew(msg.hpos, msg.mid) ) {
       boolean newCrashed = viewDesc.notifyMessage(gid, msg);
       if (log.isDebugEnabled())
         log.debug("Delivered multicast message (stable=false)");
       msgsStable = false;
 
       /*
        * If the current view size is equal to one, it contains just the
        * local host; there is no need to send this message to it
        */
       if (viewDesc.size() > 1) {
         /* Sends an ack message */
         try {
           MsgAck tosend = MsgAck.marshal(gid, pvid, viewDesc.getLocalIndex(),
               viewDesc.getAcks(), viewDesc.size());
           mss.msend(R_ACK, tosend, viewDesc.getEndPoints());
         } catch (IOException e) {
           log.error("Cannot marshal ACK message", e);
         }
       }
       
       /*  
        *  During ESTIMATE EXCHANGE phase, forwards the message. If the 
        *  current estimate size is equal to one, it contains just the
        *  local host; there is no need to send this message to it
        */
       if (status == S_ESTIMATE && estimate.size() > 1)
         mss.msend(R_FORWARD, msg, estimate.getEndPoints());
       
       /*
        *  If the message has not been delivered because of a new
        *  crashed hosts, start a new agreement protocol (if needed),
        *  or complete the member list and send a proposal, otherwise.
        */
       if (newCrashed) {
         if (log.isDebugEnabled()) {
           log.debug("Detected new crashed host; running agreement protocol");
         }
         runAgreementProtocol();
       }
     }
   }
 
   /**
    *
    *  @param     msg     the m-received message.
    *  @param     src     the sender of the message.
    */
   void handleMsgSynch(MsgSynch msg, EndPoint src)
   {
     HostData sender = thosts.get(src);
     if (sender != null) {
       if (log.isDebugEnabled()) {
         log.debug(msg);
         log.debug("  Sender: " + sender);
         log.debug("Receiver: " + me);
       }
 
       switch(status) {
 
       /*  IDLE PHASE */
       case S_IDLE:
         if (log.isDebugEnabled())
           log.debug("Group: Idle phase");
         if (sender.getVersion() < msg.vsend) {
           sender.setVersion(msg.vsend);
           if (Util.in(rset, src))
             synchronizationPhaseInit();
         }
         break;
 
       /* SYNCHRONIZATION PHASE */
       case S_SYNCH:
         if (log.isDebugEnabled()) {
           log.debug("Group: S phase sender version: " + sender.getVersion() + " <? " + msg.vsend);
           log.debug("Group: S phase local version:  " + me.getVersion() + " ==? " + msg.vdest);
         }
         if (me.getVersion() == msg.vdest) {
           estimate.setSynchronized(src);
           if (log.isDebugEnabled())
             log.debug(estimate);
           /*
            * Note that the algorithm in the thesis (fig. 5.7 page 91) is slightly different
            * than this; here the agreed for sender is only set if the sender is synchronized, while
            * the thesis version sets it as long as the local version number of the sender is less than
            * that of the sender (msg.vsend).  It is likely that this version is correct.
            */
           sender.setAgreed(msg.vsend);
         }
         if (sender.getVersion() < msg.vsend) {
           //FIXME The thesis claims that this updating should be done here rather than above:
 //          sender.setAgreed(msg.vsend);
 
           sender.setVersion(msg.vsend);
           EndPoint[] symset = grouptable.getSymset(src);
           try {
             MsgSynch msynch = MsgSynch.marshal(gid, me.getVersion(), sender.getVersion(), symset);
             if (log.isDebugEnabled())
               log.debug("Sending R_SYNCH to " + src);
             mss.send(R_SYNCH, msynch, src);
           } catch (IOException e) {
             log.error("Unable to marshal synchronization message", e);
             return;
           }
         }
         checkSynchronization();
         break;
         
       /* ESTIMATE EXCHANGE PHASE */
       case S_ESTIMATE:
         if (log.isDebugEnabled())
           log.debug("Group: EE phase ");
         if (sender.getVersion() < msg.vsend)
           sender.setVersion(msg.vsend);
         if (log.isDebugEnabled())
           log.debug("Group: sender.agreed " + sender.getAgreed() + " msg.vsend " + msg.vsend);
         if (sender.getAgreed() < msg.vsend && estimate.contains(src)) {
           estimate.remove(msg.rset);
           if (log.isDebugEnabled())
             log.debug(estimate);
           sendEstimate();
         }
 
       } // END switch
     }
   }
 
 
   /**
    *
    *  @param     msg     the m-received message.
    *  @param     src     the sender of the message.
    */
   void handleMsgSymm(MsgSymm msg, EndPoint src)
   {
     if (log.isDebugEnabled())
       log.debug(msg + " from src=" + src);
 
     // In the original algorithm, MsgSymm messages are simply discard 
     // when received in the IDLE phase.
     if (status == S_IDLE)
       return;
 
     int pos = Util.valueAt(msg.hosts, myaddress);
     if (pos >= 0) {
       int vdest = msg.version[pos];
 
       if (status == S_SYNCH) {             // SYNCHRONIZATION PHASE
         if (log.isDebugEnabled())
           log.debug("Group:MsgSymm: S PHASE");
         if (estimate.contains(src) && me.getVersion() == vdest) {
           if (log.isDebugEnabled())
             log.debug("Group:MsgSymm: accepted");
           estimate.remove(msg.rset);
           if (log.isDebugEnabled())
             log.debug(estimate);
           checkSynchronization();
         }
 
       } else if (status == S_ESTIMATE) {   // ESTIMATE EXCHANGE PHASE
         if (log.isDebugEnabled())
           log.debug("Group:MsgSymm: EE PHASE");
         HostData sender = estimate.get(src);
         // FIXME the commented line below is the thesis version;
         // which appears to be wrong, but needs to be verified
 //        if (sender != null && (me.getAgreed() == vdest || sender.getAgreed() <= msg.vsend)) { // Thesis version
         if (sender != null && me.getAgreed() == vdest && sender.getAgreed() == msg.vsend) {
           if (log.isDebugEnabled()) {
             log.debug("Group:MsgSymm: accepted " + estimate);
             for (EndPoint endp : msg.rset) {
               log.debug("removing from estimate: " + endp);
             }
           }
           estimate.remove(msg.rset);
           if (log.isDebugEnabled())
             log.debug(estimate);
           sendEstimate();
         }
       }
     }
   }
 
   /**
    *
    * @param     msg     the m-received message.
    * @param     src     the sender of the message.
    */
   void handleMsgEstim(MsgEstim msg, EndPoint src)
   {
     if (log.isDebugEnabled())
       log.debug("handleMsgEstim: " + msg + " from src = " + src);
 
     /*
      * Locate my position in the estimate in order to determine the
      * estimate agreed value for me.  If I'm not in the estimate, I
      * discard it; do nothing.
      */
     int pos = Util.valueAt(msg.hosts, myaddress);
     if (pos >= 0) {
       int vdest = msg.agreed[pos];
       if (log.isDebugEnabled()) {
         log.debug("Group:vdest " + vdest);
         log.debug("Group:me.getVersion() " + me.getVersion());
       }
 
       if (status == S_SYNCH) {           // SYNCHRONIZATION PHASE
 
         HostData sender = thosts.get(src);
         /* If the sender is valid */
         if (sender != null) {
           if (log.isDebugEnabled())
             log.debug(sender + " is valid");
           if (sender.getVersion() < msg.vsend)
             sender.setVersion(msg.vsend);
           if (!estimate.contains(src)) {
             if (log.isDebugEnabled())
               log.debug("Sending MsgSymm to " + src);
             try {
               MsgSymm tosend = MsgSymm.marshal(gid, me.getVersion(), sender,
                   estimate.getEndPoints(), 1);
               mss.send(R_SYMM, tosend, src);
             } catch (IOException e) {
               log.error("Unable to marshal symmetrical reachability message", e);
               return;
             }
           } else if (me.getVersion() == vdest) {
             if (log.isDebugEnabled())
               log.debug("Before intersect: " + estimate);
             estimate.intersect(msg.hosts);
             estimate.updateAgreed(msg.hosts, msg.agreed);
             if (log.isDebugEnabled())
               log.debug(" After intersect: " + estimate);
             checkSynchronization();
           }
         } else {
           if (log.isDebugEnabled())
             log.debug(sender + " is not valid");
         }
 
       } else if (status == S_ESTIMATE) {         // ESTIMATE EXCHANGE PHASE
 
         if (estimate.contains(src) && estimate.checkAgreed(msg.hosts, msg.agreed)) {
           estimate.intersect(msg.hosts);
           if (log.isDebugEnabled())
             log.debug(estimate);
           sendEstimate();
         } else {
           // Only for debugging
           if (log.isDebugEnabled()) {
             log.debug(estimate);
             if (!estimate.contains(src))
               log.debug("Estimate does not contain " + src);
           }
         }
       }
     } else if (status == S_ESTIMATE && estimate.contains(src)) {
       /*
        * Line 24-25 of Fig. 5.8 pg 94 of Alberto's thesis.
        */
       if (log.isDebugEnabled())
         log.debug("p not in P");
       HostData sender = thosts.get(src);
       /* If the sender is valid */
       if (sender != null && sender.getAgreed() <= msg.vsend) {
         estimate.intersect(msg.hosts);
         if (log.isDebugEnabled())
           log.debug(estimate);
         sendEstimate();
       }
     }
   }
 
 
   /**
    *
    *  @param     msg     the m-received message.
    *  @param     src     the sender of the proposal message.
    */
   void handleMsgProp(MsgProp msg, EndPoint src)
     throws IOException, ClassNotFoundException
   {
     if (log.isDebugEnabled())
       log.debug("handleMsgProp: " + msg + " from src = " + src);
 
     /* Identify the sender */
     HostData sender = thosts.get(src);
     
     /* If the sender is valid */
     if ( status != S_IDLE && sender != null && estimate.contains(src) ) {
       sender.setProposal(msg);
       if (log.isDebugEnabled())
         log.debug("handleMsgProp: sender is valid");
 
       if (status == S_ESTIMATE && checkAgreement()) {
         if (log.isDebugEnabled())
           log.debug("handleMsgProp: ESTIMATE PHASE; agreement reached");
 
         // Creates a new complete view identifier and decodes the next view composition
         long cvid = ViewId.create(myaddress, ++ncoordinated);
         EndPoint[] cvcomp = me.getProposal().decodeHostProp_hosts();
 
         /*
          * Allocates arrays
          */
         
         // View composition length
         int len = cvcomp.length;
         
         // Proposals
         MsgProp[] props = new MsgProp[len];
         
         // Previous view identifiers
         long[] prev_vid = new long[len];
         
         // Previous view compositions
         EndPoint[][] prev_vcomp  = new EndPoint[len][];
 
         // Decoded local members
         LocalId[][] members = new LocalId[len][];
 
         // Destination list
         EndPoint[] dlist = new EndPoint[len];
 
         // Debug related StringBuilder
         StringBuilder buf = new StringBuilder("Proposal: ");
 
         int i,j,k;
 
         // Initializes props and dlist
         for (i=0; i < len; i++) {
           HostData host = (HostData) thosts.get(cvcomp[i]);
           if (host == null)
             throw new IllegalStateException("Corrupted proposal; unknown host in proposal");
           if (log.isDebugEnabled()) {
             buf.append(cvcomp[i]);
             buf.append(" ");
           }
           props[i] = host.getProposal();
           dlist[i] = host.getEndPoint();
 
           // Decodes proposals
           prev_vid[i]   = props[i].cvid;
           prev_vcomp[i] = props[i].decodeLastProp();
           members[i]    = props[i].decodeMembProp();
         }
         if (log.isDebugEnabled()) 
           log.debug(buf.toString());
 
         // Checks whether partial views are needed
         // FIXME OPTIMIZATION: Note that this check could be performed in O(n^2)
         boolean partial = false;
         for (i=0; i < len && !partial; i++)
           for (j=0; j < prev_vcomp[i].length && !partial; j++) {
             k = Util.valueAt(cvcomp, prev_vcomp[i][j]);
             partial = ( k >= 0 && prev_vid[i] != prev_vid[k] );
           }
 
         // Constructs partial view index
         int[] pvids = new int[len];
         int pvid = 0;
         if (partial) {
           int npartial = 0;
           long[] diff_vid = new long[len];
           for (i=0; i < len; i++) {
             for (j=0; j < npartial && prev_vid[i] != diff_vid[j]; j++);
             if (j == npartial) {
               diff_vid[j] = prev_vid[i];
               npartial++;
             }
             pvids[i] = j;
             if (cvcomp[i].equals(myaddress)) {
               pvid = j;
             }
           }
           ncoordinated += npartial;
         }
 
         /* 
          * M-send the view message containing the information
          * Note: in the previous versions, mss.rmsend was used here. rmsend
          * has been removed; instead, daemons receiving and installing a
          * view must msend it to the other members in the view itself.
          * This costs n multicast messages in a cluster of n daemons, but
          * is needed to avoid rare scenarios in which only part of the
          * group members receive a view. 
          * FIXME OPTIMIZATION: instead of forwarding the view, store it
          * and sends only when the coordinator is suspected. In this way
          * we optimize the common case.
          */
         /*
          * If dlist contains just one host, this is the local host.
          * There is no need to send the view to it, as the next lines
          * will perform a local installation without passing through the
          * mss.
          */
         if (dlist.length > 1) {
           MsgView tosend = MsgView.marshal(gid, cvid, me.getProposal(),
             pvids, props, dlist.length);
           mss.msend(R_VIEW, tosend, dlist);
         }
 
         // Locally installs the view
         int[] incarns = me.getProposal().decodeHostProp_incarns();
         installView(cvid, cvcomp, incarns, members, pvids, pvid);
       }
     }
   }
 
   /**
    *  Locates the position of this host in the complete view and checks
    *  whether the version number in the message is equal to the current
    *  one.
    *
    *  @param     msg     the m-received message.
    *  @param     src     the sender of the message.
    */
   void handleMsgView(MsgView msg, EndPoint src)
   {
     if (log.isDebugEnabled()) {
       log.debug("handleMsgView: " + msg + " from src = " + src);
       log.debug(toString(true));
     }
     if (status == S_ESTIMATE && estimate.contains(src)) {
       int pos = Util.valueAt(msg.cvcomp, myaddress);
       if (pos >= 0 && msg.agreed[pos] == ninstalled+1) {
         /* 
          * Re-send the view to all members, to guarantee reliability.
          * FIXME OPTIMIZATION: should be optimized by postponing the
          * send when the coordinator is suspected.
          */
         mss.msend(R_VIEW, msg.marshal(), msg.cvcomp);
         /* Install the view. */
         installView(msg.cvid, msg.cvcomp, msg.incarns, msg.members, msg.pvids, msg.pvids[pos]);
       }
     }
   }
 
   /**
    *
    *  @param     msg     the m-received message.
    *  @param     src     the sender of the message.
    */
   void handleMsgResult(MsgResult msg, EndPoint src)
   {
     if (log.isDebugEnabled())
       log.debug("handleMsgResult: " + msg + " from src = " + src);
     sbuffer.insertMsgResult(msg);
   }
 
 
   /**
    *  Method invoked in response to receiving a JOIN message from a
    *  member that wants to join the specified group.
    *
    *  @param msg
    *    The m-received <code>MsgGroup</code> message.
    *  @param src
    *    The sender <code>EndPoint</code> of the JOIN message; that is
    *    the joining member endpoint.
    */
   boolean handleMsgJoin(MsgGroup msg, EndPoint src)
   {
     if (log.isDebugEnabled())
       log.debug("JOIN:" + msg);
     /*
      * Received a JOIN message, notify membership service to prepare for
      * view change and reply with a JOINACK to the joining member.
      */
     try {
       MsgGroup tosend = MsgGroup.marshal(gid, getHosts(), 1);
       mss.send(R_JOINACK, tosend, src);
     } catch (IOException e) {
       log.error("Unable to marshal group message", e);
     }
     /* Invoke code in common with receiving a JOINACK message */
     return handleMsgJoinAck(msg, src);
   }
 
 
   /**
    *  Method invoked in response to receiving a JOINACK message from the
    *  joining member.
    *
    *  @param msg
    *    The m-received <code>MsgGroup</code> message.
    *  @param src
    *    The sender <code>EndPoint</code> of the JOINACK message; that is
    *    the joining member endpoint.
    */
   boolean handleMsgJoinAck(MsgGroup msg, EndPoint src)
   {
     if (log.isDebugEnabled())
       log.debug("JOINACK: " + msg);
 
     /* Search the hosts in the hosttable */
     MssHost[] hosts = new MssHost[msg.hosts.length];
     for (int i=0; i < hosts.length; i++)
       hosts[i] = hosttable.lookup(msg.hosts[i]);
     MssHost srcHost = hosttable.lookup(src);
 
     /* Join the group */
     return join(srcHost, hosts);
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods related to the total composition of the group
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Removes <code>MemberData</code> items of members that have
    *  requested to leave or have been excluded from the current view.
    *  It also updates the prepared status for any remaining members.
    */
   private void removeLeavingMembers()
   {
     /* 
      * Declares an ACK event for the leave operation; it will be
      * instantiated only when needed.  This is used to notify the member
      * that it has now left the group.
      */
     MemberLeftEvent event = null;
     
     /* Search local leaving members */
     for (Iterator iter = getMembers().iterator(); iter.hasNext(); ) {
       MemberData member = (MemberData) iter.next();
       if (log.isDebugEnabled()) {
         log.debug("Checking member: " + member);
       }
       /* Remove member because it has requested to leave */
       if (member.isLeaving()) {
         /* Creates an ACK event, only when needed */
         if (event == null)
           event = new MemberLeftEvent(gid);
         boolean success = DaemonInteraction.addEvent(member, event);
         if (!success)
           member.setCrashed();
         /* Removes 'member' from the underlying tmembers HashMap. */
         iter.remove();
       } else if (member.isCrashed()) {
         iter.remove();
       } else {
         /* Reset the prepared status for this remaining member. */
         member.resetPrepared();
       }
     }
   }
 
 
   /**
    *  Removes <code>HostData</code> items that have requested to leave
    *  or have been excluded from the current view.
    */
   private void removeLeavingHosts()
   {
     for (Iterator<HostData> iter = getHosts().iterator(); iter.hasNext(); ) {
       HostData host = iter.next();
       if (log.isDebugEnabled())
         log.debug("Checking host: " + host);
       if (host.isLeaving() && !host.getEndPoint().isLocal()) {
         if (log.isDebugEnabled())
           log.debug("Sending R_LEAVE to: " + host);
         try {
           MsgGroup tosend = MsgGroup.marshal(gid, getHosts(), 1);
           mss.send(R_LEAVE, tosend, host.getEndPoint());
         } catch (IOException e) {
           log.error("Unable to marshal group message", e);
         }
         /* Removes 'host' from the underlying thosts HashMap. */
         iter.remove();
       }
     }
   }
 
 
   /**
    *  Send leave notification to all other hosts known to this group.
    */
   void sendLeave()
   {
     for (EndPoint endpoint : thosts.keySet()) {
       if (!endpoint.isLocal()) {
         if (log.isDebugEnabled())
           log.debug("Sending R_LEAVE to: " + endpoint);
         try {
           MsgGroup tosend = MsgGroup.marshal(gid, getHosts(), 1);
           mss.send(R_LEAVE, tosend, endpoint);
         } catch (IOException e) {
           log.error("Unable to marshal group message", e);
         }
       }
     }
     /*
      * Perform leave on the local member last to avoid starting the agreement
      * protocol before having sent all the leave requests.
      */
     if (log.isDebugEnabled())
       log.debug("Leaving: " + myaddress + "; it is the localhost");
     leave(myaddress);
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods related to the current view estimate
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Used to initialize agreement protocol.
    */
   private void runAgreementProtocol()
   {
     if (log.isDebugEnabled())
       log.debug("Running agreement protocol");
     if (status == S_IDLE) {
       synchronizationPhaseInit();
     } else if (status == S_ESTIMATE) {
       try {
         proposal.encodeMembProp(tmembers, nmembers);
       } catch (IOException e) {
         log.error("Unable to encode membership proposal", e);
         return;
       }
       sendProposal();
     }
   }
 
 
   /**
    *  Initialize the synchronization phase; we are in the process of running
    *  a new agreement protocol to agree on a new view.
    */
   private void synchronizationPhaseInit()
   {
     if (log.isDebugEnabled())
       log.debug("Preparing ESTIMATE for SYNCHRONIZATION phase");
       /* Initialize a new estimate */
       if (estimate == null) {
         estimate = new Estimate(sbuffer, thosts, rset);
       } else {
         estimate.reset(thosts, rset);
       }
  
       /* Enters SYNCHRONIZATION phase */
       status = S_SYNCH;
       if (log.isDebugEnabled()) {
         MDC.put("group", toString());
       log.debug("Entering SYNCHRONIZATION phase");
     }
 
     /* 
      * We are in the process of running a new agreement protocol;
      * notify local members that the current view has become invalid.
      */
     PrepareEvent prepareEvent = null;
     for (Iterator iter = getMembers().iterator(); iter.hasNext();) {
       MemberData md = (MemberData) iter.next();
       if (!md.isPrepared() && !md.isLeaving() && !md.isCrashed()) {
         /* Creates the PrepareEvent, only when needed */
         if (prepareEvent == null)
           prepareEvent = new PrepareEvent(gid);
         if (log.isDebugEnabled())
           log.debug("Sending PrepareEvent to " + md);
         boolean success = DaemonInteraction.addEvent(md, prepareEvent);
         if (!success)
           md.setCrashed();
       }
     }
 
     /* Increment the version number */
     me.newVersion();
     
     /* Send a SYNCH message to the hosts in the estimate */
     Iterator iterator = estimate.getHosts().iterator();
     while (iterator.hasNext()) {
       HostData host = (HostData) iterator.next();
       if (!host.equals(me)) {
         EndPoint[] symset = grouptable.getSymset(host.getEndPoint());
         try {
           MsgSynch tosend = MsgSynch.marshal(gid, me.getVersion(), host.getVersion(), symset);
           if (log.isDebugEnabled())
             log.debug("Sending R_SYNCH to " + host.getEndPoint());
           mss.send(R_SYNCH, tosend, host.getEndPoint());
         } catch (IOException e) {
           log.error("Unable to marshal synchronization message", e);
           return;
         }
       }
     }
     
     /* If synchronization has ended, go to ESTIMATE EXCHANGE phase */
     checkSynchronization();
   }
 
 
   /**
    * Check if all in estimate has been synchronized; if so enter the
    * ESTIMATE EXCHANGE phase.  Otherwise, do nothing and await further
    * R_SYNCH messages.
    */
   private void checkSynchronization()
   {
     if (estimate.isSynchronized())
       estimateExchangePhaseInit();
   }
 
 
   /**
    *
    */
   private void estimateExchangePhaseInit()
   {
     /* Enter the estimate exchange phase */
     status = S_ESTIMATE;
     if (log.isDebugEnabled()) {
       MDC.put("group", toString());
       log.debug("Group:estimateExchangePhaseInit: start");
       log.debug(estimate);
     }
 
     try {
 
       /*
        * Send a new estimate.  If the current estimate size is equal to
        * one, it contains just the local host; there is no need to send
        * this message to it.
        */
       if (estimate.size() > 1) {
         MsgEstim tosend = MsgEstim.marshal(gid, me.getVersion(), estimate.getHosts());
         mss.msend(R_ESTIM, tosend, estimate.getEndPoints());
       }
 
       /* Builds a new proposal and (possibily) sends it */
       proposal.encodeLastProp(viewDesc.getEndPoints());
       proposal.encodeHostProp(estimate.getHosts());
       proposal.encodeMembProp(tmembers, nmembers);
       if (msgsStable) {
         int[] delivered = viewDesc.getDelivered(estimate.getHosts());
         proposal.encodeMsgsProp(delivered);
         sendProposal();
       } else {
         if (log.isDebugEnabled()) {
           log.debug("Messages are not stable: ");
           log.debug(viewDesc);
         }
       }
     } catch (IOException e) {
       log.error("Unable to encode proposal for the EE-phase", e);
       return;
     }
 
     /* Forward unacknowledged messages to the other members */
     viewDesc.forwardMessages(mss);
     
     if (log.isDebugEnabled()) {
       log.debug("Group:estimateExchangePhaseInit: stop");
     }
   }
 
 
   /**
    *  Sends a new view proposal to the coordinator.
    */
   private void sendProposal()
   {
     /*
      * This loop only logs problems if a member has outstanding ACKs while
      * it is deemed that messages are stable.  This happens rarely, therefore
      * we need to collect data for these cases.
      */
     for (MemberData md : tmembers.values()) {
       log.assertLog(!(md.hasOutstandingAcks() && msgsStable), md + " has outstanding ACKs while msgs are stable");
     }
     if (log.isDebugEnabled())
       log.debug("Messages are " + (msgsStable ? "stable": "not stable")); 
     if (msgsStable && installationComplete) {
 
       try {
         MsgProp msg = proposal.marshal(gid, cvid, pvid);
 
         /*
          * Obtain the coordinator; note that we should never get here if
          * the estimate is empty and no coordinator can be selected.
          */
         EndPoint coordinator = estimate.getCoordinator();
         if (coordinator.isLocal()) {
           /*
            * This deamon is the coordinator; so, locally deliver the
            * proposal message without passing it through the mss.
            */
           if (log.isDebugEnabled())
             log.debug("Group:sendProposal: local coordinator");
           handleMsgProp(new MsgProp(msg.getOutMessage()), coordinator);
 
         } else {
           /*
            * Send a proposal message to the coordinator (distinct from
            * this daemon).
            */
           if (log.isDebugEnabled())
             log.debug("Group:sendProposal: remote coordinator: " + coordinator);
           mss.send(R_PROP, msg, coordinator);
 
         }
       } catch (IOException e) {
         log.error("Unable to marshal and send proposal message", e);
       } catch (ClassNotFoundException e) {
         log.error("Unable to unmarshal (local) proposal message", e);
       }
     }
     if (log.isDebugEnabled())
       log.debug("Group:sendProposal: end");
   }
 
 
   /**
    *  If the estimate has changed (and is not empty), send it
    *  all members in the estimate.  If the message buffer is
    *  consistent between members of the previous view, send also
    *  a proposal to the coordinator.
    */
   private void sendEstimate()
   {
     if (estimate.hasChanged() && !estimate.isEmpty()) {
       try {
         /*
          * If estimate contains just one host, this is the local host.
          * There is no need to send the estimate to it, as the local
          * host is already aware of its estimate.
          */
         if (estimate.size() > 1) {
           MsgEstim tosend = MsgEstim.marshal(gid, me.getVersion(), estimate.getHosts());
           mss.msend(R_ESTIM, tosend, estimate.getEndPoints());
         }
         proposal.encodeHostProp(estimate.getHosts());
         if (msgsStable) {
           int[] delivered = viewDesc.getDelivered(estimate.getHosts());
           proposal.encodeMsgsProp(delivered);
           sendProposal();
         } else {
           if (log.isDebugEnabled()) {
             log.debug("Messages are not stable: ");
             log.debug(viewDesc);
           }
         }
       } catch (IOException e) {
         log.error("Unable to marshal estimate proposal message", e);
       }
     }
   }
 
   /**
    * Checks if an agreement has been established. A set of proposals are
    * in agreement if and only if: (i) all proposals contains the same estimate
    * for the next view, and (ii) if two proposals in the set have been generated
    * in the same view, the sending hosts must have delivered the same set
    * of messages. <br>
    * Point (i) is checked by verifying that all hosts have the same estimate
    * as the coordinator; method checkEstimate() compares two estimates.
    * Point (ii) is checked by verifying that given any pair of hosts in the
    * set, if they have the same view they must have delivered the same set
    * of messages. This check is done by checkMessages().
    */
   private boolean checkAgreement()
   {
     MsgProp proposalp = me.getProposal();
     if (proposalp == null) {
       if (log.isDebugEnabled())
         log.debug("checkAgreement: no local proposal found");
       return false;
     }
     if (log.isDebugEnabled())
       log.debug("checkAgreement: found my proposal " + me);
     for (HostData hostq : estimate.getHosts()) {
       MsgProp proposalq = hostq.getProposal();
       if (!proposalp.checkEstimate(proposalq)) {
         if (log.isDebugEnabled()) {
           log.debug("PROPOSAL DIFFERS FROM LOCAL ESTIMATE");
           log.debug("proposalp: " + proposalp);
           log.debug("proposalq: " + proposalq);
         }
         return false;
       }
       if (log.isDebugEnabled())
         log.debug("checkAgreement: ok estimate from " + hostq);
       for (HostData hostr : estimate.getHosts()) {
         // No need to compare the proposals of the same host
         if (hostq == hostr)
           continue;
         if (log.isDebugEnabled()) {
           log.debug("Checking proposals from: ");
           log.debug(hostq);
           log.debug(hostr);
         }
         MsgProp proposalr = hostr.getProposal();
         if (!proposalq.checkMessages(proposalr)) {
           if (log.isDebugEnabled()) {
             log.debug("PROPOSALS DIFFERS (MESSAGES)");
             log.debug("proposalq: " + proposalq);
             log.debug("proposalr: " + proposalr);
           }
           return false;
         }
       }
     }
     if (log.isDebugEnabled())
       log.debug("checkAgreement: all ok");
     return true;
   }
 
 
   /**
    *
    */
   private void acceptMessage(MsgMcast msg, boolean late)
   {
     if (log.isDebugEnabled())
       log.debug("acceptMessage; " + (late ? "late" : ""));
     try {
       msg.complete(viewDesc.getLocalIndex(), pvid, sbuffer.getMid());
     } catch (IOException e) {
       log.error("Cannot complete marshalling multicast message", e);
       return;
     }
 
     /* Add message to local buffer */
     sbuffer.insertMsgMcast(msg);
     
     /* Local delivery of the message */
     boolean newCrashed = viewDesc.notifyMessage(gid, msg);
     if (log.isDebugEnabled())
       log.debug("Accept multicast message (stable=false)");
 
     /* Unstable until message is delivered */
     msgsStable = false;
 
     /**
      *  If the current view size is equal to one, it contains just the
      *  local host; there is no need to send this message to it
      */
     if (viewDesc.size() > 1) {   
       if (late) {
         mss.msend(R_FORWARD, msg, viewDesc.getEndPoints());
       } else {
         mss.msend(R_MCAST, msg, viewDesc.getEndPoints());
       }
     }
   
     /*
      *  If the message has not been delivered because of a new
      *  crashed hosts, start a new agreement protocol (if needed),
      *  or complete the member list and send a proposal, otherwise.
      */
     if (newCrashed) {
       if (log.isDebugEnabled()) {
         log.debug("Detected new crashed host; running agreement protocol");
       }
       runAgreementProtocol();
     }
   }
 
 
   /**
    *
    */
   private void installView(long cvid, EndPoint[] hosts, int[] incarns,
                            LocalId[][] members, int[] pvids, int pid)
   {
     // Stores complete view information
     this.cvid = cvid;
 
     /*
      * Reset the view index for all hosts, indicating that there isn't
      * any hosts is in the current view.  This will be followed by
      * setting the view index to the correct value for hosts that are
      * included in the current view.
      */
     for (HostData hd : thosts.values()) {
       hd.resetViewIndex();
     }
 
     /*
      * Computes the host positions in the current view and removes
      * information related to hosts and members not participating in the
      * next (partial) view.
      */
     List<MemberId> memberSet = new ArrayList<MemberId>(hosts.length);
     Set<Integer> uniquePids = new HashSet<Integer>(pvids.length);
     LocalId[] localMembers = null;
     int hostCount = 0;
     int hpos = 0;
     for (int i = 0; i < hosts.length; i++) {
       uniquePids.add(pvids[i]);
       if (pvids[i] == pid) {
         if (hosts[i].equals(myaddress)) {
           hpos = hostCount;
           localMembers = members[i];
         }
         HostData hd = (HostData) thosts.get(hosts[i]);
         if (hd == null)
           throw new IllegalStateException("Estimate contains an unknown host");
         for (int j=0; j < members[i].length; j++) {
           memberSet.add(new MemberIdImpl(hd.getEndPoint(), incarns[i], members[i][j]));
         }
         /*
          * Reorder the hosts[] array; note that the 'hostCount' variable will always
          * be less than or equal to 'i' and hence it should not cause problems for
          * the iteration over the hosts[] array.
          */
         hosts[hostCount] = hosts[i];
         hd.setViewIndex(hostCount++);
       }
     }
 
     // Stores current view information
     viewDesc = new ViewDescription(thosts, hosts, rset, tmembers, localMembers, hpos);
 
     // Updates group flags
     msgsStable = true;
     installationComplete = false;
     status = S_IDLE;
     if (log.isDebugEnabled()) {
       MDC.put("group", toString());
       log.debug("New view installed (stable=true)");
     }
 
     /* Create the new view and delivers it to non-leaving members */
     pvid = ViewId.createPartial(cvid, pid);
     MemberId[] memberIds = (MemberId[]) memberSet.toArray(new MemberId[memberSet.size()]);
     //FIXME this may cause problems for the merging layer since the uniquePids appears to not reflect
     // the new view and the number of partitions that the members come from; implement test case for this??
     View view = new ViewImpl(gid, pvid, uniquePids.size(), memberIds);
     ninstalled++;
     viewDesc.notifyView(gid, view);
 
     // Remove hosts and members that have request to leave
     removeLeavingMembers();
     removeLeavingHosts();
 
     if (!tmembers.isEmpty()) {            // The group has not to be removed
 
       // Updates HostData information
       for (Iterator iter = thosts.values().iterator(); iter.hasNext(); ) {
         HostData hd = (HostData) iter.next();
         hd.setProposal(null);
         // PVIEW Should we move also ctbl to ViewDescription
       }
 
       // Updates information about messages
       sbuffer.setView(view, hostCount);
       MsgMcast scan;
       while ((scan = (MsgMcast) delayedSent.removeFirst()) != null)
         acceptMessage(scan, false);
       while ((scan = (MsgMcast) delayedRcvd.removeFirst()) != null)
         if (scan.vid == pvid)
           viewDesc.notifyMessage(gid, scan);
 
       /*
        * If the current view size is equal to one, it contains just the
        * local host; there is no need to send this message to it.
        */
       if (!sbuffer.isEmpty() && viewDesc.size() > 1) {
         try {
           MsgAck tosend = MsgAck.marshal(gid, pvid, viewDesc.getLocalIndex(),
             viewDesc.getAcks(), viewDesc.size());
           mss.msend(R_ACK, tosend, viewDesc.getEndPoints());
         } catch (IOException e) {
           log.error("Cannot marshal ACK message", e);
           return;
         }
       }
 
       // Checks if the view is stable
       if (!viewDesc.isStable() || viewDesc.containsCrashedMembers())
         synchronizationPhaseInit();
     }
   }
 
 } // END Group