/*
    * 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.arm;
  
  import java.io.BufferedReader;
  import java.io.File;
  import java.io.InputStream;
  import java.io.InputStreamReader;
  import java.io.OutputStream;
  import java.io.PrintWriter;
  import java.lang.reflect.InvocationTargetException;
  import java.lang.reflect.Method;
  import java.lang.reflect.Modifier;
  import java.net.DatagramSocket;
  import java.net.SocketException;
  import java.rmi.AlreadyBoundException;
  import java.rmi.Remote;
  import java.rmi.RemoteException;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Map;
  import java.util.Set;
  
  import jgroup.core.ConfigManager;
  import jgroup.core.ConfigurationException;
  import jgroup.core.JgroupException;
  import jgroup.core.arm.ExecException;
  import jgroup.core.arm.ExecService;
  import jgroup.core.arm.ReplicationManager;
  import jgroup.core.registry.BootstrapRegistry;
  import jgroup.core.registry.DependableRegistry;
  import jgroup.core.registry.RegistryFactory;
  import jgroup.relacs.config.AppConfig;
  import jgroup.relacs.config.ClassData;
  import jgroup.relacs.config.DistributedSystemConfig;
  import jgroup.relacs.config.Host;
  import jgroup.relacs.daemon.DaemonInteraction;
  import jgroup.util.Abort;
  import jgroup.util.Network;
  import jgroup.util.log.Eventlogger;
  import jgroup.util.log.ReplicaEvent;
  import static jgroup.util.log.ReplicaEvent.Type.Removed;
  import static jgroup.util.log.ReplicaEvent.Type.Shutdown;
  import net.jini.export.Exporter;
  import net.jini.jeri.BasicILFactory;
  import net.jini.jeri.BasicJeriExporter;
  import net.jini.jeri.tcp.TcpServerEndpoint;
  
  import org.apache.log4j.LogManager;
  import org.apache.log4j.Logger;
  
  
  /**
   *  ExecDaemon implements the ExecService allowing a replica manager or
   *  other entity to start replicas on hosts running the ExecDaemon.
   *
   *  Note that an important property of ExecDaemon is that it does not
   *  have any state (except the map of local processes).  Thus the rest
   *  of the distributed system does not depend on its state.
   *
   *  @author Hein Meling
   *  @since Jgroup 1.2
   */
  public class ExecDaemon
    implements ExecService
  {
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Logger
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Obtain logger for this class */
    private static final Logger log = Logger.getLogger(ExecDaemon.class);
  
  
    ////////////////////////////////////////////////////////////////////////////////////////////
    // Fields
    ////////////////////////////////////////////////////////////////////////////////////////////
  
    /** Map of replica classes running on this host */
    private Map<ClassData,ReplicaThread> replicaMap =
      Collections.synchronizedMap(new HashMap<ClassData,ReplicaThread>());
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Static fields and block
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /** Variable holding JVM executable, classpath and system properties */
   private static StringBuilder jvmCmd      = new StringBuilder();
 
   /** The application starter class name */
   private static final String  appStarter  = ApplicationStarter.class.getName();
 
   /** The Runtime environment for this VM */
   private static final Runtime rt          = Runtime.getRuntime();
 
   /** Determine the local host name */
   private static final String  thisHost    = Network.getLocalHostName();
 
   /** Determine the local host name */
   private static final String  thisMachine = Network.getLocalMachineName();
 
 
   static {
     String s = File.separator;
     jvmCmd.append(System.getProperty("java.home"));
     jvmCmd.append(s);
     jvmCmd.append("bin");
     jvmCmd.append(s);
     jvmCmd.append("java");
     /*
      * Jgroup properties to pass on to JVMs started from the ExecDaemon.
      */
     String[] jgroupProperties = new String[] {
       "java.library.path",
       "jgroup.simulator",
       "jgroup.log.dir",
       "jgroup.log.config",
       "jgroup.log.msgcontent",
       "jgroup.log.measurements",
       "jgroup.system.config",
       "jgroup.system.services",
       "jgroup.system.applications",
       "java.util.logging.config.file",
     };
     for (int i = 0; i < jgroupProperties.length; i++) {
       String property = System.getProperty(jgroupProperties[i]);
       if (property != null && property.length() > 0) {
         jvmCmd.append(" -D");
         jvmCmd.append(jgroupProperties[i]);
         jvmCmd.append("=");
         jvmCmd.append(property);
       }
     }
     String classpath = System.getProperty("java.class.path");
     if (classpath != null && classpath.length() > 0) {
       jvmCmd.append(" -classpath ");
       jvmCmd.append(classpath);
     }
     // Enable assertions
     jvmCmd.append(" -ea");
 //    jvmCmd.append(" -verbose:gc -Xloggc:/tmp/meling-logs/ed-gc.log");
 //    System.out.println("JVM command line:\n" + jvmCmd.toString());
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Constructor
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   public ExecDaemon()
     throws RemoteException, JgroupException, ConfigurationException
   {
     ConfigManager.init();
     DistributedSystemConfig dsc = ConfigManager.getDistributedSystem();
     if (!dsc.containsLocalHost()) {
       JgroupException e = new JgroupException(
         "Distributed system configuration does not contain local host: " + Network.getLocalHostName() +
         "\nThis is required by the execdaemon to start servers.");
       throw e;
     }
     Host host = DistributedSystemConfig.getLocalHost();
     try {
       DatagramSocket socket = new DatagramSocket(host.getPort(), host.getAddress());
       // close the socket to allow the Jgroup/daemon to use it once a replica starts
       socket.close();
     } catch (SocketException e1) {
       JgroupException e = new JgroupException(
         "The datagram socket required by the Jgroup/daemon is occupied.\n" +
         "This could be due to an old copy of the Jgroup/daemon that was not properly\n" +
         "shutdown in a previous run.  Please ensure to kill any remaining java processes,\n" +
         "or switch to a different host port: " + host.getAddress() + ":" + host.getPort(), e1);
       throw e;
     }
 
     /*
      * Set up a shutdown hook for the ExecDaemon to ensure that
      * replicas are removed when the ExecDaemon is shutdown or crashes.
      */
     Runtime.getRuntime().addShutdownHook(new Thread("ED-ShutdownHook") {
       public void run() {
         log.info("ED-ShutdownHook: shutdown commenced (removing replicas)...");
         if (Eventlogger.ENABLED)
           Eventlogger.logEventFlush("HostShutdown");
         removeAllReplicas();
         LogManager.shutdown();
       }});
 
     /*
      * Ensure that we create a local bootstrap registry for this machine,
      * and that there is no other registry occupying the registry access port.
      */
     try {
       BootstrapRegistry.createRegistry();
     } catch (RemoteException e) {
       log.error("Could not create bootstrap registry", e);
       throw new Abort("ExecDaemon could not be started on " + thisHost, e);
     }
 
     Exporter exporter =
       new BasicJeriExporter(TcpServerEndpoint.getInstance(0), new BasicILFactory());
     Remote execProxy = exporter.export(this);
     try {
       BootstrapRegistry.bind(EXEC_DAEMON_NAME, execProxy);
     } catch (AlreadyBoundException e) {
       throw new Abort("Execution daemon already running on " + thisHost, e);
     }
 
     if (Boolean.getBoolean("jgroup.factory.notify.presence")) {
       notifyPresence();
     }
 
     DaemonInteraction.initDaemon();
 //    if (Boolean.getBoolean("jgroup.daemon")) {
 //      /*
 //       * Start the Jgroup daemon local to the execdaemon; thus all group managers
 //       * will use RMI to communicate with the Jgroup daemon.
 //       */
 //      DaemonInteraction.initDaemon();
 //    }
   }
 
   /**
    * If the ExecDaemon is started after the ReplicaManager, then lookup
    * the replication manager in the dependable registry and notify our presence.
    */
   private void notifyPresence()
   {
     /* Obtain the replication manager registry name. */
     AppConfig rmApp = AppConfig.getApplication(ReplicaManagerImpl.class);
     String rmRegName = rmApp.getRegistryName();
 
     try {
       /*
        * Obtain a proxy for the dependable registry running in the
        * distributed system.
        */
       DependableRegistry registry = RegistryFactory.getRegistry();
       ReplicationManager replicationManager = (ReplicationManager) registry.lookup(rmRegName);
       /* Notify the replication manager of our presence. */
       replicationManager.notifyEvent(new EDPresentEvent(Network.getLocalHost()));
 
     } catch (Exception e) {
       if (log.isDebugEnabled())
         log.debug(rmRegName + " not yet available.");
     }
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Main method
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   public static void main(String[] argv) 
     throws Exception
   {
     // Start the ExecDaemon
     try {
       ExecDaemon exec = new ExecDaemon();
       System.out.println("Replica object factory ready: " + Network.getLocalHostName());
     } catch(Exception e) {
       Abort.exit("Failed to start the execution daemon on the local host", e, 1);
     }
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Methods from ExecService interface
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Comments inherited from jgroup.arm.ExecService
    */
   public synchronized boolean createExecReplica(ClassData classData)
     throws RemoteException, ExecException
   {
     if (replicaMap.containsKey(classData)) {
       // Replica is already running on this host.
       return false;
     } else {
       Class c = classData.getClassObject();
       String mainError = c.getName() + " does not implement a 'public static void"
                          + " main(String[] args)' method.";
 
       /*
        * Check for method:  main(String[] args)  in the specified class.
        */
       try {
 
         Method main = c.getDeclaredMethod("main", new Class[] { String[].class });
         int m = main.getModifiers();
         if (!Modifier.isPublic(m) || !Modifier.isStatic(m)) {
           throw new ExecException(mainError);
         }
 
         ReplicaThread replicaThrd = new ReplicaThread(classData, main);
         replicaThrd.setDaemon(true);
         replicaThrd.start();
 
         /* Keep the replica thread object in case we want to remove it */
         replicaMap.put(classData, replicaThrd);
         if (log.isDebugEnabled())
           log.debug("Created replica: " + classData.getShortName());
         return true;
 
       } catch (NoSuchMethodException e) {
         throw new ExecException(mainError, e);
       }
     }
   }
 
 
   /**
    *  Instantiate and start the given class (replica) in a JVM process
    *  separate from the execution service JVM process.
    *
    *  Creates a new replica in a seperate JVM process.  Only one replica
    *  for each matching <code>ClassData</code> object (or group) can be
    *  started on each host.  This makes the operation idempotent,
    *  allowing several replication managers to invoke the same create
    *  operation on the same host as another replication manager already
    *  may have done.  This prevents creating multiple instances of the
    *  same class with identical arguments.  Thus, different arguments
    *  are not considered the same instance.
    *
    *  @param classData 
    *    The class data object representing the replica to start.
    *  @return
    *    True if the replica was created; false is returned if the
    *    replica is already running on this host.
    * 
    *  @throws RemoteException 
    *    If remote operation failed.
    *  @throws ExecException 
    *    If the specified class (replica) could not be instantiated.
    */
   public synchronized boolean createReplica(ClassData classData)
     throws RemoteException, ExecException
   {
     if (replicaMap.containsKey(classData)) {
       // Replica is already running on this host.
       return false;
     } else {
       ReplicaThread replicaThrd = new ReplicaThread(classData);
       replicaThrd.setPriority(Thread.NORM_PRIORITY+1);
       replicaThrd.setDaemon(true);
       replicaThrd.start();
 
       /* Keep the replica thread object in case we want to remove it */
       replicaMap.put(classData, replicaThrd);
       if (log.isDebugEnabled())
         log.debug("Created replica: " + classData.getShortName());
       return true;
     }
   }
 
 
   public synchronized boolean removeReplica(ClassData classData)
     throws RemoteException
   {
     if (log.isDebugEnabled())
       log.debug("Removing replica: " + classData.getShortName());
     if (replicaMap.containsKey(classData)) {
       /*
        * Remove the local replica thread for the specified class
        * (group) from the replicaMap.
        */
       ReplicaThread replicaThrd = (ReplicaThread) replicaMap.remove(classData);
       /* Remove the local replica and its thread. */
       replicaThrd.remove();
       if (Eventlogger.ENABLED) {
         int groupId = AppConfig.getApplication(classData.getClassObject()).getGroupId();
         Eventlogger.logEventFlush(new ReplicaEvent(Removed, groupId));
       }
       /* Further clean up is performed by the replica thread. */
       return true;
     } else {
       // Replica not running on this host; may have been removed already
       return false;
     }
   }
 
 
   /**
    *  Remove all local replicas associated with this ExecDaemon.
    */
   private void removeAllReplicas()
   {
     try {
       Set<ClassData> replicas = queryReplicas();
       for (ClassData cl : replicas) {
         removeReplica(cl);
       }
     } catch (RemoteException e) {
       // Cannot happen since we are local to this JVM
       e.printStackTrace();
     }
   }
 
 
   /**
    * @see jgroup.core.arm.ExecService#queryReplicas()
    */
   public synchronized Set<ClassData> queryReplicas()
     throws RemoteException
   {
     // clone the set of replicas to avoid concurrent access to the map
     synchronized (replicaMap) {
       return new HashSet<ClassData>(replicaMap.keySet());
     }
   }
 
 
   /**
    * @see jgroup.core.arm.ExecService#ping()
    */
   public void ping()
     throws RemoteException
   {
     if (log.isDebugEnabled())
       log.debug("PING");
     return;
   }
 
   /* (non-Javadoc)
    * @see jgroup.core.arm.ExecService#shutdown()
    */
   public void shutdown(final int delay)
     throws RemoteException
   {
     new Thread() {
       public void run() {
         if (delay < 0) {
           System.out.println("Halting ExecDaemon and all local replicas");
           removeAllReplicas();
           try { sleep(2000); } catch (InterruptedException e) { }
           // ensure that log files are closed
           if (Eventlogger.ENABLED)
             Eventlogger.close();
           LogManager.shutdown();
           // do not run the shutdown hooks
           rt.halt(0);
         } else if (delay > 0) {
           System.out.println("I'm shuting down in " + delay + " milliseconds.");
           try { sleep(delay); } catch (InterruptedException e) { }
         } else {
           // delay == 0
           System.out.println("I'm shuting down now");
         }
         if (Eventlogger.ENABLED)
           Eventlogger.logEventFlush("InjectingFailure");
         // ensure that log files are closed
         LogManager.shutdown();
         System.out.println(new jgroup.util.log.Event(Shutdown, "InjectingFailure"));
         System.exit(0);
         // The ExecDaemon shutdown hook will deal with removing replicas.
       }
     }.start();
     return;
   }
 
 
   ////////////////////////////////////////////////////////////////////////////////////////////
   // Private inner classfor handling replica threads.
   ////////////////////////////////////////////////////////////////////////////////////////////
 
   /**
    *  Private inner class <code>ReplicaThread</code> is used to start
    *  replicas, and will keep waiting for the replica to end, vulentarely
    *
    *  @author Hein Meling
    */
   private class ReplicaThread
     extends Thread
   {
     /**
      *  The process in which the replica may be running, assuming it is
      *  running in a JVM external to the execution daemon.
      */
     private Process replica;
 
     /** Short classname used for logging. */
     private String shortClass;
 
     /** The full class information for this replica. */
     private ClassData classData;
 
     /**
      *  Indicator used to trigger the removal of a replica.
      *  Used only for replicas internal to the execution daemon.
      */
     private boolean removed;
 
 
     //////////////////////////////////////////////////////////////////////////////////////////
     // Constructors for the replica thread
     //////////////////////////////////////////////////////////////////////////////////////////
 
     /**
      *  Creates a new <code>ReplicaThread</code> instance for a replica
      *  started in the same JVM as the execution daemon.
      *
      *  @param classData
      *    Class and argument information for the replica to be started.
      *  @exception ExecException
      *    Raised if the replica could not be started.
      */
     ReplicaThread(ClassData classData, Method main)
       throws ExecException
     {
       super("Local-ReplicaThread-" + classData.getShortName());
 
       removed = false;
       this.classData = classData;
       shortClass = classData.getShortName();
 
       try {
         /*
          * Invoke the main method to start the replica in the same
          * JVM as the execution daemon.
          */
         main.invoke(null, new Object[] { classData.getArgs() });
       } catch (InvocationTargetException e) {
         throw new ExecException(classData.getClassObject()
           + " has thrown an exception during invocation of the main method.", e.getTargetException());
       } catch (IllegalAccessException e) {
         throw new ExecException(shortClass + " failed to start", e);
       }
       log.info("Starting replica: " + shortClass);
     }
 
 
     /**
      *  Creates a new <code>ReplicaThread</code> instance for a replica
      *  started external to the execution daemon, i.e., in a separate JVM.
      *
      *  @param classData
      *    Class and argument information for the replica to be started.
      *  @exception ExecException
      *    Raised if the replica could not be started.
      */
     ReplicaThread(ClassData classData)
       throws ExecException
     {
       super("External-ReplicaThread-" + classData.getShortName());
 
       this.classData = classData;
       shortClass = classData.getShortName();
       String cmd = classData.getCommand(jvmCmd.toString(), appStarter);
       try {
         replica = rt.exec(cmd);
 
         /*
          * Create a new thread to read and print the <i>stdout</i> for
          * this replica.
          */
         Thread outThrd = new Thread(shortClass + "-out") {
           public void run() {
             InputStream drOut = replica.getInputStream();
             BufferedReader out = new BufferedReader(new InputStreamReader(drOut));
             String stdout;
             try {
               while ((stdout = out.readLine()) != null) {
                 synchronized(this) {
                   if (log.isDebugEnabled()) {
                     log.debug(stdout);
                   }
                   System.out.println(shortClass + "-" + thisMachine + ": " + stdout);
                 }
               }
               out.close();
             } catch (java.io.IOException e) {
               log.warn("Replica " + shortClass + " failed.", e);
             }
           }
         };
         outThrd.setDaemon(true);
         outThrd.start();
 
         /*
          * Create a new thread to read and print the <i>stderr</i> for
          * this replica.
          */
         Thread errThrd = new Thread(shortClass + "-err") {
           public void run() {
             InputStream drErr = replica.getErrorStream();
             BufferedReader err = new BufferedReader(new InputStreamReader(drErr));
             String stderr;
             try {
               while ((stderr = err.readLine()) != null) {
                 synchronized(this) {
                   log.error(stderr);
                   System.out.println(shortClass + "-" + thisMachine + "-err: " + stderr);
                 }
               }
               err.close();
             } catch (java.io.IOException e) {
               log.warn("Replica " + shortClass + " failed.", e);
             }
           }
         };
         errThrd.setDaemon(true);
         errThrd.start();
 
         if (log.isDebugEnabled())
           log.debug("Starting replica: " + cmd);
         else
           log.info("Starting replica: " + shortClass);
 
       } catch (java.io.IOException ioe) {
         throw new ExecException("Replica could not be started.", ioe);
       }
     }
 
 
     /**
      *  Invoked by the <code>removeReplica()</code> method to stop the
      *  thread, causing the replica itself to exit.
      *
      *  For the case when the replica is executing in a separate process,
      *  invoking the <code>destory()</code> method will cause the
      *  <code>run()</code> method below to pass through the
      *  <code>waitFor()</code> method and thus it will clean up its
      *  tables and exit this thread.
      */
     synchronized void remove()
     {
       if (replica == null) {
         /* The replica is running in this JVM. */
         removed = true;
         notifyAll();
       } else {
         /* The replica is running in a separate JVM process. */
         OutputStream os = replica.getOutputStream();
         PrintWriter writer = new PrintWriter(os, true);
         /* Send a shutdown message to the child replica process. */
         if (log.isDebugEnabled())
           log.debug("Sending SHUTDOWN message to replica");
         writer.println(SHUTDOWN_REPLICA);
         writer.close();
       }
     }
 
 
     /**
      *  This method will also wait until the replica exits and will thus
      *  clean up the execution service tables.
      */
     public void run()
     {
       try {
         if (replica == null) {
           synchronized (this) {
             while (!removed) {
               wait();
             }
           }
         } else {
           if (replica.waitFor() != 0) {
             log.warn("Got non-zero return value; replica failed or was removed.");
           }
         }
 
       } catch (InterruptedException e) {
         log.warn("Replica " + shortClass + " failed.", e);
 
       } finally {
         /* Replica exit; always clean up. */
         log.info("Replica exiting: " + shortClass);
         synchronized (this) {
           replicaMap.remove(classData);
           if (replica != null)
             replica.destroy();
         }
 //        try {
 //          /* Notify the replication manager of this replica failure. */
 //          replicaManager.notifyEvent(new ReplicaFailureEvent(classData));
 //        } catch (Exception e) {
 //          log.warn("Unable to notify Replication Manager", e);
 //        }
       }
     }
   } // END of private inner class ReplicaThread
 
 } // END of public class ExecDaemon