/*
    * 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.mss;
  
  import java.io.IOException;
  import java.net.DatagramPacket;
  
  import jgroup.core.ConfigurationException;
  import jgroup.core.EndPoint;
  import jgroup.core.JgroupException;
  import jgroup.relacs.config.DistributedSystemConfig;
  import jgroup.relacs.config.TransportConfig;
  import jgroup.relacs.daemon.DaemonMsg;
  import jgroup.relacs.events.Event;
  import jgroup.util.InMessage;
  
  import org.apache.log4j.Logger;
  
  
  /**
   *  The <code>Mss</code> class implements a Multi-Send Service.
   *
   *  @author Salvatore Cammarata
   *  @author Alberto Montresor
   *  @author Hein Meling
   *  @since Jgroup 1.2
   */
  public final class Mss
    implements EhandlerUser, MssConstants, MssTag
  {
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Logger
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Obtain logger for this class */
    private static final Logger log = Logger.getLogger(Mss.class);
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Unique message instance
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /**
     *  As this message is never stored in a queue, a single message
     *  instance is sufficient.  This allows us to save the cost of
     *  allocating and garbage collecting messages.  In fact, this message
     *  does not change at all.
     */
    private static byte[] sentFrag;
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Private Fields
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** The host on which the daemon is running. */
    private MssHost localHost;
  
    /** Upper layer */
    private MssUser mssuser;
    
    /** Distributed System configuration */
    private MssDS mssds;
  
    /** Event scheduler */
    private EventHandler ehandler;
  
    /** Message Flow Control Layer */
    private MsgCntrl msgCntrl;
  
    // Timeouts
    
    /** Routing timeout; how often the routing table are sent */
    private long routingTimeout;
  
    /** Congestion timeout */
    private long congestionTimeout;
  
    /** Last routing event scheduled; used to avoid having multiple timeouts */ 
    private ScheduledEvent lastRoutingScheduled;
   
    // Fragmentation
 
   /** Payload size */
   private int payload;
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Constructors
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Constructs a new <code>Mss</code> object.
    *
    *  @param mssuser
    *    The upper level layer, using this layer.
    *  @param dsc
    *    The distributed system configuration.
    *  @param tconf
    *    The transport parameter configuration.
    *
    *  @exception ConfigurationException
    *    Thrown if the distributed system is incorrectly configured.
    *  @exception IOException
    *    Thrown if there are problems at socket layer.
    */
   public Mss(MssUser mssuser, DistributedSystemConfig dsc, TransportConfig tconf)
     throws ConfigurationException, IOException
   {
     this.mssuser = mssuser;
 
     /* Timeout configuration */
     routingTimeout = tconf.getRoutingTimeout();
     congestionTimeout = tconf.getCongestionTimeout();
 
     /* Message configuration parameters */
     payload = tconf.getPayload();
 
     /* Configure other <code>Mss</code> layers */
     ehandler = new EventHandler(this);
     mssds = new MssDS(ehandler, dsc, tconf);
     localHost = HostTable.getLocalHost();
 
     /*
      * Initialize the single instance SENT message.  Needs to be done
      * after local host initialization, since FragmentHeader marshal
      * needs to obtain the incarnation identifier from the local
      * MssHost object.
      */
     sentFrag = FragmentHeader.marshal(SENT, false, UNDEF);
 
     /* Flow Control Layer initialization */
     msgCntrl = new MsgCntrl(ehandler, mssuser, mssds);
     mssds.setControl(msgCntrl);
 
     /*
      * Compute the next routing timeout; the timeout are choosen in a
      * probabilistic way in order to have only a routing table sent out
      * in the cluster.
      */
     RandomGenerator.setRoutingTimeout(routingTimeout);
 
     /*
      * Schedule the topology timeouts
      */
     iamAliveTimeout();
     routingTimeout();
     // congestion timeout
     congTimeout();
     // now we are visible on the net: start the network layer
     mssds.doStart();
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Multicast send methods
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Sends message the <code>msg</code> to a single host specified by
    *  <code>receiver</code>.  This method is optimized for sending
    *  directly to the given <code>receiver</code>, bypassing the cluster
    *  leader.
    */
   public void send(byte tag, DaemonMsg msg, EndPoint receiver)
   {
     if (receiver.isLocal()) {
       throw new IllegalArgumentException("Local message sending not allowed");
     }
 
     /* Add the host to the receiver set for the cluster */
     MssHost host = mssds.hostLookup(receiver);
     Cluster cluster = host.getCluster();
     cluster.addReceiver(receiver);
 
     /*
      * Marshal the mss level data into the message stream.
      */
     MsgJG msgjg = null;
     try {
       msgjg = MsgJG.marshal(msg, tag, new EndPoint[] { receiver }, mssds);
     } catch (IOException e) {
       log.error("Cannot marshal the mss level data into the message stream", e);
       return;
     }
     
     /*
      * If there exists at least one destination host belonging to the
      * cluster that is reachable we attempt to send to the cluster.
      */
     if (cluster.isReachable()) {
       /*
        * Send the message to the cluster of the receiving host
        */
       cluster.send(msgjg);
 
     } else if (log.isDebugEnabled()) {
       log.debug("msend failed: no reachable member for cluster " + cluster.getEndPoint());
     }
     cluster.clearReceivers();
   }
 
 
   /**
    *  M-sends message <code>msg</code> to all the hosts in the
    *  hosttable.  Only a single message is sent to each cluster leader,
    *  and from there sent to the corresponding members in that cluster.
    */
   public void msend(byte tag, DaemonMsg msg)
   {
     /*
      * Marshal the mss level data into the message stream.
      */
     MsgJG msgjg = null;
     try {
       msgjg = MsgJG.marshal(msg, tag, mssds);
     } catch (IOException e) {
       log.error("Cannot marshal the mss level data into the message stream", e);
       return;
     }
 
     /*
      * Send the message to all reachable clusters.
      */
     ClusterTable clustertable = mssds.getClusterTable();
     for (int i = 0, size = clustertable.size(); i < size; i++) {
       Cluster cluster = clustertable.get(i);
       if (cluster.isReachable()) {
 
         cluster.send(msgjg);
 
       } else if (log.isDebugEnabled()) {
         log.debug("msend failed: no reachable member for cluster " + cluster.getEndPoint());
       }
     }
   }
 
 
   /**
    *  M-sends message <code>msg</code> to the hosts in the destination
    *  array <code>receivers</code>.
    */
   public void msend(byte tag, DaemonMsg msg, EndPoint[] receivers) 
   {
     /* Add each receiver to the destination set of its cluster */
     for (int i = 0; i < receivers.length; i++) {
       if (!receivers[i].isLocal()) {
         MssHost host = mssds.hostLookup(receivers[i]);
         host.getCluster().addReceiver(receivers[i]);
       } else if (log.isDebugEnabled()) {
         log.debug("Local host has been removed from the destination list");
       }
     }
 
     /*
      * Marshal the mss level data into the message stream.
      */
     MsgJG msgjg = null;
     try {
       msgjg = MsgJG.marshal(msg, tag, receivers, mssds);
     } catch (IOException e) {
       log.error("Cannot marshal the mss level data into the message stream", e);
       return;
     }
 
     /*
      * Send the message to all reachable clusters with receivers.
      */
     ClusterTable clustertable = mssds.getClusterTable();
     for (int i = 0, size = clustertable.size(); i < size; i++) {
       Cluster cluster = clustertable.get(i);
       /*
        * If there exists at least one destination host belonging to the
        * cluster that is reachable we attempt to send to the cluster.
        */
       if (cluster.hasReceivers()) {
         if (cluster.isReachable()) {
 
           cluster.send(msgjg);
 
         } else if (log.isDebugEnabled()) {
           log.debug("msend failed: no reachable member for cluster " + cluster.getEndPoint());
         }
       }
       cluster.clearReceivers();
     }
   }
 
   /**
    * Returns true if the <code>Ehandler</code> local message queue
    * contains undelivered messages; otherwise, false is returned.
    */
 //  public boolean hasUndeliveredLocalMsgs(int gid)
 //  {
 //    return ehandler.hasUndeliveredLocalMsgs(gid);
 //  }
 
   /**
    * Returns a string containing the current content of the queues.
    */
   public String toString()
   {
     return ehandler.toString();
   }
   
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Implementation of abstract methods inherithed from NIListener
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Handles a message received from a remote host
    */ 
   public void rreceive(DatagramPacket packet)
   {
     byte[] buf = packet.getData();
     int bufLen = packet.getLength();
 
     log.assertLog(buf.length <= payload + OVERHEAD_SIZE,
       "Buffer length (" + buf.length + ") should not exceed the payload plus overhead ("
       + (payload + OVERHEAD_SIZE) + ")");
 
     try {
       FragmentHeader header = FragmentHeader.unmarshal(mssds, buf, bufLen);
       if (log.isDebugEnabled())
         log.debug("packet header=" + header);
 
       /*
        *  This cluster is the destination for the message fragment.
        */
 
       /* Check if the sender has the same incarnation */
       if (mssds.hasNewIncarnation(header))
         generateSuspect();
 
       switch(header.getTag()) {
 
       case SENT:
         {
           if (header.isLocal()) {
             if (log.isDebugEnabled())
               log.debug("Discarding locally sent SENT message");
           } else {
             log.assertLog(lastRoutingScheduled != null, "Last scheduled ROUTING event is null");
 
             if (log.isDebugEnabled())
               log.debug("rreceived packet SENT header= " + header);
             /* Delete the scheduled ROUTING timeout */
             ehandler.abortTimeout(lastRoutingScheduled);
 
             /* Schedule a new timeout for the last ROUTING event */
             long timeout = RandomGenerator.getRoutingTimeout();
             lastRoutingScheduled = ehandler.setTimeout(timeout, new ScheduledEvent(ROUTING));
           }
         }
         break;
 
       case IAMALIVE:
         {
           /*
            * Note that we need to process local IAMALIVE messages.
            *
            * Note also that IAMALIVE messages are not fragmented since
            * the message identifier for these messages are set to UNDEF,
            * and this conflicts with the need to give each fragment a
            * unique number.  This can potentially be a problem for large
            * distributed systems, using a very low payload value.
            *
            * FIXME: We should add a check; if the payload is too low to
            * keep an entire IAMALIVE message.
            */
           InMessage inmsg = new InMessage(bufLen, HEADER_SIZE, TRAILER_SIZE, bufLen);
           inmsg.insert(buf, bufLen);
           handleIamAliveMsg(MsgIamAlive.unmarshal(inmsg, header));
         }
         break;
 
       case ROUTING:
         {
           log.assertLog(!header.isLocal(),
             "Received ROUTING msg sent by me; shouldn't happen.");
 
           /*
            * Forward the message to my cluster directly, modifying only
            * the fragment tag, indicating that it is a forwarded routing
            * message.  This does not change the sender field of the
            * message, so that the receiver will know the actual
            * originator of the message.
            */
           header.setTag(FWDROUTING);
           localHost.getCluster().forward(header.getFragment(), header.getFragmentLength());
         }
         break;
 
       case FWDROUTING:
         {
           /*
            * Note that FWDROUTING messages are forwarded without
            * modifying the sender field of the message fragment.
            */
           log.assertLog(!header.isLocal(),
             "Received FWDROUTING msg sent by me; should this happen?");
 
           /*
            * Note that (FWD)ROUTING messages are not fragmented since
            * the message identifier for these messages are set to UNDEF,
            * and this conflicts with the need to give each fragment a
            * unique number.  This can potentially be a problem for large
            * distributed systems, using a very low payload value.
            *
            * FIXME: We should add a check; if the payload is too low to
            * keep an entire ROUTING message.
            */
           InMessage inmsg = new InMessage(bufLen, HEADER_SIZE, TRAILER_SIZE, bufLen);
           inmsg.insert(buf, bufLen);
           MsgRouting msg = MsgRouting.unmarshal(inmsg, header);
           handleRoutingMsg(msg);
         }
         break;
 
       case NACK:
         {
           if (header.isLocal()) {
             if (log.isDebugEnabled())
               log.debug("Discarding locally sent NACK message");
           } else {
             /*
              * NACK messages are expected to fit within a single message
              * fragment; is it does not require fragmentation support.
              */
             InMessage inmsg = new InMessage(bufLen, HEADER_SIZE, TRAILER_SIZE, bufLen);
             inmsg.insert(buf, bufLen);
             MsgNACK nack = MsgNACK.unmarshal(inmsg, header, mssds);
             handleNACKMsg(nack);
           }
         }
         break;
 
       case SYN:
         {
           /*
            * SYN messages are expected to fit within a single message
            * fragment; it does not require fragmentation support.
            */
           InMessage inmsg = new InMessage(bufLen, HEADER_SIZE, TRAILER_SIZE, bufLen);
           inmsg.insert(buf, bufLen);
           MsgSYN syn = MsgSYN.unmarshal(inmsg, header);
           handleSYNMsg(syn);
         }
         break;
 
       case NOTLASTFRAGMENT:
       case JG:
         {
           /* Avoid delivering messages sent locally */
           if (header.isLocal()) {
             if (log.isDebugEnabled()) {
               log.debug("Discarding local upper layer message (" + header + ")");
             }
             return;
           }
 
           // FIXME Add support to guarantee order of messages at low level
           if (!handleJGInfo(header)) {
             /*
              * If we fail to deliver the msg, reschedule it.
              * 
              * Note that most messages received here will be discarded
              * since they are received in a different view; however,
              * we may get important synchronization messages before
              * we are able to handle them, thus we reschedule all
              * of these messages.
              */
             ehandler.rnotify(packet);
           }
         }
         break;
 
       default:
         log.warn("Invalid packet tag: " + header.getTag());
       }
     } catch (IOException e) {
       /* Malformed message, just return without trying to deliver it. */
       log.error("rreceive: malformed packet", e);
     } catch (ClassCastException e) {
       log.error("rreceive: malformed packet", e);
     } catch (ClassNotFoundException e) {
       log.error("rreceive: malformed packet", e);
     } catch (JgroupException e) {
       log.error("rreceive: packet from unknown host; all hosts must be in the config.xml", e);
     }
   }
 
 
   /**
    *  Handles a message received by a local member
    */
   public void lreceive(Event event)
   {
     mssuser.localReceive(event);
   }
 
 
   /**
    *  Handle expired timeout event
    */
   public void treceive(ScheduledEvent event)
   {
     switch (event.getTag()) {
 
       case IAMALIVE:
         iamAliveTimeout();
         break;
 
       case ROUTING:
         routingTimeout();
         break;
 
       case REMOTENACK:
         nackTimeout((MsgNACK) event.getData());
         break;  
 
       case CONGESTION:
         congTimeout();
         break;
 
       default:
         log.warn("Unknown timeout event received");
     }
   }
 
 
   public void lsend(Event msg)
   {
     ehandler.lnotify(msg);
   }
 
 
   /**
    *  Returns the hosttable.
    */
   public HostTable getHostTable()
   {
     return mssds.getHostTable();
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Message handlers
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Handle a synchronization message
    */
   private void handleSYNMsg(MsgSYN msg) 
   {
     if (log.isDebugEnabled())
       log.debug("handleSYNMsg: " + msg);
 
     /* The source member cannot be unreachable here */
 
     MssHost sender = msg.getSender();
     switch (msg.getSYNType()) {
 
       case QSYN:
         /*
          * (SYN_REQUEST) The msg sender is asking for a SYN msg from me.
          * We send an ASYN message as reply.
          */
         try {
           /*
            * PATCH added by Hein 22/7-03.  The message flow was not
            * initialized correctly after two clusters had been partitioned
            * and were merging, that is they didn't synchronize.  Therefore,
            * we check if the sender side message flow is already initialized;
            * if not, we reset the message flow for the sending side.
            */
           MsgFlowSndrSide msgFlow = msg.getMsgFlow();
           if (!msgFlow.isInitialized()) {
             Cluster cluster = msg.getSender().getCluster();
             cluster.resetMsgFlow();
             if (log.isDebugEnabled())
               log.debug("cluster.resetMsgFlow()");
           }
           MsgSYN syn = MsgSYN.marshal(ASYN, msg, msgFlow.getLastMsgSent());
           sender.unicastRouteSend(syn);
           if (log.isDebugEnabled())
             log.debug("Sent " + syn);
         } catch (IOException e) {
           log.error("Unable to marshal ASYN message; nothing sent", e);
         }
         break;
 
       case ASYN:
         /*
          * (SYN_REPLY) The msg sender answer to our SYN request.  We
          * check that the sender has the right nonce.
          */
         if (sender.checkSynId(msg.getMid())) {
 
           /* Abort timeout for synchronization since we received a reply. */
           sender.abortTimeout();
 
           /* Reset message flow control (receiver side) */
           sender.resetMsgFlow(msg.getLastMsgSent());
 
           /* PATCH: added by Hein 28/7-02. See email.
            *
            * Reset message flow control for the cluster (sender side) if
            * it was previously unreachable; that is the message flow was
            * previously reset.
            */
           Cluster cluster = sender.getCluster();
           if (!cluster.getMsgFlow().isInitialized()) {
             cluster.resetMsgFlow(msg.getLastMsgSent());
           }
 
           /*
            * Notify the upper layer that the sender of the ASYN message is
            * now reachable, and we have synchronized our sequence numbers.
            */
           mssds.getUpperView().setAsReachable(sender);
           if (log.isDebugEnabled())
             log.debug("Synchronized to " + sender + " nonce (" + msg.getMid()
                       + ") - Msg flow reset to lastMsgSent=" + msg.getLastMsgSent());
 
           /* The two members are now synchronized. */
         }
         break;
 
       default:
         log.warn("Invalid synchronization message received");
     }
     if (log.isDebugEnabled())
       log.debug("handleSYNMsg end");
   }
 
 
   /**
    *  Handle NACK message
    */
   private void handleNACKMsg(MsgNACK nack) 
   {
     if (log.isDebugEnabled())
       log.debug(nack);
 
     /* The source member cannot be unreachable here */
 
     if (NACKSUPPRESSION) {
       /*
        * NACK suppression sends only a single NACK for each cluster.
        */
 
       MssHost src = nack.getSource();
       if (src.isLocal()) {
         /*
          * I'm the source of the missing message; resend it and update
          * the congestion information (NACK count).
          */
         resendMsg(nack);
         nack.getMsgFlow().incNackCount();
 
       } else if (!nack.isRemoteNACK()) {
         /*
          * Another member of the cluster has sent a NACK message; check
          * if also this member is waiting (have scheduled) a NACK event
          * for this particular message.  In which case we abort the
          * current NACK event and reschedule it at a later point in
          * time, expecting to receive the missing message soon, due to
          * the REMOTENACK sent by the NACK leader of this cluster.
          */
         ScheduledEvent eventNack = src.getMsgFlow().getScheduledEvent(nack.mid);
         if (eventNack == null) {
           log.warn("No scheduled NACK event for " + nack.mid);
         } else {
           /* Delete the scheduled NACK timeout */
           ehandler.abortTimeout(eventNack);
           /*
            * Reschedule a new timeout for the NACK (to allow additional
            * time for receiving the resent message)
            */
           long randomTimeout = RandomGenerator.getRandomTimeout(src.getTimeout());
           ehandler.setTimeout(randomTimeout, eventNack);
         }
       }
 
     } else if (nack.isRemoteNACK()) {
       /*
        * Without NACK suppression the source of the missing message will
        * receive <i>cluster.size</i> messages for a single missing
        * message.  Even more NACKs will be sent if the message is
        * missing also in other clusters.
        */
       resendMsg(nack);
 
     }
     if (log.isDebugEnabled())
       log.debug("handleNACKMsg end");
   }
 
 
   /**
    *  Recover the message corresponding to the given NACK message, and
    *  resend it to the destination cluster.
    */
   private void resendMsg(MsgNACK nack)
   {
     if (log.isDebugEnabled())
       log.debug(nack);
     Cluster destCluster = nack.getCluster();
     MsgFlowSndrSide msgFlow = destCluster.getMsgFlow();
 
     FragmentIterator fragIter = msgFlow.getSentMsgFrag(nack);
     if (fragIter != null) {
       if (log.isDebugEnabled()) {
         log.debug("fragIter=" + fragIter);
       }
       destCluster.resend(fragIter, nack.mid);
       destCluster.clearReceivers();
 
       if (log.isDebugEnabled()) {
         log.debug("Resent msg fragment (" + nack.mid + ") to cluster " + destCluster.getEndPoint());
       }
     } else {
       log.warn("Cannot resend msg fragment (" + nack.mid + "); already removed from cluster queue "
                + destCluster.getEndPoint());
     }
   }
 
 
   /**
    *  Handle routing message (forwarded from a cluster member).
    */
   private void handleRoutingMsg(MsgRouting msg)
   {
     if (log.isDebugEnabled())
       log.debug("handleRoutingMsg start");
 
     /*
      * Check the incarnations associated with the FWDROUTING message.
      */
     if (mssds.checkIncarnation(msg))
       generateSuspect();
 
     /* Update routing table */
     mssds.updateRoutingTable(msg);
 
     /* Update Flow Control information */
     msg.updateFC();
     generateSuspect();
 
     if (log.isDebugEnabled())
       log.debug("handleRoutingMsg end");
   }
 
 
   /**
    * Handle I am alive message.
    */
   private void handleIamAliveMsg(MsgIamAlive msg)
   {
     if (log.isDebugEnabled())
       log.debug("handleIamAliveMsg start");
 
     /* Update reachability info if message sender is not local */
     if (!msg.isLocal())
       mssds.updateReachability(msg);
     /* Update Flow Control information */
     msg.updateFC(mssds);
     generateSuspect();
 
     if (log.isDebugEnabled())
       log.debug("handleIamAliveMsg end");
   }
 
 
   /**
    * Send synchronization requests to all newly reachable hosts,
    * and reset the synchronization status for the new unreachable
    * hosts.  Finally, notify the upper layer if the view changed.
    */
   private void generateSuspect() 
   {
 
     /* Obtain the view */  
     DSView view = mssds.getView();
 
     if (log.isDebugEnabled())
       log.debug("generateSuspect(): view = "+ view);
 
     /* Send SYN request to all newly reachable hosts. */
     MssHost[] newReachable = view.getNewReachableHosts();
     for (int i=0; i < newReachable.length; i++) {
       if (log.isDebugEnabled())
         log.debug("newReachableHost: " + newReachable[i]);
       newReachable[i].scheduleTimeout();
     }
 
     /* Reset the SYN status for the new unreachable hosts. */
     MssHost[] newUnreachable = view.getNewUnreachableHosts();
     for (int i=0; i < newUnreachable.length; i++) {
       if (log.isDebugEnabled())
         log.debug("newUnreachableHost: " + newUnreachable[i]);
       newUnreachable[i].abortTimeout();
     }
 
     /* If the reachability set has changed, notify the upper layer. */
     DSView upperview = mssds.getUpperView();
     if (upperview.hasChanged()) {
       if (log.isDebugEnabled()) {
         log.debug("generateSuspect(): Upper view changed to: " + upperview);
       }
 
       mssuser.remoteSuspect(upperview.getReachableEndPoints(), 
         upperview.getNewReachableEndPoints(), 
         upperview.getNewUnreachableEndPoints(), 
         upperview.getNewIncarnationEndPoints());
       upperview.clear();
     }
 
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Timeout handling
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Update the topology information (due to this timeout), and send a
    *  IAMALIVE message to the local cluster.
    */
   private void iamAliveTimeout()
   {
     // Update topology info
     mssds.update();
 
     try {
       MsgIamAlive msg = MsgIamAlive.marshal(mssds.getAllFCEntry());
       localHost.getCluster().send(msg);
     } catch (IOException e) {
       log.error("Cannot marshal IAMALIVE flow control data into the message stream", e);
       return;
     }
 
     // Set a new timeout
     ehandler.setTimeout(routingTimeout, new ScheduledEvent(IAMALIVE));
     if (log.isDebugEnabled())
       log.debug("Sent IAMALIVE message; set timeout " + routingTimeout);
 
     // Generate suspect
     generateSuspect();
   }
 
 
   /**
    *  Send ROUTING topology information (flow control information) to
    *  all clusters, except the local cluster.  Also, notify the local
    *  cluster that we sent this information, using the SENT topology
    *  message.
    */
   private void routingTimeout()
   {
     log.assertLog(localHost.isReachable(),
                   "Localhost unreachable - alive value=" + localHost.getReachability() +
                   ", live cluster members=" + localHost.getCluster().getReachableCounter());
 
     /* Create a routing message */
     MsgRouting msg = null;
     try {
       msg = MsgRouting.marshal(mssds);
     } catch (IOException e) {
       log.error("Cannot marshal the ROUTING data into the message stream", e);
       return;
     }
 
     /*
      * Send the routing and flow control information to all clusters,
      * except the local cluster.
      */
     ClusterTable clustertable = mssds.getClusterTable();
     for (int i = 0, size = clustertable.size(); i < size; i++) {
       Cluster cluster = clustertable.get(i);
       /* Avoid sending to the local cluster */
       if (cluster.isLocal())
         continue;
 
       EndPoint clusterEP = cluster.getEndPoint();
       /*
        * To avoid creating a message Object for each cluster, the
        * split horizon technique is applied and disapplied (see
        * below) on the same Object.
        */
       msg.splitHorizonOn(clusterEP);
 
       /*
        * Update the flow control information for the topology
        * message; reusing the same <code>MsgRouting</code> object
        * for each cluster.
        */
       msg.setFCData(clusterEP);
 
       cluster.send(msg);
       if (log.isDebugEnabled())
         log.debug("Sent FLOW CONTROL and ROUTING info to cluster: " + clusterEP);
       msg.splitHorizonOff();
     }
 
     // Schedule a new timeout
     long timeout = RandomGenerator.getRoutingTimeout();
     lastRoutingScheduled = ehandler.setTimeout(timeout, new ScheduledEvent(ROUTING));
 
     /*
      * Notify the local cluster that a routing message was sent.
      */
     localHost.getCluster().forward(sentFrag, sentFrag.length);
   }
 
 
   /**
    *  Method to update the congestion information for all clusters in
    *  the distributed system.
    */
   private void congTimeout()
   {
     ClusterTable clustertable = mssds.getClusterTable();
     for (int i=0; i < clustertable.size(); i++) {
       MsgFlowSndrSide msgFlow = clustertable.get(i).getMsgFlow();
       if (msgFlow.isCongested())
         msgFlow.decCongWin();
       msgFlow.clearNackCount();
     }
     ehandler.setTimeout(congestionTimeout, new ScheduledEvent(CONGESTION));
   }
 
 
   private void nackTimeout(MsgNACK nack) 
   {
     MssHost src = nack.getSource();
     MsgFlowRcvrSide msgFlow = src.getMsgFlow();
 
     if (NACKSUPPRESSION) {
       if (msgFlow.isDelivered(nack.mid)) {
         log.warn("NACK timeout for already delivered msg (" + nack.mid + ")");
         msgFlow.removeScheduledEvent(nack.mid);
       } else {
         if (src.getCluster().isLocal()) {
           if (log.isDebugEnabled())
             log.debug("Sending NACK for msg (" + nack.mid + ") to cluster " + localHost.getCluster());
           localHost.getCluster().send(nack);
         } else {
           if (log.isDebugEnabled())
             log.debug("Sending NACK for msg (" + nack.mid + ") to " + src);
           src.unicastRouteSend(nack);
           try {
             MsgNACK sentNack = MsgNACK.marshal(SENTNACK, src, nack.mid);
             if (log.isDebugEnabled())
               log.debug("Sending SENTNACK to cluster " + localHost.getCluster());
             localHost.getCluster().send(sentNack);
           } catch (IOException e) {
             log.error("Unable to marshal SENTNACK for local cluster: " + localHost.getCluster(), e);
             return;
           }
         }
         ScheduledEvent eventNack = msgFlow.getScheduledEvent(nack.mid);
         if (eventNack == null) {
           log.warn("No scheduled NACK event for " + nack.mid);
         } else {
           /* Delete the scheduled NACK timeout */
           ehandler.abortTimeout(eventNack);
           /*
            * Reschedule a new timeout for the NACK (to allow additional
            * time for receiving the resent message)
            */
           long randomTimeout = RandomGenerator.getRandomTimeout(src.getTimeout());
           ehandler.setTimeout(randomTimeout, eventNack);
         }
       }
     } else if (nack.mid <= msgFlow.getLastMsgDlvr()) {
       log.warn("NACK timeout for already delivered msg (" + nack.mid + ")");
       msgFlow.removeScheduledEvent(nack.mid);
     } else {
       if (log.isDebugEnabled())
         log.debug("Sending NACK for msg (" + nack.mid + ") to " + src);
       src.unicastRouteSend(nack);
       // remove the old nack from the list
       msgFlow.removeScheduledEvent(nack.mid);
       // set again the nack timeout
       ScheduledEvent eventNack = new ScheduledEvent(nack.getNACKType(), nack);
       ehandler.setTimeout(src.getTimeout(), eventNack);
       msgFlow.putScheduledEvent(nack.mid, eventNack);
     }
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Message receiving section
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Handle the receiption of a JG message fragment.  This means to
    *  forward it to the local cluster, if the fragment has its broadcast
    *  flag set to true.  Otherwise, the message fragment will be passed
    *  to the <code>MsgCntrl</code> for delivery or queuing.
    *
    *  @see jgroup.relacs.mss.MsgCntrl#msgReceive
    *  @param fragment
    *    The message fragment to add to deliver or add to the message
    *    queue.
    *  @return
    *    False if the message could not be deliverd due to uninitialized
    *    datastructures; the message flow.
    */
   private boolean handleJGInfo(FragmentHeader fragment)
   {
     if (fragment.isBroadcast()) {
       /*
        * The fragment has to be broadcast within the local cluster.
        * This means that I'm cluster leader for this message, and I must
        * forward the message to all members of my cluster.
        */
       Cluster localCluster = ClusterTable.getLocalCluster();
       if (log.isDebugEnabled())
         log.debug("Broadcasting msg fragment (" + fragment + ") within local cluster "
                   + localCluster.getEndPoint());
       fragment.broadcastOff();
       localCluster.forward(fragment.getFragment(), fragment.getFragmentLength());
       /*
        * Note that we don't deliver the message at this point, since we
        * will receive it from the above send in the local cluster; note
        * that before invoking the forward() on the local cluster, we turn
        * off broadcasting.  This causes the next delivery to skip to the
        * else part of this if-statement when received again.
        */
     } else {
       if (log.isDebugEnabled())
         log.debug("Received msg (" + fragment.fragId + ") from " + fragment.getSender());
       return msgCntrl.msgReceive(fragment);
     }
     return true;
   }
 
 } // END Mss