policy.cpp

一个非常好的人工智能开发工具开源软件

/***
 *** See the file "L2_RTI_EO1/disclaimers-and-notices-L2.txt" for 
 *** information on usage and redistribution of this file, 
 *** and for a DISCLAIMER OF ALL WARRANTIES.
 ***/
#include <string.h>                       // for memset
#include "Policy.hpp"
#include "ExceptionAction.hpp"
#include "TimerTask.hpp"
#include "taskLib.h"

#define UNKNOWN 0
#define PEND    1
#define SUSPEND 2
#define READY   3
#define DELAY   4
#define DEAD    5

#define NORMAL_TRANSIENT                  0
#define ABNORMAL_TRANSIENT                1
#define NO_TRANSIENT                      2


Policy::Policy( void ) : outstandingTimers(0), spontaneousObservations(0) {

  beat.numberOfHeartbeats = 0;
  beat.numberOfCommandsReceived = 0;
  beat.numberOfObservationsReceived = 0;
  beat.numberOfTimersExpired = 0;
  beat.numberOfErrors = 0;
  beat.modelLoadedIndicator = 0;
  beat.faultsDiagnosedIndicator = 0;
  beat.rtiTaskStatus = 0;
  beat.l2TaskStatus = 0;
  beat.timerTaskStatus = 0;
  beat.loggerTaskStatus = 0;

  requestHeartbeat();
}


Policy::~Policy( void ) { }


void Policy::commandPolicy(const MONITOR_DATA& monitor) {

  // Send the command
  const RTIMessage cmd(COMMAND, monitor.time,
                               monitor.variable, monitor.value);
  sendCommand(cmd);

  // Request a timer
  requestTimer(monitor.delay,
	       monitor.timeoutVariable,
	       monitor.timeoutValue);

  outstandingTimers++;
  spontaneousObservations=0;

  char str[MAX_MSG_LENGTH];
  sprintf(str,
	  "Policy:: Started timer with timeout %d, command %d, value %d \n",
	  monitor.delay, monitor.timeoutVariable, monitor.timeoutValue);
  Logging::log (Logging::L2_DEBUG, str);
}


void Policy::observationPolicy(const MONITOR_DATA& monitor ) {

  // Add the observation
  const RTIMessage obs(OBSERVATION, monitor.time,
                               monitor.variable, monitor.value);

    // Send the observation to L2
    sendToDispatcher(obs);

    // Report the observation
    const RTIMessage report(REPORT_OBSERVATION,
                                    obs.getTimeStamp(),
                                    obs.getVariableID(),
                                    obs.getValueIndex());
    sendToDispatcher(report);

  // Extension to handle spontaneous observations in steady state. May want
  // to specify msgType as OBSERVATION_TIMEOUT instead of TIMER_EXPIRATION.
  // Currently this information is conveyed in the timeoutValue, which is not
  // used as the observation timeout does not result in a transition from a
  // transient mode (whereas commands do).

  if (outstandingTimers == 0)
      spontaneousObservations++;

  if ((outstandingTimers == 0) && (spontaneousObservations >= MAX_STEADY_STATE_OBSERVATIONS)) {
      // Request a timer
      requestTimer(monitor.delay, 0, OBSERVATION_TIMEOUT);
      outstandingTimers++;
  }
}


void Policy::timeoutPolicy(const MONITOR_DATA& monitor) {

  outstandingTimers--;

  if (monitor.timeoutValue != OBSERVATION_TIMEOUT) {
  // Compose command to end transient
     const RTIMessage endTransient(COMMAND,
				monitor.time,
				monitor.timeoutVariable,
				monitor.timeoutValue);
     sendCommand(endTransient);
  }
  requestFindCandidates(monitor.time);
  reportDiagnosis(monitor.time);
}



void Policy::heartbeatPolicy(const MONITOR_DATA& monitor) {

  // HeartbeatData contains L2 counters, indicators, and task status.

  L2ResultMsg d_l2ResultMsg;

  // Initialize all fields to zero/null.
  memset(&d_l2ResultMsg, 0, sizeof(L2ResultMsg));

  d_l2ResultMsg.msgType = HEARTBEAT_MSG;
  d_l2ResultMsg.time.seconds = monitor.time.seconds;
  d_l2ResultMsg.time.nanosec = monitor.time.nanosec;

  beat.numberOfErrors = ExceptionAction::getErrorCount();
  beat.modelLoadedIndicator = ExceptionAction::getModelLoadedIndicator();
  beat.faultsDiagnosedIndicator = ExceptionAction::getFaultsDiagnosedIndicator();

  beat.rtiTaskStatus    = taskStatus( RTI_TASK_NAME );
  beat.l2TaskStatus     = taskStatus( LIVINGSTONE_TASK_NAME );
  beat.timerTaskStatus  = taskStatus( TIMER_TASK_NAME );
  beat.loggerTaskStatus = taskStatus( LOGGING_TASK_NAME );

  d_l2ResultMsg.heartbeatData = beat;

  Logging::sendToTelemetry(d_l2ResultMsg);

  // Request a timer for the next heartbeat
  requestHeartbeat();

}


unsigned short Policy::taskStatus( const char taskName [L2_TASK_NAME_MAX_LENGTH] ) {
  char statusString[10];
  int taskId = taskNameToId(const_cast<char*>(taskName));

  if (taskId == ERROR) {
    char message[MAX_MSG_LENGTH];
    sprintf(message, "Policy:: %s task died.\n", taskName);
    Logging::log(Logging::L2_ERROR, message);
    return DEAD;
  }

  taskStatusString(taskId, statusString);
  switch (statusString[0]) {
    case 'R':
    	return READY;
    case 'P':
    	return PEND;
    case 'D':
    	return DELAY;
    case 'S':
    	return SUSPEND;
  }

  return UNKNOWN;
}


void Policy::sendCommand(const RTIMessage& msg) {
  // Send the command
  if (send(msg) != 0) {
    ExceptionAction::sendFailure(ExceptionAction::SND_POLICY,
				 ExceptionAction::RCV_LIVINGSTONE_DISPATCHER,
				 errno);
  }

  const RTIMessage reportCommandMessage(REPORT_COMMAND,
					msg.getTimeStamp(),
					msg.getVariableID(),
					msg.getValueIndex());
  if (send(reportCommandMessage) == 0) {
    char message[MAX_MSG_LENGTH];
    sprintf(message, "Policy:: sent Command to Dispatcher:%d %d\n",
	    msg.getVariableID(),
	    msg.getValueIndex());
    Logging::log(Logging::L2_DEBUG, message);
  } else {
    ExceptionAction::sendFailure(ExceptionAction::SND_POLICY,
				 ExceptionAction::RCV_LIVINGSTONE_DISPATCHER,
				 errno);
  }
}


void Policy::sendObservation(const RTIMessage& obs) {
  if (send(obs) != 0) {
    ExceptionAction::sendFailure(ExceptionAction::SND_POLICY,
				 ExceptionAction::RCV_LIVINGSTONE_DISPATCHER,
				 errno);
  }

  const RTIMessage reportObservationMessage(REPORT_OBSERVATION,
						    obs.getTimeStamp(),
						    obs.getVariableID(),
						    obs.getValueIndex());
  if (send(reportObservationMessage) == 0) {
    // What is to be done?
  } else {
    ExceptionAction::sendFailure(ExceptionAction::SND_POLICY,
				 ExceptionAction::RCV_LIVINGSTONE_DISPATCHER,
				 errno);
  }
}



// REDUNDANT

void Policy::sendToDispatcher(const RTIMessage&
                                          livingstoneMessage) {
  if (msgQSend(RTI_TO_L2_MQID,
               (char*)&(livingstoneMessage),
               sizeof(RTIMessage),
               WAIT_FOREVER,
               MSG_PRI_NORMAL) != 0) {
    ExceptionAction::sendFailure(ExceptionAction::SND_OBSERVATIONS_BUFFER,
                                 ExceptionAction::RCV_LIVINGSTONE_DISPATCHER,
                                 errno);
  }
}



void Policy::requestFindCandidates(const TIME_TAG& time ) {
  const RTIMessage msg(FIND_CANDIDATES, time);
  if (send(msg) == 0) {
    Logging::log(Logging::L2_DEBUG,
		 "Policy:: sent FIND_CANDIDATES to Dispatcher\n");
  } else {
    ExceptionAction::sendFailure(ExceptionAction::SND_POLICY,
				 ExceptionAction::RCV_LIVINGSTONE_DISPATCHER,
				 errno);
  }
  reportFindCandidates(time);
}


void Policy::reportFindCandidates(const TIME_TAG& time ) {
  const RTIMessage msg(REPORT_FIND_CANDIDATES, time);
  if (send(msg) == 0) {
    Logging::log(Logging::L2_DEBUG,
		 "Policy:: sent REPORT_FIND_CANDIDATES to Dispatcher\n");
  } else {
    ExceptionAction::sendFailure(ExceptionAction::SND_POLICY,
				 ExceptionAction::RCV_LIVINGSTONE_DISPATCHER,
				 errno);
  }
}


void Policy::reportDiagnosis(const TIME_TAG& time ) {
  const RTIMessage msg(REPORT_DIAGNOSIS, time);
  if (send(msg) == 0) {
    Logging::log(Logging::L2_DEBUG,
		 "Policy:: sent REPORT_DIAGNOSIS to Dispatcher\n");
  } else {
    ExceptionAction::sendFailure(ExceptionAction::SND_POLICY,
				 ExceptionAction::RCV_LIVINGSTONE_DISPATCHER,
				 errno);
  }
}


void Policy::reportFullState( void ) {
  TIME_TAG dummyTime;
  RTIMessage msg(REPORT_FULL_STATE, dummyTime);
  int msgStatus = send(msg);

  if (msgStatus == 0) {
    // What is to be done?
  } else {
    ExceptionAction::sendFailure(ExceptionAction::SND_POLICY,
				 ExceptionAction::RCV_LIVINGSTONE_DISPATCHER,
				 errno);
  }
}



int Policy::send(const RTIMessage& message) {
  return msgQSend(RTI_TO_L2_MQID,
		  (char*)&message,
		  sizeof(RTIMessage),
		  WAIT_FOREVER,
		  MSG_PRI_NORMAL);
}