/*
    * 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.util.Timer;
  import java.util.TimerTask;
  
  import jgroup.core.ConfigurationException;
  import jgroup.core.EndPoint;
  import jgroup.relacs.config.TransportConfig;
  
  import org.apache.log4j.Logger;
  
  /**
   *  The <code>MssHost</code> class
   *
   *  @author Salvatore Cammarrata
   *  @author Hein Meling
   *  @since Jgroup 0.5
   */
  public class MssHost
    implements MssConstants, MssTag
  {
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Logger
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Obtain logger for this class */
    private static final Logger log = Logger.getLogger(MssHost.class);
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Fields
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Configuration parameters */
    private TransportConfig config;
  
    /** Datalink layer */
    private NI ni;
  
    /** Event scheduler */
    private EventHandler ehandler;
  
    /** The endpoint used to identify a host (and port) */
    private EndPoint endpoint;
  
    /** Incarnation number */
    private int incarnationId;
  
    /** Index in the HostTable */
    private int hostIndex;
  
    /** Index in the Cluster */
    private int clusterIndex; 
  
    /** The local cluster */
    private Cluster cluster;
  
    /** Roundtrip timeout */
    private int rtt;
  
    /** Standard deviation */
    private int dev;
  
    /** Next timeout */
    private int timeout;
  
    /**
     *  Set to max TTL when the member is considered reachable.  This
     *  value is decremented for every TOPOLOGY timeout.  If it reaches
     *  zero the member is considered unreachable.
     */
    private int alive;  
  
    /**
     *  Last reachability information.  Used to determine if the
     *  reachability perception is changed.
     */
    private int wasAlive;
  
   /** Flow Control information */
   private MsgFlowRcvrSide msgFlow;
 
   /** The most recent synchronization message */
   private MsgSYN syn = null;
 
   /** The timer used to resend the QSYN message (if no ASYN is received) */
   private Timer synTimer = null;
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Constructors
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Builds a host object identified by <code>hostEndPoint</code>.
    */
   public MssHost(TransportConfig config, EndPoint hostEndPoint, NI ni, Cluster cluster, EventHandler ehandler)
     throws ConfigurationException
   {
     this.ni = ni;
     this.config = config;
     this.endpoint = hostEndPoint;
     this.ehandler = ehandler;
 
     if (log.isDebugEnabled())
       log.debug("MssHost:<init>: " + endpoint + ", " + cluster);
 
     if (endpoint.isLocal()) {
       alive = wasAlive = config.getMaxTTL();
       incarnationId = (int) (System.currentTimeMillis()/1000);
     } else {
       alive = wasAlive = UNREACHABLE;
       incarnationId = UNDEF;
     }
 
     /*
      *  Timeout initialization
      */
     updateTimeout(100); //FIXME constant value?
 
     /*
      *  Flow control initialization
      */
     msgFlow = new MsgFlowRcvrSide(config);
 
     /*
      *  Cluster initialization
      */
     this.cluster = cluster;
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *
    */
   public EndPoint getEndPoint()
   {
     return endpoint;
   }
 
   /**
    *
    */
   public boolean isLocal()
   {
     return endpoint.isLocal();
   }
 
   /**
    *
    */
   public int getIncarnationId()
   {
     return incarnationId;   
   }
 
   /**
    *
    */
   public void setIncarnationId(int incarnationId)
   {
     this.incarnationId = incarnationId;   
   }
 
 
   /**
    *  Returns true if this host is in the given cluster; otherwise false
    *  is returned.
    */
   public boolean isIn(Cluster theCluster)
   {
     return theCluster.equals(cluster);
   }
 
 
   /**
    *
    */
   public Cluster getCluster() 
   {
     return cluster;
   }
 
 
   /**
    *  Get this host's position (index) in the cluster.
    */
   public int getClusterIndex()
   {
     return clusterIndex;
   }
 
 
   /**
    *  Set this host's position (index) in the cluster.
    */
   public void setClusterIndex(int index)
   {
     clusterIndex = index;
   }
 
 
   /**
    *  Set this host's position (index) in the hosttable.
    */
   public void setIndex(int index)
   {
     hostIndex = index;
   }
 
 
   /**
    *  Get this host's position (index) in the hosttable.
    */
   public int getIndex()
   {
     return hostIndex;
   }
 
 
   /**
    *  Returns the timeout value associated with this host.
    */
   public int getTimeout()
   {
     return timeout;
   }
 
 
   /**
    *  Recompute the round-trip timeout, standard deviation and timeout
    *  values associated with this host.
    *
    *  @param t  an <code>int</code> value
    */
   public void updateTimeout(int t)
   {
     int alfan = config.getAlfan();
     int alfad = config.getAlfad();
     rtt = (rtt * alfan + t * (alfad - alfan)) / alfad;
     dev = (dev * alfan + (rtt > t ? rtt - t : t - rtt) * (alfad - alfan)) / alfad;
     timeout = rtt + dev * 4;
     if (timeout < 1)
       timeout = 1;
     log.assertLog(timeout > 3, "Timeout is too small: " + timeout + ", t=" + t
         + ", rtt=" + rtt + ", dev=" + dev + ", alphaN=" + alfan + ", alphaD=" + alfad);
     if (log.isDebugEnabled())
       log.debug("rtt: " + rtt + ", dev: " + dev + ", timeout: " + timeout);
   }
 
 
   /**
    *  Aborts the timeout that has been set up for performing periodic
    *  synchronization ping.
    */
   public void abortTimeout()
   {
     if (synTimer != null) {
       synTimer.cancel();
       syn = null;
       synTimer = null;
       if (log.isDebugEnabled())
         log.debug("Aborted SYN-timer for " + endpoint);
       if (log.isDebugEnabled())
         printReachParams();
     }
   }
 
 
   /**
    *  Schedule a timer for the synchronization (QSYN) message sent
    *  to this member.  The initial QSYN message is sent after no delay.
    */
   public void scheduleTimeout()
   {
     try {
       syn = MsgSYN.marshal(QSYN, this);
     } catch (IOException e) {
       log.error("Unable to marshal QSYN message for: " + this, e);
       return;
     }
     TimerTask synSender = new TimerTask() {
       public void run() {
         if (syn != null) {
           unicastRouteSend(syn);
           if (log.isDebugEnabled())
             log.debug("Sent " + syn + " for " + endpoint);
         }
       }
     };
     synTimer = new Timer("SYN-timer" + endpoint, true);
     synTimer.schedule(synSender, 0, timeout);
     if (log.isDebugEnabled())
       log.debug("Scheduled SYN-timer for " + endpoint);
   }
 
 
   /**
    *  Checks if the given message identifier corresponds to the last SYN
    *  message sent from this host.  True is returned if this is the
    *  case; false otherwise.
    *
    *  @param msgId
    *    The message identifier to check if was the last SYN message sent.
    *  @return 
    *    True is returned if the last SYN message sent by this <code> MssHost 
    *    </code> is equal to the given message identifier; otherwise false
    *    is returned.
    */
   public boolean checkSynId(int msgId)
   {
     if (syn != null) {
       int nonce = syn.getMid();
       if (log.isDebugEnabled())
         log.debug(endpoint + " received ASYN with " + msgId + "=?" + nonce);
       return nonce == msgId;
     }
     return false;
   }
 
 
   /**
    *  Returns true if this host is considered reachable within the given
    *  <code>threshold</code>.
    */
   public boolean isReachable(int threshold)
   {
     return (alive >= threshold);
   }
 
 
   /**
    *  Returns true if this host is considered reachable.  That is with a
    *  reachability <code>threshold</code> equal to
    *  <code>MINIMUM_ALIVE_VALUE</code>.
    */
   public boolean isReachable()
   {
     return isReachable(MINIMUM_ALIVE_VALUE);
   }
 
 
   /**
    *  Returns true if this host was reachable on prior to the last
    *  reachability update.
    */
   public boolean wasReachable()
   {
     return (wasAlive != UNREACHABLE);
   }
 
 
   /**
    *  Returns the current reachability value.
    */
   int getReachability()
   {
     return alive;
   }
 
 
   /**
    *  Set the alive counter to the specified value and return the
    *  previous reachability status.
    *
    *  @param newAlive
    *    The new reachability value to be set.
    *  @return 
    *    True is returned if this host was reachable; false otherwise.
    */
   boolean pingOK(int newAlive)
   {
     wasAlive = alive;
     alive = newAlive;
     if (log.isDebugEnabled()) 
       log.debug("MssHost:pingOK:" + endpoint + ": alive=" + alive + ", wasAlive=" + wasAlive);
     return (wasAlive != UNREACHABLE);
   }
 
 
   /**
    *  Set the alive counter to the maximum TTL value and return the
    *  previous status.
    *
    *  @return 
    *    True is returned if this host was reachable; false otherwise.
    */
   boolean pingOK()
   {
     return pingOK(config.getMaxTTL());
   }
     
 
   /**
    *  Decrements the alive counter and returns true if the member is
    *  still reachable; false is returned otherwise.
    */
   boolean pingKO()
   {
     wasAlive = alive;
     alive = (alive == UNREACHABLE ? alive : alive - 1);
     if (log.isDebugEnabled()) 
       log.debug("MssHost:pingKO:" + endpoint + ": alive=" + alive + ", wasAlive=" + wasAlive);
     return (alive != UNREACHABLE);
   }
 
 
   /**
    *  Mark this host as unreachable.
    */
   void setAsUnreachable()
   {
     wasAlive = alive;
     alive = UNREACHABLE;
   }
 
 
   /**
    *  Mark this host as reachable.
    */
   public void setAsReachable()
   {
     wasAlive = alive;
     alive = config.getMaxTTL();
   }
 
 
   /**
    * Flush the sent queue up until the last message ACKed.
    */
   public void flush()
   {
     /* Clear the queues */
     cluster.getMsgFlow().flush(msgFlow.clusterWindow.getLastMsgDelivered());
   }
 
 
   /**
    * Flush the sent queue up until the last message ACKed.
    *
    * @param lastMsgAcked The last message ACKed.
    */
   public void flush(int lastMsgAcked)
   {
     /* Clear the queues */
     cluster.getMsgFlow().flush(lastMsgAcked);
   }
 
 
   /**
    *
    */
   public MsgFlowRcvrSide getMsgFlow()
   {
     return msgFlow;   
   }
 
 
   /**
    *
    */
   void resetMsgFlow() 
   {
     msgFlow.reset();
   }
 
 
   /**
    *
    */
   void resetMsgFlow(int mid)
   {
     msgFlow.reset(mid);
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods for debugging
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   int monitorAlive = 0;
   long[] measure = new long[10];
   int position = 0;
 
   /**
    * Method used to print the reachability and timeout parameters associated
    * with this host, if they have changed since its last invocation.
    */
   public void printReachParams()
   {
     if (monitorAlive != alive) {
       monitorAlive = alive;
       if (alive == config.getMaxTTL()) {
         // reset measurements
         position = 0;
       } else {
         // read clock and compute diff to the first measurement
         long now = System.currentTimeMillis();
         long diff = now - measure[0];
         if (log.isDebugEnabled()) {
           log.debug(endpoint.toString() + ": alive=" + alive + ", diff=" + diff);
         }
         measure[position++] = now;
       }
     }
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods for sending to this host
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   void sendNACK(int mid)
   {
     assert !isLocal() : "Trying to send REMOTENACK to myself";
     MsgNACK nack = null;
     try {
       nack = MsgNACK.marshal(REMOTENACK, this, mid);
     } catch (IOException e) {
       log.warn("Unable to marshal REMOTENACK for host: " + this, e);
       return;
     }
     ScheduledEvent event = new ScheduledEvent(REMOTENACK, nack);
 
     if (NACKSUPPRESSION) {
       long random_timeout = RandomGenerator.getRandomTimeout(getTimeout());
       ehandler.setTimeout(random_timeout, event);
       msgFlow.putScheduledEvent(nack.mid, event);
     } else {
       ehandler.setTimeout(getTimeout(), event);
       msgFlow.putScheduledEvent(nack.mid, event);
       // Comment the following line to delay the ack sending
 //      unicastRouteSend(nack);
     }
   }
 
 
   void simpleSend(byte[] buf, int buflen)
   {
     assert !isLocal() : "Trying to send to myself";
     if (log.isDebugEnabled())
       log.debug("MssHost:simpleSend: send to " + endpoint);
     ni.send(endpoint, buf, buflen);
   }
 
 
   /**
    *
    */
   void unicastRouteSend(Msg msg)
   {
     assert !isLocal() : "Trying to send to myself: " + msg;
     if (log.isDebugEnabled())
       log.debug("MssHost:unicastRouteSend: send to " + endpoint);
 
     if (cluster.directlyConnected()) {
 
       MsgCntrl msgCntrl = cluster.getControl();
       for (FragmentIterator frag = msg.iterator(msgCntrl); frag.hasNext(); ) {
         /* Generate and send a fragment */
         byte[] buffer = frag.next(false);
         int bufLen = frag.fragmentLength();
         if (log.isDebugEnabled())
           log.debug("send: Sending msg fragment (" + frag.getFid() + ") to host " + endpoint);
         ni.send(endpoint, buffer, bufLen);
       }
 
     } else {
       log.warn("Cluster appears not to be directly connected (sending through cluster instead): " + cluster);
       cluster.send(msg);
 
     }
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods from Object
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Returns a hash code value for this host.
    */
   public int hashCode()
   {
     return endpoint.hashCode();
   }
 
 
   /**
    *  Returns a string representation of this object.
    */
   public String toString()
   {
     return endpoint.toString();
   }
 
 
   /**
    *  Compares two objects for content equality.
    *
    *  @param obj the object to compare with
    *  @return  true if these objects are equal; false otherwise.
    */
   public boolean equals(Object obj)
   {
     if (this == obj)
       return true;
     else
       if (obj instanceof MssHost)
         return endpoint.equals(((MssHost)obj).endpoint);
       else
         return false;
   }
 
 } // END MssHost