brti.c

基于linux环境的ns2多机并行仿真补丁

/*---------------------------------------------------------------------------*/
/* BRTI implementation.                                                      */
/* Author(s): Richard Fujimoto, Steve Ferenci, Thom McLean, Kalyan Perumalla.*/
/* $Revision: 1.4 $ $Name: v26apr05 $ $Date: 2003/12/29 11:33:13 $ */
/*---------------------------------------------------------------------------*/
#include "brti.h"

struct BRTIstate
{
   HEAP_PQ TSOQueue;
}BRTI;

void MyHereProc(long, struct MsgS*); /*SLFXXX*/
void BRTI_InitTSOQueue(int size, int incr);
TM_Time BRTI_TSOMin(void);
char *BRTI_TSOPop(TM_Time *minTS,long *MsgSize, long *MsgType);

/*---------------------------------------------------------------------------*/
static char RTI; /*Dummy context determinant*/
static int MessagesAnihilated;
/*---------------------------------------------------------------------------*/
/**************************************************************************
 * Message Handler
 * Called when a user message comes in off the wire.
 * The incoming message is placed into the TSO queue
 * Note: if there are multiple subscribers in this PE, a separate TSO queue
 * entry is created for each subscriber, but only one physical copy of the
 * message is created.
 */
void MyHandler (MCAST_Addr Msg, long MsgSize, MCAST_Addr Context, long MsgType)
{
  struct MsgS *m;
  fdkErrorCode retCode;

  /* consistency checks */
  if (MsgType != 0 && MsgType != MSG_RETRACTION)
  {printf ("Handler: unexpected msg type\n"); fflush(stdout); exit (1);}

  /* insert into TSO queue */
  m = (struct MsgS *) Msg;
 
  if (MsgType == MSG_RETRACTION)
  {
    if ((retCode = RHM_In(m->erh,m->TimeStamp,Msg,MsgSize,MsgType)) !=
	fdkSUCCEEDED)
    {
       printf("MyHandler Error, could not store Retraction Handle :\n");
       printf("         %s \n",fdkErrorMessage(retCode));
       exit(1);
    }
  }
  else
  {
    HEAP_InsertWithType(BRTI.TSOQueue, m->TimeStamp, m, MsgSize, MsgType);
  }
}

/*---------------------------------------------------------------------------*/
void RTI_SetLookAhead(TM_Time NewLA)
{
    Core_SetLookAhead(NewLA); 
}

/*---------------------------------------------------------------------------*/
TM_Time RTI_GetLookAhead( void )
{
    return(Core_GetLookAhead());
}

/*---------------------------------------------------------------------------*/
/**************************************************************************
 *  TM Handler: Called once for each incoming TSO message
 *  This handler is used to inform TM-Kit when a message has arrived by
 * calling TM_In..
 */
void TMHandler(MCAST_Addr Msg, long MsgSize, MCAST_Addr Context, long MsgType )
{
  struct MsgS *m;

  /* consistency checks */
  if (Context != &RTI)
  {printf ("TMHandler: unexpected context\n"); fflush(stdout); exit (1);}

  /* notify TM-Kit the message has been received */
  m = Msg;
  Core_ConsiderIncomingTSOMsg(m->TimeStamp, m->TMStuff); /*KALYAN*/
}

/*---------------------------------------------------------------------------*/
/**************************************************************************
*  Deliver TSO messages with time stamp <= TARTime */

fdkErrorCode DeliverOneMessage(TM_Time TimeStamp, char *cMsg, long MsgSize,
   long MsgType)
{
  EventRetractionHandle erh;
  struct MsgS *Msg;

  Msg = (struct MsgS *)cMsg;

  erh = Msg->erh;

  if (MsgType == MSG_RETRACTION)
  {
     /*  Forward retraction to federate and free memory when done */
     RequestRetraction(erh);
     MyHereProc(MsgSize,Msg);

  }
  else
  {
     /* If messages is not a retraction, first check if a retraction
	for this message has arrived, if not deliver to federate,
	if one has arrived anihilate message and free the retraction
	and message memory
     */
     int action;
     MCAST_Addr   MCAST_RetractMsg;
     struct MsgS *RetractMsg; 
     int          RMsgSize; 

     action = RHM_IsDeliverable(erh,&MCAST_RetractMsg,&RMsgSize); 
     RetractMsg = (struct MsgS*)MCAST_RetractMsg; 
     switch (action)
     {
       case NOTDELIVERABLE:
	     {
	          /* Retract Message, call my here proc for msg and retract */
		  MessagesAnihilated++;
	 
		  MyHereProc(MsgSize,Msg);
                  MyHereProc(RMsgSize,RetractMsg);
  		  break;
	     }
       case ISDELIVERABLE:
	     {
	          /* deliver message to federate */

  	    	  ReflectAttributeValues (TimeStamp, Msg, MsgSize, MsgType);

  		  break;
	     }
      default:
             {
	          /* error */
	          return(fdkDOMIMT); 
	     }
      }	
   }
   return(fdkSUCCEEDED);
}
 

/*---------------------------------------------------------------------------*/
fdkErrorCode DeliverEvents (TM_Time TARTime)
{
  void *Data;
  TM_Time minTS;
  long MsgSize, MsgType;

  /* deliver events from TSO queue with time stamp <= TARTime */
  while (TM_LE(HEAP_Min(BRTI.TSOQueue), TARTime))
  {
    Data = HEAP_DeleteWithType (BRTI.TSOQueue, &minTS, &MsgSize, &MsgType);
    DeliverOneMessage(minTS,Data,MsgSize,MsgType);
  }
  return(fdkSUCCEEDED);
}

/*---------------------------------------------------------------------------*/
void RTI_TimeAdvanceRequest (TM_Time AdvanceTime)
{
    fdkErrorCode retCode;

    if ((retCode = Core_TimeAdvanceRequest(AdvanceTime)) != fdkSUCCEEDED)
    {
       printf("TAR Error: %s\n",fdkErrorMessage(retCode));
       exit(1);
    }

}

/*---------------------------------------------------------------------------*/
void RTI_NextEventRequest (TM_Time TimeOfNextEvent)
{
    fdkErrorCode retCode;

    if ((retCode = Core_NextEventRequest(TimeOfNextEvent)) != fdkSUCCEEDED)
    {
       printf("NER Error: %s\n",fdkErrorMessage(retCode));
       exit(1);
    }
}

/*---------------------------------------------------------------------------*/
/**************************************************************************
 *  Procedure to make an MCAST group suitable for receiving TSO messages
 *  Main task is to set up group handlers and to add one TM handler
 *  to the group.
 */
void TSOifyGroup (MCAST_Handle Handle, MCAST_WhereProc WProc, void *Context)
{
  long dummy;

  /* set up incoming message group */
  if (MCAST_SetWhereProc(Handle, WProc, Context) != MCAST_Success)
    { printf ("SetWhereProc Failed\n"); exit (1); }

  /* attach time management handler (to call TM_In) to group */
  if (MCAST_Subscribe(Handle, TMHandler, &RTI, &dummy) != MCAST_Success)
    { printf ("Subscribe Failed\n"); exit (1); }

  /* tell core about this time-managed group*/
  Core_TSOifyGroup( Handle ); /*KALYAN*/
}

/*************************************************************************
 *************************************************************************
 *  DECLARATION MANAGEMENT SERVICES
 *  One multicast group is created for each object class.
 *  An object class designator is synonymous with a MCAST group handle
 *************************************************************************
 ************************************************************************/

/*************************************************************************
 *  RTI_GetObjClassHandle: convert class name to class designator
 ************************************************************************/
RTI_ObjClassDesignator RTI_GetObjClassHandle (char *ObjClassName)
{
  MCAST_Handle rtnval;

  if (MCAST_GetHandle(ObjClassName, &rtnval) != MCAST_SUCCESS)
    return (NULL);
  else
    return ((RTI_ObjClassDesignator) rtnval);
}


/*************************************************************************
 *  RTI_PublishObjClass: declare intent to publish objects of a class
 ************************************************************************/
void RTI_PublishObjClass (RTI_ObjClassDesignator Handle)
{
  /* nothing to do in current implementation */
  if (Handle == NULL)
  {printf ("RTI_PublishObjClass: invalid class designator\n"); exit(1);}
  Core_TSOifyGroup( Handle ); /*KALYAN*/
}


/*************************************************************************
 *  RTI_InitObjClassSubscription
 ************************************************************************/
void RTI_InitObjClassSubscription (RTI_ObjClassDesignator Handle,
	MCAST_WhereProc WProc, void *Context)
{
  if (MCAST_NLocalSubscriptions(Handle) != 0)
  {
    printf ("RTI_InitObjClassSubscription: Object Class already initialized\n");
    exit (1);
  }

  /* TSOify group */
  TSOifyGroup (Handle, WProc, Context);
}


/*************************************************************************
 *  RTI_TestIsSubscriptionInitialized
 ************************************************************************/
BOOLEAN RTI_IsClassSubscriptionInitialized (RTI_ObjClassDesignator Handle)
{
  if (MCAST_NLocalSubscriptions(Handle) != 0)
    return (TRUE);
  else
    return (FALSE);
}


/*************************************************************************
 *  RTI_SubscribeObjClassAttributes: subscribe to a class
 ************************************************************************/
void RTI_SubscribeObjClassAttributes (RTI_ObjClassDesignator Handle)
{
  long SubID;

  if (Handle == NULL)
  {
    printf ("RTI_SubscribeObjClassAttributes: invalid class designator\n");
    exit(1);
  }

  /* if subscription not already initialized, error */
  if (MCAST_NLocalSubscriptions(Handle) == 0)
  {
    printf ("RTI_SubscribeObjClassAttributes: uninitialized class subscription\n");
    printf ("Call RTI_InitObjClassSubscription to init class subscription\n");
    exit(1);
  }

  /* subscribe to group */
  if (MCAST_Subscribe(Handle, MyHandler, &RTI, &SubID) != MCAST_Success)
    { printf ("Subscribe Failed\n"); exit (1); }
}