// BSD License (http://www.galagosearch.org/license)

 package org.galagosearch.tupleflow.execution;

 import java.net.UnknownHostException;

 import org.mortbay.jetty.Server;

 import org.galagosearch.tupleflow.Utility;

 import org.galagosearch.tupleflow.Processor;

 import org.galagosearch.tupleflow.TypeReader;

 import org.galagosearch.tupleflow.execution.Job.StagePoint;

 import org.galagosearch.tupleflow.execution.StageGroupDescription.DataPipeRegion;

 import org.galagosearch.tupleflow.execution.StageInstanceDescription.PipeInput;

 import org.galagosearch.tupleflow.execution.StageInstanceDescription.PipeOutput;

 import java.io.File;

 import java.io.IOException;

 import java.io.StringReader;

 import java.net.InetAddress;

 import java.util.ArrayList;

 import java.util.Arrays;

 import java.util.Collection;

 import java.util.Collections;

 import java.util.Date;

 import java.util.HashMap;

 import java.util.HashSet;

 import java.util.Iterator;

 import java.util.List;

 import java.util.Map;

 import java.util.TreeMap;

 import java.util.TreeSet;

 import java.util.concurrent.ExecutionException;

 import java.util.concurrent.Future;

 import javax.xml.parsers.ParserConfigurationException;

 import javax.xml.parsers.SAXParser;

 import javax.xml.parsers.SAXParserFactory;

 import java.util.logging.Level;

 import java.util.logging.Logger;

 import org.mortbay.jetty.bio.SocketConnector;

 import org.xml.sax.InputSource;

 import org.xml.sax.SAXException;

 import org.xml.sax.SAXParseException;

 /**

  * <p>This class is responsible for executing TupleFlow jobs.</p>

  *

  * <p>A job is specified using the TupleFlow Job class.  The Job class has an XML form

  * which can be parsed by JobConstructor, but you can create one programmatically as well.</p>

  *

  * <p>Before the job is executed, it is verified.  JobExecutor verifies that all the

  * classes references by the Job object actually exist, and that the connections point

  * sensible places.  Once it has been verified, the JobExecutor builds an execution plan

  * that will execute the job with as much parallelism as possible while not violate any

  * ordering constraints dictated by stage connections.  After the plan is generated,

  * the job is sent to a StageExecutor for the low-level details of execution.</p>

  *

  * <p>TupleFlow has many different kinds of StageExecutors you can use.  To get started

  * and to debug your code, use the LocalStageExecutor or ThreadedExecutor.  To harness

  * more parallelism, use the SSHStageExecutor or the DRMAAStageExecutor.</p>

  * 

  * @author trevor

  */

 public class JobExecutor {

     ErrorStore store;

     Job job;

     String temporaryStorage;

     int defaultHashCount = 10;

     long maximumFileInputs = 200;

     HashMap<String, ConnectionDescription> connections = new HashMap<String, ConnectionDescription>();

     HashMap<String, StageGroupDescription> stages = new HashMap<String, StageGroupDescription>();

     ArrayList<String> stageOrder = new ArrayList<String>();

     HashMap<String, HashSet<String>> stageChildren = new HashMap<String, HashSet<String>>();

     HashMap<String, HashSet<String>> stageParents = new HashMap<String, HashSet<String>>();

     ArrayList<DataPipe> pipes = new ArrayList<DataPipe>();

     public JobExecutor(Job job, String temporaryStorage, ErrorStore store) {

         this.store = store;

         this.temporaryStorage = temporaryStorage;

         this.job = job;

     }

     /**

      * Parses the XML text in the file specified by the filename parameter into a Job.

      * 

      * @param filename A path to the XML file to parse.

      * @param store An ErrorStore object where all error information should be sent.

      * @return A job instance based on the XML description in the text parameter.

      * @throws org.xml.sax.SAXException

      * @throws javax.xml.parsers.ParserConfigurationException

      * @throws java.io.IOException

      */

     public static Job parseFile(String filename, ErrorStore store) throws SAXException, ParserConfigurationException, IOException {

         // Parse the file

         JobConstructor jobParseSaxHandler = new JobConstructor(filename, store);

         SAXParser parser = SAXParserFactory.newInstance().newSAXParser();

         try {

             parser.parse(new File(filename), jobParseSaxHandler);

         } catch (SAXParseException e) {

             store.addError(filename, e);

         }

         return jobParseSaxHandler.getJob();

     }

     /**

      * Parses the XML text in the text parameter into a Job.

      * 

      * @param text XML text to parse into a Job.

      * @param store An ErrorStore object where all error information should be sent.

      * @return A job instance based on the XML description in the text parameter.

      * @throws org.xml.sax.SAXException

      * @throws javax.xml.parsers.ParserConfigurationException

      * @throws java.io.IOException

      */

     public static Job parseText(String text, ErrorStore store) throws SAXException, ParserConfigurationException, IOException {

         // Parse the file

         JobConstructor jobParseSaxHandler = new JobConstructor(store);

         SAXParser parser = SAXParserFactory.newInstance().newSAXParser();

         try {

             StringReader reader = new StringReader(text);

             parser.parse(new InputSource(reader), jobParseSaxHandler);

         } catch (SAXParseException e) {

             store.addError("none", e);

         }

         return jobParseSaxHandler.getJob();

     }

     /**

      * This method tries to combine stages together to reduce overhead.

      * 

      * In particular, this method looks for two stages, A and B, where each

      * copy of B takes input from only one copy of stage A.  In this case, 

      * all the steps from B are moved into stage A, saving a lot of file overhead

      * in transferring tuples from A to B.

      * 

      * This method is particularly useful for jobs that are created

      * automatically.

      * 

      * @param job  The job instance to optimize.

      * @return A new job instance, perhaps with fewer stages.

      */

     public static Job optimize(Job job) {

         // First, figure out what source gets output from the stage, if any.

         HashMap<String, String> outputs = new HashMap<String, String>();

         HashMap<String, String> inputs = new HashMap<String, String>();

         for (Stage stage : job.stages.values()) {

             // Output mapping

             if (stage.steps.size() > 0) {

                 Step lastStep = stage.steps.get(stage.steps.size() - 1);

                 if (lastStep instanceof OutputStep) {

                     OutputStep output = (OutputStep) lastStep;

                     outputs.put(stage.name, output.getId());

                 }

                 Step firstStep = stage.steps.get(0);

                 if (firstStep instanceof InputStep) {

                     InputStep input = (InputStep) firstStep;

                     inputs.put(stage.name, input.getId());

                 }

             }

         }

         // Create a mapping from String -> Stage

         HashMap<String, Stage> stages = new HashMap<String, Stage>(job.stages);

         // Now, rip through the connections, and try to find a connection that links

         // one of these inputs to one of the outputs.

         Iterator<Connection> iterator = job.connections.iterator();

         Connection connection;

         innerLoop:

         while (iterator.hasNext()) {

             connection = iterator.next();

             // For simplicity, find just connections with single inputs and outputs and no hashing

             if (connection.inputs.size() != 1) {

                 continue;

             }

             if (connection.outputs.size() < 1) {

                 continue;

             }

             if (connection.getHash() != null) {

                 continue;

             }

             ConnectionEndPoint connectionInput = connection.inputs.get(0);

             String stageOutputPointId = outputs.get(connectionInput.getStageName());

             // if the input to this connection is not the main output of a stage, skip

             if (!connectionInput.getPointName().equals(stageOutputPointId)) {

                 continue;            // make sure all of the outputs are the inputs of stages

             }

             for (ConnectionEndPoint connectionOutput : connection.outputs) {

                 String stageInputPointId = inputs.get(connectionOutput.getStageName());

                 if (!connectionOutput.getPointName().equals(stageInputPointId)) {

                     continue innerLoop;

                 }

                 if (connectionOutput.getAssignment() == ConnectionAssignmentType.Combined) {

                     continue innerLoop;

                 }

             }

             // now we've verified that these stages can be combined together.

             Stage source = stages.get(connectionInput.getStageName());

             MultiStep multi = new MultiStep();

             multi.groups = new ArrayList<ArrayList<Step>>();

             for (ConnectionEndPoint connectionOutput : connection.outputs) {

                 Stage destination = stages.get(connectionOutput.getStageName());

                 // getting ready: remove the first step, add on to the multi

                 int length = destination.steps.size();

                 ArrayList<Step> steps = new ArrayList<Step>(destination.steps.subList(1, length));

                 multi.groups.add(steps);

                 renameConnections(job, source, destination);

                 // combine dependence information

                 source.connections.remove(connectionInput.getPointName());

                 destination.connections.remove(connectionOutput.getPointName());

                 source.connections.putAll(destination.connections);

                 // remove the destination stage

                 job.stages.remove(destination);

             }

             source.steps.remove(source.steps.size() - 1);

             // only add a multi step if there were multiple outputs

             if (multi.groups.size() == 1) {

                 source.steps.addAll(multi.groups.get(0));

             } else {

                 source.steps.add(multi);

             }

             // remove this connection

             iterator.remove();

             // recurse to remove other connections

             return optimize(job);

         }

         return job;

     }

     public static void renameConnections(Job job, Stage source, Stage destination) {

         // for each connection, rename dest -> source.

         for (Connection connection : job.connections) {

             for (ConnectionEndPoint input : connection.inputs) {

                 if (input.getStageName().equals(destination.name)) {

                     input.setStageName(source.name);

                 }

             }

             for (ConnectionEndPoint output : connection.outputs) {

                 if (output.getStageName().equals(destination.name)) {

                     output.setStageName(source.name);

                 }

             }

         }

     }

     public void prepare() {

         boolean successful = constructAndVerify();

         if (!successful) {

             return;

         }

         // look through each stage to see how many open files it will have.

         // if a particular stage will have a lot of open files, add some

         // intermediate merge stages.

         if (needsMergeStages()) {

             addMergeStages(job);

             constructAndVerify();

         }

         for (DataPipe pipe : pipes) {

             pipe.makeDirectories();

         }

     }

     public void clear() {

         connections.clear();

         stageChildren.clear();

         stageOrder.clear();

         stageParents.clear();

         stages.clear();

         pipes.clear();

     }

     /**

      * Checks to see if any stage has too many inputs.

      * 

      * @return true, if there is a stage with more than maximumFileInputs.

      */

     public boolean needsMergeStages() {

         for (StageGroupDescription stage : stages.values()) {

             long totalInputs = 0;

             if (stage.getName().endsWith("mergeStage")) {

                 continue;

             }

             for (DataPipeRegion region : stage.inputs.values()) {

                 totalInputs += region.fileCount();

             }

             if (totalInputs > this.maximumFileInputs) {

                 return true;

             }

         }

         return false;

     }

     /**

      * Add a merge stage to this job that merges the output of

      * stageName called pointName.

      * 

      * @param job  The job that should get the new merge stage.

      * @param stageName  The stage that contains the output that needs merging.

      * @param pointName  The output point that needs merging in the stage stageName.

      */

     public void addMergeStage(Job job, String stageName, String pointName) {

         // find the stage and the point, initialize class/order information

         Stage inputStage = job.stages.get(stageName);

         StageConnectionPoint inputPoint = inputStage.getConnection(pointName);

         String className = inputPoint.getClassName();

         String[] typeOrder = inputPoint.getOrder();

         String mergedStageName = stageName + "-" + pointName + "-mergeStage";

         String mergedPointName = pointName + "-merged";

         // if this merge stage has already been added, don't add it again

         if (job.stages.containsKey(mergedStageName)) {

             return;        // create the stage itself

         }

         Stage s = new Stage(mergedStageName);

         s.add(new StageConnectionPoint(ConnectionPointType.Input,

                                        pointName,

                                        className,

                                        typeOrder,

                                        null));

         s.add(new StageConnectionPoint(ConnectionPointType.Output,

                                        pointName + "-merged",

                                        className,

                                        typeOrder,

                                        null));

         s.add(new InputStep(pointName));

         s.add(new OutputStep(mergedPointName));

         job.add(s);

         String[] hash = null;

         int hashCount = -1;

         // run through the connections list, find all inputs for the previous data

         for (Connection connection : job.connections) {

             for (ConnectionEndPoint input : connection.inputs) {

                 if (input.getStageName().equals(stageName) &&

                         input.getPointName().equals(pointName)) {

                     if (hash != null && connection.hash != null &&

                         !Arrays.equals(hash,connection.hash)) {

                         continue;

                     }

                     if (connection.hash != null) {

                         hash = connection.hash;

                         connection.hash = null;

                     }

                     input.setStageName(mergedStageName);

                     input.setPointName(mergedPointName);

                 }

             }

         }

         // now, add a connection between the producing stage and the merge stage

         job.connect(new StagePoint(stageName, pointName),

                     new StagePoint(mergedStageName, pointName),

                     ConnectionAssignmentType.Each,

                     hash,

                     hashCount);

     }

     /**

      * Find stages that need to open a lot of files for reading when running,

      * and add some intermediate merge stages to reduce problems with open files.

      * 

      * @param job

      */

     public void addMergeStages(Job job) {

         // look at each stage in the job

         for (StageGroupDescription stage : stages.values()) {

             long totalInputs = 0;

             // add up every file this stage will need to open

             for (DataPipeRegion region : stage.inputs.values()) {

                 totalInputs += region.fileCount();

             }

             // if this stage needs to open too many files, it might crash.

             // therefore, we want to add in an extra merge stage for every input

             // that contains a lot of files.

             if (totalInputs > this.maximumFileInputs) {

                 Stage s = stage.getStage();

                 ArrayList<Connection> relevantConnections = new ArrayList<Connection>();

                 // look for connections that point to this stage and store them

                 for (Connection connection : job.connections) {

                     for (ConnectionEndPoint point : connection.outputs) {

                         if (point.getStageName().equals(s.name)) {

                             relevantConnections.add(connection);

                         }

                     }

                 }

                 // for each stage we marked in the last loop, find ones where all

                 // the inputs come from a single stage, and that stage has a large

                 // instanceCount (it generates a lot of files), and make a merge stage.

                 for (Connection connection : relevantConnections) {

                     if (connection.inputs.size() != 1) {

                         continue;

                     }

                     ConnectionEndPoint endPoint = connection.inputs.get(0);

                     String inputStageName = endPoint.getStageName();

                     String inputPointName = endPoint.getPointName();

                     StageGroupDescription inputStageDesc = stages.get(inputStageName);

                     // if there's no description, that means we just added it

                     if (inputStageDesc == null ||

                             inputStageName.endsWith("mergeStage")) {

                         continue;

                     }

                     if (inputStageDesc.instanceCount > 1) {

                         addMergeStage(job, inputStageName, inputPointName);

                     }

                 }

             }

         }

     }

     private static class EndPointName implements Comparable<EndPointName> {

         public String stageName;

         public String pointName;

         public ConnectionPointType type;

         public FileLocation location;

         public EndPointName(String stageName, String pointName, ConnectionPointType type, FileLocation location) {

             this.stageName = stageName;

             this.pointName = pointName;

             this.type = type;

             this.location = location;

         }

         public EndPointName(String stageName, String pointName, ConnectionPointType type) {

             this(stageName, pointName, type, null);

         }

         public int compareTo(EndPointName other) {

             int result = stageName.compareTo(other.stageName);

             if (result == 0) {

                 result = pointName.compareTo(other.pointName);

             }

             if (result == 0) {

                 result = type.compareTo(other.type);

             }

             return result;

         }

         @Override

         public int hashCode() {

             return stageName.hashCode() + 7 * pointName.hashCode() + 15 * type.toString().hashCode();

         }

         @Override

         public String toString() {

             return String.format("%s %s %s", stageName, pointName, type.toString());

         }

     }

     /**

      * In the parameter file, each stage has a connections section that describes a set of

      * connection endpoints for the stage (inputs and outputs).  This method verifies that

      * all of those endpoints are connected to valid connections, defined under the 

      * connections tag in the job.  If the method finds an dangling (unconnected) endpoint, 

      * an error message is added to the ErrorStore.

      */

     public void findDanglingEndpoints(final Job job) {

         TreeSet<EndPointName> endPointNames = new TreeSet();

         // First, make a list of all endpoints referenced in all stages.

         for (Stage stage : job.stages.values()) {

             // find the corresponding description object

             StageGroupDescription description = stages.get(stage.name);

             // add all connection points to the set

             for (StageConnectionPoint point : stage.connections.values()) {

                 EndPointName ep = new EndPointName(stage.name, point.getExternalName(), point.

                                                    getType(), point.location);

                 endPointNames.add(ep);

             }

         }

         // Now we have a list of referenced names.  We now remove every endpoint that

         // is referenced in the connections section.

         for (ConnectionDescription connection : connections.values()) {

             for (EndPointDescription input : connection.inputs) {

                 EndPointName ep = new EndPointName(input.stage.getName(),

                                                    input.stagePoint.getExternalName(),

                                                    input.stagePoint.getType());

                 endPointNames.remove(ep);

             }

             for (EndPointDescription output : connection.outputs) {

                 EndPointName ep = new EndPointName(output.stage.getName(),

                                                    output.stagePoint.getExternalName(),

                                                    output.stagePoint.getType());

                 endPointNames.remove(ep);

             }

         }

         for (EndPointName ep : endPointNames) {

             store.addError(ep.location,

                            ep.stageName + ": No connection references the " + ep.type +

                            " with the name '" + ep.pointName + "'.");

         }

     }

     private boolean constructAndVerify() {

         clear();

         // first, we make stage group descriptions, getting ready to add connections in

         for (Stage stage : job.stages.values()) {

             stages.put(stage.name, new StageGroupDescription(stage));

         }

         // verify each stage in the job to make sure that

         // the connections between individual steps are typesafe, that

         // step classes exist, etc.

         Verification.verify(job, store);

         if (store.getErrors().size() > 0) {

             return false;

         }

         // build data about connections between stages, while verifying

         // type safety between stage connections.

         buildConnections(job);

         if (store.getErrors().size() > 0) {

             return false;

         }

         findDanglingEndpoints(job);

         if (store.getErrors().size() > 0) {

             return false;

         }

         generateDependencies();

         if (store.getErrors().size() > 0) {

             return false;

         }

         determineStageOrder();

         if (store.getErrors().size() > 0) {

             return false;

         }

         countStages();

         if (store.getErrors().size() > 0) {

             return false;

         }

         createPipeObjects();

         return true;

     }

     private void generateDependencies() {

         // generate a list of stages

         for (StageGroupDescription stage : stages.values()) {

             stageChildren.put(stage.getName(), new HashSet());

             stageParents.put(stage.getName(), new HashSet());

         }

         for (ConnectionDescription connection : connections.values()) {

             for (EndPointDescription input : connection.inputs) {

                 for (EndPointDescription output : connection.outputs) {

                     stageChildren.get(input.getStageName()).add(output.getStageName());

                     stageParents.get(output.getStageName()).add(input.getStageName());

                 }

             }

         }

     }

     private void determineStageOrder() {

         ArrayList<String> result = new ArrayList();

         HashSet<String> used = new HashSet();

         HashSet<String> batch = new HashSet();

         for (String stageName : stageParents.keySet()) {

             if (stageParents.get(stageName).size() == 0) {

                 batch.add(stageName);

             }

         }

         while (batch.size() > 0) {

             HashSet<String> nextBatch = new HashSet();

             for (String stageName : batch) {

                 result.add(stageName);

                 used.add(stageName);

                 HashSet<String> children = stageChildren.get(stageName);

                 // adding this stage to the list may have unblocked a child

                 for (String child : children) {

                     HashSet<String> childParents = stageParents.get(child);

                     if (!used.contains(child) && used.containsAll(childParents)) {

                         nextBatch.add(child);

                     }

                 }

             }

             batch = nextBatch;

             nextBatch = new HashSet();

         }

         assert result.size() == stages.size();

         stageOrder = result;

     }

     private static class EndPointDescription {

         public EndPointDescription(ConnectionDescription connection,

                                    StageGroupDescription stage,

                                    ConnectionEndPoint connectionPoint,

                                    StageConnectionPoint stagePoint) {

             this.connectionPoint = connectionPoint;

             this.stagePoint = stagePoint;

             this.stage = stage;

             this.connection = connection;

         }

         public String getStageName() {

             return stage.getName();

         }

         public StageConnectionPoint getStagePoint() {

             return stagePoint;

         }

         public ConnectionEndPoint getConnectionPoint() {

             return connectionPoint;

         }

         public ConnectionDescription getConnection() {

             return connection;

         }

         public StageGroupDescription stage;

         public StageConnectionPoint stagePoint;

         public ConnectionEndPoint connectionPoint;

         public ConnectionDescription connection;

     }

     private class ConnectionDescription {

         public ConnectionDescription(Connection connection) {

             this.connection = connection;

             this.inputs = new ArrayList();

             this.outputs = new ArrayList();

         }

         public boolean isHashed() {

             return connection.hash != null;

         }

         public int getOutputCount() {

             int result = 1;

             if (isHashed()) {

                 String hashCount = job.properties.get("hashCount");

                 if (connection.getHashCount() > 0) {

                     result = connection.getHashCount();

                 } else if (hashCount != null &&

                         Utility.isInteger(hashCount)) {

                     result = Integer.parseInt(hashCount);

                 } else {

                     result = defaultHashCount;

                 }

             } else {

                 result = getInputCount();

             }

             return result;

         }

         public int getInputCount() {

             int total = 0;

             for (EndPointDescription input : inputs) {

                 total += input.stage.instanceCount;

             }

             return total;

         }

         public String getName() {

             return connection.getName();

         }

         public String[] getOrder() {

             return connection.getOrder();

         }

         public String[] getHash() {

             return connection.getHash();

         }

         private String getClassName() {

             return connection.getClassName();

         }

         public DataPipe getPipe() {

             return pipe;

         }

         public void setPipe(DataPipe pipe) {

             this.pipe = pipe;

         }

         @Override

         public String toString() {

             return String.format("%s %s", getClassName(), getName());

         }

         public Connection connection;

         public ArrayList<EndPointDescription> inputs;

         public ArrayList<EndPointDescription> outputs;

         public DataPipe pipe;

     }

     private EndPointDescription createEndPoint(ConnectionDescription connection,

                                                ConnectionEndPoint endPoint) {

         StageGroupDescription stageDescription = stages.get(endPoint.getStageName());

         if (stageDescription == null) {

             store.addError(endPoint.location,

                            "The stage '" + endPoint.getStageName() + "' was not found.");

         } else {

             Stage stage = stageDescription.getStage();

             StageConnectionPoint point = stage.getConnection(endPoint.getPointName());

             if (point == null) {

                 store.addError(endPoint.location, "The endpoint '" + endPoint.getPointName() + "' wasn't found in this stage, " +

                                "even though there is a connection to it.");

             } else if (!ConnectionPointType.connectable(endPoint.getType(), point.getType())) {

                 store.addError(endPoint.location,

                                "The endpoint '" + endPoint.getPointName() + "' is in this stage, but it's going the wrong direction.");

             } else if (!point.getClassName().equals(connection.connection.getClassName())) {

                 store.addError(endPoint.location, "This " + point.getType() + " has a different class name '" + point.getClassName() +

                                " than the connection that connects to it: " + connection.connection.

                                getClassName() + ".");

             } else if (!Arrays.equals(point.getOrder(), connection.connection.getOrder())) {

                 store.addError(endPoint.location, "This " + point.getType() + " has a different order " + Arrays.toString(point.getOrder()) +

                                " than the connection that connects to it: " + Arrays.toString(

                                connection.connection.getOrder()));

             } else {

                 return new EndPointDescription(connection, stageDescription, endPoint, point);

             }

         }

         return null;

     }

     private ArrayList<EndPointDescription> createEndPoints(ConnectionDescription connection,

                                                            ArrayList<ConnectionEndPoint> endPoints) {

         ArrayList<EndPointDescription> results = new ArrayList<EndPointDescription>();

         for (ConnectionEndPoint endPoint : endPoints) {

             EndPointDescription epd = createEndPoint(connection, endPoint);

             if (epd != null) {

                 results.add(epd);

             }

         }

         return results;

     }

     /**

      * Creates ConnectionDescription objects for each connection listed in

      * the Job object.  The ConnectionDescription objects combine information

      * from the StageConnectionPoints (in the Stages) and the ConnectionEndPoints

      * (in the Job.Connection objects) to make them easier to access.  In the process

      * of making these objects, this method typechecks all of the connections.

      */

     private void buildConnections(final Job job) {

         for (Connection connection : job.connections) {

             ConnectionDescription description = new ConnectionDescription(connection);

             // verify that the class, order, and hash exist

             ErrorHandler handler = store.getErrorHandler(connection.location);

             Verification.requireClass(connection.getClassName(), handler);

             Verification.requireOrder(connection.getClassName(), connection.getOrder(), handler);

             if (connection.getHash() != null) {

                 Verification.requireOrder(connection.getClassName(),

                                                                         connection.getHash(),

                                                                         handler);

             }

             description.inputs = createEndPoints(description, connection.inputs);

             description.outputs = createEndPoints(description, connection.outputs);

             connections.put(connection.getName(), description);

         }

     }

     /**

      * This method computes the number of copies of each stage to run.  

      * The execution count of stage depends on its inputs.  If the input

      * for a stage is hashed 200 ways, for instance, then there will need

      * to be 200 copies of the stage. 

      */

     private void countStages() {

         HashMap<String, HashSet<EndPointDescription>> stageInputs = new HashMap();

         HashMap<String, HashSet<EndPointDescription>> stageOutputs = new HashMap();

         for (String stageName : stages.keySet()) {

             stageInputs.put(stageName, new HashSet());

             stageOutputs.put(stageName, new HashSet());

         }

         for (ConnectionDescription connection : connections.values()) {

             for (EndPointDescription endPoint : connection.inputs) {

                 stageOutputs.get(endPoint.stage.getName()).add(endPoint);

             }

             for (EndPointDescription endPoint : connection.outputs) {

                 stageInputs.get(endPoint.stage.getName()).add(endPoint);

             }

         }

         for (String stageName : stageOrder) {

             StageGroupDescription stage = stages.get(stageName);

             // if stage has no inputs, then we store 1 in stageCounts

             if (stageInputs.get(stageName).size() == 0) {

                 stage.instanceCount = 1;

             } else {

                 // find out what the assignment is for this connection.

                 int instanceCount = 1;

                 boolean unknown = true;

                 HashSet<EndPointDescription> inputs = stageInputs.get(stageName);

                 for (EndPointDescription description : inputs) {

                     ConnectionEndPoint point = description.getConnectionPoint();

                     ConnectionAssignmentType assignment = point.getAssignment();

                     switch (assignment) {

                         case One:

                             store.addError(point.location,

                                            "The 'one' mode is not currently supported.");

                             break;

                         case Each:

                             int inputCount = description.connection.getOutputCount();

                             if (unknown) {

                                 instanceCount = inputCount;

                                 unknown = false;

                             } else if (!unknown && instanceCount != inputCount) {

                                 store.addError(point.location, "The number of stage instances for '" +

                                                stageName + "' is ambiguous (" + inputCount +

                                                " or " + instanceCount + ")");

                             }

                             break;

                         case Combined:

                             // this doesn't affect the number of stages.

                             break;

                     }

                 }

                 if (unknown) {

                     instanceCount = 1;

                 }

                 stage.instanceCount = instanceCount;

             }

         }

     }

     /**

      * Creates the directory structures to hold the files for this job.

      */

     private void createPipeObjects() {

         // Now, we need to create the connections

         for (ConnectionDescription connection : connections.values()) {

             // Make the parent directory

             String directoryName = temporaryStorage + File.separator + connection.getName();

             new File(directoryName).mkdir();

             DataPipe pipe = new DataPipe(directoryName,

                                          connection.getName(),

                                          connection.getClassName(),

                                          connection.getOrder(),

                                          connection.getHash(),

                                          connection.getInputCount(),

                                          connection.getOutputCount());

             int startIndex = 0;

             connection.setPipe(pipe);

             for (EndPointDescription input : connection.inputs) {

                 StageGroupDescription description = stages.get(input.getStageName());

                 description.outputs.put(input.getStagePoint().getInternalName(),

                                         new DataPipeRegion(pipe,

                                                            startIndex,

                                                            startIndex + description.getInstanceCount(),

                                                            ConnectionPointType.Input));

                 startIndex += description.getInstanceCount();

             }

             for (EndPointDescription output : connection.outputs) {

                 StageGroupDescription description = stages.get(output.getStageName());

                 description.inputs.put(output.getStagePoint().getInternalName(),

                                        new DataPipeRegion(pipe,

                                                           0,

                                                           connection.getOutputCount(),

                                                           ConnectionPointType.Output));

             }

             pipes.add(pipe);

         }

     }

     public static class JobExecutionStatus {

         // these are the names of all stages that have completed

         HashMap<String, StageExecutionStatus> completedStages = new HashMap<String, StageExecutionStatus>();

         // named of all stages that have been launched (contains all completed stages too)

         HashSet<String> launchedStages = new HashSet<String>();

         // all stages that are currently running.  Includes a Future object that can be used

         // to wait for the stage to complete (and to get exceptions thrown by the stage)

         HashMap<String, StageExecutionStatus> runningStages = new HashMap<String, StageExecutionStatus>();

         // names of connections that are complete, meaning that all their inputs have been created

         HashSet<String> completedConnections = new HashSet<String>();

         // map from connection names to the names of stages that provide inputs to the connection

         HashMap<String, HashSet<String>> connectionDependencies = new HashMap<String, HashSet<String>>();

         // reference to the parent class.

         HashMap<String, StageGroupDescription> stages;

         // reference to the parent class.

         String temporaryStorage;

         StageExecutor executor;

         Date startDate;

         public JobExecutionStatus(HashMap<String, StageGroupDescription> stages,

                 String temporaryStorage, StageExecutor executor, String masterURL) {

             this.stages = stages;

             this.temporaryStorage = temporaryStorage;

             this.executor = executor;

             this.startDate = new Date();

             for (StageGroupDescription description : stages.values()) {

                 // build a list of dependencies from pipe inputs to stage names

                 for (DataPipeRegion region : description.outputs.values()) {

                     String pipeName = region.pipe.pipeName;

                     if (!connectionDependencies.containsKey(pipeName)) {

                         connectionDependencies.put(

                                 pipeName, new HashSet<String>());

                     }

                     connectionDependencies.get(pipeName).add(description.getName());

                 }

                 description.setMasterURL(masterURL);

             }

         }

         class BlockedExecutionStatus implements StageExecutionStatus {

             String name;

             int instances;

             BlockedExecutionStatus(String name, int instances) {

                 this.name = name;

                 this.instances = instances;

             }

             public String getName() { return name; }

             public int getBlockedInstances() { return instances; }

             public int getQueuedInstances() { return 0; }

             public int getRunningInstances() { return 0; }

             public int getCompletedInstances() { return 0; }

             public boolean isDone() { return false; }

             public List<Exception> getExceptions() { return Collections.EMPTY_LIST; }

         }

         public synchronized boolean isComplete() {

             return stages.size() == completedStages.size();

         }

         public synchronized Map<String, StageExecutionStatus> getStageStatus() {

             Map<String, StageExecutionStatus> result = new TreeMap();

             for (String stageName : stages.keySet()) {

                 int instanceCount = stages.get(stageName).getInstanceCount();

                 if (completedStages.containsKey(stageName)) {

                     result.put(stageName, completedStages.get(stageName));

                 } else if (runningStages.containsKey(stageName)) {

                     result.put(stageName, runningStages.get(stageName));

                 } else {

                     result.put(stageName, new BlockedExecutionStatus(stageName, instanceCount));

                 }

             }

             return result;

         }

         /**

          * Returns the start date for this job.

          */

         public Date getStartDate() {

             return startDate;

         }

         /**

          * Returns the total amount of free memory in this JVM.

          */

         public long getFreeMemory() {

             return Runtime.getRuntime().freeMemory();

         }

         /**

          * Returns the maximum amount of memory that can be used by this

          * Java virtual machine.

          */

         public long getMaxMemory() {

             return Runtime.getRuntime().maxMemory();

         }

         public void run() throws InterruptedException, ExecutionException {

             // while there are incomplete stages, choose one to execute

             while (launchedStages.size() < stages.size()) {

                 // look for stages where all of their inputs are complete

                 StageGroupDescription description = findRunnableStage(stages.values(), launchedStages,

                                                                       completedConnections);

                 // didn't find any runnable stages, so we need to check to

                 // see if any other stages have finished recently that might have

                 // generated interesting input for pending stages.

                 while (description == null) {

                     // wait for at least one stage to complete

                     waitForStages(runningStages, completedStages);

                     updateCompletedConnections(completedStages, completedConnections,

                                                connectionDependencies);

                     // now, try again to find a runnable stage

                     description = findRunnableStage(stages.values(), launchedStages,

                                                     completedConnections);

                 }

                 StageExecutionStatus result = executor.execute(description, temporaryStorage);

                 synchronized(this) {

                     launchedStages.add(description.stage.name);

                     runningStages.put(description.stage.name, result);

                 }

             }

             // wait for everything to complete

             while (runningStages.size() > 0) {

                 waitForStages(runningStages, completedStages);

             }

         }

         /**

          * Finds a stage that is ready to run by checking stage dependencies.

          *

          * @param descriptions

          * @param launchedJobs

          * @param completedConnections

          */

         private synchronized StageGroupDescription findRunnableStage(

                 Collection<StageGroupDescription> descriptions,

                 HashSet<String> launchedJobs,

                 HashSet<String> completedConnections) {

             // for each job we might want to launch

             for (StageGroupDescription description : descriptions) {

                 // if it has already been launched, we don't need to do it again

                 if (launchedJobs.contains(description.getName())) {

                     continue;            // are the inputs to this stage ready?

                 }

                 boolean allComplete = true;

                 for (DataPipeRegion region : description.inputs.values()) {

                     // if this input is incomplete, we can't run this stage yet

                     if (!completedConnections.contains(region.pipe.pipeName)) {

                         allComplete = false;

                         break;

                     }

                 }

                 if (allComplete) {

                     return description;

                 }

             }

             // there are no stages ready to run

             return null;

         }

         /**

          * Polls all the running stages to see if they've completed.  When one completes,

          * it is added to completedStages and the method returns.

          * 

          * @param runningStages

          * @param completedStages

          * @throws java.lang.InterruptedException

          * @throws java.util.concurrent.ExecutionException

          */

         private void waitForStages(

                 HashMap<String, StageExecutionStatus> runningStages,

                 final HashMap<String, StageExecutionStatus> completedStages)

                 throws InterruptedException, ExecutionException {

             long delay = 1;

             while (runningStages.size() > 0) {

                 synchronized(this) {

                     for (String name : runningStages.keySet()) {

                         StageExecutionStatus status = runningStages.get(name);

                         if (status.isDone()) {

                             // force the exception to throw

                             List<Exception> exceptions = status.getExceptions();

                             if (exceptions.size() > 0) {

                                 throw new ExecutionException("Stage threw an exception: ", exceptions.get(0));

                             }

                             completedStages.put(name, status);

                             runningStages.remove(name);

                             return;

                         }

                     }

                 }

                 // check at least once a second, but poll faster at first

                 delay = Math.min(delay * 2, 1000);

                 Thread.sleep(delay);

             }

         }

         /**

          * Reads through all completed stages, trying to find any connections

          * that have been satisfied.  By satisfied, we mean that all the input

          * has been generated for a particular connection.  Once a connection

          * is satisfied, stages that require that read input from that connection

          * can start running.

          *

          * @param completedStages

          * @param completedConnections

          * @param connectionDependencies

          */

         private synchronized void updateCompletedConnections(

                 final HashMap<String, StageExecutionStatus> completedStages,

                 final HashSet<String> completedConnections,

                 final HashMap<String, HashSet<String>> connectionDependencies) {

             // Loop over all connections:

             for (String pipeName : connectionDependencies.keySet()) {

                 // These are all the stages that need to be completed before

                 // the connection pipeName is satisfied.

                 HashSet<String> pipeInputStages = connectionDependencies.get(pipeName);

                 if (completedStages.keySet().containsAll(pipeInputStages)) {

                     completedConnections.add(pipeName);

                 }

             }

         }

     }

     public void run(StageExecutor executor) throws InterruptedException, ExecutionException, UnknownHostException, IOException {

         Server server = new Server(Utility.getFreePort());

         runWithServer(executor, server);

     }

     public void runWithServer(StageExecutor executor, Server server) throws ExecutionException, InterruptedException, UnknownHostException {

         // FIXME: all of this needs to be refactored.

         InetAddress address = java.net.InetAddress.getLocalHost();

         int port = server.getConnectors()[0].getPort();

         String masterURL = String.format("http://%s:%d", address.getHostAddress(), port);

         JobExecutionStatus status = new JobExecutionStatus(stages, temporaryStorage, executor, masterURL);

         MasterWebHandler handler = new MasterWebHandler(status);

         server.addHandler(handler);

         status.run();

         handler.waitForFinalPage();

         server.removeHandler(handler);

     }

     public static boolean runLocally(Job job, ErrorStore store) throws IOException,

             InterruptedException, ExecutionException, Exception {

         StageExecutor executor = StageExecutorFactory.newInstance("local", new String[] {});

         File tempFile = Utility.createTemporary();

         tempFile.delete();

         tempFile.mkdir();

         JobExecutor jobExecutor = new JobExecutor(job, tempFile.getAbsolutePath(), store);

         jobExecutor.prepare();

         if (store.hasStatements()) {

             return false;

         }

         int port = Utility.getFreePort();

         Server server = new Server(port);

         server.start();

         System.out.println("Status: http://localhost:" + port);

         try {

             jobExecutor.runWithServer(executor, server);

         } finally {

             server.stop();

             executor.shutdown();

         }

         return !store.hasStatements();

     }

     public static void main(String[] args) throws ParserConfigurationException, SAXException,

             IOException, InterruptedException, ExecutionException {

         if (args.length < 3) {

             System.out.println("usage: executionModel parameterFile temporaryStorage");

             System.out.println(

                     "   where executionModel is one of: local, drmaa, ssh, debug, remotedebug");

             System.exit(-1);

         }

         String executionModel = args[0];

         String parameterFile = args[1];

         String temporaryStorage = args[2];

         String[] remaining = Utility.subarray(args, 3);

         StageExecutor executor = null;

         Logger logger = Logger.getLogger(JobExecutor.class.toString());

         logger.setLevel(Level.INFO);

         executor = StageExecutorFactory.newInstance(executionModel, remaining);

         ErrorStore store = new ErrorStore();

         // First, parse the job

         Job job = parseFile(parameterFile, store);

         if (store.hasStatements()) {