/* Play a Recording from a Queryable.
 *
 * Copyright (c) 2015 The Regents of the University of California.
 * All rights reserved.
 *
 * '$Author: crawl $'
 * '$Date: 2017-07-07 16:27:30 +0000 (Fri, 07 Jul 2017) $' 
 * '$Revision: 34587 $'
 * 
 * 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.provenance;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import javax.swing.SwingUtilities;

import org.kepler.objectmanager.lsid.KeplerLSID;

import ptolemy.actor.Actor;
import ptolemy.actor.AtomicActor;
import ptolemy.actor.CompositeActor;
import ptolemy.actor.Director;
import ptolemy.actor.FiringEvent;
import ptolemy.actor.IOPort;
import ptolemy.actor.IOPortEvent;
import ptolemy.actor.IOPortEventListener;
import ptolemy.actor.IORelation;
import ptolemy.actor.Manager;
import ptolemy.actor.NoRoomException;
import ptolemy.actor.Receiver;
import ptolemy.actor.TypedIOPort;
import ptolemy.actor.gui.Configuration;
import ptolemy.actor.gui.Effigy;
import ptolemy.actor.gui.ModelDirectory;
import ptolemy.actor.gui.PtolemyEffigy;
import ptolemy.actor.parameters.PortParameter;
import ptolemy.data.Token;
import ptolemy.data.expr.Parameter;
import ptolemy.domains.pn.kernel.PNDirector;
import ptolemy.kernel.CompositeEntity;
import ptolemy.kernel.Entity;
import ptolemy.kernel.Relation;
import ptolemy.kernel.util.Attribute;
import ptolemy.kernel.util.IllegalActionException;
import ptolemy.kernel.util.NameDuplicationException;
import ptolemy.kernel.util.NamedObj;
import ptolemy.vergil.basic.KeplerDocumentationAttribute;
import ptolemy.vergil.kernel.attributes.TextAttribute;

/** A class to convert provenance formats by reading from a
 *  a Queryable and writing to a Recording.
 * 
 *  @see Queryable
 *  @see Recording
 *  
 *  @author Daniel Crawl
 *  @version $Id: RecordPlayer.java 34587 2017-07-07 16:27:30Z crawl $
 * 
 */
public class RecordPlayer {

    /** Create a new RecordPlayer using a Queryable as the source
     *  and no Recording output. This is useful if actors within
     *  workflow will be re-executed.
     *  @see RecordPlayer#setExecuteGraphicalActors
     */
    public RecordPlayer(Queryable queryable) {
        this(queryable, null);
    }
    
    /** Create a new RecordPlayer using a Queryable as the source,
     *  and Recording for the output.
     */
    public RecordPlayer(Queryable queryable, Recording recording) {
        _queryable = queryable;
        _recording = recording;
    }
            
    /** Close any effigies created while replaying workflows. */
    public static void closeAllEffigies() throws IllegalActionException, NameDuplicationException {
        synchronized(_effigies) {
            for(Effigy effigy : _effigies.values()) {
                effigy.closeTableaux();
                effigy.setContainer(null);
            }
        }
    }
    
    /** Re-execute an actor.
     * @param className the class name of the actor to re-execute.
     */
    public void executeActor(String className) {
        _executeActorNames.add(className);
    }
    
    /** Get the queryable. */
    public Queryable getQueryable() {
        return _queryable;
    }
    
    /** Play a specific execution. */
    public void play(Integer execId) throws QueryException, RecordingException {

        CompositeActor workflow;
        try {
            workflow = (CompositeActor) _queryable.getWorkflowForExecution(execId);
        } catch (Exception e) {
            throw new RecordingException("Error parsing workflow.", e);
        }

        play(execId, workflow);
    }
    
    /** Play a specific execution for a workflow. */
    public void play(Integer execId, final CompositeActor workflow) throws QueryException, RecordingException {
        
        // port events must be played if executing actors.
        if(!_playPortEvents && 
                (!_executeActorNames.isEmpty() || !_executeActorPackages.isEmpty())) {
            throw new QueryException("Must play port events if executing actors.");
        }
        
        _workflow = workflow;
        
        KeplerLSID execLSID = _queryable.getExecutionLSIDForExecution(execId);
        if(execLSID == null) {
            throw new RecordingException("Could not get execution LSID.");
        }

        // NOTE: if the workflow has never been executed, we 
        // create receivers in all the actors so that if the
        // output Recording is using a PortConnector, then
        // PortConnector.sendIdToConnections() will find all
        // the connected ports. 
       
        Manager manager = workflow.getManager();
        if(manager == null) {
            try {
                manager = new Manager(workflow.workspace(), "_Manager");
               workflow.setManager(manager);
            } catch (IllegalActionException e) {
                throw new QueryException("Error setting manager.", e);
            }
        }
        if(manager.getPreinitializeVersion() != workflow.workspace().getVersion()) {
            
            // when creating receivers, an exception can sometimes occur due
            // to not being able to uniquely infer the width of relations.
            // to avoid this, create a top level parameter _defaultInferredWidthTo1
            // with value to true.
            // see ptolemy.actor.RelationWidthInference
            
            Attribute defaultWidth = workflow.getAttribute("_defaultInferredWidthTo1");
            if(defaultWidth == null) {
                try {
                    Parameter defaultWidthParameter = new Parameter(workflow, "_defaultInferredWidthTo1");
                    defaultWidthParameter.setToken("true");
                    defaultWidthParameter.setPersistent(false);
                } catch(IllegalActionException | NameDuplicationException e) {
                    throw new QueryException("Error creating parameter to set default relation width.", e);
                }
            }
            
            List<Actor> actorList = workflow.entityList(Actor.class);
            while(!actorList.isEmpty()) {
                Actor actor = actorList.remove(0);
                // create receivers in atomic actors, and opaque composite actors.
                if(actor instanceof AtomicActor ||
                    (actor instanceof CompositeActor && ((CompositeActor)actor).isOpaque())) {
                    try {
                       //System.out.println("creating receivers for " + actor);
                       actor.createReceivers();
                    } catch (IllegalActionException e) {
                       throw new QueryException("Error creating receivers for " +
                           actor.getFullName(), e);
                   }
               }
               // if the actor is a composite actor, add any contained
               // actors so we create their receivers.
               if(actor instanceof CompositeActor) {
                   actorList.addAll(((CompositeActor)actor).entityList(Actor.class));
                }
            }
        }
        
        Date[] workflowExecutionTimestamps = null;
        
        List<ProvenanceRecorder> recorders = _workflow.attributeList(ProvenanceRecorder.class);
        if(recorders.isEmpty()) {
            Map<String,String> defaults = ProvenanceRecorder.getDefaultsMap();
            try {
                _recorder = new ProvenanceRecorder(_workflow, defaults.get("Provenance Recorder Name"));
            } catch (IllegalActionException | NameDuplicationException e) {
                throw new QueryException("Error creating default provenance recorder in workflow.", e);
            }
        } else {
            _recorder = recorders.get(0);
        }

        /*
        try {
            _recorder.recordingType.setToken(ProvenanceRecorder.DEFAULT_REC_TYPE_IGNORE);
        } catch (IllegalActionException e) {
            throw new RecordingException("Error turning off provenance recording.", e);
        }
        */

        if(_recording != null) {

            // NOTE: must add the recording to the recorder since the
            // recorder access the recorder in _recorderContainer.
            _recorder.addPiggyback(_recording);
                    
            // TODO set the user and machine names, tags, etc.        
            
            if(_recording.regContents()) {
                
                // specification start
                _recording.specificationStart();
            
                // reg contents
                _recordContainerContents(_workflow);
                
                // specification stop
                _recording.specificationStop();
            }
            
            // TODO play evolutions
            
            workflowExecutionTimestamps = _queryable.getTimestampsForExecution(execId);
    
            // execution start
            if(workflowExecutionTimestamps == null || workflowExecutionTimestamps.length != 2 ||
                    workflowExecutionTimestamps[0] == null) {
                System.err.println("WARNING: could not query start/stop timestamps for execution.");
                _recording.executionStart(execLSID);
            } else {
                _recording.executionStart(execLSID, workflowExecutionTimestamps[0]);
            }
        }
        
        boolean createdEffigies = false;
        
        if(!_executeActorNames.isEmpty() || !_executeActorPackages.isEmpty()) {

            if(_recorder != null) {
                workflow.removeInitializable(_recorder);
            }
            
            _createEffigyIfNecessary(execId);
            createdEffigies = true;
            
            try {
                workflow.preinitialize();
            } catch (IllegalActionException e) {
                System.err.println("Error preinitializing the workflow: " + e.getMessage());
            }
            
            try {
                workflow.initialize();
            } catch (IllegalActionException e) {
                System.err.println("Error initializing the workflow: " + e.getMessage());
            }
        }
        
        try {
            // replay the execution events (actor fires and port read/writes)
            _replayExecution(execId, execLSID);
            
            if(!_executeActorNames.isEmpty() || !_executeActorPackages.isEmpty()) {
                
                // call wrapup without a timeout of 5 seconds since
                // NondeterministicMerge can block in wrapup.
                ExecutorService service = Executors.newSingleThreadExecutor();
                try {
                    service.submit(new Runnable() {
                        @Override
                        public void run() {
                            try {
                                workflow.wrapup();
                            } catch (IllegalActionException e) {
                                System.err.println("Error wrapping-up the workflow: " + e.getMessage());
                            }
                        }
                    }).get(5000, TimeUnit.MILLISECONDS);
                } catch (InterruptedException | ExecutionException | TimeoutException e) {
                    System.err.println("WARNING: Timeout wrapping up replay.");
                } finally {
                    service.shutdownNow();
                }                
            }
            
            if(_recording != null) {
                // execution stop
                if(workflowExecutionTimestamps == null || workflowExecutionTimestamps[1] == null) {
                    _recording.executionStop(execLSID);
                } else {
                    _recording.executionStop(execLSID, workflowExecutionTimestamps[1]);
                }
            
                // see if there was an error.
                if(_queryable.isErrorForExecution(execLSID)) {
        
                    String errorStr = _queryable.getErrorForExecution(execLSID);
                    
                    // TODO need to get the error source and throwable
                    Throwable throwable = new Throwable(errorStr);
                    _recording.executionError(_workflow, throwable, execLSID);
                }
            }
        } finally {
            // if we are replaying actors, set any created effigies to
            // be not modified. this prevents a 'Save Changes?' dialog from
            // appearing when closing windows created by replayed actors.
            if(createdEffigies) {
                final Effigy effigy = _effigies.get(execId);
                if(effigy != null) {
                    // NOTE: perform this in the event thread in case there
                    // are effigies about to be created (new windows).
                    SwingUtilities.invokeLater(new Runnable() {
                        @Override
                        public void run() {
                            _setEffigiesToNotModified(effigy);
                        }
                    });
                }
            }
        }
    }
    
    /** Execute graphical actors in workflow. */
    public void setExecuteGraphicalActors(boolean execute) {
        
        // TODO find a list of graphical actors
        
        if(execute) {
            _executeActorPackages.add("ptolemy.actor.lib.gui");            
            _executeActorPackages.add("org.kepler.actor.gui");
            _executeActorNames.add("pl.psnc.kepler.common.actor.MultipleTabDisplay");
        } else {           
            _executeActorPackages.remove("ptolemy.actor.lib.gui");            
            _executeActorPackages.remove("org.kepler.actor.gui");            
            _executeActorNames.remove("pl.psnc.kepler.common.actor.MultipleTabDisplay");
        }
    }
    
    /** Set if port events are to be played. */
    public void setPlayPortEvents(boolean playPortEvents) {
        _playPortEvents = playPortEvents;
    }

    ///////////////////////////////////////////////////////////////////
    ////                       private methods                     ////

    /** Create an effigy for a workflow execution if one does not already exist. */
    private void _createEffigyIfNecessary(Integer execId) {
        
        synchronized(_effigies) {
        
            PtolemyEffigy effigy = _effigies.remove(execId);
            if(effigy != null) {
                //System.out.println("found existing cached effigy for run " + execId + "; closing open tableaux");
                // close existing windows opened.
                effigy.closeTableaux();
                effigy.setModel(null);
                try {
                    effigy.setContainer(null);
                    //System.out.println("removing effigy " + effigy.getFullName());
                } catch (IllegalActionException | NameDuplicationException e) {
                    System.err.println("Error removing effigy: " + e.getMessage());
                }
            }
            
            // see if the Configuration has an effigy for the workflow.
            effigy = (PtolemyEffigy) Configuration.findEffigy(_workflow);
            if(effigy != null) {
                // close existing windows opened.
                effigy.closeTableaux();
                effigy.setModel(null);
                try {
                    effigy.setContainer(null);
                    //System.out.println("removing effigy " + effigy.getFullName());
                } catch (IllegalActionException | NameDuplicationException e) {
                    System.err.println("Error removing effigy: " + e.getMessage());
                }
            }
                        
            // make sure there is at least one Configuration. there are none when
            // running headless.
            if(!Configuration.configurations().isEmpty()) {
                ptolemy.actor.gui.Configuration configuration = 
                        (ptolemy.actor.gui.Configuration) Configuration.configurations().iterator().next();

                try {
                    ModelDirectory directory = configuration.getDirectory();
                    //System.out.println("Creating effigy for " +
                        //_workflow.getName() + " in " + directory.getFullName());
                    
                    effigy = new PtolemyEffigy(directory,
                            directory.uniqueName(_workflow.getName()));
                    //System.out.println("created effigy " + effigy.getFullName());
                    effigy.setModel(_workflow);                                                
                    effigy.identifier.setExpression(effigy.getName());
                } catch(IllegalActionException | NameDuplicationException e) {
                    System.err.println("Error creating effigy: " + e.getMessage());
                    return;
                }
            }
            
            if(effigy != null) {
                _effigies.put(execId, effigy);
            }
        }
    }
    
    /** Output the model contents.
     *  TODO copied from ProvenanceRecorder
     */ 
    private void _recordContainerContents(NamedObj namedObj)
        throws RecordingException
    {
        // whether to recurse based on Recording register method.
        boolean recurse = true;
                
        if(namedObj instanceof Actor)
        {            
            recurse = _recording.regActor((Actor)namedObj);
        }
        else if(namedObj instanceof Director)
        {
            recurse = _recording.regDirector((Director)namedObj);       
        }
        else if(namedObj instanceof TypedIOPort)
        {
            TypedIOPort port = (TypedIOPort)namedObj;
            _recording.regPort(port);
            
            // do not record information about attributes contained by the port
            recurse = false;
        }
        else if(namedObj instanceof IORelation)
        {
            IORelation rel = (IORelation)namedObj;
            recurse = _recording.regRelation(rel);
        }
        else if(namedObj != _recorder &&
            // NOTE: do not register a connected port parameter as a parameter
            // since during execution, reading a token by the port generates
            // both port and value change events. value change events are
            // treated as a workflow evolution, which can be expensive.
            (!(namedObj instanceof PortParameter) || 
            ((PortParameter)namedObj).getPort().numberOfSources() == 0))
        {
            recurse = _recording.regParameter(namedObj);
        }
        else // namedObj == _recorder
        {
            recurse = false;
        }

        if(recurse)
        {
            // recursively register contents of the current namedObj
            List<NamedObj> contentsList = new LinkedList<NamedObj>();

            boolean isPLOpaque = false;

            if(namedObj instanceof CompositeEntity)
            {
                CompositeEntity composite = (CompositeEntity)namedObj;
                contentsList.addAll(composite.entityList());

                // NOTE: we add the relations last since for each one,
                // regLink() is called and both ends of the link must be
                // registered first.
                contentsList.addAll(composite.relationList());

                // check to see if named obj contains an instance of
                // provenance listener
                List<?> provs = namedObj.attributeList(ProvenanceRecorder.class);
                if(provs != null && provs.size() > 0 && namedObj != namedObj.toplevel())
                {
                    isPLOpaque = true;
                }
                else if((namedObj instanceof CompositeActor) &&
                    !ProvenanceRecorder.containsSupportedDirector((CompositeActor)namedObj))
                {
                    isPLOpaque = true;
                }
                
            }

            // add the ports to the beginning
            if(namedObj instanceof Entity)
            {
                contentsList.addAll(0, ((Entity<?>)namedObj).portList());
            }

            // add the attributes to the beginning
            contentsList.addAll(0, namedObj.attributeList());

            //System.out.println("is pl opaque: " + isPLOpaque);
            
            for(NamedObj containedNamedObj : contentsList)
            {
                if(isPLOpaque &&
                    // do not record contents of these types of objects:
                    ((containedNamedObj instanceof Actor) ||
                     (containedNamedObj instanceof Director) ||
                     (containedNamedObj instanceof IORelation) ||
                     (containedNamedObj instanceof TextAttribute) ||
                     (containedNamedObj instanceof ProvenanceRecorder) ||
                     (containedNamedObj instanceof
                        KeplerDocumentationAttribute)))
                {
                    //_Debug("not recursing down " +
                        //containedNamedObj.getFullName());
                }
                else
                {
                    _recordContainerContents(containedNamedObj);
                }
            }

            // now that all the ports and relations have been registered
            // (if namedObj is a CompositeEntity), it is safe to register
            // the links (since each end point must already registered).

            if(!isPLOpaque && (namedObj instanceof CompositeEntity))
            {
                // iterate over the contained relations
                List<?> relationList = ((CompositeEntity)namedObj).relationList();
                Iterator<?> relationIter = relationList.iterator();
                while(relationIter.hasNext())
                {
                    Relation curRelation = (Relation)relationIter.next();

                    ///if(!adding)
                    //{
                        //containedName = oldName + "." +
                            //curRelation.getName();
                    //}
                    _recordLinksInRelation(curRelation);
                }
            }
        }
    }

    /** Record all the links within a relation.
     *  TODO copied from ProvenanceRecorder
     */
    private void _recordLinksInRelation(Relation relation)
        throws RecordingException
    {
        // iterate over the object linked in the current relation
        List<?> linkedObjList = relation.linkedObjectsList();
        Iterator<?> linkedObjsIter = linkedObjList.iterator();
        while(linkedObjsIter.hasNext())
        {
            NamedObj linkedObj = (NamedObj)linkedObjsIter.next();

            // see if current relation is linked to a port
            if(linkedObj instanceof TypedIOPort)
            {
                _recording.regLink(linkedObj, relation);                 
            }
            // otherwise the link is between two relations.
            // we do not want to register this link twice,
            // e.g., a <--> b and b <--> a
            // so only perform registration if the current
            // relation's name lexigraphically precedes the
            // linked relation's name.
            else if(relation.getName().compareTo(
                    linkedObj.getName()) < 0)
            {
                _recording.regLink(relation, linkedObj);                
            }
        }
    }
    
    /** Replay a workflow execution. */
    private void _replayExecution(int execId, KeplerLSID execLSID) throws QueryException, RecordingException {        
        
        List<Event> events = new ArrayList<Event>();
        
        long startNano = System.nanoTime();
        
        // collect the provenance events to replay. this consists of
        // firing start, firing stop, port read, and port write events.
        
        System.out.print("Replaying..collecting:..");
        
        // collect the firings
        int eventCount = 0;
        List<Integer> firings = _queryable.getActorFirings(execId);
        for(Integer firingId : firings) {
                        
            Date[] timestamps = _queryable.getTimestampsForActorFiring(firingId);
            if(timestamps == null || timestamps.length != 2 ||
                timestamps[0] == null || timestamps[1] == null) {
                System.err.println("WARNING: missing timestamps for firing: " + firingId);
                continue;
            }
            
            // add the firing start
            Event event = new Event(EventType.FiringStart, timestamps[0]);
            event.firingId = firingId;
            events.add(event);
            
            eventCount++;
            if(eventCount % 10000 == 0) {
                System.out.print(eventCount + "..");
            }

            // add the firing stop
            event = new Event(EventType.FiringStop, timestamps[1]);            
            event.firingId = firingId;
            events.add(event);
            
            eventCount++;
            if(eventCount % 10000 == 0) {
                System.out.print(eventCount + "..");
            }

        }
        
        // collect the port events
        if(_playPortEvents) {
            List<Integer> tokens = _queryable.getTokensForExecution(execId, false);
            for(Integer tokenId : tokens) {
                
                // add the port write for this token
                Date writeTimestamp = _queryable.getTimestampForTokenWrite(tokenId);
                if(writeTimestamp == null) {
                    System.err.println("WARNING: missing timestamp for token write: " + tokenId);
                    continue;
                }
                
                Event event = new Event(EventType.PortWrite, writeTimestamp);
                event.portEventId = tokenId;
    
                List<Integer> writeFirings = _queryable.getActorFiringForToken(tokenId, false);
                event.firingId = writeFirings.get(0);
                
                events.add(event);
    
                eventCount++;
                if(eventCount % 10000 == 0) {
                    System.out.print(eventCount + "..");
                }
    
                // add any port reads for this token
                Date[] timestamps = _queryable.getTimestampsForTokenRead(tokenId);
                if(timestamps != null && timestamps.length > 0) {
                    List<Integer> readFirings = _queryable.getActorFiringForToken(tokenId, true);
                    if(timestamps.length != readFirings.size()) {
                        System.err.println("WARNING: number of timestamps for token read " +
                                "does not match number of firings for token read.");
                        continue;
                    }
                    Integer[] readFiringArray = readFirings.toArray(
                        new Integer[readFirings.size()]);
                    Arrays.sort(readFiringArray);
                    
                    int i = 0;
                    for(Date readTimestamp : timestamps) {
                        event = new Event(EventType.PortRead, readTimestamp);
                        event.portEventId = tokenId;
                        event.firingId = readFiringArray[i];
                        i++;
                        events.add(event);
                        
                        eventCount++;
                        if(eventCount % 10000 == 0) {
                            System.out.print(eventCount + "..");
                        }
                    }
                }            
            }
        }
        
        // sort the events.
        // we perform 3 sorts to get the events in order that:
        //
        // 1. the events are sorted chronologically (primary key)
        //
        // 2. events can occur at the same time, so sort on
        // firing id (secondary key).
        //
        // 3. events can occur at the same time and with the
        // same firing id, e.g., a port write and firing stop,
        // so sort on event type order (tertiary key).
        //
        // the 3 sorts are performed sorting the least-significant key
        // first: tertiary key (event type), secondary key (firing id),
        // and primary key (event timestamp).
        
        System.out.print("sorting..");
        
        if(_playPortEvents) {
            Collections.sort(events, new Comparator<Event>() {
                @Override
                public int compare(Event o1, Event o2) {
                    return o1.type.compareTo(o2.type);
                }
            });        
        }
        
        //System.out.print("sort2..");

        Collections.sort(events, new Comparator<Event>() {
            @Override
            public int compare(Event o1, Event o2) {
                return o1.firingId.compareTo(o2.firingId);
            }
        });

        //System.out.print("sort3..");

        Collections.sort(events, new Comparator<Event>() {
            @Override
            public int compare(Event o1, Event o2) {
                return o1.timestamp.compareTo(o2.timestamp);
            }
        });

        System.out.print("processing " + events.size() + "events..");

        /*
        for(Event event : events) {
            System.out.println(event);
        }
        */
        
        // replay the events
        
        Map<Integer, Firing> firingMap = new HashMap<Integer, Firing>();
        eventCount = 0;
        for(Event event : events) {
            
            eventCount++;
            if(eventCount % 1000 == 0) {
                System.out.print(eventCount + "..");
            }

            switch(event.type) {
            case FiringStart:
                Firing firing = new Firing(event.firingId, event.timestamp);
                firingMap.put(event.firingId, firing);
                break;
            case FiringStop:
                firing = firingMap.remove(event.firingId);
                firing.stop(event.timestamp);
                break;
            case PortRead:
                firing = firingMap.get(event.firingId);
                if(firing == null) {
                    System.out.println("null firing " + event.firingId +
                        " for pe " + event.timestamp);
                }
                firing.replayPortEvent(event.portEventId, true, event.timestamp);                
                break;
            case PortWrite:
                firing = firingMap.get(event.firingId);
                /*if(firing == null) {
                    System.out.println("null firing " + event.firingId +
                            " for pe " + event.timestamp);
                }*/
                firing.replayPortEvent(event.portEventId, false, event.timestamp);
                break;
            default:
                System.err.println("WARNING: unknown type of event: " + event.type);
                break;            
            }                            }
        
        if(!firingMap.isEmpty()) {
            System.err.println("WARNING: Firings not stopped: " + firingMap.size());
        }
        
        System.out.print("done.");
        double elapsed = (System.nanoTime() - startNano) / 1000000;
        if(elapsed >= 5000) {
            System.out.print(" elapsed time = " + elapsed + " ms");
        }
        System.out.println();
    }

    /** Set an effigy and any contained effigies to be not modified. */
    private void _setEffigiesToNotModified(Effigy effigy) {
        //System.out.println("setting not modified for : " + effigy.getFullName());
        effigy.setModified(false);
        for(Effigy containedEffigy : effigy.entityList(Effigy.class)) {
            _setEffigiesToNotModified(containedEffigy);
        }
    }

    /** A provenance event to replay. */
    private static class Event {
        
        /** Create a new event with a type and timestamp. */
        public Event(EventType type, Date timestamp) {
            this.type = type;
            this.timestamp = timestamp;
        }

        /** Get a string representation of the event. */
        @Override
        public String toString() {
            return timestamp.toString() + " " + type.toString() + " f=" + firingId;
        }
        
        /** The type of event. */
        public EventType type;
        
        /** The firing id. If the event is a port read or write,
         *  this value is the firing id in which the port event
         *  occurs.
         */
        public Integer firingId;
        
        /** The port event id. If the event is a firing start or
         *  stop, this is null.
         */
        public Integer portEventId;
        
        /** The timestamp of the event. */
        public Date timestamp;
    }
    
    /** A class that represents an actor firing. */
    private class Firing {
        
        public Firing(Integer firingId, Date startTimestamp) throws QueryException, RecordingException {  
            
            _firingId = firingId;
            
            String actorName = _queryable.getActorName(firingId);
            //System.out.println(firingId + " " + actorName);
            
            if(actorName == null) {
                throw new QueryException(
                    "WARNING: Could not find actor name for firing " +
                    firingId);
            }
                        
            // remove the leading "."
            actorName = actorName.substring(1);
            
            // start firing
            _actor = (Actor) _workflow.getEntity(actorName);
            if(_actor == null) {
                System.err.println("WARNING: could not find actor " + actorName +
                        " in workflow.");
                return;
            }
            
            //System.out.println("    " + startTimestamp +
                //" start firing " + firingId +
                //": " + _actor.getFullName());

            _director = _actor.getDirector();
            if(_recording != null) {
                FiringEvent firingEvent = new FiringEvent(_director,
                    _actor,
                    FiringEvent.BEFORE_ITERATE);
                _recording.actorFire(firingEvent, startTimestamp);
            }
            
            if(_executeActorNames.contains(_actor.getClass().getName()) ||
                _executeActorPackages.contains(_actor.getClass().getPackage().getName())) {
                _executeActor = true;                                
            }
            
            if(_actor.getExecutiveDirector() instanceof PNDirector) {
                _directorIsPN = true;
            }
        }
        
        /** Replay a port event and stop the actor firing if there are
         *  no more port events that occurred during this firing.
         *  @param tokenId the port event id to replay
         *  @param read If true, the port event is a read, otherwise
         *  port event is a write.
         *  @return Returns true if there are no more port events
         *  for this firing. Returns false if there are more port events.
         */
        public void replayPortEvent(Integer tokenId, boolean read, Date timestamp)
            throws RecordingException, QueryException {
            
            Token token = _queryable.getToken(tokenId);
            
            if(token == null) {
                System.err.println("WARNING: could not get token " +
                    (read ? "read." : "written."));
                return;
            }
                            
            String portName = null;
            if(read) {
                portName = _queryable.getInputRoleForTokenAndFireId(tokenId, _firingId);
            } else {
                portName = _queryable.getOutputRoleToken(tokenId);
            }
            
            if(portName == null) {
                System.err.println("WARNING: could not get port for token " +
                    (read ? "read " : "written ") + " " + tokenId);
                return;
            }
            
            Integer channel = _queryable.getChannelForToken(tokenId, read, _firingId);
            if(channel == null) {
                System.err.println("WARNING: could not get channel for token " +
                    (read ? "read." : "written."));
                channel = 0;
            }
            
            // portName is the fully-qualified name; get just the 
            // last part
            portName = portName.substring(portName.lastIndexOf('.') + 1);
            IOPort port = (IOPort) ((Entity<?>) _actor).getPort(portName);
        
            //System.out.println("    " + timestamp +
                //(read ? " read " : " write ") +
                //"firing " + _firingId); 
        
            List<IOPortEventListener> portListeners = port.getIOPortEventListeners();
            if(_recording != null || !portListeners.isEmpty()) {
                IOPortEvent event = new IOPortEvent(port,
                        read ? IOPortEvent.GET_BEGIN : IOPortEvent.SEND_BEGIN,
                        channel,
                        true,
                        read ? null : token); 
                
                if(_recording != null) {
                    _recording.portEvent(event, timestamp);
                }
                
                for(IOPortEventListener listener: portListeners) {
                    try {
                        listener.portEvent(event);
                    } catch (IllegalActionException e) {
                        System.err.println("Port event error: " + e.getMessage());
                    }
                }

                event = new IOPortEvent(port,
                        read ? IOPortEvent.GET_END : IOPortEvent.SEND_END,
                        channel,
                        true,
                        token);     
                
                if(_recording != null) {
                    _recording.portEvent(event, timestamp);
                }

                for(IOPortEventListener listener: portListeners) {
                    try {
                        listener.portEvent(event);
                    } catch (IllegalActionException e) {
                        System.err.println("Port event error: " + e.getMessage());
                    }
                }

            }
            
            if(_executeActor && read) {
                                                
                Receiver[][] receivers = port.getReceivers();
                if(receivers[channel] == null) {
                    System.err.println("missing receiver for channel " + channel + " in " + port.getFullName());
                } else {
                    try {
                        //System.out.println("putting token inside " + port.getFullName());
                        receivers[channel][0].put(token);
                    } catch (NoRoomException | IllegalActionException e) {
                        System.err.println("Error putting token in " + port.getFullName() + ": " + e.getMessage());
                    }
                }
                
            }
        }
              
        public void stop(Date stopTimestamp) throws RecordingException {
            
            if(_executeActor) {
                
                // if the director is PN, set a time out before calling
                // iterate, since the last iteration will block waiting
                // for input tokens that will never arrive.
                if(_directorIsPN) {
                    ExecutorService service = Executors.newSingleThreadExecutor();
                    try {
                        service.submit(new Runnable() {
                            @Override
                            public void run() {
                                try {
                                    //System.out.println("executing actor " + _actor);
                                    _actor.iterate(1);
                                } catch (IllegalActionException e) {
                                    System.err.println("Error iterating " + _actor.getFullName() + ": " + e.getMessage());
                                }                            }
                        }).get(_ACTOR_ITERATE_TIMEOUT, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException | ExecutionException | TimeoutException e) {
                        // do nothing since this always occurs during the last iteration
                        //System.err.println("WARNING: Timeout iterating actor.");
                    } finally {
                        service.shutdownNow();
                    }                
                } else {
                    try {
                        //System.out.println("executing actor " + _actor);
                        _actor.iterate(1);
                    } catch (IllegalActionException e) {
                        System.err.println("Error iterating " + _actor.getFullName() + ": " + e.getMessage());
                    }
                }
            }

            //System.out.println("    " + stopTimestamp + 
                //" stop firing " + _firingId +
                //": " + _actor.getFullName());

            if(_recording != null) {
                FiringEvent firingEvent = new FiringEvent(_director,
                    _actor, FiringEvent.AFTER_ITERATE);
                _recording.actorFire(firingEvent, stopTimestamp);
            }
        }
        
        /** Print the actor name. */
        @Override
        public String toString() {
            return _actor.getName();// +
                    //" r=" + _readTokens.size() +
                    //" w=" + _writeTokens.size();            
        }        

        /** The firing id of this firing. */
        private Integer _firingId;
                
        /** The firing actor. */
        private Actor _actor;
        
        /** The director of the firing actor. */
        private Director _director;
        
        /** If true, execute this actor and place any
         *  tokens this actor reads into the input port receivers.
         */
        private boolean _executeActor = false;
        
        /** If true, executive director of actor is PN. */
        private boolean _directorIsPN = false;
    }

    ///////////////////////////////////////////////////////////////////
    ////                       private variables                   ////

    /** The input queryable containing the provenance. */
    private Queryable _queryable;
    
    /** The output recording. */
    private Recording _recording;

    /** The workflow read from provenance. */
    private CompositeActor _workflow;
    
    /** The provenance recorder in the workflow. */
    private ProvenanceRecorder _recorder;

    /** Type of replay event. NOTE: the declaration order determines
     *  how the events are sorted: events occurring at the same time
     *  with the same firing id are sorted based on the type declaration
     *  order. Events at the same time and with the same firing id
     *  (same actor) must have firing start before any port events before
     *  firing stop. By convention port reads occur before port writes,
     *  but this is not strictly necessary.
     */
    private enum EventType { FiringStart, PortRead, PortWrite, FiringStop };

    /** A set of actor class names to re-execute. */
    private final Set<String> _executeActorNames = new HashSet<String>();

    /** A set of actor package names to re-execute. */
    private final Set<String> _executeActorPackages = new HashSet<String>();

    /** A mapping of execution id to effigy. */
    private static final Map<Integer,PtolemyEffigy> _effigies =
        new HashMap<Integer,PtolemyEffigy>();
    
    /** If true, port events are played. */
    private boolean _playPortEvents = true;
    
    /** Timeout (in ms) to wait for an actor to iterate. */
    private static final int _ACTOR_ITERATE_TIMEOUT = 2000;
}