/* An engine than runs models in Spark.
 * 
 * Copyright (c) 2014 The Regents of the University of California.
 * All rights reserved.
 *
 * '$Author: crawl $'
 * '$Date: 2016-10-17 19:14:15 +0000 (Mon, 17 Oct 2016) $' 
 * '$Revision: 34532 $'
 * 
 * Permission is hereby granted, without written agreement and without
 * license or royalty fees, to use, copy, modify, and distribute this
 * software and its documentation for any purpose, provided that the above
 * copyright notice and the following two paragraphs appear in all copies
 * of this software.
 *
 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY
 * FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES
 * ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF
 * THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE
 * PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND THE UNIVERSITY OF
 * CALIFORNIA HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES,
 * ENHANCEMENTS, OR MODIFICATIONS.
 *
 */
package org.kepler.spark.director;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.StringWriter;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.URI;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.kepler.build.project.ProjectLocator;
import org.kepler.configuration.ConfigurationProperty;
import org.kepler.ddp.Utilities;
import org.kepler.ddp.actor.pattern.AtomicPatternActor;
import org.kepler.ddp.actor.pattern.CoGroup;
import org.kepler.ddp.actor.pattern.Cross;
import org.kepler.ddp.actor.pattern.DDPDataSink;
import org.kepler.ddp.actor.pattern.DDPDataSource;
import org.kepler.ddp.actor.pattern.DDPPatternActor;
import org.kepler.ddp.actor.pattern.Map;
import org.kepler.ddp.actor.pattern.Match;
import org.kepler.ddp.actor.pattern.Reduce;
import org.kepler.ddp.actor.pattern.SingleInputPatternActor;
import org.kepler.ddp.director.DDPDirector;
import org.kepler.ddp.director.DDPEngine;
import org.kepler.spark.actor.SparkConnection;
import org.kepler.spark.operator.CoGroupOperator;
import org.kepler.spark.operator.CrossOperator;
import org.kepler.spark.operator.DataSink;
import org.kepler.spark.operator.DataSource;
import org.kepler.spark.operator.FileDataSink;
import org.kepler.spark.operator.FileDataSource;
import org.kepler.spark.operator.JoinOperator;
import org.kepler.spark.operator.MapOperator;
import org.kepler.spark.operator.NullSink;
import org.kepler.spark.operator.Operator;
import org.kepler.spark.operator.ReduceOperator;
import org.kepler.spark.operator.TokenSink;
import org.kepler.spark.operator.TokenSource;
import org.kepler.spark.stub.KeplerCoGroupStub;
import org.kepler.spark.stub.KeplerCrossStub;
import org.kepler.spark.stub.KeplerMapStub;
import org.kepler.spark.stub.KeplerMatchStub;
import org.kepler.spark.stub.KeplerReduceStub;
import org.kepler.spark.stub.ScriptEngineMapStub;
import org.kepler.spark.stub.ScriptEngineReduceStub;

import ptolemy.actor.IOPort;
import ptolemy.data.ArrayToken;
import ptolemy.data.IntToken;
import ptolemy.data.StringToken;
import ptolemy.data.Token;
import ptolemy.data.expr.Parameter;
import ptolemy.data.expr.StringParameter;
import ptolemy.data.type.BaseType;
import ptolemy.kernel.util.Attribute;
import ptolemy.kernel.util.IllegalActionException;
import ptolemy.kernel.util.NameDuplicationException;
import ptolemy.kernel.util.NamedObj;
import ptolemy.kernel.util.Workspace;

/** An engine than runs workflows in Spark. This engine
 *  converts DDP pattern actors (Map, Reduce, Cross, CoGroup, and
 *  Match) and I/O actors (DDPDataSink and DDPDataSource) into a
 *  Spark job and runs it on the server.
 *  <p>
 *  <b>NOTE:</b> Only DDP pattern and I/O actors may be present in
 *  the workflow. Other actors must placed inside the composite
 *  pattern actors or in a different sub-workflow.
 *  </p>
 * 
 *  @author Daniel Crawl
 *  @version $Id: SparkEngine.java 34532 2016-10-17 19:14:15Z crawl $
 *
 */
public class SparkEngine extends DDPEngine {

    public SparkEngine(DDPDirector director) throws IllegalActionException,
            NameDuplicationException {
        super(director);

        _engineName = SPARK_ENGINE_NAME;
        
        // initialize this here since _addParameters() accesses it
        // before the field can be initialized.
        _numSameJVMWorkers = 8;
    }

    /** React to a parameter change. */
    @Override
        public void attributeChanged(Attribute attribute) throws IllegalActionException {
                
        if(attribute == masterHostAndPort) {
            InetSocketAddress address = null;
            String masterHostAndPortStr = ((StringToken)masterHostAndPort.getToken()).stringValue();
            if(!masterHostAndPortStr.trim().isEmpty()) {
                // parse the string to check for errors.
                address = SparkConnection.getMasterAndPortAddress(masterHostAndPortStr);
            }
                        
            if(address != null && address.getHostString().equals("localhost")) {
                try {
                    _masterHostAndPortStr = InetAddress.getLocalHost().getHostName() +
                            ":" + address.getPort();
                } catch (UnknownHostException e) {
                    throw new IllegalActionException(_director, e, "Unknown host.");
                }
                System.out.println("Using " + _masterHostAndPortStr +
                        " instead of 'localhost' since Spark has problems binding to 'localhost'."); 
            } else {
                _masterHostAndPortStr = masterHostAndPortStr;
            }
            
        } else if(attribute == numSameJVMWorkers) {
            Token token = numSameJVMWorkers.getToken();
            if(token == null) {
                throw new IllegalActionException(_director,
                        "numSameJVMWorkers must be specified.");
            }
            _numSameJVMWorkers = ((IntToken)token).intValue();
                            
        } else if(attribute == driverMemory) {
            String driverMemoryStr = ((StringToken)driverMemory.getToken()).stringValue();
            if(driverMemoryStr.trim().isEmpty()) {
                driverMemoryStr = SparkConnection.DEFAULT_DRIVER_MEMORY;
            }
                            
        } else {
                super.attributeChanged(attribute);
        }
        
    }

    /** Clone the SparkEngine into the specified workspace.
     *  @param workspace The workspace for the cloned object.
     *  @exception CloneNotSupportedException Not thrown in this base class
     *  @return The new SparkEngine.
     */
    @Override
    public Object clone(Workspace workspace) throws CloneNotSupportedException {
        SparkEngine newObject = (SparkEngine) super.clone(workspace);
        newObject._context = null;
        newObject._inputsAlreadyDone = new HashSet<Operator>();
        newObject._job = null;
        return newObject;
    }
    
    @Override
    public void preinitialize() throws IllegalActionException {
        
        super.preinitialize();

        // get the degree of parallelism even though it may be done
        // in the parent class. the parent class may set it to 1
        // as the default.
        _degreeOfParallelism = ((IntToken)_director.degreeOfParallelism.getToken()).intValue();

        // load the default config dir if necessary
        _checkConfigDir();
        
    }

    public static void shutdownServer() throws IllegalActionException {
        
        // shut down any servers we started.
        synchronized(_serverStartStopLock) {
                for(String stopScriptStr : _stopScriptsForShutdown) {
                    
                    System.out.println("Stopping Spark server by running " +
                            stopScriptStr);
                    
                    ProcessBuilder builder = new ProcessBuilder(stopScriptStr);
                    
                    // make sure JAVA_HOME is set
                    java.util.Map<String,String> env = builder.environment();
                    if(env.get("JAVA_HOME") == null) {
                        env.put("JAVA_HOME", System.getProperty("java.home"));
                    }
                    
                    try {
                        Process process = builder.start();
                        process.waitFor();
                    } catch (Exception e) {
                        throw new IllegalActionException("Unable to stop Spark: " + e.getMessage());
                    }            
                }
        }
    }

    /** Stop any running Spark jobs. */
    @Override
    public void stop() throws IllegalActionException {
        super.stop();
        
        synchronized(_jobLock) {
            if(_job != null) {
                _job.stop();
            }
        }
    }

    ///////////////////////////////////////////////////////////////////
    ////                      public fields                      //////

    public StringParameter masterHostAndPort;
    
    public Parameter numSameJVMWorkers;
    
    public StringParameter driverMemory;
    
    ///////////////////////////////////////////////////////////////////
    ////                      protected methods                  //////


    /** Check the existence of the Spark assembly jar. */
    @Override
    protected boolean _checkFilesBeforeStartingServer() throws IllegalActionException {
        
        if(!Utilities.isExecutingUnderResourceManager()) {
            // look for the assembly jar. the assembly jar is not shipped with
            // the spark module, so give the URL building it if not found.
            if(!super._checkFilesBeforeStartingServer()) {
                throw new IllegalActionException(
                    "Cannot start the Spark Master since the assembly jar does not exist. " +
                    "This jar is not shipped\n" +
                    "with Kepler and must be built; " +
                    "the instructions for building it can be found at:\n" +
                    "https://kepler-project.org/developers/interest-groups/distributed/configuring-hadoop-for-biokepler-or-ddp-suite");
            }
        }
        return true;
    }

    @Override
    protected void _executeDDPJob() throws IllegalActionException {
        
        // see what jars are necessary. call this before creating the job
        // since a new classloader be created for external classes.
        final List<URI> jarURIs = _getJarList();            

        _getContext();
        
        try {
            _sparkParallelism = _context.defaultParallelism();
            
            if(_degreeOfParallelism == DDPDirector.DEFAULT_INTTOKEN.intValue()) {
                _degreeOfParallelism = _sparkParallelism;
            }

            if(_degreeOfParallelism > _sparkParallelism) {
                System.err.println("WARNING: the default degree of parallelism for " +
                        _director.getName() + " is " + _degreeOfParallelism +
                        ", which is greater than Spark's parallelism." +
                        "(" + _sparkParallelism + ").");
                if(_sameJVM) {
                    System.err.println("Increase the number of local workers to achieve " +
                        "this degree of parallelism.");
                }
            }
            
            System.out.println("Default Spark parallelism is " +
                    _degreeOfParallelism + ".");
            
            for (URI jarURI : jarURIs) {
                String jarStr = jarURI.toString();
                // Spark 2 throws an exception when more than one jar with
                // the same name is added (even when in different directories).
                // So we need to exclude the (empty) kepler-tasks.jar that is
                // built for the kepler-tasks module.
                if(!jarStr.endsWith("target/kepler-tasks.jar")) {
                    _context.addJar(jarStr);
                }
            }

            synchronized(_jobLock) {
                _job = _getModelPlan();
            
                //System.out.println("Starting Spark job.");
            
                _job.start();
            }
            
            try {
                _job.waitForFinish();
            } catch (Exception e) {
                throw new IllegalActionException(_director, e, "Error in Spark job.");
            }
                
            //System.out.println("Finished Spark job.");
            
            // set the _job to null here instead of wrapup() to prevent
            // stop() from calling _job.stop() when the context has been
            // stopped.
            synchronized(_jobLock) {
                _job = null;
            }        

        } finally {
            if(_context != null) {
                SparkConnection.releaseContext();
                _context = null;
            }
        }
    }

        /** Add parameters to the containing director. */
    @Override
    protected void _addParameters() throws IllegalActionException, NameDuplicationException {

        masterHostAndPort = (StringParameter) _director.getAttribute("masterHostAndPort");
        if(masterHostAndPort == null) {
                masterHostAndPort = new StringParameter(_director, "masterHostAndPort");
        }
        
        numSameJVMWorkers = (Parameter) _director.getAttribute("numSameJVMWorkers");
        if(numSameJVMWorkers == null) {
            numSameJVMWorkers = new Parameter(_director, "numSameJVMWorkers");
            numSameJVMWorkers.setTypeEquals(BaseType.INT);                          
            numSameJVMWorkers.setToken(String.valueOf(_numSameJVMWorkers));
        }
        
        driverMemory = (StringParameter) _director.getAttribute("driverMemory");
        if(driverMemory == null) {
            driverMemory = new StringParameter(_director, "driverMemory");
            driverMemory.setToken(SparkConnection.DEFAULT_DRIVER_MEMORY);
        }       
    }

    /** Remove engine-specific parameters from the director. */
    @Override
    protected void _removeParameters() throws IllegalActionException, NameDuplicationException {      
        masterHostAndPort.setContainer(null);
        numSameJVMWorkers.setContainer(null);
        driverMemory.setContainer(null);
    }

    /** Constructs a SparkJob from the model. */
    protected SparkJob _getModelPlan() throws IllegalActionException {
     
        _inputsAlreadyDone.clear();

        SparkJob plan = null;
        
        // find all the sinks in the model
        final List<DDPDataSink> sinks = _container.entityList(DDPDataSink.class);
        
        if(sinks.isEmpty()) {
            throw new IllegalActionException(_director, "No data sinks found.");
        }
        
        // for each sink, traverse the graph to the source, adding to the plan
        for(DDPDataSink sink : sinks) {
            
            // get the operator for this sink
            DataSink operator = (DataSink) _getContract(sink);
            
            // add sink to the plan
            if(plan == null) {
                plan = new SparkJob(_context, operator, "Kepler Spark Job " + _container.getName());
            } else {
                plan.addDataSink(operator);
            }
            
            // traverse graph for this sink
            _addInputsForContract(sink, operator);
        }
                
        return plan;
    }

    /** Recursively add inputs for a contract by traversing the model graph.
     * 
     *  @param pactor the current pactor in the model.
     *  @param contract the contract for the current pactor. 
     */
    private void _addInputsForContract(DDPPatternActor pactor, Operator operator) throws IllegalActionException {
        
        // see if we've already done this contract
        if(!_inputsAlreadyDone.contains(operator)) {
        
            // see if the execution class name or execution code is set
            final String executionClassName = pactor.getExecutionClassName();
            final String executionCodeType = pactor.getExecutionCodeType();
            if(executionClassName.isEmpty() &&
                    executionCodeType == null &&
                    (pactor instanceof SingleInputPatternActor)) {
                // add sub-wf to plan configuration
                _addSubModelToContract((SingleInputPatternActor)pactor, operator);
            }
            
            // see how many inputs are required
            int numRequiredInputs = operator.numInputs();
            if(numRequiredInputs < 0 || numRequiredInputs > 2) {
                throw new IllegalActionException(_director, "Unknown type of contract: " + operator.getClass());
            }
            
            // see if there's at least one input
            if(numRequiredInputs > 0) {
                
                // get the first input port
                final IOPort inputPort1 = (IOPort) pactor.getPort("in");
                if(inputPort1 == null) {
                    throw new IllegalActionException(_director, "DDPPatternActor " + pactor.getFullName() +
                        "is missing input port \"in\".");
                }
                
                _addInputForPactorPort(inputPort1, pactor, operator);

                // see if there's a second input
                if(numRequiredInputs > 1) {
                    
                    final IOPort inputPort2 = (IOPort) pactor.getPort("in2");
                    if(inputPort2 == null) {
                        throw new IllegalActionException(_director, "DDPPatternActor " + pactor.getFullName() +
                            "is missing input port \"in2\".");
                    }
    
                    _addInputForPactorPort(inputPort2, pactor, operator);
                }
            }
        }
    }

    /** Set the input contract for a single input port and traverse the model
     *  graph to recursively add input contracts.
     *  
     *  @param port the input port for 
     *  @param pactor the current pactor in the model.
     *  @param contract the contract for the current pactor. 
     */
    private void _addInputForPactorPort(IOPort port, DDPPatternActor pactor, Operator operator) 
        throws IllegalActionException {
        
        // get the connected actor
        final List<?> outputPorts = port.sourcePortList();
        if(outputPorts.isEmpty()) {
            throw new IllegalActionException(_director, "DDPPatternActor input port " +
                    port.getName() + " must be connected.");
        } else if(outputPorts.size() > 1) {
            throw new IllegalActionException(_director, "DDPPatternActor input port " +
                    port.getName() + " may only be connected to one port.");                
        }
        
        final IOPort outputPort1 = (IOPort) outputPorts.get(0);
        final NamedObj outputNamedObj = outputPort1.getContainer();
                
        // FIXME
        if(!(outputNamedObj instanceof DDPPatternActor)) {
            throw new IllegalActionException(_director, "Actor " + pactor.getFullName() +
                " is connected to a non-DDPPatternActor: " + outputNamedObj.getFullName());
        }
        
        final Operator outputContract = _getContract((DDPPatternActor)outputNamedObj);

        int numInputs = operator.numInputs();
        if(numInputs == 1) {
            operator.setInput(outputContract);
        } else if(numInputs == 2) {
            if(port.getName().equals("in")) {
                operator.setInput(0, outputContract);
            } else if(port.getName().equals("in2")) {
                operator.setInput(1, outputContract);                
            } else {
                throw new IllegalActionException(port, "Input port must be named either \"in\" or \"in2\".");
            }
        }
        
        _inputsAlreadyDone.add(operator);
        
        // recursively add the inputs for output pactor
        _addInputsForContract((DDPPatternActor)outputNamedObj, outputContract);
    }

    /** Write the sub-workflow of a SingleInputPatternActor either to a parameter
     *  in the Spark job configuration or to a file.
     */
    protected void _addSubModelToContract(SingleInputPatternActor pactor, Operator contract)
            throws IllegalActionException {
                         
        final String name = pactor.getName();
                
        final Configuration contractConfiguration = contract.getParameters();
        contractConfiguration.set(Utilities.CONFIGURATION_KEPLER_MODEL_NAME, name);

        contractConfiguration.setBoolean(Utilities.CONFIGURATION_KEPLER_RUN_WORKFLOW_LIFECYCLE_PER_INPUT,
                pactor.getRunWorkflowLifecyclePerInput());
        
        // make a clone of the pactor so we can add TypeAttributes to ports
        // and add parameters from the hierarchy
        
        Workspace workspaceForClone;
        if(_sameJVM) {
            workspaceForClone = new Workspace(name);
        } else {
            workspaceForClone = _director.workspace();
        }
        
        SingleInputPatternActor clonePactor;
        try {
            clonePactor = (SingleInputPatternActor)pactor.clone(workspaceForClone);
        } catch (CloneNotSupportedException e) {
            throw new IllegalActionException(_director, e, "Unable to clone " + name);
        }
        
        // copy the port types to the clone
        _setPortTypes(clonePactor);
        
        // copy the parameters to the clone
        _copyParameters(pactor, clonePactor);
        
        // set display redirect path
        String directDir = pactor.getDisplayRedirectDir();
        NamedObj redirectSpecifier = pactor;
        if(directDir.isEmpty()) {
            redirectSpecifier = _director;
            directDir = _displayRedirectDir;
        }
        if(!directDir.isEmpty()) {
            
            // display redirection when running in same jvm is not supported
            if(_sameJVM) {
                throw new IllegalActionException(redirectSpecifier,
                        "Redirecting display actors is not supported for"
                        + " Spark server running in the same JVM as Kepler.");
            }
            
            final File file = new File(directDir);
            if (!file.exists() && !file.mkdirs()) {
                throw new IllegalActionException(_director, "Could not create directories " + file);
            }
            if (!file.isDirectory() || !file.canWrite()) {
                throw new IllegalActionException(redirectSpecifier,
                        "Parameter '" + _displayRedirectDir + "' must be a directory and writable.");
            }
            contractConfiguration.set(Utilities.CONFIGURATION_KEPLER_REDIRECT_DIR, directDir);
        }

        if(_sameJVM) {
            _subWorkflows.put(name, clonePactor);
        } else {
        
            // remove top-level ports and relations
            Utilities.removeModelPorts(clonePactor);
            
            // serialize the clone pactor
            
            if(_writeSubWorkflowsToFiles) {
                
                if(_jobDir == null) {
                    _createJobDirectory();
                }
                
                final String modelPath = _jobDir + name + ".xml";
                FileWriter writer = null;
                try {
                    writer = new FileWriter(modelPath);
                    clonePactor.exportMoML(writer);
                } catch(IOException e) {
                    throw new IllegalActionException(_director, e, "Error writing model to " + modelPath);
                } finally {
                    if(writer != null) {
                        try {
                            writer.close();
                        } catch (IOException e) {
                            throw new IllegalActionException(_director, e, "Error writing model to " + modelPath);
                        }
                    }
                }
                
                // add model file to plan configuration
                contractConfiguration.set(Utilities.CONFIGURATION_KEPLER_MODEL_PATH, modelPath);
                
            } else {
             
                // write model to a string
                final StringWriter writer = new StringWriter();
                try {
                    clonePactor.exportMoML(writer);
                } catch (IOException e) {
                    throw new IllegalActionException(_director, e, "Error serializing model.");
                }
                
                // add string to configuration
                contractConfiguration.set(Utilities.CONFIGURATION_KEPLER_MODEL, writer.toString());
            }            
            
            // set the location of the kepler installation directory.
            // NOTE: this is done so that the stub can initialize kepler and set
            // the java properties for each module's workflow directory, e.g.:
            // property("spark.workflowdir")
            // if the modules directory does not exist on the stub, e.g., the file
            // system is not shared, then initialization is not done and the stub 
            // workflow cannot use the module workflow directory properties.
            contractConfiguration.set(Utilities.CONFIGURATION_KEPLER_MODULES_DIR,
                    ProjectLocator.getKeplerModulesDir().getAbsolutePath());
        
        }
    }    
    
    /** Get the Spark Context. */
    private void _getContext() throws IllegalActionException {

        String typeStr = ((StringToken)_director.startServerType.getToken()).stringValue().trim();
        
        if(typeStr.isEmpty() || typeStr.equals("default")) {
            
            if(!_masterHostAndPortStr.trim().isEmpty()) {
                throw new IllegalActionException(_director,
                    "masterHostAndPort must not be specified if startServerType is default (or empty).");
            }
            
            _context = SparkConnection.getContext(
                "",
                _numSameJVMWorkers,
                _driverMemoryStr,
                _container.getFullName());        
        
            // see if we were able to get a context
            if(_context == null) {
                _sameJVM = false;
            } else {
                // see if the context is to a local master.
                String sparkMasterStr = _context.getConf().get("spark.master");
                _sameJVM = sparkMasterStr.startsWith("local");
            }
            
        } else if(typeStr.equals(DDPDirector.SAME_JVM_STRING)) {
            
            _context = SparkConnection.getContext("local",
                _numSameJVMWorkers,
                _driverMemoryStr,
                _container.getFullName());
            
            _sameJVM = true;
            
            if(_context == null) {
                throw new IllegalActionException(_director,
                        "Unable to create Spark context in same JVM.");
            }
            
        } else if(typeStr.equals(DDPDirector.DISTRIBUTED_STRING)) {
                       
            _sameJVM = false;

            InetSocketAddress socketAddress;
            if(_masterHostAndPortStr.trim().isEmpty()) {
                socketAddress = SparkConnection.getDefaultMasterSocketAddress();
            } else {
                socketAddress = SparkConnection.getMasterAndPortAddress(_masterHostAndPortStr);
            }
            
            // see if we can connect to standalone server
            boolean connected = false;
            try(Socket socket = new Socket();) {
                socket.connect(socketAddress, _CONNECT_TIMEOUT);
                connected = true;
            } catch(IOException e) {
                //System.err.println("IOException connecting to " + socketAddress +
                    //": " + e.getMessage());
                connected = false;
            }

            // start standalone server since could not connect
            if(!connected) {

                File configDirFile = new File(_configDirStr);
                String parentDirStr = configDirFile.getParent();
                String startScriptStr;
                String stopScriptStr;
                
                ConfigurationProperty engineProperty = 
                    Utilities.getEngineProperty(SPARK_ENGINE_NAME, _director);
                
                ConfigurationProperty scriptsProperty = engineProperty.getProperty("Server.Scripts");
                if(scriptsProperty == null) {
                    throw new IllegalActionException(_director, "Server Scripts not found in configuration file.");
                }
                
                final String pathStr = parentDirStr + File.separator + "sbin" +
                        File.separator;

                String startName;
                String stopName;
               
                // see if we are running under a resource manager
                if(Utilities.isExecutingUnderResourceManager()) {
                    System.out.println("Running under cluster resource manager.");
                    startName = "ClusterStart";
                    stopName = "ClusterStop";
                } else {
                    startName = "Start";
                    stopName = "Stop";                      
                }

                ConfigurationProperty property = scriptsProperty.getProperty(startName);
                if(property == null) {
                    throw new IllegalActionException(_director, "No start script property in configuration file: " + startName);
                }
                startScriptStr = pathStr + property.getValue();
                

                property = scriptsProperty.getProperty(stopName);
                if(property == null) {
                    throw new IllegalActionException(_director, "No stop script property in configuration file: " + startName);
                }
                stopScriptStr = pathStr + property.getValue();
                
                System.out.println("Spark server start script is " + startScriptStr);
                System.out.println("Spark server stop script is " + stopScriptStr);

                if(_checkServer(socketAddress, startScriptStr)) {
                    _stopScriptsForShutdown.add(stopScriptStr);
                }
            }
            
            _context = SparkConnection.getContext(            
                    _masterHostAndPortStr,
                    _numSameJVMWorkers,
                    _driverMemoryStr,
                    _container.getFullName());            
            
            
        } else {
            throw new IllegalActionException(_director,
                "Unsupported value for startServerType: " + typeStr);
        }        
            
        if(_context == null) {
            throw new IllegalActionException(_director,
                "Unable to create Spark context.");
        }

    }
    
    /** Parse the output from the script that starts the server.
     *  @return If the start script specifies a server URL, returns the
     *  socket address for that URL. Otherwise, returns null.  
     */
    @Override
    protected InetSocketAddress _parseOutputFromStartingServer(InputStream input) throws IOException, IllegalActionException {
        InetSocketAddress retval = null;
        if(Utilities.isExecutingUnderResourceManager()) {
            String hostAndPortStr = SparkConnection.parseOutputFromStartingServer(input);
            if(hostAndPortStr != null) {
                _masterHostAndPortStr = hostAndPortStr;
                retval = SparkConnection.getMasterAndPortAddress(_masterHostAndPortStr);
            }
        }
        return retval;
    }

    /** Get the contract for a DDPPatternActor. */
    private Operator _getContract(DDPPatternActor actor) throws IllegalActionException {

        Operator operator = null;
        
        final String actorName = actor.getName();
        
        final String stubClassName = actor.getExecutionClassName();
        Class<?> stubClass = null;
        
        // see if the stub was set
        if(!stubClassName.isEmpty()) {
            try {
                stubClass = _classLoader.loadClass(stubClassName);
            } catch (ClassNotFoundException e) {
                throw new IllegalActionException(actor, e,
                        "Could not find execution class " + stubClassName); 
            }
        }
        
        final String stubCodeTypeName = actor.getExecutionCodeType();
        if(stubCodeTypeName != null) {
            
            // TODO
            if(!(actor instanceof Map) && !(actor instanceof Reduce)) {
                throw new IllegalActionException(actor, "Code execution not yet supported for this pattern.");
            }
        }
                
        final boolean stubIsSubWorkflow = (stubClass == null) && (stubCodeTypeName == null);
        
        if(actor instanceof DDPDataSource) {
            operator = _getSourceSinkContract((DDPDataSource)actor, true);            
        } else if(actor instanceof Map) {
            
            PairFlatMapFunction stub;
            
            if(stubIsSubWorkflow) {
                // execute sub-workflow as stub
                stub = new KeplerMapStub();
            } else if(stubCodeTypeName != null) {
                stub = new ScriptEngineMapStub();
            } else if(!PairFlatMapFunction.class.isAssignableFrom(stubClass)) {
                throw new IllegalActionException(actor, "Execution class " +
                        stubClassName + " must be a subclass of " +
                        PairFlatMapFunction.class.getName());
            } else {
                System.out.println("Using Map execution class " + stubClassName +
                        " for " + actor.getFullName());

                try {
                    stub = (PairFlatMapFunction) stubClass.newInstance();
                } catch(Throwable t) {
                    throw new IllegalActionException(actor, t,
                            "Error instantiating map class " + stubClass.getName());
                }

            }
            
            
            operator = new MapOperator(stub, actorName);
            
        } else if (actor instanceof Reduce) {
            
            Function2<?,?,?> stub = null;
            PairFlatMapFunction stubNewKeys = null;
            
            if(stubIsSubWorkflow) {
                // execute sub-workflow as stub
                
                // see if reducer is also a combiner
                // TODO
                //if(((BooleanToken)((Reduce)actor).useAsCombiner.getToken()).booleanValue()) {
                //    stubClass = KeplerCombineAndReduceStub.class;
                //} else {          
                    stubNewKeys = new KeplerReduceStub();
                //}
                
            } else if(stubCodeTypeName != null) {
                stubNewKeys = new ScriptEngineReduceStub();
            } else if(Function2.class.isAssignableFrom(stubClass)) {

                System.out.println("Using Reduce execution class " + stubClassName +
                        " for " + actor.getFullName());
                
                try {
                    stub = (Function2<?, ?, ?>) stubClass.newInstance();
                } catch(Throwable t) {
                    throw new IllegalActionException(actor, t,
                            "Error instantiating reduce class " + stubClass.getName());
                }

            } else if(PairFlatMapFunction.class.isAssignableFrom(stubClass)) {
                
                System.out.println("Using Reduce execution class " + stubClassName +
                        " for " + actor.getFullName());

                try {
                    stubNewKeys = (PairFlatMapFunction) stubClass.newInstance();
                } catch(Throwable t) {
                    throw new IllegalActionException(actor, t,
                            "Error instantiating map class " + stubClass.getName());
                }
                
            } else {            
                throw new IllegalActionException(actor, "Execution class " +
                        stubClassName + " must be a subclass of " +
                        Function2.class.getName() + " or " +
                        PairFlatMapFunction.class.getName());
              }
        
            if(stub != null) {
                operator = new ReduceOperator(stub, actorName);
            } else {
                operator = new ReduceOperator(stubNewKeys, actorName);
            }

        } else if(actor instanceof Cross) {
            
            PairFlatMapFunction stub;
            
            // see if execution class was set
            if(stubClass == null) {
                // execute sub-workflow as stub
                stub = new KeplerCrossStub();
            } else if(!PairFlatMapFunction.class.isAssignableFrom(stubClass)) {
                throw new IllegalActionException(actor, "Execution class " +
                        stubClassName + " must be a subclass of " +
                        PairFlatMapFunction.class.getName());
            } else {
                System.out.println("Using Cross execution class " + stubClassName +
                        " for " + actor.getFullName());

                try {
                    stub = (PairFlatMapFunction) stubClass.newInstance();
                } catch(Throwable t) {
                    throw new IllegalActionException(actor, t,
                            "Error instantiating cross class " + stubClass.getName());
                }
            }
            
            operator = new CrossOperator(stub, actorName);

        } else if(actor instanceof CoGroup) {
            
            PairFlatMapFunction stub;
            
            // see if execution class was set
            if(stubClass == null) {
                // execute sub-workflow as stub
                stub = new KeplerCoGroupStub();
            } else if(!PairFlatMapFunction.class.isAssignableFrom(stubClass)) {
                throw new IllegalActionException(actor, "Execution class " +
                        stubClassName + " must be a subclass of " +
                        PairFlatMapFunction.class.getName());
            } else {
                System.out.println("Using CoGroup execution class " + stubClassName +
                        " for " + actor.getFullName());

                try {
                    stub = (PairFlatMapFunction) stubClass.newInstance();
                } catch(Throwable t) {
                    throw new IllegalActionException(actor, t,
                            "Error instantiating cogroup class " + stubClass.getName());
                }
            }

            operator = new CoGroupOperator(stub, actorName);

        } else if(actor instanceof Match) {
            
            PairFlatMapFunction stub;
            
            // see if execution class was set
            if(stubClass == null) {
                // execute sub-workflow as stub
                stub = new KeplerMatchStub();
            } else if(!PairFlatMapFunction.class.isAssignableFrom(stubClass)) {
                throw new IllegalActionException(actor, "Execution class " +
                        stubClassName + " must be a subclass of " +
                        PairFlatMapFunction.class.getName());
            } else {
                System.out.println("Using Match execution class " + stubClassName +
                        " for " + actor.getFullName());

                try {
                    stub = (PairFlatMapFunction) stubClass.newInstance();
                } catch(Throwable t) {
                    throw new IllegalActionException(actor, t,
                            "Error instantiating match class " + stubClass.getName());
                }
            }

            operator = new JoinOperator(stub, actorName);
            
        } else if(actor instanceof DDPDataSink) {
            operator = _getSourceSinkContract((DDPDataSink)actor, false);                        
        } else {
            throw new IllegalActionException(_director, "Cannot determine Operator type for " +
                    actor.getFullName());
        }
        
        // set the number of parallel instances.
        int numInstances = actor.getDegreeOfParallelism();
        
        // if less than 1, use the director's value
        if(numInstances <= 0) {
            numInstances = _degreeOfParallelism;
        }
        // check if greater than num local workers
        else if(numInstances > _sparkParallelism) {
            System.err.println("WARNING: degree of parallelism for " +
                    actor.getName() + " is " + numInstances + ", which is " +
                    "greater than Spark's parallelism " +
                    "(" + _sparkParallelism + ").");
            if(_sameJVM) {
                System.err.println("Increase the number of local workers to achieve " +
                    "this degree of parallelism.");
            }
        }
        
        operator.setDegreeOfParallelism(numInstances);
        
        // set any job arguments if an execution class is used (not a sub-workflow)
        final Configuration configuration = operator.getParameters();
        if(!stubIsSubWorkflow) {
            for(java.util.Map.Entry<String,String> entry : _jobArgumentsMap.entrySet()) {
                // XXX this assumes they are all strings.
                configuration.set(entry.getKey(), entry.getValue());
            }
        }

        // add any actor parameters to the contract configuration
        final java.util.Map<String,String> contractParameters = actor.getParameters();
        final java.util.Map<String,String> paraNames = actor.getParaImplNames(_engineName);
        for(java.util.Map.Entry<String, String> entry : contractParameters.entrySet()) {
            String keplerParaName = entry.getKey();
            if (paraNames.get(keplerParaName) != null) {
                configuration.set(paraNames.get(keplerParaName), entry.getValue());
            } else { 
                configuration.set(Utilities.CONFIGURATION_KEPLER_PARA_PREFIX + "::" + keplerParaName, entry.getValue());
            }
        }
        //set print execution info parameter
        configuration.setBoolean(Utilities.CONFIGURATION_KEPLER_PRINT_EXE_INFO, actor.getPrintExeInfo());

        // set the same JVM flag.
        if(_sameJVM) {
            configuration.setBoolean(Utilities.CONFIGURATION_KEPLER_SAME_JVM, true);        
        }
        
        if(stubCodeTypeName != null) {
                        
            String scriptEngineFactoryName = Utilities.getScriptEngineFactoryName(stubCodeTypeName);
            if(scriptEngineFactoryName == null) {
                throw new IllegalActionException(actor,
                        "No script engine has been configured for " + stubCodeTypeName);
            }
            
            if(!_sameJVM) {
                List<String> jars = Utilities.getJarsForLanguage(stubCodeTypeName);
                _additionalJars.addAll(jars);
            }
            
            configuration.set(Utilities.CONFIGURATION_KEPLER_SCRIPT_ENGINE_FACTORY_NAME, scriptEngineFactoryName);
            
            configuration.set(Utilities.CONFIGURATION_KEPLER_STUB_CODE, actor.getExecutionCode());     
            
        }
        
        return operator;
    }
    
    /** Get a contract for a source or sink.
     *  @param actor the Pactor
     *  @param parameterName the name of the parameter containing the class name.
     *  @param input if true, get a source contract. if false, get a sink contract.
     */
    private Operator _getSourceSinkContract(AtomicPatternActor actor, boolean input) throws IllegalActionException {
        
        final String actorName = actor.getName();
        
        String className = actor.getFormatClassName(_engineName);
        if(className == null) {
            throw new IllegalActionException(_director, "Could not find format class name for " +
                    " actor " + actorName);
        }
        
        // try to get the class
        Class<?> clazz;
        try {
            clazz = _classLoader.loadClass(className);
        } catch (ClassNotFoundException e) {
            throw new IllegalActionException(actor, "Format type " + className +
                " was not found in the format types configurations or is a class not on the classpath.");
        }
        
        Operator operator = null;
        
        if(input) {
                        
            if(FileInputFormat.class.isAssignableFrom(clazz)) {
                operator = new FileDataSource((Class<? extends FileInputFormat<?,?>>) clazz,
                    ((DDPDataSource) actor).getPathAsURI(), actorName);
            } else if(TokenSource.class.isAssignableFrom(clazz)) {

                ArrayToken token = DDPDataSource.getToken(actor.getFullName());
                
                if(token == null) {
                    throw new IllegalActionException(actor, 
                        "No input token found for source actor " + actorName);
                }

                operator = new TokenSource(token, actor.getName());
                
            } else {
                throw new IllegalActionException(actor, "Unsupported type of format class: " +
                        clazz.getName());
            }
            
            ((DataSource) operator).setContext(_context);

            
        } else {
            
            if(FileOutputFormat.class.isAssignableFrom(clazz)) {
                
                operator = new FileDataSink((Class<? extends FileOutputFormat<?,?>>) clazz,
                    ((DDPDataSink)actor).getPathAsURI(), actorName);
            
            } else if(TokenSink.class.isAssignableFrom(clazz)) {
                
                operator = new TokenSink(actor.getFullName(), actorName);
                
            } else if(clazz == NullOutputFormat.class) {
                operator = new NullSink(actorName);
            } else {
                throw new IllegalActionException(actor, "Unsupported type of format class: " +
                        clazz.getName());
            }
        }
        
        return operator;

    }

    ///////////////////////////////////////////////////////////////////
    ////                      private fields                     //////

    /** The name of the spark engine. */
    public final static String SPARK_ENGINE_NAME = "Spark";
    
    /**  */
    private JavaSparkContext _context;

    private Set<Operator> _inputsAlreadyDone = new HashSet<Operator>();
    
    /** A Spark job that can be cancelled. */
    private SparkJob _job;

        /**
         * A list of scripts to execute to stop each spark server that is
         * started.
         */
        private final static Set<String> _stopScriptsForShutdown = Collections
                        .synchronizedSet(new HashSet<String>());
        
        /** The number of workers for the local context. */
        private int _numSameJVMWorkers = SparkConnection.DEFAULT_NUM_LOCAL_WORKERS;
        
        private String _driverMemoryStr = SparkConnection.DEFAULT_DRIVER_MEMORY;
        
        private String _masterHostAndPortStr = "";
        
        private int _sparkParallelism;
        
}