Package ptolemy.actor.process

Class ProcessDirector

  • All Implemented Interfaces:
    java.lang.Cloneable, Executable, Initializable, Changeable, Debuggable, DebugListener, Derivable, ModelErrorHandler, MoMLExportable, Moveable, Nameable
    Direct Known Subclasses:
    CompositeProcessDirector, SysMLADirector, SysMLConcurrentDirector
    public class ProcessDirector
extends Director
    The base class for directors for the process oriented domains. It provides default implementations for methods that are common across such domains.

    In the process oriented domains, the director controlling a model needs to keep track of the state of the model. In particular it needs to maintain an accurate count of the number of active processes under its control and any processes that are blocked for whatever reason (trying to read from an empty channel as in PN). These counts, and perhaps other counts, are needed by the director to control and respond when deadlock is detected (no processes can make progress), or to respond to requests from higher in the hierarchy.

    The methods that control how the director detects and responds to deadlocks are _areActorsDeadlocked() and _resolveDeadlock(). These methods should be overridden in derived classes to get domain-specific behaviour. The implementations given here are trivial and suffice only to illustrate the approach that should be followed.

    This base class is not sufficient for executing hierarchical, heterogeneous models. In order to accommodate hierarchical, heterogeneity the subclass CompositeProcessDirector must be used.

    Since:
    Ptolemy II 0.2
    Version:
    $Id$
    Author:
    Mudit Goel, Neil Smyth, John S. Davis II
    See Also:
    Director
    Pt.AcceptedRating:
    Yellow (mudit)
    Pt.ProposedRating:
    Green (mudit)

    • Field Detail

      • _newActorThreadList

        protected java.util.LinkedList _newActorThreadList
        A list of threads created but not started.

      • _notDone

        protected boolean _notDone
        A flag for determining whether successive iterations will be permitted.

    • Constructor Detail

      • ProcessDirector

        public ProcessDirector​(CompositeEntity container,
                       java.lang.String name)
                throws IllegalActionException,
                       NameDuplicationException
        Construct a director in the given container with the given name. If the container argument must not be null, or a NullPointerException will be thrown. If the name argument is null, then the name is set to the empty string. Increment the version number of the workspace.
        Parameters:
        container - The container
        name - Name of this director.
        Throws:
        IllegalActionException - If the name contains a period, or if the director is not compatible with the specified container.
        NameDuplicationException - If the container not a CompositeActor and the name collides with an entity in the container.

    • Method Detail

      • addThread

        public void addThread​(java.lang.Thread thread)
        Notify this director that the specified thread is part of the execution of this model. This is used to keep track of whether the model is deadlocked, and also to terminate threads if necessary. It is important that the thread call _removeThread() upon exiting. Note further that this should be called before the thread is started to avoid race conditions where some threads have been started and others have not been started and deadlock is falsely detected because the not-yet-started threads are not counted.
        Parameters:
        thread - The thread.
        See Also:
        removeThread(Thread)

      • clone

        public java.lang.Object clone​(Workspace workspace)
                       throws java.lang.CloneNotSupportedException
        Clone the director into the specified workspace. The new object is not added to the directory of that workspace (It must be added by the user if he wants it to be there). The result is a new director with no container, no pending mutations, and no topology listeners. The count of active processes is zero.
        Overrides:
        clone in class Director
        Parameters:
        workspace - The workspace for the cloned object.
        Returns:
        The new ProcessDirector.
        Throws:
        java.lang.CloneNotSupportedException - If one of the attributes cannot be cloned.
        See Also:
        NamedObj.exportMoML(Writer, int, String), NamedObj.setDeferringChangeRequests(boolean)

      • finish

        public void finish()
        Request that the current iteration finishes and postfire() returns false, indicating to the environment that no more iterations should be invoked. To support domains where actor firings do not necessarily terminate, such as PN, you may wish to call stopFire() as well to request that those actors complete their firings.
        Overrides:
        finish in class Director

      • fire

        public void fire()
          throws IllegalActionException
        Wait until a deadlock is detected. Then deal with the deadlock by calling the protected method _resolveDeadlock() and return. This method is synchronized on the director.
        Specified by:
        fire in interface Executable
        Overrides:
        fire in class Director
        Throws:
        IllegalActionException - If a derived class throws it.

      • initialize

        public void initialize​(Actor actor)
                throws IllegalActionException
        Initialize the given actor. This class overrides the base class to reset the flags for all of the receivers, and to create a new ProcessThread for each actor being controlled. This class does *NOT* directly call the initialize method of the actor. That method is instead called by the actor's thread itself. This allows actors in process domains to create tokens during initialization, since sending data in a process-based domain requires threads for each actor.
        Overrides:
        initialize in class Director
        Parameters:
        actor - The actor that is to be initialized.
        Throws:
        IllegalActionException - If the actor is not acceptable to the domain. Not thrown in this base class.

      • isStopFireRequested

        public boolean isStopFireRequested()
        Return true if a stop has been requested on the director. This is used by the ProcessThread to tell the difference between a request to pause and a request to stop.
        Returns:
        True if stop() has been called.

      • isThreadActive

        public boolean isThreadActive​(java.lang.Thread thread)
        Return true if the specified thread has been registered with addThread() and has not been removed with removeThread().
        Parameters:
        thread - The thread.
        Returns:
        True if the specified thread is active.
        See Also:
        addThread(Thread), removeThread(Thread)

      • postfire

        public boolean postfire()
                 throws IllegalActionException
        Return false if a stop has been requested or if the model has reached deadlock. Return true otherwise.
        Specified by:
        postfire in interface Executable
        Overrides:
        postfire in class Director
        Returns:
        False if the director has detected a deadlock or a stop has been requested.
        Throws:
        IllegalActionException - If a derived class throws it.

      • prefire

        public boolean prefire()
                throws IllegalActionException
        Start threads for all actors that have not had threads started already (this might include actors initialized since the last invocation of prefire). This starts the threads, corresponding to all the actors, that were created in a mutation.
        Specified by:
        prefire in interface Executable
        Overrides:
        prefire in class Director
        Returns:
        True.
        Throws:
        IllegalActionException - If a derived class throws it.

      • removeThread

        public void removeThread​(java.lang.Thread thread)
        Notify this director that the specified thread has finished executing. This is used to keep track of whether the model is deadlocked, and also to terminate threads if necessary.
        Parameters:
        thread - The thread.
        See Also:
        addThread(Thread)

      • stop

        public void stop()
        Request that the director cease execution altogether. This causes a call to stop() on all actors contained by the container of this director, and a call to stopThread() on each of the process threads that contain actors controlled by this director. This also sets a flag so that the next call to postfire() returns false.
        Specified by:
        stop in interface Executable
        Overrides:
        stop in class Director

      • stopFire

        public void stopFire()
        Request that execution stop at the conclusion of the current iteration. Call stopThread() on each of the process threads that contain actors controlled by this director and call stopFire() on the actors that are contained by these threads. This method is non-blocking.
        Specified by:
        stopFire in interface Executable
        Overrides:
        stopFire in class Director

      • terminate

        public void terminate()
        Terminate all threads under control of this director immediately. This abrupt termination will not allow normal cleanup actions to be performed, and the model should be recreated after calling this method. This method uses Thread.stop(), a deprecated method in Java.
        Specified by:
        terminate in interface Executable
        Overrides:
        terminate in class Director

      • threadBlocked

        public void threadBlocked​(java.lang.Thread thread,
                          ProcessReceiver receiver)
        Notify the director that the specified thread is blocked on an I/O operation. If the thread has not been registered with addThread(), then this call is ignored.
        Parameters:
        thread - The thread.
        receiver - The receiver handling the I/O operation, or null if it is not a specific receiver.
        See Also:
        addThread(Thread)

      • threadHasPaused

        public void threadHasPaused​(java.lang.Thread thread)
        Notify the director that the specified thread has paused in response to a call to stopFire(). If the thread has not been registered with addThread(), then this call is ignored. If the thread has been identified as blocked, it is removed from the set of blocked threads (so it doesn't get counted twice).
        Parameters:
        thread - The thread.
        See Also:
        addThread(Thread)

      • threadHasResumed

        public void threadHasResumed​(java.lang.Thread thread)
        Notify the director that the specified thread has resumed. If the director has not previously been notified that it was paused, then this call is ignored.
        Parameters:
        thread - The thread.
        See Also:
        threadHasPaused(Thread)

      • threadUnblocked

        public void threadUnblocked​(java.lang.Thread thread,
                            ProcessReceiver receiver)
        Notify the director that the specified thread is unblocked on an I/O operation. If the thread has not been registered with threadBlocked(), then this call is ignored.
        Parameters:
        thread - The thread.
        receiver - The receiver handling the I/O operation, or null if it is not a specific receiver.
        See Also:
        *

      • transferInputs

        public boolean transferInputs​(IOPort port)
                       throws IllegalActionException
        Do nothing. Input transfers in process domains are handled by branches, which transfer inputs in a separate thread.
        Overrides:
        transferInputs in class Director
        Parameters:
        port - The port.
        Returns:
        False, to indicate that no tokens were transferred.
        Throws:
        IllegalActionException - Not thrown in this base class.

      • transferOutputs

        public boolean transferOutputs​(IOPort port)
                        throws IllegalActionException
        Do nothing. Output transfers in process domains are handled by branches, which transfer inputs in a separate thread.
        Overrides:
        transferOutputs in class Director
        Parameters:
        port - The port.
        Returns:
        False, to indicate that no tokens were transferred.
        Throws:
        IllegalActionException - Not thrown in this base class.

      • wrapup

        public void wrapup()
            throws IllegalActionException
        End the execution of the model under the control of this director. A flag is set in all the receivers that causes each process to terminate at the earliest communication point. Prior to setting receiver flags, this method wakes up the threads if they all are stopped. If the container is not an instance of CompositeActor, then this method does nothing.

        This method is not synchronized on the workspace, so the caller should be.

        Specified by:
        wrapup in interface Initializable
        Overrides:
        wrapup in class Director
        Throws:
        IllegalActionException - If an error occurs while accessing the receivers of all actors under the control of this director.

      • _areAllThreadsStopped

        protected boolean _areAllThreadsStopped()
        Return true if the count of active processes equals the number of paused and blocked threads. Otherwise return false.
        Returns:
        True if there are no active processes in the container.

      • _areThreadsDeadlocked

        protected boolean _areThreadsDeadlocked()
        Return true if the count of active processes in the container is 0. Otherwise return false. Derived classes must override this method to return true to any other forms of deadlocks that they might introduce.
        Returns:
        True if there are no active processes in the container.

      • _getActiveThreadsCount

        protected final int _getActiveThreadsCount()
        Return the number of active threads under the control of this director.
        Returns:
        The number of active threads.

      • _getBlockedThreadsCount

        protected final int _getBlockedThreadsCount()
        Return the number of threads that are currently blocked.
        Returns:
        Return the number of threads that are currently blocked.

      • _getStoppedThreadsCount

        protected final int _getStoppedThreadsCount()
        Return the number of threads that are currently stopped.
        Returns:
        Return the number of threads that are currently stopped.

      • _newProcessThread

        protected ProcessThread _newProcessThread​(Actor actor,
                                          ProcessDirector director)
                                   throws IllegalActionException
        Create a new ProcessThread for controlling the actor that is passed as a parameter of this method. Subclasses are encouraged to override this method as necessary for domain specific functionality.
        Parameters:
        actor - The actor that the created ProcessThread will control.
        director - The director that manages the model that the created thread is associated with.
        Returns:
        Return a new ProcessThread that will control the actor passed as a parameter for this method.
        Throws:
        IllegalActionException - If creating an new ProcessThread throws it.

      • _resolveDeadlock

        protected boolean _resolveDeadlock()
                            throws IllegalActionException
        Return false indicating that deadlock has not been resolved and that execution will be discontinued. In derived classes, override this method to obtain domain specific handling of deadlocks. Return false if a real deadlock has occurred and the simulation can be ended. Return true if the simulation can proceed given additional data and need not be terminated.
        Returns:
        False.
        Throws:
        IllegalActionException - Not thrown in this base class.

      • _requestFinishOnReceivers

        protected void _requestFinishOnReceivers()
        Call requestFinish() on all receivers.