// BSD License (http://www.galagosearch.org/license)
   package org.galagosearch.tupleflow.execution;
   
   import java.util.ArrayList;
   import java.util.Collections;
   import java.util.List;
   import java.util.concurrent.CountDownLatch;
   import java.util.concurrent.ExecutionException;
   import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.TimeUnit;
  
  /***
   * This will eventually allow you to run TupleFlow workers on other machines using
   * SSH.  For now, it doesn't work.  If you really need job distribution, try the Grid
   * Engine plugin.
   * 
   * @author trevor
   */
  public class SSHStageExecutor extends RemoteStageExecutor {
      private static String machineEndMarker = "STOP#MACHINE#SHUTDOWN";
      private LinkedBlockingQueue<String> machines = new LinkedBlockingQueue();
      private LinkedBlockingQueue<StageTask> tasks = new LinkedBlockingQueue();
      private ExecutorService pool = null;
      private String commandName;
  
      public SSHStageExecutor(String commandName, List<String> machines) {
          this.commandName = commandName;
          this.machines.addAll(machines);
          this.pool = Executors.newCachedThreadPool();
      }
  
      public class SSHStageExecutionContext implements Runnable, StageExecutionStatus {
          String name;
          List<String> jobPaths;
          boolean done;
  
          public SSHStageExecutionContext(String name, List<String> jobPaths) {
              this.name = name;
              this.jobPaths = jobPaths;
              this.done = false;
          }
  
          public void run() {
              CountDownLatch latch = new CountDownLatch(jobPaths.size());
  
              for (String jobPath : jobPaths) {
                  // submit this job to the queue
                  StageTask task = new StageTask(commandName, jobPath, latch);
                  pool.execute(task);
              }
  
              synchronized(this) { done = true; }
          }
  
          public synchronized boolean isDone() {
              return done;
          }
  
          public String getName() {
              return name;
          }
  
          public int getBlockedInstances() {
              return 0;
          }
  
          public int getQueuedInstances() {
              // FIXME
              return jobPaths.size();
          }
  
          public int getRunningInstances() {
              // FIXME
              return 0;
          }
  
          public int getCompletedInstances() {
              // FIXME
              return 0;
          }
  
          public List<Exception> getExceptions() {
              // FIXME
              return Collections.EMPTY_LIST;
          }
      }
  
      public class StageTask implements Runnable {
          private String commandName;
          private String jobFileArgument;
          private CountDownLatch latch;
  
          public StageTask(String commandName, String jobFileArgument, CountDownLatch latch) {
              this.commandName = commandName;
              this.jobFileArgument = jobFileArgument;
              this.latch = latch;
         }
 
         public void run() {
             String machineName;
 
             try {
                 // first, wait for a machine reservation
                 machineName = machines.poll();
 
                 // if we see a shutdown marker, quit, but put the marker back
                 if (machineName.equals(machineEndMarker)) {
                     machines.offer(machineEndMarker);
                     return;
                 }
 
                 String[] arguments = new String[]{machineName, jobFileArgument};
                 Process process = Runtime.getRuntime().exec(commandName, arguments);
 
                 // close the process stdin
                 process.getOutputStream().close();
                 // BUGBUG: someday we need to trap process stdout/stderr here
 
                 process.waitFor();
             } catch (Exception e) {
                 // BUGBUG: fix this too
                 e.printStackTrace();
             } finally {
                 latch.countDown();
             }
         }
 
         public boolean isNullTask() {
             return commandName == null && jobFileArgument == null && latch == null;
         }
     }
 
     public StageExecutionStatus submit(String stageName, ArrayList<String> jobPaths, String temporary) {
         SSHStageExecutionContext context = new SSHStageExecutionContext(stageName, jobPaths);
         new Thread(context).start();
         return context;
     }
 
     public void shutdown() {
         machines.add(machineEndMarker);
         pool.shutdown();
     }
 }