/*
    * 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.config;
  
  import static jgroup.relacs.simulator.SocketStatus.SOCKET_STATUS;
  import static jgroup.util.log.ConnectionPatternEvent.Type.Merge;
  import static jgroup.util.log.ConnectionPatternEvent.Type.Partition;
  
  import java.io.Externalizable;
  import java.io.IOException;
  import java.io.ObjectInput;
  import java.io.ObjectOutput;
  import java.net.InetAddress;
  import java.net.UnknownHostException;
  import java.rmi.RemoteException;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.Map;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.locks.Condition;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantLock;
  
  import jgroup.core.EndPoint;
  import jgroup.core.registry.BootstrapRegistry;
  import jgroup.relacs.simulator.SocketStatus;
  import jgroup.relacs.types.EndPointImpl;
  import jgroup.util.Abort;
  import jgroup.util.Network;
  import jgroup.util.log.ConnectionPatternEvent;
  import jgroup.util.log.Eventlogger;
  
  import org.apache.log4j.Logger;
  
  /**
   *  Class for holding domain information.
   *
   *  @author Hein Meling
   *  @since Jgroup 1.2
   */
  public class Domain
    extends EndPointImpl
    implements Cloneable, Comparable, Externalizable
  {
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Logger
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Obtain logger for this class */
    private static final Logger log = Logger.getLogger(Domain.class);
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Static fields
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    private static final long serialVersionUID = -2367179058534386338L;
  
    /** In this JVM there can be only one local multicast address. */
    private static InetAddress localMcastAddress = null;
  
    private static final String DEFAULT_MCAST_ADR = "226.7.8.9";
  
    private static final int DEFAULT_MCAST_PORT   = 61341;
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Fields
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Domain name */
    private String domainName;
  
    /** Map of hosts contained within this domain */
    private HostSet hosts = new HostSet();
  
    /** Map for storing content associated with this <code>Domain</code> object */
    private Map<String,Object> content = new HashMap<String,Object>();
  
    /**
     *  The number of daemons in this domain.  The default is that all hosts
     *  should contain a Jgroup daemon; this is indicated by value -1.
    */
   private int numOfDaemons = -1;
 
   /** The number of replicas in this domain; initially zero. */
   private int replicaCount = 0;
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Constructors
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    * Constructor for deserialization.
    */
   public Domain()
   {
   }
 
   /**
    *  Constructs a domain from another domain object.  If the <code>copyHosts</code>
    *  parameter is set to false, no hosts will be copied from the domain object.
    * 
    *  @param domain the domain to copy
    *  @param copyHosts if true, the internal host set of the provided domain
    *     is also copied; otherwise the internal host set will be empty.
    */
   public Domain(Domain domain, boolean copyHosts)
   {
     this(domain.domainName, domain.address, domain.port, domain.numOfDaemons);
     if (copyHosts)
       this.hosts = (HostSet) domain.hosts.clone();
   }
 
   public Domain(String domainName)
     throws UnknownHostException
   {
     this(domainName, InetAddress.getByName(DEFAULT_MCAST_ADR), DEFAULT_MCAST_PORT, -1);
   }
 
   public Domain(String domainName, String mcastAdr, int mcastPort)
     throws UnknownHostException
   {
     this(domainName, InetAddress.getByName(mcastAdr), mcastPort, -1);
   }
 
   public Domain(String domainName, String mcastAdr, int mcastPort, int jdaemons)
     throws UnknownHostException
   {
     this(domainName, InetAddress.getByName(mcastAdr), mcastPort, jdaemons);
   }
 
   private Domain(String domainName, InetAddress mcastAddress, int mcastPort, int jdaemons)
   {
     super(mcastAddress, mcastPort);
     this.numOfDaemons = jdaemons;
     if (jdaemons < -1) {
       throw new IllegalArgumentException("Invalid number Jgroup daemons");
     }
 
     this.domainName = domainName;
     /*
      * We cannot rely on the local detection in EndPointImpl for
      * multicast addresses.  Thus, we require that local has to be set
      * using the <code>setLocal()</code> method.
      */
     local = false;
     if (!mcastAddress.isMulticastAddress())
       throw new IllegalArgumentException("Domain IP address is not a multicast address.");
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Overridden EndPointImpl methods from the EndPoint interface
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Returns true if this is a local domain.
    */
   public boolean isLocal()
   {
     if (localMcastAddress == null)
       throw new IllegalStateException("Local domain multicast address not initialized");
     return local;
   }
 
 
   /**
    *  Returns true always, since a domain endpoint must be a multicast
    *  endpoint.  There is a check for this in the class constructor.
    */
   public boolean isMulticastEndPoint()
   {
     return true;
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Domain interface methods
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Set the local flag for this domain.
    */
   public void setLocal(boolean local)
   {
     this.local = local;
     if (local)
       localMcastAddress = address;
   }
 
 
   /**
    *  Returns the domain name.
    */
   public String getName() 
   {
     return domainName;
   }
 
   public EndPoint getEndpoint()
   {
     return new EndPointImpl(super.address, super.port);
   }
 
   /**
    *  Returns the number of hosts contained in this domain.
    */
   public int size() 
   {
     return hosts.size();
   }
 
 
   /**
    *  Returns true if there are no hosts in this domain; false otherwise.
    */
   public boolean isEmpty()
   {
     return hosts.isEmpty();
   }
 
 
   /**
    *  Returns the <code>HostSet</code> for the hosts in this domain.
    */
   public HostSet getHostSet()
   {
     return hosts;
   }
 
   
   /**
    *  Add a host to this domain.
    *
    *  @param host
    *    The host to add to this domain.
    *  @return 
    *    True if the <code>Domain</code> did not already contain this host.
    */
   public boolean addHost(Host host)
   {
     return hosts.addHost(host);
   }
 
 
   /**
    *  Retrieve the value associated with this domain for the given key.
    */
   public Object get(String key)
   {
     return content.get(key);
   }
 
 
   /**
    *  Associate the given value with this domain under the given key.
    */
   public Object put(String key, Object value)
   {
     return content.put(key, value);
   }
 
 
   /**
    *  Returns true if all hosts in this domain should have a daemon.
    */
   public boolean allDaemons()
   {
     return (numOfDaemons == -1);
   }
 
 
   /**
    *  Returns true if this domain should not have any daemons.
    */
   public boolean hasNoDaemons()
   {
     return (numOfDaemons == 0);
   }
 
 
   /**
    *  Returns the number of daemons that should reside in this domain.
    *  A value of -1 indicates that all hosts in this domain should have
    *  a Jgroup daemon.  A value of 0 indicates that this domain have no
    *  daemons, and that we should look in other domains for a daemon.
    */
   public int numOfDaemons()
   {
     return numOfDaemons;
   }
 
   //FIXME these three methods should probably be part of an aspect??!
   public void decReplicaCount()
   {
     replicaCount--;
   }
 
   public void incReplicaCount()
   {
     replicaCount++;
   }
 
   public int getReplicaCount()
   {
     return replicaCount;
   }
 
   public void setReplicaCount(int newReplicaCount)
   {
     this.replicaCount = newReplicaCount;
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Cloneable interface
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Returns a shallow copy of this <code>Domain</code> instance.
    *  (The elements themselves are not copied.)
    *
    *  @return 
    *    a clone of this <code>Domain</code> instance.
    */
   @SuppressWarnings("unchecked")
   public Object clone()
   {
     try {
       Domain newDomain = (Domain) super.clone();
       newDomain.content = (HashMap) ((HashMap) content).clone();
       newDomain.hosts = (HostSet) hosts.clone();
       return newDomain;
     } catch (CloneNotSupportedException e) {
       /* Cannot happen since HashMap and HostSet supports clone. */
       throw new Abort("Internal error", e);
     }
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Comparable interface
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    * Compares this object with the specified object for ordering the objects.
    * The replicaCount is the primary ordering value; the size of the domain 
    * (the number of hosts in the domain) is the secondary ordering value, while
    * the multicast IP and port number is the third ordering value.  Note that
    * we rank domains with the largest number of hosts first.
    */
   public int compareTo(Object obj)
   {
     if (this == obj) {
       return 0;
     } else {
       Domain domain = (Domain) obj;
       if (replicaCount < domain.replicaCount) {
         return -1;
       } else if (replicaCount == domain.replicaCount) {
         if (size() < domain.size()) {
           // Note that we do the reverse ordering for domain size;
           // that is the highest value should be first
           return 1;
         } else if (size() == domain.size()) {
           // If the two domains have the same number of hosts,
           // use the IP and port number ordering defined by the EndPointImpl
           return super.compareTo(obj);
         } else {
           return -1;
         }
       } else {
         return 1;
       }
     }
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Override methods in Object/EndPointImpl
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Returns a string representation of this object
    */
    public String toString() 
    {
      return toString(false);
    }
 
   /**
    *  Returns a string representation of this object
    */
    public String toString(boolean full) 
    {
      StringBuilder buf = new StringBuilder();
 //     buf.append("[");
 //     buf.append(domainName);
 //     buf.append(": ");
      buf.append(super.toString());
 //     buf.append("]");
      if (full) {
        for (Iterator<Host> i = hosts.iterator(); i.hasNext();) {
          buf.append("\n  ");
          buf.append((i.next()).toString());
        }
      }
      return buf.toString();
    }
 
 
    ////////////////////////////////////////////////////////////////////////////////////////////
    //  Marshaling/unmarshaling methods
    ////////////////////////////////////////////////////////////////////////////////////////////
 
    /**
     *  Restores the content of this object from the marshalled data contained
     *  in the specified input stream.
     * 
     *  @param in the stream to be read
     */
    public void readExternal(ObjectInput in)
      throws ClassNotFoundException, IOException
    {
      super.readExternal(in);
      replicaCount = in.readInt();
      hosts = (HostSet) in.readObject();
      content = (Map) in.readObject();
    }
 
    /**
     *  Marshals the content of this object to the specified output stream.
     * 
     *  @param out the stream to be written
     */
    public void writeExternal(ObjectOutput out)
      throws IOException
    {
      super.writeExternal(out);
      out.writeInt(replicaCount);
      out.writeObject(hosts);
      out.writeObject(content);
    }
 
 
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Methods for the partition simulator
    //
    // //FIXME these methods should be placed in an aspect instead
    ////////////////////////////////////////////////////////////////////////////////////////////
 
    /** List of socket status objects that should be committed */
    private static final Map<Host, SocketStatus> statusMap = new HashMap<Host, SocketStatus>();
 
    /** The connection event to be committed */
    private static ConnectionPatternEvent connEvent = null;
 
    /** The number of remaining commits in the current injection campaign */
    private static volatile int remainingCommits;
 
    /** Lock and condition to exclude concurrent access */
    private static final Lock lock = new ReentrantLock();
    private static final Condition completed = lock.newCondition();
 
 
    /**
     * Inject a partition on this domain so that it becomes unreachable from the given
     * domain.
     * 
     * This method will only configure the reachability pattern, and thus it must
     * be followed by an invocation of the <code>commit()</code> method below.
     */
    public void partition(Domain domain)
    {
      if (connEvent == null)
        connEvent = new ConnectionPatternEvent();
      connEvent.addEvent(Partition, this, domain);
      boolean success = setReachability(0, domain);
      if (!success) {
        // Clear and retry
        clearStatus();
        if (log.isDebugEnabled())
          log.debug("Retrying partition: " + domain);
        success = setReachability(0, domain);
        if (!success)
          log.error("FAILED TO RETRY PARTITION: " + domain);
      }
    }
 
    /**
     * Inject a merge on this domain so that it becomes reachable from the given domain.
     * 
     * This method will only configure the reachability pattern, and thus it must
     * be followed by an invocation of the <code>commit()</code> method below.
     */
    public void merge(Domain domain)
    {
      if (connEvent == null)
        connEvent = new ConnectionPatternEvent();
      connEvent.addEvent(Merge, this, domain);
      boolean success = setReachability(100, domain);
      if (!success) {
        // Clear and retry
        clearStatus();
        if (log.isDebugEnabled())
          log.debug("Retrying merge: " + domain);
        success = setReachability(100, domain);
        if (!success)
          log.error("FAILED TO RETRY MERGE: " + domain);
      }
    }
 
    /**
     * Commit the currently configured reachability pattern, essentially activating
     * the pattern.
     * 
     * Calls to this method <i>must</i> only occur after a call to either the
     * <code>partition()</code> or <code>merge()</code> method, otherwise an
     * <code>IllegalStateException</code> is thrown.  It cannot be called multiple
     * times per call to <code>partition()</code> or <code>merge()</code>.
     */
    public static void commit(String pattern)
    {
      if (connEvent == null)
        throw new IllegalStateException("Cannot call commit() before partition() or merge()");
      ConnectionPatternEvent preEvent =
        new ConnectionPatternEvent("PRE-COMMIT", pattern, connEvent.getConnectionEvents());
      if (Eventlogger.ENABLED)
        Eventlogger.logEventFlush(preEvent);
 
      // initialize the remaining commits counter
      lock.lock();
      try {
        remainingCommits = statusMap.size();
      } finally {
        lock.unlock();
      }
 
      /*
       * Iterate over the stored socket status objects and perform commit
       * once for each host.
       */
      for (final Map.Entry<Host, SocketStatus> entry : statusMap.entrySet()) {
        final Host host = entry.getKey();
        Thread commitThread = new Thread("Commit-" + host) {
          public void run() {
            if (log.isDebugEnabled())
              log.debug("Committing: " + host);
            try {
              entry.getValue().commit();
              if (log.isDebugEnabled())
                log.debug("Have committed: " + host);
            } catch (RemoteException e) {
              log.warn("Failed to commit: " + host, e);
            } finally {
              lock.lock();
              try {
                // decrement the remaining commits counter
                remainingCommits--;
                /*
                 * Only signal when all commit threads have completed.
                 * This reduces unecessary context-switching.
                 */
                if (remainingCommits == 0)
                  completed.signal();
              } finally {
                lock.unlock();
              }
            }
          }
        };
        commitThread.setDaemon(true);
        commitThread.setPriority(Thread.MAX_PRIORITY);
        commitThread.start();
      }
 
      // Check if all commit threads have completed
      lock.lock();
      try {
        // We still need a while loop here since the await call can have spurious wakeups.
        while (remainingCommits > 0) {
          log.info("Commit threads remaining: " + remainingCommits);
          try {
            boolean timeout = !completed.await(5, TimeUnit.SECONDS);
            if (timeout && remainingCommits > 0)
              throw new RuntimeException("Timeout during commit; threads remaining: " + remainingCommits);
          } catch (InterruptedException e) {
            log.warn("Waiting for commit thread interrupted (should not happen)", e);
          }
        }
      } finally {
        lock.unlock();
      }
 
      ConnectionPatternEvent postEvent =
        new ConnectionPatternEvent("POST-COMMIT", pattern, connEvent.getConnectionEvents());
      if (Eventlogger.ENABLED)
        Eventlogger.logEventFlush(postEvent);
      if (log.isDebugEnabled())
        log.debug("ALL COMMITTED");
      /*
       * Reset the connection pattern to ensure that calls to commit() are
       * always proceeded by calls to either partition() or merge() first.
       */
      connEvent = null;
    }
 
   /**
    * Clear out the status map to avoid that the socket status remote
    * reference objects are propagated from one experiment to the next.
    */
   public void clearStatus()
   {
     statusMap.clear();
   }
 
   /**
    * Set the reachability between this domain and the given domain.
    * 
    * @param reachability The reachability percentage to set for this domain
    * @param domain The domain for which this domain the reachability is to be updated.
    * @return True if all nodes were successfully updated with the given
    *   reachability percentage.  False is returned if updating failed on at
    *   least one of the nodes.
    */
   private boolean setReachability(int reachability, Domain domain)
    {
     if (!domain.equals(this)) {
       for (Host h1 : hosts) {
         StringBuilder buf = new StringBuilder();
         buf.append(h1.getHostName());
         buf.append("(");
         buf.append(reachability);
         buf.append("): ");
         try {
           SocketStatus h1status = getStatus(h1);
           for (Host h2 : domain.getHostSet()) {
             h1status.setStatus(h2.getAddress(), reachability);
             buf.append(h2.getHostName());
             buf.append(" ");
             // Ensure to establish also the reverse reachability relation
             SocketStatus h2status = getStatus(h2);
             h2status.setStatus(h1.getAddress(), reachability);
           }
           if (log.isDebugEnabled())
             log.debug(buf);
         } catch (Exception e) {
           log.warn("Failed to reach host", e);
           // Abort early if one failed
           return false;
         }
       }
     } else {
       log.warn(domain + " is the same as this");
     }
      // All successful
      return true;
    }
 
    private SocketStatus getStatus(Host host)
      throws Exception
    {
      SocketStatus status = statusMap.get(host);
      if (status == null) {
        try {
          status = (SocketStatus) host.lookup(SOCKET_STATUS);
          if (log.isDebugEnabled())
            log.debug("SocketStatus for " + host + ": " + status);
        } catch (Exception e) {
          log.warn(host.getHostName() + " is not available", e);
          throw e;
        }
        statusMap.put(host, status);
      }
      return status;
    }
 
 } // END Domain