Package fr.dyade.aaa.agent

Class Engine

  • All Implemented Interfaces:
    AgentEngine, EngineMBean, MessageComparator, MessageConsumer, Runnable
    class Engine
extends Object
implements Runnable, AgentEngine, EngineMBean
    The Engine class provides multiprogramming of agents. It realizes the program loop which successively gets the notifications from the message queue and calls the relevant reaction function member of the target agent. The engine's basic behaviour is: 
     While (true) {
   // get next message in channel
   Message msg = qin.get();
   // get the agent to process event
   Agent agent = load(msg.to);
   // execute relevant reaction, all notification sent during this
   // reaction is inserted into persistent queue in order to processed
   // by the channel.
   agent.react(msg.from, msg.not);
   // save changes, then commit.
   <BEGIN TRANSACTION>
   qin.pop();
   channel.dispatch();
   agent.save();
   <COMMIT TRANSACTION>
 }

    The Engine class ensures the atomic handling of an agent reacting to a notification:

    • if the reaction completes, a COMMIT ensures all changes related to the reaction are committed (state change of the agent, notifications signaled during the reaction, deletion of the handled notification);
    • if anything goes wrong during the reaction, a ROLLBACK undoes the changes; depending on the error kind it may be necessary to execute additional operations to resynchronize the database and the memory objects, and to allow the main program to continue.
    Handling errors.

    Two types of errors may occur: errors of first type are detected in the source code and signaled by an Exception; serious errors lead to an Error being raised then the engine exits. In the first case the exception may be handled at any level, even partially. Most of them are signaled up to the engine loop. Two cases are then distinguished depending on the recovery policy:

    • if recoveryPolicy is set to RP_EXC_NOT (default value) then the agent state and the message queue are restored (ROLLBACK); an ExceptionNotification notification is sent to the sender and the engine may then proceed with next notification;
    • if recoveryPolicy is set to RP_EXIT the engine stops the agent server.

    • Field Summary

      Fields 
      Modifier and Type Field Description
      (package private) Agent agent
      The current agent running.
      boolean agentProfiling
      Boolean value indicating if the agent profiling is on, by default false.
      (package private) Hashtable<AgentId,​Agent> agents
      This table is used to maintain a list of agents already in memory using the AgentId as primary key.
      (package private) AverageCPUTask averageCPUTask
      The average use of CPU by the Engine thread during last minute.
      (package private) Engine.EngineAverageLoadTask averageLoadTask  
      protected boolean canStop
      Boolean variable used to stop the engine properly.
      private long commitTime
      Time consumed during reaction commit.
      private static boolean DEBUG  
      (package private) Vector<AgentId> fixedAgentIdList
      Vector containing id's of all fixed agents.
      protected boolean isRunning
      Boolean variable used to stop the engine properly.
      private Logger logmon  
      private boolean modified
      True if the timestamp is modified since last save.
      protected Queue mq  
      (package private) Message msg
      The message in progress.
      private String name  
      (package private) int NbMaxAgents
      Maximum number of memory loaded agents.
      protected boolean needToBeCommited  
      (package private) boolean noTxIfTransient
      Flag to avoid transaction when not needed.
      (package private) long now
      Virtual time counter use in FIFO swap-in/swap-out mechanisms.
      private boolean persistentPush  
      protected MessageQueue qin
      Queue of messages to be delivered to local agents.
      private long reactTime
      Time consumed during agent's reaction.
      (package private) int recoveryPolicy
      recovery policy in case of exception in agent specific code.
      (package private) static int RP_EXC_NOT
      Send ExceptionNotification notification in case of exception in agent specific code.
      (package private) static int RP_EXIT
      Stop agent server in case of exception in agent specific code.
      (package private) static String[] rpStrings
      String representations of RP_* constant values for the recoveryPolicy variable.
      private int stamp
      Logical timestamp information for messages in "local" domain.
      private byte[] stampBuf
      Buffer used to optimize
      (package private) EngineThread thread
      The active component of this engine.
      protected long timeout  
      (package private) long workInProgress
      the tick counter that reflects activity in engine.

    • Constructor Summary

      Constructors 
      Modifier Constructor Description
      protected Engine()
      Initializes a new Engine object (can only be used by subclasses).

    • Method Summary

      All Methods Static Methods Instance Methods Concrete Methods 
      Modifier and Type Method Description
      (package private) void abort​(Exception exc)
      Abort the agent reaction in case of error during execution.
      (package private) void addFixedAgentId​(AgentId id)
      Adds an AgentId in the fixedAgentIdList Vector.
      (package private) void clean()
      Cleans the Channel queue of all pushed notifications.
      (package private) void commit()
      Commit the agent reaction in case of right termination: suppress the processed notification from message queue, then deletes it ; push all new notifications in qin and qout, and saves them ; saves the agent state ; then commit the transaction to validate all changes.
      (package private) void createAgent​(Agent agent)
      Creates and initializes an agent.
      void createAgent​(AgentId id, Agent agent)
      Creates and initializes an agent.
      void delete()
      This operation always throws an IllegalStateException.
      void deleteAgent​(AgentId id)
      Deletes an agent.
      (package private) void dispatch()
      Dispatch messages between the MessageConsumer: Engine component and Network components.
      String dumpAgent​(AgentId id)
      Returns a string representation of the specified agent.
      String dumpAgent​(String id)
      Returns a string representation of the specified agent.
      (package private) void garbage()
      The garbage method should be called regularly , to swap out from memory all the agents which have not been accessed for a time.
      int getAverageCPU()
      Returns the average use of CPU by the Engine thread during last minute.
      float getAverageLoad1()
      Returns the load averages for the last minute.
      float getAverageLoad15()
      Returns the load averages for the past 15 minutes.
      float getAverageLoad5()
      Returns the load averages for the past 5 minutes.
      long getCommitTime()
      Returns the total commit time calculated for this engine.
      String getDomainName()
      Returns the corresponding domain's name.
      (package private) AgentId[] getLoadedAgentIdlist()
      Method used for debug and monitoring.
      String getName()
      Returns this Engine's name.
      int getNbAgents()
      Returns the number of agents actually loaded in memory.
      int getNbFixedAgents()
      Returns the number of fixed agents.
      int getNbMaxAgents()
      Returns the maximum number of agents loaded in memory.
      int getNbMessages()
      Gets the number of messages posted to this engine since creation.
      long getNbReactions()
      Returns the number of agent's reaction since last boot.
      int getNbWaitingMessages()
      Gets the number of waiting messages in this engine.
      MessageQueue getQueue()
      Get this engine's MessageQueue qin.
      long getReactTime()
      Returns the total reaction time calculated for this engine.
      String getRunningAgent()
      Returns the unique id. of the running agent if any, null otherwise.
      private int getStamp()
      Gets the current value of stamp counter.
      long getWorkInProgress()
      Returns the tick counter that reflects activity in engine.
      void incWorkInProgress()
      Increments the tick counter that reflects activity in engine.
      void init​(AgentEngineContext agentEngineContext)
      Initializes the engine.
      void insert​(Message msg)
      Insert a message in the MessageQueue.
      boolean isAgentProfiling()
      Returns true if the agent profiling is on.
      boolean isEngineThread()
      Checks if the current thread calling this method belongs to the engine.
      boolean isNoTxIfTransient()
      Returns the flag to avoid transactions.
      boolean isPrior​(Message m1, Message m2)
      Compares the 2 messages for time ordering.
      boolean isRunning()
      Tests if the engine is alive.
      (package private) Agent load​(AgentId id)
      The load method return the Agent object designed by the AgentId parameter.
      (package private) static Engine newInstance()
      Creates a new instance of Engine (real class depends of server type).
      protected void onTimeOut()  
      void post​(Message msg)
      Adds a message in "ready to deliver" list.
      void postAndValidate​(Message msg)
      Posts a message and validates it at the same time.
      void push​(AgentId from, AgentId to, Notification not)
      Push a new message in temporary queue until the end of current reaction.
      void push​(AgentId to, Notification not)
      Push a new message in temporary queue until the end of current reaction.
      (package private) Agent reload​(AgentId id)
      The reload method return the Agent object loaded from the storage.
      (package private) void removeFixedAgentId​(AgentId id)
      Removes an AgentId in the fixedAgentIdList Vector.
      String report()
      Returns a report about the distribution of messages type in queue.
      void resetAverageLoad()  
      void resetCommitTime()
      reset the commitTime
      void resetReactTime()
      reset the reactTime
      void resetTimer()
      Reset reactTime and commitTime
      void restore()
      Restores logical clock information from persistent storage.
      void run()
      Main loop of agent server Engine.
      void save()
      Saves logical clock information to persistent storage.
      void setAgentProfiling​(boolean agentProfiling)
      Sets the agent profiling.
      void setNbMaxAgents​(int NbMaxAgents)
      Sets the maximum number of agents that can be loaded simultaneously in memory.
      protected void stamp​(Message msg)
      Sets the stamp of the given message.
      void start()
      Causes this engine to begin execution.
      void stop()
      Forces the engine to stop executing.
      (package private) void terminate()  
      String toString()
      Returns a string representation of this engine.
      void validate()
      Validates all messages pushed in queue during transaction session.

    • Field Detail

      • DEBUG

        private static final boolean DEBUG

      • logmon

        private final Logger logmon

      • qin

        protected MessageQueue qin
        Queue of messages to be delivered to local agents.

      • isRunning

        protected volatile boolean isRunning
        Boolean variable used to stop the engine properly. The engine tests this variable between each reaction, and stops if it is false.

      • canStop

        protected volatile boolean canStop
        Boolean variable used to stop the engine properly. If this variable is true then the engine is waiting and it can interrupted, else it handles a notification and it will exit after (the engine tests the isRunning variable between each reaction)

      • stamp

        private int stamp
        Logical timestamp information for messages in "local" domain.

      • stampBuf

        private byte[] stampBuf
        Buffer used to optimize

      • modified

        private boolean modified
        True if the timestamp is modified since last save.

      • agents

        Hashtable<AgentId,​Agent> agents
        This table is used to maintain a list of agents already in memory using the AgentId as primary key.

      • now

        long now
        Virtual time counter use in FIFO swap-in/swap-out mechanisms.

      • NbMaxAgents

        int NbMaxAgents
        Maximum number of memory loaded agents.

      • noTxIfTransient

        boolean noTxIfTransient
        Flag to avoid transaction when not needed. By default true, we run a transaction only if there is transaction operations: - incoming persistent notification, - persistent objects modifications, - outgoing persistent notifications. The value could be set using "NoTxIfTransient" property.

      • fixedAgentIdList

        Vector<AgentId> fixedAgentIdList
        Vector containing id's of all fixed agents.

      • agent

        Agent agent
        The current agent running.

      • msg

        Message msg
        The message in progress.

      • thread

        EngineThread thread
        The active component of this engine.

      • RP_EXC_NOT

        static final int RP_EXC_NOT
        Send ExceptionNotification notification in case of exception in agent specific code. Constant value for the recoveryPolicy variable.
        See Also:
        Constant Field Values

      • RP_EXIT

        static final int RP_EXIT
        Stop agent server in case of exception in agent specific code. Constant value for the recoveryPolicy variable.
        See Also:
        Constant Field Values

      • rpStrings

        static final String[] rpStrings
        String representations of RP_* constant values for the recoveryPolicy variable.

      • recoveryPolicy

        int recoveryPolicy
        recovery policy in case of exception in agent specific code. Default value is RP_EXC_NOT.

      • persistentPush

        private boolean persistentPush

      • needToBeCommited

        protected boolean needToBeCommited

      • timeout

        protected long timeout

      • agentProfiling

        public boolean agentProfiling
        Boolean value indicating if the agent profiling is on, by default false. If true, the cumulative time of reaction and commit is kept for each agent. In addition the total reaction and commit time is calculated for this engine. This value can be adjusted through the AgentProfiling system property.

      • reactTime

        private long reactTime
        Time consumed during agent's reaction.

      • commitTime

        private long commitTime
        Time consumed during reaction commit.

      • workInProgress

        long workInProgress
        the tick counter that reflects activity in engine.

      • averageCPUTask

        AverageCPUTask averageCPUTask
        The average use of CPU by the Engine thread during last minute.

    • Constructor Detail

      • Engine

        protected Engine()
          throws Exception
        Initializes a new Engine object (can only be used by subclasses).
        Throws:
        Exception - an error occurs.

    • Method Detail

      • getNbReactions

        public long getNbReactions()
        Returns the number of agent's reaction since last boot.
        Specified by:
        getNbReactions in interface AgentEngine
        Specified by:
        getNbReactions in interface EngineMBean
        Returns:
        the number of agent's reaction since last boot

      • getNbMaxAgents

        public int getNbMaxAgents()
        Returns the maximum number of agents loaded in memory.
        Specified by:
        getNbMaxAgents in interface EngineMBean
        Returns:
        the maximum number of agents loaded in memory

      • setNbMaxAgents

        public void setNbMaxAgents​(int NbMaxAgents)
        Sets the maximum number of agents that can be loaded simultaneously in memory.
        Specified by:
        setNbMaxAgents in interface EngineMBean
        Parameters:
        NbMaxAgents - the maximum number of agents

      • getNbAgents

        public int getNbAgents()
        Returns the number of agents actually loaded in memory.
        Specified by:
        getNbAgents in interface EngineMBean
        Returns:
        the maximum number of agents actually loaded in memory

      • getNbMessages

        public int getNbMessages()
        Gets the number of messages posted to this engine since creation. return the number of messages.
        Specified by:
        getNbMessages in interface EngineMBean
        Returns:
        the number of messages.

      • getNbWaitingMessages

        public int getNbWaitingMessages()
        Gets the number of waiting messages in this engine. return the number of waiting messages.
        Specified by:
        getNbWaitingMessages in interface AgentEngine
        Specified by:
        getNbWaitingMessages in interface EngineMBean
        Returns:
        the number of waiting messages.

      • getNbFixedAgents

        public int getNbFixedAgents()
        Returns the number of fixed agents.
        Specified by:
        getNbFixedAgents in interface EngineMBean
        Returns:
        the number of fixed agents

      • getRunningAgent

        public String getRunningAgent()
        Returns the unique id. of the running agent if any, null otherwise.
        Specified by:
        getRunningAgent in interface AgentEngine
        Specified by:
        getRunningAgent in interface EngineMBean
        Returns:
        the unique id. of the running agent if any, null otherwise.

      • getDomainName

        public final String getDomainName()
        Returns the corresponding domain's name.
        Specified by:
        getDomainName in interface MessageConsumer
        Returns:
        this domain's name.

      • newInstance

        static Engine newInstance()
                   throws Exception
        Creates a new instance of Engine (real class depends of server type).
        Returns:
        the corresponding engine's instance.
        Throws:
        Exception - an error occurs.

      • push

        public final void push​(AgentId from,
                       AgentId to,
                       Notification not)
        Push a new message in temporary queue until the end of current reaction. As this method is only called by engine's thread it does not need to be synchronized.
        Specified by:
        push in interface AgentEngine
        Parameters:
        from - the source
        to - the destination
        not - the notification to push

      • push

        public final void push​(AgentId to,
                       Notification not)
        Push a new message in temporary queue until the end of current reaction. As this method is only called by engine's thread it does not need to be synchronized.
        Specified by:
        push in interface AgentEngine
        Parameters:
        to - the destination
        not - the notification to push

      • dispatch

        final void dispatch()
             throws Exception
        Dispatch messages between the MessageConsumer: Engine component and Network components.

        Handle persistent information in respect with engine transaction.

        Be careful, this method must only be used during a transaction in order to ensure the mutual exclusion.
        Throws:
        IOException - error when accessing the local persistent storage.
        Exception

      • clean

        final void clean()
        Cleans the Channel queue of all pushed notifications.
        Be careful, this method must only be used during a transaction in order to ensure the mutual exclusion.

      • init

        public void init​(AgentEngineContext agentEngineContext)
          throws Exception
        Description copied from interface: AgentEngine
        Initializes the engine. The AgentEngineContext parameter should be for the unique private use of this AgentEngine and should not be shared with any other components otherwise the agent server security would be broken.
        Specified by:
        init in interface AgentEngine
        Parameters:
        agentEngineContext - context enabling this AgentEngine to invoke operations that cannot be accessed outside of the fr.dyade.aaa.agent package. The AgentEngineContext should be for the unique private use of this AgentEngine and should not be shared with any other components otherwise the agent server security would be broken.
        Throws:
        Exception - an error occurs.

      • terminate

        void terminate()

      • createAgent

        public final void createAgent​(AgentId id,
                              Agent agent)
                       throws Exception
        Creates and initializes an agent.
        Specified by:
        createAgent in interface AgentEngine
        Parameters:
        agent - agent object to create
        id - the unique identifier
        Throws:
        Exception - unspecialized exception

      • createAgent

        final void createAgent​(Agent agent)
                throws Exception
        Creates and initializes an agent.
        Parameters:
        agent - agent object to create
        Throws:
        Exception - unspecialized exception

      • deleteAgent

        public void deleteAgent​(AgentId id)
                 throws Exception
        Deletes an agent.
        Specified by:
        deleteAgent in interface AgentEngine
        Parameters:
        id - unique identifier of agent to delete
        Throws:
        Exception - unspecialized exception

      • garbage

        void garbage()
        The garbage method should be called regularly , to swap out from memory all the agents which have not been accessed for a time.

      • removeFixedAgentId

        void removeFixedAgentId​(AgentId id)
                 throws IOException
        Removes an AgentId in the fixedAgentIdList Vector.
        Parameters:
        id - the AgentId of no more used fixed agent.
        Throws:
        IOException - an error occurs.

      • addFixedAgentId

        void addFixedAgentId​(AgentId id)
              throws IOException
        Adds an AgentId in the fixedAgentIdList Vector.
        Parameters:
        id - the AgentId of new fixed agent.
        Throws:
        IOException - an error occurs.

      • getLoadedAgentIdlist

        AgentId[] getLoadedAgentIdlist()
        Method used for debug and monitoring. It returns an enumeration of all agents loaded in memory.
        Returns:
        an array containing the unique identifier of all agents in memory.

      • dumpAgent

        public String dumpAgent​(String id)
                 throws Exception
        Returns a string representation of the specified agent.
        Specified by:
        dumpAgent in interface EngineMBean
        Parameters:
        id - The string representation of the agent's unique identification.
        Returns:
        A string representation of the specified agent.
        Throws:
        Exception - an error occurs.
        See Also:
        dumpAgent(AgentId)

      • dumpAgent

        public String dumpAgent​(AgentId id)
                 throws IOException,
                        ClassNotFoundException
        Returns a string representation of the specified agent. If the agent is not present it is loaded in memory, be careful it is not initialized (agentInitialize) nor cached in agents vector.
        Specified by:
        dumpAgent in interface AgentEngine
        Parameters:
        id - The agent's unique identification.
        Returns:
        A string representation of specified agent.
        Throws:
        IOException - an error occurs.
        ClassNotFoundException - an error occurs.

      • load

        final Agent load​(AgentId id)
          throws IOException,
                 ClassNotFoundException,
                 Exception
        The load method return the Agent object designed by the AgentId parameter. If the Agent object is not already present in the server memory, it is loaded from the storage. Be careful, if the save method can be overloaded to optimize the save process, the load procedure used by engine is always load.
        Parameters:
        id - The agent identification.
        Returns:
        The corresponding agent.
        Throws:
        IOException - If an I/O error occurs.
        ClassNotFoundException - Should never happen (the agent has already been loaded in deploy).
        UnknownAgentException - There is no corresponding agent on secondary storage.
        Exception - when executing class specific initialization

      • insert

        public void insert​(Message msg)
        Insert a message in the MessageQueue. This method is used during initialization to restore the component state from persistent storage.
        Specified by:
        insert in interface MessageConsumer
        Parameters:
        msg - the message

      • validate

        public void validate()
        Validates all messages pushed in queue during transaction session.
        Specified by:
        validate in interface MessageConsumer

      • isRunning

        public boolean isRunning()
        Tests if the engine is alive.
        Specified by:
        isRunning in interface EngineMBean
        Specified by:
        isRunning in interface MessageConsumer
        Returns:
        true if this MessageConsumer is alive; false otherwise.

      • restore

        public void restore()
             throws Exception
        Restores logical clock information from persistent storage.
        Specified by:
        restore in interface MessageConsumer
        Throws:
        Exception - an error occurs.

      • getStamp

        private final int getStamp()
        Gets the current value of stamp counter.
        Returns:
        the current value of stamp counter.

      • stamp

        protected final void stamp​(Message msg)
        Sets the stamp of the given message.
        Parameters:
        msg - The message to stamp.

      • isPrior

        public boolean isPrior​(Message m1,
                       Message m2)
        Description copied from interface: MessageComparator
        Compares the 2 messages for time ordering. Returns true if the first message is earlier than the second, false otherwise.
        Specified by:
        isPrior in interface MessageComparator
        Parameters:
        m1 - The first message.
        m2 - The second message.
        Returns:
        true if the first message is earlier than the second, false otherwise.

      • post

        public void post​(Message msg)
          throws Exception
        Adds a message in "ready to deliver" list. This method allocates a new time stamp to the message ; be Careful, changing the stamp imply the filename change too.
        Specified by:
        post in interface MessageConsumer
        Parameters:
        msg - the message to deliver.
        Throws:
        Exception - an error occurs.

      • resetTimer

        public void resetTimer()
        Reset reactTime and commitTime
        Specified by:
        resetTimer in interface EngineMBean

      • run

        public void run()
        Main loop of agent server Engine.
        Specified by:
        run in interface Runnable

      • commit

        void commit()
     throws Exception
        Commit the agent reaction in case of right termination:
        • suppress the processed notification from message queue, then deletes it ;
        • push all new notifications in qin and qout, and saves them ;
        • saves the agent state ;
        • then commit the transaction to validate all changes.
        Throws:
        Exception - an error occurs.

      • abort

        void abort​(Exception exc)
    throws Exception
        Abort the agent reaction in case of error during execution. In case of unrecoverable error during the reaction we have to rollback:
        • reload the previous state of agent ;
        • remove the failed notification ;
        • clean the Channel queue of all pushed notifications ;
        • send an error notification to the sender ;
        • then commit the transaction to validate all changes.
        Parameters:
        exc - the cause.
        Throws:
        Exception - an error occurs.

      • getWorkInProgress

        public final long getWorkInProgress()
        Returns the tick counter that reflects activity in engine.
        Specified by:
        getWorkInProgress in interface EngineMBean
        Returns:
        the tick counter that reflects activity in engine.

      • getAverageCPU

        public final int getAverageCPU()
        Returns the average use of CPU by the Engine thread during last minute.
        Specified by:
        getAverageCPU in interface EngineMBean
        Returns:
        the average use of CPU by the Engine thread during last minute.

      • report

        public String report()
        Returns a report about the distribution of messages type in queue.
        Specified by:
        report in interface EngineMBean
        Returns:
        a report about the distribution of messages type in queue.

      • toString

        public String toString()
        Returns a string representation of this engine.
        Specified by:
        toString in interface EngineMBean
        Overrides:
        toString in class Object
        Returns:
        A string representation of this engine.

      • isEngineThread

        public boolean isEngineThread()
        Checks if the current thread calling this method belongs to the engine.
        Specified by:
        isEngineThread in interface AgentEngine
        Returns:
        true if the current thread calling this method belongs to the engine